feat(operator): prepare staging reconciliation infrastructure

Add status tracking helpers, resource validation functions, and observed
generation tracking to prepare for staging reconciliation. Includes
InternalStatus helpers and component creation utilities.

Author: DanielBlei

.golangci.yml

@@ -22,6 +22,8 @@ linters:
     - unparam
     - unused
   settings:
+    lll:
+      line-length: 120
     revive:
       rules:
         - name: comment-spacings
@@ -47,6 +49,11 @@ formatters:
   enable:
     - gofmt
     - goimports
+    - golines
+  settings:
+    golines:
+      max-len: 120
+      tab-len: 1
   exclusions:
     generated: lax
     paths:

operator/Makefile

@@ -1,6 +1,9 @@
 # Image URL to use all building/pushing image targets
 IMG ?= controller:latest
 
+# Golangci Config File
+GOLANGCI_CONFIG ?= ../.golangci.yml
+
 # Get the currently used golang install path (in GOPATH/bin, unless GOBIN is set)
 ifeq (,$(shell go env GOBIN))
 GOBIN=$(shell go env GOPATH)/bin
@@ -73,11 +76,16 @@ cleanup-test-e2e: ## Tear down the Kind cluster used for e2e tests
 
 .PHONY: lint
 lint: golangci-lint ## Run golangci-lint linter
-	$(GOLANGCI_LINT) run
+	$(GOLANGCI_LINT) run --config $(GOLANGCI_CONFIG)
 
 .PHONY: lint-fix
-lint-fix: golangci-lint ## Run golangci-lint linter and perform fixes
-	$(GOLANGCI_LINT) run --fix
+lint-fix: golangci-lint ## Run golangci-lint linter and perform fixes, then format long lines
+	$(GOLANGCI_LINT) run --fix --config $(GOLANGCI_CONFIG)
+	$(MAKE) format-lines
+
+.PHONY: format-lines
+format-lines: golines ## Format long lines using golines
+	$(GOLINES) -w -m 120 -t 1 .
 
 .PHONY: lint-config
 lint-config: golangci-lint ## Verify golangci-lint linter configuration
@@ -180,6 +188,7 @@ KUSTOMIZE ?= $(LOCALBIN)/kustomize
 CONTROLLER_GEN ?= $(LOCALBIN)/controller-gen
 ENVTEST ?= $(LOCALBIN)/setup-envtest
 GOLANGCI_LINT = $(LOCALBIN)/golangci-lint
+GOLINES ?= $(LOCALBIN)/golines
 
 # Tool Versions
 KUSTOMIZE_VERSION ?= v5.6.0
