wip

api/v1beta2/statefulset_webhhok.go

@@ -0,0 +1,99 @@
+package v1beta2
+
+import (
+ "context"
+ "fmt"
+
+ "github.com/cybozu-go/moco/pkg/constants"
+ admissionv1 "k8s.io/api/admission/v1"
+ appsv1 "k8s.io/api/apps/v1"
+ "k8s.io/apimachinery/pkg/api/equality"
+ "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+ "k8s.io/apimachinery/pkg/runtime"
+ "k8s.io/utils/ptr"
+ ctrl "sigs.k8s.io/controller-runtime"
+ "sigs.k8s.io/controller-runtime/pkg/webhook/admission"
+)
+
+func SetupStatefulSetWebhookWithManager(mgr ctrl.Manager) error {
+ return ctrl.NewWebhookManagedBy(mgr).
+ For(&appsv1.StatefulSet{}).
+ WithDefaulter(&StatefulSetDefaulter{}).
+ Complete()
+}
+
+//+kubebuilder:webhook:path=/mutate-apps-v1-statefulset,mutating=true,failurePolicy=fail,sideEffects=None,groups=apps,resources=statefulsets,verbs=create;update,versions=v1,name=statefulset.kb.io,admissionReviewVersions=v1
+
+type StatefulSetDefaulter struct{}
+
+var _ admission.CustomDefaulter = &StatefulSetDefaulter{}
+
+// Default implements webhook.Defaulter so a webhook will be registered for the type
+func (*StatefulSetDefaulter) Default(ctx context.Context, obj runtime.Object) error {
+ sts, ok := obj.(*appsv1.StatefulSet)
+ if !ok {
+ return fmt.Errorf("unknown obj type %T", obj)
+ }
+
+ req, err := admission.RequestFromContext(ctx)
+ if err != nil {
+ return fmt.Errorf("failed to get admission request from context: %w", err)
+ }
+
+ if req.Operation != admissionv1.Update {
+ return nil
+ }
+
+ if len(sts.OwnerReferences) != 1 {
+ return nil
+ }
+
+ if sts.OwnerReferences[0].Kind != "MySQLCluster" && sts.OwnerReferences[0].APIVersion != GroupVersion.String() {
+ return nil
+ }
+
+ if sts.Annotations[constants.AnnForceRollingUpdate] == "true" {
+ sts.Spec.UpdateStrategy.RollingUpdate = nil
+ return nil
+ }
+
+ if sts.Spec.UpdateStrategy.RollingUpdate == nil || sts.Spec.UpdateStrategy.RollingUpdate.Partition == nil {
+ sts.Spec.UpdateStrategy.RollingUpdate = &appsv1.RollingUpdateStatefulSetStrategy{
+ Partition: ptr.To[int32](*sts.Spec.Replicas),
+ }
+ return nil
+ }
+
+ oldSts, err := readStatefulSet(req.OldObject.Raw)
+ if err != nil {
+ return fmt.Errorf("failed to read old statefulset: %w", err)
+ }
+
+ partition := *sts.Spec.UpdateStrategy.RollingUpdate.Partition
+ oldPartition := *oldSts.Spec.UpdateStrategy.RollingUpdate.Partition
+
+ newSts := sts.DeepCopy()
+ newSts.Spec.UpdateStrategy = oldSts.Spec.UpdateStrategy
+
+ if partition != oldPartition && equality.Semantic.DeepEqual(newSts.Spec, oldSts.Spec) {
+ return nil
+ }
+
+ sts.Spec.UpdateStrategy.RollingUpdate = &appsv1.RollingUpdateStatefulSetStrategy{
+ Partition: ptr.To[int32](*sts.Spec.Replicas),
+ }
+
+ return nil
+}
+
+func readStatefulSet(raw []byte) (*appsv1.StatefulSet, error) {
+ var sts appsv1.StatefulSet
+
+ if _, _, err := unstructured.UnstructuredJSONScheme.Decode(raw, nil, &sts); err != nil {
+ return nil, err
+ }
+
+ sts.TypeMeta.APIVersion = appsv1.SchemeGroupVersion.Group + "/" + appsv1.SchemeGroupVersion.Version
+
+ return &sts, nil
+}

api/v1beta2/statefulset_webhhok_test.go

