diff --git a/pkg/pillar/cmd/zedkube/zedkube.go b/pkg/pillar/cmd/zedkube/zedkube.go index 5ef4baa7355..ef88412e1c2 100644 --- a/pkg/pillar/cmd/zedkube/zedkube.go +++ b/pkg/pillar/cmd/zedkube/zedkube.go @@ -50,6 +50,12 @@ const ( // kube-API failures and the boot race where ENCC arrives before leader // election settles are self-healing instead of stuck-until-next-config. pruneStaleMasterInterval = 600 + // longhornDiskReservedErrRelogInterval: how often a still-unchanged + // applyLonghornDiskReserved failure is re-logged at Error level. That + // reconcile runs every kubeCfgInterval forever with no success latch, so + // unthrottled logging would bury the journal during any lengthy kube-API + // or Longhorn outage. + longhornDiskReservedErrRelogInterval = 30 * time.Minute inlineCmdKubeClusterUpdateStatus = "pubKubeClusterUpdateStatus" inlineCmdVmiDetach = "vmiDetach" @@ -130,8 +136,12 @@ type zedkube struct { deschedulerOnBootStarted bool receivedENCC bool - // longhornDiskReservedSet is true once the desired reservation has been applied to the Longhorn node - longhornDiskReservedSet bool + // longhornDiskReservedErr / longhornDiskReservedErrLogged throttle the error + // logging in applyLonghornDiskReserved. That reconcile has no success latch, + // so a persistent failure would otherwise log identically every + // kubeCfgInterval for as long as it lasts. + longhornDiskReservedErr string + longhornDiskReservedErrLogged time.Time // longhornSnapshotSet is true once the desired recurring snapshot interval has been applied longhornSnapshotSet bool // longhornDrainPolicySet is true once the desired node-drain-policy has been applied @@ -949,9 +959,10 @@ func handleGlobalConfigImpl(ctxArg interface{}, key string, newReservedGB := newConfigItemValueMap.GlobalValueInt(types.LonghornDiskReservedGB) existingReservedGB := currentConfigItemValueMap.GlobalValueInt(types.LonghornDiskReservedGB) if newReservedGB != existingReservedGB { + // No latch to clear: applyLonghornDiskReserved below reconciles + // against the new value, as does every kubeCfgTimer tick after it. log.Functionf("handleGlobalConfigImpl: LonghornDiskReservedGB changed %d -> %d", existingReservedGB, newReservedGB) - z.longhornDiskReservedSet = false } newSnapshotCron := newConfigItemValueMap.GlobalValueString(types.LonghornSnapshotCron) @@ -981,26 +992,63 @@ func handleGlobalConfigImpl(ctxArg interface{}, key string, log.Functionf("handleGlobalConfigImpl(%s): done", key) } -// applyLonghornDiskReserved attempts to set the per-disk reserved space on the local Longhorn -// node. It is a no-op if the node name is not yet known or the value has already been applied. -// Callers should retry periodically until longhornDiskReservedSet is true. +// applyLonghornDiskReserved reconciles the per-disk reserved space on the local Longhorn +// node with storage.longhorn.disk.reserved.gigabytes. +// +// This runs on every kubeCfgTimer tick and deliberately keeps no "already applied" latch. +// Longhorn deletes and recreates the node object whenever this node leaves and rejoins the +// cluster, and the recreated object comes back carrying Longhorn's own default reservation +// (30% of the disk, per storage-reserved-percentage-for-default-disk). A latched apply +// would never notice, leaving the EVE value silently unenforced until zedbox restarts — +// and on a full disk that difference is what stops new replicas from being scheduled. +// Steady state costs one Get per interval; an Update is only issued on drift. func (z *zedkube) applyLonghornDiskReserved() { - if z.nodeName == "" || z.longhornDiskReservedSet { + if z.nodeName == "" { + return + } + // The tie-breaker node holds no Longhorn replicas, so a disk reservation is + // meaningless there and the Longhorn validating webhook rejects the update + // anyway. Identify it from the controller's cluster config only — the node's + // Longhorn Schedulable condition tracks the Kubernetes cordon, which every + // node passes through at boot until nodeOnBootHealthStatusWatcher uncordons + // it, and during every drain. + if z.clusterConfig.IsTieBreakerNode(z.nodeuuid) { return } reservedGB := z.globalConfig.GlobalValueInt(types.LonghornDiskReservedGB) if reservedGB == types.LonghornDiskReservedGBDisabled { // Operator disabled EVE's override; leave Longhorn's current value in place. - z.longhornDiskReservedSet = true return } reservedBytes := int64(reservedGB) * 1024 * 1024 * 1024 - applied, err := kubeapi.SetLonghornNodeDiskReserved(z.nodeName, reservedBytes) + updated, err := kubeapi.SetLonghornNodeDiskReserved(z.nodeName, reservedBytes) if err != nil { - log.Errorf("applyLonghornDiskReserved: %v", err) + z.logLonghornDiskReservedErr(err) + return + } + z.longhornDiskReservedErr = "" + if updated { + log.Noticef("applyLonghornDiskReserved: set reserved space to %d GB on node %s", + reservedGB, z.nodeName) + } +} + +// logLonghornDiskReservedErr logs a reconcile failure at Error the first time it is seen +// and then at most once per longhornDiskReservedErrRelogInterval while the same failure +// persists; repeats in between drop to Function level. Without this an unreachable kube +// API would emit one identical Error every kubeCfgInterval indefinitely, since +// applyLonghornDiskReserved never stops retrying. +func (z *zedkube) logLonghornDiskReservedErr(err error) { + msg := err.Error() + now := time.Now() + if msg == z.longhornDiskReservedErr && + now.Sub(z.longhornDiskReservedErrLogged) < longhornDiskReservedErrRelogInterval { + log.Functionf("applyLonghornDiskReserved: %v", err) return } - z.longhornDiskReservedSet = applied + z.longhornDiskReservedErr = msg + z.longhornDiskReservedErrLogged = now + log.Errorf("applyLonghornDiskReserved: %v", err) } // applyLonghornRecurringSnapshot creates or updates the Longhorn recurring snapshot job diff --git a/pkg/pillar/kubeapi/longhorninfo.go b/pkg/pillar/kubeapi/longhorninfo.go index 0b8c0837137..ef58d870ceb 100644 --- a/pkg/pillar/kubeapi/longhorninfo.go +++ b/pkg/pillar/kubeapi/longhorninfo.go @@ -708,10 +708,19 @@ func longhornVolumeSetNode(lhVolName string, kubeNodeName string) error { } // SetLonghornNodeDiskReserved sets StorageReserved on every disk of the named Longhorn node. -// Returns (false, nil) if the Longhorn API is absent or the node object does not exist yet, -// so callers should retry until (true, nil) is returned. -// Returns (true, nil) for non-schedulable (tie-breaker) nodes: the reservation is not -// needed and the Longhorn admission webhook would reject any update attempt. +// It reads the node on every call and only issues an Update when a disk differs from +// reservedBytes, so it is cheap to call repeatedly and is meant to be driven as a standing +// reconcile rather than once per boot. Longhorn recreates the node object (with its own +// 30% default reservation) whenever the node leaves and rejoins the cluster, and only a +// standing reconcile notices that. +// +// Returns (true, nil) when an Update was issued, (false, nil) when the node already matched, +// when Longhorn is not installed yet, or when the Longhorn node object does not exist yet. +// +// This function deliberately makes no attempt to recognize tie-breaker nodes. Callers must +// skip those themselves using EdgeNodeClusterConfig.IsTieBreakerNode: a node's Longhorn +// Schedulable condition is not a usable proxy, since Longhorn sets it False for any +// Kubernetes cordon — including the boot-time cordon every node passes through. func SetLonghornNodeDiskReserved(nodeName string, reservedBytes int64) (bool, error) { apiExists, err := longhornAPIExists() if !apiExists && err == nil { @@ -739,7 +748,8 @@ func SetLonghornNodeDiskReserved(nodeName string, reservedBytes int64) (bool, er // setLonghornNodeDiskReservedInner is the testable core of SetLonghornNodeDiskReserved. // All Longhorn node I/O is injected through the nodes interface argument so unit tests -// can supply hand-written mocks without a live cluster. +// can supply hand-written mocks without a live cluster. The returned bool reports +// whether an Update was issued, so a steady state costs one Get per call. func setLonghornNodeDiskReservedInner(ctx context.Context, nodeName string, reservedBytes int64, nodes lhNodeGetUpdater) (bool, error) { node, err := nodes.Get(ctx, nodeName, metav1.GetOptions{}) @@ -750,22 +760,6 @@ func setLonghornNodeDiskReservedInner(ctx context.Context, nodeName string, return false, fmt.Errorf("SetLonghornNodeDiskReserved: get node %s: %v", nodeName, err) } - // Tie breaker nodes will have a non-deployed engine and the longhorn - // validator will return an error. - // example: - // 'admission webhook "validator.longhorn.io" denied the request: - // spec and status of disks on node are being syncing - // and please retry later.' - // - // Skip this node, the reservation isn't necessary here. - // Return true so the caller stops retrying — the reservation is not needed. - for _, cond := range node.Status.Conditions { - if cond.Type == lhv1beta2.NodeConditionTypeSchedulable && - cond.Status != lhv1beta2.ConditionStatusTrue { - return true, nil - } - } - changed := false for key, disk := range node.Spec.Disks { if disk.StorageReserved != reservedBytes { @@ -775,7 +769,8 @@ func setLonghornNodeDiskReservedInner(ctx context.Context, nodeName string, } } if !changed { - return true, nil + // Already in sync; nothing to write. + return false, nil } _, err = nodes.Update(ctx, node, metav1.UpdateOptions{}) diff --git a/pkg/pillar/kubeapi/longhorninfo_test.go b/pkg/pillar/kubeapi/longhorninfo_test.go index 4a8f2f7429e..62c6400a19b 100644 --- a/pkg/pillar/kubeapi/longhorninfo_test.go +++ b/pkg/pillar/kubeapi/longhorninfo_test.go @@ -816,16 +816,18 @@ func TestSetLonghornNodeDiskReservedInner(t *testing.T) { name string getFn func(string) (*lhv1beta2.Node, error) updateFn func(*lhv1beta2.Node) (*lhv1beta2.Node, error) - wantApplied bool + wantUpdated bool wantErr bool // updateCalled asserts whether Update was (or was not) invoked. wantUpdateCalled bool }{ { - // Non-schedulable node (tie-breaker): reservation is not needed and - // the Longhorn admission webhook would reject any update. Return true - // so the caller stops retrying. - name: "non-schedulable node returns true without updating", + // A cordoned node reports Schedulable=False. That must NOT be read as + // "tie-breaker, skip": every node is cordoned at boot and during drains, + // so skipping here would silently leave ordinary storage nodes on + // Longhorn's default reservation. Tie-breakers are excluded by the + // caller via EdgeNodeClusterConfig.IsTieBreakerNode. + name: "schedulable=False node is still reconciled", getFn: func(string) (*lhv1beta2.Node, error) { node := lhNodeWithDisks(differentVal) node.Status.Conditions = []lhv1beta2.Condition{ @@ -833,13 +835,20 @@ func TestSetLonghornNodeDiskReservedInner(t *testing.T) { } return node, nil }, - updateFn: func(n *lhv1beta2.Node) (*lhv1beta2.Node, error) { return n, nil }, - wantApplied: true, + updateFn: func(n *lhv1beta2.Node) (*lhv1beta2.Node, error) { + for _, disk := range n.Spec.Disks { + if disk.StorageReserved != wantReserved { + return nil, errors.New("disk not updated to wantReserved") + } + } + return n, nil + }, + wantUpdated: true, wantErr: false, - wantUpdateCalled: false, + wantUpdateCalled: true, }, { - // Schedulable condition present and True: normal node, proceeds to disk check. + // Steady state: value already correct, so no write and no drift to report. name: "schedulable=True node with correct reservation is a no-op", getFn: func(string) (*lhv1beta2.Node, error) { node := lhNodeWithDisks(alreadySet) @@ -849,25 +858,24 @@ func TestSetLonghornNodeDiskReservedInner(t *testing.T) { return node, nil }, updateFn: func(n *lhv1beta2.Node) (*lhv1beta2.Node, error) { return n, nil }, - wantApplied: true, + wantUpdated: false, wantErr: false, wantUpdateCalled: false, }, { - // No schedulable condition at all (node not yet registered by Longhorn): treat as - // schedulable and proceed to disk check. + // No conditions at all (node object freshly created by Longhorn). name: "no schedulable condition, reservation already set — no-op", getFn: func(string) (*lhv1beta2.Node, error) { return lhNodeWithDisks(alreadySet), nil }, updateFn: func(n *lhv1beta2.Node) (*lhv1beta2.Node, error) { return n, nil }, - wantApplied: true, + wantUpdated: false, wantErr: false, wantUpdateCalled: false, }, { - // Disks have the wrong reservation: Update must be called with the - // corrected value and the function must return (true, nil). + // Drift correction: this is the path a recreated node object takes when + // it comes back with Longhorn's 30% default instead of the EVE value. name: "disks need update — Update called with corrected value", getFn: func(string) (*lhv1beta2.Node, error) { return lhNodeWithDisks(differentVal), nil @@ -880,18 +888,30 @@ func TestSetLonghornNodeDiskReservedInner(t *testing.T) { } return n, nil }, - wantApplied: true, + wantUpdated: true, wantErr: false, wantUpdateCalled: true, }, { - // Node object not yet created by Longhorn: signal retry with (false, nil). + // Node object not yet created by Longhorn: nothing to write, no error. name: "node not found returns false without error", getFn: func(string) (*lhv1beta2.Node, error) { return nil, nodeNotFound }, updateFn: func(n *lhv1beta2.Node) (*lhv1beta2.Node, error) { return n, nil }, - wantApplied: false, + wantUpdated: false, + wantErr: false, + wantUpdateCalled: false, + }, + { + // An empty disk map must not be mistaken for "in sync and done" in a + // way that stops future reconciles; the caller retries next tick. + name: "node with no disks yet is a no-op", + getFn: func(string) (*lhv1beta2.Node, error) { + return &lhv1beta2.Node{}, nil + }, + updateFn: func(n *lhv1beta2.Node) (*lhv1beta2.Node, error) { return n, nil }, + wantUpdated: false, wantErr: false, wantUpdateCalled: false, }, @@ -901,11 +921,13 @@ func TestSetLonghornNodeDiskReservedInner(t *testing.T) { return nil, errors.New("kube api unavailable") }, updateFn: func(n *lhv1beta2.Node) (*lhv1beta2.Node, error) { return n, nil }, - wantApplied: false, + wantUpdated: false, wantErr: true, wantUpdateCalled: false, }, { + // The Longhorn validator rejects updates while disk spec and status are + // syncing. That is transient: surface the error so the caller retries. name: "Update returns error", getFn: func(string) (*lhv1beta2.Node, error) { return lhNodeWithDisks(differentVal), nil @@ -913,7 +935,7 @@ func TestSetLonghornNodeDiskReservedInner(t *testing.T) { updateFn: func(n *lhv1beta2.Node) (*lhv1beta2.Node, error) { return nil, errors.New("webhook denied") }, - wantApplied: false, + wantUpdated: false, wantErr: true, wantUpdateCalled: true, }, @@ -928,7 +950,7 @@ func TestSetLonghornNodeDiskReservedInner(t *testing.T) { } mock := funcLHNodeGetUpdater{getFn: tc.getFn, updateFn: wrappedUpdate} - applied, err := setLonghornNodeDiskReservedInner( + updated, err := setLonghornNodeDiskReservedInner( context.Background(), nodeName, wantReserved, mock) if tc.wantErr { @@ -940,8 +962,8 @@ func TestSetLonghornNodeDiskReservedInner(t *testing.T) { t.Errorf("unexpected error: %v", err) } } - if applied != tc.wantApplied { - t.Errorf("applied = %v, want %v", applied, tc.wantApplied) + if updated != tc.wantUpdated { + t.Errorf("updated = %v, want %v", updated, tc.wantUpdated) } if updateCalled != tc.wantUpdateCalled { t.Errorf("updateCalled = %v, want %v", updateCalled, tc.wantUpdateCalled) diff --git a/pkg/pillar/types/clustertypes.go b/pkg/pillar/types/clustertypes.go index f1b3a61eb58..305c2607999 100644 --- a/pkg/pillar/types/clustertypes.go +++ b/pkg/pillar/types/clustertypes.go @@ -185,6 +185,30 @@ func (config EdgeNodeClusterConfig) NativeK8sOrchestrationEnabled() bool { config.EnableNativeK8SOrchestration } +// IsTieBreakerNode reports whether nodeUUID is the tie-breaker node the +// controller designated for this cluster. A tie-breaker node exists only to +// hold a quorum vote: it is kept cordoned, runs no workloads and carries no +// Longhorn replicas, so node-local storage configuration does not apply to it. +// +// This is the authoritative test. Do not infer tie-breaker-ness from a node +// being unschedulable: Longhorn drives its node Schedulable condition off the +// Kubernetes cordon, which every node passes through at boot and during +// drains, so that signal would misclassify ordinary storage nodes. +// +// Returns false when the controller designated no tie-breaker (zero UUID) or +// nodeUUID does not parse, so a node is only treated as the tie-breaker on +// positive evidence from the EVE API config. +func (config EdgeNodeClusterConfig) IsTieBreakerNode(nodeUUID string) bool { + if config.TieBreakerNodeID.UUID == uuid.Nil { + return false + } + parsed, err := uuid.FromString(nodeUUID) + if err != nil { + return false + } + return parsed == config.TieBreakerNodeID.UUID +} + // EdgeNodeClusterStatus - Status of the multi-node cluster published by zedkube type EdgeNodeClusterStatus struct { ClusterName string diff --git a/pkg/pillar/types/clustertypes_test.go b/pkg/pillar/types/clustertypes_test.go index 7d86586ea76..a8aa9552279 100644 --- a/pkg/pillar/types/clustertypes_test.go +++ b/pkg/pillar/types/clustertypes_test.go @@ -11,6 +11,7 @@ import ( "strings" "testing" + uuid "github.com/satori/go.uuid" "github.com/stretchr/testify/assert" ) @@ -72,6 +73,41 @@ func TestNativeK8sOrchestrationEnabled(t *testing.T) { } } +// EdgeNodeClusterConfig.IsTieBreakerNode + +func TestIsTieBreakerNode(t *testing.T) { + const ( + tieBreaker = "5a0283de-d69a-4d4a-877f-63f9fcfcb929" + otherNode = "e3342f18-c354-49e7-9a00-f3ff3a2f3db0" + ) + tbUUID, err := uuid.FromString(tieBreaker) + assert.NoError(t, err) + + cases := []struct { + name string + configured uuid.UUID + nodeUUID string + want bool + }{ + {"designated tie-breaker matches", tbUUID, tieBreaker, true}, + {"tie-breaker matches upper case", tbUUID, strings.ToUpper(tieBreaker), true}, + {"storage node does not match", tbUUID, otherNode, false}, + // No tie-breaker designated: no node may be treated as one, otherwise a + // node with an empty UUID would match the zero value. + {"no tie-breaker designated", uuid.Nil, otherNode, false}, + {"no tie-breaker designated, empty node uuid", uuid.Nil, "", false}, + {"unparsable node uuid", tbUUID, "not-a-uuid", false}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + cfg := EdgeNodeClusterConfig{ + TieBreakerNodeID: UUIDandVersion{UUID: tc.configured}, + } + assert.Equal(t, tc.want, cfg.IsTieBreakerNode(tc.nodeUUID)) + }) + } +} + func TestVmiVNCConfig_JSONRoundTrip(t *testing.T) { cases := []struct { name string