Stop reconciliation and clustering (#578)

* Stop reconciliation and clustering

Signed-off-by: d-kuro <[email protected]>

api/v1beta2/mysqlcluster_types.go

@@ -602,11 +602,13 @@ type MySQLClusterStatus struct {
 }
 
 const (
- ConditionInitialized string = "Initialized"
- ConditionAvailable string = "Available"
- ConditionHealthy string = "Healthy"
- ConditionStatefulSetReady string = "StatefulSetReady"
- ConditionReconcileSuccess string = "ReconcileSuccess"
+ ConditionInitialized string = "Initialized"
+ ConditionAvailable string = "Available"
+ ConditionHealthy string = "Healthy"
+ ConditionStatefulSetReady string = "StatefulSetReady"
+ ConditionReconcileSuccess string = "ReconcileSuccess"
+ ConditionReconciliationActive string = "ReconciliationActive"
+ ConditionClusteringActive string = "ClusteringActive"
 )
 
 // BackupStatus represents the status of the last successful backup.
@@ -669,6 +671,8 @@ type ReconcileInfo struct {
 // +kubebuilder:printcolumn:name="Primary",type="integer",JSONPath=".status.currentPrimaryIndex"
 // +kubebuilder:printcolumn:name="Synced replicas",type="integer",JSONPath=".status.syncedReplicas"
 // +kubebuilder:printcolumn:name="Errant replicas",type="integer",JSONPath=".status.errantReplicas"
+// +kubebuilder:printcolumn:name="Clustering Active",type="string",JSONPath=".status.conditions[?(@.type=='ClusteringActive')].status"
+// +kubebuilder:printcolumn:name="Reconcile Active",type="string",JSONPath=".status.conditions[?(@.type=='ReconciliationActive')].status"
 // +kubebuilder:printcolumn:name="Last backup",type="string",JSONPath=".status.backup.time"
 
 // MySQLCluster is the Schema for the mysqlclusters API

charts/moco/templates/generated/crds/moco_crds.yaml

@@ -2099,6 +2099,12 @@ spec:
  - jsonPath: .status.errantReplicas
  name: Errant replicas
  type: integer
+ - jsonPath: .status.conditions[?(@.type=='ClusteringActive')].status
+ name: Clustering Active
+ type: string
+ - jsonPath: .status.conditions[?(@.type=='ReconciliationActive')].status
+ name: Reconcile Active
+ type: string
  - jsonPath: .status.backup.time
  name: Last backup
  type: string

clustering/manager.go

@@ -27,6 +27,7 @@ type ClusterManager interface {
  UpdateNoStart(types.NamespacedName, string)
  Stop(types.NamespacedName)
  StopAll()
+ Pause(types.NamespacedName)
 }
 
 func NewClusterManager(interval time.Duration, m manager.Manager, opf dbop.OperatorFactory, af AgentFactory, log logr.Logger) ClusterManager {
@@ -122,3 +123,17 @@ func (m *clusterManager) StopAll() {
  m.wg.Wait()
  m.stopped = true
 }
+
+// Pause halts the manager process for the cluster.
+// Unlike Stop, the metrics held by the manager are not deleted.
+func (m *clusterManager) Pause(name types.NamespacedName) {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+
+ key := name.String()
+ p, ok := m.processes[key]
+ if ok {
+ p.Pause()
+ delete(m.processes, key)
+ }
+}

clustering/process.go

@@ -3,6 +3,7 @@ package clustering
 import (
  "context"
  "fmt"
+ "math"
  "time"
 
  mocov1beta2 "github.com/cybozu-go/moco/api/v1beta2"
@@ -49,10 +50,12 @@ type managerProcess struct {
  agentf AgentFactory
  name types.NamespacedName
  cancel func()
+ pause bool
 
  ch chan string
  metrics metricsSet
  deleteMetrics func()
+ pauseMetrics func()
 }
 
 func newManagerProcess(c client.Client, r client.Reader, recorder record.EventRecorder, dbf dbop.OperatorFactory, agentf AgentFactory, name types.NamespacedName, cancel func()) *managerProcess {
@@ -101,6 +104,12 @@ func newManagerProcess(c client.Client, r client.Reader, recorder record.EventRe
  metrics.BackupWorkDirUsage.DeleteLabelValues(name.Name, name.Namespace)
  metrics.BackupWarnings.DeleteLabelValues(name.Name, name.Namespace)
  },
+ pauseMetrics: func() {
+ metrics.AvailableVec.WithLabelValues(name.Name, name.Namespace).Set(math.NaN())
+ metrics.HealthyVec.WithLabelValues(name.Name, name.Namespace).Set(math.NaN())
+ metrics.ReadyReplicasVec.WithLabelValues(name.Name, name.Namespace).Set(math.NaN())
+ metrics.ErrantReplicasVec.WithLabelValues(name.Name, name.Namespace).Set(math.NaN())
+ },
  }
 }
 
@@ -115,10 +124,22 @@ func (p *managerProcess) Cancel() {
  p.cancel()
 }
 
+// Pause pauses the manager process.
+// Unlike Cancel, it does not delete metrics.
+// Also, it sets NaN to some metrics.
+func (p *managerProcess) Pause() {
+ p.pause = true
+ p.cancel()
+}
+
 func (p *managerProcess) Start(ctx context.Context, rootLog logr.Logger, interval time.Duration) {
  tick := time.NewTicker(interval)
  defer func() {
  tick.Stop()
+ if p.pause {
+ p.pauseMetrics()
+ return
+ }
  p.deleteMetrics()
  }()
 
@@ -266,6 +287,15 @@ func (p *managerProcess) updateStatus(ctx context.Context, ss *StatusSet) error
  meta.SetStatusCondition(&cluster.Status.Conditions, updateCond(mocov1beta2.ConditionAvailable, available))
  meta.SetStatusCondition(&cluster.Status.Conditions, updateCond(mocov1beta2.ConditionHealthy, healthy))
 
+ meta.SetStatusCondition(&cluster.Status.Conditions,
+ metav1.Condition{
+ Type: mocov1beta2.ConditionClusteringActive,
+ Status: metav1.ConditionTrue,
+ Reason: "ClusteringActive",
+ Message: "clustering is active",
+ },
+ )
+
  if available == metav1.ConditionTrue {
  p.metrics.available.Set(1)
  } else {

cmd/kubectl-moco/cmd/start.go

@@ -0,0 +1,103 @@
+package cmd
+
+import (
+ "context"
+ "fmt"
+ "os"
+
+ mocov1beta2 "github.com/cybozu-go/moco/api/v1beta2"
+ "github.com/cybozu-go/moco/pkg/constants"
+ "github.com/spf13/cobra"
+ "k8s.io/apimachinery/pkg/api/equality"
+ "k8s.io/apimachinery/pkg/types"
+)
+
+func init() {
+ rootCmd.AddCommand(startCmd)
+ startCmd.AddCommand(startClusteringCmd)
+ startCmd.AddCommand(startReconciliationCmd)
+}
+
+var startCmd = &cobra.Command{
+ Use: "start",
+ Short: "Starts the MySQLCluster reconciliation or clustering",
+ Long: "The start command is used to start the reconciliation or clustering of MySQLCluster",
+}
+
+var startClusteringCmd = &cobra.Command{
+ Use: "clustering CLUSTER_NAME",
+ Short: "Start the specified MySQLCluster's clustering",
+ Long: "start clustering is a command to start the clustering of the specified MySQLCluster. It requires the cluster name as the parameter.",
+ Args: cobra.ExactArgs(1),
+ RunE: func(cmd *cobra.Command, args []string) error {
+ return startClustering(cmd.Context(), args[0])
+ },
+ ValidArgsFunction: func(cmd *cobra.Command, args []string, toComplete string) ([]string, cobra.ShellCompDirective) {
+ return mysqlClusterCandidates(cmd.Context(), cmd, args, toComplete)
+ },
+}
+
+func startClustering(ctx context.Context, name string) error {
+ cluster := &mocov1beta2.MySQLCluster{}
+ if err := kubeClient.Get(ctx, types.NamespacedName{Namespace: namespace, Name: name}, cluster); err != nil {
+ return err
+ }
+
+ orig := cluster.DeepCopy()
+
+ if ann, ok := cluster.Annotations[constants.AnnClusteringStopped]; ok && ann == "true" {
+ delete(cluster.Annotations, constants.AnnClusteringStopped)
+ }
+
+ if equality.Semantic.DeepEqual(orig, cluster) {
+ fmt.Fprintf(os.Stdout, "The clustering is already running.")
+ return nil
+ }
+
+ if err := kubeClient.Update(ctx, cluster); err != nil {
+ return fmt.Errorf("failed to start clustering of MySQLCluster: %w", err)
+ }
+
+ fmt.Fprintf(os.Stdout, "started clustering of MySQLCluster %q\n", fmt.Sprintf("%s/%s", namespace, name))
+
+ return nil
+}
+
+var startReconciliationCmd = &cobra.Command{
+ Use: "reconciliation CLUSTER_NAME",
+ Short: "Start the specified MySQLCluster's reconciliation",
+ Long: "start reconciliation is a command to start the reconciliation process for the specified MySQLCluster. This requires the cluster name as the parameter.",
+ Args: cobra.ExactArgs(1),
+ RunE: func(cmd *cobra.Command, args []string) error {
+ return startReconciliation(cmd.Context(), args[0])
+ },
+ ValidArgsFunction: func(cmd *cobra.Command, args []string, toComplete string) ([]string, cobra.ShellCompDirective) {
+ return mysqlClusterCandidates(cmd.Context(), cmd, args, toComplete)
+ },
+}
+
+func startReconciliation(ctx context.Context, name string) error {
+ cluster := &mocov1beta2.MySQLCluster{}
+ if err := kubeClient.Get(ctx, types.NamespacedName{Namespace: namespace, Name: name}, cluster); err != nil {
+ return err
+ }
+
+ orig := cluster.DeepCopy()
+
+ if ann, ok := cluster.Annotations[constants.AnnReconciliationStopped]; ok && ann == "true" {
+ delete(cluster.Annotations, constants.AnnReconciliationStopped)
+ }
+
+ if equality.Semantic.DeepEqual(orig, cluster) {
+ fmt.Fprintf(os.Stdout, "The reconciliation is already running.")
+ return nil
+ }
+
+ if err := kubeClient.Update(ctx, cluster); err != nil {
+ return fmt.Errorf("failed to start reconciliation of MySQLCluster: %w", err)
+ }
+
+ fmt.Fprintf(os.Stdout, "started reconciliation of MySQLCluster %q\n", fmt.Sprintf("%s/%s", namespace, name))
+
+ return nil
+}

cmd/kubectl-moco/cmd/stop.go

@@ -0,0 +1,105 @@
+package cmd
+
+import (
+ "context"
+ "fmt"
+ "os"
+
+ mocov1beta2 "github.com/cybozu-go/moco/api/v1beta2"
+ "github.com/cybozu-go/moco/pkg/constants"
+ "github.com/spf13/cobra"
+ "k8s.io/apimachinery/pkg/api/equality"
+ "k8s.io/apimachinery/pkg/types"
+)
+
+func init() {
+ rootCmd.AddCommand(stopCmd)
+ stopCmd.AddCommand(stopClusteringCmd)
+ stopCmd.AddCommand(stopReconciliationCmd)
+}
+
+var stopCmd = &cobra.Command{
+ Use: "stop",
+ Short: "Stops the MySQLCluster reconciliation or clustering",
+ Long: "The stop command is used to halt the reconciliation or clustering of MySQLCluster",
+}
+
+var stopClusteringCmd = &cobra.Command{
+ Use: "clustering CLUSTER_NAME",
+ Short: "Stop the specified MySQLCluster's clustering",
+ Long: "stop clustering is a command to stop the clustering of the specified MySQLCluster. It requires the cluster name as the parameter.",
+ Args: cobra.ExactArgs(1),
+ RunE: func(cmd *cobra.Command, args []string) error {
+ return stopClustering(cmd.Context(), args[0])
+ },
+ ValidArgsFunction: func(cmd *cobra.Command, args []string, toComplete string) ([]string, cobra.ShellCompDirective) {
+ return mysqlClusterCandidates(cmd.Context(), cmd, args, toComplete)
+ },
+}
+
+func stopClustering(ctx context.Context, name string) error {
+ cluster := &mocov1beta2.MySQLCluster{}
+ if err := kubeClient.Get(ctx, types.NamespacedName{Namespace: namespace, Name: name}, cluster); err != nil {
+ return err
+ }
+
+ orig := cluster.DeepCopy()
+
+ if cluster.Annotations == nil {
+ cluster.Annotations = make(map[string]string)
+ }
+ cluster.Annotations[constants.AnnClusteringStopped] = "true"
+
+ if equality.Semantic.DeepEqual(orig, cluster) {
+ fmt.Fprintf(os.Stdout, "The clustering is already stopped.")
+ return nil
+ }
+
+ if err := kubeClient.Update(ctx, cluster); err != nil {
+ return fmt.Errorf("failed to stop clustering of MySQLCluster: %w", err)
+ }
+
+ fmt.Fprintf(os.Stdout, "stopped clustering of MySQLCluster %q\n", fmt.Sprintf("%s/%s", namespace, name))
+
+ return nil
+}
+
+var stopReconciliationCmd = &cobra.Command{
+ Use: "reconciliation CLUSTER_NAME",
+ Short: "Stop the specified MySQLCluster's reconciliation",
+ Long: "stop reconciliation is a command to stop the reconciliation process for the specified MySQLCluster. This requires the cluster name as the parameter.",
+ Args: cobra.ExactArgs(1),
+ RunE: func(cmd *cobra.Command, args []string) error {
+ return stopReconciliation(cmd.Context(), args[0])
+ },
+ ValidArgsFunction: func(cmd *cobra.Command, args []string, toComplete string) ([]string, cobra.ShellCompDirective) {
+ return mysqlClusterCandidates(cmd.Context(), cmd, args, toComplete)
+ },
+}
+
+func stopReconciliation(ctx context.Context, name string) error {
+ cluster := &mocov1beta2.MySQLCluster{}
+ if err := kubeClient.Get(ctx, types.NamespacedName{Namespace: namespace, Name: name}, cluster); err != nil {
+ return err
+ }
+
+ orig := cluster.DeepCopy()
+
+ if cluster.Annotations == nil {
+ cluster.Annotations = make(map[string]string)
+ }
+ cluster.Annotations[constants.AnnReconciliationStopped] = "true"
+
+ if equality.Semantic.DeepEqual(orig, cluster) {
+ fmt.Fprintf(os.Stdout, "The reconciliation is already stopped.")
+ return nil
+ }
+
+ if err := kubeClient.Update(ctx, cluster); err != nil {
+ return fmt.Errorf("failed to stop reconciliation of MySQLCluster: %w", err)
+ }
+
+ fmt.Fprintf(os.Stdout, "stopped reconciliation of MySQLCluster %q\n", fmt.Sprintf("%s/%s", namespace, name))
+
+ return nil
+}

config/crd/bases/moco.cybozu.com_mysqlclusters.yaml

@@ -30,6 +30,12 @@ spec:
  - jsonPath: .status.errantReplicas
  name: Errant replicas
  type: integer
+ - jsonPath: .status.conditions[?(@.type=='ClusteringActive')].status
+ name: Clustering Active
+ type: string
+ - jsonPath: .status.conditions[?(@.type=='ReconciliationActive')].status
+ name: Reconcile Active
+ type: string
  - jsonPath: .status.backup.time
  name: Last backup
  type: string

config/crd/tests/apiextensions.k8s.io_v1_customresourcedefinition_mysqlclusters.moco.cybozu.com.yaml

@@ -30,6 +30,12 @@ spec:
  - jsonPath: .status.errantReplicas
  name: Errant replicas
  type: integer
+ - jsonPath: .status.conditions[?(@.type=='ClusteringActive')].status
+ name: Clustering Active
+ type: string
+ - jsonPath: .status.conditions[?(@.type=='ReconciliationActive')].status
+ name: Reconcile Active
+ type: string
  - jsonPath: .status.backup.time
  name: Last backup
  type: string

controllers/mock_test.go

@@ -38,6 +38,13 @@ func (m *mockManager) Stop(key types.NamespacedName) {
 
 func (m *mockManager) StopAll() {}
 
+func (m *mockManager) Pause(key types.NamespacedName) {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+
+ delete(m.clusters, key.String())
+}
+
 func (m *mockManager) getKeys() map[string]bool {
  m.mu.Lock()
  defer m.mu.Unlock()