@@ -0,0 +1,170 @@
+package v1beta2_test
+
+import (
+ "context"
+
+ . "github.com/onsi/ginkgo"
+ . "github.com/onsi/gomega"
+ appsv1 "k8s.io/api/apps/v1"
+ corev1 "k8s.io/api/core/v1"
+ apierrors "k8s.io/apimachinery/pkg/api/errors"
+ metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+ "k8s.io/utils/ptr"
+ "sigs.k8s.io/controller-runtime/pkg/client"
+
+ "github.com/cybozu-go/moco/pkg/constants"
+)
+
+func makeStatefulSet() *appsv1.StatefulSet {
+ return &appsv1.StatefulSet{
+ ObjectMeta: metav1.ObjectMeta{
+ Name: "test",
+ Namespace: "default",
+ OwnerReferences: []metav1.OwnerReference{
+ {
+ APIVersion: "moco.cybozu.com/v1beta2",
+ Kind: "MySQLCluster",
+ Name: "test",
+ UID: "uid",
+ },
+ },
+ },
+ Spec: appsv1.StatefulSetSpec{
+ Replicas: ptr.To[int32](3),
+ Selector: &metav1.LabelSelector{MatchLabels: map[string]string{"foo": "bar"}},
+ UpdateStrategy: appsv1.StatefulSetUpdateStrategy{
+ Type: appsv1.RollingUpdateStatefulSetStrategyType,
+ },
+ Template: corev1.PodTemplateSpec{
+ ObjectMeta: metav1.ObjectMeta{
+ Labels: map[string]string{"foo": "bar"},
+ },
+ Spec: corev1.PodSpec{
+ Containers: []corev1.Container{
+ {
+ Name: "mysql",
+ Image: "mysql:examle",
+ },
+ },
+ },
+ },
+ },
+ }
+}
+
+func deleteStatefulSet() error {
+ r := &appsv1.StatefulSet{}
+ err := k8sClient.Get(ctx, client.ObjectKey{Namespace: "default", Name: "test"}, r)
+ if apierrors.IsNotFound(err) {
+ return nil
+ }
+
+ if err != nil {
+ return err
+ }
+
+ r.Finalizers = nil
+ if err := k8sClient.Update(ctx, r); err != nil {
+ return err
+ }
+
+ if err := k8sClient.Delete(ctx, r); err != nil {
+ return err
+ }
+
+ return nil
+}
+
+var _ = Describe("MySQLCluster Webhook", func() {
+ ctx := context.TODO()
+
+ BeforeEach(func() {
+ err := deleteStatefulSet()
+ Expect(err).NotTo(HaveOccurred())
+ })
+
+ It("should not set partition when creating StatefulSet", func() {
+ r := makeStatefulSet()
+ err := k8sClient.Create(ctx, r)
+ Expect(err).NotTo(HaveOccurred())
+ Expect(r.Spec.UpdateStrategy.RollingUpdate).To(BeNil())
+ })
+
+ It("should set partition when updating StatefulSet", func() {
+ r := makeStatefulSet()
+ err := k8sClient.Create(ctx, r)
+ Expect(err).NotTo(HaveOccurred())
+
+ err = k8sClient.Update(ctx, r)
+ Expect(err).NotTo(HaveOccurred())
+ Expect(r.Spec.UpdateStrategy.RollingUpdate.Partition).To(Equal(r.Spec.Replicas))
+ })
+
+ It("should not set partition when forcing updating StatefulSet", func() {
+ r := makeStatefulSet()
+ err := k8sClient.Create(ctx, r)
+ Expect(err).NotTo(HaveOccurred())
+
+ r.Annotations = map[string]string{constants.AnnForceRollingUpdate: "true"}
+ err = k8sClient.Update(ctx, r)
+ Expect(err).NotTo(HaveOccurred())
+ Expect(r.Spec.UpdateStrategy.RollingUpdate).To(BeNil())
+ })
+
+ It("should set partition when forcing updating StatefulSet with invalid value", func() {
+ r := makeStatefulSet()
+ err := k8sClient.Create(ctx, r)
+ Expect(err).NotTo(HaveOccurred())
+
+ r.Annotations = map[string]string{constants.AnnForceRollingUpdate: "false"}
+ err = k8sClient.Update(ctx, r)
+ Expect(err).NotTo(HaveOccurred())
+ Expect(r.Spec.UpdateStrategy.RollingUpdate.Partition).To(Equal(r.Spec.Replicas))
+ })
+
+ It("should not update partition when updating StatefulSet with only partition changed", func() {
+ r := makeStatefulSet()
+ err := k8sClient.Create(ctx, r)
+ Expect(err).NotTo(HaveOccurred())
+
+ err = k8sClient.Update(ctx, r)
+ Expect(err).NotTo(HaveOccurred())
+ Expect(r.Spec.UpdateStrategy.RollingUpdate.Partition).To(Equal(r.Spec.Replicas))
+
+ r.Spec.UpdateStrategy.RollingUpdate.Partition = ptr.To[int32](2)
+ err = k8sClient.Update(ctx, r)
+ Expect(err).NotTo(HaveOccurred())
+ Expect(r.Spec.UpdateStrategy.RollingUpdate.Partition).To(Equal(ptr.To[int32](2)))
+ })
+
+ It("should update partition when updating StatefulSet with partition and same field changed", func() {
+ r := makeStatefulSet()
+ err := k8sClient.Create(ctx, r)
+ Expect(err).NotTo(HaveOccurred())
+
+ err = k8sClient.Update(ctx, r)
+ Expect(err).NotTo(HaveOccurred())
+ Expect(r.Spec.UpdateStrategy.RollingUpdate.Partition).To(Equal(r.Spec.Replicas))
+
+ r.Spec.Replicas = ptr.To[int32](5)
+ r.Spec.UpdateStrategy.RollingUpdate.Partition = ptr.To[int32](2)
+ err = k8sClient.Update(ctx, r)
+ Expect(err).NotTo(HaveOccurred())
+ Expect(r.Spec.UpdateStrategy.RollingUpdate.Partition).To(Equal(ptr.To[int32](5)))
+ })
+
+ It("should update partition when updating StatefulSet with partition unchanged", func() {
+ r := makeStatefulSet()
+ err := k8sClient.Create(ctx, r)
+ Expect(err).NotTo(HaveOccurred())
+
+ err = k8sClient.Update(ctx, r)
+ Expect(err).NotTo(HaveOccurred())
+ Expect(r.Spec.UpdateStrategy.RollingUpdate.Partition).To(Equal(r.Spec.Replicas))
+
+ r.Spec.Replicas = ptr.To[int32](5)
+ err = k8sClient.Update(ctx, r)
+ Expect(err).NotTo(HaveOccurred())
+ Expect(r.Spec.UpdateStrategy.RollingUpdate.Partition).To(Equal(ptr.To[int32](5)))
+ })
+})

