diff --git a/cmd/virtwork/main.go b/cmd/virtwork/main.go index 3cc6a44..6be3a0c 100644 --- a/cmd/virtwork/main.go +++ b/cmd/virtwork/main.go @@ -14,23 +14,16 @@ import ( "os/signal" "strings" "syscall" - "time" "github.com/spf13/cobra" - "golang.org/x/sync/errgroup" - kubevirtv1 "kubevirt.io/api/core/v1" "sigs.k8s.io/controller-runtime/pkg/client" - sigyaml "sigs.k8s.io/yaml" "github.com/opdev/virtwork/internal/audit" "github.com/opdev/virtwork/internal/cleanup" "github.com/opdev/virtwork/internal/cluster" "github.com/opdev/virtwork/internal/config" - "github.com/opdev/virtwork/internal/constants" "github.com/opdev/virtwork/internal/logging" - "github.com/opdev/virtwork/internal/resources" - "github.com/opdev/virtwork/internal/vm" - "github.com/opdev/virtwork/internal/wait" + "github.com/opdev/virtwork/internal/orchestrator" "github.com/opdev/virtwork/internal/workloads" ) @@ -40,9 +33,6 @@ var ( date = "" ) -// ErrReadinessCheck: readiness check failed -var ErrReadinessCheck = errors.New("readiness check failed") - func main() { if err := newRootCmd().Execute(); err != nil { os.Exit(1) @@ -115,7 +105,6 @@ func newCleanupCmd() *cobra.Command { return cmd } -// initAuditor creates the appropriate Auditor based on configuration flags. func initAuditor(cmd *cobra.Command, cfg *config.Config) (audit.Auditor, error) { noAudit, _ := cmd.Flags().GetBool("no-audit") if noAudit || !cfg.AuditEnabled { @@ -130,67 +119,15 @@ func initAuditor(cmd *cobra.Command, cfg *config.Config) (audit.Auditor, error) return audit.NewSQLiteAuditor(dbPath) } -// vmPlan describes a single VM to be created during orchestration. -type vmPlan struct { - workload workloads.Workload - vmSpec *vm.VMSpecOpts - vmName string - component string - role string -} - -// namespaceDataVolumes appends the VM name to DataVolume template names and -// updates corresponding volume references to prevent name collisions when -// deploying multiple VMs of the same workload type. -func namespaceDataVolumes( - baseTemplates []kubevirtv1.DataVolumeTemplateSpec, - baseVolumes []kubevirtv1.Volume, - vmName string, -) ([]kubevirtv1.DataVolumeTemplateSpec, []kubevirtv1.Volume) { - if len(baseTemplates) == 0 { - return baseTemplates, baseVolumes - } - - // Build a map of original DataVolume names to namespaced names - nameMap := make(map[string]string, len(baseTemplates)) - templates := make([]kubevirtv1.DataVolumeTemplateSpec, len(baseTemplates)) - for i, tmpl := range baseTemplates { - oldName := tmpl.Name - newName := fmt.Sprintf("%s-%s", oldName, vmName) - nameMap[oldName] = newName - - templates[i] = tmpl - templates[i].Name = newName - } - - // Update volume references to use the namespaced names - volumes := make([]kubevirtv1.Volume, len(baseVolumes)) - for i, vol := range baseVolumes { - volumes[i] = vol - if vol.DataVolume != nil { - if newName, ok := nameMap[vol.DataVolume.Name]; ok { - volumes[i].DataVolume = &kubevirtv1.DataVolumeSource{ - Name: newName, - } - } - } - } - - return templates, volumes -} - -// runE is the main orchestration flow for the "run" subcommand. func runE(cmd *cobra.Command, args []string) error { cfg, err := config.LoadConfig(cmd) if err != nil { return fmt.Errorf("loading config: %w", err) } - // Initialize logger verbose, _ := cmd.Flags().GetBool("verbose") logger := logging.NewLogger(cmd.OutOrStdout(), verbose) - // Initialize auditor auditor, err := initAuditor(cmd, cfg) if err != nil { return fmt.Errorf("initializing auditor: %w", err) @@ -202,7 +139,6 @@ func runE(cmd *cobra.Command, args []string) error { ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() - // Connect to cluster early (unless dry-run) to capture context for audit var c client.Client if !cfg.DryRun { var contextName string @@ -213,7 +149,6 @@ func runE(cmd *cobra.Command, args []string) error { cfg.ClusterContext = contextName } - // Start audit execution cmdName := "run" if cfg.DryRun { cmdName = "dry-run" @@ -234,403 +169,31 @@ func runE(cmd *cobra.Command, args []string) error { Message: fmt.Sprintf("Starting %s with run-id %s", cmdName, runID), }) - // Determine which workloads to deploy workloadNames, _ := cmd.Flags().GetStringSlice("workloads") vmCountFlag, _ := cmd.Flags().GetInt("vm-count") - registry := workloads.DefaultRegistry() - registryOpts := []workloads.Option{ - workloads.WithNamespace(cfg.Namespace), - workloads.WithSSHCredentials(cfg.SSHUser, cfg.SSHPassword, cfg.SSHAuthorizedKeys), - workloads.WithDataDiskSize(cfg.DataDiskSize), - } - - // Build workload instances - var plans []vmPlan - var vmNames []string - auditWorkloadIDs := make(map[string]int64) // workload name -> audit workload ID - workloadInstances := make(map[string]workloads.Workload) // workload name -> configured instance - - for _, name := range workloadNames { - // Check if workload is explicitly disabled in YAML config - if fileCfg, ok := cfg.Workloads[name]; ok { - if fileCfg.Enabled != nil && !*fileCfg.Enabled { - _ = auditor.RecordEvent(ctx, execID, audit.EventRecord{ - EventType: "workload_skipped", - Message: fmt.Sprintf("Workload %q disabled via config (enabled: false)", name), - }) - logger.Info("workload skipped", - slog.String("workload", name), - slog.String("reason", "disabled in config")) - continue - } - } - - wlCfg := config.WorkloadConfig{ - Enabled: new(true), - VMCount: vmCountFlag, - CPUCores: cfg.CPUCores, - Memory: cfg.Memory, - } - // Override with per-workload config from YAML if present - if fileCfg, ok := cfg.Workloads[name]; ok { - if fileCfg.CPUCores > 0 { - wlCfg.CPUCores = fileCfg.CPUCores - } - if fileCfg.Memory != "" { - wlCfg.Memory = fileCfg.Memory - } - if fileCfg.VMCount > 0 { - wlCfg.VMCount = fileCfg.VMCount - } - if len(fileCfg.Params) > 0 { - wlCfg.Params = fileCfg.Params - } - } - - w, err := registry.Get(name, wlCfg, registryOpts...) - if err != nil { - return fmt.Errorf("creating workload %q: %w", name, err) - } - workloadInstances[name] = w - - vmCount := w.VMCount() - res := w.VMResources() - - // Record workload in audit - dvTemplatesForAudit, err := w.DataVolumeTemplates() - if err != nil { - return fmt.Errorf("building data volume templates for %q: %w", name, err) - } - wlID, _ := auditor.RecordWorkload(ctx, execID, audit.WorkloadRecord{ - WorkloadType: name, - Enabled: true, - VMCount: vmCount, - CPUCores: res.CPUCores, - Memory: res.Memory, - HasDataDisk: len(dvTemplatesForAudit) > 0, - DataDiskSize: cfg.DataDiskSize, - RequiresService: w.RequiresService(), - }) - auditWorkloadIDs[name] = wlID - - if multiVM, isMulti := w.(workloads.MultiVMWorkload); !isMulti { - userdata, err := w.CloudInitUserdata() - if err != nil { - return fmt.Errorf("generating cloud-init for %q: %w", name, err) - } - - for i := range vmCount { - vmName := fmt.Sprintf("virtwork-%s-%d", name, i) - - // Namespace DataVolume names to avoid collisions across VMs - dvts, err := w.DataVolumeTemplates() - if err != nil { - return fmt.Errorf("building data volume templates for %q vm %s: %w", name, vmName, err) - } - dvTemplates, extraVols := namespaceDataVolumes( - dvts, - w.ExtraVolumes(), - vmName, - ) - - plans = append(plans, vmPlan{ - workload: w, - component: name, - vmName: vmName, - vmSpec: &vm.VMSpecOpts{ - Name: vmName, - Namespace: cfg.Namespace, - ContainerDiskImage: cfg.ContainerDiskImage, - CloudInitUserdata: userdata, - CPUCores: res.CPUCores, - Memory: res.Memory, - Labels: map[string]string{ - constants.LabelAppName: fmt.Sprintf("virtwork-%s", name), - constants.LabelManagedBy: constants.ManagedByValue, - constants.LabelComponent: name, - constants.LabelRunID: runID, - }, - ExtraDisks: w.ExtraDisks(), - ExtraVolumes: extraVols, - DataVolumeTemplates: dvTemplates, - }, - }) - vmNames = append(vmNames, vmName) - } - } else { - // Multi-VM workload — use UserdataForRole - roles := multiVM.Roles() - perRole := vmCount / len(roles) - if remainder := vmCount % len(roles); remainder != 0 { - logger.Warn("VM count not evenly divisible by roles; remainder VMs will not be created", - slog.String("workload", name), - slog.Int("requested", vmCount), - slog.Int("per_role", perRole), - slog.Int("dropped", remainder)) - } - for _, role := range roles { - userdata, err := multiVM.UserdataForRole(role, cfg.Namespace) - if err != nil { - return fmt.Errorf("generating cloud-init for %q role %q: %w", name, role, err) - } - - for i := range perRole { - vmName := fmt.Sprintf("virtwork-%s-%s-%d", name, role, i) - labels := map[string]string{ - constants.LabelAppName: fmt.Sprintf("virtwork-%s", name), - constants.LabelManagedBy: constants.ManagedByValue, - constants.LabelComponent: name, - constants.LabelRunID: runID, - "virtwork/role": role, - } - - // Namespace DataVolume names to avoid collisions across VMs - dvts, err := w.DataVolumeTemplates() - if err != nil { - return fmt.Errorf( - "building data volume templates for %q role %q vm %s: %w", name, role, vmName, err, - ) - } - dvTemplates, extraVols := namespaceDataVolumes( - dvts, - w.ExtraVolumes(), - vmName, - ) - - plans = append(plans, vmPlan{ - workload: w, - component: name, - vmName: vmName, - role: role, - vmSpec: &vm.VMSpecOpts{ - Name: vmName, - Namespace: cfg.Namespace, - ContainerDiskImage: cfg.ContainerDiskImage, - CloudInitUserdata: userdata, - CPUCores: res.CPUCores, - Memory: res.Memory, - Labels: labels, - ExtraDisks: w.ExtraDisks(), - ExtraVolumes: extraVols, - DataVolumeTemplates: dvTemplates, - }, - }) - vmNames = append(vmNames, vmName) - } - } - } - } - - // Update audit with total counts - _ = auditor.RecordEvent(ctx, execID, audit.EventRecord{ - EventType: "execution_started", - Message: fmt.Sprintf("Planned %d VMs across %d workloads", len(plans), len(workloadNames)), - }) - - // Dry-run: print specs and return - if cfg.DryRun { - if err := printDryRun(logger, plans); err != nil { - return err - } - _ = auditor.CompleteExecution(ctx, execID, "success", "") - err = nil // clear for defer - return nil - } - - // Ensure namespace exists - if err := resources.EnsureNamespace(ctx, c, cfg.Namespace, map[string]string{ - constants.LabelManagedBy: constants.ManagedByValue, - }); err != nil { - return fmt.Errorf("ensuring namespace %q: %w", cfg.Namespace, err) - } - logger.Info("namespace ensured", slog.String("namespace", cfg.Namespace)) - - // Create services before VMs (DNS must resolve for client VMs) - servicesCreated := 0 - for name, w := range workloadInstances { - if w.RequiresService() { - svc := w.ServiceSpec() - if svc != nil { - // Add run-id label to service - if svc.Labels == nil { - svc.Labels = make(map[string]string) - } - svc.Labels[constants.LabelRunID] = runID - - if err := resources.CreateService(ctx, c, svc); err != nil { - return fmt.Errorf("creating service for %q: %w", name, err) - } - servicesCreated++ - logger.Info("service created", - slog.String("service_name", svc.Name), - slog.String("namespace", svc.Namespace)) - - _, _ = auditor.RecordResource(ctx, execID, audit.ResourceRecord{ - ResourceType: "Service", - ResourceName: svc.Name, - Namespace: svc.Namespace, - }) - _ = auditor.RecordEvent(ctx, execID, audit.EventRecord{ - EventType: "service_created", - Message: fmt.Sprintf("Service %s created", svc.Name), - }) - } - } - } - - // Create cloud-init secrets before VMs - secretsCreated := 0 - for i := range plans { - secretName := plans[i].vmName + "-cloudinit" - secretLabels := map[string]string{ - constants.LabelAppName: plans[i].vmSpec.Labels[constants.LabelAppName], - constants.LabelManagedBy: constants.ManagedByValue, - constants.LabelComponent: plans[i].component, - constants.LabelRunID: runID, - } - if err := resources.CreateCloudInitSecret(ctx, c, secretName, - cfg.Namespace, plans[i].vmSpec.CloudInitUserdata, secretLabels); err != nil { - return fmt.Errorf("creating cloud-init secret for %q: %w", plans[i].vmName, err) - } - plans[i].vmSpec.CloudInitSecretName = secretName - secretsCreated++ - logger.Info("secret created", - slog.String("secret_name", secretName), - slog.String("namespace", cfg.Namespace)) - - _, _ = auditor.RecordResource(ctx, execID, audit.ResourceRecord{ - ResourceType: "Secret", - ResourceName: secretName, - Namespace: cfg.Namespace, - }) - } - - // Create VMs concurrently via errgroup - g, gctx := errgroup.WithContext(ctx) - for _, plan := range plans { - p := plan // capture loop variable - g.Go(func() error { - vmObj, err := vm.BuildVMSpec(*p.vmSpec) - if err != nil { - return fmt.Errorf("building VM spec for %q: %w", p.vmName, err) - } - if err := vm.CreateVM(gctx, c, vmObj); err != nil { - _ = auditor.RecordEvent(ctx, execID, audit.EventRecord{ - EventType: "vm_failed", - Message: fmt.Sprintf("Failed to create VM %s", p.vmName), - ErrorDetail: err.Error(), - }) - return fmt.Errorf("creating VM %q: %w", p.vmName, err) - } - logger.Info("vm created", - slog.String("vm_name", p.vmName), - slog.String("namespace", cfg.Namespace), - slog.String("workload", p.component)) - - wlID := auditWorkloadIDs[p.component] - _, _ = auditor.RecordVM(ctx, execID, wlID, audit.VMRecord{ - VMName: p.vmName, - Namespace: cfg.Namespace, - Component: p.component, - Role: p.role, - CPUCores: p.vmSpec.CPUCores, - Memory: p.vmSpec.Memory, - ContainerDiskImage: p.vmSpec.ContainerDiskImage, - HasDataDisk: len(p.vmSpec.DataVolumeTemplates) > 0, - DataDiskSize: cfg.DataDiskSize, - }) - _ = auditor.RecordEvent(ctx, execID, audit.EventRecord{ - EventType: "vm_created", - Message: fmt.Sprintf("VM %s created", p.vmName), - }) - return nil - }) - } - if err := g.Wait(); err != nil { - for _, wlID := range auditWorkloadIDs { - _ = auditor.UpdateWorkloadStatus(ctx, wlID, "failed") - } - return fmt.Errorf("creating VMs: %w", err) - } - - // Build VM-name → workload-component mapping for readiness tracking - vmToComponent := make(map[string]string, len(plans)) - for _, p := range plans { - vmToComponent[p.vmName] = p.component - } - - // Wait for readiness - if cfg.WaitForReady { - timeout := time.Duration(cfg.ReadyTimeoutSeconds) * time.Second - logger.Info("waiting for VMs to become ready", - slog.Int("vm_count", len(vmNames)), - slog.Duration("timeout", timeout)) - results := wait.WaitForAllVMsReady(ctx, c, logger, vmNames, cfg.Namespace, - timeout, constants.DefaultPollInterval) - - failures := 0 - failedWorkloads := make(map[string]bool) - for name, err := range results { - if err != nil { - logger.Error("vm readiness check failed", - slog.String("vm_name", name), - slog.String("error", err.Error())) - failures++ - _ = auditor.RecordEvent(ctx, execID, audit.EventRecord{ - EventType: "vm_timeout", - Message: fmt.Sprintf("VM %s failed readiness check", name), - ErrorDetail: err.Error(), - }) - if comp, ok := vmToComponent[name]; ok { - failedWorkloads[comp] = true - } - } else { - _ = auditor.RecordEvent(ctx, execID, audit.EventRecord{ - EventType: "vm_ready", - Message: fmt.Sprintf("VM %s is ready", name), - }) - } - } - if failures > 0 { - for comp, wlID := range auditWorkloadIDs { - if failedWorkloads[comp] { - _ = auditor.UpdateWorkloadStatus(ctx, wlID, "failed") - } else { - _ = auditor.UpdateWorkloadStatus(ctx, wlID, "created") - } - } - return fmt.Errorf("%d of %d VMs failed; %w", failures, len(vmNames), ErrReadinessCheck) - } - logger.Info("all VMs ready", slog.Int("vm_count", len(vmNames))) - } - - // Mark all workloads as created - for _, wlID := range auditWorkloadIDs { - _ = auditor.UpdateWorkloadStatus(ctx, wlID, "created") + ro := orchestrator.NewRunOrchestrator(logger, c, cfg, auditor, cmd.OutOrStdout()) + result, err := ro.Run(ctx, execID, runID, workloadNames, vmCountFlag) + if err != nil { + return err } - // Complete audit _ = auditor.CompleteExecution(ctx, execID, "success", "") - err = nil // clear for defer + err = nil - // Print summary - printSummary(logger, len(plans), servicesCreated, secretsCreated, cfg, runID) + printSummary(logger, result, cfg) return nil } -// cleanupE is the cleanup flow for the "cleanup" subcommand. func cleanupE(cmd *cobra.Command, args []string) error { cfg, err := config.LoadConfig(cmd) if err != nil { return fmt.Errorf("loading config: %w", err) } - // Initialize logger verbose, _ := cmd.Flags().GetBool("verbose") logger := logging.NewLogger(cmd.OutOrStdout(), verbose) - // Initialize auditor auditor, err := initAuditor(cmd, cfg) if err != nil { return fmt.Errorf("initializing auditor: %w", err) @@ -642,11 +205,9 @@ func cleanupE(cmd *cobra.Command, args []string) error { ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() - // Get cleanup flags deleteNS, _ := cmd.Flags().GetBool("delete-namespace") targetRunID, _ := cmd.Flags().GetString("run-id") - // Set cleanup mode for audit if cfg.DryRun { cfg.CleanupMode = "dry-run" } else if targetRunID != "" { @@ -655,14 +216,12 @@ func cleanupE(cmd *cobra.Command, args []string) error { cfg.CleanupMode = "all" } - // Connect to cluster early to capture context for audit c, contextName, err := cluster.Connect(cluster.ResolveKubeconfigPath(cfg.KubeconfigPath)) if err != nil { return fmt.Errorf("connecting to cluster: %w", err) } cfg.ClusterContext = contextName - // Start audit execution cmdName := "cleanup" if cfg.DryRun { cmdName = "cleanup --dry-run" @@ -679,13 +238,9 @@ func cleanupE(cmd *cobra.Command, args []string) error { } }() - _ = auditor.RecordEvent(ctx, execID, audit.EventRecord{ - EventType: "cleanup_started", - Message: fmt.Sprintf("Started %s: (namespace: %s, run-id filter: %q)", cmdName, cfg.Namespace, targetRunID), - }) + co := orchestrator.NewCleanupOrchestrator(logger, c, cfg, auditor, cmd.OutOrStdout()) - // Preview resources before deletion - preview, err := cleanup.PreviewCleanup(ctx, c, cfg, targetRunID) + preview, err := co.Preview(ctx, execID, targetRunID) if err != nil { return fmt.Errorf("previewing cleanup: %w", err) } @@ -721,31 +276,13 @@ func cleanupE(cmd *cobra.Command, args []string) error { } } - result, err := cleanup.CleanupAll(ctx, c, cfg, deleteNS, targetRunID) + result, err := co.Execute(ctx, execID, deleteNS, targetRunID) if err != nil { return fmt.Errorf("cleanup failed: %w", err) } - // Link cleanup to discovered run IDs - if len(result.RunIDs) > 0 { - _ = auditor.LinkCleanupToRuns(ctx, execID, result.RunIDs) - } - - // Record cleanup counts - _ = auditor.RecordCleanupCounts(ctx, execID, - result.VMsDeleted, result.ServicesDeleted, result.SecretsDeleted, - result.DVsDeleted, result.PVCsDeleted, result.NamespaceDeleted) - - _ = auditor.RecordEvent(ctx, execID, audit.EventRecord{ - EventType: "cleanup_completed", - Message: fmt.Sprintf("Deleted %d VMs, %d services, %d secrets, %d DVs, %d PVCs", - result.VMsDeleted, result.ServicesDeleted, result.SecretsDeleted, - result.DVsDeleted, result.PVCsDeleted), - }) - - // Complete audit _ = auditor.CompleteExecution(ctx, execID, "success", "") - err = nil // clear for defer + err = nil logger.Info("cleanup complete", slog.Bool("dry_run", cfg.DryRun), @@ -765,33 +302,13 @@ func cleanupE(cmd *cobra.Command, args []string) error { return nil } -// printDryRun outputs VM specs in YAML without connecting to a cluster. -func printDryRun(logger *slog.Logger, plans []vmPlan) error { - logger.Info("dry run mode", slog.Int("total_vms", len(plans))) - - for _, p := range plans { - vmObj, err := vm.BuildVMSpec(*p.vmSpec) - if err != nil { - return fmt.Errorf("building VM spec for %q: %w", p.vmName, err) - } - data, err := sigyaml.Marshal(vmObj) - if err != nil { - return fmt.Errorf("marshaling VM spec for %q: %w", p.vmName, err) - } - // Output YAML to stdout for dry-run inspection - _, _ = fmt.Printf("# VM: %s (workload: %s)\n%s\n%s\n", p.vmName, p.component, string(data), "---") - } - return nil -} - -// printSummary outputs a deployment summary. -func printSummary(logger *slog.Logger, vmCount, svcCount, secCount int, cfg *config.Config, runID string) { +func printSummary(logger *slog.Logger, result *orchestrator.RunResult, cfg *config.Config) { logger.Info("deployment summary", - slog.String("run_id", runID), + slog.String("run_id", result.RunID), slog.String("namespace", cfg.Namespace), - slog.Int("vms_created", vmCount), - slog.Int("services_created", svcCount), - slog.Int("secrets_created", secCount), + slog.Int("vms_created", result.VMCount), + slog.Int("services_created", result.ServiceCount), + slog.Int("secrets_created", result.SecretCount), slog.String("container_image", cfg.ContainerDiskImage)) } diff --git a/internal/orchestrator/cleanup.go b/internal/orchestrator/cleanup.go new file mode 100644 index 0000000..4bcd070 --- /dev/null +++ b/internal/orchestrator/cleanup.go @@ -0,0 +1,94 @@ +// Copyright 2026 Red Hat +// SPDX-License-Identifier: Apache-2.0 + +package orchestrator + +import ( + "context" + "fmt" + "io" + "log/slog" + + "sigs.k8s.io/controller-runtime/pkg/client" + + "github.com/opdev/virtwork/internal/audit" + "github.com/opdev/virtwork/internal/cleanup" + "github.com/opdev/virtwork/internal/config" +) + +// CleanupOrchestrator coordinates the "cleanup" workflow: previewing and +// deleting managed resources. Dependencies are injected at construction +// for testability. +type CleanupOrchestrator struct { + logger *slog.Logger + client client.Client + config *config.Config + auditor audit.Auditor + writer io.Writer +} + +// NewCleanupOrchestrator creates a CleanupOrchestrator with the given dependencies. +func NewCleanupOrchestrator( + logger *slog.Logger, + c client.Client, + cfg *config.Config, + auditor audit.Auditor, + writer io.Writer, +) *CleanupOrchestrator { + return &CleanupOrchestrator{ + logger: logger, + client: c, + config: cfg, + auditor: auditor, + writer: writer, + } +} + +// Preview returns what would be deleted without modifying any resources. +func (co *CleanupOrchestrator) Preview( + ctx context.Context, + execID int64, + runID string, +) (*cleanup.CleanupPreview, error) { + preview, err := cleanup.PreviewCleanup(ctx, co.client, co.config, runID) + if err != nil { + return nil, fmt.Errorf("previewing cleanup: %w", err) + } + return preview, nil +} + +// Execute performs the cleanup, deleting managed resources and recording +// audit data. +func (co *CleanupOrchestrator) Execute( + ctx context.Context, + execID int64, + deleteNamespace bool, + runID string, +) (*cleanup.CleanupResult, error) { + _ = co.auditor.RecordEvent(ctx, execID, audit.EventRecord{ + EventType: "cleanup_started", + Message: fmt.Sprintf("Starting cleanup (namespace: %s, run-id filter: %q)", co.config.Namespace, runID), + }) + + result, err := cleanup.CleanupAll(ctx, co.client, co.config, deleteNamespace, runID) + if err != nil { + return nil, fmt.Errorf("cleanup failed: %w", err) + } + + if len(result.RunIDs) > 0 { + _ = co.auditor.LinkCleanupToRuns(ctx, execID, result.RunIDs) + } + + _ = co.auditor.RecordCleanupCounts(ctx, execID, + result.VMsDeleted, result.ServicesDeleted, result.SecretsDeleted, + result.DVsDeleted, result.PVCsDeleted, result.NamespaceDeleted) + + _ = co.auditor.RecordEvent(ctx, execID, audit.EventRecord{ + EventType: "cleanup_completed", + Message: fmt.Sprintf("Deleted %d VMs, %d services, %d secrets, %d DVs, %d PVCs", + result.VMsDeleted, result.ServicesDeleted, result.SecretsDeleted, + result.DVsDeleted, result.PVCsDeleted), + }) + + return result, nil +} diff --git a/internal/orchestrator/cleanup_test.go b/internal/orchestrator/cleanup_test.go new file mode 100644 index 0000000..b30c0f8 --- /dev/null +++ b/internal/orchestrator/cleanup_test.go @@ -0,0 +1,183 @@ +// Copyright 2026 Red Hat +// SPDX-License-Identifier: Apache-2.0 + +package orchestrator_test + +import ( + "bytes" + "context" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + kubevirtv1 "kubevirt.io/api/core/v1" + "sigs.k8s.io/controller-runtime/pkg/client" + + "github.com/opdev/virtwork/internal/audit" + "github.com/opdev/virtwork/internal/config" + "github.com/opdev/virtwork/internal/constants" + "github.com/opdev/virtwork/internal/logging" + "github.com/opdev/virtwork/internal/orchestrator" +) + +var _ = Describe("CleanupOrchestrator", func() { + var ( + ctx context.Context + buf *bytes.Buffer + cfg *config.Config + auditor audit.Auditor + ) + + BeforeEach(func() { + ctx = context.Background() + buf = &bytes.Buffer{} + cfg = defaultConfig() + auditor = audit.NoOpAuditor{} + }) + + Describe("Preview", func() { + It("should return zero counts when no resources exist", func() { + c := newFakeClient() + logger := logging.NewLogger(buf, false) + co := orchestrator.NewCleanupOrchestrator(logger, c, cfg, auditor, buf) + + preview, err := co.Preview(ctx, 0, "") + Expect(err).NotTo(HaveOccurred()) + Expect(preview.TotalCount).To(Equal(0)) + }) + + It("should count managed VMs", func() { + vm := &kubevirtv1.VirtualMachine{ + ObjectMeta: metav1.ObjectMeta{ + Name: "virtwork-cpu-0", + Namespace: constants.DefaultNamespace, + Labels: map[string]string{ + constants.LabelManagedBy: constants.ManagedByValue, + }, + }, + } + ns := &corev1.Namespace{ + ObjectMeta: metav1.ObjectMeta{ + Name: constants.DefaultNamespace, + }, + } + c := newFakeClient(ns, vm) + logger := logging.NewLogger(buf, false) + co := orchestrator.NewCleanupOrchestrator(logger, c, cfg, auditor, buf) + + preview, err := co.Preview(ctx, 0, "") + Expect(err).NotTo(HaveOccurred()) + Expect(preview.VMCount).To(Equal(1)) + Expect(preview.TotalCount).To(BeNumerically(">=", 1)) + }) + + It("should filter by run-id when provided", func() { + vm1 := &kubevirtv1.VirtualMachine{ + ObjectMeta: metav1.ObjectMeta{ + Name: "virtwork-cpu-0", + Namespace: constants.DefaultNamespace, + Labels: map[string]string{ + constants.LabelManagedBy: constants.ManagedByValue, + constants.LabelRunID: "run-1", + }, + }, + } + vm2 := &kubevirtv1.VirtualMachine{ + ObjectMeta: metav1.ObjectMeta{ + Name: "virtwork-cpu-1", + Namespace: constants.DefaultNamespace, + Labels: map[string]string{ + constants.LabelManagedBy: constants.ManagedByValue, + constants.LabelRunID: "run-2", + }, + }, + } + ns := &corev1.Namespace{ + ObjectMeta: metav1.ObjectMeta{ + Name: constants.DefaultNamespace, + }, + } + c := newFakeClient(ns, vm1, vm2) + logger := logging.NewLogger(buf, false) + co := orchestrator.NewCleanupOrchestrator(logger, c, cfg, auditor, buf) + + preview, err := co.Preview(ctx, 0, "run-1") + Expect(err).NotTo(HaveOccurred()) + Expect(preview.VMCount).To(Equal(1)) + }) + }) + + Describe("Execute", func() { + It("should delete managed VMs", func() { + vm := &kubevirtv1.VirtualMachine{ + ObjectMeta: metav1.ObjectMeta{ + Name: "virtwork-cpu-0", + Namespace: constants.DefaultNamespace, + Labels: map[string]string{ + constants.LabelManagedBy: constants.ManagedByValue, + }, + }, + } + ns := &corev1.Namespace{ + ObjectMeta: metav1.ObjectMeta{ + Name: constants.DefaultNamespace, + }, + } + c := newFakeClient(ns, vm) + logger := logging.NewLogger(buf, false) + co := orchestrator.NewCleanupOrchestrator(logger, c, cfg, auditor, buf) + + result, err := co.Execute(ctx, 0, false, "") + Expect(err).NotTo(HaveOccurred()) + Expect(result.VMsDeleted).To(Equal(1)) + + // Verify VM is gone + vmObj := &kubevirtv1.VirtualMachine{} + err = c.Get(ctx, client.ObjectKey{ + Name: "virtwork-cpu-0", + Namespace: constants.DefaultNamespace, + }, vmObj) + Expect(err).To(HaveOccurred()) + }) + + It("should delete managed secrets", func() { + secret := &corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{ + Name: "virtwork-cpu-0-cloudinit", + Namespace: constants.DefaultNamespace, + Labels: map[string]string{ + constants.LabelManagedBy: constants.ManagedByValue, + }, + }, + } + ns := &corev1.Namespace{ + ObjectMeta: metav1.ObjectMeta{ + Name: constants.DefaultNamespace, + }, + } + c := newFakeClient(ns, secret) + logger := logging.NewLogger(buf, false) + co := orchestrator.NewCleanupOrchestrator(logger, c, cfg, auditor, buf) + + result, err := co.Execute(ctx, 0, false, "") + Expect(err).NotTo(HaveOccurred()) + Expect(result.SecretsDeleted).To(Equal(1)) + }) + + It("should record cleanup counts in audit", func() { + ns := &corev1.Namespace{ + ObjectMeta: metav1.ObjectMeta{ + Name: constants.DefaultNamespace, + }, + } + c := newFakeClient(ns) + logger := logging.NewLogger(buf, false) + co := orchestrator.NewCleanupOrchestrator(logger, c, cfg, auditor, buf) + + result, err := co.Execute(ctx, 0, false, "") + Expect(err).NotTo(HaveOccurred()) + Expect(result).NotTo(BeNil()) + }) + }) +}) diff --git a/internal/orchestrator/orchestrator.go b/internal/orchestrator/orchestrator.go new file mode 100644 index 0000000..1cf2e79 --- /dev/null +++ b/internal/orchestrator/orchestrator.go @@ -0,0 +1,549 @@ +// Copyright 2026 Red Hat +// SPDX-License-Identifier: Apache-2.0 + +package orchestrator + +import ( + "context" + "errors" + "fmt" + "io" + "log/slog" + "time" + + "golang.org/x/sync/errgroup" + "sigs.k8s.io/controller-runtime/pkg/client" + sigyaml "sigs.k8s.io/yaml" + + "github.com/opdev/virtwork/internal/audit" + "github.com/opdev/virtwork/internal/config" + "github.com/opdev/virtwork/internal/constants" + "github.com/opdev/virtwork/internal/resources" + "github.com/opdev/virtwork/internal/vm" + "github.com/opdev/virtwork/internal/wait" + "github.com/opdev/virtwork/internal/workloads" +) + +var ErrReadinessCheck = errors.New("readiness check failed") + +// RunOrchestrator coordinates the "run" workflow: planning VMs, creating +// resources, and waiting for readiness. Dependencies are injected at +// construction for testability. +type RunOrchestrator struct { + logger *slog.Logger + client client.Client + config *config.Config + auditor audit.Auditor + writer io.Writer +} + +// NewRunOrchestrator creates a RunOrchestrator with the given dependencies. +func NewRunOrchestrator( + logger *slog.Logger, + c client.Client, + cfg *config.Config, + auditor audit.Auditor, + writer io.Writer, +) *RunOrchestrator { + return &RunOrchestrator{ + logger: logger, + client: c, + config: cfg, + auditor: auditor, + writer: writer, + } +} + +// Run executes the full orchestration flow: plan VMs, create resources, +// wait for readiness. The caller owns the audit execution lifecycle +// (StartExecution/CompleteExecution); Run records workloads, VMs, resources, +// and events internally. +func (ro *RunOrchestrator) Run( + ctx context.Context, + execID int64, + runID string, + workloadNames []string, + vmCountFlag int, +) (*RunResult, error) { + cfg := ro.config + + registry := workloads.DefaultRegistry() + registryOpts := []workloads.Option{ + workloads.WithNamespace(cfg.Namespace), + workloads.WithSSHCredentials(cfg.SSHUser, cfg.SSHPassword, cfg.SSHAuthorizedKeys), + workloads.WithDataDiskSize(cfg.DataDiskSize), + } + + plans, vmNames, workloadInstances, auditWorkloadIDs, err := ro.planVMs( + ctx, execID, runID, workloadNames, vmCountFlag, registry, registryOpts, + ) + if err != nil { + return nil, err + } + + _ = ro.auditor.RecordEvent(ctx, execID, audit.EventRecord{ + EventType: "execution_started", + Message: fmt.Sprintf("Planned %d VMs across %d workloads", len(plans), len(workloadNames)), + }) + + if cfg.DryRun { + if err := ro.printDryRun(plans); err != nil { + return nil, err + } + return &RunResult{ + RunID: runID, + VMCount: len(plans), + }, nil + } + + svcCount, err := ro.createResources(ctx, execID, runID, plans, workloadInstances) + if err != nil { + return nil, err + } + + secretCount, err := ro.createSecrets(ctx, execID, runID, plans) + if err != nil { + return nil, err + } + + if err := ro.createVMs(ctx, execID, cfg, plans, auditWorkloadIDs); err != nil { + for _, wlID := range auditWorkloadIDs { + _ = ro.auditor.UpdateWorkloadStatus(ctx, wlID, "failed") + } + return nil, fmt.Errorf("creating VMs: %w", err) + } + + if err := ro.waitForReadiness(ctx, execID, vmNames, auditWorkloadIDs, plans); err != nil { + return nil, err + } + + for _, wlID := range auditWorkloadIDs { + _ = ro.auditor.UpdateWorkloadStatus(ctx, wlID, "created") + } + + return &RunResult{ + RunID: runID, + VMCount: len(plans), + ServiceCount: svcCount, + SecretCount: secretCount, + }, nil +} + +func (ro *RunOrchestrator) planVMs( + ctx context.Context, + execID int64, + runID string, + workloadNames []string, + vmCountFlag int, + registry workloads.Registry, + registryOpts []workloads.Option, +) ([]VMPlan, []string, map[string]workloads.Workload, map[string]int64, error) { + cfg := ro.config + + var plans []VMPlan + var vmNames []string + auditWorkloadIDs := make(map[string]int64) + workloadInstances := make(map[string]workloads.Workload) + + for _, name := range workloadNames { + if fileCfg, ok := cfg.Workloads[name]; ok { + if fileCfg.Enabled != nil && !*fileCfg.Enabled { + _ = ro.auditor.RecordEvent(ctx, execID, audit.EventRecord{ + EventType: "workload_skipped", + Message: fmt.Sprintf("Workload %q disabled via config (enabled: false)", name), + }) + ro.logger.Info("workload skipped", + slog.String("workload", name), + slog.String("reason", "disabled in config")) + continue + } + } + + wlCfg := config.WorkloadConfig{ + Enabled: new(true), + VMCount: vmCountFlag, + CPUCores: cfg.CPUCores, + Memory: cfg.Memory, + } + if fileCfg, ok := cfg.Workloads[name]; ok { + if fileCfg.CPUCores > 0 { + wlCfg.CPUCores = fileCfg.CPUCores + } + if fileCfg.Memory != "" { + wlCfg.Memory = fileCfg.Memory + } + if fileCfg.VMCount > 0 { + wlCfg.VMCount = fileCfg.VMCount + } + if len(fileCfg.Params) > 0 { + wlCfg.Params = fileCfg.Params + } + } + + w, err := registry.Get(name, wlCfg, registryOpts...) + if err != nil { + return nil, nil, nil, nil, fmt.Errorf("creating workload %q: %w", name, err) + } + workloadInstances[name] = w + + vmCount := w.VMCount() + res := w.VMResources() + + dvTemplatesForAudit, err := w.DataVolumeTemplates() + if err != nil { + return nil, nil, nil, nil, fmt.Errorf("building data volume templates for %q: %w", name, err) + } + wlID, _ := ro.auditor.RecordWorkload(ctx, execID, audit.WorkloadRecord{ + WorkloadType: name, + Enabled: true, + VMCount: vmCount, + CPUCores: res.CPUCores, + Memory: res.Memory, + HasDataDisk: len(dvTemplatesForAudit) > 0, + DataDiskSize: cfg.DataDiskSize, + RequiresService: w.RequiresService(), + }) + auditWorkloadIDs[name] = wlID + + if multiVM, isMulti := w.(workloads.MultiVMWorkload); !isMulti { + userdata, err := w.CloudInitUserdata() + if err != nil { + return nil, nil, nil, nil, fmt.Errorf("generating cloud-init for %q: %w", name, err) + } + + for i := range vmCount { + vmName := fmt.Sprintf("virtwork-%s-%d", name, i) + + dvts, err := w.DataVolumeTemplates() + if err != nil { + return nil, nil, nil, nil, fmt.Errorf( + "building data volume templates for %q vm %s: %w", + name, + vmName, + err, + ) + } + dvTemplates, extraVols := NamespaceDataVolumes(dvts, w.ExtraVolumes(), vmName) + + plans = append(plans, VMPlan{ + WorkloadName: name, + Component: name, + VMName: vmName, + VMSpec: &VMSpecInput{ + Name: vmName, + Namespace: cfg.Namespace, + ContainerDiskImage: cfg.ContainerDiskImage, + CloudInitUserdata: userdata, + CPUCores: res.CPUCores, + Memory: res.Memory, + Labels: map[string]string{ + constants.LabelAppName: fmt.Sprintf("virtwork-%s", name), + constants.LabelManagedBy: constants.ManagedByValue, + constants.LabelComponent: name, + constants.LabelRunID: runID, + }, + ExtraDisks: w.ExtraDisks(), + ExtraVolumes: extraVols, + DataVolumeTemplates: dvTemplates, + }, + }) + vmNames = append(vmNames, vmName) + } + } else { + roles := multiVM.Roles() + perRole := vmCount / len(roles) + if remainder := vmCount % len(roles); remainder != 0 { + ro.logger.Warn("VM count not evenly divisible by roles; remainder VMs will not be created", + slog.String("workload", name), + slog.Int("requested", vmCount), + slog.Int("per_role", perRole), + slog.Int("dropped", remainder)) + } + for _, role := range roles { + userdata, err := multiVM.UserdataForRole(role, cfg.Namespace) + if err != nil { + return nil, nil, nil, nil, fmt.Errorf("generating cloud-init for %q role %q: %w", name, role, err) + } + + for i := range perRole { + vmName := fmt.Sprintf("virtwork-%s-%s-%d", name, role, i) + labels := map[string]string{ + constants.LabelAppName: fmt.Sprintf("virtwork-%s", name), + constants.LabelManagedBy: constants.ManagedByValue, + constants.LabelComponent: name, + constants.LabelRunID: runID, + "virtwork/role": role, + } + + dvts, err := w.DataVolumeTemplates() + if err != nil { + return nil, nil, nil, nil, fmt.Errorf( + "building data volume templates for %q role %q vm %s: %w", name, role, vmName, err, + ) + } + dvTemplates, extraVols := NamespaceDataVolumes(dvts, w.ExtraVolumes(), vmName) + + plans = append(plans, VMPlan{ + WorkloadName: name, + Component: name, + VMName: vmName, + Role: role, + VMSpec: &VMSpecInput{ + Name: vmName, + Namespace: cfg.Namespace, + ContainerDiskImage: cfg.ContainerDiskImage, + CloudInitUserdata: userdata, + CPUCores: res.CPUCores, + Memory: res.Memory, + Labels: labels, + ExtraDisks: w.ExtraDisks(), + ExtraVolumes: extraVols, + DataVolumeTemplates: dvTemplates, + }, + }) + vmNames = append(vmNames, vmName) + } + } + } + } + + return plans, vmNames, workloadInstances, auditWorkloadIDs, nil +} + +func (ro *RunOrchestrator) createResources( + ctx context.Context, + execID int64, + runID string, + plans []VMPlan, + workloadInstances map[string]workloads.Workload, +) (int, error) { + cfg := ro.config + + if err := resources.EnsureNamespace(ctx, ro.client, cfg.Namespace, map[string]string{ + constants.LabelManagedBy: constants.ManagedByValue, + }); err != nil { + return 0, fmt.Errorf("ensuring namespace %q: %w", cfg.Namespace, err) + } + ro.logger.Info("namespace ensured", slog.String("namespace", cfg.Namespace)) + + servicesCreated := 0 + for name, w := range workloadInstances { + if w.RequiresService() { + svc := w.ServiceSpec() + if svc != nil { + if svc.Labels == nil { + svc.Labels = make(map[string]string) + } + svc.Labels[constants.LabelRunID] = runID + + if err := resources.CreateService(ctx, ro.client, svc); err != nil { + return 0, fmt.Errorf("creating service for %q: %w", name, err) + } + servicesCreated++ + ro.logger.Info("service created", + slog.String("service_name", svc.Name), + slog.String("namespace", svc.Namespace)) + + _, _ = ro.auditor.RecordResource(ctx, execID, audit.ResourceRecord{ + ResourceType: "Service", + ResourceName: svc.Name, + Namespace: svc.Namespace, + }) + _ = ro.auditor.RecordEvent(ctx, execID, audit.EventRecord{ + EventType: "service_created", + Message: fmt.Sprintf("Service %s created", svc.Name), + }) + } + } + } + + return servicesCreated, nil +} + +func (ro *RunOrchestrator) createSecrets( + ctx context.Context, + execID int64, + runID string, + plans []VMPlan, +) (int, error) { + cfg := ro.config + + secretsCreated := 0 + for i := range plans { + secretName := plans[i].VMName + "-cloudinit" + secretLabels := map[string]string{ + constants.LabelAppName: plans[i].VMSpec.Labels[constants.LabelAppName], + constants.LabelManagedBy: constants.ManagedByValue, + constants.LabelComponent: plans[i].Component, + constants.LabelRunID: runID, + } + if err := resources.CreateCloudInitSecret(ctx, ro.client, secretName, + cfg.Namespace, plans[i].VMSpec.CloudInitUserdata, secretLabels); err != nil { + return 0, fmt.Errorf("creating cloud-init secret for %q: %w", plans[i].VMName, err) + } + plans[i].VMSpec.CloudInitSecretName = secretName + secretsCreated++ + ro.logger.Info("secret created", + slog.String("secret_name", secretName), + slog.String("namespace", cfg.Namespace)) + + _, _ = ro.auditor.RecordResource(ctx, execID, audit.ResourceRecord{ + ResourceType: "Secret", + ResourceName: secretName, + Namespace: cfg.Namespace, + }) + } + + return secretsCreated, nil +} + +func (ro *RunOrchestrator) createVMs( + ctx context.Context, + execID int64, + cfg *config.Config, + plans []VMPlan, + auditWorkloadIDs map[string]int64, +) error { + g, gctx := errgroup.WithContext(ctx) + for _, plan := range plans { + p := plan + g.Go(func() error { + vmObj, err := vm.BuildVMSpec(vm.VMSpecOpts{ + Name: p.VMSpec.Name, + Namespace: p.VMSpec.Namespace, + ContainerDiskImage: p.VMSpec.ContainerDiskImage, + CloudInitUserdata: p.VMSpec.CloudInitUserdata, + CloudInitSecretName: p.VMSpec.CloudInitSecretName, + CPUCores: p.VMSpec.CPUCores, + Memory: p.VMSpec.Memory, + Labels: p.VMSpec.Labels, + ExtraDisks: p.VMSpec.ExtraDisks, + ExtraVolumes: p.VMSpec.ExtraVolumes, + DataVolumeTemplates: p.VMSpec.DataVolumeTemplates, + }) + if err != nil { + return fmt.Errorf("building VM spec for %q: %w", p.VMName, err) + } + if err := vm.CreateVM(gctx, ro.client, vmObj); err != nil { + _ = ro.auditor.RecordEvent(ctx, execID, audit.EventRecord{ + EventType: "vm_failed", + Message: fmt.Sprintf("Failed to create VM %s", p.VMName), + ErrorDetail: err.Error(), + }) + return fmt.Errorf("creating VM %q: %w", p.VMName, err) + } + ro.logger.Info("vm created", + slog.String("vm_name", p.VMName), + slog.String("namespace", cfg.Namespace), + slog.String("workload", p.Component)) + + wlID := auditWorkloadIDs[p.Component] + _, _ = ro.auditor.RecordVM(ctx, execID, wlID, audit.VMRecord{ + VMName: p.VMName, + Namespace: cfg.Namespace, + Component: p.Component, + Role: p.Role, + CPUCores: p.VMSpec.CPUCores, + Memory: p.VMSpec.Memory, + ContainerDiskImage: p.VMSpec.ContainerDiskImage, + HasDataDisk: len(p.VMSpec.DataVolumeTemplates) > 0, + DataDiskSize: cfg.DataDiskSize, + }) + _ = ro.auditor.RecordEvent(ctx, execID, audit.EventRecord{ + EventType: "vm_created", + Message: fmt.Sprintf("VM %s created", p.VMName), + }) + return nil + }) + } + return g.Wait() +} + +func (ro *RunOrchestrator) waitForReadiness( + ctx context.Context, + execID int64, + vmNames []string, + auditWorkloadIDs map[string]int64, + plans []VMPlan, +) error { + if !ro.config.WaitForReady { + return nil + } + + vmToComponent := make(map[string]string, len(plans)) + for _, p := range plans { + vmToComponent[p.VMName] = p.Component + } + + timeout := time.Duration(ro.config.ReadyTimeoutSeconds) * time.Second + ro.logger.Info("waiting for VMs to become ready", + slog.Int("vm_count", len(vmNames)), + slog.Duration("timeout", timeout)) + results := wait.WaitForAllVMsReady(ctx, ro.client, ro.logger, vmNames, ro.config.Namespace, + timeout, constants.DefaultPollInterval) + + failures := 0 + failedWorkloads := make(map[string]bool) + for name, err := range results { + if err != nil { + ro.logger.Error("vm readiness check failed", + slog.String("vm_name", name), + slog.String("error", err.Error())) + failures++ + _ = ro.auditor.RecordEvent(ctx, execID, audit.EventRecord{ + EventType: "vm_timeout", + Message: fmt.Sprintf("VM %s failed readiness check", name), + ErrorDetail: err.Error(), + }) + if comp, ok := vmToComponent[name]; ok { + failedWorkloads[comp] = true + } + } else { + _ = ro.auditor.RecordEvent(ctx, execID, audit.EventRecord{ + EventType: "vm_ready", + Message: fmt.Sprintf("VM %s is ready", name), + }) + } + } + if failures > 0 { + for comp, wlID := range auditWorkloadIDs { + if failedWorkloads[comp] { + _ = ro.auditor.UpdateWorkloadStatus(ctx, wlID, "failed") + } else { + _ = ro.auditor.UpdateWorkloadStatus(ctx, wlID, "created") + } + } + return fmt.Errorf("%d of %d VMs failed; %w", failures, len(vmNames), ErrReadinessCheck) + } + ro.logger.Info("all VMs ready", slog.Int("vm_count", len(vmNames))) + return nil +} + +func (ro *RunOrchestrator) printDryRun(plans []VMPlan) error { + ro.logger.Info("dry run mode", slog.Int("total_vms", len(plans))) + + for _, p := range plans { + vmObj, err := vm.BuildVMSpec(vm.VMSpecOpts{ + Name: p.VMSpec.Name, + Namespace: p.VMSpec.Namespace, + ContainerDiskImage: p.VMSpec.ContainerDiskImage, + CloudInitUserdata: p.VMSpec.CloudInitUserdata, + CloudInitSecretName: p.VMSpec.CloudInitSecretName, + CPUCores: p.VMSpec.CPUCores, + Memory: p.VMSpec.Memory, + Labels: p.VMSpec.Labels, + ExtraDisks: p.VMSpec.ExtraDisks, + ExtraVolumes: p.VMSpec.ExtraVolumes, + DataVolumeTemplates: p.VMSpec.DataVolumeTemplates, + }) + if err != nil { + return fmt.Errorf("building VM spec for %q: %w", p.VMName, err) + } + data, err := sigyaml.Marshal(vmObj) + if err != nil { + return fmt.Errorf("marshaling VM spec for %q: %w", p.VMName, err) + } + _, _ = fmt.Fprintf(ro.writer, "# VM: %s (workload: %s)\n%s\n%s\n", p.VMName, p.Component, string(data), "---") + } + return nil +} diff --git a/internal/orchestrator/orchestrator_suite_test.go b/internal/orchestrator/orchestrator_suite_test.go new file mode 100644 index 0000000..362e9e8 --- /dev/null +++ b/internal/orchestrator/orchestrator_suite_test.go @@ -0,0 +1,16 @@ +// Copyright 2026 Red Hat +// SPDX-License-Identifier: Apache-2.0 + +package orchestrator_test + +import ( + "testing" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +func TestOrchestrator(t *testing.T) { + RegisterFailHandler(Fail) + RunSpecs(t, "Orchestrator Suite") +} diff --git a/internal/orchestrator/orchestrator_test.go b/internal/orchestrator/orchestrator_test.go new file mode 100644 index 0000000..adc2faa --- /dev/null +++ b/internal/orchestrator/orchestrator_test.go @@ -0,0 +1,277 @@ +// Copyright 2026 Red Hat +// SPDX-License-Identifier: Apache-2.0 + +package orchestrator_test + +import ( + "bytes" + "context" + "fmt" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/runtime" + kubevirtv1 "kubevirt.io/api/core/v1" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + "github.com/opdev/virtwork/internal/audit" + "github.com/opdev/virtwork/internal/cluster" + "github.com/opdev/virtwork/internal/config" + "github.com/opdev/virtwork/internal/constants" + "github.com/opdev/virtwork/internal/logging" + "github.com/opdev/virtwork/internal/orchestrator" +) + +func newFakeClient(objs ...client.Object) client.Client { + scheme := cluster.NewScheme() + builder := fake.NewClientBuilder().WithScheme(scheme) + if len(objs) > 0 { + runtimeObjs := make([]runtime.Object, len(objs)) + for i, o := range objs { + runtimeObjs[i] = o + } + builder = builder.WithRuntimeObjects(runtimeObjs...) + } + return builder.Build() +} + +func defaultConfig() *config.Config { + return &config.Config{ + Namespace: constants.DefaultNamespace, + ContainerDiskImage: constants.DefaultContainerDiskImage, + DataDiskSize: constants.DefaultDiskSize, + CPUCores: constants.DefaultCPUCores, + Memory: constants.DefaultMemory, + SSHUser: constants.DefaultSSHUser, + WaitForReady: false, + ReadyTimeoutSeconds: 600, + AuditEnabled: false, + } +} + +var _ = Describe("RunOrchestrator", func() { + var ( + ctx context.Context + buf *bytes.Buffer + cfg *config.Config + auditor audit.Auditor + ) + + BeforeEach(func() { + ctx = context.Background() + buf = &bytes.Buffer{} + cfg = defaultConfig() + auditor = audit.NoOpAuditor{} + }) + + Describe("Run", func() { + Context("dry-run mode", func() { + BeforeEach(func() { + cfg.DryRun = true + }) + + It("should build plans and print YAML without requiring a client", func() { + logger := logging.NewLogger(buf, false) + ro := orchestrator.NewRunOrchestrator(logger, nil, cfg, auditor, buf) + + result, err := ro.Run(ctx, 0, "test-run-id", []string{"cpu"}, 1) + Expect(err).NotTo(HaveOccurred()) + Expect(result).NotTo(BeNil()) + Expect(result.VMCount).To(Equal(1)) + Expect(buf.String()).To(ContainSubstring("virtwork-cpu-0")) + }) + + It("should handle multiple workloads in dry-run", func() { + logger := logging.NewLogger(buf, false) + ro := orchestrator.NewRunOrchestrator(logger, nil, cfg, auditor, buf) + + result, err := ro.Run(ctx, 0, "test-run-id", []string{"cpu", "memory"}, 1) + Expect(err).NotTo(HaveOccurred()) + Expect(result.VMCount).To(Equal(2)) + }) + + It("should respect vm-count flag in dry-run", func() { + logger := logging.NewLogger(buf, false) + ro := orchestrator.NewRunOrchestrator(logger, nil, cfg, auditor, buf) + + result, err := ro.Run(ctx, 0, "test-run-id", []string{"cpu"}, 3) + Expect(err).NotTo(HaveOccurred()) + Expect(result.VMCount).To(Equal(3)) + }) + + It("should handle multi-VM workloads in dry-run", func() { + logger := logging.NewLogger(buf, false) + ro := orchestrator.NewRunOrchestrator(logger, nil, cfg, auditor, buf) + + result, err := ro.Run(ctx, 0, "test-run-id", []string{"network"}, 2) + Expect(err).NotTo(HaveOccurred()) + // network workload with VMCount=2 creates 4 VMs (2 server + 2 client) + Expect(result.VMCount).To(Equal(4)) + }) + }) + + Context("workload filtering", func() { + It("should skip disabled workloads", func() { + cfg.DryRun = true + enabled := false + cfg.Workloads = map[string]config.WorkloadConfig{ + "cpu": {Enabled: &enabled}, + } + + logger := logging.NewLogger(buf, false) + ro := orchestrator.NewRunOrchestrator(logger, nil, cfg, auditor, buf) + + result, err := ro.Run(ctx, 0, "test-run-id", []string{"cpu"}, 1) + Expect(err).NotTo(HaveOccurred()) + Expect(result.VMCount).To(Equal(0)) + }) + + It("should include explicitly enabled workloads", func() { + cfg.DryRun = true + enabled := true + cfg.Workloads = map[string]config.WorkloadConfig{ + "cpu": {Enabled: &enabled, VMCount: 2}, + } + + logger := logging.NewLogger(buf, false) + ro := orchestrator.NewRunOrchestrator(logger, nil, cfg, auditor, buf) + + result, err := ro.Run(ctx, 0, "test-run-id", []string{"cpu"}, 1) + Expect(err).NotTo(HaveOccurred()) + Expect(result.VMCount).To(Equal(2)) + }) + + It("should return error for unknown workload", func() { + cfg.DryRun = true + logger := logging.NewLogger(buf, false) + ro := orchestrator.NewRunOrchestrator(logger, nil, cfg, auditor, buf) + + _, err := ro.Run(ctx, 0, "test-run-id", []string{"nonexistent"}, 1) + Expect(err).To(HaveOccurred()) + Expect(err.Error()).To(ContainSubstring("nonexistent")) + }) + }) + + Context("normal mode with fake client", func() { + var c client.Client + + BeforeEach(func() { + c = newFakeClient() + }) + + It("should create namespace, services, secrets, and VMs", func() { + logger := logging.NewLogger(buf, false) + ro := orchestrator.NewRunOrchestrator(logger, c, cfg, auditor, buf) + + result, err := ro.Run(ctx, 0, "test-run-id", []string{"cpu"}, 1) + Expect(err).NotTo(HaveOccurred()) + Expect(result.VMCount).To(Equal(1)) + Expect(result.SecretCount).To(Equal(1)) + + // Verify namespace was created + ns := &corev1.Namespace{} + err = c.Get(ctx, client.ObjectKey{Name: constants.DefaultNamespace}, ns) + Expect(err).NotTo(HaveOccurred()) + Expect(ns.Labels[constants.LabelManagedBy]).To(Equal(constants.ManagedByValue)) + + // Verify VM was created + vmObj := &kubevirtv1.VirtualMachine{} + err = c.Get(ctx, client.ObjectKey{ + Name: "virtwork-cpu-0", + Namespace: constants.DefaultNamespace, + }, vmObj) + Expect(err).NotTo(HaveOccurred()) + Expect(vmObj.Labels[constants.LabelRunID]).To(Equal("test-run-id")) + + // Verify secret was created + secret := &corev1.Secret{} + err = c.Get(ctx, client.ObjectKey{ + Name: "virtwork-cpu-0-cloudinit", + Namespace: constants.DefaultNamespace, + }, secret) + Expect(err).NotTo(HaveOccurred()) + }) + + It("should create service for workloads that require it", func() { + logger := logging.NewLogger(buf, false) + ro := orchestrator.NewRunOrchestrator(logger, c, cfg, auditor, buf) + + result, err := ro.Run(ctx, 0, "test-run-id", []string{"network"}, 2) + Expect(err).NotTo(HaveOccurred()) + Expect(result.ServiceCount).To(BeNumerically(">", 0)) + }) + + It("should set run-id labels on all resources", func() { + logger := logging.NewLogger(buf, false) + ro := orchestrator.NewRunOrchestrator(logger, c, cfg, auditor, buf) + + _, err := ro.Run(ctx, 0, "my-run-id", []string{"cpu"}, 1) + Expect(err).NotTo(HaveOccurred()) + + vmObj := &kubevirtv1.VirtualMachine{} + err = c.Get(ctx, client.ObjectKey{ + Name: "virtwork-cpu-0", + Namespace: constants.DefaultNamespace, + }, vmObj) + Expect(err).NotTo(HaveOccurred()) + Expect(vmObj.Labels[constants.LabelRunID]).To(Equal("my-run-id")) + }) + + It("should handle per-workload config overrides from YAML", func() { + cfg.Workloads = map[string]config.WorkloadConfig{ + "cpu": {CPUCores: 8, Memory: "16Gi"}, + } + logger := logging.NewLogger(buf, false) + ro := orchestrator.NewRunOrchestrator(logger, c, cfg, auditor, buf) + + result, err := ro.Run(ctx, 0, "test-run-id", []string{"cpu"}, 1) + Expect(err).NotTo(HaveOccurred()) + Expect(result.VMCount).To(Equal(1)) + }) + + It("should create multiple VMs concurrently", func() { + logger := logging.NewLogger(buf, false) + ro := orchestrator.NewRunOrchestrator(logger, c, cfg, auditor, buf) + + result, err := ro.Run(ctx, 0, "test-run-id", []string{"cpu"}, 3) + Expect(err).NotTo(HaveOccurred()) + Expect(result.VMCount).To(Equal(3)) + + for i := range 3 { + vmObj := &kubevirtv1.VirtualMachine{} + err := c.Get(ctx, client.ObjectKey{ + Name: fmt.Sprintf("virtwork-cpu-%d", i), + Namespace: constants.DefaultNamespace, + }, vmObj) + Expect(err).NotTo(HaveOccurred()) + } + }) + }) + + Context("workloads with DataVolumes", func() { + It("should namespace DataVolume names across VMs", func() { + cfg.DryRun = true + logger := logging.NewLogger(buf, false) + ro := orchestrator.NewRunOrchestrator(logger, nil, cfg, auditor, buf) + + result, err := ro.Run(ctx, 0, "test-run-id", []string{"disk"}, 2) + Expect(err).NotTo(HaveOccurred()) + Expect(result.VMCount).To(Equal(2)) + }) + }) + + Context("RunResult", func() { + It("should set RunID from provided value", func() { + cfg.DryRun = true + logger := logging.NewLogger(buf, false) + ro := orchestrator.NewRunOrchestrator(logger, nil, cfg, auditor, buf) + + result, err := ro.Run(ctx, 0, "expected-run-id", []string{"cpu"}, 1) + Expect(err).NotTo(HaveOccurred()) + Expect(result.RunID).To(Equal("expected-run-id")) + }) + }) + }) +}) diff --git a/internal/orchestrator/types.go b/internal/orchestrator/types.go new file mode 100644 index 0000000..ddf26c9 --- /dev/null +++ b/internal/orchestrator/types.go @@ -0,0 +1,80 @@ +// Copyright 2026 Red Hat +// SPDX-License-Identifier: Apache-2.0 + +package orchestrator + +import ( + "fmt" + + kubevirtv1 "kubevirt.io/api/core/v1" +) + +// VMSpecInput holds the inputs needed to build a KubeVirt VirtualMachine spec. +type VMSpecInput struct { + Name string + Namespace string + ContainerDiskImage string + CloudInitUserdata string + CloudInitSecretName string + CPUCores int + Memory string + Labels map[string]string + ExtraDisks []kubevirtv1.Disk + ExtraVolumes []kubevirtv1.Volume + DataVolumeTemplates []kubevirtv1.DataVolumeTemplateSpec +} + +// VMPlan describes a single VM to be created during orchestration. +type VMPlan struct { + WorkloadName string + VMName string + Component string + Role string + VMSpec *VMSpecInput +} + +// RunResult holds the summary of a completed run. +type RunResult struct { + RunID string + VMCount int + ServiceCount int + SecretCount int +} + +// NamespaceDataVolumes appends the VM name to DataVolume template names and +// updates corresponding volume references to prevent name collisions when +// deploying multiple VMs of the same workload type. +func NamespaceDataVolumes( + baseTemplates []kubevirtv1.DataVolumeTemplateSpec, + baseVolumes []kubevirtv1.Volume, + vmName string, +) ([]kubevirtv1.DataVolumeTemplateSpec, []kubevirtv1.Volume) { + if len(baseTemplates) == 0 { + return baseTemplates, baseVolumes + } + + nameMap := make(map[string]string, len(baseTemplates)) + templates := make([]kubevirtv1.DataVolumeTemplateSpec, len(baseTemplates)) + for i, tmpl := range baseTemplates { + oldName := tmpl.Name + newName := fmt.Sprintf("%s-%s", oldName, vmName) + nameMap[oldName] = newName + + templates[i] = tmpl + templates[i].Name = newName + } + + volumes := make([]kubevirtv1.Volume, len(baseVolumes)) + for i, vol := range baseVolumes { + volumes[i] = vol + if vol.DataVolume != nil { + if newName, ok := nameMap[vol.DataVolume.Name]; ok { + volumes[i].DataVolume = &kubevirtv1.DataVolumeSource{ + Name: newName, + } + } + } + } + + return templates, volumes +} diff --git a/internal/orchestrator/types_test.go b/internal/orchestrator/types_test.go new file mode 100644 index 0000000..1f851bc --- /dev/null +++ b/internal/orchestrator/types_test.go @@ -0,0 +1,186 @@ +// Copyright 2026 Red Hat +// SPDX-License-Identifier: Apache-2.0 + +package orchestrator_test + +import ( + "fmt" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + kubevirtv1 "kubevirt.io/api/core/v1" + + "github.com/opdev/virtwork/internal/orchestrator" +) + +var _ = Describe("NamespaceDataVolumes", func() { + It("should return inputs unchanged when there are no templates", func() { + var templates []kubevirtv1.DataVolumeTemplateSpec + volumes := []kubevirtv1.Volume{ + {Name: "containerdisk"}, + } + + outTemplates, outVolumes := orchestrator.NamespaceDataVolumes(templates, volumes, "vm-0") + Expect(outTemplates).To(BeEmpty()) + Expect(outVolumes).To(Equal(volumes)) + }) + + It("should append VM name to DataVolume template names", func() { + templates := []kubevirtv1.DataVolumeTemplateSpec{ + {ObjectMeta: metav1.ObjectMeta{Name: "data"}}, + } + volumes := []kubevirtv1.Volume{ + { + Name: "datadisk", + VolumeSource: kubevirtv1.VolumeSource{ + DataVolume: &kubevirtv1.DataVolumeSource{Name: "data"}, + }, + }, + } + + outTemplates, outVolumes := orchestrator.NamespaceDataVolumes(templates, volumes, "virtwork-disk-0") + Expect(outTemplates).To(HaveLen(1)) + Expect(outTemplates[0].Name).To(Equal("data-virtwork-disk-0")) + Expect(outVolumes[0].DataVolume.Name).To(Equal("data-virtwork-disk-0")) + }) + + It("should handle multiple DataVolume templates", func() { + templates := []kubevirtv1.DataVolumeTemplateSpec{ + {ObjectMeta: metav1.ObjectMeta{Name: "wal"}}, + {ObjectMeta: metav1.ObjectMeta{Name: "pgdata"}}, + } + volumes := []kubevirtv1.Volume{ + { + Name: "wal-vol", + VolumeSource: kubevirtv1.VolumeSource{ + DataVolume: &kubevirtv1.DataVolumeSource{Name: "wal"}, + }, + }, + { + Name: "pgdata-vol", + VolumeSource: kubevirtv1.VolumeSource{ + DataVolume: &kubevirtv1.DataVolumeSource{Name: "pgdata"}, + }, + }, + } + + outTemplates, outVolumes := orchestrator.NamespaceDataVolumes(templates, volumes, "virtwork-db-1") + Expect(outTemplates).To(HaveLen(2)) + Expect(outTemplates[0].Name).To(Equal("wal-virtwork-db-1")) + Expect(outTemplates[1].Name).To(Equal("pgdata-virtwork-db-1")) + Expect(outVolumes[0].DataVolume.Name).To(Equal("wal-virtwork-db-1")) + Expect(outVolumes[1].DataVolume.Name).To(Equal("pgdata-virtwork-db-1")) + }) + + It("should not modify volumes without DataVolume source", func() { + templates := []kubevirtv1.DataVolumeTemplateSpec{ + {ObjectMeta: metav1.ObjectMeta{Name: "data"}}, + } + volumes := []kubevirtv1.Volume{ + {Name: "containerdisk"}, + { + Name: "datadisk", + VolumeSource: kubevirtv1.VolumeSource{ + DataVolume: &kubevirtv1.DataVolumeSource{Name: "data"}, + }, + }, + } + + _, outVolumes := orchestrator.NamespaceDataVolumes(templates, volumes, "vm-0") + Expect(outVolumes[0].DataVolume).To(BeNil()) + Expect(outVolumes[1].DataVolume.Name).To(Equal("data-vm-0")) + }) + + It("should not modify volumes referencing unknown DataVolumes", func() { + templates := []kubevirtv1.DataVolumeTemplateSpec{ + {ObjectMeta: metav1.ObjectMeta{Name: "data"}}, + } + volumes := []kubevirtv1.Volume{ + { + Name: "other", + VolumeSource: kubevirtv1.VolumeSource{ + DataVolume: &kubevirtv1.DataVolumeSource{Name: "not-in-templates"}, + }, + }, + } + + _, outVolumes := orchestrator.NamespaceDataVolumes(templates, volumes, "vm-0") + Expect(outVolumes[0].DataVolume.Name).To(Equal("not-in-templates")) + }) +}) + +var _ = Describe("VMPlan", func() { + It("should be constructible with all fields", func() { + plan := orchestrator.VMPlan{ + WorkloadName: "cpu", + VMName: "virtwork-cpu-0", + Component: "cpu", + Role: "", + VMSpec: &orchestrator.VMSpecInput{ + Name: "virtwork-cpu-0", + Namespace: "virtwork", + ContainerDiskImage: "quay.io/containerdisks/fedora:41", + CloudInitUserdata: "#cloud-config\n", + CPUCores: 2, + Memory: "2Gi", + Labels: map[string]string{ + "app.kubernetes.io/name": "virtwork-cpu", + }, + }, + } + Expect(plan.VMName).To(Equal("virtwork-cpu-0")) + Expect(plan.Component).To(Equal("cpu")) + Expect(plan.VMSpec.CPUCores).To(Equal(2)) + }) + + It("should support role field for multi-VM workloads", func() { + plan := orchestrator.VMPlan{ + WorkloadName: "network", + VMName: "virtwork-network-server-0", + Component: "network", + Role: "server", + VMSpec: &orchestrator.VMSpecInput{ + Name: "virtwork-network-server-0", + Namespace: "virtwork", + Labels: map[string]string{ + "virtwork/role": "server", + }, + }, + } + Expect(plan.Role).To(Equal("server")) + Expect(plan.VMSpec.Labels["virtwork/role"]).To(Equal("server")) + }) +}) + +var _ = Describe("RunResult", func() { + It("should hold summary counts", func() { + result := orchestrator.RunResult{ + RunID: "abc-123", + VMCount: 5, + ServiceCount: 1, + SecretCount: 5, + } + Expect(result.RunID).To(Equal("abc-123")) + Expect(result.VMCount).To(Equal(5)) + Expect(result.ServiceCount).To(Equal(1)) + Expect(result.SecretCount).To(Equal(5)) + }) +}) + +var _ = Describe("NewRunOrchestrator", func() { + It("should construct with all dependencies", func() { + ro := orchestrator.NewRunOrchestrator(nil, nil, nil, nil, nil) + Expect(ro).NotTo(BeNil()) + }) +}) + +var _ = Describe("NewCleanupOrchestrator", func() { + It("should construct with all dependencies", func() { + co := orchestrator.NewCleanupOrchestrator(nil, nil, nil, nil, nil) + Expect(co).NotTo(BeNil()) + }) +}) + +// Ensure ObjectMeta Name field is accessible in test context. +var _ = fmt.Sprintf