@@ -189,6 +198,7 @@ ENVTEST_VERSION ?= $(shell go list -m -f "{{ .Version }}" sigs.k8s.io/controller
 #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.6
+GOLINES_VERSION ?= v0.12.1
 
 .PHONY: kustomize
 kustomize: $(KUSTOMIZE) ## Download kustomize locally if necessary.
@@ -218,6 +228,11 @@ golangci-lint: $(GOLANGCI_LINT) ## Download golangci-lint locally if necessary.
 $(GOLANGCI_LINT): $(LOCALBIN)
 	$(call go-install-tool,$(GOLANGCI_LINT),github.com/golangci/golangci-lint/v2/cmd/golangci-lint,$(GOLANGCI_LINT_VERSION))
 
+.PHONY: golines
+golines: $(GOLINES) ## Download golines locally if necessary.
+$(GOLINES): $(LOCALBIN)
+	$(call go-install-tool,$(GOLINES),github.com/segmentio/golines,$(GOLINES_VERSION))
+
 # go-install-tool will 'go install' any package with custom target and name of binary, if it doesn't exist
 # $1 - target path with name of binary
 # $2 - package url which can be installed

operator/api/v1alpha1/common.go

@@ -40,3 +40,16 @@ const (
 	IngestTypeJob     IngestType = "job"
 	IngestTypeTrigger IngestType = "trigger"
 )
+
+// ObjectCondition represents the condition of an object (Ingest, Transform, Trigger, etc.)
+type ObjCondition string
+
+const (
+	ObjectStatusCompleted  ObjCondition = "Completed"
+	ObjConditionFailed     ObjCondition = "Failed"
+	ObjConditionPending    ObjCondition = "Pending"
+	ObjConditionRunning    ObjCondition = "Running"
+	ObjConditionSuspended  ObjCondition = "Suspended"
+	ObjConditionReady      ObjCondition = "Ready"
+	ObjConditionInitiating ObjCondition = "Initiating"
+)

operator/api/v1alpha1/staging_status.go

@@ -75,12 +75,10 @@ type IngestSpec struct {
 	// +optional
 	Image string `json:"image"`
 
-	// Args specifies the command-line arguments to pass to the container,
-	// overriding the default container command (entrypoint).
-	// If set, these arguments will replace the container's default command.
+	// Command specifies the command to run in the container.
 	//
 	// +optional
-	Args []string `json:"args,omitempty"`
+	Command []string `json:"command,omitempty"`
 
 	// Resources defines CPU and memory requests/limits for the Cronjob/Job container.
 	// If Type is "trigger", this field is ignored.
@@ -220,7 +218,7 @@ type StagingSpec struct {
 // +kubebuilder:object:generate=true
 type InternalStatus struct {
 	// Status of the associated staging object (e.g., Complete, Running, Failed, Pending, Unknown).
-	Status *StagingCondition `json:"status,omitempty"`
+	Status *ObjCondition `json:"status,omitempty"`
 
 	// Message provides additional information or error messages about the step status.
 	Message string `json:"message,omitempty"`
@@ -256,10 +254,15 @@ type InternalStatus struct {
 	FailedAttempts int32 `json:"failedAttempts,omitempty"`
 }
 
+// SetInternalStatus is a helper function to set the internal status condition
+func (s *InternalStatus) SetInternalStatus(condition ObjCondition) {
+	s.Status = &condition
+}
+
 // StagingStatus defines the observed state of a Staging resource.
 type StagingStatus struct {
 	// Status is the status of the Staging (e.g., Deployed, Failed, Pending, Running).
-	Status *StagingCondition `json:"status,omitempty"`
+	Status *ObjCondition `json:"status,omitempty"`
 
 	// ObservedGeneration is the most recent generation observed by the controller.
 	ObservedGeneration int64 `json:"observedGeneration,omitempty"`
@@ -274,6 +277,11 @@ type StagingStatus struct {
 	Conditions []metav1.Condition `json:"conditions,omitempty" patchStrategy:"merge" patchMergeKey:"type"`
 }
 
+// SetStagingStatus is a helper function to set the staging status condition
+func (s *StagingStatus) SetStagingStatus(condition ObjCondition) {
+	s.Status = &condition
+}
+
 // +kubebuilder:object:root=true
 // +kubebuilder:storageversion
 // +kubebuilder:subresource:status
@@ -300,16 +308,11 @@ type Staging struct {
 // +kubebuilder:object:root=true
 // StagingList contains a list of Staging
 type StagingList struct {
-	metav1.TypeMeta `json:",inline"`
-	metav1.ListMeta `json:"metadata,omitempty"`
+	metav1.TypeMeta `          json:",inline"`
+	metav1.ListMeta `          json:"metadata,omitempty"`
 	Items           []Staging `json:"items"`
 }
 
-// SetStatus is a helper function to set the staging status condition
-func (s *StagingStatus) SetStatus(condition StagingCondition) {
-	s.Status = &condition
-}
-
 func init() {
 	SchemeBuilder.Register(&Staging{}, &StagingList{})
 }

operator/api/v1alpha1/staging_types.go

@@ -84,7 +84,7 @@ type TriggerSpec struct {
 	//
 	// +kubebuilder:validation:Enum=gcs;pubsub;bigquery
 	// +kubebuilder:validation:Required
-	Type string `json:"type"`
+	Type TriggerType `json:"type"`
 
 	// Name of the trigger resource.
 	//
@@ -175,7 +175,7 @@ type TriggerSpec struct {
 // +kubebuilder:object:generate=true
 type TriggerStatus struct {
 	// Status is the status of the Trigger (e.g., Deployed, Failed, Pending, Running).
-	Status *TriggerCondition `json:"status,omitempty"`
+	Status *ObjCondition `json:"status,omitempty"`
 
 	// ObservedGeneration is the most recent generation observed by the controller.
 	ObservedGeneration int64 `json:"observedGeneration,omitempty"`
@@ -221,6 +221,11 @@ type TriggerStatus struct {
 	LastFailureTime *metav1.Time `json:"lastFailureTime,omitempty"`
 }
 
+// SetTriggerStatus is a helper function to set the trigger status condition
+func (t *TriggerStatus) SetTriggerStatus(condition ObjCondition) {
+	t.Status = &condition
+}
+
 // +kubebuilder:object:root=true
 // +kubebuilder:storageversion
 // +kubebuilder:subresource:status
@@ -249,16 +254,11 @@ type Trigger struct {
 
 // TriggerList contains a list of Trigger
 type TriggerList struct {
-	metav1.TypeMeta `json:",inline"`
-	metav1.ListMeta `json:"metadata,omitempty"`
+	metav1.TypeMeta `          json:",inline"`
+	metav1.ListMeta `          json:"metadata,omitempty"`
 	Items           []Trigger `json:"items"`
 }
 
-// SetStatus is a helper function to set the trigger status condition
-func (t *TriggerStatus) SetStatus(condition TriggerCondition) {
-	t.Status = &condition
-}
-
 func init() {
 	SchemeBuilder.Register(&Trigger{}, &TriggerList{})
 }

operator/api/v1alpha1/trigger_status.go

@@ -112,7 +112,9 @@ func main() {
 
 	if len(webhookCertPath) > 0 {
 		setupLog.Info("Initializing webhook certificate watcher using provided certificates",
-			"webhook-cert-path", webhookCertPath, "webhook-cert-name", webhookCertName, "webhook-cert-key", webhookCertKey)
+			"webhook-cert-path", webhookCertPath,
+			"webhook-cert-name", webhookCertName,
+			"webhook-cert-key", webhookCertKey)
 
 		var err error
 		webhookCertWatcher, err = certwatcher.New(

operator/api/v1alpha1/trigger_types.go

@@ -75,11 +75,8 @@ spec:
                   by referencing a Kubernetes CronJob or Job or by referencing a Trigger
                   CRD
                 properties:
-                  args:
-                    description: |-
-                      Args specifies the command-line arguments to pass to the container,
-                      overriding the default container command (entrypoint).
-                      If set, these arguments will replace the container's default command.
+                  command:
+                    description: Command specifies the command to run in the container.
                     items:
                       type: string
                     type: array