diff --git a/api/v1alpha1/status.go b/api/v1alpha1/status.go index b6ccf400437..4f98bca2bea 100644 --- a/api/v1alpha1/status.go +++ b/api/v1alpha1/status.go @@ -191,8 +191,6 @@ const ( RuntimeFusesScaledIn RuntimeConditionType = "FusesScaledIn" // RuntimeFusesScaledOut means the fuses of runtime just scaled out RuntimeFusesScaledOut RuntimeConditionType = "FusesScaledOut" - // RuntimeWorkerDecommissioning means the runtime is draining workers ahead of a scale-down - RuntimeWorkerDecommissioning RuntimeConditionType = "WorkerDecommissioning" ) const ( @@ -216,8 +214,6 @@ const ( RuntimeFusesScaledInReason = "Fuses scaled in" // RuntimeFusesScaledInReason means the fuses of runtime just scaled out RuntimeFusesScaledOutReason = "Fuses scaled out" - // RuntimeWorkerDecommissioningReason means workers are being decommissioned ahead of a scale-down - RuntimeWorkerDecommissioningReason = "Workers are being decommissioned" ) // Condition describes the state of the cache at a certain point. diff --git a/pkg/ddc/alluxio/const.go b/pkg/ddc/alluxio/const.go index 08c72ba9815..833988ec453 100644 --- a/pkg/ddc/alluxio/const.go +++ b/pkg/ddc/alluxio/const.go @@ -16,8 +16,6 @@ limitations under the License. package alluxio -import "time" - const ( // NON_NATIVE_MOUNT_DATA_NAME also used in master 'statefulset.yaml' and config 'alluxio-mount.conf.yaml' NON_NATIVE_MOUNT_DATA_NAME = "mount.info" @@ -59,16 +57,6 @@ const ( defaultGracefulShutdownLimits int32 = 3 defaultCleanCacheGracePeriodSeconds int32 = 60 - // defaultWorkerRPCPort is the Alluxio worker Thrift RPC port used when the - // runtime spec does not override alluxio.worker.rpc.port. - defaultWorkerRPCPort = 29999 - MountConfigStorage = "ALLUXIO_MOUNT_CONFIG_STORAGE" ConfigmapStorageName = "configmap" - - // defaultWorkerDecommissionDeadline bounds how long the engine keeps - // retrying a stuck worker drain (e.g. an unhealthy master, unreplicable - // blocks) before forcing the scale-down to proceed anyway, rather than - // stalling on every reconcile indefinitely. - defaultWorkerDecommissionDeadline = 10 * time.Minute ) diff --git a/pkg/ddc/alluxio/operations/decommission.go b/pkg/ddc/alluxio/operations/decommission.go deleted file mode 100644 index 61c0a7b7b13..00000000000 --- a/pkg/ddc/alluxio/operations/decommission.go +++ /dev/null @@ -1,86 +0,0 @@ -/* -Copyright 2026 The Fluid Authors. - -Licensed under the Apache License, Version 2.0 (the "License"); -you may not use this file except in compliance with the License. -You may obtain a copy of the License at - - http://www.apache.org/licenses/LICENSE-2.0 - -Unless required by applicable law or agreed to in writing, software -distributed under the License is distributed on an "AS IS" BASIS, -WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -See the License for the specific language governing permissions and -limitations under the License. -*/ - -package operations - -import "strings" - -// DecommissionWorkers signals the Alluxio master to decommission the given -// workers. Each address must be in ":" form. -// The call is idempotent: re-issuing it against an already-decommissioned -// worker is safe. -// -// Requires Alluxio >= 2.9, where "fsadmin decommissionWorker" was introduced; -// against older masters this subcommand does not exist. -func (a AlluxioFileUtils) DecommissionWorkers(addresses []string) error { - if len(addresses) == 0 { - return nil - } - command := []string{ - "alluxio", "fsadmin", "decommissionWorker", - "--addresses", strings.Join(addresses, ","), - } - _, _, err := a.exec(command, false) - if err != nil { - a.log.Error(err, "AlluxioFileUtils.DecommissionWorkers() failed", "addresses", addresses) - } - return err -} - -// CountActiveWorkers returns the number of live workers according to -// "alluxio fsadmin report capacity -live". The "-live" flag is what makes -// this safe to compare against immediately after a decommission: it asks the -// master for currently live workers rather than every worker it still has a -// record of, so a worker that was just decommissioned doesn't linger in the -// count until its heartbeat times out. -func (a AlluxioFileUtils) CountActiveWorkers() (int, error) { - report, _, err := a.exec([]string{"alluxio", "fsadmin", "report", "capacity", "-live"}, false) - if err != nil { - a.log.Error(err, "AlluxioFileUtils.CountActiveWorkers() failed") - return 0, err - } - return parseActiveWorkerCount(report), nil -} - -// parseActiveWorkerCount counts workers in the capacity report produced by -// "alluxio fsadmin report capacity". Worker entries begin at the non-indented -// line after the "Worker Name" header; the indented line that follows each -// entry contains the used-capacity detail. -// -// Worker Name Last Heartbeat Storage MEM -// 192.168.1.147 0 capacity 2048.00MB <- worker entry -// used 443.89MB (21%) <- detail, indented -// 192.168.1.146 0 capacity 2048.00MB <- worker entry -// used 0B (0%) -func parseActiveWorkerCount(report string) int { - inWorkerSection := false - count := 0 - for _, line := range strings.Split(report, "\n") { - if strings.HasPrefix(line, "Worker Name") { - inWorkerSection = true - continue - } - if !inWorkerSection || strings.TrimSpace(line) == "" { - continue - } - // Non-indented lines are new worker entries; indented lines are - // the used-capacity continuation for the previous entry. - if line[0] != ' ' && line[0] != '\t' { - count++ - } - } - return count -} diff --git a/pkg/ddc/alluxio/operations/decommission_test.go b/pkg/ddc/alluxio/operations/decommission_test.go deleted file mode 100644 index 9b1dbc2486f..00000000000 --- a/pkg/ddc/alluxio/operations/decommission_test.go +++ /dev/null @@ -1,227 +0,0 @@ -/* -Copyright 2026 The Fluid Authors. - -Licensed under the Apache License, Version 2.0 (the "License"); -you may not use this file except in compliance with the License. -You may obtain a copy of the License at - - http://www.apache.org/licenses/LICENSE-2.0 - -Unless required by applicable law or agreed to in writing, software -distributed under the License is distributed on an "AS IS" BASIS, -WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -See the License for the specific language governing permissions and -limitations under the License. -*/ - -package operations - -import ( - "errors" - "testing" - - "github.com/agiledragon/gomonkey/v2" - "github.com/fluid-cloudnative/fluid/pkg/utils/fake" -) - -func TestAlluxioFileUtils_DecommissionWorkers(t *testing.T) { - a := &AlluxioFileUtils{log: fake.NullLogger()} - - t.Run("empty address list is a no-op", func(t *testing.T) { - if err := a.DecommissionWorkers(nil); err != nil { - t.Fatalf("want nil, got: %v", err) - } - if err := a.DecommissionWorkers([]string{}); err != nil { - t.Fatalf("want nil, got: %v", err) - } - }) - - t.Run("exec error is propagated", func(t *testing.T) { - patches := gomonkey.ApplyFunc(AlluxioFileUtils.exec, - func(_ AlluxioFileUtils, _ []string, _ bool) (string, string, error) { - return "", "", errors.New("exec failed") - }) - defer patches.Reset() - - if err := a.DecommissionWorkers([]string{"192.168.1.1:29999"}); err == nil { - t.Error("want error, got nil") - } - }) - - t.Run("address is forwarded to the alluxio CLI", func(t *testing.T) { - var capturedCmd []string - patches := gomonkey.ApplyFunc(AlluxioFileUtils.exec, - func(_ AlluxioFileUtils, cmd []string, _ bool) (string, string, error) { - capturedCmd = cmd - return "", "", nil - }) - defer patches.Reset() - - addr := "192.168.1.1:29999" - if err := a.DecommissionWorkers([]string{addr}); err != nil { - t.Fatalf("want nil, got: %v", err) - } - found := false - for _, arg := range capturedCmd { - if arg == addr { - found = true - break - } - } - if !found { - t.Errorf("address %q not found in command: %v", addr, capturedCmd) - } - }) - - t.Run("multiple addresses are joined with commas", func(t *testing.T) { - var capturedCmd []string - patches := gomonkey.ApplyFunc(AlluxioFileUtils.exec, - func(_ AlluxioFileUtils, cmd []string, _ bool) (string, string, error) { - capturedCmd = cmd - return "", "", nil - }) - defer patches.Reset() - - if err := a.DecommissionWorkers([]string{"10.0.0.1:29999", "10.0.0.2:29999"}); err != nil { - t.Fatalf("want nil, got: %v", err) - } - found := false - for _, arg := range capturedCmd { - if arg == "10.0.0.1:29999,10.0.0.2:29999" { - found = true - break - } - } - if !found { - t.Errorf("joined addresses not found in command: %v", capturedCmd) - } - }) -} - -func TestAlluxioFileUtils_CountActiveWorkers(t *testing.T) { - a := &AlluxioFileUtils{log: fake.NullLogger()} - - t.Run("exec error returns zero and the error", func(t *testing.T) { - patches := gomonkey.ApplyFunc(AlluxioFileUtils.exec, - func(_ AlluxioFileUtils, _ []string, _ bool) (string, string, error) { - return "", "", errors.New("exec failed") - }) - defer patches.Reset() - - count, err := a.CountActiveWorkers() - if err == nil { - t.Error("want error, got nil") - } - if count != 0 { - t.Errorf("want 0 on error, got %d", count) - } - }) - - t.Run("requests live workers only", func(t *testing.T) { - var capturedCmd []string - patches := gomonkey.ApplyFunc(AlluxioFileUtils.exec, - func(_ AlluxioFileUtils, cmd []string, _ bool) (string, string, error) { - capturedCmd = cmd - return "", "", nil - }) - defer patches.Reset() - - if _, err := a.CountActiveWorkers(); err != nil { - t.Fatalf("want nil, got: %v", err) - } - found := false - for _, arg := range capturedCmd { - if arg == "-live" { - found = true - break - } - } - if !found { - t.Errorf("-live flag not found in command: %v", capturedCmd) - } - }) - - t.Run("two active workers", func(t *testing.T) { - report := `Capacity information for all workers: - Total Capacity: 4096.00MB - Used Capacity: 443.89MB - -Worker Name Last Heartbeat Storage MEM -192.168.1.147 0 capacity 2048.00MB - used 443.89MB (21%) -192.168.1.146 0 capacity 2048.00MB - used 0B (0%) -` - patches := gomonkey.ApplyFunc(AlluxioFileUtils.exec, - func(_ AlluxioFileUtils, _ []string, _ bool) (string, string, error) { - return report, "", nil - }) - defer patches.Reset() - - count, err := a.CountActiveWorkers() - if err != nil { - t.Fatalf("want nil, got: %v", err) - } - if count != 2 { - t.Errorf("want 2, got %d", count) - } - }) -} - -func TestParseActiveWorkerCount(t *testing.T) { - cases := []struct { - name string - input string - expect int - }{ - { - name: "empty report", - input: "", - expect: 0, - }, - { - name: "no worker section header", - input: "Capacity information for all workers:\n Total Capacity: 0B\n", - expect: 0, - }, - { - name: "single worker", - input: `Worker Name Last Heartbeat Storage MEM -192.168.1.1 0 capacity 1024.00MB - used 0B (0%) -`, - expect: 1, - }, - { - name: "three workers", - input: `Worker Name Last Heartbeat Storage MEM -10.0.0.1 0 capacity 2048.00MB - used 100MB (5%) -10.0.0.2 0 capacity 2048.00MB - used 0B (0%) -10.0.0.3 0 capacity 2048.00MB - used 500MB (25%) -`, - expect: 3, - }, - { - name: "trailing blank lines are ignored", - input: `Worker Name Last Heartbeat Storage MEM -10.0.0.1 0 capacity 1024.00MB - used 0B (0%) - - -`, - expect: 1, - }, - } - - for _, tc := range cases { - t.Run(tc.name, func(t *testing.T) { - got := parseActiveWorkerCount(tc.input) - if got != tc.expect { - t.Errorf("want %d, got %d", tc.expect, got) - } - }) - } -} diff --git a/pkg/ddc/alluxio/replicas.go b/pkg/ddc/alluxio/replicas.go index 83848c890f2..f149578869f 100644 --- a/pkg/ddc/alluxio/replicas.go +++ b/pkg/ddc/alluxio/replicas.go @@ -18,32 +18,19 @@ package alluxio import ( "context" - stderrors "errors" "fmt" "reflect" - "time" data "github.com/fluid-cloudnative/fluid/api/v1alpha1" "github.com/fluid-cloudnative/fluid/pkg/ctrl" - "github.com/fluid-cloudnative/fluid/pkg/ddc/alluxio/operations" - "github.com/fluid-cloudnative/fluid/pkg/features" cruntime "github.com/fluid-cloudnative/fluid/pkg/runtime" "github.com/fluid-cloudnative/fluid/pkg/utils" - utilfeature "github.com/fluid-cloudnative/fluid/pkg/utils/feature" corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/errors" - metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/types" "k8s.io/client-go/util/retry" ) -// errWorkersNotYetDrained marks the normal, transient state during scale-in -// where the targeted workers have not finished migrating their cached blocks -// to the surviving workers yet. It lets the caller log this at Info level -// instead of Error, while still propagating a non-nil error so the existing -// fixed-interval reconcile requeue (see runtime_controller.go) kicks in. -var errWorkersNotYetDrained = stderrors.New("workers not yet drained") - // SyncReplicas syncs the replicas func (e *AlluxioEngine) SyncReplicas(ctx cruntime.ReconcileRequestContext) (err error) { err = retry.RetryOnConflict(retry.DefaultBackoff, func() error { @@ -101,190 +88,12 @@ func (e *AlluxioEngine) SyncReplicas(ctx cruntime.ReconcileRequestContext) (err return err } runtimeToUpdate := runtime.DeepCopy() - - // When the GracefulWorkerScaleDown feature is enabled and we detect a - // scale-in, decommission the targeted workers before the StatefulSet - // controller terminates them. This gives the Alluxio master a chance to - // migrate their cached blocks to the surviving workers. The reconciler - // requeues until the active worker count has dropped to the desired - // level. - // - // workers.Status.Replicas (the number of Pods the StatefulSet controller - // has actually created) is used rather than workers.Spec.Replicas: the - // spec is the target this engine itself lowers once a drain succeeds, so - // relying on it could under-count pods that still exist but whose spec - // update already landed. - if utilfeature.DefaultFeatureGate.Enabled(features.GracefulWorkerScaleDown) && - runtime.Replicas() < workers.Status.Replicas { - - decommissionStart, alreadyTracked := getDecommissionStart(runtime) - if !alreadyTracked { - decommissionStart = time.Now() - } - - drained, drainErr := e.drainScalingDownWorkers(ctx, runtime, runtime.Replicas(), workers.Status.Replicas) - if drainErr != nil { - return drainErr - } - - if !drained { - elapsed := time.Since(decommissionStart) - if elapsed > defaultWorkerDecommissionDeadline { - // A worker that never finishes draining (unhealthy master, - // unreplicable blocks, ...) would otherwise stall scale-down - // forever. Past the deadline we fall through and proceed - // anyway so the StatefulSet still converges; any data loss - // risk this avoided is the same the cluster accepts today - // without this feature. - e.Log.Info("Worker decommission exceeded the deadline; forcing scale-down to proceed", - "elapsed", elapsed, "deadline", defaultWorkerDecommissionDeadline) - } else { - if !alreadyTracked { - runtimeToUpdate.Status.Conditions = utils.UpdateRuntimeCondition( - runtimeToUpdate.Status.Conditions, newDecommissioningCondition(decommissionStart)) - if updateErr := e.Client.Status().Update(ctx, runtimeToUpdate); updateErr != nil { - return updateErr - } - } - return fmt.Errorf("%w: scale-in to %d replicas will resume on next reconcile", - errWorkersNotYetDrained, runtime.Replicas()) - } - } - - if alreadyTracked { - runtimeToUpdate.Status.Conditions = clearDecommissioningCondition(runtimeToUpdate.Status.Conditions) - if updateErr := e.Client.Status().Update(ctx, runtimeToUpdate); updateErr != nil { - return updateErr - } - } - } - err = e.Helper.SyncReplicas(ctx, runtimeToUpdate, runtimeToUpdate.Status, workers) return err }) if err != nil { - if stderrors.Is(err, errWorkersNotYetDrained) { - e.Log.Info(err.Error(), "name", e.name, "namespace", e.namespace) - } else { - _ = utils.LoggingErrorExceptConflict(e.Log, err, "Failed to sync replicas", types.NamespacedName{Namespace: e.namespace, Name: e.name}) - } + _ = utils.LoggingErrorExceptConflict(e.Log, err, "Failed to sync replicas", types.NamespacedName{Namespace: e.namespace, Name: e.name}) } return } - -// drainScalingDownWorkers decommissions the Alluxio workers that are about to be -// removed when scaling from currentReplicas down to desiredReplicas. -// -// A standard StatefulSet removes the highest-ordinal pods first, so the targets -// are ordinals [desiredReplicas, currentReplicas). The function issues a -// decommission request via the master and returns whether Alluxio's active -// worker count has already dropped to the desired level. -func (e *AlluxioEngine) drainScalingDownWorkers(ctx context.Context, runtime *data.AlluxioRuntime, desiredReplicas, currentReplicas int32) (bool, error) { - masterPodName, masterContainerName := e.getMasterPodInfo() - fileUtils := operations.NewAlluxioFileUtils(masterPodName, masterContainerName, e.namespace, e.Log) - - workerRPCPort := e.getWorkerRPCPort(runtime) - workerStsName := e.getWorkerName() - - // Collect RPC addresses of the pods that will be terminated on scale-down. - // The worker registers with the master under its node's IP (see the - // ALLUXIO_WORKER_HOSTNAME wiring in charts/alluxio, which sources - // alluxio.worker.hostname from status.hostIP), not its pod IP, so that is - // the identity "fsadmin decommissionWorker" must be addressed by. - // - // Pods sharing a node produce the same HostIP; seen tracks addresses - // already added so the request doesn't list the same worker twice. - var toDecommission []string - seen := make(map[string]struct{}) - for ord := desiredReplicas; ord < currentReplicas; ord++ { - podName := fmt.Sprintf("%s-%d", workerStsName, ord) - pod := &corev1.Pod{} - if err := e.Client.Get(ctx, - types.NamespacedName{Name: podName, Namespace: e.namespace}, pod); err != nil { - if errors.IsNotFound(err) { - // Pod is already gone; nothing to decommission here. - continue - } - return false, err - } - if pod.Status.HostIP == "" { - e.Log.Info("Worker pod has no host IP yet, will retry", "pod", podName) - return false, nil - } - addr := fmt.Sprintf("%s:%d", pod.Status.HostIP, workerRPCPort) - if _, dup := seen[addr]; dup { - continue - } - seen[addr] = struct{}{} - toDecommission = append(toDecommission, addr) - } - - if len(toDecommission) == 0 { - // All targeted pods are already gone from the cluster. - return true, nil - } - - if err := fileUtils.DecommissionWorkers(toDecommission); err != nil { - return false, err - } - - activeCount, err := fileUtils.CountActiveWorkers() - if err != nil { - return false, err - } - - if int32(activeCount) > desiredReplicas { - e.Log.Info("Workers are still draining, will retry", - "activeWorkers", activeCount, "desired", desiredReplicas) - return false, nil - } - - return true, nil -} - -// getWorkerRPCPort returns the configured Alluxio worker RPC port, falling back -// to the Alluxio default when the runtime does not override it. -func (e *AlluxioEngine) getWorkerRPCPort(runtime *data.AlluxioRuntime) int { - if port, ok := runtime.Spec.Worker.Ports["rpc"]; ok && port > 0 { - return port - } - return defaultWorkerRPCPort -} - -// getDecommissionStart returns when the current worker-drain attempt began, -// based on the RuntimeWorkerDecommissioning condition set the first time a -// scale-down's drain didn't finish within one reconcile. The bool reports -// whether such an in-progress attempt is already being tracked. -func getDecommissionStart(runtime *data.AlluxioRuntime) (time.Time, bool) { - _, cond := utils.GetRuntimeCondition(runtime.Status.Conditions, data.RuntimeWorkerDecommissioning) - if cond == nil || cond.Status != corev1.ConditionTrue { - return time.Time{}, false - } - return cond.LastTransitionTime.Time, true -} - -// newDecommissioningCondition marks the start of a worker-drain attempt that -// didn't complete within one reconcile, so subsequent reconciles can measure -// elapsed time against defaultWorkerDecommissionDeadline. -func newDecommissioningCondition(start time.Time) data.RuntimeCondition { - cond := utils.NewRuntimeCondition(data.RuntimeWorkerDecommissioning, data.RuntimeWorkerDecommissioningReason, - "Workers are being decommissioned ahead of a scale-down.", corev1.ConditionTrue) - cond.LastTransitionTime = metav1.NewTime(start) - return cond -} - -// clearDecommissioningCondition marks a tracked drain attempt as finished, -// whether because it succeeded or because defaultWorkerDecommissionDeadline -// forced the scale-down to proceed anyway. -func clearDecommissioningCondition(conditions []data.RuntimeCondition) []data.RuntimeCondition { - idx, cond := utils.GetRuntimeCondition(conditions, data.RuntimeWorkerDecommissioning) - if cond == nil { - return conditions - } - cleared := *cond - cleared.Status = corev1.ConditionFalse - cleared.LastTransitionTime = metav1.Now() - conditions[idx] = cleared - return conditions -} diff --git a/pkg/ddc/alluxio/replicas_drain_test.go b/pkg/ddc/alluxio/replicas_drain_test.go deleted file mode 100644 index e18eeb4fbcb..00000000000 --- a/pkg/ddc/alluxio/replicas_drain_test.go +++ /dev/null @@ -1,319 +0,0 @@ -/* -Copyright 2026 The Fluid Authors. - -Licensed under the Apache License, Version 2.0 (the "License"); -you may not use this file except in compliance with the License. -You may obtain a copy of the License at - - http://www.apache.org/licenses/LICENSE-2.0 - -Unless required by applicable law or agreed to in writing, software -distributed under the License is distributed on an "AS IS" BASIS, -WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -See the License for the specific language governing permissions and -limitations under the License. -*/ - -package alluxio - -import ( - "context" - "errors" - "fmt" - "time" - - "github.com/agiledragon/gomonkey/v2" - . "github.com/onsi/ginkgo/v2" - . "github.com/onsi/gomega" - - "github.com/fluid-cloudnative/fluid/api/v1alpha1" - "github.com/fluid-cloudnative/fluid/pkg/ddc/alluxio/operations" - "github.com/fluid-cloudnative/fluid/pkg/features" - cruntime "github.com/fluid-cloudnative/fluid/pkg/runtime" - "github.com/fluid-cloudnative/fluid/pkg/utils" - "github.com/fluid-cloudnative/fluid/pkg/utils/fake" - utilfeature "github.com/fluid-cloudnative/fluid/pkg/utils/feature" - appsv1 "k8s.io/api/apps/v1" - corev1 "k8s.io/api/core/v1" - metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "k8s.io/apimachinery/pkg/runtime" - "k8s.io/apimachinery/pkg/types" - "k8s.io/client-go/tools/record" - "k8s.io/utils/ptr" -) - -const testDrainWorkerSts = "drain-worker" -const testDrainNamespace = "fluid" - -var _ = Describe("AlluxioEngine drainScalingDownWorkers", Label("pkg.ddc.alluxio.replicas_drain_test.go"), func() { - var ( - engine *AlluxioEngine - rt *v1alpha1.AlluxioRuntime - ) - - BeforeEach(func() { - rt = &v1alpha1.AlluxioRuntime{ - ObjectMeta: metav1.ObjectMeta{ - Name: testDrainWorkerSts, - Namespace: testDrainNamespace, - }, - } - }) - - newEngineWithPods := func(pods ...*corev1.Pod) *AlluxioEngine { - objs := []runtime.Object{} - for _, p := range pods { - objs = append(objs, p.DeepCopy()) - } - fakeClient := fake.NewFakeClientWithScheme(testScheme, objs...) - return newAlluxioEngineREP(fakeClient, testDrainWorkerSts, testDrainNamespace) - } - - // hostIP mirrors status.hostIP, which is what ALLUXIO_WORKER_HOSTNAME (and - // therefore the worker's registered identity with the master) is sourced - // from in charts/alluxio - not the pod's own IP. - workerPod := func(ordinal int, hostIP string) *corev1.Pod { - return &corev1.Pod{ - ObjectMeta: metav1.ObjectMeta{ - Name: fmt.Sprintf("%s-worker-%d", testDrainWorkerSts, ordinal), - Namespace: testDrainNamespace, - }, - Status: corev1.PodStatus{ - HostIP: hostIP, - }, - } - } - - Context("when the pod targeted for removal is already gone", func() { - It("treats a NotFound pod as already decommissioned", func() { - engine = newEngineWithPods() - drained, err := engine.drainScalingDownWorkers(context.TODO(), rt, 1, 2) - Expect(err).NotTo(HaveOccurred()) - Expect(drained).To(BeTrue()) - }) - }) - - Context("when the pod has not yet been assigned a host IP", func() { - It("returns not drained without error", func() { - engine = newEngineWithPods(workerPod(1, "")) - drained, err := engine.drainScalingDownWorkers(context.TODO(), rt, 1, 2) - Expect(err).NotTo(HaveOccurred()) - Expect(drained).To(BeFalse()) - }) - }) - - Context("when the decommission call fails", func() { - It("propagates the error", func() { - engine = newEngineWithPods(workerPod(1, "10.0.0.1")) - patch := gomonkey.ApplyFunc(operations.AlluxioFileUtils.DecommissionWorkers, - func(_ operations.AlluxioFileUtils, _ []string) error { - return errors.New("decommission failed") - }) - defer patch.Reset() - - drained, err := engine.drainScalingDownWorkers(context.TODO(), rt, 1, 2) - Expect(err).To(HaveOccurred()) - Expect(drained).To(BeFalse()) - }) - }) - - Context("when active workers are still above the desired count", func() { - It("returns not drained and requests a retry", func() { - engine = newEngineWithPods(workerPod(1, "10.0.0.1")) - patch1 := gomonkey.ApplyFunc(operations.AlluxioFileUtils.DecommissionWorkers, - func(_ operations.AlluxioFileUtils, _ []string) error { - return nil - }) - defer patch1.Reset() - patch2 := gomonkey.ApplyFunc(operations.AlluxioFileUtils.CountActiveWorkers, - func(_ operations.AlluxioFileUtils) (int, error) { - return 2, nil - }) - defer patch2.Reset() - - drained, err := engine.drainScalingDownWorkers(context.TODO(), rt, 1, 2) - Expect(err).NotTo(HaveOccurred()) - Expect(drained).To(BeFalse()) - }) - }) - - Context("when the worker has successfully drained", func() { - It("returns drained with no error", func() { - engine = newEngineWithPods(workerPod(1, "10.0.0.1")) - patch1 := gomonkey.ApplyFunc(operations.AlluxioFileUtils.DecommissionWorkers, - func(_ operations.AlluxioFileUtils, _ []string) error { - return nil - }) - defer patch1.Reset() - patch2 := gomonkey.ApplyFunc(operations.AlluxioFileUtils.CountActiveWorkers, - func(_ operations.AlluxioFileUtils) (int, error) { - return 1, nil - }) - defer patch2.Reset() - - drained, err := engine.drainScalingDownWorkers(context.TODO(), rt, 1, 2) - Expect(err).NotTo(HaveOccurred()) - Expect(drained).To(BeTrue()) - }) - }) - - Context("when multiple targeted pods share the same node", func() { - It("deduplicates the decommission address list", func() { - engine = newEngineWithPods(workerPod(1, "10.0.0.1"), workerPod(2, "10.0.0.1")) - - var capturedAddrs []string - patch1 := gomonkey.ApplyFunc(operations.AlluxioFileUtils.DecommissionWorkers, - func(_ operations.AlluxioFileUtils, addrs []string) error { - capturedAddrs = append([]string(nil), addrs...) - return nil - }) - defer patch1.Reset() - patch2 := gomonkey.ApplyFunc(operations.AlluxioFileUtils.CountActiveWorkers, - func(_ operations.AlluxioFileUtils) (int, error) { - return 1, nil - }) - defer patch2.Reset() - - drained, err := engine.drainScalingDownWorkers(context.TODO(), rt, 1, 3) - Expect(err).NotTo(HaveOccurred()) - Expect(drained).To(BeTrue()) - Expect(capturedAddrs).To(HaveLen(1)) - }) - }) -}) - -var _ = Describe("AlluxioEngine getWorkerRPCPort", Label("pkg.ddc.alluxio.replicas_drain_test.go"), func() { - var engine *AlluxioEngine - - BeforeEach(func() { - engine = newAlluxioEngineREP(fake.NewFakeClientWithScheme(testScheme), testDrainWorkerSts, testDrainNamespace) - }) - - It("returns the configured rpc port when set", func() { - rt := &v1alpha1.AlluxioRuntime{ - Spec: v1alpha1.AlluxioRuntimeSpec{ - Worker: v1alpha1.AlluxioCompTemplateSpec{ - Ports: map[string]int{"rpc": 12345}, - }, - }, - } - Expect(engine.getWorkerRPCPort(rt)).To(Equal(12345)) - }) - - It("falls back to the default port when unset", func() { - rt := &v1alpha1.AlluxioRuntime{} - Expect(engine.getWorkerRPCPort(rt)).To(Equal(defaultWorkerRPCPort)) - }) - - It("falls back to the default port when the configured value is not positive", func() { - rt := &v1alpha1.AlluxioRuntime{ - Spec: v1alpha1.AlluxioRuntimeSpec{ - Worker: v1alpha1.AlluxioCompTemplateSpec{ - Ports: map[string]int{"rpc": 0}, - }, - }, - } - Expect(engine.getWorkerRPCPort(rt)).To(Equal(defaultWorkerRPCPort)) - }) -}) - -var _ = Describe("AlluxioEngine SyncReplicas worker decommission deadline", Label("pkg.ddc.alluxio.replicas_drain_test.go"), func() { - const ( - deadlineTestRuntime = "deadline-worker" - deadlineTestNs = "fluid" - ) - - newFixtures := func(existingCond *v1alpha1.RuntimeCondition) *AlluxioEngine { - rt := &v1alpha1.AlluxioRuntime{ - ObjectMeta: metav1.ObjectMeta{Name: deadlineTestRuntime, Namespace: deadlineTestNs}, - Spec: v1alpha1.AlluxioRuntimeSpec{Replicas: 1}, - Status: v1alpha1.RuntimeStatus{DesiredWorkerNumberScheduled: 2}, - } - if existingCond != nil { - rt.Status.Conditions = []v1alpha1.RuntimeCondition{*existingCond} - } - sts := &appsv1.StatefulSet{ - ObjectMeta: metav1.ObjectMeta{Name: deadlineTestRuntime + "-worker", Namespace: deadlineTestNs}, - Spec: appsv1.StatefulSetSpec{Replicas: ptr.To[int32](2)}, - Status: appsv1.StatefulSetStatus{Replicas: 2}, - } - pod := &corev1.Pod{ - ObjectMeta: metav1.ObjectMeta{Name: deadlineTestRuntime + "-worker-1", Namespace: deadlineTestNs}, - Status: corev1.PodStatus{HostIP: "10.0.0.5"}, - } - // BuildWorkersAffinity (invoked when Helper.SyncReplicas updates the - // StatefulSet's replica count) requires the Dataset to exist. - dataset := &v1alpha1.Dataset{ - ObjectMeta: metav1.ObjectMeta{Name: deadlineTestRuntime, Namespace: deadlineTestNs}, - } - fakeClient := fake.NewFakeClientWithScheme(testScheme, rt, sts, pod, dataset) - return newAlluxioEngineREP(fakeClient, deadlineTestRuntime, deadlineTestNs) - } - - getCondition := func(engine *AlluxioEngine) *v1alpha1.RuntimeCondition { - rt, err := engine.getRuntime() - Expect(err).NotTo(HaveOccurred()) - _, cond := utils.GetRuntimeCondition(rt.Status.Conditions, v1alpha1.RuntimeWorkerDecommissioning) - return cond - } - - BeforeEach(func() { - Expect(utilfeature.DefaultMutableFeatureGate.Set(string(features.GracefulWorkerScaleDown) + "=true")).To(Succeed()) - }) - - AfterEach(func() { - Expect(utilfeature.DefaultMutableFeatureGate.Set(string(features.GracefulWorkerScaleDown) + "=false")).To(Succeed()) - }) - - Context("when a drain doesn't finish within one reconcile", func() { - It("records when the decommission attempt started and keeps requeuing", func() { - engine := newFixtures(nil) - patch1 := gomonkey.ApplyFunc(operations.AlluxioFileUtils.DecommissionWorkers, - func(_ operations.AlluxioFileUtils, _ []string) error { return nil }) - defer patch1.Reset() - patch2 := gomonkey.ApplyFunc(operations.AlluxioFileUtils.CountActiveWorkers, - func(_ operations.AlluxioFileUtils) (int, error) { return 2, nil }) - defer patch2.Reset() - - err := engine.SyncReplicas(cruntime.ReconcileRequestContext{ - Log: fake.NullLogger(), Recorder: record.NewFakeRecorder(300), - }) - Expect(errors.Is(err, errWorkersNotYetDrained)).To(BeTrue()) - - cond := getCondition(engine) - Expect(cond).NotTo(BeNil()) - Expect(cond.Status).To(Equal(corev1.ConditionTrue)) - Expect(time.Since(cond.LastTransitionTime.Time)).To(BeNumerically("<", time.Minute)) - }) - }) - - Context("when a drain is still stuck past the deadline", func() { - It("forces the scale-down to proceed and clears the marker", func() { - staleCond := utils.NewRuntimeCondition(v1alpha1.RuntimeWorkerDecommissioning, - v1alpha1.RuntimeWorkerDecommissioningReason, "started earlier", corev1.ConditionTrue) - staleCond.LastTransitionTime = metav1.NewTime(time.Now().Add(-defaultWorkerDecommissionDeadline - time.Minute)) - engine := newFixtures(&staleCond) - - patch1 := gomonkey.ApplyFunc(operations.AlluxioFileUtils.DecommissionWorkers, - func(_ operations.AlluxioFileUtils, _ []string) error { return nil }) - defer patch1.Reset() - patch2 := gomonkey.ApplyFunc(operations.AlluxioFileUtils.CountActiveWorkers, - func(_ operations.AlluxioFileUtils) (int, error) { return 2, nil }) - defer patch2.Reset() - - err := engine.SyncReplicas(cruntime.ReconcileRequestContext{ - Log: fake.NullLogger(), Recorder: record.NewFakeRecorder(300), - }) - Expect(err).NotTo(HaveOccurred()) - - cond := getCondition(engine) - Expect(cond).NotTo(BeNil()) - Expect(cond.Status).To(Equal(corev1.ConditionFalse)) - - var sts appsv1.StatefulSet - Expect(engine.Client.Get(context.TODO(), - types.NamespacedName{Name: deadlineTestRuntime + "-worker", Namespace: deadlineTestNs}, &sts)).To(Succeed()) - Expect(*sts.Spec.Replicas).To(Equal(int32(1))) - }) - }) -}) diff --git a/pkg/features/features.go b/pkg/features/features.go deleted file mode 100644 index 329b938d709..00000000000 --- a/pkg/features/features.go +++ /dev/null @@ -1,46 +0,0 @@ -/* -Copyright 2026 The Fluid Authors. - -Licensed under the Apache License, Version 2.0 (the "License"); -you may not use this file except in compliance with the License. -You may obtain a copy of the License at - - http://www.apache.org/licenses/LICENSE-2.0 - -Unless required by applicable law or agreed to in writing, software -distributed under the License is distributed on an "AS IS" BASIS, -WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -See the License for the specific language governing permissions and -limitations under the License. -*/ - -package features - -import ( - utilfeature "github.com/fluid-cloudnative/fluid/pkg/utils/feature" - "k8s.io/apimachinery/pkg/util/runtime" - "k8s.io/component-base/featuregate" -) - -const ( - // GracefulWorkerScaleDown gates graceful worker scale-down for AlluxioRuntime - // on a standard Kubernetes StatefulSet. - // - // When enabled, workers targeted for removal are decommissioned from the - // Alluxio cluster before the pod is terminated, giving the master time to - // migrate their cached blocks to the surviving workers. Without this gate, - // cached data held on removed workers is lost immediately on scale-in. - // - // This only supports the standard StatefulSet scale-down order (highest - // ordinal first); it does not yet integrate with OpenKruise's Advanced - // StatefulSet for selective deletion of non-highest-ordinal pods. - GracefulWorkerScaleDown featuregate.Feature = "GracefulWorkerScaleDown" -) - -var defaultFeatureGates = map[featuregate.Feature]featuregate.FeatureSpec{ - GracefulWorkerScaleDown: {Default: false, PreRelease: featuregate.Alpha}, -} - -func init() { - runtime.Must(utilfeature.DefaultMutableFeatureGate.Add(defaultFeatureGates)) -} diff --git a/test/gha-e2e/curvine/read_job.yaml b/test/gha-e2e/curvine/read_job.yaml index bb96ee806e1..e7d584f682f 100644 --- a/test/gha-e2e/curvine/read_job.yaml +++ b/test/gha-e2e/curvine/read_job.yaml @@ -24,14 +24,7 @@ spec: command: ['sh'] args: - -c - - | - for i in $(seq 1 12); do - content=$(cat /data/minio/bar 2>/dev/null) && [ -n "$content" ] && exit 0 - echo "Attempt $i: /data/minio/bar not readable yet, retrying in 5s..." - sleep 5 - done - echo "ERROR: /data/minio/bar is not readable after 60 seconds" - exit 1 + - set -ex; test -n "$(cat /data/minio/bar)" volumeMounts: - name: data-vol mountPath: /data