Add OTEL metrics to cel input

changelog/fragments/1764199697-add-cel-input-otel-metrics.yaml

@@ -0,0 +1,45 @@
+# REQUIRED
+# Kind can be one of:
+# - breaking-change: a change to previously-documented behavior
+# - deprecation: functionality that is being removed in a later release
+# - bug-fix: fixes a problem in a previous version
+# - enhancement: extends functionality but does not break or fix existing behavior
+# - feature: new functionality
+# - known-issue: problems that we are aware of in a given version
+# - security: impacts on the security of a product or a user’s deployment.
+# - upgrade: important information for someone upgrading from a prior version
+# - other: does not fit into any of the other categories
+kind: feature
+
+# REQUIRED for all kinds
+# Change summary; a 80ish characters long description of the change.
+summary: add OTEL metrics to CEL inputs
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# Long description; in case the summary is not enough to describe the change
+# this field accommodate a description without length limits.
+# description:
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# impact:
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# action:
+
+# REQUIRED for all kinds
+# Affected component; usually one of "elastic-agent", "fleet-server", "filebeat", "metricbeat", "auditbeat", "all", etc.
+component: filebeat
+
+# AUTOMATED
+# OPTIONAL to manually add other PR URLs
+# PR URL: A link the PR that added the changeset.
+# If not present is automatically filled by the tooling finding the PR where this changelog fragment has been added.
+# NOTE: the tooling supports backports, so it's able to fill the original PR number instead of the backport PR number.
+# Please provide it if you are adding a fragment for a different PR.
+# pr: https://github.com/owner/repo/1234
+
+# AUTOMATED
+# OPTIONAL to manually add other issue URLs
+# Issue URL; optional; the GitHub issue related to this changeset (either closes or is part of).
+# If not present is automatically filled by the tooling with the issue linked to the PR number.
+# issue: https://github.com/owner/repo/1234

go.mod

