fix(kafkatopic): wait service running

aiven · Jul 16, 2024 · 7afde9e · 7afde9e
1 parent e401591
commit 7afde9e
Show file tree

Hide file tree

Showing 15 changed files with 32 additions and 21 deletions.
diff --git a/controllers/clickhousedatabase_controller.go b/controllers/clickhousedatabase_controller.go
@@ -115,7 +115,7 @@ func (h *ClickhouseDatabaseHandler) checkPreconditions(ctx context.Context, avn
 	meta.SetStatusCondition(&db.Status.Conditions,
 		getInitializedCondition("Preconditions", "Checking preconditions"))
 
-	return checkServiceIsRunning(ctx, avn, avnGen, db.Spec.Project, db.Spec.ServiceName)
+	return checkServiceIsRunning(ctx, avnGen, db.Spec.Project, db.Spec.ServiceName)
 }
 
 func (h *ClickhouseDatabaseHandler) convert(i client.Object) (*v1alpha1.ClickhouseDatabase, error) {

diff --git a/controllers/clickhousegrant_controller.go b/controllers/clickhousegrant_controller.go
@@ -125,7 +125,7 @@ func (h *ClickhouseGrantHandler) checkPreconditions(ctx context.Context, avn *ai
 	meta.SetStatusCondition(&g.Status.Conditions,
 		getInitializedCondition("Preconditions", "Checking preconditions"))
 
-	serviceIsRunning, err := checkServiceIsRunning(ctx, avn, avnGen, g.Spec.Project, g.Spec.ServiceName)
+	serviceIsRunning, err := checkServiceIsRunning(ctx, avnGen, g.Spec.Project, g.Spec.ServiceName)
 	if !serviceIsRunning || err != nil {
 		return false, err
 	}

diff --git a/controllers/clickhouserole_controller.go b/controllers/clickhouserole_controller.go
@@ -111,7 +111,7 @@ func (h *clickhouseRoleHandler) checkPreconditions(ctx context.Context, avn *aiv
 	meta.SetStatusCondition(&role.Status.Conditions,
 		getInitializedCondition("Preconditions", "Checking preconditions"))
 
-	return checkServiceIsRunning(ctx, avn, avnGen, role.Spec.Project, role.Spec.ServiceName)
+	return checkServiceIsRunning(ctx, avnGen, role.Spec.Project, role.Spec.ServiceName)
 }
 
 func (h *clickhouseRoleHandler) convert(i client.Object) (*v1alpha1.ClickhouseRole, error) {

diff --git a/controllers/clickhouseuser_controller.go b/controllers/clickhouseuser_controller.go
@@ -156,7 +156,7 @@ func (h *clickhouseUserHandler) checkPreconditions(ctx context.Context, avn *aiv
 	meta.SetStatusCondition(&user.Status.Conditions,
 		getInitializedCondition("Preconditions", "Checking preconditions"))
 
-	return checkServiceIsRunning(ctx, avn, avnGen, user.Spec.Project, user.Spec.ServiceName)
+	return checkServiceIsRunning(ctx, avnGen, user.Spec.Project, user.Spec.ServiceName)
 }
 
 func (h *clickhouseUserHandler) convert(i client.Object) (*v1alpha1.ClickhouseUser, error) {

diff --git a/controllers/common.go b/controllers/common.go
@@ -48,7 +48,7 @@ const (
 
 var errTerminationProtectionOn = errors.New("termination protection is on")
 
-func checkServiceIsRunning(ctx context.Context, _ *aiven.Client, avnGen avngen.Client, project, serviceName string) (bool, error) {
+func checkServiceIsRunning(ctx context.Context, avnGen avngen.Client, project, serviceName string) (bool, error) {
 	s, err := avnGen.ServiceGet(ctx, project, serviceName)
 	if err != nil {
 		// if service is not found, it is not running
@@ -58,11 +58,11 @@ func checkServiceIsRunning(ctx context.Context, _ *aiven.Client, avnGen avngen.C
 		}
 		return false, err
 	}
-	return serviceIsRunning(s.State), nil
+	return serviceIsOperational(s.State), nil
 }
 
-// serviceIsRunning returns "true" when a service is in operational state, i.e. "running"
-func serviceIsRunning[T service.ServiceStateType | string](state T) bool {
+// serviceIsOperational returns "true" when a service is in operational state, i.e. "running"
+func serviceIsOperational[T service.ServiceStateType | string](state T) bool {
 	switch service.ServiceStateType(state) {
 	case service.ServiceStateTypeRunning, service.ServiceStateTypeRebalancing:
 		return true

diff --git a/controllers/connectionpool_controller.go b/controllers/connectionpool_controller.go
@@ -211,7 +211,7 @@ func (h ConnectionPoolHandler) checkPreconditions(ctx context.Context, avn *aive
 	meta.SetStatusCondition(&cp.Status.Conditions,
 		getInitializedCondition("Preconditions", "Checking preconditions"))
 
-	isRunning, err := checkServiceIsRunning(ctx, avn, avnGen, cp.Spec.Project, cp.Spec.ServiceName)
+	isRunning, err := checkServiceIsRunning(ctx, avnGen, cp.Spec.Project, cp.Spec.ServiceName)
 	if err != nil {
 		return false, err
 	}

diff --git a/controllers/database_controller.go b/controllers/database_controller.go
@@ -136,7 +136,7 @@ func (h DatabaseHandler) checkPreconditions(ctx context.Context, avn *aiven.Clie
 	meta.SetStatusCondition(&db.Status.Conditions,
 		getInitializedCondition("Preconditions", "Checking preconditions"))
 
-	return checkServiceIsRunning(ctx, avn, avnGen, db.Spec.Project, db.Spec.ServiceName)
+	return checkServiceIsRunning(ctx, avnGen, db.Spec.Project, db.Spec.ServiceName)
 }
 
 func (h DatabaseHandler) convert(i client.Object) (*v1alpha1.Database, error) {

diff --git a/controllers/generic_service_handler.go b/controllers/generic_service_handler.go
@@ -174,13 +174,13 @@ func (h *genericServiceHandler) get(ctx context.Context, avn *aiven.Client, avnG
 	}
 
 	spec := o.getServiceCommonSpec()
-	s, err := avn.Services.Get(ctx, spec.Project, o.getObjectMeta().Name)
+	s, err := avn.Services.Get(ctx, spec.Project, o.GetName())
 	if err != nil {
 		return nil, fmt.Errorf("failed to get service from Aiven: %w", err)
 	}
 
 	status := o.getServiceStatus()
-	if !serviceIsRunning(s.State) {
+	if !serviceIsOperational(s.State) {
 		status.State = s.State
 		return nil, nil
 	}
@@ -232,7 +232,7 @@ func (h *genericServiceHandler) checkPreconditions(ctx context.Context, avn *aiv
 		// Validates that read_replica is running
 		// If not, the wrapper controller will try later
 		if s.IntegrationType == "read_replica" {
-			r, err := checkServiceIsRunning(ctx, avn, avnGen, spec.Project, s.SourceServiceName)
+			r, err := checkServiceIsRunning(ctx, avnGen, spec.Project, s.SourceServiceName)
 			if !r || err != nil {
 				return false, err
 			}

diff --git a/controllers/kafkaacl_controller.go b/controllers/kafkaacl_controller.go
@@ -165,7 +165,7 @@ func (h KafkaACLHandler) checkPreconditions(ctx context.Context, avn *aiven.Clie
 	meta.SetStatusCondition(&acl.Status.Conditions,
 		getInitializedCondition("Preconditions", "Checking preconditions"))
 
-	return checkServiceIsRunning(ctx, avn, avnGen, acl.Spec.Project, acl.Spec.ServiceName)
+	return checkServiceIsRunning(ctx, avnGen, acl.Spec.Project, acl.Spec.ServiceName)
 }
 
 func (h KafkaACLHandler) convert(i client.Object) (*v1alpha1.KafkaACL, error) {

diff --git a/controllers/kafkaconnector_controller.go b/controllers/kafkaconnector_controller.go
@@ -220,7 +220,7 @@ func (h KafkaConnectorHandler) checkPreconditions(ctx context.Context, avn *aive
 	meta.SetStatusCondition(&conn.Status.Conditions,
 		getInitializedCondition("Preconditions", "Checking preconditions"))
 
-	return checkServiceIsRunning(ctx, avn, avnGen, conn.Spec.Project, conn.Spec.ServiceName)
+	return checkServiceIsRunning(ctx, avnGen, conn.Spec.Project, conn.Spec.ServiceName)
 }
 
 func (h KafkaConnectorHandler) convert(o client.Object) (*v1alpha1.KafkaConnector, error) {

diff --git a/controllers/kafkaschema_controller.go b/controllers/kafkaschema_controller.go
@@ -134,7 +134,7 @@ func (h KafkaSchemaHandler) checkPreconditions(ctx context.Context, avn *aiven.C
 		return false, err
 	}
 
-	return checkServiceIsRunning(ctx, avn, avnGen, schema.Spec.Project, schema.Spec.ServiceName)
+	return checkServiceIsRunning(ctx, avnGen, schema.Spec.Project, schema.Spec.ServiceName)
 }
 
 func (h KafkaSchemaHandler) convert(i client.Object) (*v1alpha1.KafkaSchema, error) {

diff --git a/controllers/kafkaschemaregistryacl_controller.go b/controllers/kafkaschemaregistryacl_controller.go
@@ -157,7 +157,7 @@ func (h KafkaSchemaRegistryACLHandler) checkPreconditions(ctx context.Context, a
 	meta.SetStatusCondition(&acl.Status.Conditions,
 		getInitializedCondition("Preconditions", "Checking preconditions"))
 
-	return checkServiceIsRunning(ctx, avn, avnGen, acl.Spec.Project, acl.Spec.ServiceName)
+	return checkServiceIsRunning(ctx, avnGen, acl.Spec.Project, acl.Spec.ServiceName)
 }
 
 func (h KafkaSchemaRegistryACLHandler) convert(i client.Object) (*v1alpha1.KafkaSchemaRegistryACL, error) {

diff --git a/controllers/kafkatopic_controller.go b/controllers/kafkatopic_controller.go
@@ -9,6 +9,7 @@ import (
 
 	"github.com/aiven/aiven-go-client/v2"
 	avngen "github.com/aiven/go-client-codegen"
+	"github.com/aiven/go-client-codegen/handler/service"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -174,7 +175,17 @@ func (h KafkaTopicHandler) checkPreconditions(ctx context.Context, avn *aiven.Cl
 	meta.SetStatusCondition(&topic.Status.Conditions,
 		getInitializedCondition("Preconditions", "Checking preconditions"))
 
-	return checkServiceIsRunning(ctx, avn, avnGen, topic.Spec.Project, topic.Spec.ServiceName)
+	if topic.Spec.Replication != 0 {
+		// Replication factor requires all nodes running
+		s, err := avnGen.ServiceGet(ctx, topic.Spec.Project, topic.Spec.ServiceName)
+		if isNotFound(err) {
+			return false, nil
+		}
+
+		return s.State == service.ServiceStateTypeRunning, nil
+	}
+
+	return checkServiceIsRunning(ctx, avnGen, topic.Spec.Project, topic.Spec.ServiceName)
 }
 
 func (h KafkaTopicHandler) getState(ctx context.Context, avn *aiven.Client, topic *v1alpha1.KafkaTopic) (string, error) {

diff --git a/controllers/serviceintegration_controller.go b/controllers/serviceintegration_controller.go
@@ -174,7 +174,7 @@ func (h ServiceIntegrationHandler) checkPreconditions(ctx context.Context, avn *
 		if project == "" {
 			project = si.Spec.Project
 		}
-		running, err := checkServiceIsRunning(ctx, avn, avnGen, project, si.Spec.SourceServiceName)
+		running, err := checkServiceIsRunning(ctx, avnGen, project, si.Spec.SourceServiceName)
 		if !running || err != nil {
 			return false, err
 		}
@@ -185,7 +185,7 @@ func (h ServiceIntegrationHandler) checkPreconditions(ctx context.Context, avn *
 		if project == "" {
 			project = si.Spec.Project
 		}
-		running, err := checkServiceIsRunning(ctx, avn, avnGen, project, si.Spec.DestinationServiceName)
+		running, err := checkServiceIsRunning(ctx, avnGen, project, si.Spec.DestinationServiceName)
 		if !running || err != nil {
 			return false, err
 		}

diff --git a/controllers/serviceuser_controller.go b/controllers/serviceuser_controller.go
@@ -161,7 +161,7 @@ func (h ServiceUserHandler) checkPreconditions(ctx context.Context, avn *aiven.C
 	meta.SetStatusCondition(&user.Status.Conditions,
 		getInitializedCondition("Preconditions", "Checking preconditions"))
 
-	return checkServiceIsRunning(ctx, avn, avnGen, user.Spec.Project, user.Spec.ServiceName)
+	return checkServiceIsRunning(ctx, avnGen, user.Spec.Project, user.Spec.ServiceName)
 }
 
 func (h ServiceUserHandler) convert(i client.Object) (*v1alpha1.ServiceUser, error) {