Gate nodesets on their deployOnAllNodeSets service jobs

docs/assemblies/proc_creating-a-custom-service.adoc

@@ -92,7 +92,9 @@ spec:
 +
 A mount is created for each `ConfigMap` and `Secret` in the `OpenStackAnsibleEE` pod with a filename that matches the resource value. The mounts are created under `/var/lib/openstack/configs/<service name>`.
 
-. Optional: It may be necessary to run some services on all nodesets at the same time. These services need to have their `deployOnAllNodeSets` field set to true. If these services are repated in multiple nodeset specs included in a deployment, they would be ignored from subsequent nodeset services and would be run only once.
+. Optional: It may be necessary to run some services on all nodesets at the same time. These services need to have their `deployOnAllNodeSets` field set to true. Such a service produces a single `OpenStackAnsibleEE` job for the whole deployment, targeting the hosts of every nodeset, even nodesets that do not list the service.
++
+A deployment scoped service is an execution barrier for the nodesets that list it: its job starts only once every service ordered before it in each of those nodesets has completed, and none of those nodesets starts a service ordered after it until the job completed. A nodeset that does not list the service neither waits for it nor delays it, so list the service in every nodeset whose deployment ordering depends on it.
 +
 ----
 apiVersion: dataplane.openstack.org/v1beta1

internal/controller/dataplane/openstackdataplanedeployment_controller.go

@@ -21,6 +21,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"slices"
 	"time"
 
 	"github.com/go-playground/validator/v10"