@@ -253,6 +253,13 @@ require (
 	go.opentelemetry.io/collector/processor v1.45.0
 	go.opentelemetry.io/collector/processor/processorhelper v0.139.0
 	go.opentelemetry.io/collector/receiver/receivertest v0.139.0
+	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.63.0
+	go.opentelemetry.io/otel v1.38.0
+	go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc v1.38.0
+	go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp v1.38.0
+	go.opentelemetry.io/otel/exporters/stdout/stdoutmetric v1.38.0
+	go.opentelemetry.io/otel/metric v1.38.0
+	go.opentelemetry.io/otel/sdk v1.38.0
 	go.opentelemetry.io/otel/sdk/metric v1.38.0
 	go.uber.org/goleak v1.3.0
 	sigs.k8s.io/kind v0.29.0
@@ -468,25 +475,18 @@ require (
 	go.opentelemetry.io/contrib/bridges/otelzap v0.13.0 // indirect
 	go.opentelemetry.io/contrib/detectors/gcp v1.36.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.61.0 // indirect
-	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.63.0 // indirect
 	go.opentelemetry.io/contrib/otelconf v0.18.0 // indirect
 	go.opentelemetry.io/contrib/propagators/b3 v1.38.0 // indirect
 	go.opentelemetry.io/ebpf-profiler v0.0.202540 // indirect
-	go.opentelemetry.io/otel v1.38.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploggrpc v0.14.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploghttp v0.14.0 // indirect
-	go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc v1.38.0 // indirect
-	go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp v1.38.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.38.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.38.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.38.0 // indirect
 	go.opentelemetry.io/otel/exporters/prometheus v0.60.0 // indirect
 	go.opentelemetry.io/otel/exporters/stdout/stdoutlog v0.14.0 // indirect
-	go.opentelemetry.io/otel/exporters/stdout/stdoutmetric v1.38.0 // indirect
 	go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.38.0 // indirect
 	go.opentelemetry.io/otel/log v0.14.0 // indirect
-	go.opentelemetry.io/otel/metric v1.38.0 // indirect
-	go.opentelemetry.io/otel/sdk v1.38.0 // indirect
 	go.opentelemetry.io/otel/sdk/log v0.14.0 // indirect
 	go.opentelemetry.io/otel/trace v1.38.0 // indirect
 	go.opentelemetry.io/proto/otlp v1.7.1 // indirect

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

@@ -70,6 +70,19 @@ type config struct {
 	// RecordCoverage indicates whether a program should
 	// record and log execution coverage.
 	RecordCoverage bool `config:"record_coverage"`
+
+	Package map[string]string `config:"package"`
+}
+
+func (c config) GetPackageData(key string) string {
+	if c.Package == nil {
+		return "unknown"
+	}
+	value, ok := c.Package[key]
+	if !ok {
+		return "unknown"
+	}
+	return value
 }
 
 type redact struct {

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

@@ -28,6 +28,10 @@ import (
 	"strings"
 	"time"
 
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/sdk/resource"
+	semconv "go.opentelemetry.io/otel/semconv/v1.34.0"
+
 	"github.com/hashicorp/go-retryablehttp"
 	"github.com/icholy/digest"
 	"github.com/rcrowley/go-metrics"
@@ -42,16 +46,6 @@ import (
 	"github.com/google/cel-go/ext"
 	"google.golang.org/protobuf/types/known/structpb"
 
-	v2 "github.com/elastic/beats/v7/filebeat/input/v2"
-	inputcursor "github.com/elastic/beats/v7/filebeat/input/v2/input-cursor"
-	"github.com/elastic/beats/v7/libbeat/beat"
-	"github.com/elastic/beats/v7/libbeat/feature"
-	"github.com/elastic/beats/v7/libbeat/management/status"
-	"github.com/elastic/beats/v7/libbeat/statestore"
-	"github.com/elastic/beats/v7/libbeat/version"
-	"github.com/elastic/beats/v7/x-pack/filebeat/input/internal/httplog"
-	"github.com/elastic/beats/v7/x-pack/filebeat/input/internal/httpmon"
-	"github.com/elastic/beats/v7/x-pack/libbeat/common/aws"
 	"github.com/elastic/elastic-agent-libs/logp"
 	"github.com/elastic/elastic-agent-libs/mapstr"
 	"github.com/elastic/elastic-agent-libs/monitoring"
@@ -62,6 +56,18 @@ import (
 	"github.com/elastic/go-concert/ctxtool"
 	"github.com/elastic/go-concert/timed"
 	"github.com/elastic/mito/lib"
+
+	v2 "github.com/elastic/beats/v7/filebeat/input/v2"
+	inputcursor "github.com/elastic/beats/v7/filebeat/input/v2/input-cursor"
+	"github.com/elastic/beats/v7/libbeat/beat"
+	"github.com/elastic/beats/v7/libbeat/feature"
+	"github.com/elastic/beats/v7/libbeat/management/status"
+	"github.com/elastic/beats/v7/libbeat/statestore"
+	"github.com/elastic/beats/v7/libbeat/version"
+	"github.com/elastic/beats/v7/x-pack/filebeat/input/internal/httplog"
+	"github.com/elastic/beats/v7/x-pack/filebeat/input/internal/httpmon"
+	"github.com/elastic/beats/v7/x-pack/filebeat/otel"
+	"github.com/elastic/beats/v7/x-pack/libbeat/common/aws"
 )
 
 const (
@@ -100,7 +106,12 @@ func (i input) now() time.Time {
 func (input) Name() string { return inputName }
 
 func (input) Test(src inputcursor.Source, _ v2.TestContext) error {
-	cfg := src.(*source).cfg
+	srcp, ok := src.(*source)
+	if !ok {
+		return fmt.Errorf("input type %T is not a source", src)
+	}
+
+	cfg := srcp.cfg
 	if !wantClient(cfg) {
 		return nil
 	}
@@ -110,7 +121,11 @@ func (input) Test(src inputcursor.Source, _ v2.TestContext) error {
 // Run starts the input and blocks until it ends completes. It will return on
 // context cancellation or type invalidity errors, any other error will be retried.
 func (input) Run(env v2.Context, src inputcursor.Source, crsr inputcursor.Cursor, pub inputcursor.Publisher) error {
-	dataStreamName := src.(*source).cfg.DataStream // May be empty.
+	srcP, ok := src.(*source)
+	if !ok {
+		return errors.New("inputcursor.Source is not a *source type")
+	}
+	dataStreamName := srcP.cfg.DataStream // May be empty.
 
 	var cursor map[string]interface{}
 	env.UpdateStatus(status.Starting, dataStreamName)
@@ -128,7 +143,7 @@ func (input) Run(env v2.Context, src inputcursor.Source, crsr inputcursor.Cursor
 			parent: &env,
 		}
 	}
-	err := input{}.run(env, src.(*source), cursor, pub, health)
+	err := input{}.run(env, srcP, cursor, pub, health)
 	if err != nil {
 		msg := "failed to run: " + err.Error()
 		if dataStreamName != "" {
@@ -179,7 +194,7 @@ func (i input) run(env v2.Context, src *source, cursor map[string]interface{}, p
 		cfg.Resource.Tracer.Filename = strings.ReplaceAll(cfg.Resource.Tracer.Filename, "*", id)
 	}
 
-	client, trace, err := newClient(ctx, cfg, log, reg)
+	client, trace, otelMetrics, err := newClient(ctx, cfg, log, reg, env)
 	if err != nil {
 		return err
 	}
@@ -268,7 +283,12 @@ func (i input) run(env v2.Context, src *source, cursor map[string]interface{}, p
 	// from mito/lib, a global, useragent, is available to use
 	// in requests.
 	err = periodically(ctx, cfg.Interval, func() error {
-		log.Info("process periodic request")
+		log.Debug("Starting otel periodic")
+		otelMetrics.AddPeriodicRun(ctx, 1)
+		otelMetrics.StartPeriodic()
+		defer otelMetrics.EndPeriodic(ctx)
+
+		log.Debug("process periodic request")
 		var (
 			budget    = *cfg.MaxExecutions
 			waitUntil time.Time
@@ -306,9 +326,14 @@ func (i input) run(env v2.Context, src *source, cursor map[string]interface{}, p
 				log.Debugw("previous transaction", "transaction.id", trace.TxID())
 			}
 			log.Debugw("request state", logp.Namespace("cel"), "state", redactor{state: state, cfg: cfg.Redact})
+			otelMetrics.AddProgramExecution(ctx, 1)
 			metrics.executions.Add(1)
 			start := i.now().In(time.UTC)
+			defer func() {
+				otelMetrics.AddTotalDuration(ctx, time.Since(start))
+			}()
 			state, err = evalWith(ctx, prg, ast, state, start, wantDump, budget-1)
+			otelMetrics.AddCELDuration(ctx, time.Since(start))
 			log.Debugw("response state", logp.Namespace("cel"), "state", redactor{state: state, cfg: cfg.Redact})
 			if err != nil {
 				var dump dumpError
@@ -443,11 +468,13 @@ func (i input) run(env v2.Context, src *source, cursor map[string]interface{}, p
 			switch e := e.(type) {
 			case []interface{}:
 				if len(e) == 0 {
+					otelMetrics.AddProgramSuccessExecution(ctx, 1)
 					return nil
 				}
 				events = e
 			case map[string]interface{}:
 				if e == nil {
+					otelMetrics.AddProgramSuccessExecution(ctx, 1)
 					return nil
 				}
 				if _, ok := e["error"]; ok {
@@ -481,7 +508,8 @@ func (i input) run(env v2.Context, src *source, cursor map[string]interface{}, p
 			// We have a non-empty batch of events to process.
 			metrics.batchesReceived.Add(1)
 			metrics.eventsReceived.Add(uint64(len(events)))
-
+			otelMetrics.AddGeneratedBatch(ctx, 1)
+			otelMetrics.AddEvents(ctx, int64(len(events)))
 			// Drop events from state. If we fail during the publication,
 			// we will re-request these events.
 			delete(state, "events")
@@ -558,8 +586,11 @@ func (i input) run(env v2.Context, src *source, cursor map[string]interface{}, p
 				}
 				if i == 0 {
 					metrics.batchesPublished.Add(1)
+					otelMetrics.AddPublishedBatch(ctx, 1)
+
 				}
 				metrics.eventsPublished.Add(1)
+				otelMetrics.AddPublishedEvents(ctx, 1)
 
 				err = ctx.Err()
 				if err != nil {
@@ -572,11 +603,12 @@ func (i input) run(env v2.Context, src *source, cursor map[string]interface{}, p
 			}
 
 			metrics.batchProcessingTime.Update(time.Since(start).Nanoseconds())
-
+			otelMetrics.AddPublishDuration(ctx, time.Since(start))
 			// Advance the cursor to the final state if there was no error during
 			// publications. This is needed to transition to the next set of events.
 			if !hadPublicationError {
 				goodCursor = cursor
+				otelMetrics.AddProgramSuccessExecution(ctx, 1)
 			}
 
 			// Replace the last known good cursor.
@@ -603,6 +635,7 @@ func (i input) run(env v2.Context, src *source, cursor map[string]interface{}, p
 		log.Infof("input stopped because context was cancelled with: %v", err)
 		err = nil
 	}
+	otelMetrics.Shutdown(ctx)
 	return err
 }
 
@@ -827,10 +860,10 @@ func getLimit(which string, rateLimit map[string]interface{}, log *logp.Logger)
 // https://github.com/natefinch/lumberjack/blob/4cb27fcfbb0f35cb48c542c5ea80b7c1d18933d0/lumberjack.go#L39
 const lumberjackTimestamp = "[0-9][0-9][0-9][0-9]-[0-9][0-9]-[0-9][0-9]T[0-9][0-9]-[0-9][0-9]-[0-9][0-9].[0-9][0-9][0-9]"
 
-func newClient(ctx context.Context, cfg config, log *logp.Logger, reg *monitoring.Registry) (*http.Client, *httplog.LoggingRoundTripper, error) {
+func newClient(ctx context.Context, cfg config, log *logp.Logger, reg *monitoring.Registry, env v2.Context) (*http.Client, *httplog.LoggingRoundTripper, *otel.OTELCELMetrics, error) {
 	c, err := cfg.Resource.Transport.Client(clientOptions(cfg.Resource.URL.URL, cfg.Resource.KeepAlive.settings(), log)...)
 	if err != nil {
-		return nil, nil, err
+		return nil, nil, nil, err
 	}
 
 	if cfg.Auth.Digest.isEnabled() {
@@ -850,13 +883,13 @@ func newClient(ctx context.Context, cfg config, log *logp.Logger, reg *monitorin
 		tr, err := aws.InitializeSignerTransport(*cfg.Auth.AWS, log, c.Transport)
 		if err != nil {
 			log.Errorw("failed to initialize aws config failed for signer", "error", err)
-			return nil, nil, err
+			return nil, nil, nil, err
 		}
 		c.Transport = tr
 	} else if cfg.Auth.File.isEnabled() {
 		tr, err := newFileAuthTransport(cfg.Auth.File, c.Transport)
 		if err != nil {
-			return nil, nil, err
+			return nil, nil, nil, err
 		}
 		c.Transport = tr
 	}
@@ -941,19 +974,75 @@ func newClient(ctx context.Context, cfg config, log *logp.Logger, reg *monitorin
 	}
 
 	if cfg.Auth.OAuth2.isEnabled() {
-		authClient, err := cfg.Auth.OAuth2.client(ctx, c)
+		c, err = cfg.Auth.OAuth2.client(ctx, c)
 		if err != nil {
-			return nil, nil, err
+			return nil, nil, nil, err
 		}
-		return authClient, trace, nil
 	}
 
 	c.Transport = userAgentDecorator{
 		UserAgent: userAgent,
 		Transport: c.Transport,
 	}
 
-	return c, trace, nil
+	resource := resource.NewWithAttributes(
+		semconv.SchemaURL, GetResourceAttributes(env, cfg)...,
+	)
+
+	log.Infof("created cel input resource %s", resource.String())
+	exporter, exporterType, err := otel.GetGlobalExporterFactory(log).GetExporter(ctx)
+	if err != nil {
+		log.Errorw("failed to get exporter", "error", err)
+	}
+	if err != nil {
+		log.Errorw("failed to get collection period", "error", err)
+	}
+	log.Infof("created OTEL cel input exporter %s for input %s", exporterType, env.IDWithoutName)
+	otelMetrics, otelTransport, err := otel.NewOTELCELMetrics(log, env.IDWithoutName, *resource, c.Transport, exporter)
+	if err != nil {
+		return nil, nil, nil, err
+	}
+	c.Transport = otelTransport
+	return c, trace, otelMetrics, nil
+}
+
+func GetResourceAttributes(env v2.Context, cfg config) []attribute.KeyValue {
+	attrs := []attribute.KeyValue{semconv.ServiceInstanceID(env.IDWithoutName),
+		attribute.String("package.name", cfg.GetPackageData("name")),
+		attribute.String("package.version", cfg.GetPackageData("version")),
+		attribute.String("package.data_stream", cfg.DataStream),
+		attribute.String("agent.version", env.Agent.Version),
+		attribute.String("agent.id", env.Agent.ID.String())}
+
+	usedKeys := make(map[string]struct{})
+
+	for _, attr := range attrs {
+		// Access the Key field of the KeyValue struct
+		usedKeys[string(attr.Key)] = struct{}{}
+	}
+	attributesStr, ok := os.LookupEnv("OTEL_RESOURCE_ATTRIBUTES")
+	if ok && len(attributesStr) > 0 {
+		attributes := make([]attribute.KeyValue, 0)
+		pairs := strings.Split(attributesStr, ",")
+		for _, pair := range pairs {
+			kv := strings.SplitN(pair, "=", 2)
+			if len(kv) == 2 {
+				key := strings.TrimSpace(kv[0])
+				value := strings.TrimSpace(kv[1])
+				if key != "" {
+					// don't overwrite existing keys
+					_, used := usedKeys[key]
+					if !used {
+						attributes = append(attributes, attribute.String(key, value))
+					}
+				}
+			}
+		}
+		attrs = append(attrs, attributes...)
+	}
+
+	return attrs
+
 }
 
 func wantClient(cfg config) bool {