feat(filebeat): add input_type resource attribute for CEL OTel telemetry (#50645)

* feat(filebeat): add input_type resource attribute for CEL OTel telemetry

CEL emits both custom input metrics and HTTP client metrics via
different instrumentation libraries, so they do not share a common
filter key by default. Add a built-in input_type resource attribute so
both metric families (and CEL traces) can be filtered together
consistently across backends that preserve resource attributes as
queryable labels/fields.

* Apply suggestions from code review

Co-authored-by: Dan Kortschak <dan.kortschak@elastic.co>

* add logptest import

* goimports -local github.com/elastic

---------

Co-authored-by: Dan Kortschak <dan.kortschak@elastic.co>

Author: andrewkroh

changelog/fragments/1778632723-filebeat-otel-input-type-resource-attribute.yaml

@@ -0,0 +1,3 @@
+kind: enhancement
+summary: Add an input_type resource attribute to CEL input OTel metrics for consistent filtering across custom and HTTP client metrics.
+component: filebeat

x-pack/filebeat/input/cel/cel_metrics_test.go

@@ -21,6 +21,7 @@ import (
 	"github.com/elastic/beats/v7/x-pack/filebeat/otel"
 	conf "github.com/elastic/elastic-agent-libs/config"
 	"github.com/elastic/elastic-agent-libs/logp"
+	"github.com/elastic/elastic-agent-libs/logp/logptest"
 	"github.com/elastic/elastic-agent-libs/monitoring"
 
 	"go.opentelemetry.io/otel/exporters/stdout/stdoutmetric"
@@ -220,6 +221,51 @@ func (e *inMemoryExporter) getMetrics() []metricdata.ResourceMetrics {
 	return e.metrics
 }
 
+func TestCreateOTELMetricsSetsInputTypeResourceAttribute(t *testing.T) {
+	exporter := &inMemoryExporter{}
+	otel.GetGlobalMetricsExporterFactory().SetGlobalMetricsExporter(exporter)
+	defer otel.GetGlobalMetricsExporterFactory().SetGlobalMetricsExporter(nil)
+
+	cfg := defaultConfig()
+	cfg.DataStream = "foo.bar"
+	cfg.Package = map[string]string{
+		"name":    "foo",
+		"version": "1.2.3",
+	}
+	env := v2.Context{
+		IDWithoutName: "test_input",
+		Agent: beat.Info{
+			Version: "9.0.0",
+		},
+	}
+
+	ctx := context.Background()
+	otelMetrics, _, err := createOTELMetrics(ctx, cfg, logptest.NewTestingLogger(t, "cel_metrics_test"), env, http.DefaultTransport, nil)
+	if err != nil {
+		t.Fatalf("failed to create OTEL metrics collector: %v", err)
+	}
+	defer otelMetrics.Shutdown(ctx)
+
+	otelMetrics.StartPeriodic(ctx)
+	otelMetrics.AddProgramExecutionStarted(ctx, 1)
+	otelMetrics.EndPeriodic(ctx)
+
+	deadline := time.Now().Add(5 * time.Second)
+	for len(exporter.getMetrics()) == 0 && time.Now().Before(deadline) {
+		time.Sleep(10 * time.Millisecond)
+	}
+	if got := len(exporter.getMetrics()); got == 0 {
+		t.Fatalf("expected OTEL metrics to be exported, got %d", got)
+	}
+
+	for _, rm := range exporter.getMetrics() {
+		if rm.Resource != nil && strings.Contains(rm.Resource.String(), "input_type=cel") {
+			return
+		}
+	}
+	t.Error("expected exported OTEL metrics resource attributes to include input_type=cel")
+}
+
 // testPublisher is a publisher that signals when events are published.
 type testPublisher struct {
 	mu        sync.Mutex

x-pack/filebeat/input/cel/doc.go

@@ -36,6 +36,7 @@ These Resource Attributes are included for every CEL input instance:
 	agent.version        version of agent
 	agent.id             the id of the agent
 	service.instance.id  id of the cel input instance
+	input_type           input type name emitting metrics, for example "cel"
 	package.name         name of the integration package
 	package.version      version of the integration package
 	package.data_stream  the datastream name in the integration package

x-pack/filebeat/input/cel/input.go

@@ -1319,6 +1319,7 @@ func createOTELMetrics(ctx context.Context, cfg config, log *logp.Logger, env v2
 func getResourceAttributes(env v2.Context, cfg config) []attribute.KeyValue {
 	attrs := []attribute.KeyValue{
 		semconv.ServiceInstanceID(env.IDWithoutName),
+		attribute.String("input_type", inputName),
 		attribute.String("package.name", cfg.GetPackageData("name")),
 		attribute.String("package.version", cfg.GetPackageData("version")),
 		attribute.String("package.data_stream", cfg.DataStream),

x-pack/filebeat/input/cel/input_test.go

@@ -27,6 +27,7 @@ import (
 
 	"github.com/google/go-cmp/cmp"
 	"github.com/icholy/digest"
+	"go.opentelemetry.io/otel/attribute"
 
 	v2 "github.com/elastic/beats/v7/filebeat/input/v2"
 	inputcursor "github.com/elastic/beats/v7/filebeat/input/v2/input-cursor"
@@ -3250,6 +3251,55 @@ func TestRedactor(t *testing.T) {
 	}
 }
 
+func TestGetResourceAttributesIncludesInputType(t *testing.T) {
+	env := v2.Context{IDWithoutName: "input-id"}
+	cfg := config{
+		DataStream: "foo.bar",
+		Package: map[string]string{
+			"name":    "foo",
+			"version": "1.2.3",
+		},
+	}
+
+	attrs := getResourceAttributes(env, cfg)
+	attrsMap := toResourceAttributeMap(attrs)
+
+	if got, want := attrsMap["input_type"], "cel"; got != want {
+		t.Errorf("input_type should be set from input name: got %q, want %q", got, want)
+	}
+}
+
+func TestGetResourceAttributesInputTypeCannotBeOverridden(t *testing.T) {
+	t.Setenv("OTEL_RESOURCE_ATTRIBUTES", "input_type=httpjson,deployment.environment=production")
+
+	env := v2.Context{IDWithoutName: "input-id"}
+	cfg := config{
+		DataStream: "foo.bar",
+		Package: map[string]string{
+			"name":    "foo",
+			"version": "1.2.3",
+		},
+	}
+
+	attrs := getResourceAttributes(env, cfg)
+	attrsMap := toResourceAttributeMap(attrs)
+
+	if got, want := attrsMap["input_type"], "cel"; got != want {
+		t.Errorf("built-in input_type should not be overridden from OTEL_RESOURCE_ATTRIBUTES: got %q, want %q", got, want)
+	}
+	if got, want := attrsMap["deployment.environment"], "production"; got != want {
+		t.Errorf("custom resource attributes from OTEL_RESOURCE_ATTRIBUTES should still be included: got %q, want %q", got, want)
+	}
+}
+
+func toResourceAttributeMap(attrs []attribute.KeyValue) map[string]string {
+	result := make(map[string]string, len(attrs))
+	for _, attr := range attrs {
+		result[string(attr.Key)] = attr.Value.AsString()
+	}
+	return result
+}
+
 // sameErrorOrContains reports whether got matches want: both nil, or got's
 // message contains want's message.
 func sameErrorOrContains(got, want error) bool {