From 9d6a910cd7794070995fabf1343c86b8d651bd57 Mon Sep 17 00:00:00 2001 From: Rabi Mishra Date: Fri, 2 Oct 2026 13:38:29 +0530 Subject: [PATCH] Gate nodesets on their deployOnAllNodeSets service jobs A deployOnAllNodeSets service runs as a single AnsibleEE Job for the whole deployment. DedupeServices kept such a service in the first nodeSet's service list only, so every later nodeSet lost its ordering relationship to it: dependsOn resolves against the nodeSet's own list and silently skipped the missing entry. Those nodeSets advanced past the shared Job while it was still mutating their hosts. Keep the service in every listing nodeSet's plan, where it acts as a barrier. The deployment starts its Job before the nodeSet loop, once every service ordered before it in each listing nodeSet has completed, and a nodeSet blocks until that Job succeeded. Readiness is read from the Job rather than from conditions, which are rebuilt every reconcile, so a restart mid-deployment resumes from cluster state. A nodeSet that does not list the service neither waits for it nor delays it. Applies to both serial and parallel execution. Related: OSPRH-37768 Signed-off-by: Rabi Mishra --- .../proc_creating-a-custom-service.adoc | 4 +- ...openstackdataplanedeployment_controller.go | 138 ++++++-- internal/dataplane/deployment.go | 119 ++++++- internal/dataplane/global.go | 102 ++++++ internal/dataplane/service.go | 77 +++-- internal/dataplane/util/ansible_execution.go | 15 +- test/functional/dataplane/base_test.go | 31 ++ ...tackdataplanedeployment_controller_test.go | 300 ++++++++++++++---- 8 files changed, 663 insertions(+), 123 deletions(-) create mode 100644 internal/dataplane/global.go diff --git a/docs/assemblies/proc_creating-a-custom-service.adoc b/docs/assemblies/proc_creating-a-custom-service.adoc index 34753ec7d1..fadb8a5701 100644 --- a/docs/assemblies/proc_creating-a-custom-service.adoc +++ b/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/`. -. 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 diff --git a/internal/controller/dataplane/openstackdataplanedeployment_controller.go b/internal/controller/dataplane/openstackdataplanedeployment_controller.go index e9bac1ac19..050b53f5be 100644 --- a/internal/controller/dataplane/openstackdataplanedeployment_controller.go +++ b/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, diff --git a/internal/dataplane/deployment.go b/internal/dataplane/deployment.go index b52a2b1a55..fa22e3999c 100644 --- a/internal/dataplane/deployment.go +++ b/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 { diff --git a/internal/dataplane/global.go b/internal/dataplane/global.go new file mode 100644 index 0000000000..04a15ff627 --- /dev/null +++ b/internal/dataplane/global.go @@ -0,0 +1,102 @@ +/* +Copyright 2026. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package deployment + +import ( + "context" + "slices" + + k8s_errors "k8s.io/apimachinery/pkg/api/errors" + + "github.com/openstack-k8s-operators/lib-common/modules/common/helper" + dataplanev1 "github.com/openstack-k8s-operators/openstack-operator/api/dataplane/v1beta1" + dataplaneutil "github.com/openstack-k8s-operators/openstack-operator/internal/dataplane/util" +) + +// JobState caches AnsibleEE Job outcomes for one reconcile pass. Deployment +// scoped services share one Job name across nodesets. +type JobState map[string]bool + +// Deployment scoped services are looked up without the nodeset label: their +// single Job is not nodeset owned. +func serviceJobCompleted( + ctx context.Context, + helper *helper.Helper, + deployment *dataplanev1.OpenStackDataPlaneDeployment, + service dataplanev1.OpenStackDataPlaneService, + nodeSetName string, + state JobState, +) (bool, error) { + name, labels := dataplaneutil.GetAnsibleExecutionNameAndLabels( + &service, deployment.GetName(), nodeSetName) + if service.Spec.DeployOnAllNodeSets { + labels = dataplaneutil.GetGlobalAnsibleExecutionLabels(service.Name, deployment.GetName()) + } + + if completed, ok := state[name]; ok { + return completed, nil + } + + ansibleJob, err := dataplaneutil.GetAnsibleExecution(ctx, helper, deployment, labels) + if err != nil { + if k8s_errors.IsNotFound(err) { + state[name] = false + return false, nil + } + return false, err + } + + completed := ansibleJob.Status.Succeeded > 0 + state[name] = completed + return completed, nil +} + +// GlobalServiceGateOpen reports whether a deployment scoped service may start. +// Every service ordered before it in each listing nodeset must have completed; +// nodesets that do not list it impose no constraint. +func GlobalServiceGateOpen( + ctx context.Context, + helper *helper.Helper, + plan *ServicePlan, + deployment *dataplanev1.OpenStackDataPlaneDeployment, + service string, + serviceLevels map[string][][]string, + state JobState, +) (bool, error) { + for _, nodeSetName := range plan.NodesetsWithGlobalService[service] { + for _, level := range serviceLevels[nodeSetName] { + if slices.Contains(level, service) { + // levels are ordered, so everything before it completed + break + } + for _, predecessor := range level { + foundService, err := plan.Cache.Get(ctx, helper, predecessor) + if err != nil { + return false, err + } + completed, err := serviceJobCompleted(ctx, helper, deployment, foundService, nodeSetName, state) + if err != nil { + return false, err + } + if !completed { + return false, nil + } + } + } + } + return true, nil +} diff --git a/internal/dataplane/service.go b/internal/dataplane/service.go index 769b6373d7..c86f85e2f8 100644 --- a/internal/dataplane/service.go +++ b/internal/dataplane/service.go @@ -194,49 +194,65 @@ func EnsureServices(ctx context.Context, helper *helper.Helper, instance *datapl return nil } -// DedupeServices dedupes each nodeset's services into an ordered list. +// ServicePlan is the execution plan of one OpenStackDataPlaneDeployment. +// Deployment scoped services (deployOnAllNodeSets) stay in every listing +// nodeset's plan, where their single shared Job acts as a barrier. +type ServicePlan struct { + Nodesets []string + Services map[string][]string + // GlobalServices is in the order the nodesets list them + GlobalServices []string + // NodesetsWithGlobalService is keyed by service name, unlike Services + NodesetsWithGlobalService map[string][]string + Cache *ServiceCache +} + +// DedupeServices builds the ServicePlan of a deployment. func DedupeServices(ctx context.Context, helper *helper.Helper, nodesets []dataplanev1.OpenStackDataPlaneNodeSet, serviceOverride []string, -) (map[string][]string, *ServiceCache, error) { - nodeSetServiceMap := make(map[string][]string) - var globalServices []string - var services []string +) (*ServicePlan, error) { + plan := &ServicePlan{ + Nodesets: make([]string, 0, len(nodesets)), + Services: make(map[string][]string, len(nodesets)), + NodesetsWithGlobalService: make(map[string][]string), + Cache: NewServiceCache(), + } var err error - serviceCache := NewServiceCache() for i := range nodesets { nodeset := &nodesets[i] + plan.Nodesets = append(plan.Nodesets, nodeset.Name) var nodeSetServices []string if len(serviceOverride) != 0 { nodeSetServices = serviceOverride } else { nodeSetServices = nodeset.Spec.Services } - services, globalServices, err = dedupe(ctx, helper, serviceCache, nodeset.Name, nodeSetServices, globalServices) + var services []string + services, err = dedupe(ctx, helper, plan, nodeset.Name, nodeSetServices) if err != nil { - return nil, nil, err + return nil, err } - nodeSetServiceMap[nodeset.Name] = services + plan.Services[nodeset.Name] = services } - helper.GetLogger().Info("Current global services", "services", globalServices) - return nodeSetServiceMap, serviceCache, nil + helper.GetLogger().Info("Current global services", "services", plan.GlobalServices) + return plan, nil } func dedupe(ctx context.Context, helper *helper.Helper, - serviceCache *ServiceCache, + plan *ServicePlan, nodeSetName string, - services []string, globalServices []string) ([]string, []string, error) { + services []string) ([]string, error) { var dedupedServices []string var nodeSetServiceTypes []string - updatedglobalServices := globalServices for _, svc := range services { - service, err := serviceCache.Get(ctx, helper, svc) + service, err := plan.Cache.Get(ctx, helper, svc) if err != nil { if !k8s_errors.IsNotFound(err) { - return dedupedServices, updatedglobalServices, err + return dedupedServices, err } - return dedupedServices, updatedglobalServices, &MissingServiceError{ + return dedupedServices, &MissingServiceError{ NodeSet: nodeSetName, Service: svc, Err: err, @@ -247,17 +263,22 @@ func dedupe(ctx context.Context, helper *helper.Helper, if serviceType == "" { serviceType = service.Name } - if !slices.Contains(nodeSetServiceTypes, serviceType) && !slices.Contains(dedupedServices, svc) { - if service.Spec.DeployOnAllNodeSets { - if !slices.Contains(globalServices, svc) { - updatedglobalServices = append(globalServices, svc) - } else { - continue - } - } - nodeSetServiceTypes = append(nodeSetServiceTypes, serviceType) - dedupedServices = append(dedupedServices, svc) + if slices.Contains(nodeSetServiceTypes, serviceType) || slices.Contains(dedupedServices, svc) { + continue } + if service.Spec.DeployOnAllNodeSets && !slices.Contains(plan.GlobalServices, svc) { + plan.GlobalServices = append(plan.GlobalServices, svc) + } + nodeSetServiceTypes = append(nodeSetServiceTypes, serviceType) + dedupedServices = append(dedupedServices, svc) } - return dedupedServices, updatedglobalServices, nil + + for _, svc := range plan.GlobalServices { + if slices.Contains(dedupedServices, svc) && + !slices.Contains(plan.NodesetsWithGlobalService[svc], nodeSetName) { + plan.NodesetsWithGlobalService[svc] = append(plan.NodesetsWithGlobalService[svc], nodeSetName) + } + } + + return dedupedServices, nil } diff --git a/internal/dataplane/util/ansible_execution.go b/internal/dataplane/util/ansible_execution.go index 2fa60e8983..7f318129e2 100644 --- a/internal/dataplane/util/ansible_execution.go +++ b/internal/dataplane/util/ansible_execution.go @@ -262,12 +262,19 @@ func GetAnsibleExecutionNameAndLabels(service *dataplanev1.OpenStackDataPlaneSer executionName = fmt.Sprintf("%s-%s", prefix, hashSuffix) } - labels := map[string]string{ - "openstackdataplaneservice": service.Name, + labels := GetGlobalAnsibleExecutionLabels(service.Name, deploymentName) + labels["openstackdataplanenodeset"] = nodeSetName + return executionName, labels +} + +// GetGlobalAnsibleExecutionLabels returns the labels of the single AnsibleEE Job +// a deployOnAllNodeSets service produces. It is not nodeset owned, so its +// selector must not carry the nodeset label. +func GetGlobalAnsibleExecutionLabels(serviceName string, deploymentName string) map[string]string { + return map[string]string{ + "openstackdataplaneservice": serviceName, "openstackdataplanedeployment": deploymentName, - "openstackdataplanenodeset": nodeSetName, } - return executionName, labels } // BuildAeeJobSpec builds the job specification for Ansible Execution Environment diff --git a/test/functional/dataplane/base_test.go b/test/functional/dataplane/base_test.go index 8738da1249..a206a25a7d 100644 --- a/test/functional/dataplane/base_test.go +++ b/test/functional/dataplane/base_test.go @@ -422,6 +422,37 @@ func ParallelExecutionDeploymentSpec() map[string]interface{} { } } +// BarrierNodeSetSpec builds a pre-provisioned nodeset for barrier tests. Each +// nodeset needs its own ansibleHost: the nodeset webhook rejects a node address +// that already exists in another nodeset. +func BarrierNodeSetSpec(nodeSetName string, services []string, ansibleHost string) map[string]interface{} { + nodeName := fmt.Sprintf("%s-node-1", nodeSetName) + return map[string]interface{}{ + "services": services, + "nodeTemplate": map[string]interface{}{ + "ansibleSSHPrivateKeySecret": "dataplane-ansible-ssh-private-key-secret", + "ansible": map[string]interface{}{ + "ansibleUser": "cloud-user", + }, + }, + "nodes": map[string]interface{}{ + nodeName: map[string]interface{}{ + "hostName": nodeName, + "networks": []infrav1.IPSetNetwork{ + {Name: "networkinternal", SubnetName: "subnet1"}, + {Name: "ctlplane", SubnetName: "subnet1"}, + }, + "ansible": map[string]interface{}{ + "ansibleHost": ansibleHost, + }, + }, + }, + "secretMaxSize": 1048576, + "tlsEnabled": true, + "preProvisioned": true, + } +} + func DefaultNetConfigSpec() map[string]interface{} { return map[string]interface{}{ "networks": []map[string]interface{}{{ diff --git a/test/functional/dataplane/openstackdataplanedeployment_controller_test.go b/test/functional/dataplane/openstackdataplanedeployment_controller_test.go index b1ba9cbb14..18a23eef15 100644 --- a/test/functional/dataplane/openstackdataplanedeployment_controller_test.go +++ b/test/functional/dataplane/openstackdataplanedeployment_controller_test.go @@ -898,8 +898,7 @@ var _ = Describe("Dataplane Deployment Test", func() { }, th.Timeout, th.Interval).Should(Succeed()) - // Create all services necessary for deployment - for _, serviceName := range nodeSetAlpha.Spec.Services { + markJobSucceeded := func(serviceName string, nodeSetName string) { dataplaneServiceName := types.NamespacedName{ Name: serviceName, Namespace: namespace, @@ -907,10 +906,8 @@ var _ = Describe("Dataplane Deployment Test", func() { service := GetService(dataplaneServiceName) deployment := GetDataplaneDeployment(dataplaneMultiNodesetDeploymentName) aeeName, _ := dataplaneutil.GetAnsibleExecutionNameAndLabels( - service, deployment.GetName(), nodeSetAlpha.GetName()) - //Retrieve service AnsibleEE and set JobStatus to Successful + service, deployment.GetName(), nodeSetName) Eventually(func(g Gomega) { - // Make an AnsibleEE name for each service ansibleeeName := types.NamespacedName{ Name: aeeName, Namespace: dataplaneMultiNodesetDeploymentName.Namespace, @@ -921,30 +918,16 @@ var _ = Describe("Dataplane Deployment Test", func() { }, th.Timeout, th.Interval).Should(Succeed()) } - servicesExcludingGlobal := []string{"foo-service", "foo-update-service"} - // Create all services necessary for deployment - for _, serviceName := range servicesExcludingGlobal { - dataplaneServiceName := types.NamespacedName{ - Name: serviceName, - Namespace: namespace, - } - service := GetService(dataplaneServiceName) - deployment := GetDataplaneDeployment(dataplaneMultiNodesetDeploymentName) - aeeName, _ := dataplaneutil.GetAnsibleExecutionNameAndLabels( - service, deployment.GetName(), nodeSetBeta.GetName()) - - //Retrieve service AnsibleEE and set JobStatus to Successful - Eventually(func(g Gomega) { - // Make an AnsibleEE name for each service - ansibleeeName := types.NamespacedName{ - Name: aeeName, - Namespace: dataplaneMultiNodesetDeploymentName.Namespace, - } - ansibleEE := GetAnsibleee(ansibleeeName) - ansibleEE.Status.Succeeded = 1 - g.Expect(th.K8sClient.Status().Update(th.Ctx, ansibleEE)).To(Succeed()) - }, th.Timeout, th.Interval).Should(Succeed()) + // The deployment scoped Job starts once every nodeset that lists it + // finished the services it ordered before the barrier + for _, serviceName := range []string{"foo-service", "foo-update-service"} { + markJobSucceeded(serviceName, nodeSetAlpha.GetName()) } + markJobSucceeded("foo-service", nodeSetBeta.GetName()) + + markJobSucceeded("global-service", nodeSetAlpha.GetName()) + + markJobSucceeded("foo-update-service", nodeSetBeta.GetName()) th.ExpectCondition( dataplaneMultiNodesetDeploymentName, @@ -1434,8 +1417,7 @@ var _ = Describe("Dataplane Deployment Test", func() { }, th.Timeout, th.Interval).Should(Succeed()) - // Create all services necessary for deployment - for _, serviceName := range []string{"foo-service", "global-service", "foo-update-service"} { + markJobSucceeded := func(serviceName string, nodeSetName string) { dataplaneServiceName := types.NamespacedName{ Name: serviceName, Namespace: namespace, @@ -1443,10 +1425,8 @@ var _ = Describe("Dataplane Deployment Test", func() { service := GetService(dataplaneServiceName) deployment := GetDataplaneDeployment(dataplaneMultiNodesetDeploymentName) aeeName, _ := dataplaneutil.GetAnsibleExecutionNameAndLabels( - service, deployment.GetName(), nodeSetAlpha.GetName()) - //Retrieve service AnsibleEE and set JobStatus to Successful + service, deployment.GetName(), nodeSetName) Eventually(func(g Gomega) { - // Make an AnsibleEE name for each service ansibleeeName := types.NamespacedName{ Name: aeeName, Namespace: dataplaneMultiNodesetDeploymentName.Namespace, @@ -1457,29 +1437,12 @@ var _ = Describe("Dataplane Deployment Test", func() { }, th.Timeout, th.Interval).Should(Succeed()) } - servicesExcludingGlobal := []string{"foo-service", "foo-update-service"} - // Create all services necessary for deployment - for _, serviceName := range servicesExcludingGlobal { - dataplaneServiceName := types.NamespacedName{ - Name: serviceName, - Namespace: namespace, - } - service := GetService(dataplaneServiceName) - deployment := GetDataplaneDeployment(dataplaneMultiNodesetDeploymentName) - aeeName, _ := dataplaneutil.GetAnsibleExecutionNameAndLabels( - service, deployment.GetName(), nodeSetBeta.GetName()) - - //Retrieve service AnsibleEE and set JobStatus to Successful - Eventually(func(g Gomega) { - // Make an AnsibleEE name for each service - ansibleeeName := types.NamespacedName{ - Name: aeeName, - Namespace: dataplaneMultiNodesetDeploymentName.Namespace, - } - ansibleEE := GetAnsibleee(ansibleeeName) - ansibleEE.Status.Succeeded = 1 - g.Expect(th.K8sClient.Status().Update(th.Ctx, ansibleEE)).To(Succeed()) - }, th.Timeout, th.Interval).Should(Succeed()) + for _, nodeSetName := range []string{nodeSetAlpha.GetName(), nodeSetBeta.GetName()} { + markJobSucceeded("foo-service", nodeSetName) + } + markJobSucceeded("global-service", nodeSetAlpha.GetName()) + for _, nodeSetName := range []string{nodeSetAlpha.GetName(), nodeSetBeta.GetName()} { + markJobSucceeded("foo-update-service", nodeSetName) } th.ExpectCondition( @@ -2067,4 +2030,229 @@ var _ = Describe("Dataplane Deployment Test", func() { }) }) + When("A dataplaneDeployment lists a deployment scoped service in two nodesets", func() { + var barrierDeploymentName types.NamespacedName + var barrierAlphaName types.NamespacedName + var barrierBetaName types.NamespacedName + + barrierServices := []string{"svc-first", "svc-global", "svc-last"} + + BeforeEach(func() { + barrierDeploymentName = types.NamespacedName{Name: "barrier-deployment", Namespace: namespace} + barrierAlphaName = types.NamespacedName{Name: "barrier-alpha", Namespace: namespace} + barrierBetaName = types.NamespacedName{Name: "barrier-beta", Namespace: namespace} + CreateSSHSecret(dataplaneSSHSecretName) + CreateCABundleSecret(caBundleSecretName) + + DeferCleanup(th.DeleteInstance, CreateDataPlaneServiceFromSpec( + types.NamespacedName{Name: "svc-first", Namespace: namespace}, + map[string]interface{}{"edpmServiceType": "svc-first"})) + DeferCleanup(th.DeleteInstance, CreateDataPlaneServiceFromSpec( + types.NamespacedName{Name: "svc-global", Namespace: namespace}, + map[string]interface{}{"edpmServiceType": "svc-global", "deployOnAllNodeSets": true})) + DeferCleanup(th.DeleteInstance, CreateDataPlaneServiceFromSpec( + types.NamespacedName{Name: "svc-last", Namespace: namespace}, + map[string]interface{}{"edpmServiceType": "svc-last"})) + + DeferCleanup(th.DeleteInstance, CreateNetConfig(dataplaneNetConfigName, DefaultNetConfigSpec())) + DeferCleanup(th.DeleteInstance, CreateDNSMasq(dnsMasqName, DefaultDNSMasqSpec())) + SimulateDNSMasqComplete(dnsMasqName) + + for idx, nodeSetName := range []types.NamespacedName{barrierAlphaName, barrierBetaName} { + DeferCleanup(th.DeleteInstance, CreateDataplaneNodeSet( + nodeSetName, BarrierNodeSetSpec(nodeSetName.Name, barrierServices, + fmt.Sprintf("192.168.122.%d", 100+idx)))) + SimulateIPSetComplete(types.NamespacedName{ + Name: fmt.Sprintf("%s-node-1", nodeSetName.Name), Namespace: namespace}) + SimulateDNSDataComplete(nodeSetName) + } + + DeferCleanup(th.DeleteInstance, CreateDataplaneDeployment(barrierDeploymentName, + map[string]interface{}{ + "nodeSets": []string{barrierAlphaName.Name, barrierBetaName.Name}, + })) + }) + + It("should run the shared Job once after every listing nodeset reached the barrier", func() { + nodeSetJobName := func(service, nodeSet string) types.NamespacedName { + return types.NamespacedName{ + Name: fmt.Sprintf("%s-%s-%s", service, barrierDeploymentName.Name, nodeSet), + Namespace: namespace, + } + } + globalJobName := types.NamespacedName{ + Name: fmt.Sprintf("svc-global-%s", barrierDeploymentName.Name), + Namespace: namespace, + } + getJob := func(name types.NamespacedName) *batchv1.Job { + job := &batchv1.Job{} + err := k8sClient.Get(ctx, name, job) + if k8s_errors.IsNotFound(err) { + return nil + } + Expect(err).NotTo(HaveOccurred()) + return job + } + completeJob := func(name types.NamespacedName) { + Eventually(func(g Gomega) { + job := GetAnsibleee(name) + job.Status.Succeeded = 1 + g.Expect(k8sClient.Status().Update(ctx, job)).To(Succeed()) + }, timeout, interval).Should(Succeed()) + } + + // Both nodesets start the service ordered before the barrier + Eventually(func(g Gomega) { + g.Expect(getJob(nodeSetJobName("svc-first", barrierAlphaName.Name))).ToNot(BeNil()) + g.Expect(getJob(nodeSetJobName("svc-first", barrierBetaName.Name))).ToNot(BeNil()) + }, timeout, interval).Should(Succeed()) + + // The shared Job must not start while a nodeset still has work + // ordered before the barrier + Consistently(func() *batchv1.Job { + return getJob(globalJobName) + }, "5s", interval).Should(BeNil()) + + completeJob(nodeSetJobName("svc-first", barrierAlphaName.Name)) + completeJob(nodeSetJobName("svc-first", barrierBetaName.Name)) + + // One Job for the whole deployment, not one per nodeset + Eventually(func(g Gomega) { + g.Expect(getJob(globalJobName)).ToNot(BeNil()) + g.Expect(getJob(types.NamespacedName{ + Name: fmt.Sprintf("svc-global-%s-%s", barrierDeploymentName.Name, barrierBetaName.Name), + Namespace: namespace, + })).To(BeNil()) + }, timeout, interval).Should(Succeed()) + + // Neither nodeset advances past the barrier before it completed + Consistently(func(g Gomega) { + g.Expect(getJob(nodeSetJobName("svc-last", barrierAlphaName.Name))).To(BeNil()) + g.Expect(getJob(nodeSetJobName("svc-last", barrierBetaName.Name))).To(BeNil()) + }, "5s", interval).Should(Succeed()) + + // A blocked nodeset reports what it is waiting on + Eventually(func(g Gomega) { + deployment := GetDataplaneDeployment(barrierDeploymentName) + for _, nodeSetName := range []string{barrierAlphaName.Name, barrierBetaName.Name} { + nsConditions := deployment.Status.NodeSetConditions[nodeSetName] + c := nsConditions.Get(dataplanev1.NodeSetDeploymentReadyCondition) + g.Expect(c).ToNot(BeNil()) + g.Expect(c.Message).To(ContainSubstring("svc-global")) + } + }, timeout, interval).Should(Succeed()) + + completeJob(globalJobName) + + Eventually(func(g Gomega) { + g.Expect(getJob(nodeSetJobName("svc-last", barrierAlphaName.Name))).ToNot(BeNil()) + g.Expect(getJob(nodeSetJobName("svc-last", barrierBetaName.Name))).ToNot(BeNil()) + }, timeout, interval).Should(Succeed()) + + completeJob(nodeSetJobName("svc-last", barrierAlphaName.Name)) + completeJob(nodeSetJobName("svc-last", barrierBetaName.Name)) + + th.ExpectCondition( + barrierDeploymentName, + ConditionGetterFunc(DataplaneDeploymentConditionGetter), + condition.DeploymentReadyCondition, + corev1.ConditionTrue, + ) + }) + }) + + When("A nodeset does not list the deployment scoped service", func() { + var freeDeploymentName types.NamespacedName + var freeAlphaName types.NamespacedName + var freeBetaName types.NamespacedName + + BeforeEach(func() { + freeDeploymentName = types.NamespacedName{Name: "free-deployment", Namespace: namespace} + freeAlphaName = types.NamespacedName{Name: "free-alpha", Namespace: namespace} + freeBetaName = types.NamespacedName{Name: "free-beta", Namespace: namespace} + CreateSSHSecret(dataplaneSSHSecretName) + CreateCABundleSecret(caBundleSecretName) + + DeferCleanup(th.DeleteInstance, CreateDataPlaneServiceFromSpec( + types.NamespacedName{Name: "svc-first", Namespace: namespace}, + map[string]interface{}{"edpmServiceType": "svc-first"})) + DeferCleanup(th.DeleteInstance, CreateDataPlaneServiceFromSpec( + types.NamespacedName{Name: "svc-global", Namespace: namespace}, + map[string]interface{}{"edpmServiceType": "svc-global", "deployOnAllNodeSets": true})) + DeferCleanup(th.DeleteInstance, CreateDataPlaneServiceFromSpec( + types.NamespacedName{Name: "svc-last", Namespace: namespace}, + map[string]interface{}{"edpmServiceType": "svc-last"})) + + DeferCleanup(th.DeleteInstance, CreateNetConfig(dataplaneNetConfigName, DefaultNetConfigSpec())) + DeferCleanup(th.DeleteInstance, CreateDNSMasq(dnsMasqName, DefaultDNSMasqSpec())) + SimulateDNSMasqComplete(dnsMasqName) + + DeferCleanup(th.DeleteInstance, CreateDataplaneNodeSet(freeAlphaName, + BarrierNodeSetSpec(freeAlphaName.Name, []string{"svc-first", "svc-global", "svc-last"}, "192.168.122.100"))) + SimulateIPSetComplete(types.NamespacedName{ + Name: fmt.Sprintf("%s-node-1", freeAlphaName.Name), Namespace: namespace}) + SimulateDNSDataComplete(freeAlphaName) + + DeferCleanup(th.DeleteInstance, CreateDataplaneNodeSet(freeBetaName, + BarrierNodeSetSpec(freeBetaName.Name, []string{"svc-first", "svc-last"}, "192.168.122.101"))) + SimulateIPSetComplete(types.NamespacedName{ + Name: fmt.Sprintf("%s-node-1", freeBetaName.Name), Namespace: namespace}) + SimulateDNSDataComplete(freeBetaName) + + DeferCleanup(th.DeleteInstance, CreateDataplaneDeployment(freeDeploymentName, + map[string]interface{}{ + "nodeSets": []string{freeAlphaName.Name, freeBetaName.Name}, + "useParallelExecution": true, + })) + }) + + It("should not hold that nodeset behind the barrier", func() { + nodeSetJobName := func(service, nodeSet string) types.NamespacedName { + return types.NamespacedName{ + Name: fmt.Sprintf("%s-%s-%s", service, freeDeploymentName.Name, nodeSet), + Namespace: namespace, + } + } + globalJobName := types.NamespacedName{ + Name: fmt.Sprintf("svc-global-%s", freeDeploymentName.Name), + Namespace: namespace, + } + getJob := func(name types.NamespacedName) *batchv1.Job { + job := &batchv1.Job{} + err := k8sClient.Get(ctx, name, job) + if k8s_errors.IsNotFound(err) { + return nil + } + Expect(err).NotTo(HaveOccurred()) + return job + } + completeJob := func(name types.NamespacedName) { + Eventually(func(g Gomega) { + job := GetAnsibleee(name) + job.Status.Succeeded = 1 + g.Expect(k8sClient.Status().Update(ctx, job)).To(Succeed()) + }, timeout, interval).Should(Succeed()) + } + + Eventually(func(g Gomega) { + g.Expect(getJob(nodeSetJobName("svc-first", freeAlphaName.Name))).ToNot(BeNil()) + g.Expect(getJob(nodeSetJobName("svc-first", freeBetaName.Name))).ToNot(BeNil()) + }, timeout, interval).Should(Succeed()) + + completeJob(nodeSetJobName("svc-first", freeAlphaName.Name)) + completeJob(nodeSetJobName("svc-first", freeBetaName.Name)) + + // free-beta never listed svc-global, so it keeps advancing while the + // shared Job for free-alpha has not completed + Eventually(func(g Gomega) { + g.Expect(getJob(globalJobName)).ToNot(BeNil()) + g.Expect(getJob(nodeSetJobName("svc-last", freeBetaName.Name))).ToNot(BeNil()) + }, timeout, interval).Should(Succeed()) + + Consistently(func() *batchv1.Job { + return getJob(nodeSetJobName("svc-last", freeAlphaName.Name)) + }, "5s", interval).Should(BeNil()) + }) + }) + })