@@ -253,11 +254,10 @@ func (r *OpenStackDataPlaneDeploymentReconciler) Reconcile(ctx context.Context,
 	shouldRequeue := false
 	haveError := false
 	deploymentErrMsg := ""
-	var nodesetServiceMap map[string][]string
-	var serviceCache *deployment.ServiceCache
+	var servicePlan *deployment.ServicePlan
 	backoffLimitReached := false
 
-	if nodesetServiceMap, serviceCache, err = deployment.DedupeServices(ctx, helper, nodeSets.Items,
+	if servicePlan, err = deployment.DedupeServices(ctx, helper, nodeSets.Items,
 		instance.Spec.ServicesOverride); err != nil {
 		util.LogErrorForObject(helper, err, "OpenStackDeployment error for deployment", instance)
 		var missingServiceErr *deployment.MissingServiceError
@@ -304,22 +304,14 @@ func (r *OpenStackDataPlaneDeploymentReconciler) Reconcile(ctx context.Context,
 		return ctrl.Result{}, err
 	}
 
-	// Deploy each nodeSet
-	// The loop starts and checks NodeSet deployments sequentially. However, after they
-	// are started, they are running in parallel, since the loop does not wait
-	// for the first started NodeSet to finish before starting the next.
-	if !instance.Spec.UseParallelExecution {
-		Log.Info("useParallelExecution is not set to true, services will run sequentially in list order and service dependencies (dependsOn) are ignored")
-	}
-	for _, nodeSet := range nodeSets.Items {
+	// Built up front: the deployment starts its deployment scoped services itself,
+	// independent of which nodeset reconciles first.
+	deployers := make(map[string]*deployment.Deployer, len(nodeSets.Items))
+	serviceLevels := make(map[string][][]string, len(nodeSets.Items))
+
+	for i := range nodeSets.Items {
+		nodeSet := &nodeSets.Items[i]
 
-		Log.Info("Deploying NodeSet", "nodeSet", nodeSet.Name)
-		Log.Info("Set Status.Deployed to false", "instance", instance)
-		instance.Status.Deployed = false
-		Log.Info("Set DeploymentReadyCondition false")
-		instance.Status.Conditions.MarkFalse(
-			condition.DeploymentReadyCondition, condition.RequestedReason,
-			condition.SeverityInfo, condition.DeploymentReadyRunningMessage)
 		ansibleEESpec := nodeSet.GetAnsibleEESpec()
 		ansibleEESpec.AnsibleTags = instance.Spec.AnsibleTags
 		ansibleEESpec.AnsibleSkipTags = instance.Spec.AnsibleSkipTags
@@ -334,22 +326,22 @@ func (r *OpenStackDataPlaneDeploymentReconciler) Reconcile(ctx context.Context,
 			}
 		}
 
-		deployer := deployment.Deployer{
+		deployers[nodeSet.Name] = &deployment.Deployer{
 			Ctx:                         ctx,
 			Helper:                      helper,
-			NodeSet:                     &nodeSet,
+			NodeSet:                     nodeSet,
 			Deployment:                  instance,
 			Status:                      &instance.Status,
 			AeeSpec:                     &ansibleEESpec,
 			InventorySecrets:            globalInventorySecrets,
 			AnsibleSSHPrivateKeySecrets: globalSSHKeySecrets,
-			ServiceCache:                serviceCache,
+			ServiceCache:                servicePlan.Cache,
 			Version:                     version,
 		}
 
-		var serviceLevels [][]string
 		if instance.Spec.UseParallelExecution {
-			serviceLevels, err = deployment.BuildServiceLevels(ctx, helper, serviceCache, nodesetServiceMap[nodeSet.Name])
+			serviceLevels[nodeSet.Name], err = deployment.BuildServiceLevels(
+				ctx, helper, servicePlan.Cache, servicePlan.Services[nodeSet.Name])
 			if err != nil {
 				util.LogErrorForObject(helper, err, fmt.Sprintf("error building service dependency graph for NodeSet %s", nodeSet.Name), instance)
 				graphErrMsg := fmt.Sprintf("nodeSet: %s error: %s", nodeSet.Name, err.Error())
@@ -370,14 +362,52 @@ func (r *OpenStackDataPlaneDeploymentReconciler) Reconcile(ctx context.Context,
 				return ctrl.Result{}, fmt.Errorf("error building service dependency graph for nodeset %s: %w", nodeSet.Name, err)
 			}
 		} else {
-			serviceLevels = deployment.SerialServiceLevels(nodesetServiceMap[nodeSet.Name])
+			serviceLevels[nodeSet.Name] = deployment.SerialServiceLevels(servicePlan.Services[nodeSet.Name])
 		}
+		deployers[nodeSet.Name].AllServices = slices.Concat(serviceLevels[nodeSet.Name]...)
+	}
+
+	// Start them only once every listing nodeset reached its barrier, else a
+	// nodeset advances while the shared Job still runs against its hosts.
+	if err := r.deployGlobalServices(ctx, helper, instance, servicePlan, serviceLevels, deployers); err != nil {
+		return ctrl.Result{}, err
+	}
+
+	// Deploy each nodeSet
+	// The loop starts and checks NodeSet deployments sequentially. However, after they
+	// are started, they are running in parallel, since the loop does not wait
+	// for the first started NodeSet to finish before starting the next.
+	if !instance.Spec.UseParallelExecution {
+		Log.Info("useParallelExecution is not set to true, services will run sequentially in list order and service dependencies (dependsOn) are ignored")
+	}
+	for i := range nodeSets.Items {
+		nodeSet := &nodeSets.Items[i]
+
+		Log.Info("Deploying NodeSet", "nodeSet", nodeSet.Name)
+		Log.Info("Set Status.Deployed to false", "instance", instance)
+		instance.Status.Deployed = false
+		Log.Info("Set DeploymentReadyCondition false")
+		instance.Status.Conditions.MarkFalse(
+			condition.DeploymentReadyCondition, condition.RequestedReason,
+			condition.SeverityInfo, condition.DeploymentReadyRunningMessage)
 
 		var deployResult *ctrl.Result
-		deployResult, err = deployer.Deploy(serviceLevels)
+		deployResult, err = deployers[nodeSet.Name].Deploy(serviceLevels[nodeSet.Name])
 
 		nsConditions := instance.Status.NodeSetConditions[nodeSet.Name]
 		nsConditions.Set(nsConditions.Mirror(dataplanev1.NodeSetDeploymentReadyCondition))
+		// A barrier wait sets no service condition, so surface it on the nodeset.
+		if err == nil {
+			if blockedOn := deployers[nodeSet.Name].BlockedOn; blockedOn != "" {
+				nsConditions.Set(condition.FalseCondition(
+					dataplanev1.NodeSetDeploymentReadyCondition,
+					condition.RequestedReason,
+					condition.SeverityInfo,
+					"%s", fmt.Sprintf("%s: it runs once for the whole deployment",
+						fmt.Sprintf(dataplanev1.NodeSetServiceDeploymentReadyWaitingMessage, blockedOn))))
+			}
+		}
+		instance.Status.NodeSetConditions[nodeSet.Name] = nsConditions
 
 		if err != nil {
 			util.LogErrorForObject(helper, err, fmt.Sprintf("OpenStackDeployment error for NodeSet %s", nodeSet.Name), instance)
@@ -445,6 +475,64 @@ func (r *OpenStackDataPlaneDeploymentReconciler) Reconcile(ctx context.Context,
 	return ctrl.Result{}, nil
 }
 
+// deployGlobalServices starts the single AnsibleEE Job of every deployment
+// scoped service whose barrier has cleared.
+func (r *OpenStackDataPlaneDeploymentReconciler) deployGlobalServices(
+	ctx context.Context,
+	helper *helper.Helper,
+	instance *dataplanev1.OpenStackDataPlaneDeployment,
+	servicePlan *deployment.ServicePlan,
+	serviceLevels map[string][][]string,
+	deployers map[string]*deployment.Deployer,
+) error {
+	Log := r.GetLogger(ctx)
+	jobState := deployment.JobState{}
+
+	for _, serviceName := range servicePlan.GlobalServices {
+		gateOpen, err := deployment.GlobalServiceGateOpen(
+			ctx, helper, servicePlan, instance, serviceName, serviceLevels, jobState)
+		if err != nil {
+			return err
+		}
+		if !gateOpen {
+			Log.Info("Deployment scoped service is behind a barrier that has not cleared",
+				"service", serviceName, "nodeSets", servicePlan.NodesetsWithGlobalService[serviceName])
+			continue
+		}
+
+		foundService, err := servicePlan.Cache.Get(ctx, helper, serviceName)
+		if err != nil {
+			instance.Status.Conditions.MarkFalse(
+				condition.InputReadyCondition,
+				condition.ErrorReason,
+				condition.SeverityError,
+				dataplanev1.ServiceErrorMessage,
+				err.Error())
+			return err
+		}
+
+		listingNodeSets := servicePlan.NodesetsWithGlobalService[serviceName]
+		if len(listingNodeSets) == 0 {
+			continue
+		}
+		// Any listing nodeset works; it only supplies the Job labels and mounts.
+		if err := deployers[listingNodeSets[0]].DeployGlobalService(foundService); err != nil {
+			util.LogErrorForObject(helper, err,
+				fmt.Sprintf("Unable to execute deployment scoped service %s", serviceName), instance)
+			instance.Status.Conditions.MarkFalse(
+				condition.DeploymentReadyCondition,
+				condition.ErrorReason,
+				condition.SeverityError,
+				condition.DeploymentReadyErrorMessage,
+				err.Error())
+			return err
+		}
+		Log.Info("Started deployment scoped service", "service", serviceName, "nodeSets", listingNodeSets)
+	}
+
+	return nil
+}
+
 // GetService retrieves a service for the OpenStackDataPlaneDeployment
 func (r *OpenStackDataPlaneDeploymentReconciler) GetService(
 	ctx context.Context,

internal/dataplane/deployment.go

@@ -57,7 +57,12 @@ type Deployer struct {
 	AnsibleSSHPrivateKeySecrets map[string]string
 	ServiceCache                *ServiceCache
 	Version                     *openstackv1.OpenStackVersion
-	allServices                 []string
+	// AllServices is the nodeset's flat service plan, used to resolve cert
+	// mounts. Set it before any Job is built, including the deployment scoped
+	// ones the deployment starts outside the nodeset loop.
+	AllServices []string
+	// BlockedOn names the barrier a nodeset is stalled on, for its status.
+	BlockedOn string
 }
 
 // Deploy runs the nodeset's leveled service plan, waiting for each level
@@ -67,9 +72,6 @@ func (d *Deployer) Deploy(serviceLevels [][]string) (*ctrl.Result, error) {
 		d.ServiceCache = NewServiceCache()
 	}
 
-	// Flat list for cert-mount lookups
-	d.allServices = slices.Concat(serviceLevels...)
-
 	for levelIdx, level := range serviceLevels {
 		d.Helper.GetLogger().Info("Deploying service level", "level", levelIdx, "services", level)
 
@@ -78,7 +80,11 @@ func (d *Deployer) Deploy(serviceLevels [][]string) (*ctrl.Result, error) {
 			return &ctrl.Result{}, err
 		}
 
-		if !d.isLevelReady(levelIdx, level) {
+		ready, err := d.isLevelReady(levelIdx, level)
+		if err != nil {
+			return &ctrl.Result{}, err
+		}
+		if !ready {
 			return &ctrl.Result{}, nil
 		}
 	}
@@ -112,6 +118,11 @@ func (d *Deployer) startLevel(levelIdx int, level []string) error {
 			continue
 		}
 
+		// Deployment scoped Jobs are started by the deployment; isLevelReady waits.
+		if foundService.Spec.DeployOnAllNodeSets {
+			continue
+		}
+
 		serviceAeeSpec, err := d.buildServiceAeeSpec(foundService)
 		if err != nil {
 			d.setServiceError(readyCondition, readyErrorMessage, err)
@@ -136,17 +147,107 @@ func (d *Deployer) startLevel(levelIdx int, level []string) error {
 }
 
 // isLevelReady reports whether every service in the level completed.
-func (d *Deployer) isLevelReady(levelIdx int, level []string) bool {
+func (d *Deployer) isLevelReady(levelIdx int, level []string) (bool, error) {
 	log := d.Helper.GetLogger()
 	nsConditions := d.Status.NodeSetConditions[d.NodeSet.Name]
 	for _, service := range level {
+		foundService, err := d.ServiceCache.Get(d.Ctx, d.Helper, service)
+		if err != nil {
+			return false, err
+		}
+
+		if foundService.Spec.DeployOnAllNodeSets {
+			completed, err := d.globalServiceCompleted(foundService)
+			if err != nil {
+				return false, err
+			}
+			if !completed {
+				d.BlockedOn = service
+				log.Info("Waiting for deployment scoped service",
+					"service", service, "level", levelIdx, "nodeSet", d.NodeSet.Name)
+				return false, nil
+			}
+			continue
+		}
+
 		readyCondition := d.serviceReadyCondition(service)
 		if !nsConditions.IsTrue(readyCondition) {
 			log.Info("Condition not ready in service level, waiting", "condition", readyCondition, "level", levelIdx)
-			return false
+			return false, nil
+		}
+	}
+	return true, nil
+}
+
+// globalServiceCompleted reports whether the Job of a deployment scoped service
+// finished successfully. A Job past its backoff limit is recorded on the nodeset
+// conditions, so it fails the same way a nodeset owned Job would.
+func (d *Deployer) globalServiceCompleted(
+	service dataplanev1.OpenStackDataPlaneService,
+) (bool, error) {
+	log := d.Helper.GetLogger()
+
+	ansibleJob, err := dataplaneutil.GetAnsibleExecution(
+		d.Ctx, d.Helper, d.Deployment,
+		dataplaneutil.GetGlobalAnsibleExecutionLabels(service.Name, d.Deployment.Name))
+	if err != nil {
+		if k8s_errors.IsNotFound(err) {
+			return false, nil
+		}
+		return false, err
+	}
+
+	if ansibleJob.Status.Succeeded > 0 {
+		d.storeExecutionSummary(ansibleJob)
+		d.recordContainerImages(service)
+		return true, nil
+	}
+
+	if ansibleJob.Status.Failed > *ansibleJob.Spec.BackoffLimit {
+		reason := condition.Reason(condition.ErrorReason)
+		for _, jobCondition := range ansibleJob.Status.Conditions {
+			if jobCondition.Type == batchv1.JobFailed {
+				reason = condition.Reason(jobCondition.Reason)
+			}
+		}
+		d.storeExecutionSummary(ansibleJob)
+		errorMsg := fmt.Sprintf("execution.name %s execution.namespace %s failed pods: %d",
+			ansibleJob.Name, ansibleJob.Namespace, ansibleJob.Status.Failed)
+		if reason == condition.JobReasonBackoffLimitExceeded {
+			errorMsg = fmt.Sprintf("backoff limit reached for execution.name %s execution.namespace %s",
+				ansibleJob.Name, ansibleJob.Namespace)
 		}
+		log.Info("Deployment scoped service failed", "service", service.Name, "execution", ansibleJob.Name)
+
+		nsConditions := d.Status.NodeSetConditions[d.NodeSet.Name]
+		nsConditions.Set(condition.FalseCondition(
+			dataplanev1.NodeSetDeploymentReadyCondition,
+			reason,
+			condition.SeverityError,
+			"%s", errorMsg))
+		d.Status.NodeSetConditions[d.NodeSet.Name] = nsConditions
+		return false, fmt.Errorf("%s", errorMsg)
 	}
-	return true
+
+	return false, nil
+}
+
+// DeployGlobalService creates the single AnsibleEE Job of a deployment scoped
+// service. The bound nodeset only supplies the Job labels and the mounts; the
+// play targets every nodeset's hosts.
+func (d *Deployer) DeployGlobalService(
+	service dataplanev1.OpenStackDataPlaneService,
+) error {
+	if !service.Spec.DeployOnAllNodeSets {
+		return fmt.Errorf("service %s is not deployment scoped", service.Name)
+	}
+
+	serviceAeeSpec, err := d.buildServiceAeeSpec(service)
+	if err != nil {
+		return err
+	}
+
+	return d.DeployService(service, serviceAeeSpec)
 }
 
 // serviceReadyCondition returns the per-service readiness condition type.
@@ -314,7 +415,7 @@ func (d *Deployer) buildServiceAeeSpec(
 	}
 
 	if foundService.Spec.AddCertMounts {
-		if err := d.addCertMounts(serviceAeeSpec, d.allServices); err != nil {
+		if err := d.addCertMounts(serviceAeeSpec, d.AllServices); err != nil {
 			return nil, err
 		}
 	} else if len(foundService.Spec.CACerts) > 0 {