fix: Incorrect handling of parallel CDStageDeploy (#71)

Added in-queue status for CDStageDeploy.
This status is for CDStageDeploy,
which waits for the previous deployment to be ready.
We can only have one CDStageDeploy status in-queue;
all other  CDStageDeploy will be skipped.
This will allow us to get the latest versions
and not run redundant deployments.

api/v1/cdstagedeploy_types.go

@@ -6,12 +6,11 @@ import (
 	metaV1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
-// NOTE: json tags are required.  Any new fields you add must have json tags for the fields to be serialized.
-
 const (
 	CDStageDeployStatusFailed    = "failed"
 	CDStageDeployStatusRunning   = "running"
 	CDStageDeployStatusPending   = "pending"
+	CDStageDeployStatusInQueue   = "in-queue"
 	CDStageDeployStatusCompleted = "completed"
 )
 
@@ -39,7 +38,7 @@ type CodebaseTag struct {
 type CDStageDeployStatus struct {
 	// Specifies a current status of CDStageDeploy.
 	// +optional
-	// +kubebuilder:validation:Enum=failed;running;pending;completed
+	// +kubebuilder:validation:Enum=failed;running;pending;completed;in-queue
 	// +kubebuilder:default=pending
 	Status string `json:"status"`
 

config/crd/bases/v2.edp.epam.com_cdstagedeployments.yaml

@@ -93,6 +93,7 @@ spec:
                 - running
                 - pending
                 - completed
+                - in-queue
                 type: string
             type: object
         type: object

controllers/cdstagedeploy/chain/resolve_status.go

@@ -32,26 +32,52 @@ func (r *ResolveStatus) ServeRequest(ctx context.Context, stageDeploy *codebaseA
 		return nil
 	}
 
+	pipelineRun, err := r.getRunningPipelines(ctx, stageDeploy)
+	if err != nil {
+		return fmt.Errorf("failed to get running pipelines: %w", err)
+	}
+
 	if stageDeploy.Status.Status == codebaseApi.CDStageDeployStatusPending {
-		log.Info("CDStageDeploy has pending status. Start deploying.")
+		if allPipelineRunsCompleted(pipelineRun.Items) {
+			log.Info("CDStageDeploy has pending status. Start deploying.")
+
+			return nil
+		}
+
+		log.Info("Put CDStageDeploy in queue. Some PipelineRuns are still running.")
+
+		stageDeploy.Status.Status = codebaseApi.CDStageDeployStatusInQueue
 
 		return nil
 	}
 
-	pipelineRun, err := r.getRunningPipelines(ctx, stageDeploy)
-	if err != nil {
-		return fmt.Errorf("failed to get running pipelines: %w", err)
-	}
+	if stageDeploy.Status.Status == codebaseApi.CDStageDeployStatusInQueue {
+		if allPipelineRunsCompleted(pipelineRun.Items) {
+			log.Info("All PipelineRuns have been completed.")
+
+			stageDeploy.Status.Status = codebaseApi.CDStageDeployStatusPending
 
-	if allPipelineRunsCompleted(pipelineRun.Items) {
-		log.Info("All PipelineRuns have been completed.")
+			return nil
+		}
 
-		stageDeploy.Status.Status = codebaseApi.CDStageDeployStatusCompleted
+		log.Info("Some PipelineRuns are still running.")
 
 		return nil
 	}
 
-	log.Info("Some PipelineRuns are still running.")
+	if stageDeploy.Status.Status == codebaseApi.CDStageDeployStatusRunning {
+		if allPipelineRunsCompleted(pipelineRun.Items) {
+			log.Info("All PipelineRuns have been completed.")
+
+			stageDeploy.Status.Status = codebaseApi.CDStageDeployStatusCompleted
+
+			return nil
+		}
+
+		log.Info("Some PipelineRuns are still running.")
+
+		return nil
+	}
 
 	return nil
 }

controllers/codebaseimagestream/chain/factory.go

@@ -1,17 +1,13 @@
 package chain
 
 import (
-	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/epam/edp-codebase-operator/v2/controllers/codebaseimagestream/chain/handler"
 )
 
-var log = ctrl.Log.WithName("codebase-image-stream")
-
 func CreateDefChain(c client.Client) handler.CodebaseImageStreamHandler {
 	return PutCDStageDeploy{
 		client: c,
-		log:    log.WithName("create-chain").WithName("put-cd-stage-deploy"),
 	}
 }

controllers/codebaseimagestream/chain/handler/codebase_image_stream_handler.go

@@ -1,9 +1,11 @@
 package handler
 
 import (
+	"context"
+
 	codebaseApi "github.com/epam/edp-codebase-operator/v2/api/v1"
 )
 
 type CodebaseImageStreamHandler interface {
-	ServeRequest(imageStream *codebaseApi.CodebaseImageStream) error
+	ServeRequest(ctx context.Context, imageStream *codebaseApi.CodebaseImageStream) error
 }

controllers/codebaseimagestream/chain/put_cd_stage_deploy.go

@@ -7,10 +7,9 @@ import (
 	"regexp"
 	"strings"
 
-	"github.com/go-logr/logr"
-	k8sErrors "k8s.io/apimachinery/pkg/api/errors"
 	metaV1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
+	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 
@@ -23,7 +22,6 @@ import (
 
 type PutCDStageDeploy struct {
 	client client.Client
-	log    logr.Logger
 }
 
 type cdStageDeployCommand struct {
@@ -35,26 +33,25 @@ type cdStageDeployCommand struct {
 	Tags      []codebaseApi.CodebaseTag
 }
 
-const (
-	logNameKey = "name"
-)
+func (h PutCDStageDeploy) ServeRequest(ctx context.Context, imageStream *codebaseApi.CodebaseImageStream) error {
+	l := ctrl.LoggerFrom(ctx)
 
-func (h PutCDStageDeploy) ServeRequest(imageStream *codebaseApi.CodebaseImageStream) error {
-	l := h.log.WithValues(logNameKey, imageStream.Name)
-	l.Info("creating/updating CDStageDeploy.")
+	l.Info("Creating CDStageDeploy.")
 
-	if err := h.handleCodebaseImageStreamEnvLabels(imageStream); err != nil {
+	if err := h.handleCodebaseImageStreamEnvLabels(ctx, imageStream); err != nil {
 		return fmt.Errorf("failed to handle %v codebase image stream: %w", imageStream.Name, err)
 	}
 
-	l.Info("creating/updating CDStageDeploy has been finished.")
+	l.Info("Creating CDStageDeploy has been finished.")
 
 	return nil
 }
 
-func (h PutCDStageDeploy) handleCodebaseImageStreamEnvLabels(imageStream *codebaseApi.CodebaseImageStream) error {
+func (h PutCDStageDeploy) handleCodebaseImageStreamEnvLabels(ctx context.Context, imageStream *codebaseApi.CodebaseImageStream) error {
+	l := ctrl.LoggerFrom(ctx)
+
 	if imageStream.ObjectMeta.Labels == nil || len(imageStream.ObjectMeta.Labels) == 0 {
-		h.log.Info("codebase image stream does not contain env labels. skip CDStageDeploy creating...")
+		l.Info("CodebaseImageStream does not contain env labels. Skip CDStageDeploy creating.")
 		return nil
 	}
 
@@ -65,7 +62,7 @@ func (h PutCDStageDeploy) handleCodebaseImageStreamEnvLabels(imageStream *codeba
 			return errors.New(strings.Join(errs, "; "))
 		}
 
-		if err := h.putCDStageDeploy(envLabel, imageStream.Namespace, imageStream.Spec); err != nil {
+		if err := h.putCDStageDeploy(ctx, envLabel, imageStream.Namespace, imageStream.Spec); err != nil {
 			return err
 		}
 	}
@@ -85,66 +82,78 @@ func validateCbis(imageStream *codebaseApi.CodebaseImageStream, envLabel string,
 	}
 
 	if !labelValueRegexp.MatchString(envLabel) {
-		errs = append(errs, "Label must be in format cd-pipeline-name/stage-name")
+		errs = append(errs, "label must be in format cd-pipeline-name/stage-name")
 	}
 
 	return errs
 }
 
-func (h PutCDStageDeploy) putCDStageDeploy(envLabel, namespace string, spec codebaseApi.CodebaseImageStreamSpec) error {
+func (h PutCDStageDeploy) putCDStageDeploy(ctx context.Context, envLabel, namespace string, spec codebaseApi.CodebaseImageStreamSpec) error {
+	l := ctrl.LoggerFrom(ctx)
 	// use name for CDStageDeploy, it is converted from envLabel and cdpipeline/stage now is cdpipeline-stage
 	name := strings.ReplaceAll(envLabel, "/", "-")
 
-	stageDeploy, err := h.getCDStageDeploy(name, namespace)
+	skip, err := h.skipCDStageDeployCreation(ctx, envLabel, namespace)
 	if err != nil {
-		return fmt.Errorf("failed to get %v cd stage deploy: %w", name, err)
+		return fmt.Errorf("failed to check if CDStageDeploy exists: %w", err)
 	}
 
-	if stageDeploy != nil {
-		h.log.Info("CDStageDeploy already exists. skip creating.", logNameKey, stageDeploy.Name)
+	if skip {
+		l.Info("Skip CDStageDeploy creation.")
 
 		return nil
 	}
 
-	cdsd, err := getCreateCommand(envLabel, name, namespace, spec.Codebase, spec.Tags, log)
+	cdsd, err := getCreateCommand(ctx, envLabel, name, namespace, spec.Codebase, spec.Tags)
 	if err != nil {
 		return fmt.Errorf("failed to construct command to create %v cd stage deploy: %w", name, err)
 	}
 
-	if err := h.create(cdsd); err != nil {
+	if err = h.create(ctx, cdsd); err != nil {
 		return fmt.Errorf("failed to create %v cd stage deploy: %w", name, err)
 	}
 
 	return nil
 }
 
-func (h PutCDStageDeploy) getCDStageDeploy(name, namespace string) (*codebaseApi.CDStageDeploy, error) {
-	h.log.Info("getting cd stage deploy", logNameKey, name)
+func (h PutCDStageDeploy) skipCDStageDeployCreation(ctx context.Context, envLabel, namespace string) (bool, error) {
+	l := ctrl.LoggerFrom(ctx)
+	l.Info("Getting CDStageDeploys.")
 
-	ctx := context.Background()
-	i := &codebaseApi.CDStageDeploy{}
-	nn := types.NamespacedName{
-		Namespace: namespace,
-		Name:      name,
-	}
-
-	if err := h.client.Get(ctx, nn, i); err != nil {
-		if k8sErrors.IsNotFound(err) {
-			return nil, nil
-		}
+	env := strings.Split(envLabel, "/")
 
-		return nil, fmt.Errorf("failed to fetch CDStageDeploy resource %q: %w", name, err)
+	list := &codebaseApi.CDStageDeployList{}
+	if err := h.client.List(
+		ctx,
+		list,
+		client.InNamespace(namespace),
+		client.MatchingLabels{
+			"app.edp.epam.com/cdpipeline": env[0],
+			"app.edp.epam.com/cdstage":    fmt.Sprintf("%s-%s", env[0], env[1]),
+		},
+	); err != nil {
+		return false, fmt.Errorf("failed to get CDStageDeploys: %w", err)
 	}
 
-	return i, nil
+	switch len(list.Items) {
+	case 0:
+		l.Info("CDStageDeploy is not present in cluster.")
+		return false, nil
+	case 1:
+		l.Info("One CDStageDeploy is present in cluster.")
+		return false, nil
+	default:
+		l.Info("More than one CDStageDeploy is present in cluster.")
+		return true, nil
+	}
 }
 
-func getCreateCommand(envLabel, name, namespace, codebase string, tags []codebaseApi.Tag, log logr.Logger) (*cdStageDeployCommand, error) {
+func getCreateCommand(ctx context.Context, envLabel, name, namespace, codebase string, tags []codebaseApi.Tag) (*cdStageDeployCommand, error) {
 	env := strings.Split(envLabel, "/")
 
-	lastTag, err := codebaseimagestream.GetLastTag(tags, log)
+	lastTag, err := codebaseimagestream.GetLastTag(tags, ctrl.LoggerFrom(ctx))
 	if err != nil {
-		return nil, fmt.Errorf("failed to create cdStageDeployCommand with name %v: %w", name, err)
+		return nil, fmt.Errorf("failed to get last tag: %w", err)
 	}
 
 	return &cdStageDeployCommand{
@@ -165,19 +174,18 @@ func getCreateCommand(envLabel, name, namespace, codebase string, tags []codebas
 	}, nil
 }
 
-func (h PutCDStageDeploy) create(command *cdStageDeployCommand) error {
-	log := h.log.WithValues(logNameKey, command.Name)
-	log.Info("cd stage deploy is not present in cluster. start creating...")
+func (h PutCDStageDeploy) create(ctx context.Context, command *cdStageDeployCommand) error {
+	l := ctrl.LoggerFrom(ctx)
+	l.Info("CDStageDeploy is not present in cluster. Start creating.")
 
-	ctx := context.Background()
 	stageDeploy := &codebaseApi.CDStageDeploy{
 		TypeMeta: metaV1.TypeMeta{
 			APIVersion: util.V2APIVersion,
 			Kind:       util.CDStageDeployKind,
 		},
 		ObjectMeta: metaV1.ObjectMeta{
-			Name:      command.Name,
-			Namespace: command.Namespace,
+			GenerateName: command.Name,
+			Namespace:    command.Namespace,
 		},
 		Spec: codebaseApi.CDStageDeploySpec{
 			Pipeline: command.Pipeline,
@@ -187,6 +195,11 @@ func (h PutCDStageDeploy) create(command *cdStageDeployCommand) error {
 		},
 	}
 
+	stageDeploy.SetLabels(map[string]string{
+		"app.edp.epam.com/cdpipeline": command.Pipeline,
+		"app.edp.epam.com/cdstage":    stageDeploy.GetStageCRName(),
+	})
+
 	stage := &pipelineApi.Stage{}
 	if err := h.client.Get(
 		ctx,
@@ -208,7 +221,7 @@ func (h PutCDStageDeploy) create(command *cdStageDeployCommand) error {
 		return fmt.Errorf("failed to create CDStageDeploy resource %q: %w", command.Name, err)
 	}
 
-	log.Info("cd stage deploy has been created.")
+	l.Info("CDStageDeploy has been created.")
 
 	return nil
 }