feat: Allow collecting Telemetry module status metrics in development variant

.github/workflows/pr-integration.yml

@@ -37,7 +37,7 @@ jobs:
           github-token: ${{ secrets.GITHUB_TOKEN }}
 
       - name: Run tests
-        run: bin/ginkgo run --tags e2e --label-filter="${{ matrix.ginkgo-labels }} && !v1beta1" test/e2e
+        run: bin/ginkgo run --tags e2e --label-filter="${{ matrix.ginkgo-labels }} && !experimental" test/e2e
 
       - name: Finalize test
         uses: "./.github/template/finalize-test"
@@ -64,7 +64,7 @@ jobs:
           github-token: ${{ secrets.GITHUB_TOKEN }}
 
       - name: Run tests
-        run: bin/ginkgo run --tags e2e --label-filter="${{ matrix.ginkgo-labels }} && v1beta1" test/e2e
+        run: bin/ginkgo run --tags e2e --label-filter="${{ matrix.ginkgo-labels }} && experimental" test/e2e
 
       - name: Finalize test
         uses: "./.github/template/finalize-test"

controllers/telemetry/metricpipeline_controller.go

@@ -64,7 +64,6 @@ type MetricPipelineControllerConfig struct {
 
 	SelfMonitorName    string
 	TelemetryNamespace string
-	KymaInputAllowed   bool
 }
 
 func NewMetricPipelineController(client client.Client, reconcileTriggerChan <-chan event.GenericEvent, config MetricPipelineControllerConfig) (*MetricPipelineController, error) {

internal/otelcollector/config/metric/agent/config.go

@@ -4,6 +4,7 @@ import (
 	"time"
 
 	"github.com/kyma-project/telemetry-manager/internal/otelcollector/config"
+	"github.com/kyma-project/telemetry-manager/internal/otelcollector/config/metric"
 )
 
 type Config struct {
@@ -115,11 +116,11 @@ const (
 type Processors struct {
 	config.BaseProcessors `yaml:",inline"`
 
-	DeleteServiceName                 *config.ResourceProcessor `yaml:"resource/delete-service-name,omitempty"`
-	DropInternalCommunication         *FilterProcessor          `yaml:"filter/drop-internal-communication,omitempty"`
-	SetInstrumentationScopeRuntime    *TransformProcessor       `yaml:"transform/set-instrumentation-scope-runtime,omitempty"`
-	SetInstrumentationScopePrometheus *TransformProcessor       `yaml:"transform/set-instrumentation-scope-prometheus,omitempty"`
-	SetInstrumentationScopeIstio      *TransformProcessor       `yaml:"transform/set-instrumentation-scope-istio,omitempty"`
+	DeleteServiceName                 *config.ResourceProcessor  `yaml:"resource/delete-service-name,omitempty"`
+	DropInternalCommunication         *FilterProcessor           `yaml:"filter/drop-internal-communication,omitempty"`
+	SetInstrumentationScopeRuntime    *metric.TransformProcessor `yaml:"transform/set-instrumentation-scope-runtime,omitempty"`
+	SetInstrumentationScopePrometheus *metric.TransformProcessor `yaml:"transform/set-instrumentation-scope-prometheus,omitempty"`
+	SetInstrumentationScopeIstio      *metric.TransformProcessor `yaml:"transform/set-instrumentation-scope-istio,omitempty"`
 }
 
 type Exporters struct {
@@ -133,8 +134,3 @@ type FilterProcessor struct {
 type FilterProcessorMetrics struct {
 	Metric []string `yaml:"metric,omitempty"`
 }
-
-type TransformProcessor struct {
-	ErrorMode        string                                `yaml:"error_mode"`
-	MetricStatements []config.TransformProcessorStatements `yaml:"metric_statements"`
-}

internal/otelcollector/config/metric/agent/config_builder.go

@@ -43,7 +43,7 @@ func (b *Builder) Build(pipelines []telemetryv1alpha1.MetricPipeline, opts Build
 			Extensions: config.DefaultExtensions(),
 		},
 		Receivers:  makeReceiversConfig(inputs, opts),
-		Processors: makeProcessorsConfig(inputs, opts),
+		Processors: makeProcessorsConfig(inputs, opts.InstrumentationScopeVersion),
 		Exporters:  makeExportersConfig(b.Config.GatewayOTLPServiceName),
 	}
 }

internal/otelcollector/config/metric/agent/processors.go

@@ -5,7 +5,7 @@ import (
 	"github.com/kyma-project/telemetry-manager/internal/otelcollector/config/metric"
 )
 
-func makeProcessorsConfig(inputs inputSources, opts BuildOptions) Processors {
+func makeProcessorsConfig(inputs inputSources, instrumentationScopeVersion string) Processors {
 	processorsConfig := Processors{
 		BaseProcessors: config.BaseProcessors{
 			Batch:         makeBatchProcessorConfig(),
@@ -17,16 +17,16 @@ func makeProcessorsConfig(inputs inputSources, opts BuildOptions) Processors {
 		processorsConfig.DeleteServiceName = makeDeleteServiceNameConfig()
 
 		if inputs.runtime {
-			processorsConfig.SetInstrumentationScopeRuntime = makeInstrumentationScopeProcessor(metric.InputSourceRuntime, opts)
+			processorsConfig.SetInstrumentationScopeRuntime = metric.MakeInstrumentationScopeProcessor(metric.InputSourceRuntime, instrumentationScopeVersion)
 		}
 
 		if inputs.prometheus {
-			processorsConfig.SetInstrumentationScopePrometheus = makeInstrumentationScopeProcessor(metric.InputSourcePrometheus, opts)
+			processorsConfig.SetInstrumentationScopePrometheus = metric.MakeInstrumentationScopeProcessor(metric.InputSourcePrometheus, instrumentationScopeVersion)
 		}
 
 		if inputs.istio {
 			processorsConfig.DropInternalCommunication = makeFilterToDropMetricsForTelemetryComponents()
-			processorsConfig.SetInstrumentationScopeIstio = makeInstrumentationScopeProcessor(metric.InputSourceIstio, opts)
+			processorsConfig.SetInstrumentationScopeIstio = metric.MakeInstrumentationScopeProcessor(metric.InputSourceIstio, instrumentationScopeVersion)
 		}
 	}
 

internal/otelcollector/config/metric/config.go

@@ -0,0 +1,8 @@
+package metric
+
+import "github.com/kyma-project/telemetry-manager/internal/otelcollector/config"
+
+type TransformProcessor struct {
+	ErrorMode        string                                `yaml:"error_mode"`
+	MetricStatements []config.TransformProcessorStatements `yaml:"metric_statements"`
+}

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

@@ -2,6 +2,7 @@ package gateway
 
 import (
 	"github.com/kyma-project/telemetry-manager/internal/otelcollector/config"
+	"github.com/kyma-project/telemetry-manager/internal/otelcollector/config/metric"
 )
 
 type Config struct {
@@ -13,7 +14,35 @@ type Config struct {
 }
 
 type Receivers struct {
-	OTLP config.OTLPReceiver `yaml:"otlp"`
+	OTLP                              config.OTLPReceiver                `yaml:"otlp"`
+	SingletonKymaStatsReceiverCreator *SingletonKymaStatsReceiverCreator `yaml:"singleton_receiver_creator/kymastats,omitempty"`
+}
+
+type SingletonKymaStatsReceiverCreator struct {
+	AuthType                   string                     `yaml:"auth_type"`
+	LeaderElection             LeaderElection             `yaml:"leader_election"`
+	SingletonKymaStatsReceiver SingletonKymaStatsReceiver `yaml:"receiver"`
+}
+
+type LeaderElection struct {
+	LeaseName      string `yaml:"lease_name"`
+	LeaseNamespace string `yaml:"lease_namespace"`
+}
+
+type SingletonKymaStatsReceiver struct {
+	KymaStatsReceiver KymaStatsReceiver `yaml:"kymastats"`
+}
+
+type KymaStatsReceiver struct {
+	AuthType           string      `yaml:"auth_type"`
+	CollectionInterval string      `yaml:"collection_interval"`
+	Modules            []ModuleGVR `yaml:"modules"`
+}
+
+type ModuleGVR struct {
+	Group    string `yaml:"group"`
+	Version  string `yaml:"version"`
+	Resource string `yaml:"resource"`
 }
 
 type Processors struct {
@@ -30,6 +59,7 @@ type Processors struct {
 	DropRuntimePodMetrics                        *FilterProcessor               `yaml:"filter/drop-runtime-pod-metrics,omitempty"`
 	DropRuntimeContainerMetrics                  *FilterProcessor               `yaml:"filter/drop-runtime-container-metrics,omitempty"`
 	ResolveServiceName                           *TransformProcessor            `yaml:"transform/resolve-service-name,omitempty"`
+	SetInstrumentationScopeKyma                  *metric.TransformProcessor     `yaml:"transform/set-instrumentation-scope-kyma,omitempty"`
 
 	// NamespaceFilters contains filter processors, which need different configurations per pipeline
 	NamespaceFilters NamespaceFilters `yaml:",inline,omitempty"`

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

@@ -11,14 +11,21 @@ import (
 	"github.com/kyma-project/telemetry-manager/internal/otelcollector/config"
 	"github.com/kyma-project/telemetry-manager/internal/otelcollector/config/metric"
 	"github.com/kyma-project/telemetry-manager/internal/otelcollector/config/otlpexporter"
-	"github.com/kyma-project/telemetry-manager/internal/otelcollector/ports"
 )
 
+const KymaInputAnnotation = "telemetry.kyma-project.io/experimental-kyma-input"
+
 type Builder struct {
 	Reader client.Reader
 }
 
-func (b *Builder) Build(ctx context.Context, pipelines []telemetryv1alpha1.MetricPipeline) (*Config, otlpexporter.EnvVars, error) {
+type BuildOptions struct {
+	GatewayNamespace            string
+	InstrumentationScopeVersion string
+	KymaInputAllowed            bool
+}
+
+func (b *Builder) Build(ctx context.Context, pipelines []telemetryv1alpha1.MetricPipeline, opts BuildOptions) (*Config, otlpexporter.EnvVars, error) {
 	cfg := &Config{
 		Base: config.Base{
 			Service:    config.DefaultService(make(config.Pipelines)),
@@ -45,41 +52,41 @@ func (b *Builder) Build(ctx context.Context, pipelines []telemetryv1alpha1.Metri
 			queueSize,
 			otlpexporter.SignalTypeMetric,
 		)
-		if err := declareComponentsForMetricPipeline(ctx, otlpExporterBuilder, &pipeline, cfg, envVars); err != nil {
+		if err := declareComponentsForMetricPipeline(ctx, otlpExporterBuilder, &pipeline, cfg, envVars, opts); err != nil {
 			return nil, nil, err
 		}
 
 		pipelineID := fmt.Sprintf("metrics/%s", pipeline.Name)
-		cfg.Service.Pipelines[pipelineID] = makeServicePipelineConfig(&pipeline)
+		cfg.Service.Pipelines[pipelineID] = makeServicePipelineConfig(&pipeline, opts)
 	}
 
 	return cfg, envVars, nil
 }
 
-func makeReceiversConfig() Receivers {
-	return Receivers{
-		OTLP: config.OTLPReceiver{
-			Protocols: config.ReceiverProtocols{
-				HTTP: config.Endpoint{
-					Endpoint: fmt.Sprintf("${%s}:%d", config.EnvVarCurrentPodIP, ports.OTLPHTTP),
-				},
-				GRPC: config.Endpoint{
-					Endpoint: fmt.Sprintf("${%s}:%d", config.EnvVarCurrentPodIP, ports.OTLPGRPC),
-				},
-			},
-		},
-	}
-}
-
-// declareComponentsForMetricPipeline enriches a Config (exporters, processors, etc.) with components for a given telemetryv1alpha1.MetricPipeline.
-func declareComponentsForMetricPipeline(ctx context.Context, otlpExporterBuilder *otlpexporter.ConfigBuilder, pipeline *telemetryv1alpha1.MetricPipeline, cfg *Config, envVars otlpexporter.EnvVars) error {
+// declareComponentsForMetricPipeline enriches a Config (receivers, processors, exporters etc.) with components for a given telemetryv1alpha1.MetricPipeline.
+func declareComponentsForMetricPipeline(
+	ctx context.Context,
+	otlpExporterBuilder *otlpexporter.ConfigBuilder,
+	pipeline *telemetryv1alpha1.MetricPipeline,
+	cfg *Config,
+	envVars otlpexporter.EnvVars,
+	opts BuildOptions,
+) error {
+	declareSingletonKymaStatsReceiverCreator(pipeline, cfg, opts)
 	declareDiagnosticMetricsDropFilters(pipeline, cfg)
 	declareInputSourceFilters(pipeline, cfg)
 	declareRuntimeResourcesFilters(pipeline, cfg)
 	declareNamespaceFilters(pipeline, cfg)
+	declareInstrumentationScopeTransform(pipeline, cfg, opts)
 	return declareOTLPExporter(ctx, otlpExporterBuilder, pipeline, cfg, envVars)
 }
 
+func declareSingletonKymaStatsReceiverCreator(pipeline *telemetryv1alpha1.MetricPipeline, cfg *Config, opts BuildOptions) {
+	if isKymaInputEnabled(pipeline.Annotations, opts.KymaInputAllowed) {
+		cfg.Receivers.SingletonKymaStatsReceiverCreator = makeSingletonKymaStatsReceiverCreatorConfig(opts.GatewayNamespace)
+	}
+}
+
 func declareDiagnosticMetricsDropFilters(pipeline *telemetryv1alpha1.MetricPipeline, cfg *Config) {
 	input := pipeline.Spec.Input
 
@@ -143,6 +150,12 @@ func declareNamespaceFilters(pipeline *telemetryv1alpha1.MetricPipeline, cfg *Co
 	}
 }
 
+func declareInstrumentationScopeTransform(pipeline *telemetryv1alpha1.MetricPipeline, cfg *Config, opts BuildOptions) {
+	if isKymaInputEnabled(pipeline.Annotations, opts.KymaInputAllowed) {
+		cfg.Processors.SetInstrumentationScopeKyma = metric.MakeInstrumentationScopeProcessor(metric.InputSourceKyma, opts.InstrumentationScopeVersion)
+	}
+}
+
 func declareOTLPExporter(ctx context.Context, otlpExporterBuilder *otlpexporter.ConfigBuilder, pipeline *telemetryv1alpha1.MetricPipeline, cfg *Config, envVars otlpexporter.EnvVars) error {
 	otlpExporterConfig, otlpExporterEnvVars, err := otlpExporterBuilder.MakeConfig(ctx)
 	if err != nil {
@@ -157,7 +170,7 @@ func declareOTLPExporter(ctx context.Context, otlpExporterBuilder *otlpexporter.
 	return nil
 }
 
-func makeServicePipelineConfig(pipeline *telemetryv1alpha1.MetricPipeline) config.Pipeline {
+func makeServicePipelineConfig(pipeline *telemetryv1alpha1.MetricPipeline, opts BuildOptions) config.Pipeline {
 	processors := []string{"memory_limiter", "k8sattributes"}
 
 	input := pipeline.Spec.Input
@@ -167,10 +180,14 @@ func makeServicePipelineConfig(pipeline *telemetryv1alpha1.MetricPipeline) confi
 	processors = append(processors, makeRuntimeResourcesFiltersIDs(input)...)
 	processors = append(processors, makeDiagnosticMetricFiltersIDs(input)...)
 
+	if isKymaInputEnabled(pipeline.Annotations, opts.KymaInputAllowed) {
+		processors = append(processors, "transform/set-instrumentation-scope-kyma")
+	}
+
 	processors = append(processors, "resource/insert-cluster-name", "transform/resolve-service-name", "batch")
 
 	return config.Pipeline{
-		Receivers:  []string{"otlp"},
+		Receivers:  makeReceiversIDs(pipeline.Annotations, opts.KymaInputAllowed),
 		Processors: processors,
 		Exporters:  []string{makeOTLPExporterID(pipeline)},
 	}
@@ -248,6 +265,18 @@ func formatNamespaceFilterID(pipelineName string, inputSourceType metric.InputSo
 	return fmt.Sprintf("filter/%s-filter-by-namespace-%s-input", pipelineName, inputSourceType)
 }
 
+func makeReceiversIDs(annotations map[string]string, kymaInputAllowed bool) []string {
+	var receivers []string
+
+	receivers = append(receivers, "otlp")
+
+	if isKymaInputEnabled(annotations, kymaInputAllowed) {
+		receivers = append(receivers, "singleton_receiver_creator/kymastats")
+	}
+
+	return receivers
+}
+
 func makeOTLPExporterID(pipeline *telemetryv1alpha1.MetricPipeline) string {
 	return otlpexporter.ExporterID(pipeline.Spec.Output.Otlp.Protocol, pipeline.Name)
 }
@@ -297,3 +326,7 @@ func isRuntimeContainerMetricsEnabled(input telemetryv1alpha1.MetricPipelineInpu
 
 	return !isRuntimeContainerMetricsDisabled
 }
+
+func isKymaInputEnabled(annotations map[string]string, kymaInputAllowed bool) bool {
+	return kymaInputAllowed && annotations[KymaInputAnnotation] == "true"
+}