diff --git a/.golangci.yml b/.golangci.yml index e5b21b0..9d01b03 100644 --- a/.golangci.yml +++ b/.golangci.yml @@ -1,6 +1,8 @@ version: "2" run: allow-parallel-runners: true +issues: + fix: true linters: default: none enable: @@ -28,6 +30,8 @@ linters: - name: import-shadowing exclusions: generated: lax + presets: + - std-error-handling rules: - linters: - lll @@ -36,6 +40,9 @@ linters: - dupl - lll path: internal/* + - linters: + - errcheck + path: _test\.go paths: - third_party$ - builtin$ @@ -44,6 +51,11 @@ formatters: enable: - gofmt - goimports + settings: + gofmt: + rewrite-rules: + - pattern: interface{} + replacement: any exclusions: generated: lax paths: diff --git a/Makefile b/Makefile index 1f1c0bb..fdeac28 100644 --- a/Makefile +++ b/Makefile @@ -309,7 +309,7 @@ CONTROLLER_TOOLS_VERSION ?= v0.18.0 ENVTEST_VERSION ?= $(shell go list -m -f "{{ .Version }}" sigs.k8s.io/controller-runtime | awk -F'[v.]' '{printf "release-%d.%d", $$2, $$3}') #ENVTEST_K8S_VERSION is the version of Kubernetes to use for setting up ENVTEST binaries (i.e. 1.31) ENVTEST_K8S_VERSION ?= $(shell go list -m -f "{{ .Version }}" k8s.io/api | awk -F'[v.]' '{printf "1.%d", $$3}') -GOLANGCI_LINT_VERSION ?= v2.1.0 +GOLANGCI_LINT_VERSION ?= v2.11.4 .PHONY: kustomize kustomize: $(KUSTOMIZE) ## Download kustomize locally if necessary. diff --git a/cmd/controller/main.go b/cmd/controller/main.go index efbdac2..78b73aa 100644 --- a/cmd/controller/main.go +++ b/cmd/controller/main.go @@ -176,7 +176,9 @@ func convertToResourceProfiles(profiles []ResourceProfileConfig) ([]loomv1alpha1 } // initializeSpindle creates a spindle server with KubernetesEngine -func initializeSpindle(ctx context.Context, cfg *config.Config, mgr ctrl.Manager, loomCfg *LoomConfig) (*spindle.Spindle, error) { +func initializeSpindle( + ctx context.Context, cfg *config.Config, mgr ctrl.Manager, loomCfg *LoomConfig, +) (*spindle.Spindle, error) { // Initialize Kubernetes engine // Get namespace from environment (injected via Downward API) namespace := os.Getenv("POD_NAMESPACE") diff --git a/config/crd/bases/loom.j5t.io_spindlesets.yaml b/config/crd/bases/loom.j5t.io_spindlesets.yaml index ccd9d61..8ba1c14 100644 --- a/config/crd/bases/loom.j5t.io_spindlesets.yaml +++ b/config/crd/bases/loom.j5t.io_spindlesets.yaml @@ -1240,9 +1240,10 @@ spec: operator: description: |- Operator represents a key's relationship to the value. - Valid operators are Exists and Equal. Defaults to Equal. + Valid operators are Exists, Equal, Lt, and Gt. Defaults to Equal. Exists is equivalent to wildcard for value, so that a pod can tolerate all taints of a particular category. + Lt and Gt perform numeric comparisons (requires feature gate TaintTolerationComparisonOperators). type: string tolerationSeconds: description: |- diff --git a/internal/controller/spindleset_controller.go b/internal/controller/spindleset_controller.go index ad4f0b0..f20515f 100644 --- a/internal/controller/spindleset_controller.go +++ b/internal/controller/spindleset_controller.go @@ -42,6 +42,11 @@ import ( "tangled.org/evan.jarrett.net/loom/internal/jobbuilder" ) +const ( + phaseSucceeded = "Succeeded" + phaseFailed = "Failed" +) + // SpindleSetReconciler reconciles a SpindleSet object type SpindleSetReconciler struct { client.Client @@ -129,7 +134,7 @@ func (r *SpindleSetReconciler) Reconcile(ctx context.Context, req ctrl.Request) } // If SpindleSet is terminal, don't create new jobs — check for TTL cleanup - if spindleSet.Status.Phase == "Succeeded" || spindleSet.Status.Phase == "Failed" { + if spindleSet.Status.Phase == phaseSucceeded || spindleSet.Status.Phase == phaseFailed { age := time.Since(spindleSet.CreationTimestamp.Time) const spindleSetMaxLifetime = 4 * time.Hour if age > spindleSetMaxLifetime { @@ -316,16 +321,16 @@ func (r *SpindleSetReconciler) updateStatus(ctx context.Context, spindleSet *loo // Compute Phase with latch semantics: once terminal, never revert. // This prevents job recreation after TTL deletes completed Jobs. - if spindleSet.Status.Phase != "Succeeded" && spindleSet.Status.Phase != "Failed" { + if spindleSet.Status.Phase != phaseSucceeded && spindleSet.Status.Phase != phaseFailed { expectedWorkflows := int32(len(spindleSet.Spec.PipelineRun.Workflows)) totalTerminal := completed + failed var newPhase string switch { case expectedWorkflows > 0 && totalTerminal >= expectedWorkflows && failed > 0: - newPhase = "Failed" + newPhase = phaseFailed case expectedWorkflows > 0 && totalTerminal >= expectedWorkflows: - newPhase = "Succeeded" + newPhase = phaseSucceeded case running > 0: newPhase = "Running" default: @@ -405,7 +410,7 @@ func setCondition(conditions *[]metav1.Condition, newCondition metav1.Condition) // ensurePipelineJobs ensures Jobs are created for all workflows in a pipeline run func (r *SpindleSetReconciler) ensurePipelineJobs(ctx context.Context, spindleSet *loomv1alpha1.SpindleSet) error { // Don't recreate jobs for completed pipelines - if spindleSet.Status.Phase == "Succeeded" || spindleSet.Status.Phase == "Failed" { + if spindleSet.Status.Phase == phaseSucceeded || spindleSet.Status.Phase == phaseFailed { return nil } @@ -465,7 +470,7 @@ func (r *SpindleSetReconciler) ensurePipelineJobs(ctx context.Context, spindleSe // List nodes for profile selection (to validate nodeSelector labels exist) var nodeList corev1.NodeList - if err := r.Client.List(ctx, &nodeList); err != nil { + if err := r.List(ctx, &nodeList); err != nil { return fmt.Errorf("failed to list nodes: %w", err) } diff --git a/internal/engine/kubernetes_engine.go b/internal/engine/kubernetes_engine.go index bbb06a6..29fe292 100644 --- a/internal/engine/kubernetes_engine.go +++ b/internal/engine/kubernetes_engine.go @@ -11,7 +11,7 @@ import ( "sync" "time" - "github.com/cyphar/filepath-securejoin" + securejoin "github.com/cyphar/filepath-securejoin" "gopkg.in/yaml.v3" batchv1 "k8s.io/api/batch/v1" corev1 "k8s.io/api/core/v1" @@ -59,9 +59,9 @@ type KubernetesEngine struct { } // NewKubernetesEngine creates a new Kubernetes-based spindle engine. -func NewKubernetesEngine(client client.Client, config *rest.Config, namespace string, template loomv1alpha1.SpindleTemplate, vault secrets.Manager) *KubernetesEngine { +func NewKubernetesEngine(k8sClient client.Client, config *rest.Config, namespace string, template loomv1alpha1.SpindleTemplate, vault secrets.Manager) *KubernetesEngine { return &KubernetesEngine{ - client: client, + client: k8sClient, config: config, namespace: namespace, template: template, @@ -164,7 +164,7 @@ func (e *KubernetesEngine) InitWorkflow(twf tangled.Pipeline_Workflow, tpl tangl // SetupWorkflow creates a SpindleSet CR for the workflow. func (e *KubernetesEngine) SetupWorkflow(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, wfLogger models.WorkflowLogger) error { - logger := log.FromContext(ctx).WithValues("workflow", wid.Name, "pipeline", wid.PipelineId.Rkey) + logger := log.FromContext(ctx).WithValues("workflow", wid.Name, "pipeline", wid.Rkey) // Extract pre-computed workflow data data, ok := wf.Data.(*kubernetesWorkflowData) @@ -216,7 +216,7 @@ func (e *KubernetesEngine) SetupWorkflow(ctx context.Context, wid models.Workflo sanitizedWorkflowName = strings.TrimSuffix(sanitizedWorkflowName, ".yml") sanitizedWorkflowName = strings.ReplaceAll(sanitizedWorkflowName, ".", "-") - spindleSetName := fmt.Sprintf("spindle-%s-%s", sanitizedWorkflowName, wid.PipelineId.Rkey) + spindleSetName := fmt.Sprintf("spindle-%s-%s", sanitizedWorkflowName, wid.Rkey) if len(spindleSetName) > 63 { // Kubernetes names must be 63 chars or less spindleSetName = spindleSetName[:63] @@ -226,7 +226,7 @@ func (e *KubernetesEngine) SetupWorkflow(ctx context.Context, wid models.Workflo // Knot is extracted from the pipeline ID provided by the framework skipClone := len(data.CloneStep.Commands()) == 0 pipelineRun := &loomv1alpha1.PipelineRunSpec{ - PipelineID: wid.PipelineId.Rkey, + PipelineID: wid.Rkey, SkipClone: skipClone, Secrets: repoSecrets, Workflows: []loomv1alpha1.WorkflowSpec{data.Spec}, @@ -244,7 +244,7 @@ func (e *KubernetesEngine) SetupWorkflow(ctx context.Context, wid models.Workflo Namespace: e.namespace, Labels: map[string]string{ "loom.j5t.io/component": "spindle", - "loom.j5t.io/pipeline-id": wid.PipelineId.Rkey, + "loom.j5t.io/pipeline-id": wid.Rkey, "loom.j5t.io/workflow": wf.Name, }, }, @@ -286,7 +286,7 @@ func (e *KubernetesEngine) getSpindleSet(ctx context.Context, wid models.Workflo if err := e.client.List(ctx, spindleSetList, client.InNamespace(e.namespace), client.MatchingLabels{ - "loom.j5t.io/pipeline-id": wid.PipelineId.Rkey, + "loom.j5t.io/pipeline-id": wid.Rkey, "loom.j5t.io/workflow": wid.Name, }); err != nil { return nil, fmt.Errorf("failed to query SpindleSet: %w", err) @@ -304,7 +304,7 @@ func (e *KubernetesEngine) getSpindleSet(ctx context.Context, wid models.Workflo // DestroyWorkflow cleans up the SpindleSet after completion. func (e *KubernetesEngine) DestroyWorkflow(ctx context.Context, wid models.WorkflowId) error { - logger := log.FromContext(ctx).WithValues("workflow", wid.Name, "pipeline", wid.PipelineId.Rkey) + logger := log.FromContext(ctx).WithValues("workflow", wid.Name, "pipeline", wid.Rkey) spindleSet, err := e.getSpindleSet(ctx, wid) if err != nil { @@ -352,8 +352,8 @@ func (e *KubernetesEngine) DestroyWorkflow(ctx context.Context, wid models.Workf // RunStep streams logs for the specific step and waits for that step to complete. // For Kubernetes engine, all steps run in a single Job, but we stream logs incrementally // as each step executes. Each RunStep call blocks until that step's "end" control event is received. -func (e *KubernetesEngine) RunStep(ctx context.Context, wid models.WorkflowId, w *models.Workflow, idx int, secrets []secrets.UnlockedSecret, wfLogger models.WorkflowLogger) error { - logger := log.FromContext(ctx).WithValues("workflow", wid.Name, "pipeline", wid.PipelineId.Rkey, "step", idx) +func (e *KubernetesEngine) RunStep(ctx context.Context, wid models.WorkflowId, w *models.Workflow, idx int, wfSecrets []secrets.UnlockedSecret, wfLogger models.WorkflowLogger) error { + logger := log.FromContext(ctx).WithValues("workflow", wid.Name, "pipeline", wid.Rkey, "step", idx) // Query for the Job created by SpindleSetReconciler (only on first step) var job *batchv1.Job @@ -379,7 +379,7 @@ func (e *KubernetesEngine) RunStep(ctx context.Context, wid models.WorkflowId, w client.MatchingLabels{ "loom.j5t.io/spindleset": spindleSet.Name, "loom.j5t.io/workflow": w.Name, - "loom.j5t.io/pipeline-id": wid.PipelineId.Rkey, + "loom.j5t.io/pipeline-id": wid.Rkey, }) if err != nil { return fmt.Errorf("failed to list jobs: %w", err) @@ -437,7 +437,7 @@ func (e *KubernetesEngine) getOrCreateLogStream(ctx context.Context, wid models. } // Create new stream - logger := log.FromContext(ctx).WithValues("workflow", wid.Name, "pipeline", wid.PipelineId.Rkey) + logger := log.FromContext(ctx).WithValues("workflow", wid.Name, "pipeline", wid.Rkey) // Create kubernetes clientset for log streaming clientset, err := kubernetes.NewForConfig(e.config) @@ -591,7 +591,7 @@ func (e *KubernetesEngine) readUntilStepEnd(ctx context.Context, stream *workflo logLine.Stream = "stdout" // Default to stdout } dataWriter := wfLogger.DataWriter(logLine.StepId, logLine.Stream) - dataWriter.Write([]byte(logLine.Content + "\n")) + _, _ = dataWriter.Write([]byte(logLine.Content + "\n")) } } diff --git a/internal/jobbuilder/job_template.go b/internal/jobbuilder/job_template.go index 7f6aa80..43c8bf1 100644 --- a/internal/jobbuilder/job_template.go +++ b/internal/jobbuilder/job_template.go @@ -287,7 +287,7 @@ func BuildJob(config WorkflowConfig, nodes *corev1.NodeList) (*batchv1.Job, erro // - Engine-specific vars (PATH, TANGLED_ARCHITECTURE, HOME) set in InitWorkflow // - Pipeline-level vars (TANGLED_REPO_*, TANGLED_REF, CI, etc.) injected by framework func buildEnvironmentVariables(config WorkflowConfig) []corev1.EnvVar { - var env []corev1.EnvVar + env := make([]corev1.EnvVar, 0, len(config.WorkflowSpec.Environment)) for key, value := range config.WorkflowSpec.Environment { env = append(env, corev1.EnvVar{ Name: key,