controllers: cleanup component manager

 - split component manager construction and fetching status
 - cleanup and describe logic
 - cleanup components state helpers

ComponentState:
"Ready"      // Running, reconciliation is not required
"Blocked"    // Reconciliation is impossible
"Pending"    // Reconciliation is possible
"NeedUpdate" // Running, update is required
"Updating"   // Update in progress

ComponentManagerStatus:
allReady    (was !needSync)
allRunning  (was !needInit)
allReadyOrUpdating

Update state machine uses these aggregates for making decisions,
and performs actions according to update plan.

Signed-off-by: Konstantin Khlebnikov <[email protected]>

Author: koct9i

controllers/component_manager.go

@@ -22,18 +22,18 @@ type ComponentManager struct {
 }
 
 type ComponentManagerStatus struct {
-	needSync           bool
-	needInit           bool
-	needUpdate         []components.Component
-	allReadyOrUpdating bool
+	allReady           bool // All components are Ready - no reconciliations required
+	allRunning         bool // All components are Ready or NeedUpdate - can start updates
+	allReadyOrUpdating bool // All components are Ready or Updating - no new updates
+
+	needUpdate []components.Component // Components in state NeedUpdate
 }
 
 func NewComponentManager(
 	ctx context.Context,
 	ytsaurus *apiProxy.Ytsaurus,
 	clusterDomain string,
 ) (*ComponentManager, error) {
-	logger := log.FromContext(ctx)
 	resource := ytsaurus.GetResource()
 
 	if clusterDomain == "" {
@@ -164,79 +164,87 @@ func NewComponentManager(
 	tt := components.NewTimbertruck(cfgen, ytsaurus, tnds, yc)
 	allComponents = append(allComponents, tt)
 
+	return &ComponentManager{
+		ytsaurus:      ytsaurus,
+		allComponents: allComponents,
+	}, nil
+}
+
+func (cm *ComponentManager) FetchStatus(ctx context.Context) error {
+	logger := log.FromContext(ctx)
+	resource := cm.ytsaurus.GetResource()
+
 	// Fetch component status.
 	var readyComponents []string
+	var needUpdateComponents []string
 	var updatingComponents []string
 	var notReadyComponents []string
 
-	status := ComponentManagerStatus{
-		needInit:           false,
-		needSync:           false,
-		needUpdate:         nil,
+	cm.status = ComponentManagerStatus{
+		allReady:           true,
+		allRunning:         true,
 		allReadyOrUpdating: true,
+		needUpdate:         nil,
 	}
-	for _, c := range allComponents {
-		err := c.Fetch(ctx)
-		if err != nil {
-			logger.Error(err, "failed to fetch status for controller", "component", c.GetFullName())
-			return nil, err
-		}
 
-		componentStatus, err := c.Status(ctx)
+	for _, component := range cm.allComponents {
+		err := component.Fetch(ctx)
 		if err != nil {
-			return nil, fmt.Errorf("failed to get component %s status: %w", c.GetFullName(), err)
+			return fmt.Errorf("failed to fetch component %s: %w", component.GetFullName(), err)
 		}
 
-		c.SetReadyCondition(componentStatus)
-		syncStatus := componentStatus.SyncStatus
-
-		if syncStatus == components.SyncStatusNeedLocalUpdate {
-			status.needUpdate = append(status.needUpdate, c)
+		status, err := component.Status(ctx)
+		if err != nil {
+			return fmt.Errorf("failed to get component %s status: %w", component.GetFullName(), err)
 		}
 
-		if !components.IsRunningStatus(syncStatus) {
-			if ytsaurus.GetClusterState() == ytv1.ClusterStateRunning {
-				msg := fmt.Sprintf("component `%s` status is neither Ready nor NeedLocalUpdate, but `%s`, "+
-					"needInit=true will be set, which will lead to Reconfiguration since cluster is in Running state",
-					c.GetFullName(), syncStatus,
-				)
-				logger.Info(msg,
-					"component", c.GetFullName(),
-					"syncStatus", syncStatus,
+		component.SetReadyCondition(status)
+		logger.Info("Component status",
+			"component", component.GetFullName(),
+			"status", status.SyncStatus,
+			"message", status.Message,
+		)
+
+		switch status.SyncStatus {
+		case components.SyncStatusReady:
+			readyComponents = append(readyComponents, component.GetFullName())
+		case components.SyncStatusNeedUpdate:
+			needUpdateComponents = append(needUpdateComponents, component.GetFullName())
+			cm.status.needUpdate = append(cm.status.needUpdate, component)
+			cm.status.allReady = false
+			cm.status.allReadyOrUpdating = false
+		case components.SyncStatusUpdating:
+			updatingComponents = append(updatingComponents, component.GetFullName())
+			cm.status.allReady = false
+			cm.status.allRunning = false
+		default:
+			notReadyComponents = append(notReadyComponents, component.GetFullName())
+			cm.status.allReady = false
+			cm.status.allRunning = false
+			cm.status.allReadyOrUpdating = false
+			if cm.ytsaurus.GetClusterState() == ytv1.ClusterStateRunning {
+				logger.Info("Cluster needs reconfiguration because component is not running",
+					"component", component.GetFullName(),
+					"status", status.SyncStatus,
+					"message", status.Message,
 				)
 			}
-			status.needInit = true
-		}
-
-		if syncStatus != components.SyncStatusReady && syncStatus != components.SyncStatusUpdating {
-			status.allReadyOrUpdating = false
-		}
-
-		if syncStatus != components.SyncStatusReady {
-			logger.Info("component is not ready", "component", c.GetFullName(), "syncStatus", syncStatus, "message", componentStatus.Message)
-			notReadyComponents = append(notReadyComponents, c.GetFullName())
-			status.needSync = true
-		} else {
-			readyComponents = append(readyComponents, c.GetFullName())
-		}
-
-		if syncStatus == components.SyncStatusUpdating {
-			updatingComponents = append(updatingComponents, c.GetFullName())
 		}
 	}
 
 	logger.Info("Ytsaurus sync status",
+		"clusterState", resource.Status.State,
+		"updateState", resource.Status.UpdateStatus.State,
+		"allReady", cm.status.allReady,
+		"allRunning", cm.status.allRunning,
+		"allReadyOrUpdating", cm.status.allReadyOrUpdating,
+		"needUpdateComponents", needUpdateComponents,
+		"updatingComponents", updatingComponents,
 		"notReadyComponents", notReadyComponents,
 		"readyComponents", readyComponents,
-		"updatingComponents", updatingComponents,
-		"updateState", resource.Status.UpdateStatus.State,
-		"clusterState", resource.Status.State)
+	)
 
-	return &ComponentManager{
-		ytsaurus:      ytsaurus,
-		allComponents: allComponents,
-		status:        status,
-	}, nil
+	return nil
 }
 
 func (cm *ComponentManager) Sync(ctx context.Context) (ctrl.Result, error) {
@@ -273,22 +281,6 @@ func (cm *ComponentManager) Sync(ctx context.Context) (ctrl.Result, error) {
 	return ctrl.Result{RequeueAfter: time.Second}, nil
 }
 
-func (cm *ComponentManager) needSync() bool {
-	return cm.status.needSync
-}
-
-func (cm *ComponentManager) needInit() bool {
-	return cm.status.needInit
-}
-
-func (cm *ComponentManager) needUpdate() []components.Component {
-	return cm.status.needUpdate
-}
-
-func (cm *ComponentManager) allReadyOrUpdating() bool {
-	return cm.status.allReadyOrUpdating
-}
-
 func (cm *ComponentManager) arePodsRemoved() bool {
 	for _, cmp := range cm.allComponents {
 		if cmp.GetType() == consts.YtsaurusClientType {

controllers/sync.go

@@ -151,6 +151,13 @@ func (r *YtsaurusReconciler) Sync(ctx context.Context, resource *ytv1.Ytsaurus)
 	ytsaurus := apiProxy.NewYtsaurus(resource, r.Client, r.Recorder, r.Scheme)
 	componentManager, err := NewComponentManager(ctx, ytsaurus, r.ClusterDomain)
 	if err != nil {
+		logger.Error(err, "Cannot build component manager")
+		return ctrl.Result{Requeue: true}, err
+	}
+
+	err = componentManager.FetchStatus(ctx)
+	if err != nil {
+		logger.Error(err, "Cannot fetch component manager status")
 		return ctrl.Result{Requeue: true}, err
 	}
 
@@ -162,14 +169,14 @@ func (r *YtsaurusReconciler) Sync(ctx context.Context, resource *ytv1.Ytsaurus)
 
 	case ytv1.ClusterStateInitializing:
 		// Ytsaurus has finished initializing, and is running now.
-		if !componentManager.needSync() {
+		if componentManager.status.allRunning {
 			logger.Info("Ytsaurus has synced and is running now")
 			err := ytsaurus.SaveClusterState(ctx, ytv1.ClusterStateRunning)
 			return ctrl.Result{Requeue: true}, err
 		}
 
 	case ytv1.ClusterStateRunning:
-		needUpdate := componentManager.needUpdate()
+		needUpdate := componentManager.status.needUpdate
 		canUpdate, cannotUpdate := chooseUpdatingComponents(
 			ytsaurus.GetResource().Spec,
 			convertToComponent(needUpdate),
@@ -186,10 +193,10 @@ func (r *YtsaurusReconciler) Sync(ctx context.Context, resource *ytv1.Ytsaurus)
 		}
 
 		switch {
-		case !componentManager.needSync():
+		case componentManager.status.allReady:
 			logger.Info("Ytsaurus is running and happy")
 
-		case componentManager.needInit():
+		case !componentManager.status.allRunning:
 			logger.Info("Ytsaurus needs initialization of some components")
 			err := ytsaurus.SaveClusterState(ctx, ytv1.ClusterStateReconfiguration)
 			return ctrl.Result{Requeue: true}, err
@@ -259,7 +266,7 @@ func (r *YtsaurusReconciler) Sync(ctx context.Context, resource *ytv1.Ytsaurus)
 		return ctrl.Result{Requeue: true}, err
 
 	case ytv1.ClusterStateReconfiguration:
-		if !componentManager.needInit() {
+		if componentManager.status.allRunning {
 			logger.Info("Ytsaurus has reconfigured and is running now")
 			err := ytsaurus.SaveClusterState(ctx, ytv1.ClusterStateRunning)
 			// Requeue once again to do final check and maybe update observed generation.

controllers/update_flow_steps.go

@@ -181,7 +181,7 @@ var flowConditions = map[ytv1.UpdateState]flowCondition{
 		return stepResultMarkUnsatisfied
 	},
 	ytv1.UpdateStateImpossibleToStart: func(ctx context.Context, ytsaurus *apiProxy.Ytsaurus, componentManager *ComponentManager) stepResultMark {
-		if !componentManager.needSync() || !ytsaurus.GetResource().Spec.EnableFullUpdate {
+		if componentManager.status.allReady || !ytsaurus.GetResource().Spec.EnableFullUpdate {
 			return stepResultMarkHappy
 		}
 		return stepResultMarkUnsatisfied
@@ -214,7 +214,7 @@ var flowConditions = map[ytv1.UpdateState]flowCondition{
 		return stepResultMarkUnsatisfied
 	},
 	ytv1.UpdateStateWaitingForPodsCreation: func(ctx context.Context, ytsaurus *apiProxy.Ytsaurus, componentManager *ComponentManager) stepResultMark {
-		if componentManager.allReadyOrUpdating() {
+		if componentManager.status.allReadyOrUpdating {
 			return stepResultMarkHappy
 		}
 		return stepResultMarkUnsatisfied

pkg/components/bundle_controller.go

@@ -54,7 +54,7 @@ func (bc *BundleController) doSync(ctx context.Context, dry bool) (ComponentStat
 	var err error
 
 	if ytv1.IsReadyToUpdateClusterState(bc.ytsaurus.GetClusterState()) && bc.server.needUpdate() {
-		return SimpleStatus(SyncStatusNeedLocalUpdate), err
+		return SimpleStatus(SyncStatusNeedUpdate), err
 	}
 
 	if bc.ytsaurus.GetClusterState() == ytv1.ClusterStateUpdating {
@@ -67,11 +67,11 @@ func (bc *BundleController) doSync(ctx context.Context, dry bool) (ComponentStat
 		if !dry {
 			err = bc.doServerSync(ctx)
 		}
-		return WaitingStatus(SyncStatusPending, "components"), err
+		return ComponentStatusWaitingFor("components"), err
 	}
 
 	if !bc.server.arePodsReady(ctx) {
-		return WaitingStatus(SyncStatusBlocked, "pods"), err
+		return ComponentStatusBlockedBy("pods"), err
 	}
 
 	return SimpleStatus(SyncStatusReady), err

pkg/components/chyt.go

@@ -137,7 +137,7 @@ func (c *Chyt) doSync(ctx context.Context, dry bool) (ComponentStatus, error) {
 	var err error
 
 	if c.ytsaurus.Status.State != ytv1.ClusterStateRunning {
-		return WaitingStatus(SyncStatusBlocked, "ytsaurus running"), err
+		return ComponentStatusBlockedBy("ytsaurus is not running"), err
 	}
 
 	// Create a user for chyt initialization.
@@ -150,7 +150,7 @@ func (c *Chyt) doSync(ctx context.Context, dry bool) (ComponentStatus, error) {
 			err = c.secret.Sync(ctx)
 		}
 		c.chyt.GetResource().Status.ReleaseStatus = ytv1.ChytReleaseStatusCreatingUserSecret
-		return WaitingStatus(SyncStatusPending, c.secret.Name()), err
+		return ComponentStatusWaitingFor(c.secret.Name()), err
 	}
 
 	if !dry {
@@ -200,7 +200,7 @@ func (c *Chyt) doSync(ctx context.Context, dry bool) (ComponentStatus, error) {
 
 	c.chyt.GetResource().Status.ReleaseStatus = ytv1.ChytReleaseStatusFinished
 
-	return SimpleStatus(SyncStatusReady), err
+	return ComponentStatusReady(), err
 }
 
 func (c *Chyt) Fetch(ctx context.Context) error {

pkg/components/component.go

@@ -15,30 +15,59 @@ import (
 type SyncStatus string
 
 const (
-	SyncStatusBlocked         SyncStatus = "Blocked"
-	SyncStatusNeedLocalUpdate SyncStatus = "NeedLocalUpdate"
-	SyncStatusPending         SyncStatus = "Pending"
-	SyncStatusReady           SyncStatus = "Ready"
-	SyncStatusUpdating        SyncStatus = "Updating"
+	SyncStatusReady      SyncStatus = "Ready"      // Running, reconciliation is not required
+	SyncStatusBlocked    SyncStatus = "Blocked"    // Reconciliation is impossible
+	SyncStatusPending    SyncStatus = "Pending"    // Reconciliation is possible
+	SyncStatusNeedUpdate SyncStatus = "NeedUpdate" // Running, update is required
+	SyncStatusUpdating   SyncStatus = "Updating"   // Update in progress
 )
 
-func IsRunningStatus(status SyncStatus) bool {
-	return status == SyncStatusReady || status == SyncStatusNeedLocalUpdate
-}
-
 type ComponentStatus struct {
 	SyncStatus SyncStatus
 	Message    string
 }
 
-func NewComponentStatus(status SyncStatus, message string) ComponentStatus {
-	return ComponentStatus{status, message}
+func (cs ComponentStatus) IsRunning() bool {
+	return cs.SyncStatus == SyncStatusReady || cs.SyncStatus == SyncStatusNeedUpdate
+}
+
+func ComponentStatusReady() ComponentStatus {
+	return ComponentStatus{SyncStatusReady, "Ready"}
+}
+
+func ComponentStatusReadyAfter(message string) ComponentStatus {
+	return ComponentStatus{SyncStatusReady, message}
+}
+
+func ComponentStatusBlocked(message string) ComponentStatus {
+	return ComponentStatus{SyncStatusBlocked, message}
+}
+
+func ComponentStatusPending(message string) ComponentStatus {
+	return ComponentStatus{SyncStatusPending, message}
+}
+
+func ComponentStatusNeedUpdate(message string) ComponentStatus {
+	return ComponentStatus{SyncStatusNeedUpdate, message}
+}
+
+func ComponentStatusUpdating(message string) ComponentStatus {
+	return ComponentStatus{SyncStatusUpdating, message}
+}
+
+func ComponentStatusBlockedBy(cause string) ComponentStatus {
+	return ComponentStatus{SyncStatusBlocked, fmt.Sprintf("Blocked by %s", cause)}
+}
+
+func ComponentStatusWaitingFor(event string) ComponentStatus {
+	return ComponentStatus{SyncStatusPending, fmt.Sprintf("Waiting for %s", event)}
 }
 
-func WaitingStatus(status SyncStatus, event string) ComponentStatus {
-	return ComponentStatus{status, fmt.Sprintf("Wait for %s", event)}
+func ComponentStatusUpdateStep(part string) ComponentStatus {
+	return ComponentStatus{SyncStatusUpdating, fmt.Sprintf("Updating %s", part)}
 }
 
+// TODO(khlebnikov): Replace this stub with status with meaningful message.
 func SimpleStatus(status SyncStatus) ComponentStatus {
 	return ComponentStatus{status, string(status)}
 }