feat: Refactor the config for telemetry manager after kymastats chang…

…es (#1422)

internal/otelcollector/config/metric/gateway/config.go

@@ -43,7 +43,7 @@ type SingletonKymaStatsReceiver struct {
 type KymaStatsReceiver struct {
 	AuthType           string      `yaml:"auth_type"`
 	CollectionInterval string      `yaml:"collection_interval"`
-	Modules            []ModuleGVR `yaml:"modules"`
+	Resources          []ModuleGVR `yaml:"resources"`
 }
 
 type SingletonK8sClusterReceiver struct {

internal/otelcollector/config/metric/gateway/receivers.go

@@ -33,12 +33,27 @@ func makeSingletonKymaStatsReceiverCreatorConfig(gatewayNamespace string) *Singl
 			KymaStatsReceiver: KymaStatsReceiver{
 				AuthType:           "serviceAccount",
 				CollectionInterval: "30s",
-				Modules: []ModuleGVR{
+				Resources: []ModuleGVR{
 					{
 						Group:    "operator.kyma-project.io",
 						Version:  "v1alpha1",
 						Resource: "telemetries",
 					},
+					{
+						Group:    "telemetry.kyma-project.io",
+						Version:  "v1alpha1",
+						Resource: "logpipelines",
+					},
+					{
+						Group:    "telemetry.kyma-project.io",
+						Version:  "v1alpha1",
+						Resource: "metricpipelines",
+					},
+					{
+						Group:    "telemetry.kyma-project.io",
+						Version:  "v1alpha1",
+						Resource: "tracepipelines",
+					},
 				},
 			},
 		},

internal/otelcollector/config/metric/gateway/receivers_test.go

@@ -56,10 +56,34 @@ func TestReceivers(t *testing.T) {
 		kymaStatsReceiver := singletonKymaStatsReceiverCreator.SingletonKymaStatsReceiver.KymaStatsReceiver
 		require.Equal(t, "serviceAccount", kymaStatsReceiver.AuthType)
 		require.Equal(t, "30s", kymaStatsReceiver.CollectionInterval)
-		require.Len(t, kymaStatsReceiver.Modules, 1)
-		require.Equal(t, "operator.kyma-project.io", kymaStatsReceiver.Modules[0].Group)
-		require.Equal(t, "v1alpha1", kymaStatsReceiver.Modules[0].Version)
-		require.Equal(t, "telemetries", kymaStatsReceiver.Modules[0].Resource)
+		require.Len(t, kymaStatsReceiver.Resources, 4)
+		expectedResources := []ModuleGVR{
+			{
+				Group:    "operator.kyma-project.io",
+				Version:  "v1alpha1",
+				Resource: "telemetries",
+			},
+			{
+				Group:    "telemetry.kyma-project.io",
+				Version:  "v1alpha1",
+				Resource: "logpipelines",
+			},
+			{
+				Group:    "telemetry.kyma-project.io",
+				Version:  "v1alpha1",
+				Resource: "metricpipelines",
+			},
+			{
+				Group:    "telemetry.kyma-project.io",
+				Version:  "v1alpha1",
+				Resource: "tracepipelines",
+			},
+		}
+		for i, expectedResource := range expectedResources {
+			require.Equal(t, expectedResource.Group, kymaStatsReceiver.Resources[i].Group)
+			require.Equal(t, expectedResource.Version, kymaStatsReceiver.Resources[i].Version)
+			require.Equal(t, expectedResource.Resource, kymaStatsReceiver.Resources[i].Resource)
+		}
 	})
 
 	t.Run("singleton k8s cluster receiver creator", func(t *testing.T) {

internal/resources/otelcollector/rbac.go

@@ -110,6 +110,21 @@ func makeMetricGatewayClusterRole(name types.NamespacedName, kymaInputAllowed bo
 			Resources: []string{"telemetries"},
 			Verbs:     []string{"get", "list", "watch"},
 		})
+		clusterRole.Rules = append(clusterRole.Rules, rbacv1.PolicyRule{
+			APIGroups: []string{"telemetry.kyma-project.io"},
+			Resources: []string{"metricpipelines"},
+			Verbs:     []string{"get", "list", "watch"},
+		})
+		clusterRole.Rules = append(clusterRole.Rules, rbacv1.PolicyRule{
+			APIGroups: []string{"telemetry.kyma-project.io"},
+			Resources: []string{"tracepipelines"},
+			Verbs:     []string{"get", "list", "watch"},
+		})
+		clusterRole.Rules = append(clusterRole.Rules, rbacv1.PolicyRule{
+			APIGroups: []string{"telemetry.kyma-project.io"},
+			Resources: []string{"logpipelines"},
+			Verbs:     []string{"get", "list", "watch"},
+		})
 	}
 
 	k8sClusterRules := []rbacv1.PolicyRule{{

internal/resources/otelcollector/rbac_test.go

@@ -198,6 +198,21 @@ func TestMakeMetricGatewayRBACWithKymaInputAllowed(t *testing.T) {
 				Resources: []string{"telemetries"},
 				Verbs:     []string{"get", "list", "watch"},
 			},
+			{
+				APIGroups: []string{"telemetry.kyma-project.io"},
+				Resources: []string{"metricpipelines"},
+				Verbs:     []string{"get", "list", "watch"},
+			},
+			{
+				APIGroups: []string{"telemetry.kyma-project.io"},
+				Resources: []string{"tracepipelines"},
+				Verbs:     []string{"get", "list", "watch"},
+			},
+			{
+				APIGroups: []string{"telemetry.kyma-project.io"},
+				Resources: []string{"logpipelines"},
+				Verbs:     []string{"get", "list", "watch"},
+			},
 			{
 				APIGroups: []string{""},
 				Resources: []string{"events", "namespaces", "namespaces/status", "nodes", "nodes/spec", "pods", "pods/status", "replicationcontrollers", "replicationcontrollers/status", "resourcequotas", "services"},

test/e2e/metrics_kyma_input_test.go

@@ -99,71 +99,48 @@ var _ = Describe(suite.ID(), Label(suite.LabelMetrics, suite.LabelExperimental),
 				defer resp.Body.Close()
 				g.Expect(err).NotTo(HaveOccurred())
 
-				// Check the "kyma.module.status.state" metric
-				g.Expect(bodyContent).To(HaveFlatMetrics(
-					ContainElement(SatisfyAll(
-						HaveName(Equal("kyma.module.status.state")),
-						HaveMetricAttributes(HaveKey("state")),
-						HaveResourceAttributes(HaveKeyWithValue("k8s.namespace.name", kitkyma.SystemNamespaceName)),
-						HaveResourceAttributes(HaveKeyWithValue("kyma.module.name", "Telemetry")),
-						HaveScopeName(Equal(metric.InstrumentationScopeKyma)),
-						HaveScopeVersion(SatisfyAny(
-							Equal("main"),
-							MatchRegexp("[0-9]+.[0-9]+.[0-9]+"),
-						)),
-					)),
-				))
+				// Check the "kyma.resource.status.state" metric
+				checkTelemetryModuleMetricState(g, bodyContent)
 
-				// Check the "kyma.module.status.conditions" metric for the "LogComponentsHealthy" condition type
-				g.Expect(bodyContent).To(HaveFlatMetrics(
-					ContainElement(SatisfyAll(
-						HaveName(Equal("kyma.module.status.conditions")),
-						HaveMetricAttributes(HaveKeyWithValue("type", "LogComponentsHealthy")),
-						HaveMetricAttributes(HaveKey("status")),
-						HaveMetricAttributes(HaveKey("reason")),
-						HaveResourceAttributes(HaveKeyWithValue("k8s.namespace.name", kitkyma.SystemNamespaceName)),
-						HaveResourceAttributes(HaveKeyWithValue("kyma.module.name", "Telemetry")),
-						HaveScopeName(Equal(metric.InstrumentationScopeKyma)),
-						HaveScopeVersion(SatisfyAny(
-							Equal("main"),
-							MatchRegexp("[0-9]+.[0-9]+.[0-9]+"),
-						)),
-					)),
-				))
+				// Check the "kyma.resource.status.conditions" metric for the "LogComponentsHealthy" condition type
+				checkTelemtryModuleMetricsConditions(g, bodyContent, "LogComponentsHealthy")
 
-				// Check the "kyma.module.status.conditions" metric for the "MetricComponentsHealthy" condition type
-				g.Expect(bodyContent).To(HaveFlatMetrics(
-					ContainElement(SatisfyAll(
-						HaveName(Equal("kyma.module.status.conditions")),
-						HaveMetricAttributes(HaveKeyWithValue("type", "MetricComponentsHealthy")),
-						HaveMetricAttributes(HaveKey("status")),
-						HaveMetricAttributes(HaveKey("reason")),
-						HaveResourceAttributes(HaveKeyWithValue("k8s.namespace.name", kitkyma.SystemNamespaceName)),
-						HaveResourceAttributes(HaveKeyWithValue("kyma.module.name", "Telemetry")),
-						HaveScopeName(Equal(metric.InstrumentationScopeKyma)),
-						HaveScopeVersion(SatisfyAny(
-							Equal("main"),
-							MatchRegexp("[0-9]+.[0-9]+.[0-9]+"),
-						)),
-					)),
-				))
+				// Check the "kyma.resource.status.conditions" metric for the "MetricComponentsHealthy" condition type
+				checkTelemtryModuleMetricsConditions(g, bodyContent, "MetricComponentsHealthy")
+
+				// Check the "kyma.resource.status.conditions" metric for the "TraceComponentsHealthy" condition type
+				checkTelemtryModuleMetricsConditions(g, bodyContent, "TraceComponentsHealthy")
+
+			}, periodic.TelemetryEventuallyTimeout, periodic.TelemetryInterval).Should(Succeed())
+		})
+
+		It("Ensures metric pipeline condition metrics from both pipelines are sent to the backend which is receiving metrics from the pipeline with annotation", func() {
+			Eventually(func(g Gomega) {
+				resp, err := proxyClient.Get(backendForKymaInputExportURL)
+				g.Expect(err).NotTo(HaveOccurred())
+				g.Expect(resp).To(HaveHTTPStatus(http.StatusOK))
+				bodyContent, err := io.ReadAll(resp.Body)
+				defer resp.Body.Close()
+				g.Expect(err).NotTo(HaveOccurred())
+
+				// Check the "kyma.resource.status.conditions" type ConfigurationGenerated for  metricpipeline with annotation
+				CheckMetricPipelineMetricsConditions(g, bodyContent, "ConfigurationGenerated", pipelineWithAnnotationName)
+
+				// Check the "kyma.resource.status.conditions" type AgentHealthy for metricpipeline with annotation
+				CheckMetricPipelineMetricsConditions(g, bodyContent, "AgentHealthy", pipelineWithAnnotationName)
+
+				// Check the "kyma.resource.status.conditions" type GatewayHealthy for metricpipeline with annotation
+				CheckMetricPipelineMetricsConditions(g, bodyContent, "GatewayHealthy", pipelineWithAnnotationName)
+
+				// Check the "kyma.resource.status.conditions" type ConfigurationGenerated for  metricpipeline with annotation
+				CheckMetricPipelineMetricsConditions(g, bodyContent, "ConfigurationGenerated", pipelineWithoutAnnotationName)
+
+				// Check the "kyma.resource.status.conditions" type AgentHealthy for metricpipeline with annotation
+				CheckMetricPipelineMetricsConditions(g, bodyContent, "AgentHealthy", pipelineWithoutAnnotationName)
+
+				// Check the "kyma.resource.status.conditions" type GatewayHealthy for metricpipeline with annotation
+				CheckMetricPipelineMetricsConditions(g, bodyContent, "GatewayHealthy", pipelineWithoutAnnotationName)
 
-				// Check the "kyma.module.status.conditions" metric for the "TraceComponentsHealthy" condition type
-				g.Expect(bodyContent).To(HaveFlatMetrics(
-					ContainElement(SatisfyAll(
-						HaveName(Equal("kyma.module.status.conditions")),
-						HaveMetricAttributes(HaveKeyWithValue("type", "TraceComponentsHealthy")),
-						HaveMetricAttributes(HaveKey("status")),
-						HaveMetricAttributes(HaveKey("reason")),
-						HaveResourceAttributes(HaveKeyWithValue("k8s.namespace.name", kitkyma.SystemNamespaceName)),
-						HaveResourceAttributes(HaveKeyWithValue("kyma.module.name", "Telemetry")),
-						HaveScopeName(Equal(metric.InstrumentationScopeKyma)),
-						HaveScopeVersion(SatisfyAny(
-							Equal("main"),
-							MatchRegexp("[0-9]+.[0-9]+.[0-9]+"),
-						)),
-					)),
-				))
 			}, periodic.TelemetryEventuallyTimeout, periodic.TelemetryInterval).Should(Succeed())
 		})
 
@@ -174,11 +151,71 @@ var _ = Describe(suite.ID(), Label(suite.LabelMetrics, suite.LabelExperimental),
 				g.Expect(resp).To(HaveHTTPStatus(http.StatusOK))
 				g.Expect(resp).To(HaveHTTPBody(
 					HaveFlatMetrics(SatisfyAll(
-						Not(ContainElement(HaveName(Equal("kyma.module.status.state")))),
-						Not(ContainElement(HaveName(Equal("kyma.module.status.conditions")))),
+						Not(ContainElement(HaveName(Equal("kyma.resource.status.state")))),
+						Not(ContainElement(HaveName(Equal("kyma.resource.status.conditions")))),
 					)),
 				))
-			}, periodic.TelemetryConsistentlyTimeout, periodic.TelemetryInterval).Should(Succeed())
+			}, periodic.TelemetryConsistentlyScrapeTimeout, periodic.TelemetryInterval).Should(Succeed())
 		})
 	})
 })
