feat(RELEASE-914): add collectors phase

This change enables the execution of the collectors pipeline.

Signed-off-by: David Moreno García <[email protected]>

api/v1alpha1/collectors.go

@@ -1,8 +1,22 @@
 package v1alpha1
 
-// Collector represents a reference to a Collector to be executed as part of the release workflow.
+// Collectors holds the list of collectors to be executed as part of the release workflow along with the
+// ServiceAccount to be used in the PipelineRun.
 // +kubebuilder:object:generate=true
-type Collector struct {
+type Collectors struct {
+	// Items is the list of Collectors to be executed as part of the release workflow
+	// +required
+	Items []CollectorItem `json:"items"`
+
+	// ServiceAccountName is the ServiceAccount to use during the execution of the Collectors Pipeline
+	// +kubebuilder:validation:Pattern=^[a-z0-9]([-a-z0-9]*[a-z0-9])?$
+	// +optional
+	ServiceAccountName string `json:"serviceAccountName,omitempty"`
+}
+
+// CollectorItem represents all the information about an specific collector which will be executed in the
+// CollectorsPipeline.
+type CollectorItem struct {
 	// Name of the collector
 	// +kubebuilder:validation:Pattern=^[a-z0-9]([-a-z0-9]*[a-z0-9])?$
 	// +required

api/v1alpha1/release_conditions.go

@@ -6,9 +6,15 @@ const (
 	// finalProcessedConditionType is the type used to track the status of a Release Final Pipeline processing
 	finalProcessedConditionType conditions.ConditionType = "FinalPipelineProcessed"
 
+	// managedCollectorsProcessedConditionType is the type used to track the status of a Release Managed Collectors Pipeline processing
+	managedCollectorsProcessedConditionType conditions.ConditionType = "ManagedCollectorsPipelineProcessed"
+
 	// managedProcessedConditionType is the type used to track the status of a Release Managed Pipeline processing
 	managedProcessedConditionType conditions.ConditionType = "ManagedPipelineProcessed"
 
+	// tenantCollectorsProcessedConditionType is the type used to track the status of a Release Tenant Collectors Pipeline processing
+	tenantCollectorsProcessedConditionType conditions.ConditionType = "TenantCollectorsPipelineProcessed"
+
 	// tenantProcessedConditionType is the type used to track the status of a Release Tenant Pipeline processing
 	tenantProcessedConditionType conditions.ConditionType = "TenantPipelineProcessed"
 

api/v1alpha1/release_types.go

@@ -72,6 +72,10 @@ type ReleaseStatus struct {
 	// +optional
 	Conditions []metav1.Condition `json:"conditions"`
 
+	// CollectorsProcessing contains information about the release collectors processing
+	// +optional
+	CollectorsProcessing CollectorsInfo `json:"collectorsProcessing,omitempty"`
+
 	// FinalProcessing contains information about the release final processing
 	// +optional
 	FinalProcessing PipelineInfo `json:"finalProcessing,omitempty"`
@@ -121,6 +125,15 @@ type AttributionInfo struct {
 	StandingAuthorization bool `json:"standingAuthorization,omitempty"`
 }
 
+// CollectorsInfo defines the observed state of the release collectors.
+type CollectorsInfo struct {
+	// ManagedCollectorsProcessing contains information about the release managed collectors processing
+	ManagedCollectorsProcessing PipelineInfo `json:"managedCollectorsProcessing,omitempty"`
+
+	// TenantCollectorsProcessing contains information about the release tenant collectors processing
+	TenantCollectorsProcessing PipelineInfo `json:"tenantCollectorsProcessing,omitempty"`
+}
+
 // PipelineInfo defines the observed state of a release pipeline processing.
 type PipelineInfo struct {
 	// CompletionTime is the time when the Release processing was completed
@@ -175,11 +188,21 @@ func (r *Release) HasFinalPipelineProcessingFinished() bool {
 	return r.hasPhaseFinished(finalProcessedConditionType)
 }
 
+// HasManagedCollectorsPipelineProcessingFinished checks whether the Release Managed Collectors Pipeline processing has finished, regardless of the result.
+func (r *Release) HasManagedCollectorsPipelineProcessingFinished() bool {
+	return r.hasPhaseFinished(managedCollectorsProcessedConditionType)
+}
+
 // HasManagedPipelineProcessingFinished checks whether the Release Managed Pipeline processing has finished, regardless of the result.
 func (r *Release) HasManagedPipelineProcessingFinished() bool {
 	return r.hasPhaseFinished(managedProcessedConditionType)
 }
 
+// HasTenantCollectorsPipelineProcessingFinished checks whether the Release Tenant Collectors Pipeline processing has finished, regardless of the result.
+func (r *Release) HasTenantCollectorsPipelineProcessingFinished() bool {
+	return r.hasPhaseFinished(tenantCollectorsProcessedConditionType)
+}
+
 // HasTenantPipelineProcessingFinished checks whether the Release Tenant Pipeline processing has finished, regardless of the result.
 func (r *Release) HasTenantPipelineProcessingFinished() bool {
 	return r.hasPhaseFinished(tenantProcessedConditionType)
@@ -205,11 +228,21 @@ func (r *Release) IsFinalPipelineProcessed() bool {
 	return meta.IsStatusConditionTrue(r.Status.Conditions, finalProcessedConditionType.String())
 }
 
+// IsManagedCollectorsPipelineProcessed checks whether the Release Managed Collectors Pipeline was successfully processed.
+func (r *Release) IsManagedCollectorsPipelineProcessed() bool {
+	return meta.IsStatusConditionTrue(r.Status.Conditions, managedCollectorsProcessedConditionType.String())
+}
+
 // IsManagedPipelineProcessed checks whether the Release Managed Pipeline was successfully processed.
 func (r *Release) IsManagedPipelineProcessed() bool {
 	return meta.IsStatusConditionTrue(r.Status.Conditions, managedProcessedConditionType.String())
 }
 
+// IsTenantCollectorsPipelineProcessed checks whether the Release Tenant Collectors Pipeline was successfully processed.
+func (r *Release) IsTenantCollectorsPipelineProcessed() bool {
+	return meta.IsStatusConditionTrue(r.Status.Conditions, tenantCollectorsProcessedConditionType.String())
+}
+
 // IsTenantPipelineProcessed checks whether the Release Tenant Pipeline was successfully processed.
 func (r *Release) IsTenantPipelineProcessed() bool {
 	return meta.IsStatusConditionTrue(r.Status.Conditions, tenantProcessedConditionType.String())
@@ -220,11 +253,21 @@ func (r *Release) IsFinalPipelineProcessing() bool {
 	return r.isPhaseProgressing(finalProcessedConditionType)
 }
 
+// IsManagedCollectorsPipelineProcessing checks whether the Release Managed Collectors Pipeline processing is in progress.
+func (r *Release) IsManagedCollectorsPipelineProcessing() bool {
+	return r.isPhaseProgressing(managedCollectorsProcessedConditionType)
+}
+
 // IsManagedPipelineProcessing checks whether the Release Managed Pipeline processing is in progress.
 func (r *Release) IsManagedPipelineProcessing() bool {
 	return r.isPhaseProgressing(managedProcessedConditionType)
 }
 
+// IsTenantCollectorsPipelineProcessing checks whether the Release Tenant Collectors Pipeline processing is in progress.
+func (r *Release) IsTenantCollectorsPipelineProcessing() bool {
+	return r.isPhaseProgressing(tenantCollectorsProcessedConditionType)
+}
+
 // IsTenantPipelineProcessing checks whether the Release Tenant Pipeline processing is in progress.
 func (r *Release) IsTenantPipelineProcessing() bool {
 	return r.isPhaseProgressing(tenantProcessedConditionType)
@@ -263,6 +306,24 @@ func (r *Release) MarkFinalPipelineProcessed() {
 	)
 }
 
+// MarkManagedCollectorsPipelineProcessed marks the Release Managed Collectors Pipeline as processed.
+func (r *Release) MarkManagedCollectorsPipelineProcessed() {
+	if !r.IsManagedCollectorsPipelineProcessing() || r.HasManagedCollectorsPipelineProcessingFinished() {
+		return
+	}
+
+	r.Status.CollectorsProcessing.ManagedCollectorsProcessing.CompletionTime = &metav1.Time{Time: time.Now()}
+	conditions.SetCondition(&r.Status.Conditions, managedCollectorsProcessedConditionType, metav1.ConditionTrue, SucceededReason)
+
+	go metrics.RegisterCompletedReleasePipelineProcessing(
+		r.Status.CollectorsProcessing.ManagedCollectorsProcessing.StartTime,
+		r.Status.CollectorsProcessing.ManagedCollectorsProcessing.CompletionTime,
+		SucceededReason.String(),
+		r.Status.Target,
+		metadata.ManagedCollectorsPipelineType,
+	)
+}
+
 // MarkManagedPipelineProcessed marks the Release Managed Pipeline as processed.
 func (r *Release) MarkManagedPipelineProcessed() {
 	if !r.IsManagedPipelineProcessing() || r.HasManagedPipelineProcessingFinished() {
@@ -281,6 +342,24 @@ func (r *Release) MarkManagedPipelineProcessed() {
 	)
 }
 
+// MarkTenantCollectorsPipelineProcessed marks the Release Tenant Collectors Pipeline as processed.
+func (r *Release) MarkTenantCollectorsPipelineProcessed() {
+	if !r.IsTenantCollectorsPipelineProcessing() || r.HasTenantCollectorsPipelineProcessingFinished() {
+		return
+	}
+
+	r.Status.CollectorsProcessing.TenantCollectorsProcessing.CompletionTime = &metav1.Time{Time: time.Now()}
+	conditions.SetCondition(&r.Status.Conditions, tenantCollectorsProcessedConditionType, metav1.ConditionTrue, SucceededReason)
+
+	go metrics.RegisterCompletedReleasePipelineProcessing(
+		r.Status.CollectorsProcessing.TenantCollectorsProcessing.StartTime,
+		r.Status.CollectorsProcessing.TenantCollectorsProcessing.CompletionTime,
+		SucceededReason.String(),
+		r.Status.Target,
+		metadata.TenantCollectorsPipelineType,
+	)
+}
+
 // MarkTenantPipelineProcessed marks the Release Tenant Pipeline as processed.
 func (r *Release) MarkTenantPipelineProcessed() {
 	if !r.IsTenantPipelineProcessing() || r.HasTenantPipelineProcessingFinished() {
@@ -312,14 +391,35 @@ func (r *Release) MarkFinalPipelineProcessing() {
 	conditions.SetCondition(&r.Status.Conditions, finalProcessedConditionType, metav1.ConditionFalse, ProgressingReason)
 
 	go metrics.RegisterNewReleasePipelineProcessing(
-		r.Status.FinalProcessing.StartTime,
 		r.Status.StartTime,
+		r.Status.FinalProcessing.StartTime,
 		ProgressingReason.String(),
 		r.Status.Target,
 		metadata.FinalPipelineType,
 	)
 }
 
+// MarkManagedCollectorsPipelineProcessing marks the Release Managed Collectors Pipeline as processing.
+func (r *Release) MarkManagedCollectorsPipelineProcessing() {
+	if r.HasManagedCollectorsPipelineProcessingFinished() {
+		return
+	}
+
+	if !r.IsManagedCollectorsPipelineProcessing() {
+		r.Status.CollectorsProcessing.ManagedCollectorsProcessing.StartTime = &metav1.Time{Time: time.Now()}
+	}
+
+	conditions.SetCondition(&r.Status.Conditions, managedCollectorsProcessedConditionType, metav1.ConditionFalse, ProgressingReason)
+
+	go metrics.RegisterNewReleasePipelineProcessing(
+		r.Status.StartTime,
+		r.Status.CollectorsProcessing.ManagedCollectorsProcessing.StartTime,
+		ProgressingReason.String(),
+		r.Status.Target,
+		metadata.ManagedPipelineType,
+	)
+}
+
 // MarkManagedPipelineProcessing marks the Release Managed Pipeline as processing.
 func (r *Release) MarkManagedPipelineProcessing() {
 	if r.HasManagedPipelineProcessingFinished() {
@@ -333,14 +433,35 @@ func (r *Release) MarkManagedPipelineProcessing() {
 	conditions.SetCondition(&r.Status.Conditions, managedProcessedConditionType, metav1.ConditionFalse, ProgressingReason)
 
 	go metrics.RegisterNewReleasePipelineProcessing(
-		r.Status.ManagedProcessing.StartTime,
 		r.Status.StartTime,
+		r.Status.ManagedProcessing.StartTime,
 		ProgressingReason.String(),
 		r.Status.Target,
 		metadata.ManagedPipelineType,
 	)
 }
 
+// MarkTenantCollectorsPipelineProcessing marks the Release Tenant Collectors Pipeline as processing.
+func (r *Release) MarkTenantCollectorsPipelineProcessing() {
+	if r.HasTenantCollectorsPipelineProcessingFinished() {
+		return
+	}
+
+	if !r.IsTenantCollectorsPipelineProcessing() {
+		r.Status.CollectorsProcessing.TenantCollectorsProcessing.StartTime = &metav1.Time{Time: time.Now()}
+	}
+
+	conditions.SetCondition(&r.Status.Conditions, tenantCollectorsProcessedConditionType, metav1.ConditionFalse, ProgressingReason)
+
+	go metrics.RegisterNewReleasePipelineProcessing(
+		r.Status.StartTime,
+		r.Status.CollectorsProcessing.TenantCollectorsProcessing.StartTime,
+		ProgressingReason.String(),
+		r.Status.Target,
+		metadata.TenantCollectorsPipelineType,
+	)
+}
+
 // MarkTenantPipelineProcessing marks the Release Tenant Pipeline as processing.
 func (r *Release) MarkTenantPipelineProcessing() {
 	if r.HasTenantPipelineProcessingFinished() {
@@ -354,8 +475,8 @@ func (r *Release) MarkTenantPipelineProcessing() {
 	conditions.SetCondition(&r.Status.Conditions, tenantProcessedConditionType, metav1.ConditionFalse, ProgressingReason)
 
 	go metrics.RegisterNewReleasePipelineProcessing(
-		r.Status.TenantProcessing.StartTime,
 		r.Status.StartTime,
+		r.Status.TenantProcessing.StartTime,
 		ProgressingReason.String(),
 		r.Status.Target,
 		metadata.TenantPipelineType,
@@ -380,6 +501,24 @@ func (r *Release) MarkFinalPipelineProcessingFailed(message string) {
 	)
 }
 
+// MarkManagedCollectorsPipelineProcessingFailed marks the Release Managed Collectors Pipeline processing as failed.
+func (r *Release) MarkManagedCollectorsPipelineProcessingFailed(message string) {
+	if !r.IsManagedCollectorsPipelineProcessing() || r.HasManagedCollectorsPipelineProcessingFinished() {
+		return
+	}
+
+	r.Status.CollectorsProcessing.ManagedCollectorsProcessing.CompletionTime = &metav1.Time{Time: time.Now()}
+	conditions.SetConditionWithMessage(&r.Status.Conditions, managedCollectorsProcessedConditionType, metav1.ConditionFalse, FailedReason, message)
+
+	go metrics.RegisterCompletedReleasePipelineProcessing(
+		r.Status.CollectorsProcessing.ManagedCollectorsProcessing.StartTime,
+		r.Status.CollectorsProcessing.ManagedCollectorsProcessing.CompletionTime,
+		FailedReason.String(),
+		r.Status.Target,
+		metadata.ManagedCollectorsPipelineType,
+	)
+}
+
 // MarkManagedPipelineProcessingFailed marks the Release Managed Pipeline processing as failed.
 func (r *Release) MarkManagedPipelineProcessingFailed(message string) {
 	if !r.IsManagedPipelineProcessing() || r.HasManagedPipelineProcessingFinished() {
@@ -398,6 +537,24 @@ func (r *Release) MarkManagedPipelineProcessingFailed(message string) {
 	)
 }
 
+// MarkTenantCollectorsPipelineProcessingFailed marks the Release Tenant Collectors Pipeline processing as failed.
+func (r *Release) MarkTenantCollectorsPipelineProcessingFailed(message string) {
+	if !r.IsTenantCollectorsPipelineProcessing() || r.HasTenantCollectorsPipelineProcessingFinished() {
+		return
+	}
+
+	r.Status.CollectorsProcessing.TenantCollectorsProcessing.CompletionTime = &metav1.Time{Time: time.Now()}
+	conditions.SetConditionWithMessage(&r.Status.Conditions, tenantCollectorsProcessedConditionType, metav1.ConditionFalse, FailedReason, message)
+
+	go metrics.RegisterCompletedReleasePipelineProcessing(
+		r.Status.CollectorsProcessing.TenantCollectorsProcessing.StartTime,
+		r.Status.CollectorsProcessing.TenantCollectorsProcessing.CompletionTime,
+		FailedReason.String(),
+		r.Status.Target,
+		metadata.TenantCollectorsPipelineType,
+	)
+}
+
 // MarkTenantPipelineProcessingFailed marks the Release Tenant Pipeline processing as failed.
 func (r *Release) MarkTenantPipelineProcessingFailed(message string) {
 	if !r.IsTenantPipelineProcessing() || r.HasTenantPipelineProcessingFinished() {
@@ -425,6 +582,15 @@ func (r *Release) MarkFinalPipelineProcessingSkipped() {
 	conditions.SetCondition(&r.Status.Conditions, finalProcessedConditionType, metav1.ConditionTrue, SkippedReason)
 }
 
+// MarkManagedCollectorsPipelineProcessingSkipped marks the Release Managed Collectors Pipeline processing as skipped.
+func (r *Release) MarkManagedCollectorsPipelineProcessingSkipped() {
+	if r.HasManagedCollectorsPipelineProcessingFinished() {
+		return
+	}
+
+	conditions.SetCondition(&r.Status.Conditions, managedCollectorsProcessedConditionType, metav1.ConditionTrue, SkippedReason)
+}
+
 // MarkManagedPipelineProcessingSkipped marks the Release Managed Pipeline processing as skipped.
 func (r *Release) MarkManagedPipelineProcessingSkipped() {
 	if r.HasManagedPipelineProcessingFinished() {
@@ -434,6 +600,15 @@ func (r *Release) MarkManagedPipelineProcessingSkipped() {
 	conditions.SetCondition(&r.Status.Conditions, managedProcessedConditionType, metav1.ConditionTrue, SkippedReason)
 }
 
+// MarkTenantCollectorsPipelineProcessingSkipped marks the Release Tenant Collectors Pipeline processing as skipped.
+func (r *Release) MarkTenantCollectorsPipelineProcessingSkipped() {
+	if r.HasTenantCollectorsPipelineProcessingFinished() {
+		return
+	}
+
+	conditions.SetCondition(&r.Status.Conditions, tenantCollectorsProcessedConditionType, metav1.ConditionTrue, SkippedReason)
+}
+
 // MarkTenantPipelineProcessingSkipped marks the Release Tenant Pipeline processing as skipped.
 func (r *Release) MarkTenantPipelineProcessingSkipped() {
 	if r.HasTenantPipelineProcessingFinished() {
@@ -455,7 +630,9 @@ func (r *Release) MarkReleased() {
 	go metrics.RegisterCompletedRelease(
 		r.Status.StartTime,
 		r.Status.CompletionTime,
+		r.getPhaseReason(tenantCollectorsProcessedConditionType),
 		r.getPhaseReason(tenantProcessedConditionType),
+		r.getPhaseReason(managedCollectorsProcessedConditionType),
 		r.getPhaseReason(managedProcessedConditionType),
 		r.getPhaseReason(finalProcessedConditionType),
 		SucceededReason.String(),
@@ -491,7 +668,9 @@ func (r *Release) MarkReleaseFailed(message string) {
 	go metrics.RegisterCompletedRelease(
 		r.Status.StartTime,
 		r.Status.CompletionTime,
+		r.getPhaseReason(tenantCollectorsProcessedConditionType),
 		r.getPhaseReason(tenantProcessedConditionType),
+		r.getPhaseReason(managedCollectorsProcessedConditionType),
 		r.getPhaseReason(managedProcessedConditionType),
 		r.getPhaseReason(finalProcessedConditionType),
 		FailedReason.String(),