diff --git a/api/v2/weightsandbiases_types.go b/api/v2/weightsandbiases_types.go index 1655793f..b69d1371 100644 --- a/api/v2/weightsandbiases_types.go +++ b/api/v2/weightsandbiases_types.go @@ -541,7 +541,8 @@ type RedisConfig struct { } type RedisSentinelSpec struct { - Enabled bool `json:"enabled"` + // +kubebuilder:default=true + Enabled *bool `json:"enabled,omitempty"` Config RedisSentinelConfig `json:"config,omitempty"` } diff --git a/api/v2/zz_generated.deepcopy.go b/api/v2/zz_generated.deepcopy.go index 76ab543b..5cc03b8b 100644 --- a/api/v2/zz_generated.deepcopy.go +++ b/api/v2/zz_generated.deepcopy.go @@ -1301,6 +1301,11 @@ func (in *RedisSentinelConfig) DeepCopy() *RedisSentinelConfig { // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *RedisSentinelSpec) DeepCopyInto(out *RedisSentinelSpec) { *out = *in + if in.Enabled != nil { + in, out := &in.Enabled, &out.Enabled + *out = new(bool) + **out = **in + } in.Config.DeepCopyInto(&out.Config) } diff --git a/config/crd/bases/apps.wandb.com_weightsandbiases.yaml b/config/crd/bases/apps.wandb.com_weightsandbiases.yaml index 4ee11815..2bfdb666 100644 --- a/config/crd/bases/apps.wandb.com_weightsandbiases.yaml +++ b/config/crd/bases/apps.wandb.com_weightsandbiases.yaml @@ -3864,9 +3864,8 @@ spec: type: object type: object enabled: + default: true type: boolean - required: - - enabled type: object storageSize: type: string diff --git a/config/rbac/role.yaml b/config/rbac/role.yaml index 22dac7bc..cf50f61b 100644 --- a/config/rbac/role.yaml +++ b/config/rbac/role.yaml @@ -396,3 +396,11 @@ rules: - securitycontextconstraints verbs: - use +- apiGroups: + - storage.k8s.io + resources: + - storageclasses + verbs: + - get + - list + - watch diff --git a/deploy/operator/templates/wandb-operator-wandb-role.yaml b/deploy/operator/templates/wandb-operator-wandb-role.yaml index 290ae4e8..bc729666 100644 --- a/deploy/operator/templates/wandb-operator-wandb-role.yaml +++ b/deploy/operator/templates/wandb-operator-wandb-role.yaml @@ -4,6 +4,14 @@ kind: ClusterRole metadata: name: {{ .Release.Name }}-wandb rules: + - apiGroups: + - storage.k8s.io + resources: + - storageclasses + verbs: + - get + - list + - watch - apiGroups: - apps.wandb.com resources: @@ -210,4 +218,4 @@ subjects: - kind: ServiceAccount name: {{ include "wandb-operator.fullname" . }} namespace: {{ .Release.Namespace }} -{{- end }} \ No newline at end of file +{{- end }} diff --git a/internal/controller/common/state.go b/internal/controller/common/state.go index 36119590..03d73138 100644 --- a/internal/controller/common/state.go +++ b/internal/controller/common/state.go @@ -10,5 +10,5 @@ const ( ) var NotReadyStates = []string{ - ErrorState, PendingState, UnavailableState, + ErrorState, PendingState, DegradedState, UnknownState, UnavailableState, } diff --git a/internal/controller/infra/managed/clickhouse/altinity/keeper/read.go b/internal/controller/infra/managed/clickhouse/altinity/keeper/read.go index ed24b015..67a19afb 100644 --- a/internal/controller/infra/managed/clickhouse/altinity/keeper/read.go +++ b/internal/controller/infra/managed/clickhouse/altinity/keeper/read.go @@ -48,7 +48,7 @@ func ReadState( }} } - return computeKeeperReadyCondition(ctx, podsRunning) + return computeKeeperReadyCondition(ctx, expectedKeeperPodCount(actual), podsRunning) } func keeperPodsRunningStatus( @@ -70,7 +70,7 @@ func keeperPodsRunningStatus( return result, nil } -func computeKeeperReadyCondition(ctx context.Context, podsRunning map[string]bool) []metav1.Condition { +func computeKeeperReadyCondition(ctx context.Context, expectedPodCount int, podsRunning map[string]bool) []metav1.Condition { log := logx.GetSlog(ctx) var runningCount, podCount int @@ -80,19 +80,19 @@ func computeKeeperReadyCondition(ctx context.Context, podsRunning map[string]boo runningCount++ } } - log.Info("Keeper pods status", "running", runningCount, "total", podCount) + log.Info("Keeper pods status", "running", runningCount, "reported", podCount, "expected", expectedPodCount) status := metav1.ConditionUnknown reason := common.UnknownReason message := "" switch { - case podCount > 0 && podCount == runningCount: + case expectedPodCount > 0 && podCount == expectedPodCount && podCount == runningCount: status = metav1.ConditionTrue reason = common.ResourceExistsReason - case podCount > 0: + case expectedPodCount > 0 || podCount > 0: status = metav1.ConditionFalse reason = common.NoResourceReason - message = fmt.Sprintf("%d of %d keeper pods running", runningCount, podCount) + message = fmt.Sprintf("%d of %d expected keeper pods running (%d reported)", runningCount, expectedPodCount, podCount) } return []metav1.Condition{{ @@ -102,3 +102,21 @@ func computeKeeperReadyCondition(ctx context.Context, podsRunning map[string]boo Message: message, }} } + +func expectedKeeperPodCount(chk *chkv1.ClickHouseKeeperInstallation) int { + if chk == nil || chk.Spec.Configuration == nil { + return 0 + } + var count int + for _, cluster := range chk.Spec.Configuration.Clusters { + if cluster == nil || cluster.Layout == nil { + continue + } + replicas := cluster.Layout.ReplicasCount + if replicas < 1 { + replicas = 1 + } + count += replicas + } + return count +} diff --git a/internal/controller/infra/managed/clickhouse/altinity/keeper/read_test.go b/internal/controller/infra/managed/clickhouse/altinity/keeper/read_test.go index 5db34fb3..f5a40f54 100644 --- a/internal/controller/infra/managed/clickhouse/altinity/keeper/read_test.go +++ b/internal/controller/infra/managed/clickhouse/altinity/keeper/read_test.go @@ -10,20 +10,26 @@ import ( var _ = Describe("Keeper readiness", func() { It("is ready when all pods are running", func() { - conds := computeKeeperReadyCondition(context.Background(), map[string]bool{"a": true, "b": true, "c": true}) + conds := computeKeeperReadyCondition(context.Background(), 3, map[string]bool{"a": true, "b": true, "c": true}) Expect(conds).To(HaveLen(1)) Expect(conds[0].Type).To(Equal(KeeperReportedReadyType)) Expect(conds[0].Status).To(Equal(metav1.ConditionTrue)) }) It("is not ready when some pods are not running", func() { - conds := computeKeeperReadyCondition(context.Background(), map[string]bool{"a": true, "b": false, "c": true}) + conds := computeKeeperReadyCondition(context.Background(), 3, map[string]bool{"a": true, "b": false, "c": true}) Expect(conds[0].Status).To(Equal(metav1.ConditionFalse)) Expect(conds[0].Message).To(ContainSubstring("2 of 3")) }) - It("is unknown when no pods are reported yet", func() { - conds := computeKeeperReadyCondition(context.Background(), map[string]bool{}) - Expect(conds[0].Status).To(Equal(metav1.ConditionUnknown)) + It("is not ready when no desired pods are reported yet", func() { + conds := computeKeeperReadyCondition(context.Background(), 3, map[string]bool{}) + Expect(conds[0].Status).To(Equal(metav1.ConditionFalse)) + }) + + It("is not ready when fewer pods are reported than desired", func() { + conds := computeKeeperReadyCondition(context.Background(), 3, map[string]bool{"a": true}) + Expect(conds[0].Status).To(Equal(metav1.ConditionFalse)) + Expect(conds[0].Message).To(ContainSubstring("1 of 3 expected")) }) }) diff --git a/internal/controller/infra/managed/clickhouse/altinity/keeper/spec.go b/internal/controller/infra/managed/clickhouse/altinity/keeper/spec.go index 812c757c..a952f185 100644 --- a/internal/controller/infra/managed/clickhouse/altinity/keeper/spec.go +++ b/internal/controller/infra/managed/clickhouse/altinity/keeper/spec.go @@ -42,6 +42,8 @@ func ToKeeperVendorSpec( } labels := common.BuildWandbLabels(wandb, KeeperModuleName) + settings := chiv1.NewSettings() + settings.Set("keeper_server/enable_reconfiguration", chiv1.NewSettingScalar("true")) podSpec := corev1.PodSpec{ SecurityContext: keeperPodSecurityContext(), @@ -70,6 +72,7 @@ func ToKeeperVendorSpec( }, Spec: chkv1.ChkSpec{ Configuration: &chkv1.Configuration{ + Settings: settings, Clusters: []*chkv1.Cluster{ { Name: ClusterName, diff --git a/internal/controller/infra/managed/clickhouse/altinity/keeper/spec_test.go b/internal/controller/infra/managed/clickhouse/altinity/keeper/spec_test.go index 78319ff4..21af18fc 100644 --- a/internal/controller/infra/managed/clickhouse/altinity/keeper/spec_test.go +++ b/internal/controller/infra/managed/clickhouse/altinity/keeper/spec_test.go @@ -41,6 +41,7 @@ var _ = Describe("Keeper vendor spec", func() { Expect(chk.Spec.Configuration.Clusters).To(HaveLen(1)) Expect(chk.Spec.Configuration.Clusters[0].Layout.ReplicasCount).To(Equal(5)) + Expect(chk.Spec.Configuration.Settings.Get("keeper_server/enable_reconfiguration").String()).To(Equal("true")) Expect(chk.Spec.Templates.VolumeClaimTemplates).To(HaveLen(1)) storage := chk.Spec.Templates.VolumeClaimTemplates[0].Spec.Resources.Requests[corev1.ResourceStorage] diff --git a/internal/controller/infra/managed/clickhouse/altinity/keeper/topology.go b/internal/controller/infra/managed/clickhouse/altinity/keeper/topology.go new file mode 100644 index 00000000..f8a1028f --- /dev/null +++ b/internal/controller/infra/managed/clickhouse/altinity/keeper/topology.go @@ -0,0 +1,271 @@ +package keeper + +import ( + "context" + "fmt" + "strconv" + + "github.com/wandb/operator/internal/controller/common" + chkv1 "github.com/wandb/operator/pkg/vendored/altinity-clickhouse/clickhouse-keeper.altinity.com/v1" + batchv1 "k8s.io/api/batch/v1" + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + apimeta "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + "k8s.io/utils/ptr" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +const StableReplicasAnnotation = "operator.wandb.com/keeper-stable-replicas" + +// ReconcileState grows an existing Keeper ensemble one committed Raft member +// at a time. A brand-new ensemble can safely start at its requested size. +func ReconcileState( + ctx context.Context, + cl client.Client, + desired *chkv1.ClickHouseKeeperInstallation, +) []metav1.Condition { + nsName := client.ObjectKeyFromObject(desired) + actual := &chkv1.ClickHouseKeeperInstallation{} + found, err := common.GetResource(ctx, cl, nsName, ResourceTypeName, actual) + if err != nil { + return topologyCondition(metav1.ConditionUnknown, common.ApiErrorReason, "") + } + + target := expectedKeeperPodCount(desired) + if !found { + setStableReplicas(desired, target) + return append( + WriteState(ctx, cl, nsName, desired), + topologyCondition(metav1.ConditionFalse, common.PendingCreateReason, "waiting for the ClickHouse Keeper ensemble")..., + ) + } + + current := expectedKeeperPodCount(actual) + if target < current { + return topologyCondition( + metav1.ConditionFalse, + "ReplicaReductionUnsupported", + fmt.Sprintf("refusing to reduce ClickHouse Keeper replicas from %d to %d without a member removal workflow", current, target), + ) + } + + stable, adopted := stableReplicas(actual) + if !reconfigurationEnabled(actual) || !adopted { + staged := desired.DeepCopy() + setKeeperReplicas(staged, current) + if !keeperOwnedEqual(actual, staged) { + return append( + WriteState(ctx, cl, nsName, staged), + topologyCondition(metav1.ConditionFalse, common.PendingCreateReason, "enabling safe ClickHouse Keeper scaling")..., + ) + } + ready := ReadState(ctx, cl, nsName) + if !apimeta.IsStatusConditionTrue(ready, KeeperReportedReadyType) { + return ready + } + setStableReplicas(staged, current) + return append( + WriteState(ctx, cl, nsName, staged), + topologyCondition(metav1.ConditionFalse, common.PendingCreateReason, "recording the current ClickHouse Keeper topology")..., + ) + } + + if current == stable { + ready := ReadState(ctx, cl, nsName) + if !apimeta.IsStatusConditionTrue(ready, KeeperReportedReadyType) { + return ready + } + setStableReplicas(desired, stable) + if current == target { + if keeperOwnedEqual(actual, desired) { + return ready + } + return append( + WriteState(ctx, cl, nsName, desired), + topologyCondition(metav1.ConditionFalse, common.PendingCreateReason, "waiting for the desired ClickHouse Keeper configuration")..., + ) + } + + staged := desired.DeepCopy() + setKeeperReplicas(staged, current+1) + setStableReplicas(staged, stable) + return append( + WriteState(ctx, cl, nsName, staged), + topologyCondition( + metav1.ConditionFalse, + common.PendingCreateReason, + fmt.Sprintf("adding ClickHouse Keeper replica %d of %d", current+1, target), + )..., + ) + } + + if current != stable+1 { + return topologyCondition( + metav1.ConditionFalse, + "UnexpectedKeeperTopology", + fmt.Sprintf("ClickHouse Keeper has %d configured replicas but %d stable replicas", current, stable), + ) + } + + jobConditions, complete := reconcileMembershipJob(ctx, cl, desired, stable) + if !complete { + return jobConditions + } + ready := ReadState(ctx, cl, nsName) + if !apimeta.IsStatusConditionTrue(ready, KeeperReportedReadyType) { + return ready + } + + staged := desired.DeepCopy() + setKeeperReplicas(staged, current) + setStableReplicas(staged, current) + return append( + WriteState(ctx, cl, nsName, staged), + topologyCondition( + metav1.ConditionFalse, + common.PendingCreateReason, + fmt.Sprintf("recording %d stable ClickHouse Keeper replicas", current), + )..., + ) +} + +func reconcileMembershipJob( + ctx context.Context, + cl client.Client, + desired *chkv1.ClickHouseKeeperInstallation, + replica int, +) ([]metav1.Condition, bool) { + name := common.FitDefaultInfraName(desired.Name, fmt.Sprintf("-add-%d", replica), 63) + job := &batchv1.Job{} + err := cl.Get(ctx, types.NamespacedName{Namespace: desired.Namespace, Name: name}, job) + if err != nil && !apierrors.IsNotFound(err) { + return topologyCondition(metav1.ConditionUnknown, common.ApiErrorReason, err.Error()), false + } + if apierrors.IsNotFound(err) { + job = membershipJob(desired, name, replica) + if err := cl.Create(ctx, job); err != nil { + return topologyCondition(metav1.ConditionUnknown, common.ApiErrorReason, err.Error()), false + } + return topologyCondition( + metav1.ConditionFalse, + common.PendingCreateReason, + fmt.Sprintf("created ClickHouse Keeper membership job %s", name), + ), false + } + + for _, condition := range job.Status.Conditions { + if condition.Type == batchv1.JobFailed && condition.Status == corev1.ConditionTrue { + message := condition.Message + if message == "" { + message = fmt.Sprintf("ClickHouse Keeper membership job %s failed", name) + } + return topologyCondition(metav1.ConditionFalse, "KeeperMembershipFailed", message), false + } + if condition.Type == batchv1.JobComplete && condition.Status == corev1.ConditionTrue { + return nil, true + } + } + return topologyCondition( + metav1.ConditionFalse, + common.PendingCreateReason, + fmt.Sprintf("waiting for ClickHouse Keeper membership job %s", name), + ), false +} + +func membershipJob( + desired *chkv1.ClickHouseKeeperInstallation, + name string, + replica int, +) *batchv1.Job { + podSpec := desired.Spec.Templates.PodTemplates[0].Spec + memberHost := keeperHostServiceName(desired.Name, replica) + clientHost := "keeper-" + desired.Name + member := fmt.Sprintf("server.%d=%s:9444;participant;1", replica, memberHost) + script := fmt.Sprintf(`set -u +member=%q +for attempt in $(seq 1 150); do + config=$(/usr/bin/clickhouse-keeper keeper-client -h %q -p %d -q 'get "/keeper/config"' 2>/dev/null || true) + if printf '%%s\n' "$config" | grep -Fqx "$member"; then + exit 0 + fi + /usr/bin/clickhouse-keeper keeper-client -h %q -p %d -q "reconfig add \"$member\"" >/dev/null 2>&1 || true + sleep 2 +done +exit 1 +`, member, clientHost, KeeperClientPort, clientHost, KeeperClientPort) + + return &batchv1.Job{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Namespace: desired.Namespace, + Labels: desired.Labels, + OwnerReferences: desired.OwnerReferences, + }, + Spec: batchv1.JobSpec{ + BackoffLimit: ptr.To[int32](0), + ActiveDeadlineSeconds: ptr.To[int64](360), + Template: corev1.PodTemplateSpec{ + ObjectMeta: metav1.ObjectMeta{Labels: desired.Labels}, + Spec: corev1.PodSpec{ + Affinity: podSpec.Affinity, + Tolerations: podSpec.Tolerations, + ImagePullSecrets: podSpec.ImagePullSecrets, + RestartPolicy: corev1.RestartPolicyNever, + SecurityContext: podSpec.SecurityContext, + Containers: []corev1.Container{{ + Name: "membership", + Image: podSpec.Containers[0].Image, + ImagePullPolicy: podSpec.Containers[0].ImagePullPolicy, + SecurityContext: podSpec.Containers[0].SecurityContext, + Command: []string{"/bin/sh", "-c"}, + Args: []string{script}, + }}, + }, + }, + }, + } +} + +func topologyCondition(status metav1.ConditionStatus, reason, message string) []metav1.Condition { + return []metav1.Condition{{ + Type: KeeperReportedReadyType, + Status: status, + Reason: reason, + Message: message, + }} +} + +func keeperHostServiceName(installationName string, replica int) string { + return fmt.Sprintf("chk-%s-%s-0-%d", installationName, ClusterName, replica) +} + +func stableReplicas(chk *chkv1.ClickHouseKeeperInstallation) (int, bool) { + value, ok := chk.GetAnnotations()[StableReplicasAnnotation] + if !ok { + return 0, false + } + replicas, err := strconv.Atoi(value) + return replicas, err == nil && replicas > 0 +} + +func setStableReplicas(chk *chkv1.ClickHouseKeeperInstallation, replicas int) { + annotations := chk.GetAnnotations() + if annotations == nil { + annotations = map[string]string{} + } + annotations[StableReplicasAnnotation] = strconv.Itoa(replicas) + chk.SetAnnotations(annotations) +} + +func setKeeperReplicas(chk *chkv1.ClickHouseKeeperInstallation, replicas int) { + chk.Spec.Configuration.Clusters[0].Layout.ReplicasCount = replicas +} + +func reconfigurationEnabled(chk *chkv1.ClickHouseKeeperInstallation) bool { + if chk.Spec.Configuration == nil || chk.Spec.Configuration.Settings == nil { + return false + } + return chk.Spec.Configuration.Settings.Get("keeper_server/enable_reconfiguration").String() == "true" +} diff --git a/internal/controller/infra/managed/clickhouse/altinity/keeper/topology_test.go b/internal/controller/infra/managed/clickhouse/altinity/keeper/topology_test.go new file mode 100644 index 00000000..f4537c1d --- /dev/null +++ b/internal/controller/infra/managed/clickhouse/altinity/keeper/topology_test.go @@ -0,0 +1,171 @@ +package keeper + +import ( + "context" + "fmt" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + "github.com/wandb/operator/internal/controller/common" + chkv1 "github.com/wandb/operator/pkg/vendored/altinity-clickhouse/clickhouse-keeper.altinity.com/v1" + chiv1 "github.com/wandb/operator/pkg/vendored/altinity-clickhouse/clickhouse.altinity.com/v1" + batchv1 "k8s.io/api/batch/v1" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" +) + +var _ = Describe("Keeper topology reconciliation", func() { + It("creates a new ensemble at its requested size", func() { + scheme := topologyScheme() + desired := topologyKeeper(3, 0) + cl := fake.NewClientBuilder().WithScheme(scheme).Build() + + conditions := ReconcileState(context.Background(), cl, desired) + + Expect(conditions).To(ContainElement(HaveField("Message", "waiting for the ClickHouse Keeper ensemble"))) + actual := &chkv1.ClickHouseKeeperInstallation{} + Expect(cl.Get(context.Background(), client.ObjectKeyFromObject(desired), actual)).To(Succeed()) + Expect(expectedKeeperPodCount(actual)).To(Equal(3)) + Expect(actual.Annotations[StableReplicasAnnotation]).To(Equal("3")) + }) + + It("stages only one new replica", func() { + scheme := topologyScheme() + actual := topologyKeeper(1, 1) + actual.Status = &chkv1.Status{Pods: []string{"keeper-0"}} + desired := topologyKeeper(3, 0) + cl := fake.NewClientBuilder().WithScheme(scheme).WithObjects(actual, topologyReadyPod("keeper-0")).Build() + + conditions := ReconcileState(context.Background(), cl, desired) + + Expect(conditions).To(ContainElement(HaveField("Message", "adding ClickHouse Keeper replica 2 of 3"))) + updated := &chkv1.ClickHouseKeeperInstallation{} + Expect(cl.Get(context.Background(), client.ObjectKeyFromObject(actual), updated)).To(Succeed()) + Expect(expectedKeeperPodCount(updated)).To(Equal(2)) + Expect(updated.Annotations[StableReplicasAnnotation]).To(Equal("1")) + }) + + It("enables reconfiguration before scaling an existing ensemble", func() { + scheme := topologyScheme() + actual := topologyKeeper(1, 0) + actual.Spec.Configuration.Settings = nil + desired := topologyKeeper(3, 0) + cl := fake.NewClientBuilder().WithScheme(scheme).WithObjects(actual).Build() + + conditions := ReconcileState(context.Background(), cl, desired) + + Expect(conditions).To(ContainElement(HaveField("Message", "enabling safe ClickHouse Keeper scaling"))) + updated := &chkv1.ClickHouseKeeperInstallation{} + Expect(cl.Get(context.Background(), client.ObjectKeyFromObject(actual), updated)).To(Succeed()) + Expect(expectedKeeperPodCount(updated)).To(Equal(1)) + Expect(reconfigurationEnabled(updated)).To(BeTrue()) + Expect(updated.Annotations).NotTo(HaveKey(StableReplicasAnnotation)) + }) + + It("creates a job that commits the staged member", func() { + scheme := topologyScheme() + actual := topologyKeeper(2, 1) + desired := topologyKeeper(3, 0) + cl := fake.NewClientBuilder().WithScheme(scheme).WithObjects(actual).Build() + + conditions := ReconcileState(context.Background(), cl, desired) + + Expect(conditions).To(ContainElement(HaveField( + "Message", + ContainSubstring("created ClickHouse Keeper membership job"), + ))) + job := &batchv1.Job{} + name := common.FitDefaultInfraName(actual.Name, "-add-1", 63) + Expect(cl.Get(context.Background(), client.ObjectKey{Namespace: actual.Namespace, Name: name}, job)).To(Succeed()) + Expect(job.Spec.Template.Spec.Containers[0].Args[0]).To(ContainSubstring( + "server.1=chk-wandb-clickhouse-chk-default-0-1:9444;participant;1", + )) + }) + + It("records the staged replica after membership and pods are ready", func() { + scheme := topologyScheme() + actual := topologyKeeper(2, 1) + actual.Status = &chkv1.Status{Pods: []string{"keeper-0", "keeper-1"}} + desired := topologyKeeper(3, 0) + job := membershipJob(desired, common.FitDefaultInfraName(actual.Name, "-add-1", 63), 1) + job.Status.Conditions = []batchv1.JobCondition{{ + Type: batchv1.JobComplete, + Status: corev1.ConditionTrue, + }} + cl := fake.NewClientBuilder().WithScheme(scheme).WithObjects( + actual, + job, + topologyReadyPod("keeper-0"), + topologyReadyPod("keeper-1"), + ).Build() + + conditions := ReconcileState(context.Background(), cl, desired) + + Expect(conditions).To(ContainElement(HaveField("Message", "recording 2 stable ClickHouse Keeper replicas"))) + updated := &chkv1.ClickHouseKeeperInstallation{} + Expect(cl.Get(context.Background(), client.ObjectKeyFromObject(actual), updated)).To(Succeed()) + Expect(expectedKeeperPodCount(updated)).To(Equal(2)) + Expect(updated.Annotations[StableReplicasAnnotation]).To(Equal("2")) + }) +}) + +func topologyScheme() *runtime.Scheme { + scheme := runtime.NewScheme() + Expect(corev1.AddToScheme(scheme)).To(Succeed()) + Expect(batchv1.AddToScheme(scheme)).To(Succeed()) + Expect(chiv1.AddToScheme(scheme)).To(Succeed()) + Expect(chkv1.AddToScheme(scheme)).To(Succeed()) + return scheme +} + +func topologyKeeper(replicas, stable int) *chkv1.ClickHouseKeeperInstallation { + settings := chiv1.NewSettings() + settings.Set("keeper_server/enable_reconfiguration", chiv1.NewSettingScalar("true")) + annotations := map[string]string{} + if stable > 0 { + annotations[StableReplicasAnnotation] = fmt.Sprint(stable) + } + return &chkv1.ClickHouseKeeperInstallation{ + ObjectMeta: metav1.ObjectMeta{ + Name: "wandb-clickhouse-chk", + Namespace: "wandb", + Labels: map[string]string{"app": "keeper"}, + Annotations: annotations, + }, + Spec: chkv1.ChkSpec{ + Configuration: &chkv1.Configuration{ + Settings: settings, + Clusters: []*chkv1.Cluster{{ + Name: ClusterName, + Layout: &chkv1.ChkClusterLayout{ReplicasCount: replicas}, + }}, + }, + Templates: &chiv1.Templates{ + PodTemplates: []chiv1.PodTemplate{{ + Spec: corev1.PodSpec{ + Containers: []corev1.Container{{ + Name: keeperContainerName, + Image: KeeperImage, + }}, + }, + }}, + }, + }, + } +} + +func topologyReadyPod(name string) *corev1.Pod { + return &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: "wandb"}, + Status: corev1.PodStatus{ + Phase: corev1.PodRunning, + Conditions: []corev1.PodCondition{{ + Type: corev1.PodReady, + Status: corev1.ConditionTrue, + }}, + }, + } +} diff --git a/internal/controller/infra/managed/clickhouse/altinity/keeper/write.go b/internal/controller/infra/managed/clickhouse/altinity/keeper/write.go index 2185b498..e0fd55aa 100644 --- a/internal/controller/infra/managed/clickhouse/altinity/keeper/write.go +++ b/internal/controller/infra/managed/clickhouse/altinity/keeper/write.go @@ -60,6 +60,14 @@ func applyOwnedKeeper(obj, desired *chkv1.ClickHouseKeeperInstallation) { labels[k] = v } obj.SetLabels(labels) + annotations := obj.GetAnnotations() + if annotations == nil { + annotations = map[string]string{} + } + if stableReplicas, ok := desired.GetAnnotations()[StableReplicasAnnotation]; ok { + annotations[StableReplicasAnnotation] = stableReplicas + } + obj.SetAnnotations(annotations) obj.SetOwnerReferences(desired.GetOwnerReferences()) obj.Spec = desired.Spec } @@ -67,5 +75,6 @@ func applyOwnedKeeper(obj, desired *chkv1.ClickHouseKeeperInstallation) { func keeperOwnedEqual(a, b *chkv1.ClickHouseKeeperInstallation) bool { return common.JSONEqual(a.Spec, b.Spec) && common.JSONEqual(a.Labels, b.Labels) && + a.GetAnnotations()[StableReplicasAnnotation] == b.GetAnnotations()[StableReplicasAnnotation] && common.JSONEqual(a.OwnerReferences, b.OwnerReferences) } diff --git a/internal/controller/infra/managed/clickhouse/altinity/read.go b/internal/controller/infra/managed/clickhouse/altinity/read.go index c5c3ac42..c2ca148a 100644 --- a/internal/controller/infra/managed/clickhouse/altinity/read.go +++ b/internal/controller/infra/managed/clickhouse/altinity/read.go @@ -187,22 +187,23 @@ func computeClickHouseReportedReadyCondition( runningCount++ } } + expectedPodCount := expectedClickHousePodCount(chi) log.Info( - "Clickhouse pods status", "running", runningCount, "total", podCount, + "Clickhouse pods status", "running", runningCount, "reported", podCount, "expected", expectedPodCount, ) status := metav1.ConditionUnknown reason := ctrlcommon.UnknownReason message := "" - if podCount > 0 && podCount == runningCount { + if expectedPodCount > 0 && podCount == expectedPodCount && podCount == runningCount { status = metav1.ConditionTrue reason = ctrlcommon.ResourceExistsReason - } else if podCount > 0 { + } else if expectedPodCount > 0 || podCount > 0 { status = metav1.ConditionFalse reason = ctrlcommon.NoResourceReason - message = fmt.Sprintf("%d of %d pods running", runningCount, podCount) + message = fmt.Sprintf("%d of %d expected pods running (%d reported)", runningCount, expectedPodCount, podCount) } return []metav1.Condition{ @@ -214,3 +215,25 @@ func computeClickHouseReportedReadyCondition( }, } } + +func expectedClickHousePodCount(chi *chiv1.ClickHouseInstallation) int { + if chi == nil || chi.Spec.Configuration == nil { + return 0 + } + var count int + for _, cluster := range chi.Spec.Configuration.Clusters { + if cluster == nil || cluster.Layout == nil { + continue + } + shards := cluster.Layout.ShardsCount + if shards < 1 { + shards = 1 + } + replicas := cluster.Layout.ReplicasCount + if replicas < 1 { + replicas = 1 + } + count += shards * replicas + } + return count +} diff --git a/internal/controller/infra/managed/clickhouse/altinity/status_test.go b/internal/controller/infra/managed/clickhouse/altinity/status_test.go index c3e8e11e..c5e46710 100644 --- a/internal/controller/infra/managed/clickhouse/altinity/status_test.go +++ b/internal/controller/infra/managed/clickhouse/altinity/status_test.go @@ -7,6 +7,7 @@ import ( . "github.com/onsi/gomega" "github.com/wandb/operator/internal/controller/common" "github.com/wandb/operator/internal/controller/infra/managed/clickhouse/altinity/keeper" + chiv1 "github.com/wandb/operator/pkg/vendored/altinity-clickhouse/clickhouse.altinity.com/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) @@ -55,4 +56,45 @@ var _ = Describe("ClickHouse status keeper gating", func() { Expect(events[0].Reason).To(Equal("ClickHouseInvalidName")) Expect(events[0].Message).To(ContainSubstring("too long")) }) + + It("does not report a degraded ClickHouse installation as ready", func() { + conditions := append(healthyClickHouse(), metav1.Condition{ + Type: keeper.KeeperReportedReadyType, + Status: metav1.ConditionTrue, + Reason: common.ResourceExistsReason, + }) + for i := range conditions { + if conditions[i].Type == ClickHouseReportedReadyType { + conditions[i].Status = metav1.ConditionFalse + conditions[i].Reason = common.NoResourceReason + } + } + + status, _, _ := ComputeStatus(context.Background(), true, nil, conditions, nil, 1) + + Expect(status.State).To(Equal(common.DegradedState)) + Expect(status.Ready).To(BeFalse()) + }) + + It("requires every desired ClickHouse pod to be reported and ready", func() { + chi := &chiv1.ClickHouseInstallation{ + Spec: chiv1.ChiSpec{ + Configuration: &chiv1.Configuration{ + Clusters: []*chiv1.Cluster{{ + Layout: &chiv1.ChiClusterLayout{ShardsCount: 1, ReplicasCount: 3}, + }}, + }, + }, + } + + conditions := computeClickHouseReportedReadyCondition( + context.Background(), + chi, + map[string]bool{"clickhouse-0": true}, + ) + + Expect(conditions).To(HaveLen(1)) + Expect(conditions[0].Status).To(Equal(metav1.ConditionFalse)) + Expect(conditions[0].Message).To(ContainSubstring("1 of 3 expected pods")) + }) }) diff --git a/internal/controller/infra/managed/clickhouse/altinity/write.go b/internal/controller/infra/managed/clickhouse/altinity/write.go index e5456bba..61aa1f23 100644 --- a/internal/controller/infra/managed/clickhouse/altinity/write.go +++ b/internal/controller/infra/managed/clickhouse/altinity/write.go @@ -9,6 +9,7 @@ import ( chkv1 "github.com/wandb/operator/pkg/vendored/altinity-clickhouse/clickhouse-keeper.altinity.com/v1" chiv1 "github.com/wandb/operator/pkg/vendored/altinity-clickhouse/clickhouse.altinity.com/v1" corev1 "k8s.io/api/core/v1" + apimeta "k8s.io/apimachinery/pkg/api/meta" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/types" "sigs.k8s.io/controller-runtime/pkg/client" @@ -40,11 +41,57 @@ func WriteState( return results } } - results = append(results, keeper.WriteState( - ctx, client, - types.NamespacedName{Namespace: desiredKeeper.Namespace, Name: desiredKeeper.Name}, - desiredKeeper, - )...) + keeperConditions := keeper.ReconcileState(ctx, client, desiredKeeper) + results = append(results, keeperConditions...) + if !apimeta.IsStatusConditionTrue(keeperConditions, keeper.KeeperReportedReadyType) { + results = append(results, metav1.Condition{ + Type: ClickHouseCustomResourceType, + Status: metav1.ConditionFalse, + Reason: common.PendingCreateReason, + Message: "waiting for the desired ClickHouse Keeper replicas", + }) + return results + } + + installationNsName := createNsNameBuilder(specNamespacedName).InstallationNsName() + actualClickHouse := &chiv1.ClickHouseInstallation{} + clickHouseFound, err := common.GetResource( + ctx, + client, + installationNsName, + ResourceTypeName, + actualClickHouse, + ) + if err != nil { + results = append(results, metav1.Condition{ + Type: ClickHouseCustomResourceType, + Status: metav1.ConditionUnknown, + Reason: common.ApiErrorReason, + }) + return results + } + if clickHouseFound && !common.JSONEqual(actualClickHouse.Spec, desired.Spec) { + podsRunning, err := chPodsRunningStatus(ctx, client, installationNsName.Namespace, actualClickHouse) + if err != nil { + results = append(results, metav1.Condition{ + Type: ClickHouseReportedReadyType, + Status: metav1.ConditionUnknown, + Reason: common.ApiErrorReason, + }) + return results + } + currentConditions := computeClickHouseReportedReadyCondition(ctx, actualClickHouse, podsRunning) + results = append(results, currentConditions...) + if !apimeta.IsStatusConditionTrue(currentConditions, ClickHouseReportedReadyType) { + results = append(results, metav1.Condition{ + Type: ClickHouseCustomResourceType, + Status: metav1.ConditionFalse, + Reason: common.PendingCreateReason, + Message: "waiting for the current ClickHouse topology before applying the desired topology", + }) + return results + } + } results = append(results, writeClickHouseInstallation(ctx, client, specNamespacedName, desired)...) return results diff --git a/internal/controller/infra/managed/clickhouse/altinity/write_test.go b/internal/controller/infra/managed/clickhouse/altinity/write_test.go new file mode 100644 index 00000000..0407318b --- /dev/null +++ b/internal/controller/infra/managed/clickhouse/altinity/write_test.go @@ -0,0 +1,145 @@ +package altinity + +import ( + "context" + "fmt" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + "github.com/wandb/operator/internal/controller/infra/managed/clickhouse/altinity/keeper" + chkv1 "github.com/wandb/operator/pkg/vendored/altinity-clickhouse/clickhouse-keeper.altinity.com/v1" + chiv1 "github.com/wandb/operator/pkg/vendored/altinity-clickhouse/clickhouse.altinity.com/v1" + batchv1 "k8s.io/api/batch/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" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" +) + +var _ = Describe("ClickHouse staged writes", func() { + keeperWithReplicas := func(replicas int) *chkv1.ClickHouseKeeperInstallation { + settings := chiv1.NewSettings() + settings.Set("keeper_server/enable_reconfiguration", chiv1.NewSettingScalar("true")) + return &chkv1.ClickHouseKeeperInstallation{ + ObjectMeta: metav1.ObjectMeta{ + Name: "wandb-clickhouse-chk", + Namespace: "wandb", + Annotations: map[string]string{ + keeper.StableReplicasAnnotation: fmt.Sprint(replicas), + }, + }, + Spec: chkv1.ChkSpec{ + Configuration: &chkv1.Configuration{ + Settings: settings, + Clusters: []*chkv1.Cluster{{ + Layout: &chkv1.ChkClusterLayout{ReplicasCount: replicas}, + }}, + }, + }, + } + } + clickHouseWithReplicas := func(replicas int) *chiv1.ClickHouseInstallation { + return &chiv1.ClickHouseInstallation{ + ObjectMeta: metav1.ObjectMeta{Name: "wandb-clickhouse", Namespace: "wandb"}, + Spec: chiv1.ChiSpec{ + Configuration: &chiv1.Configuration{ + Clusters: []*chiv1.Cluster{{ + Layout: &chiv1.ChiClusterLayout{ShardsCount: 1, ReplicasCount: replicas}, + }}, + }, + }, + } + } + readyPod := func(name string, ready bool) *corev1.Pod { + status := corev1.ConditionFalse + if ready { + status = corev1.ConditionTrue + } + return &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: "wandb"}, + Status: corev1.PodStatus{ + Phase: corev1.PodRunning, + Conditions: []corev1.PodCondition{{ + Type: corev1.PodReady, + Status: status, + }}, + }, + } + } + + It("does not update ClickHouse in the reconcile that changes Keeper topology", func() { + scheme := runtime.NewScheme() + Expect(corev1.AddToScheme(scheme)).To(Succeed()) + Expect(batchv1.AddToScheme(scheme)).To(Succeed()) + Expect(chkv1.AddToScheme(scheme)).To(Succeed()) + Expect(chiv1.AddToScheme(scheme)).To(Succeed()) + actualKeeper := keeperWithReplicas(1) + actualKeeper.Status = &chkv1.Status{Pods: []string{"keeper-0"}} + desiredKeeper := keeperWithReplicas(3) + desiredKeeper.Annotations = nil + desiredClickHouse := clickHouseWithReplicas(3) + cl := fake.NewClientBuilder().WithScheme(scheme).WithObjects( + actualKeeper, + readyPod("keeper-0", true), + ).Build() + + conditions := WriteState( + context.Background(), + cl, + types.NamespacedName{Name: "wandb-clickhouse", Namespace: "wandb"}, + nil, + desiredKeeper, + desiredClickHouse, + ) + + Expect(conditions).To(ContainElement(HaveField("Message", "adding ClickHouse Keeper replica 2 of 3"))) + updatedKeeper := &chkv1.ClickHouseKeeperInstallation{} + Expect(cl.Get(context.Background(), client.ObjectKeyFromObject(desiredKeeper), updatedKeeper)).To(Succeed()) + Expect(updatedKeeper.Spec.Configuration.Clusters[0].Layout.ReplicasCount).To(Equal(2)) + Expect(updatedKeeper.Annotations[keeper.StableReplicasAnnotation]).To(Equal("1")) + clickHouse := &chiv1.ClickHouseInstallation{} + err := cl.Get(context.Background(), client.ObjectKeyFromObject(desiredClickHouse), clickHouse) + Expect(err).To(HaveOccurred()) + }) + + It("does not scale ClickHouse while its current topology is degraded", func() { + scheme := runtime.NewScheme() + Expect(corev1.AddToScheme(scheme)).To(Succeed()) + Expect(batchv1.AddToScheme(scheme)).To(Succeed()) + Expect(chkv1.AddToScheme(scheme)).To(Succeed()) + Expect(chiv1.AddToScheme(scheme)).To(Succeed()) + + actualKeeper := keeperWithReplicas(3) + actualKeeper.Status = &chkv1.Status{Pods: []string{"keeper-0", "keeper-1", "keeper-2"}} + actualClickHouse := clickHouseWithReplicas(1) + actualClickHouse.Status = &chiv1.Status{Pods: []string{"clickhouse-0"}} + desiredClickHouse := clickHouseWithReplicas(3) + cl := fake.NewClientBuilder().WithScheme(scheme).WithObjects( + actualKeeper, + actualClickHouse, + readyPod("keeper-0", true), + readyPod("keeper-1", true), + readyPod("keeper-2", true), + readyPod("clickhouse-0", false), + ).Build() + + conditions := WriteState( + context.Background(), + cl, + types.NamespacedName{Name: "wandb-clickhouse", Namespace: "wandb"}, + nil, + keeperWithReplicas(3), + desiredClickHouse, + ) + + Expect(conditions).To(ContainElement(HaveField( + "Message", + "waiting for the current ClickHouse topology before applying the desired topology", + ))) + updatedClickHouse := &chiv1.ClickHouseInstallation{} + Expect(cl.Get(context.Background(), client.ObjectKeyFromObject(actualClickHouse), updatedClickHouse)).To(Succeed()) + Expect(updatedClickHouse.Spec.Configuration.Clusters[0].Layout.ReplicasCount).To(Equal(1)) + }) +}) diff --git a/internal/controller/infra/managed/objectstore/seaweedfs/status.go b/internal/controller/infra/managed/objectstore/seaweedfs/status.go index 7053b7b3..b92f2937 100644 --- a/internal/controller/infra/managed/objectstore/seaweedfs/status.go +++ b/internal/controller/infra/managed/objectstore/seaweedfs/status.go @@ -2,6 +2,7 @@ package seaweedfs import ( "context" + "strings" "time" "github.com/samber/lo" @@ -86,6 +87,13 @@ func applyDefaultConditions(conditions []metav1.Condition) []metav1.Condition { Reason: common.NoResourceReason, }) } + if !common.ContainsType(conditions, SeaweedTopologyReadyType) { + conditions = append(conditions, metav1.Condition{ + Type: SeaweedTopologyReadyType, + Status: metav1.ConditionUnknown, + Reason: common.NoResourceReason, + }) + } return conditions } @@ -106,6 +114,7 @@ func inferInfraState( impliedStates = inferStateFromCondition(ctx, SeaweedReportedReadyType, impliedStates, conditions) impliedStates = inferStateFromCondition(ctx, SeaweedWritableType, impliedStates, conditions) impliedStates = inferStateFromCondition(ctx, SeaweedS3ReachableType, impliedStates, conditions) + impliedStates = inferStateFromCondition(ctx, SeaweedTopologyReadyType, impliedStates, conditions) hasImpliedState := func(target string) bool { return len(lo.FilterValues( @@ -161,6 +170,8 @@ func inferStateFromCondition(ctx context.Context, conditionType string, impliedS impliedStates[conditionType] = inferState_SeaweedReportedReadyType(ctx, cond) case SeaweedWritableType, SeaweedS3ReachableType: impliedStates[conditionType] = inferState_SeaweedWritableType(ctx, cond) + case SeaweedTopologyReadyType: + impliedStates[conditionType] = inferState_SeaweedTopologyReadyType(ctx, cond) default: impliedStates[conditionType] = common.UnknownState } @@ -168,6 +179,25 @@ func inferStateFromCondition(ctx context.Context, conditionType string, impliedS return impliedStates } +func inferState_SeaweedTopologyReadyType(ctx context.Context, condition metav1.Condition) string { + log := logx.GetSlog(ctx) + result := common.PendingState + if condition.Status == metav1.ConditionTrue { + result = common.HealthyState + } + if condition.Status == metav1.ConditionFalse && + (strings.Contains(condition.Reason, "Failed") || + strings.Contains(condition.Reason, "Unsupported") || + strings.Contains(condition.Reason, "Unexpected")) { + result = common.ErrorState + } + log.Debug( + "implied state", "state", result, "condition", condition.Type, + "reason", condition.Reason, "status", condition.Status, + ) + return result +} + func inferState_SeaweedWritableType(ctx context.Context, condition metav1.Condition) string { log := logx.GetSlog(ctx) result := common.PendingState diff --git a/internal/controller/infra/managed/objectstore/seaweedfs/status_test.go b/internal/controller/infra/managed/objectstore/seaweedfs/status_test.go index cc6aeca4..7261ca94 100644 --- a/internal/controller/infra/managed/objectstore/seaweedfs/status_test.go +++ b/internal/controller/infra/managed/objectstore/seaweedfs/status_test.go @@ -58,6 +58,7 @@ var _ = Describe("SeaweedFS status", func() { {Type: SeaweedReportedReadyType, Status: metav1.ConditionTrue}, {Type: SeaweedWritableType, Status: metav1.ConditionTrue, Reason: "AllocationSucceeded"}, {Type: SeaweedS3ReachableType, Status: metav1.ConditionTrue, Reason: "EndpointReachable"}, + {Type: SeaweedTopologyReadyType, Status: metav1.ConditionTrue, Reason: "TopologyReady"}, }, nil, 1, @@ -66,4 +67,25 @@ var _ = Describe("SeaweedFS status", func() { Expect(status.Ready).To(BeTrue()) Expect(status.State).To(Equal(common.HealthyState)) }) + + It("reports a failed topology migration as an error", func() { + status, _, _ := ComputeStatus( + context.Background(), + true, + nil, + []metav1.Condition{ + {Type: SeaweedCustomResourceType, Status: metav1.ConditionTrue}, + {Type: SeaweedConnectionInfoType, Status: metav1.ConditionTrue}, + {Type: SeaweedReportedReadyType, Status: metav1.ConditionTrue}, + {Type: SeaweedWritableType, Status: metav1.ConditionTrue, Reason: "AllocationSucceeded"}, + {Type: SeaweedS3ReachableType, Status: metav1.ConditionTrue, Reason: "EndpointReachable"}, + {Type: SeaweedTopologyReadyType, Status: metav1.ConditionFalse, Reason: "ReplicationFailed"}, + }, + nil, + 1, + ) + + Expect(status.Ready).To(BeFalse()) + Expect(status.State).To(Equal(common.ErrorState)) + }) }) diff --git a/internal/controller/infra/managed/objectstore/seaweedfs/topology.go b/internal/controller/infra/managed/objectstore/seaweedfs/topology.go new file mode 100644 index 00000000..423736fd --- /dev/null +++ b/internal/controller/infra/managed/objectstore/seaweedfs/topology.go @@ -0,0 +1,561 @@ +package seaweedfs + +import ( + "context" + "fmt" + "strings" + + seaweedv1 "github.com/wandb/operator/pkg/vendored/seaweedfs-operator/seaweed.seaweedfs.com/v1" + appsv1 "k8s.io/api/apps/v1" + batchv1 "k8s.io/api/batch/v1" + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + "k8s.io/utils/ptr" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" +) + +const ( + SeaweedTopologyReadyType = "SeaweedTopologyReady" + + stableReplicationAnnotation = "operator.wandb.com/seaweed-stable-replication" + s3RolloutAnnotation = "operator.wandb.com/seaweed-s3-rollout-replication" + + topologyJobPreVerify = "pre-verify" + topologyJobReplicate = "replicate" + topologyJobPostVerify = "post-verify" +) + +type topologyJobState int + +type statefulSetTarget struct { + suffix string + replicas int32 +} + +const ( + topologyJobPending topologyJobState = iota + topologyJobSucceeded + topologyJobFailed +) + +func reconcileTopology( + ctx context.Context, + kubeClient client.Client, + desired, actual *seaweedv1.Seaweed, + owner client.Object, +) ([]metav1.Condition, error) { + targetReplication := defaultReplication(desired) + if actual == nil { + setStableReplication(desired, targetReplication) + return topologyResult( + metav1.ConditionTrue, + "TopologyInitialized", + fmt.Sprintf("SeaweedFS topology initialized with %d volume servers and replication %s", desired.Spec.Volume.Replicas, targetReplication), + ), nil + } + + stableReplication := actual.Annotations[stableReplicationAnnotation] + if stableReplication == "" { + return adoptTopology(ctx, kubeClient, desired, actual) + } + + setStableReplication(desired, stableReplication) + return migrateTopology(ctx, kubeClient, desired, actual, owner, stableReplication, targetReplication) +} + +func adoptTopology( + ctx context.Context, + kubeClient client.Client, + desired, actual *seaweedv1.Seaweed, +) ([]metav1.Condition, error) { + currentReplicas := volumeReplicas(actual) + stableReplication := defaultReplication(actual) + + desired.Spec.Volume.Replicas = currentReplicas + setDefaultReplication(desired, stableReplication) + + ready, err := seaweedComponentsAndWorkloadsReady(ctx, kubeClient, actual) + if err != nil { + return nil, err + } + if !ready { + return topologyResult( + metav1.ConditionFalse, + "WaitingForTopologyAdoption", + "waiting for the existing SeaweedFS components before recording their topology", + ), nil + } + + setStableReplication(desired, stableReplication) + return topologyResult( + metav1.ConditionFalse, + "TopologyAdopted", + fmt.Sprintf("recorded existing SeaweedFS topology at %d volume servers with replication %s", currentReplicas, stableReplication), + ), nil +} + +func migrateTopology( + ctx context.Context, + kubeClient client.Client, + desired, actual *seaweedv1.Seaweed, + owner client.Object, + stableReplication, targetReplication string, +) ([]metav1.Condition, error) { + targetReplicas := desired.Spec.Volume.Replicas + actualReplicas := volumeReplicas(actual) + readyReplicas := actual.Status.Volume.ReadyReplicas + + if targetReplicas < actualReplicas { + desired.Spec.Volume.Replicas = actualReplicas + setDefaultReplication(desired, stableReplication) + return topologyResult( + metav1.ConditionFalse, + "ReplicaReductionUnsupported", + fmt.Sprintf("refusing to reduce SeaweedFS volume servers from %d to %d without an evacuation workflow", actualReplicas, targetReplicas), + ), nil + } + + if actualReplicas < targetReplicas || readyReplicas < targetReplicas { + setDefaultReplication(desired, stableReplication) + return topologyResult( + metav1.ConditionFalse, + "ScalingVolumeServers", + fmt.Sprintf("scaling SeaweedFS volume servers from %d to %d before changing replication", actualReplicas, targetReplicas), + ), nil + } + + actualReplication := defaultReplication(actual) + ready, err := seaweedComponentsAndWorkloadsReady(ctx, kubeClient, actual) + if err != nil { + return nil, err + } + if !ready { + setDefaultReplication(desired, actualReplication) + return topologyResult( + metav1.ConditionFalse, + "WaitingForSeaweedComponents", + "waiting for SeaweedFS master, volume, filer, and S3 components before continuing topology migration", + ), nil + } + if targetReplication == stableReplication && actualReplication == targetReplication { + return topologyResult( + metav1.ConditionTrue, + "TopologyReady", + fmt.Sprintf("SeaweedFS has %d ready volume servers with replication %s", readyReplicas, targetReplication), + ), nil + } + + if actualReplication == stableReplication { + setDefaultReplication(desired, stableReplication) + state, message, err := reconcileTopologyJob( + ctx, kubeClient, desired, owner, topologyJobPreVerify, targetReplication, + verificationScript(desired), + ) + if err != nil { + return nil, err + } + if result := topologyJobResult( + state, + "PreMigrationVerificationFailed", + "PreMigrationVerificationRunning", + message, + ); result != nil { + return result, nil + } + + setDefaultReplication(desired, targetReplication) + return topologyResult( + metav1.ConditionFalse, + "UpdatingReplicationPolicy", + fmt.Sprintf("existing data verified; changing SeaweedFS replication from %s to %s", stableReplication, targetReplication), + ), nil + } + + if actualReplication != targetReplication { + setDefaultReplication(desired, stableReplication) + return topologyResult( + metav1.ConditionFalse, + "UnexpectedReplicationPolicy", + fmt.Sprintf("SeaweedFS replication is %s; expected stable %s or target %s", actualReplication, stableReplication, targetReplication), + ), nil + } + + state, message, err := reconcileTopologyJob( + ctx, kubeClient, desired, owner, topologyJobReplicate, targetReplication, + replicationScript(desired, targetReplication), + ) + if err != nil { + return nil, err + } + if result := topologyJobResult( + state, + "ReplicationFailed", + "ReplicationRunning", + message, + ); result != nil { + return result, nil + } + + s3Ready, err := s3GatewayRolloutReady(ctx, kubeClient, desired, targetReplication) + if err != nil { + return nil, err + } + if !s3Ready { + if err := rolloutS3Gateway(ctx, kubeClient, desired, targetReplication); err != nil { + return nil, err + } + return topologyResult( + metav1.ConditionFalse, + "RestartingS3Gateway", + "restarting the SeaweedFS S3 gateway after volume replication", + ), nil + } + + state, message, err = reconcileTopologyJob( + ctx, kubeClient, desired, owner, topologyJobPostVerify, targetReplication, + verificationScript(desired), + ) + if err != nil { + return nil, err + } + if result := topologyJobResult( + state, + "PostMigrationVerificationFailed", + "PostMigrationVerificationRunning", + message, + ); result != nil { + return result, nil + } + + setStableReplication(desired, targetReplication) + return topologyResult( + metav1.ConditionTrue, + "TopologyMigrated", + fmt.Sprintf("SeaweedFS data verified after replication changed from %s to %s", stableReplication, targetReplication), + ), nil +} + +func topologyJobResult( + state topologyJobState, + failedReason, pendingReason, message string, +) []metav1.Condition { + switch state { + case topologyJobFailed: + return topologyResult(metav1.ConditionFalse, failedReason, message) + case topologyJobPending: + return topologyResult(metav1.ConditionFalse, pendingReason, message) + default: + return nil + } +} + +func reconcileTopologyJob( + ctx context.Context, + kubeClient client.Client, + seaweed *seaweedv1.Seaweed, + owner client.Object, + stage, targetReplication, script string, +) (topologyJobState, string, error) { + name := topologyJobName(seaweed.Name, stage, targetReplication) + job := &batchv1.Job{} + err := kubeClient.Get(ctx, types.NamespacedName{Namespace: seaweed.Namespace, Name: name}, job) + if err != nil && !apierrors.IsNotFound(err) { + return topologyJobPending, "", err + } + if apierrors.IsNotFound(err) { + job = topologyJob(seaweed, name, stage, targetReplication, script) + if err := controllerutil.SetOwnerReference(owner, job, kubeClient.Scheme()); err != nil { + return topologyJobPending, "", err + } + if err := kubeClient.Create(ctx, job); err != nil { + return topologyJobPending, "", err + } + return topologyJobPending, fmt.Sprintf("SeaweedFS topology %s job %s created", stage, name), nil + } + + for _, condition := range job.Status.Conditions { + if condition.Type == batchv1.JobFailed && condition.Status == corev1.ConditionTrue { + message := condition.Message + if message == "" { + message = fmt.Sprintf("SeaweedFS topology %s job %s failed", stage, name) + } + return topologyJobFailed, message, nil + } + if condition.Type == batchv1.JobComplete && condition.Status == corev1.ConditionTrue { + return topologyJobSucceeded, fmt.Sprintf("SeaweedFS topology %s job %s succeeded", stage, name), nil + } + } + return topologyJobPending, fmt.Sprintf("waiting for SeaweedFS topology %s job %s", stage, name), nil +} + +func topologyJob( + seaweed *seaweedv1.Seaweed, + name, stage, targetReplication, script string, +) *batchv1.Job { + return &batchv1.Job{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Namespace: seaweed.Namespace, + Labels: map[string]string{ + "app.kubernetes.io/managed-by": "wandb-operator", + "app.kubernetes.io/instance": seaweed.Name, + "app.kubernetes.io/component": "seaweedfs-topology", + "operator.wandb.com/stage": stage, + }, + Annotations: map[string]string{ + "operator.wandb.com/target-replication": targetReplication, + }, + }, + Spec: batchv1.JobSpec{ + BackoffLimit: ptr.To[int32](0), + ActiveDeadlineSeconds: ptr.To[int64](86400), + Template: corev1.PodTemplateSpec{ + ObjectMeta: metav1.ObjectMeta{ + Labels: map[string]string{ + "app.kubernetes.io/managed-by": "wandb-operator", + "app.kubernetes.io/instance": seaweed.Name, + "app.kubernetes.io/component": "seaweedfs-topology", + }, + }, + Spec: corev1.PodSpec{ + RestartPolicy: corev1.RestartPolicyNever, + ImagePullSecrets: seaweed.Spec.ImagePullSecrets, + Containers: []corev1.Container{{ + Name: "topology", + Image: seaweed.Spec.Image, + ImagePullPolicy: seaweed.Spec.ImagePullPolicy, + Command: []string{"/bin/sh", "-c"}, + Args: []string{script}, + }}, + }, + }, + }, + } +} + +func verificationScript(seaweed *seaweedv1.Seaweed) string { + master := fmt.Sprintf("%s-master:%d", seaweed.Name, seaweedv1.MasterHTTPPort) + filer := fmt.Sprintf("%s-filer:%d", seaweed.Name, seaweedv1.FilerHTTPPort) + return fmt.Sprintf(`set -eu +echo "fs.verify -concurrency 8 /buckets" | weed shell -master=%s -filer=%s | awk ' + /failed verify/ { + failures++ + if (failures <= 20) print + } + /^total / { print } + /^verified / { + print + summary=$0 + } + END { + if (summary !~ /^verified [0-9]+ files, error 0 files[[:space:]]*$/) exit 1 + } +' +`, master, filer) +} + +func replicationScript(seaweed *seaweedv1.Seaweed, targetReplication string) string { + master := fmt.Sprintf("%s-master:%d", seaweed.Name, seaweedv1.MasterHTTPPort) + filer := fmt.Sprintf("%s-filer:%d", seaweed.Name, seaweedv1.FilerHTTPPort) + return fmt.Sprintf(`set -eu +printf 'lock\nvolume.configure.replication -replication=%s\nvolume.fix.replication -apply -doDelete=false\nunlock\n' | weed shell -master=%s -filer=%s +output=/tmp/seaweed-replication-check.out +echo "volume.fix.replication -verbose -doDelete=false" | weed shell -master=%s -filer=%s | tee "$output" +if grep -Eq 'under replicated|failed to place|not well placed|mismatch in topology' "$output"; then + exit 1 +fi +`, targetReplication, master, filer, master, filer) +} + +func topologyResult(status metav1.ConditionStatus, reason, message string) []metav1.Condition { + return []metav1.Condition{{ + Type: SeaweedTopologyReadyType, + Status: status, + Reason: reason, + Message: message, + }} +} + +func defaultReplication(seaweed *seaweedv1.Seaweed) string { + if seaweed != nil && seaweed.Spec.Master != nil && seaweed.Spec.Master.DefaultReplication != nil { + return *seaweed.Spec.Master.DefaultReplication + } + return "000" +} + +func volumeReplicas(seaweed *seaweedv1.Seaweed) int32 { + if seaweed == nil || seaweed.Spec.Volume == nil { + return 0 + } + return seaweed.Spec.Volume.Replicas +} + +func setDefaultReplication(seaweed *seaweedv1.Seaweed, replication string) { + if seaweed.Spec.Master != nil { + seaweed.Spec.Master.DefaultReplication = ptr.To(replication) + } +} + +func setStableReplication(seaweed *seaweedv1.Seaweed, replication string) { + if seaweed.Annotations == nil { + seaweed.Annotations = map[string]string{} + } + seaweed.Annotations[stableReplicationAnnotation] = replication +} + +func rolloutS3Gateway( + ctx context.Context, + kubeClient client.Client, + seaweed *seaweedv1.Seaweed, + replication string, +) error { + deployment := &appsv1.Deployment{} + if err := kubeClient.Get(ctx, types.NamespacedName{ + Namespace: seaweed.Namespace, + Name: fmt.Sprintf("%s-s3", seaweed.Name), + }, deployment); err != nil { + return err + } + if deployment.Spec.Template.Annotations[s3RolloutAnnotation] == replication { + return nil + } + + base := deployment.DeepCopy() + if deployment.Spec.Template.Annotations == nil { + deployment.Spec.Template.Annotations = map[string]string{} + } + deployment.Spec.Template.Annotations[s3RolloutAnnotation] = replication + return kubeClient.Patch(ctx, deployment, client.MergeFrom(base)) +} + +func s3GatewayRolloutReady( + ctx context.Context, + kubeClient client.Client, + seaweed *seaweedv1.Seaweed, + replication string, +) (bool, error) { + if seaweed.Spec.S3 == nil || seaweed.Spec.S3.Replicas == 0 { + return true, nil + } + deployment := &appsv1.Deployment{} + err := kubeClient.Get(ctx, types.NamespacedName{ + Namespace: seaweed.Namespace, + Name: fmt.Sprintf("%s-s3", seaweed.Name), + }, deployment) + if apierrors.IsNotFound(err) { + return false, nil + } + if err != nil { + return false, err + } + if deployment.Spec.Template.Annotations[s3RolloutAnnotation] != replication { + return false, nil + } + return deploymentReady(deployment, seaweed.Spec.S3.Replicas), nil +} + +func targetJobToken(replication string) string { + return strings.ReplaceAll(replication, "0", "z") +} + +func topologyJobName(seaweedName, stage, replication string) string { + stageToken := map[string]string{ + topologyJobPreVerify: "p", + topologyJobReplicate: "r", + topologyJobPostVerify: "v", + }[stage] + return fmt.Sprintf("%s-swt-%s-%s", seaweedName, stageToken, targetJobToken(replication)) +} + +func seaweedComponentsReady(seaweed *seaweedv1.Seaweed) bool { + components := []seaweedv1.ComponentStatus{ + seaweed.Status.Master, + seaweed.Status.Volume, + seaweed.Status.Filer, + seaweed.Status.S3, + } + for _, component := range components { + if component.Replicas > 0 && component.ReadyReplicas != component.Replicas { + return false + } + } + return true +} + +func seaweedComponentsAndWorkloadsReady( + ctx context.Context, + kubeClient client.Client, + seaweed *seaweedv1.Seaweed, +) (bool, error) { + if !seaweedComponentsReady(seaweed) { + return false, nil + } + + statefulSets := make([]statefulSetTarget, 0, 3) + if seaweed.Spec.Master != nil { + statefulSets = append(statefulSets, statefulSetTarget{suffix: "master", replicas: seaweed.Spec.Master.Replicas}) + } + if seaweed.Spec.Volume != nil { + statefulSets = append(statefulSets, statefulSetTarget{suffix: "volume", replicas: seaweed.Spec.Volume.Replicas}) + } + if seaweed.Spec.Filer != nil { + statefulSets = append(statefulSets, statefulSetTarget{suffix: "filer", replicas: seaweed.Spec.Filer.Replicas}) + } + for _, component := range statefulSets { + if component.replicas == 0 { + continue + } + statefulSet := &appsv1.StatefulSet{} + err := kubeClient.Get(ctx, types.NamespacedName{ + Namespace: seaweed.Namespace, + Name: fmt.Sprintf("%s-%s", seaweed.Name, component.suffix), + }, statefulSet) + if apierrors.IsNotFound(err) { + return false, nil + } + if err != nil { + return false, err + } + if !statefulSetReady(statefulSet, component.replicas) { + return false, nil + } + } + + if seaweed.Spec.S3 == nil || seaweed.Spec.S3.Replicas == 0 { + return true, nil + } + deployment := &appsv1.Deployment{} + err := kubeClient.Get(ctx, types.NamespacedName{ + Namespace: seaweed.Namespace, + Name: fmt.Sprintf("%s-s3", seaweed.Name), + }, deployment) + if apierrors.IsNotFound(err) { + return false, nil + } + if err != nil { + return false, err + } + return deploymentReady(deployment, seaweed.Spec.S3.Replicas), nil +} + +func statefulSetReady(statefulSet *appsv1.StatefulSet, expected int32) bool { + return statefulSet.Generation <= statefulSet.Status.ObservedGeneration && + ptr.Deref(statefulSet.Spec.Replicas, 1) == expected && + statefulSet.Status.Replicas == expected && + statefulSet.Status.CurrentReplicas == expected && + statefulSet.Status.UpdatedReplicas == expected && + statefulSet.Status.ReadyReplicas == expected && + statefulSet.Status.CurrentRevision == statefulSet.Status.UpdateRevision +} + +func deploymentReady(deployment *appsv1.Deployment, expected int32) bool { + return deployment.Generation <= deployment.Status.ObservedGeneration && + ptr.Deref(deployment.Spec.Replicas, 1) == expected && + deployment.Status.Replicas == expected && + deployment.Status.UpdatedReplicas == expected && + deployment.Status.AvailableReplicas == expected && + deployment.Status.ReadyReplicas == expected +} diff --git a/internal/controller/infra/managed/objectstore/seaweedfs/topology_test.go b/internal/controller/infra/managed/objectstore/seaweedfs/topology_test.go new file mode 100644 index 00000000..2fb37b7e --- /dev/null +++ b/internal/controller/infra/managed/objectstore/seaweedfs/topology_test.go @@ -0,0 +1,273 @@ +package seaweedfs + +import ( + "context" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + apiv2 "github.com/wandb/operator/api/v2" + seaweedv1 "github.com/wandb/operator/pkg/vendored/seaweedfs-operator/seaweed.seaweedfs.com/v1" + appsv1 "k8s.io/api/apps/v1" + batchv1 "k8s.io/api/batch/v1" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/utils/ptr" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" +) + +var _ = Describe("SeaweedFS topology migration", func() { + const ( + namespace = "wandb" + name = "wandb-seaweedfs" + ) + + owner := func() *apiv2.WeightsAndBiases { + return &apiv2.WeightsAndBiases{ + ObjectMeta: metav1.ObjectMeta{Name: "wandb", Namespace: namespace}, + } + } + seaweed := func(replicas, ready int32, replication, stable string) *seaweedv1.Seaweed { + annotations := map[string]string{} + if stable != "" { + annotations[stableReplicationAnnotation] = stable + } + return &seaweedv1.Seaweed{ + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: namespace, Annotations: annotations}, + Spec: seaweedv1.SeaweedSpec{ + Image: "chrislusf/seaweedfs:4.35", + Master: &seaweedv1.MasterSpec{ + Replicas: 1, + DefaultReplication: ptr.To(replication), + }, + Volume: &seaweedv1.VolumeSpec{Replicas: replicas}, + Filer: &seaweedv1.FilerSpec{Replicas: 1}, + S3: &seaweedv1.S3GatewaySpec{Replicas: 1}, + }, + Status: seaweedv1.SeaweedStatus{ + Master: seaweedv1.ComponentStatus{Replicas: 1, ReadyReplicas: 1}, + Volume: seaweedv1.ComponentStatus{Replicas: replicas, ReadyReplicas: ready}, + Filer: seaweedv1.ComponentStatus{Replicas: 1, ReadyReplicas: 1}, + S3: seaweedv1.ComponentStatus{Replicas: 1, ReadyReplicas: 1}, + }, + } + } + readyWorkloads := func(volumeReplicas int32) []client.Object { + statefulSet := func(suffix string, replicas int32) *appsv1.StatefulSet { + return &appsv1.StatefulSet{ + ObjectMeta: metav1.ObjectMeta{ + Name: name + "-" + suffix, + Namespace: namespace, + Generation: 1, + }, + Spec: appsv1.StatefulSetSpec{Replicas: ptr.To(replicas)}, + Status: appsv1.StatefulSetStatus{ + ObservedGeneration: 1, + Replicas: replicas, + CurrentReplicas: replicas, + UpdatedReplicas: replicas, + ReadyReplicas: replicas, + CurrentRevision: "ready", + UpdateRevision: "ready", + }, + } + } + return []client.Object{ + statefulSet("master", 1), + statefulSet("volume", volumeReplicas), + statefulSet("filer", 1), + &appsv1.Deployment{ + ObjectMeta: metav1.ObjectMeta{Name: name + "-s3", Namespace: namespace, Generation: 1}, + Spec: appsv1.DeploymentSpec{Replicas: ptr.To[int32](1)}, + Status: appsv1.DeploymentStatus{ + ObservedGeneration: 1, + Replicas: 1, + UpdatedReplicas: 1, + ReadyReplicas: 1, + AvailableReplicas: 1, + }, + }, + } + } + completedJob := func(stage, target string) *batchv1.Job { + job := topologyJob( + seaweed(3, 3, target, "000"), + topologyJobName(name, stage, target), + stage, + target, + "true", + ) + job.Status.Conditions = []batchv1.JobCondition{{ + Type: batchv1.JobComplete, + Status: corev1.ConditionTrue, + }} + return job + } + newClient := func(objects ...client.Object) client.Client { + return fake.NewClientBuilder(). + WithScheme(writeScheme()). + WithObjects(objects...). + Build() + } + + It("initializes a new cluster directly at its requested topology", func() { + desired := seaweed(3, 0, "001", "") + conditions, err := reconcileTopology(context.Background(), newClient(), desired, nil, owner()) + + Expect(err).NotTo(HaveOccurred()) + Expect(conditions).To(ContainElement(SatisfyAll( + HaveField("Type", SeaweedTopologyReadyType), + HaveField("Status", metav1.ConditionTrue), + ))) + Expect(desired.Annotations).To(HaveKeyWithValue(stableReplicationAnnotation, "001")) + }) + + It("records a healthy legacy cluster before changing its topology", func() { + actual := seaweed(1, 1, "000", "") + desired := seaweed(3, 0, "001", "") + cl := newClient(readyWorkloads(1)...) + + conditions, err := reconcileTopology(context.Background(), cl, desired, actual, owner()) + + Expect(err).NotTo(HaveOccurred()) + Expect(conditions[0].Reason).To(Equal("TopologyAdopted")) + Expect(desired.Annotations).To(HaveKeyWithValue(stableReplicationAnnotation, "000")) + Expect(defaultReplication(desired)).To(Equal("000")) + Expect(desired.Spec.Volume.Replicas).To(Equal(int32(1))) + jobs := &batchv1.JobList{} + Expect(cl.List(context.Background(), jobs)).To(Succeed()) + Expect(jobs.Items).To(BeEmpty()) + }) + + It("adds volume servers without restarting the master into the new policy", func() { + actual := seaweed(1, 1, "000", "000") + desired := seaweed(3, 0, "001", "") + + conditions, err := reconcileTopology(context.Background(), newClient(), desired, actual, owner()) + + Expect(err).NotTo(HaveOccurred()) + Expect(defaultReplication(desired)).To(Equal("000")) + Expect(desired.Spec.Volume.Replicas).To(Equal(int32(3))) + Expect(conditions[0].Reason).To(Equal("ScalingVolumeServers")) + }) + + It("verifies data before changing the replication policy", func() { + actual := seaweed(3, 3, "000", "000") + desired := seaweed(3, 0, "001", "") + cl := newClient(readyWorkloads(3)...) + + conditions, err := reconcileTopology(context.Background(), cl, desired, actual, owner()) + + Expect(err).NotTo(HaveOccurred()) + Expect(defaultReplication(desired)).To(Equal("000")) + Expect(conditions[0].Reason).To(Equal("PreMigrationVerificationRunning")) + jobs := &batchv1.JobList{} + Expect(cl.List(context.Background(), jobs)).To(Succeed()) + Expect(jobs.Items).To(HaveLen(1)) + Expect(jobs.Items[0].Spec.Template.Spec.Containers[0].Args[0]).To(ContainSubstring("fs.verify")) + }) + + It("waits for the volume StatefulSet rollout before verifying data", func() { + actual := seaweed(3, 3, "000", "000") + desired := seaweed(3, 0, "001", "") + workloads := readyWorkloads(3) + volume := workloads[1].(*appsv1.StatefulSet) + volume.Status.ReadyReplicas = 2 + cl := newClient(workloads...) + + conditions, err := reconcileTopology(context.Background(), cl, desired, actual, owner()) + + Expect(err).NotTo(HaveOccurred()) + Expect(conditions[0].Reason).To(Equal("WaitingForSeaweedComponents")) + jobs := &batchv1.JobList{} + Expect(cl.List(context.Background(), jobs)).To(Succeed()) + Expect(jobs.Items).To(BeEmpty()) + }) + + It("changes the master policy only after pre-verification succeeds", func() { + actual := seaweed(3, 3, "000", "000") + desired := seaweed(3, 0, "001", "") + preVerify := completedJob(topologyJobPreVerify, "001") + + conditions, err := reconcileTopology( + context.Background(), + newClient(append(readyWorkloads(3), preVerify)...), + desired, + actual, + owner(), + ) + + Expect(err).NotTo(HaveOccurred()) + Expect(defaultReplication(desired)).To(Equal("001")) + Expect(conditions[0].Reason).To(Equal("UpdatingReplicationPolicy")) + }) + + It("rolls the S3 gateway after replication and before post-verification", func() { + actual := seaweed(3, 3, "001", "000") + desired := seaweed(3, 0, "001", "") + preVerify := completedJob(topologyJobPreVerify, "001") + replicate := completedJob(topologyJobReplicate, "001") + workloads := readyWorkloads(3) + cl := newClient(append(workloads, preVerify, replicate)...) + + conditions, err := reconcileTopology( + context.Background(), + cl, + desired, + actual, + owner(), + ) + + Expect(err).NotTo(HaveOccurred()) + Expect(conditions[0].Reason).To(Equal("RestartingS3Gateway")) + s3 := &appsv1.Deployment{} + Expect(cl.Get(context.Background(), client.ObjectKey{Name: name + "-s3", Namespace: namespace}, s3)).To(Succeed()) + Expect(s3.Spec.Template.Annotations).To(HaveKeyWithValue(s3RolloutAnnotation, "001")) + }) + + It("replicates old volumes and verifies every filer chunk before completing", func() { + actual := seaweed(3, 3, "001", "000") + desired := seaweed(3, 0, "001", "") + preVerify := completedJob(topologyJobPreVerify, "001") + replicate := completedJob(topologyJobReplicate, "001") + postVerify := completedJob(topologyJobPostVerify, "001") + workloads := readyWorkloads(3) + workloads[3].(*appsv1.Deployment).Spec.Template.Annotations = map[string]string{ + s3RolloutAnnotation: "001", + } + + conditions, err := reconcileTopology( + context.Background(), + newClient(append(workloads, preVerify, replicate, postVerify)...), + desired, + actual, + owner(), + ) + + Expect(err).NotTo(HaveOccurred()) + Expect(conditions[0].Status).To(Equal(metav1.ConditionTrue)) + Expect(conditions[0].Reason).To(Equal("TopologyMigrated")) + Expect(desired.Annotations).To(HaveKeyWithValue(stableReplicationAnnotation, "001")) + }) + + It("waits for the S3 rollout generation before post-verification", func() { + actual := seaweed(3, 3, "001", "000") + desired := seaweed(3, 0, "001", "") + preVerify := completedJob(topologyJobPreVerify, "001") + replicate := completedJob(topologyJobReplicate, "001") + workloads := readyWorkloads(3) + s3 := workloads[3].(*appsv1.Deployment) + s3.Spec.Template.Annotations = map[string]string{s3RolloutAnnotation: "001"} + s3.Generation = 2 + s3.Status.ObservedGeneration = 1 + cl := newClient(append(workloads, preVerify, replicate)...) + + conditions, err := reconcileTopology(context.Background(), cl, desired, actual, owner()) + + Expect(err).NotTo(HaveOccurred()) + Expect(conditions[0].Reason).To(Equal("WaitingForSeaweedComponents")) + jobs := &batchv1.JobList{} + Expect(cl.List(context.Background(), jobs)).To(Succeed()) + Expect(jobs.Items).To(HaveLen(2)) + }) +}) diff --git a/internal/controller/infra/managed/objectstore/seaweedfs/write.go b/internal/controller/infra/managed/objectstore/seaweedfs/write.go index 7daac715..c42df9b9 100644 --- a/internal/controller/infra/managed/objectstore/seaweedfs/write.go +++ b/internal/controller/infra/managed/objectstore/seaweedfs/write.go @@ -10,7 +10,10 @@ import ( "github.com/wandb/operator/internal/controller/infra/objectstore" "github.com/wandb/operator/internal/logx" seaweedv1 "github.com/wandb/operator/pkg/vendored/seaweedfs-operator/seaweed.seaweedfs.com/v1" + appsv1 "k8s.io/api/apps/v1" corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/apimachinery/pkg/types" @@ -62,6 +65,25 @@ func WriteState( } result := make([]metav1.Condition, 0) + topologyConditions, err := reconcileTopology(ctx, kubeClient, desiredCr, actual, wandbOwner) + if err != nil { + result = append(result, metav1.Condition{ + Type: common.ReconciledType, + Status: metav1.ConditionFalse, + Reason: common.ApiErrorReason, + }) + return result, nil + } + result = append(result, topologyConditions...) + + if err := preserveClaimTemplateStorage(ctx, kubeClient, desiredCr); err != nil { + result = append(result, metav1.Condition{ + Type: common.ReconciledType, + Status: metav1.ConditionFalse, + Reason: common.ApiErrorReason, + }) + return result, nil + } action, err := common.CrudResource(ctx, kubeClient, desiredCr, actual) if err != nil { @@ -142,6 +164,68 @@ func WriteState( return result, nil } +func preserveClaimTemplateStorage( + ctx context.Context, + kubeClient client.Client, + desired *seaweedv1.Seaweed, +) error { + if desired == nil { + return nil + } + + volumeStorage, found, err := statefulSetClaimTemplateStorage( + ctx, + kubeClient, + types.NamespacedName{Namespace: desired.Namespace, Name: desired.Name + "-volume"}, + ) + if err != nil { + return err + } + if found && desired.Spec.Volume != nil { + if desired.Spec.Volume.Requests == nil { + desired.Spec.Volume.Requests = corev1.ResourceList{} + } + desired.Spec.Volume.Requests[corev1.ResourceStorage] = volumeStorage + } + + filerStorage, found, err := statefulSetClaimTemplateStorage( + ctx, + kubeClient, + types.NamespacedName{Namespace: desired.Namespace, Name: desired.Name + "-filer"}, + ) + if err != nil { + return err + } + if found && desired.Spec.Filer != nil && desired.Spec.Filer.Persistence != nil { + if desired.Spec.Filer.Persistence.Resources.Requests == nil { + desired.Spec.Filer.Persistence.Resources.Requests = corev1.ResourceList{} + } + desired.Spec.Filer.Persistence.Resources.Requests[corev1.ResourceStorage] = filerStorage + } + + return nil +} + +func statefulSetClaimTemplateStorage( + ctx context.Context, + kubeClient client.Client, + nsn types.NamespacedName, +) (resource.Quantity, bool, error) { + statefulSet := &appsv1.StatefulSet{} + if err := kubeClient.Get(ctx, nsn, statefulSet); err != nil { + if apierrors.IsNotFound(err) { + return resource.Quantity{}, false, nil + } + return resource.Quantity{}, false, fmt.Errorf("get StatefulSet %s/%s: %w", nsn.Namespace, nsn.Name, err) + } + for _, claimTemplate := range statefulSet.Spec.VolumeClaimTemplates { + if storage, ok := claimTemplate.Spec.Resources.Requests[corev1.ResourceStorage]; ok { + return storage, true, nil + } + } + return resource.Quantity{}, false, nil +} + // writeSeaweedS3Config persists the SeaweedFS S3 identity config secret, // preserving the existing secret key when one is already present so credentials // stay stable across reconciles, and returns the resolved ConnInfo. diff --git a/internal/controller/infra/managed/objectstore/seaweedfs/write_test.go b/internal/controller/infra/managed/objectstore/seaweedfs/write_test.go index 8d2f78bb..f277b89d 100644 --- a/internal/controller/infra/managed/objectstore/seaweedfs/write_test.go +++ b/internal/controller/infra/managed/objectstore/seaweedfs/write_test.go @@ -10,6 +10,9 @@ import ( "github.com/wandb/operator/internal/controller/common" seaweedv1 "github.com/wandb/operator/pkg/vendored/seaweedfs-operator/seaweed.seaweedfs.com/v1" "github.com/wandb/operator/pkg/wandb/manifest" + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/types" @@ -84,4 +87,39 @@ var _ = Describe("SeaweedFS WriteState", func() { Expect(conn).NotTo(BeNil()) Expect(hasCondition(conds, SeaweedConnectionInfoType, metav1.ConditionTrue)).To(BeTrue()) }) + + It("preserves existing StatefulSet claim template sizes", func() { + desired.Spec.Volume.Requests[corev1.ResourceStorage] = resource.MustParse("100Gi") + desired.Spec.Filer.Persistence.Resources.Requests[corev1.ResourceStorage] = resource.MustParse("50Gi") + volumeStatefulSet := statefulSetWithStorage(desired.Name+"-volume", desired.Namespace, "10Gi") + filerStatefulSet := statefulSetWithStorage(desired.Name+"-filer", desired.Namespace, "20Gi") + cl := fake.NewClientBuilder(). + WithScheme(writeScheme()). + WithObjects(volumeStatefulSet, filerStatefulSet). + Build() + + _, _ = WriteState(ctx, cl, specNsn, desired, envCfg, wandb) + + actual := &seaweedv1.Seaweed{} + Expect(cl.Get(ctx, client.ObjectKeyFromObject(desired), actual)).To(Succeed()) + Expect(actual.Spec.Volume.Requests[corev1.ResourceStorage]).To(Equal(resource.MustParse("10Gi"))) + Expect(actual.Spec.Filer.Persistence.Resources.Requests[corev1.ResourceStorage]).To(Equal(resource.MustParse("20Gi"))) + }) }) + +func statefulSetWithStorage(name, namespace, storage string) *appsv1.StatefulSet { + return &appsv1.StatefulSet{ + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: namespace}, + Spec: appsv1.StatefulSetSpec{ + VolumeClaimTemplates: []corev1.PersistentVolumeClaim{{ + Spec: corev1.PersistentVolumeClaimSpec{ + Resources: corev1.VolumeResourceRequirements{ + Requests: corev1.ResourceList{ + corev1.ResourceStorage: resource.MustParse(storage), + }, + }, + }, + }}, + }, + } +} diff --git a/internal/controller/infra/managed/redis/opstree/spec.go b/internal/controller/infra/managed/redis/opstree/spec.go index 606ac2cd..e0e39bef 100644 --- a/internal/controller/infra/managed/redis/opstree/spec.go +++ b/internal/controller/infra/managed/redis/opstree/spec.go @@ -165,7 +165,7 @@ func ToRedisStandaloneVendorSpec( return nil, nil } - if spec.Sentinel.Enabled { + if ptr.Deref(spec.Sentinel.Enabled, true) { return nil, nil } @@ -246,7 +246,7 @@ func ToRedisSentinelVendorSpec( return nil, nil } - if !spec.Sentinel.Enabled { + if !ptr.Deref(spec.Sentinel.Enabled, true) { return nil, nil } @@ -324,7 +324,7 @@ func ToRedisReplicationVendorSpec( return nil, nil } - if !spec.Sentinel.Enabled { + if !ptr.Deref(spec.Sentinel.Enabled, true) { return nil, nil } diff --git a/internal/controller/infra/managed/redis/opstree/spec_test.go b/internal/controller/infra/managed/redis/opstree/spec_test.go index 766b8b8f..ff9ea818 100644 --- a/internal/controller/infra/managed/redis/opstree/spec_test.go +++ b/internal/controller/infra/managed/redis/opstree/spec_test.go @@ -6,14 +6,15 @@ import ( . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" apiv2 "github.com/wandb/operator/api/v2" - "github.com/wandb/operator/pkg/wandb/manifest" "github.com/wandb/operator/pkg/utils" redisv1beta2 "github.com/wandb/operator/pkg/vendored/redis-operator/redis/v1beta2" redisreplicationv1beta2 "github.com/wandb/operator/pkg/vendored/redis-operator/redisreplication/v1beta2" redissentinelv1beta2 "github.com/wandb/operator/pkg/vendored/redis-operator/redissentinel/v1beta2" + "github.com/wandb/operator/pkg/wandb/manifest" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" + "k8s.io/utils/ptr" ) var _ = Describe("Redis vendor specs", func() { @@ -54,6 +55,24 @@ var _ = Describe("Redis vendor specs", func() { expectRedisWritableTmpMount(replication.Spec.Storage.VolumeMount.MountPath) }) + It("treats omitted Sentinel configuration as enabled", func() { + wandb := redisWandb(true) + wandb.Spec.Redis[apiv2.DefaultInstanceName].ManagedRedis.Sentinel.Enabled = nil + spec := wandb.Spec.Redis[apiv2.DefaultInstanceName].ManagedRedis + + standalone, err := ToRedisStandaloneVendorSpec(context.Background(), wandb, spec, redisScheme(), manifest.Manifest{}) + Expect(err).NotTo(HaveOccurred()) + Expect(standalone).To(BeNil()) + + sentinel, err := ToRedisSentinelVendorSpec(context.Background(), wandb, spec, redisScheme(), manifest.Manifest{}) + Expect(err).NotTo(HaveOccurred()) + Expect(sentinel).NotTo(BeNil()) + + replication, err := ToRedisReplicationVendorSpec(context.Background(), wandb, spec, redisScheme(), manifest.Manifest{}) + Expect(err).NotTo(HaveOccurred()) + Expect(replication).NotTo(BeNil()) + }) + It("omits fixed Redis IDs in OpenShift mode", func() { utils.SetOpenShiftMode(true) @@ -95,7 +114,7 @@ func redisWandb(sentinel bool) *apiv2.WeightsAndBiases { Namespace: "wandb", StorageSize: "1Gi", Telemetry: apiv2.Telemetry{Enabled: true}, - Sentinel: apiv2.RedisSentinelSpec{Enabled: sentinel}, + Sentinel: apiv2.RedisSentinelSpec{Enabled: ptr.To(sentinel)}, }, }, }, diff --git a/internal/controller/reconciler/clickhouse.go b/internal/controller/reconciler/clickhouse.go index f842da17..4702edb8 100644 --- a/internal/controller/reconciler/clickhouse.go +++ b/internal/controller/reconciler/clickhouse.go @@ -9,8 +9,10 @@ import ( externalch "github.com/wandb/operator/internal/controller/infra/external/clickhouse" "github.com/wandb/operator/internal/controller/infra/managed/clickhouse/altinity" "github.com/wandb/operator/internal/controller/infra/managed/clickhouse/altinity/keeper" + "github.com/wandb/operator/internal/controller/infra/managed/objectstore/seaweedfs" "github.com/wandb/operator/pkg/utils" "github.com/wandb/operator/pkg/wandb/manifest" + apimeta "k8s.io/apimachinery/pkg/api/meta" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/types" "k8s.io/client-go/tools/record" @@ -28,12 +30,14 @@ func clickHouseWriteState( client client.Client, wandb *apiv2.WeightsAndBiases, mfst manifest.Manifest, + objectStoreConditions map[string][]metav1.Condition, ) map[string][]metav1.Condition { out := map[string][]metav1.Condition{} + topologyConditions, _ := apiv2.ResolveInstance(objectStoreConditions, clickHouseObjectStoreInstance) for key, spec := range wandb.Spec.ClickHouse { switch { case spec.ManagedClickHouse != nil: - out[key] = managedClickHouseWriteState(ctx, client, wandb, spec.ManagedClickHouse, mfst) + out[key] = managedClickHouseWriteState(ctx, client, wandb, spec.ManagedClickHouse, mfst, topologyConditions) case spec.ExternalClickHouse != nil: out[key] = externalch.WriteState(ctx, client, wandb, key, spec.ExternalClickHouse) } @@ -150,6 +154,7 @@ func managedClickHouseWriteState( wandb *apiv2.WeightsAndBiases, spec *apiv2.ManagedClickHouseSpec, mfst manifest.Manifest, + objectStoreConditions []metav1.Condition, ) []metav1.Condition { log := ctrl.LoggerFrom(ctx) @@ -178,6 +183,15 @@ func managedClickHouseWriteState( objStoreStatus, _ := apiv2.ResolveInstance(wandb.Status.ObjectStoreStatus, clickHouseObjectStoreInstance) objStoreSpec, _ := apiv2.ResolveInstance(wandb.Spec.ObjectStore, clickHouseObjectStoreInstance) waitForObjectStore := objStoreSpec.ManagedObjectStore != nil + if waitForObjectStore && !apimeta.IsStatusConditionTrue(objectStoreConditions, seaweedfs.SeaweedTopologyReadyType) { + log.Info("waiting for SeaweedFS topology migration before reconciling ClickHouse") + return []metav1.Condition{{ + Type: common.ReconciledType, + Status: metav1.ConditionFalse, + Reason: common.PendingCreateReason, + Message: "waiting for SeaweedFS topology migration and verification", + }} + } // Resolve the bucket connection; wait and requeue if it isn't ready yet. objStorage, objStorageEndpoint, err := altinity.ResolveObjectStorage(ctx, client, spec, &objStoreStatus.Connection) diff --git a/internal/controller/reconciler/readiness_test.go b/internal/controller/reconciler/readiness_test.go index 6b771b24..8254c7ab 100644 --- a/internal/controller/reconciler/readiness_test.go +++ b/internal/controller/reconciler/readiness_test.go @@ -83,6 +83,29 @@ func TestSetReadyStatusKeepsBooleanAndConditionConsistent(t *testing.T) { } } +func TestInfrastructureBlockersIncludesDegradedClickHouse(t *testing.T) { + wandb := &apiv2.WeightsAndBiases{ + Spec: apiv2.WeightsAndBiasesSpec{ + ClickHouse: map[string]apiv2.ClickHouseSpec{ + apiv2.DefaultInstanceName: {ManagedClickHouse: &apiv2.ManagedClickHouseSpec{}}, + }, + }, + Status: apiv2.WeightsAndBiasesStatus{ + ClickHouseStatus: map[string]apiv2.ClickHouseInfraStatus{ + apiv2.DefaultInstanceName: { + WBInfraStatus: apiv2.WBInfraStatus{Ready: false, State: "Degraded"}, + }, + }, + }, + } + + blockers := infrastructureBlockers(wandb) + + if len(blockers) != 1 || blockers[0] != "clickhouse/default" { + t.Fatalf("unexpected infrastructure blockers: %#v", blockers) + } +} + func TestRunMigrationsSurfacesFailedJobPhaseAndReason(t *testing.T) { scheme := runtime.NewScheme() if err := apiv2.AddToScheme(scheme); err != nil { diff --git a/internal/controller/reconciler/reconcile_v2.go b/internal/controller/reconciler/reconcile_v2.go index 57891d89..a6b8a939 100644 --- a/internal/controller/reconciler/reconcile_v2.go +++ b/internal/controller/reconciler/reconcile_v2.go @@ -302,13 +302,52 @@ func Reconcile( // Apply manifest-derived infra sizing before provisioning ApplyInfraSizing(wandb, manifest) + expansionBlockers, expansionPending, err := reconcileSeaweedFSVolumeExpansion(ctx, client, wandb) + if err != nil { + return ctrl.Result{}, fmt.Errorf("reconcile SeaweedFS volume expansion: %w", err) + } + if len(expansionBlockers) > 0 { + statusBefore := wandb.DeepCopy().Status + message := volumeExpansionBlockersMessage(expansionBlockers) + if err := updateReadyStatus( + ctx, + client, + wandb, + statusBefore, + false, + "StorageExpansionUnsupported", + message, + ); err != nil { + return ctrl.Result{}, err + } + recorder.Event(wandb, corev1.EventTypeWarning, "StorageExpansionUnsupported", message) + return ctrl.Result{RequeueAfter: defaultRequeueDuration}, nil + } + + if len(expansionPending) > 0 { + statusBefore := wandb.DeepCopy().Status + message := volumeExpansionPendingMessage(expansionPending) + if err := updateReadyStatus( + ctx, + client, + wandb, + statusBefore, + false, + "StorageExpansionInProgress", + message, + ); err != nil { + return ctrl.Result{}, err + } + return ctrl.Result{RequeueAfter: defaultRequeueDuration}, nil + } + ///////////////////////// // Write Infra State redisConditions := redisWriteState(ctx, client, wandb, manifest) mysqlConditions := mysqlWriteState(ctx, client, wandb, manifest) objectStoreConditions, objectStoreConnection := objectStoreWriteState(ctx, client, wandb, manifest) kafkaConditions := kafkaWriteState(ctx, client, wandb, manifest) - clickHouseConditions := clickHouseWriteState(ctx, client, wandb, manifest) + clickHouseConditions := clickHouseWriteState(ctx, client, wandb, manifest, objectStoreConditions) ///////////////////////// // Read Infra State diff --git a/internal/controller/reconciler/storage_expansion.go b/internal/controller/reconciler/storage_expansion.go new file mode 100644 index 00000000..615b4c6a --- /dev/null +++ b/internal/controller/reconciler/storage_expansion.go @@ -0,0 +1,211 @@ +package reconciler + +import ( + "context" + "fmt" + "sort" + "strings" + + apiv2 "github.com/wandb/operator/api/v2" + corev1 "k8s.io/api/core/v1" + storagev1 "k8s.io/api/storage/v1" + "k8s.io/apimachinery/pkg/api/resource" + "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +type seaweedFSVolumeTarget struct { + component string + name string + namespace string + desired string +} + +type volumeExpansionBlocker struct { + component string + pvc types.NamespacedName + storageClass string + current resource.Quantity + desired resource.Quantity +} + +type volumeExpansionPending struct { + component string + pvc types.NamespacedName + current resource.Quantity + desired resource.Quantity +} + +func (b volumeExpansionBlocker) String() string { + class := b.storageClass + if class == "" { + class = "" + } + return fmt.Sprintf( + "%s requires PVC %s/%s to grow from %s to %s, but StorageClass %q does not allow volume expansion", + b.component, + b.pvc.Namespace, + b.pvc.Name, + b.current.String(), + b.desired.String(), + class, + ) +} + +func reconcileSeaweedFSVolumeExpansion( + ctx context.Context, + c client.Client, + wandb *apiv2.WeightsAndBiases, +) ([]volumeExpansionBlocker, []volumeExpansionPending, error) { + pvcsByNamespace := map[string][]corev1.PersistentVolumeClaim{} + storageClasses := map[string]*storagev1.StorageClass{} + var blockers []volumeExpansionBlocker + var pending []volumeExpansionPending + + for _, target := range seaweedFSVolumeTargets(wandb) { + if target.desired == "" { + continue + } + + pvcs, ok := pvcsByNamespace[target.namespace] + if !ok { + pvcList := &corev1.PersistentVolumeClaimList{} + if err := c.List(ctx, pvcList, client.InNamespace(target.namespace)); err != nil { + return nil, nil, fmt.Errorf("list PVCs in namespace %q: %w", target.namespace, err) + } + pvcs = pvcList.Items + pvcsByNamespace[target.namespace] = pvcs + } + + desired, err := resource.ParseQuantity(target.desired) + if err != nil { + return nil, nil, fmt.Errorf("parse desired storage for %s: %w", target.component, err) + } + for i := range pvcs { + pvc := &pvcs[i] + if pvc.Labels["app.kubernetes.io/instance"] != target.name || + pvc.Labels["app.kubernetes.io/component"] != target.component { + continue + } + + current := pvc.Spec.Resources.Requests[corev1.ResourceStorage] + if desired.Cmp(current) > 0 { + storageClassName := "" + if pvc.Spec.StorageClassName != nil { + storageClassName = *pvc.Spec.StorageClassName + } + expandable, err := storageClassAllowsExpansion(ctx, c, storageClasses, storageClassName) + if err != nil { + return nil, nil, fmt.Errorf( + "check StorageClass for PVC %s/%s: %w", + pvc.Namespace, + pvc.Name, + err, + ) + } + if !expandable { + blockers = append(blockers, volumeExpansionBlocker{ + component: "objectStore/" + target.component, + pvc: client.ObjectKeyFromObject(pvc), + storageClass: storageClassName, + current: current, + desired: desired, + }) + continue + } + + before := pvc.DeepCopy() + pvc.Spec.Resources.Requests[corev1.ResourceStorage] = desired + if err := c.Patch(ctx, pvc, client.MergeFrom(before)); err != nil { + return nil, nil, fmt.Errorf("expand PVC %s/%s: %w", pvc.Namespace, pvc.Name, err) + } + } + + capacity := pvc.Status.Capacity[corev1.ResourceStorage] + if desired.Cmp(capacity) > 0 { + pending = append(pending, volumeExpansionPending{ + component: "objectStore/" + target.component, + pvc: client.ObjectKeyFromObject(pvc), + current: capacity, + desired: desired, + }) + } + } + } + + sort.Slice(blockers, func(i, j int) bool { + return blockers[i].String() < blockers[j].String() + }) + sort.Slice(pending, func(i, j int) bool { + return pending[i].pvc.String() < pending[j].pvc.String() + }) + return blockers, pending, nil +} + +func storageClassAllowsExpansion( + ctx context.Context, + c client.Client, + cache map[string]*storagev1.StorageClass, + name string, +) (bool, error) { + if name == "" { + return false, nil + } + storageClass, ok := cache[name] + if !ok { + storageClass = &storagev1.StorageClass{} + if err := c.Get(ctx, types.NamespacedName{Name: name}, storageClass); err != nil { + return false, fmt.Errorf("get StorageClass %q: %w", name, err) + } + cache[name] = storageClass + } + return storageClass.AllowVolumeExpansion != nil && *storageClass.AllowVolumeExpansion, nil +} + +func seaweedFSVolumeTargets(wandb *apiv2.WeightsAndBiases) []seaweedFSVolumeTarget { + var targets []seaweedFSVolumeTarget + for _, instance := range wandb.Spec.ObjectStore { + spec := instance.ManagedObjectStore + if spec == nil { + continue + } + targets = append(targets, + seaweedFSVolumeTarget{ + component: "volume", + name: spec.Name, + namespace: spec.Namespace, + desired: spec.StorageSize, + }, + seaweedFSVolumeTarget{ + component: "filer", + name: spec.Name, + namespace: spec.Namespace, + desired: spec.SeaweedObjectStoreSpec.FilerStorageSize, + }, + ) + } + return targets +} + +func volumeExpansionBlockersMessage(blockers []volumeExpansionBlocker) string { + messages := make([]string, 0, len(blockers)) + for _, blocker := range blockers { + messages = append(messages, blocker.String()) + } + return strings.Join(messages, "; ") +} + +func volumeExpansionPendingMessage(pending []volumeExpansionPending) string { + messages := make([]string, 0, len(pending)) + for _, item := range pending { + messages = append(messages, fmt.Sprintf( + "waiting for %s PVC %s/%s capacity to grow from %s to %s", + item.component, + item.pvc.Namespace, + item.pvc.Name, + item.current.String(), + item.desired.String(), + )) + } + return strings.Join(messages, "; ") +} diff --git a/internal/controller/reconciler/storage_expansion_test.go b/internal/controller/reconciler/storage_expansion_test.go new file mode 100644 index 00000000..c82a8283 --- /dev/null +++ b/internal/controller/reconciler/storage_expansion_test.go @@ -0,0 +1,182 @@ +package reconciler + +import ( + "context" + "testing" + + apiv2 "github.com/wandb/operator/api/v2" + corev1 "k8s.io/api/core/v1" + storagev1 "k8s.io/api/storage/v1" + "k8s.io/apimachinery/pkg/api/resource" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/utils/ptr" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" +) + +func TestReconcileSeaweedFSVolumeExpansion(t *testing.T) { + scheme := runtime.NewScheme() + if err := corev1.AddToScheme(scheme); err != nil { + t.Fatal(err) + } + if err := storagev1.AddToScheme(scheme); err != nil { + t.Fatal(err) + } + + wandb := func(storageSize string) *apiv2.WeightsAndBiases { + return &apiv2.WeightsAndBiases{ + Spec: apiv2.WeightsAndBiasesSpec{ + ObjectStore: map[string]apiv2.ObjectStoreSpec{ + apiv2.DefaultInstanceName: { + ManagedObjectStore: &apiv2.ManagedObjectStoreSpec{ + Name: "wandb-seaweedfs", + Namespace: "wandb", + StorageSize: storageSize, + }, + }, + }, + }, + } + } + pvc := func(request, capacity string) *corev1.PersistentVolumeClaim { + return &corev1.PersistentVolumeClaim{ + ObjectMeta: metav1.ObjectMeta{ + Name: "mount0-wandb-seaweedfs-volume-0", + Namespace: "wandb", + Labels: map[string]string{ + "app.kubernetes.io/instance": "wandb-seaweedfs", + "app.kubernetes.io/component": "volume", + }, + }, + Spec: corev1.PersistentVolumeClaimSpec{ + StorageClassName: ptr.To("standard"), + Resources: corev1.VolumeResourceRequirements{ + Requests: corev1.ResourceList{ + corev1.ResourceStorage: resource.MustParse(request), + }, + }, + }, + Status: corev1.PersistentVolumeClaimStatus{ + Capacity: corev1.ResourceList{ + corev1.ResourceStorage: resource.MustParse(capacity), + }, + }, + } + } + storageClass := func(expandable bool) *storagev1.StorageClass { + return &storagev1.StorageClass{ + ObjectMeta: metav1.ObjectMeta{Name: "standard"}, + AllowVolumeExpansion: ptr.To(expandable), + } + } + + t.Run("blocks an increase when the StorageClass is not expandable", func(t *testing.T) { + c := fake.NewClientBuilder(). + WithScheme(scheme). + WithObjects(pvc("10Gi", "10Gi"), storageClass(false)). + Build() + + blockers, pending, err := reconcileSeaweedFSVolumeExpansion(context.Background(), c, wandb("100Gi")) + if err != nil { + t.Fatal(err) + } + if len(blockers) != 1 { + t.Fatalf("got %d blockers, want 1", len(blockers)) + } + if len(pending) != 0 { + t.Fatalf("got pending expansions %v, want none", pending) + } + want := `objectStore/volume requires PVC wandb/mount0-wandb-seaweedfs-volume-0 to grow from 10Gi to 100Gi, but StorageClass "standard" does not allow volume expansion` + if got := blockers[0].String(); got != want { + t.Fatalf("blocker = %q, want %q", got, want) + } + }) + + t.Run("grows the PVC and waits for capacity", func(t *testing.T) { + c := fake.NewClientBuilder(). + WithScheme(scheme). + WithObjects(pvc("10Gi", "10Gi"), storageClass(true)). + Build() + + blockers, pending, err := reconcileSeaweedFSVolumeExpansion(context.Background(), c, wandb("100Gi")) + if err != nil { + t.Fatal(err) + } + if len(blockers) != 0 { + t.Fatalf("got blockers %v, want none", blockers) + } + if len(pending) != 1 { + t.Fatalf("got %d pending expansions, want 1", len(pending)) + } + + actual := &corev1.PersistentVolumeClaim{} + if err := c.Get(context.Background(), client.ObjectKey{Namespace: "wandb", Name: "mount0-wandb-seaweedfs-volume-0"}, actual); err != nil { + t.Fatal(err) + } + if got := actual.Spec.Resources.Requests[corev1.ResourceStorage]; got.Cmp(resource.MustParse("100Gi")) != 0 { + t.Fatalf("PVC request = %s, want 100Gi", got.String()) + } + }) + + t.Run("grows the filer PVC", func(t *testing.T) { + filerPVC := pvc("20Gi", "20Gi") + filerPVC.Name = "data-wandb-seaweedfs-filer-0" + filerPVC.Labels["app.kubernetes.io/component"] = "filer" + desired := wandb("10Gi") + desired.Spec.ObjectStore[apiv2.DefaultInstanceName]. + ManagedObjectStore.SeaweedObjectStoreSpec.FilerStorageSize = "50Gi" + c := fake.NewClientBuilder(). + WithScheme(scheme). + WithObjects(filerPVC, storageClass(true)). + Build() + + blockers, pending, err := reconcileSeaweedFSVolumeExpansion(context.Background(), c, desired) + if err != nil { + t.Fatal(err) + } + if len(blockers) != 0 || len(pending) != 1 { + t.Fatalf("got blockers %v and pending %v, want one pending expansion", blockers, pending) + } + + actual := &corev1.PersistentVolumeClaim{} + if err := c.Get(context.Background(), client.ObjectKeyFromObject(filerPVC), actual); err != nil { + t.Fatal(err) + } + if got := actual.Spec.Resources.Requests[corev1.ResourceStorage]; got.Cmp(resource.MustParse("50Gi")) != 0 { + t.Fatalf("PVC request = %s, want 50Gi", got.String()) + } + }) + + t.Run("reports convergence when request and capacity match", func(t *testing.T) { + c := fake.NewClientBuilder().WithScheme(scheme).WithObjects(pvc("100Gi", "100Gi")).Build() + + blockers, pending, err := reconcileSeaweedFSVolumeExpansion(context.Background(), c, wandb("100Gi")) + if err != nil { + t.Fatal(err) + } + if len(blockers) != 0 || len(pending) != 0 { + t.Fatalf("got blockers %v and pending %v, want neither", blockers, pending) + } + }) + + t.Run("never shrinks a PVC", func(t *testing.T) { + c := fake.NewClientBuilder().WithScheme(scheme).WithObjects(pvc("100Gi", "100Gi")).Build() + + blockers, pending, err := reconcileSeaweedFSVolumeExpansion(context.Background(), c, wandb("10Gi")) + if err != nil { + t.Fatal(err) + } + if len(blockers) != 0 || len(pending) != 0 { + t.Fatalf("got blockers %v and pending %v, want neither", blockers, pending) + } + + actual := &corev1.PersistentVolumeClaim{} + if err := c.Get(context.Background(), client.ObjectKey{Namespace: "wandb", Name: "mount0-wandb-seaweedfs-volume-0"}, actual); err != nil { + t.Fatal(err) + } + if got := actual.Spec.Resources.Requests[corev1.ResourceStorage]; got.Cmp(resource.MustParse("100Gi")) != 0 { + t.Fatalf("PVC request = %s, want 100Gi", got.String()) + } + }) +} diff --git a/internal/controller/weightsandbiases_controller.go b/internal/controller/weightsandbiases_controller.go index 965264ee..cd7b2251 100644 --- a/internal/controller/weightsandbiases_controller.go +++ b/internal/controller/weightsandbiases_controller.go @@ -70,6 +70,7 @@ type WeightsAndBiasesReconciler struct { //+kubebuilder:rbac:groups=clickhouse-keeper.altinity.com,resources=clickhousekeeperinstallations/status,verbs=get //+kubebuilder:rbac:groups=cloud.google.com,resources=backendconfigs,verbs=update;delete;get;list;patch;create;watch //+kubebuilder:rbac:groups=events.k8s.io,resources=events,verbs=list;watch +//+kubebuilder:rbac:groups=storage.k8s.io,resources=storageclasses,verbs=get;list;watch //+kubebuilder:rbac:groups=grafana.integreatly.org,resources=grafanas;grafanadashboards;grafanadatasources,verbs=get;list;watch //+kubebuilder:rbac:groups=grafana.integreatly.org,resources=grafanas/status;grafanadashboards/status;grafanadatasources/status,verbs=get //+kubebuilder:rbac:groups=seaweed.seaweedfs.com,resources=seaweeds,verbs=get;list;watch;create;update;patch;delete diff --git a/internal/crdinstaller/crds/operator/apps.wandb.com_weightsandbiases.yaml b/internal/crdinstaller/crds/operator/apps.wandb.com_weightsandbiases.yaml index 4ee11815..2bfdb666 100644 --- a/internal/crdinstaller/crds/operator/apps.wandb.com_weightsandbiases.yaml +++ b/internal/crdinstaller/crds/operator/apps.wandb.com_weightsandbiases.yaml @@ -3864,9 +3864,8 @@ spec: type: object type: object enabled: + default: true type: boolean - required: - - enabled type: object storageSize: type: string diff --git a/internal/webhook/v2/weightsandbiases_defaulter_redis_test.go b/internal/webhook/v2/weightsandbiases_defaulter_redis_test.go index 45eb16b9..274ef70f 100644 --- a/internal/webhook/v2/weightsandbiases_defaulter_redis_test.go +++ b/internal/webhook/v2/weightsandbiases_defaulter_redis_test.go @@ -44,18 +44,34 @@ var _ = Describe("WeightsAndBiasesCustomDefaulter - Redis", func() { g.Expect(wandb.Spec.Redis[apiv2.DefaultInstanceName].ManagedRedis.Namespace).To(g.Equal("custom-redis-namespace")) }) - It("does not mutate unrelated Redis fields", func() { + It("defaults Sentinel on for managed Redis regardless of size", func() { + for _, size := range []apiv2.Size{apiv2.SizeDev, apiv2.SizeSmall} { + wandb := &apiv2.WeightsAndBiases{ + ObjectMeta: metav1.ObjectMeta{Name: "test-wandb", Namespace: "test-namespace"}, + Spec: apiv2.WeightsAndBiasesSpec{ + Size: size, + Redis: map[string]apiv2.RedisSpec{apiv2.DefaultInstanceName: {ManagedRedis: &apiv2.ManagedRedisSpec{}}}, + }, + } + + err := defaulter.Default(ctx, wandb) + g.Expect(err).ToNot(g.HaveOccurred()) + g.Expect(wandb.Spec.Redis[apiv2.DefaultInstanceName].ManagedRedis.Sentinel.Enabled).To(g.HaveValue(g.BeTrue())) + } + }) + + It("preserves explicitly disabled Sentinel", func() { wandb := &apiv2.WeightsAndBiases{ ObjectMeta: metav1.ObjectMeta{Name: "test-wandb", Namespace: "test-namespace"}, Spec: apiv2.WeightsAndBiasesSpec{ - Redis: map[string]apiv2.RedisSpec{apiv2.DefaultInstanceName: {ManagedRedis: &apiv2.ManagedRedisSpec{StorageSize: "20Gi", Sentinel: apiv2.RedisSentinelSpec{Enabled: true}}}}, + Size: apiv2.SizeSmall, + Redis: map[string]apiv2.RedisSpec{apiv2.DefaultInstanceName: {ManagedRedis: &apiv2.ManagedRedisSpec{Sentinel: apiv2.RedisSentinelSpec{Enabled: boolPtr(false)}}}}, }, } err := defaulter.Default(ctx, wandb) g.Expect(err).ToNot(g.HaveOccurred()) - g.Expect(wandb.Spec.Redis[apiv2.DefaultInstanceName].ManagedRedis.StorageSize).To(g.Equal("20Gi")) - g.Expect(wandb.Spec.Redis[apiv2.DefaultInstanceName].ManagedRedis.Sentinel.Enabled).To(g.BeTrue()) + g.Expect(wandb.Spec.Redis[apiv2.DefaultInstanceName].ManagedRedis.Sentinel.Enabled).To(g.HaveValue(g.BeFalse())) }) It("does not apply defaults when External is present", func() { diff --git a/internal/webhook/v2/weightsandbiases_webhook.go b/internal/webhook/v2/weightsandbiases_webhook.go index fd229ca4..70b322eb 100644 --- a/internal/webhook/v2/weightsandbiases_webhook.go +++ b/internal/webhook/v2/weightsandbiases_webhook.go @@ -252,8 +252,8 @@ func applyRedisDefaults(wandb *appsv2.WeightsAndBiases) { if spec.ManagedRedis.Namespace == "" { spec.ManagedRedis.Namespace = wandb.Namespace } - if wandb.Spec.Size != appsv2.SizeDev { - spec.ManagedRedis.Sentinel.Enabled = true + if spec.ManagedRedis.Sentinel.Enabled == nil { + spec.ManagedRedis.Sentinel.Enabled = ptr.To(true) } wandb.Spec.Redis[key] = spec } @@ -814,7 +814,7 @@ func validateRedisChanges(newWandb, oldWandb *appsv2.WeightsAndBiases) field.Err )) } - if oldSpec.Sentinel.Enabled != newSpec.Sentinel.Enabled { + if ptr.Deref(oldSpec.Sentinel.Enabled, true) != ptr.Deref(newSpec.Sentinel.Enabled, true) { errors = append(errors, field.Invalid( instancePath.Child("sentinel").Child("enabled"), newSpec.Sentinel.Enabled, diff --git a/internal/webhook/v2/weightsandbiases_webhook_test.go b/internal/webhook/v2/weightsandbiases_webhook_test.go index 8bb0d2ae..b2b64753 100644 --- a/internal/webhook/v2/weightsandbiases_webhook_test.go +++ b/internal/webhook/v2/weightsandbiases_webhook_test.go @@ -230,6 +230,40 @@ var _ = Describe("WeightsAndBiases Webhook", func() { Expect(warnings).To(BeEmpty()) }) + It("allows size upgrades when managed Redis topology is unchanged", func() { + oldObj.Spec.Size = appsv2.SizeDev + oldObj.Spec.Redis = map[string]appsv2.RedisSpec{appsv2.DefaultInstanceName: {ManagedRedis: &appsv2.ManagedRedisSpec{ + Namespace: "redis", + Sentinel: appsv2.RedisSentinelSpec{Enabled: boolPtr(false)}, + }}} + obj.Spec.Size = appsv2.SizeSmall + obj.Spec.Redis = map[string]appsv2.RedisSpec{appsv2.DefaultInstanceName: {ManagedRedis: &appsv2.ManagedRedisSpec{ + Namespace: "redis", + Sentinel: appsv2.RedisSentinelSpec{Enabled: boolPtr(false)}, + }}} + + Expect(defaulter.Default(ctx, obj)).To(Succeed()) + Expect(obj.Spec.Redis[appsv2.DefaultInstanceName].ManagedRedis.Sentinel.Enabled).To(HaveValue(BeFalse())) + warnings, err := validator.ValidateUpdate(ctx, oldObj, obj) + Expect(err).NotTo(HaveOccurred()) + Expect(warnings).To(BeEmpty()) + }) + + It("rejects managed Redis topology changes", func() { + oldObj.Spec.Redis = map[string]appsv2.RedisSpec{appsv2.DefaultInstanceName: {ManagedRedis: &appsv2.ManagedRedisSpec{ + Namespace: "redis", + Sentinel: appsv2.RedisSentinelSpec{Enabled: boolPtr(false)}, + }}} + obj.Spec.Redis = map[string]appsv2.RedisSpec{appsv2.DefaultInstanceName: {ManagedRedis: &appsv2.ManagedRedisSpec{ + Namespace: "redis", + Sentinel: appsv2.RedisSentinelSpec{Enabled: boolPtr(true)}, + }}} + + _, err := validator.ValidateUpdate(ctx, oldObj, obj) + Expect(err).To(HaveOccurred()) + Expect(err.Error()).To(ContainSubstring("Redis Sentinel cannot be toggled")) + }) + It("rejects decreasing managed MySQL replicas on update", func() { oldObj.Spec.MySQL = map[string]appsv2.MySQLSpec{appsv2.DefaultInstanceName: {ManagedMysql: &appsv2.ManagedMysqlSpec{Replicas: 3}}} obj.Spec.MySQL = map[string]appsv2.MySQLSpec{appsv2.DefaultInstanceName: {ManagedMysql: &appsv2.ManagedMysqlSpec{Replicas: 1}}}