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()) + }) + }) + })