fix database initializing conditions

Author: kobzonega

internal/cms/operation.go

@@ -0,0 +1,74 @@
+package cms
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/ydb-platform/ydb-go-genproto/Ydb_Operation_V1"
+	"github.com/ydb-platform/ydb-go-genproto/protos/Ydb"
+	"github.com/ydb-platform/ydb-go-genproto/protos/Ydb_Operations"
+	"github.com/ydb-platform/ydb-go-sdk/v3"
+	"sigs.k8s.io/controller-runtime/pkg/log"
+
+	"github.com/ydb-platform/ydb-kubernetes-operator/internal/connection"
+)
+
+const (
+	GetOperationTimeoutSeconds = 10
+)
+
+type Operation struct {
+	StorageEndpoint string
+	Domain          string
+	Id              string
+}
+
+func (op *Operation) GetOperation(
+	ctx context.Context,
+	opts ...ydb.Option,
+) (*Ydb_Operations.GetOperationResponse, error) {
+	logger := log.FromContext(ctx)
+
+	endpoint := fmt.Sprintf("%s/%s", op.StorageEndpoint, op.Domain)
+	ydbCtx, ydbCtxCancel := context.WithTimeout(ctx, time.Second)
+	defer ydbCtxCancel()
+	conn, err := connection.Open(ydbCtx, endpoint, ydb.MergeOptions(opts...))
+	if err != nil {
+		return nil, fmt.Errorf("error connecting to YDB: %w", err)
+	}
+	defer func() {
+		connection.Close(ydbCtx, conn)
+	}()
+
+	cmsCtx, cmsCtxCancel := context.WithTimeout(ctx, GetOperationTimeoutSeconds*time.Second)
+	defer cmsCtxCancel()
+	client := Ydb_Operation_V1.NewOperationServiceClient(ydb.GRPCConn(conn))
+	request := &Ydb_Operations.GetOperationRequest{Id: op.Id}
+
+	logger.Info("CMS GetOperation", "endpoint", endpoint, "request", request)
+	return client.GetOperation(cmsCtx, request)
+}
+
+func (op *Operation) CheckGetOperationResponse(ctx context.Context, response *Ydb_Operations.GetOperationResponse) (bool, string, error) {
+	logger := log.FromContext(ctx)
+
+	logger.Info("CMS GetOperation", "response", response)
+	return CheckOperationStatus(response.GetOperation())
+}
+
+func CheckOperationStatus(operation *Ydb_Operations.Operation) (bool, string, error) {
+	if operation == nil {
+		return false, "", ErrEmptyReplyFromStorage
+	}
+
+	if !operation.GetReady() {
+		return false, operation.Id, nil
+	}
+
+	if operation.Status == Ydb.StatusIds_ALREADY_EXISTS || operation.Status == Ydb.StatusIds_SUCCESS {
+		return true, operation.Id, nil
+	}
+
+	return true, operation.Id, fmt.Errorf("YDB response error: %v %v", operation.Status, operation.Issues)
+}

internal/cms/tenant.go

@@ -4,19 +4,21 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"time"
 
 	"github.com/ydb-platform/ydb-go-genproto/Ydb_Cms_V1"
-	"github.com/ydb-platform/ydb-go-genproto/Ydb_Operation_V1"
-	"github.com/ydb-platform/ydb-go-genproto/protos/Ydb"
 	"github.com/ydb-platform/ydb-go-genproto/protos/Ydb_Cms"
-	"github.com/ydb-platform/ydb-go-genproto/protos/Ydb_Operations"
 	ydb "github.com/ydb-platform/ydb-go-sdk/v3"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 
 	ydbv1alpha1 "github.com/ydb-platform/ydb-kubernetes-operator/api/v1alpha1"
 	"github.com/ydb-platform/ydb-kubernetes-operator/internal/connection"
 )
 