api/v1beta2/webhook_suite_test.go

@@ -92,6 +92,8 @@ var _ = BeforeSuite(func() {
  Expect(err).NotTo(HaveOccurred())
  err = (&mocov1beta2.BackupPolicy{}).SetupWebhookWithManager(mgr)
  Expect(err).NotTo(HaveOccurred())
+ err = mocov1beta2.SetupStatefulSetWebhookWithManager(mgr)
+ Expect(err).NotTo(HaveOccurred())
 
  //+kubebuilder:scaffold:webhook
 

charts/moco/templates/generated/generated.yaml

@@ -155,6 +155,12 @@ rules:
  - patch
  - update
  - watch
+ - apiGroups:
+ - ""
+ resources:
+ - pods/status
+ verbs:
+ - get
  - apiGroups:
  - ""
  resources:
@@ -460,6 +466,26 @@ webhooks:
  resources:
  - mysqlclusters
  sideEffects: None
+ - admissionReviewVersions:
+ - v1
+ clientConfig:
+ service:
+ name: moco-webhook-service
+ namespace: '{{ .Release.Namespace }}'
+ path: /mutate-apps-v1-statefulset
+ failurePolicy: Fail
+ name: statefulset.kb.io
+ rules:
+ - apiGroups:
+ - apps
+ apiVersions:
+ - v1
+ operations:
+ - CREATE
+ - UPDATE
+ resources:
+ - statefulsets
+ sideEffects: None
 ---
 apiVersion: admissionregistration.k8s.io/v1
 kind: ValidatingWebhookConfiguration

cmd/moco-controller/cmd/run.go

@@ -113,6 +113,15 @@ func subMain(ns, addr string, port int) error {
  return err
  }
 
+ if err = (&controllers.StatefulSetPartitionReconciler{
+ Client: mgr.GetClient(),
+ Recorder: mgr.GetEventRecorderFor("moco-controller"),
+ MaxConcurrentReconciles: config.maxConcurrentReconciles,
+ }).SetupWithManager(mgr); err != nil {
+ setupLog.Error(err, "unable to create controller", "controller", "Partition")
+ return err
+ }
+
  if err = (&controllers.PodWatcher{
  Client: mgr.GetClient(),
  ClusterManager: clusterMgr,
@@ -132,6 +141,11 @@ func subMain(ns, addr string, port int) error {
  return err
  }
 
+ if err := mocov1beta2.SetupStatefulSetWebhookWithManager(mgr); err != nil {
+ setupLog.Error(err, "unable to setup webhook", "webhook", "StatefulSet")
+ return err
+ }
+
  if err := mgr.AddHealthzCheck("health", healthz.Ping); err != nil {
  setupLog.Error(err, "unable to set up health check")
  return err

config/rbac/role.yaml

@@ -50,6 +50,12 @@ rules:
  - patch
  - update
  - watch
+- apiGroups:
+ - ""
+ resources:
+ - pods/status
+ verbs:
+ - get
 - apiGroups:
  - ""
  resources: