chore: Optimize reconciliation triggered by the self-monitor webhook (#…

…935)
kyma-project · Apr 4, 2024 · 05d90fd · 05d90fd
1 parent e6dfed3
commit 05d90fd
Show file tree

Hide file tree

Showing 21 changed files with 350 additions and 170 deletions.
diff --git a/controllers/telemetry/metricpipeline_controller.go b/controllers/telemetry/metricpipeline_controller.go
@@ -66,7 +66,7 @@ func (r *MetricPipelineController) SetupWithManager(mgr ctrl.Manager) error {
 
 	b.WatchesRawSource(
 		&source.Channel{Source: r.reconcileTriggerChan},
-		handler.EnqueueRequestsFromMapFunc(r.mapReconcileTriggerEvent),
+		&handler.EnqueueRequestForObject{},
 	)
 
 	ownedResourceTypesToWatch := []client.Object{
@@ -131,15 +131,6 @@ func (r *MetricPipelineController) mapTelemetryChanges(ctx context.Context, obje
 	return requests
 }
 
-func (r *MetricPipelineController) mapReconcileTriggerEvent(ctx context.Context, _ client.Object) []reconcile.Request {
-	logf.FromContext(ctx).V(1).Info("Reconcile trigger event received")
-	requests, err := r.createRequestsForAllPipelines(ctx)
-	if err != nil {
-		logf.FromContext(ctx).Error(err, "Unable to create reconcile requests")
-	}
-	return requests
-}
-
 func (r *MetricPipelineController) createRequestsForAllPipelines(ctx context.Context) ([]reconcile.Request, error) {
 	var pipelines telemetryv1alpha1.MetricPipelineList
 	var requests []reconcile.Request

diff --git a/controllers/telemetry/tracepipeline_controller.go b/controllers/telemetry/tracepipeline_controller.go
@@ -64,7 +64,7 @@ func (r *TracePipelineController) SetupWithManager(mgr ctrl.Manager) error {
 
 	b.WatchesRawSource(
 		&source.Channel{Source: r.reconcileTriggerChan},
-		handler.EnqueueRequestsFromMapFunc(r.mapReconcileTriggerEvent),
+		&handler.EnqueueRequestForObject{},
 	)
 
 	ownedResourceTypesToWatch := []client.Object{
@@ -110,15 +110,6 @@ func (r *TracePipelineController) mapTelemetryChanges(ctx context.Context, objec
 	return requests
 }
 
-func (r *TracePipelineController) mapReconcileTriggerEvent(ctx context.Context, _ client.Object) []reconcile.Request {
-	logf.FromContext(ctx).V(1).Info("Reconcile trigger event received")
-	requests, err := r.createRequestsForAllPipelines(ctx)
-	if err != nil {
-		logf.FromContext(ctx).Error(err, "Unable to create reconcile requests")
-	}
-	return requests
-}
-
 func (r *TracePipelineController) createRequestsForAllPipelines(ctx context.Context) ([]reconcile.Request, error) {
 	var pipelines telemetryv1alpha1.TracePipelineList
 	var requests []reconcile.Request

diff --git a/internal/reconciler/metricpipeline/mocks/flow_health_prober.go b/internal/reconciler/metricpipeline/mocks/flow_health_prober.go
diff --git a/internal/reconciler/metricpipeline/reconciler.go b/internal/reconciler/metricpipeline/reconciler.go
@@ -20,7 +20,7 @@ import (
 	"github.com/kyma-project/telemetry-manager/internal/overrides"
 	"github.com/kyma-project/telemetry-manager/internal/resources/otelcollector"
 	"github.com/kyma-project/telemetry-manager/internal/secretref"
-	"github.com/kyma-project/telemetry-manager/internal/selfmonitor/flowhealth"
+	"github.com/kyma-project/telemetry-manager/internal/selfmonitor/prober"
 )
 
 const defaultReplicaCount int32 = 2
@@ -44,7 +44,7 @@ type DaemonSetProber interface {
 
 //go:generate mockery --name FlowHealthProber --filename flow_health_prober.go
 type FlowHealthProber interface {
-	Probe(ctx context.Context, pipelineName string) (flowhealth.ProbeResult, error)
+	Probe(ctx context.Context, pipelineName string) (prober.ProbeResult, error)
 }
 
 type Reconciler struct {

diff --git a/internal/reconciler/metricpipeline/status.go b/internal/reconciler/metricpipeline/status.go
@@ -13,7 +13,7 @@ import (
 	telemetryv1alpha1 "github.com/kyma-project/telemetry-manager/apis/telemetry/v1alpha1"
 	"github.com/kyma-project/telemetry-manager/internal/conditions"
 	"github.com/kyma-project/telemetry-manager/internal/secretref"
-	"github.com/kyma-project/telemetry-manager/internal/selfmonitor/flowhealth"
+	"github.com/kyma-project/telemetry-manager/internal/selfmonitor/prober"
 )
 
 func (r *Reconciler) updateStatus(ctx context.Context, pipelineName string, withinPipelineCountLimit bool) error {
@@ -159,7 +159,7 @@ func (r *Reconciler) setFlowHealthCondition(ctx context.Context, pipeline *telem
 	meta.SetStatusCondition(&pipeline.Status.Conditions, condition)
 }
 
-func flowHealthReasonFor(probeResult flowhealth.ProbeResult) string {
+func flowHealthReasonFor(probeResult prober.ProbeResult) string {
 	if probeResult.AllDataDropped {
 		return conditions.ReasonAllDataDropped
 	}

diff --git a/internal/reconciler/metricpipeline/status_test.go b/internal/reconciler/metricpipeline/status_test.go
@@ -20,7 +20,7 @@ import (
 	"github.com/kyma-project/telemetry-manager/internal/conditions"
 	"github.com/kyma-project/telemetry-manager/internal/reconciler/metricpipeline/mocks"
 	"github.com/kyma-project/telemetry-manager/internal/resources/otelcollector"
-	"github.com/kyma-project/telemetry-manager/internal/selfmonitor/flowhealth"
+	"github.com/kyma-project/telemetry-manager/internal/selfmonitor/prober"
 	"github.com/kyma-project/telemetry-manager/internal/testutils"
 )
 
@@ -315,7 +315,7 @@ func TestUpdateStatus(t *testing.T) {
 	t.Run("flow healthy", func(t *testing.T) {
 		tests := []struct {
 			name           string
-			probe          flowhealth.ProbeResult
+			probe          prober.ProbeResult
 			probeErr       error
 			expectedStatus metav1.ConditionStatus
 			expectedReason string
@@ -328,31 +328,31 @@ func TestUpdateStatus(t *testing.T) {
 			},
 			{
 				name: "healthy",
-				probe: flowhealth.ProbeResult{
+				probe: prober.ProbeResult{
 					Healthy: true,
 				},
 				expectedStatus: metav1.ConditionTrue,
 				expectedReason: conditions.ReasonFlowHealthy,
 			},
 			{
 				name: "throttling",
-				probe: flowhealth.ProbeResult{
+				probe: prober.ProbeResult{
 					Throttling: true,
 				},
 				expectedStatus: metav1.ConditionFalse,
 				expectedReason: conditions.ReasonGatewayThrottling,
 			},
 			{
 				name: "buffer filling up",
-				probe: flowhealth.ProbeResult{
+				probe: prober.ProbeResult{
 					QueueAlmostFull: true,
 				},
 				expectedStatus: metav1.ConditionFalse,
 				expectedReason: conditions.ReasonBufferFillingUp,
 			},
 			{
 				name: "buffer filling up shadows other problems",
-				probe: flowhealth.ProbeResult{
+				probe: prober.ProbeResult{
 					QueueAlmostFull: true,
 					Throttling:      true,
 				},
@@ -361,15 +361,15 @@ func TestUpdateStatus(t *testing.T) {
 			},
 			{
 				name: "some data dropped",
-				probe: flowhealth.ProbeResult{
+				probe: prober.ProbeResult{
 					SomeDataDropped: true,
 				},
 				expectedStatus: metav1.ConditionFalse,
 				expectedReason: conditions.ReasonSomeDataDropped,
 			},
 			{
 				name: "some data dropped shadows other problems",
-				probe: flowhealth.ProbeResult{
+				probe: prober.ProbeResult{
 					SomeDataDropped: true,
 					Throttling:      true,
 				},
@@ -378,15 +378,15 @@ func TestUpdateStatus(t *testing.T) {
 			},
 			{
 				name: "all data dropped",
-				probe: flowhealth.ProbeResult{
+				probe: prober.ProbeResult{
 					AllDataDropped: true,
 				},
 				expectedStatus: metav1.ConditionFalse,
 				expectedReason: conditions.ReasonAllDataDropped,
 			},
 			{
 				name: "all data dropped shadows other problems",
-				probe: flowhealth.ProbeResult{
+				probe: prober.ProbeResult{
 					AllDataDropped: true,
 					Throttling:     true,
 				},

diff --git a/internal/reconciler/telemetry/reconciler.go b/internal/reconciler/telemetry/reconciler.go
@@ -23,8 +23,8 @@ import (
 	"github.com/kyma-project/telemetry-manager/internal/k8sutils"
 	"github.com/kyma-project/telemetry-manager/internal/overrides"
 	"github.com/kyma-project/telemetry-manager/internal/resources/selfmonitor"
+	"github.com/kyma-project/telemetry-manager/internal/selfmonitor/alertrules"
 	"github.com/kyma-project/telemetry-manager/internal/selfmonitor/config"
-	"github.com/kyma-project/telemetry-manager/internal/selfmonitor/flowhealth"
 	"github.com/kyma-project/telemetry-manager/internal/webhookcert"
 )
 
@@ -158,7 +158,7 @@ func (r *Reconciler) reconcileSelfMonitor(ctx context.Context, telemetry operato
 		return fmt.Errorf("failed to marshal selfmonitor config: %w", err)
 	}
 
-	rules := flowhealth.MakeRules()
+	rules := alertrules.MakeRules()
 	rulesYAML, err := yaml.Marshal(rules)
 	if err != nil {
 		return fmt.Errorf("failed to marshal rules: %w", err)

diff --git a/internal/reconciler/tracepipeline/mocks/flow_health_prober.go b/internal/reconciler/tracepipeline/mocks/flow_health_prober.go
diff --git a/internal/reconciler/tracepipeline/reconciler.go b/internal/reconciler/tracepipeline/reconciler.go
@@ -35,7 +35,7 @@ import (
 	"github.com/kyma-project/telemetry-manager/internal/overrides"
 	"github.com/kyma-project/telemetry-manager/internal/resources/otelcollector"
 	"github.com/kyma-project/telemetry-manager/internal/secretref"
-	"github.com/kyma-project/telemetry-manager/internal/selfmonitor/flowhealth"
+	"github.com/kyma-project/telemetry-manager/internal/selfmonitor/prober"
 )
 
 const defaultReplicaCount int32 = 2
@@ -53,7 +53,7 @@ type DeploymentProber interface {
 
 //go:generate mockery --name FlowHealthProber --filename flow_health_prober.go
 type FlowHealthProber interface {
-	Probe(ctx context.Context, pipelineName string) (flowhealth.ProbeResult, error)
+	Probe(ctx context.Context, pipelineName string) (prober.ProbeResult, error)
 }
 
 type Reconciler struct {

diff --git a/internal/reconciler/tracepipeline/status.go b/internal/reconciler/tracepipeline/status.go
@@ -13,7 +13,7 @@ import (
 	telemetryv1alpha1 "github.com/kyma-project/telemetry-manager/apis/telemetry/v1alpha1"
 	"github.com/kyma-project/telemetry-manager/internal/conditions"
 	"github.com/kyma-project/telemetry-manager/internal/secretref"
-	"github.com/kyma-project/telemetry-manager/internal/selfmonitor/flowhealth"
+	"github.com/kyma-project/telemetry-manager/internal/selfmonitor/prober"
 )
 
 func (r *Reconciler) updateStatus(ctx context.Context, pipelineName string, withinPipelineCountLimit bool) error {
@@ -133,7 +133,7 @@ func (r *Reconciler) setFlowHealthCondition(ctx context.Context, pipeline *telem
 	meta.SetStatusCondition(&pipeline.Status.Conditions, condition)
 }
 
-func flowHealthReasonFor(probeResult flowhealth.ProbeResult) string {
+func flowHealthReasonFor(probeResult prober.ProbeResult) string {
 	if probeResult.AllDataDropped {
 		return conditions.ReasonAllDataDropped
 	}

diff --git a/internal/reconciler/tracepipeline/status_test.go b/internal/reconciler/tracepipeline/status_test.go
@@ -19,7 +19,7 @@ import (
 	"github.com/kyma-project/telemetry-manager/internal/conditions"
 	"github.com/kyma-project/telemetry-manager/internal/reconciler/tracepipeline/mocks"
 	"github.com/kyma-project/telemetry-manager/internal/resources/otelcollector"
-	"github.com/kyma-project/telemetry-manager/internal/selfmonitor/flowhealth"
+	"github.com/kyma-project/telemetry-manager/internal/selfmonitor/prober"
 	"github.com/kyma-project/telemetry-manager/internal/testutils"
 )
 
@@ -349,7 +349,7 @@ func TestUpdateStatus(t *testing.T) {
 	t.Run("flow healthy", func(t *testing.T) {
 		tests := []struct {
 			name           string
-			probe          flowhealth.ProbeResult
+			probe          prober.ProbeResult
 			probeErr       error
 			expectedStatus metav1.ConditionStatus
 			expectedReason string
@@ -362,31 +362,31 @@ func TestUpdateStatus(t *testing.T) {
 			},
 			{
 				name: "healthy",
-				probe: flowhealth.ProbeResult{
+				probe: prober.ProbeResult{
 					Healthy: true,
 				},
 				expectedStatus: metav1.ConditionTrue,
 				expectedReason: conditions.ReasonFlowHealthy,
 			},
 			{
 				name: "throttling",
-				probe: flowhealth.ProbeResult{
+				probe: prober.ProbeResult{
 					Throttling: true,
 				},
 				expectedStatus: metav1.ConditionFalse,
 				expectedReason: conditions.ReasonGatewayThrottling,
 			},
 			{
 				name: "buffer filling up",
-				probe: flowhealth.ProbeResult{
+				probe: prober.ProbeResult{
 					QueueAlmostFull: true,
 				},
 				expectedStatus: metav1.ConditionFalse,
 				expectedReason: conditions.ReasonBufferFillingUp,
 			},
 			{
 				name: "buffer filling up shadows other problems",
-				probe: flowhealth.ProbeResult{
+				probe: prober.ProbeResult{
 					QueueAlmostFull: true,
 					Throttling:      true,
 				},
@@ -395,15 +395,15 @@ func TestUpdateStatus(t *testing.T) {
 			},
 			{
 				name: "some data dropped",
-				probe: flowhealth.ProbeResult{
+				probe: prober.ProbeResult{
 					SomeDataDropped: true,
 				},
 				expectedStatus: metav1.ConditionFalse,
 				expectedReason: conditions.ReasonSomeDataDropped,
 			},
 			{
 				name: "some data dropped shadows other problems",
-				probe: flowhealth.ProbeResult{
+				probe: prober.ProbeResult{
 					SomeDataDropped: true,
 					Throttling:      true,
 				},
@@ -412,15 +412,15 @@ func TestUpdateStatus(t *testing.T) {
 			},
 			{
 				name: "all data dropped",
-				probe: flowhealth.ProbeResult{
+				probe: prober.ProbeResult{
 					AllDataDropped: true,
 				},
 				expectedStatus: metav1.ConditionFalse,
 				expectedReason: conditions.ReasonAllDataDropped,
 			},
 			{
 				name: "all data dropped shadows other problems",
-				probe: flowhealth.ProbeResult{
+				probe: prober.ProbeResult{
 					AllDataDropped: true,
 					Throttling:     true,
 				},

diff --git a/...al/selfmonitor/flowhealth/expr_builder.go → ...al/selfmonitor/alertrules/expr_builder.go b/...al/selfmonitor/flowhealth/expr_builder.go → ...al/selfmonitor/alertrules/expr_builder.go
@@ -1,4 +1,4 @@
-package flowhealth
+package alertrules
 
 import (
 	"fmt"
@@ -16,7 +16,7 @@ type labelSelector func(string) string
 
 func selectService(serviceName string) labelSelector {
 	return func(metric string) string {
-		return fmt.Sprintf("%s{%s=\"%s\"}", metric, serviceLabelKey, serviceName)
+		return fmt.Sprintf("%s{%s=\"%s\"}", metric, LabelService, serviceName)
 	}
 }
 

diff --git a/internal/selfmonitor/alertrules/pipeline_type.go b/internal/selfmonitor/alertrules/pipeline_type.go
@@ -0,0 +1,8 @@
+package alertrules
+
+type PipelineType string
+
+const (
+	MetricPipeline PipelineType = "MetricPipeline"
+	TracePipeline  PipelineType = "TracePipeline"
+)