+const (
+	CreateDatabaseTimeoutSeconds = 10
+)
+
 var ErrEmptyReplyFromStorage = errors.New("empty reply from storage")
 
 type Tenant struct {
@@ -28,31 +30,37 @@ type Tenant struct {
 	SharedDatabasePath string
 }
 
-func (t *Tenant) Create(
+func (t *Tenant) CreateDatabase(
 	ctx context.Context,
 	opts ...ydb.Option,
-) (string, error) {
+) (*Ydb_Cms.CreateDatabaseResponse, error) {
 	logger := log.FromContext(ctx)
-	url := fmt.Sprintf("%s/%s", t.StorageEndpoint, t.Domain)
-	conn, err := connection.Open(ctx, url, opts...)
+
+	endpoint := fmt.Sprintf("%s/%s", t.StorageEndpoint, t.Domain)
+	ydbCtx, ydbCtxCancel := context.WithTimeout(ctx, time.Second)
+	defer ydbCtxCancel()
+	conn, err := connection.Open(ydbCtx, endpoint, opts...)
 	if err != nil {
-		logger.Error(err, "Error connecting to YDB storage")
-		return "", err
+		logger.Error(err, "Error connecting to YDB")
+		return nil, err
 	}
 	defer func() {
-		connection.Close(ctx, conn)
+		connection.Close(ydbCtx, conn)
 	}()
 
+	cmsCtx, cmsCtxCancel := context.WithTimeout(ctx, CreateDatabaseTimeoutSeconds*time.Second)
+	defer cmsCtxCancel()
 	client := Ydb_Cms_V1.NewCmsServiceClient(ydb.GRPCConn(conn))
-	logger.Info(fmt.Sprintf("creating tenant, url: %s", url))
 	request := t.makeCreateDatabaseRequest()
-	logger.Info(fmt.Sprintf("creating tenant, request: %s", request))
-	response, err := client.CreateDatabase(ctx, request)
-	if err != nil {
-		return "", err
-	}
-	logger.Info(fmt.Sprintf("creating tenant, response: %s", response))
-	return processDatabaseCreationOperation(response.Operation)
+	logger.Info("CMS CreateDatabase", "endpoint", endpoint, "request", request)
+	return client.CreateDatabase(cmsCtx, request)
+}
+
+func (t *Tenant) CheckCreateDatabaseResponse(ctx context.Context, response *Ydb_Cms.CreateDatabaseResponse) (bool, string, error) {
+	logger := log.FromContext(ctx)
+
+	logger.Info("CMS CreateDatabase", "response", response)
+	return CheckOperationStatus(response.GetOperation())
 }
 
 func (t *Tenant) makeCreateDatabaseRequest() *Ydb_Cms.CreateDatabaseRequest {
@@ -87,47 +95,3 @@ func (t *Tenant) makeCreateDatabaseRequest() *Ydb_Cms.CreateDatabaseRequest {
 	}
 	return request
 }
-
-func processDatabaseCreationOperation(operation *Ydb_Operations.Operation) (string, error) {
-	if operation == nil {
-		return "", ErrEmptyReplyFromStorage
-	}
-	if !operation.Ready {
-		return operation.Id, nil
-	}
-	if operation.Status == Ydb.StatusIds_ALREADY_EXISTS || operation.Status == Ydb.StatusIds_SUCCESS {
-		return "", nil
-	}
-	return "", fmt.Errorf("YDB response error: %v %v", operation.Status, operation.Issues)
-}
-
-func (t *Tenant) CheckCreateOperation(
-	ctx context.Context,
-	operationID string,
-	opts ...ydb.Option,
-) (bool, error, error) {
-	logger := log.FromContext(ctx)
-	url := fmt.Sprintf("%s/%s", t.StorageEndpoint, t.Domain)
-	conn, err := connection.Open(ctx, url, opts...)
-	if err != nil {
-		logger.Error(err, "Error connecting to YDB storage")
-		return false, nil, err
-	}
-	defer func() {
-		connection.Close(ctx, conn)
-	}()
-
-	client := Ydb_Operation_V1.NewOperationServiceClient(ydb.GRPCConn(conn))
-	request := &Ydb_Operations.GetOperationRequest{Id: operationID}
-	logger.Info(fmt.Sprintf("checking operation, url: %s, operationId: %s, request: %s", url, operationID, request))
-	response, err := client.GetOperation(ctx, request)
-	if err != nil {
-		return false, nil, err
-	}
-	logger.Info(fmt.Sprintf("checking operation, response: %s", response))
-	if response.Operation == nil {
-		return false, nil, ErrEmptyReplyFromStorage
-	}
-	oid, err := processDatabaseCreationOperation(response.Operation)
-	return len(oid) == 0, err, nil
-}

internal/controllers/constants/constants.go

@@ -45,6 +45,7 @@ const (
 	ReasonInProgress  = "InProgress"
 	ReasonNotRequired = "NotRequired"
 	ReasonCompleted   = "Completed"
+	ReasonFailed      = "Failed"
 
 	DefaultRequeueDelay                = 10 * time.Second
 	StatusUpdateRequeueDelay           = 1 * time.Second

internal/controllers/database/init.go

@@ -4,7 +4,7 @@ import (
 	"context"
 	"fmt"
 
-	"github.com/ydb-platform/ydb-go-sdk/v3"
+	ydb "github.com/ydb-platform/ydb-go-sdk/v3"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
@@ -72,62 +72,71 @@ func (r *Reconciler) setInitDatabaseCompleted(
 	return r.updateStatus(ctx, database, StatusUpdateRequeueDelay)
 }
 
-func (r *Reconciler) checkCreateTenantOperation(
+func (r *Reconciler) checkCreateDatabaseOperation(
 	ctx context.Context,
 	database *resources.DatabaseBuilder,
 	tenant *cms.Tenant,
 	ydbOptions ydb.Option,
 ) (bool, ctrl.Result, error) {
 	condition := meta.FindStatusCondition(database.Status.Conditions, CreateDatabaseOperationCondition)
-	if condition == nil || len(condition.Message) == 0 {
+	if len(condition.Message) == 0 {
 		// Something is wrong with the condition where we save operation id
 		// retry create tenant
 		meta.SetStatusCondition(&database.Status.Conditions, metav1.Condition{
 			Type:   CreateDatabaseOperationCondition,
-			Status: metav1.ConditionTrue,
-			Reason: ReasonNotRequired,
+			Status: metav1.ConditionFalse,
+			Reason: ReasonFailed,
 		})
 		return r.updateStatus(ctx, database, DatabaseInitializationRequeueDelay)
 	}
-	operationID := condition.Message
-	finished, operationErr, err := tenant.CheckCreateOperation(ctx, operationID, ydbOptions)
+
+	operation := &cms.Operation{
+		StorageEndpoint: tenant.StorageEndpoint,
+		Domain:          tenant.Domain,
+		Id:              condition.Message,
+	}
+	response, err := operation.GetOperation(ctx, ydbOptions)
 	if err != nil {
 		r.Recorder.Event(
 			database,
 			corev1.EventTypeWarning,
 			"InitializingFailed",
-			fmt.Sprintf("Error creating tenant %s: %s", tenant.Path, err),
+			fmt.Sprintf("Failed to check creation operation, operationID %s: %s", operation.Id, err),
 		)
 		return Stop, ctrl.Result{RequeueAfter: DatabaseInitializationRequeueDelay}, err
 	}
-	if operationErr != nil {
-		// Creation operation failed - retry Create Tenant
+
+	finished, operationID, err := operation.CheckGetOperationResponse(ctx, response)
+	if err != nil {
 		r.Recorder.Event(
 			database,
 			corev1.EventTypeWarning,
 			"InitializingFailed",
-			fmt.Sprintf("Error creating tenant %s: %s", tenant.Path, operationErr),
+			fmt.Sprintf("Error creating tenant %s: %s", tenant.Path, err),
 		)
 		meta.SetStatusCondition(&database.Status.Conditions, metav1.Condition{
-			Type:   CreateDatabaseOperationCondition,
-			Status: metav1.ConditionTrue,
-			Reason: ReasonNotRequired,
+			Type:    CreateDatabaseOperationCondition,
+			Status:  metav1.ConditionFalse,
+			Reason:  ReasonFailed,
+			Message: fmt.Sprintf("Failed to create tenant %s", tenant.Path),
 		})
-		return r.updateStatus(ctx, database, DatabaseInitializationRequeueDelay)
+		return Stop, ctrl.Result{RequeueAfter: DatabaseInitializationRequeueDelay}, err
 	}
+
 	if !finished {
 		r.Recorder.Event(
 			database,
 			corev1.EventTypeWarning,
-			"Pending",
+			string(DatabaseInitializing),
 			fmt.Sprintf("Tenant creation operation is not completed, operationID: %s", operationID),
 		)
 		return Stop, ctrl.Result{RequeueAfter: DatabaseInitializationRequeueDelay}, nil
 	}
+
 	r.Recorder.Event(
 		database,
 		corev1.EventTypeNormal,
-		"Initialized",
+		string(DatabaseInitializing),
 		fmt.Sprintf("Tenant %s created", tenant.Path),
 	)
 	return r.setInitDatabaseCompleted(ctx, database, "Database initialized successfully")
@@ -239,9 +248,10 @@ func (r *Reconciler) initializeTenant(
 	ydbOpts := ydb.MergeOptions(ydb.WithCredentials(creds), tlsOptions)
 
 	if meta.IsStatusConditionFalse(database.Status.Conditions, CreateDatabaseOperationCondition) {
-		return r.checkCreateTenantOperation(ctx, database, tenant, ydbOpts)
+		return r.checkCreateDatabaseOperation(ctx, database, tenant, ydbOpts)
 	}
-	operationID, err := tenant.Create(ctx, ydb.WithCredentials(creds), tlsOptions)
+
+	response, err := tenant.CreateDatabase(ctx, ydbOpts)
 	if err != nil {
 		r.Recorder.Event(
 			database,
@@ -251,11 +261,23 @@ func (r *Reconciler) initializeTenant(
 		)
 		return Stop, ctrl.Result{RequeueAfter: DatabaseInitializationRequeueDelay}, err
 	}
-	if len(operationID) > 0 {
+
+	finished, operationID, err := tenant.CheckCreateDatabaseResponse(ctx, response)
+	if err != nil {
+		r.Recorder.Event(
+			database,
+			corev1.EventTypeWarning,
+			"InitializingFailed",
+			fmt.Sprintf("Failed %s: %s", tenant.Path, err),
+		)
+		return Stop, ctrl.Result{RequeueAfter: DatabaseInitializationRequeueDelay}, err
+	}
+
+	if !finished {
 		r.Recorder.Event(
 			database,
 			corev1.EventTypeWarning,
-			"Pending",
+			string(DatabaseInitializing),
 			fmt.Sprintf("Tenant creation operation in progress, operationID: %s", operationID),
 		)
 		meta.SetStatusCondition(&database.Status.Conditions, metav1.Condition{
@@ -269,7 +291,7 @@ func (r *Reconciler) initializeTenant(
 	r.Recorder.Event(
 		database,
 		corev1.EventTypeNormal,
-		"Initialized",
+		string(DatabaseInitializing),
 		fmt.Sprintf("Tenant %s created", tenant.Path),
 	)