+
+func checkTelemetryModuleMetricState(g Gomega, body []byte) {
+	g.Expect(body).To(HaveFlatMetrics(
+		ContainElement(SatisfyAll(
+			HaveName(Equal("kyma.resource.status.state")),
+			HaveMetricAttributes(HaveKey("state")),
+			HaveResourceAttributes(HaveKeyWithValue("k8s.namespace.name", kitkyma.SystemNamespaceName)),
+			HaveResourceAttributes(HaveKeyWithValue("k8s.resource.name", "default")),
+			HaveResourceAttributes(HaveKeyWithValue("k8s.resource.group", "operator.kyma-project.io")),
+			HaveResourceAttributes(HaveKeyWithValue("k8s.resource.version", "v1alpha1")),
+			HaveResourceAttributes(HaveKeyWithValue("k8s.resource.kind", "telemetries")),
+			HaveScopeName(Equal(metric.InstrumentationScopeKyma)),
+			HaveScopeVersion(SatisfyAny(
+				Equal("main"),
+				MatchRegexp("[0-9]+.[0-9]+.[0-9]+"),
+			)),
+		)),
+	))
+}
+
+func checkTelemtryModuleMetricsConditions(g Gomega, body []byte, typeName string) {
+	g.Expect(body).To(HaveFlatMetrics(
+		ContainElement(SatisfyAll(
+			HaveName(Equal("kyma.resource.status.conditions")),
+			HaveMetricAttributes(HaveKeyWithValue("type", typeName)),
+			HaveMetricAttributes(HaveKey("status")),
+			HaveMetricAttributes(HaveKey("reason")),
+			HaveResourceAttributes(HaveKeyWithValue("k8s.namespace.name", kitkyma.SystemNamespaceName)),
+			HaveResourceAttributes(HaveKeyWithValue("k8s.resource.name", "default")),
+			HaveResourceAttributes(HaveKeyWithValue("k8s.resource.group", "operator.kyma-project.io")),
+			HaveResourceAttributes(HaveKeyWithValue("k8s.resource.version", "v1alpha1")),
+			HaveResourceAttributes(HaveKeyWithValue("k8s.resource.kind", "telemetries")),
+			HaveScopeName(Equal(metric.InstrumentationScopeKyma)),
+			HaveScopeVersion(SatisfyAny(
+				Equal("main"),
+				MatchRegexp("[0-9]+.[0-9]+.[0-9]+"),
+			)),
+		)),
+	))
+}
+
+func CheckMetricPipelineMetricsConditions(g Gomega, body []byte, typeName, pipelineName string) {
+	g.Expect(body).To(HaveFlatMetrics(
+		ContainElement(SatisfyAll(
+			HaveName(Equal("kyma.resource.status.conditions")),
+			HaveMetricAttributes(HaveKeyWithValue("type", typeName)),
+			HaveMetricAttributes(HaveKey("status")),
+			HaveMetricAttributes(HaveKey("reason")),
+			HaveResourceAttributes(HaveKeyWithValue("k8s.resource.name", pipelineName)),
+			HaveResourceAttributes(HaveKeyWithValue("k8s.resource.group", "telemetry.kyma-project.io")),
+			HaveResourceAttributes(HaveKeyWithValue("k8s.resource.version", "v1alpha1")),
+			HaveResourceAttributes(HaveKeyWithValue("k8s.resource.kind", "metricpipelines")),
+			HaveScopeName(Equal(metric.InstrumentationScopeKyma)),
+			HaveScopeVersion(SatisfyAny(
+				Equal("main"),
+				MatchRegexp("[0-9]+.[0-9]+.[0-9]+"),
+			)),
+		)),
+	))
+}

test/testkit/periodic/periodic.go

@@ -19,6 +19,10 @@ const (
 	// For example, to verify that a certain signal *does not* have provided resource attributes.
 	TelemetryConsistentlyTimeout = time.Second * 20
 
+	//TelemetryConsistentlyScrapeTimeout is used to set the timeout equal to two scrape intervals.
+	//So that we can consistently check for the presence/absence of a metric.
+	TelemetryConsistentlyScrapeTimeout = time.Second * 60
+
 	// DefaultInterval is the default interval duration used when no specialized interval is applicable.
 	DefaultInterval = time.Millisecond * 250