Add-metrics-CEL-609 checkpoint

Author: StacieClark-Elastic

NOTICE.txt

@@ -249,6 +249,16 @@ require (
 	go.opentelemetry.io/collector/processor v1.41.0
 	go.opentelemetry.io/collector/processor/processorhelper v0.135.0
 	go.opentelemetry.io/collector/receiver/receivertest v0.135.0
+	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.62.0
+	go.opentelemetry.io/otel v1.38.0
+	go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc v1.37.0
+	go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp v1.37.0
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.37.0
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.37.0
+	go.opentelemetry.io/otel/exporters/stdout/stdoutmetric v1.37.0
+	go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.37.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
@@ -466,25 +476,15 @@ require (
 	go.opentelemetry.io/contrib/bridges/otelzap v0.12.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.62.0 // indirect
 	go.opentelemetry.io/contrib/otelconf v0.17.0 // indirect
 	go.opentelemetry.io/contrib/propagators/b3 v1.37.0 // indirect
 	go.opentelemetry.io/ebpf-profiler v0.0.202531 // indirect
-	go.opentelemetry.io/otel v1.38.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploggrpc v0.13.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploghttp v0.13.0 // indirect
-	go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc v1.37.0 // indirect
-	go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp v1.37.0 // indirect
-	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.37.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.37.0 // indirect
-	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.37.0 // indirect
 	go.opentelemetry.io/otel/exporters/prometheus v0.59.0 // indirect
 	go.opentelemetry.io/otel/exporters/stdout/stdoutlog v0.13.0 // indirect
-	go.opentelemetry.io/otel/exporters/stdout/stdoutmetric v1.37.0 // indirect
-	go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.37.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.13.0 // indirect
 	go.opentelemetry.io/otel/trace v1.38.0 // indirect
 	go.opentelemetry.io/proto/otlp v1.7.0 // indirect

go.mod

@@ -66,6 +66,38 @@ type config struct {
 	// RecordCoverage indicates whether a program should
 	// record and log execution coverage.
 	RecordCoverage bool `config:"record_coverage"`
+
+	Meta map[string]interface{} `config:"meta"`
+}
+
+func (c config) GetPackageName() string {
+	if c.Meta == nil {
+		return "unknown"
+	}
+	packageinfo, ok := c.Meta["package"]
+	if !ok {
+		return "unknown"
+	}
+	name, ok := packageinfo.(map[string]interface{})["name"]
+	if !ok {
+		return "unknown"
+	}
+	return name.(string)
+}
+
+func (c config) GetPackageVersion() string {
+	if c.Meta == nil {
+		return "unknown"
+	}
+	packageinfo, ok := c.Meta["package"]
+	if !ok {
+		return "unknown"
+	}
+	version, ok := packageinfo.(map[string]interface{})["version"]
+	if !ok {
+		return "unknown"
+	}
+	return version.(string)
 }
 
 type redact struct {

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

@@ -13,6 +13,9 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/sdk/resource"
+	semconv "go.opentelemetry.io/otel/semconv/v1.34.0"
 	"io"
 	"io/fs"
 	"maps"
@@ -51,6 +54,7 @@ import (
 	"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/elastic-agent-libs/logp"
 	"github.com/elastic/elastic-agent-libs/mapstr"
 	"github.com/elastic/elastic-agent-libs/monitoring"
@@ -150,7 +154,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
 	}
@@ -238,6 +242,11 @@ 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.Debug("Starting otel periodic")
+		otelMetrics.AddPeriodicRun(ctx, 1)
+		otelMetrics.StartPeriodic()
+		defer otelMetrics.EndPeriodic(ctx)
+
 		log.Info("process periodic request")
 		var (
 			budget    = *cfg.MaxExecutions
@@ -276,8 +285,10 @@ 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 otelMetrics.AddTotalDuration(ctx, time.Since(start))
 			state, err = evalWith(ctx, prg, ast, state, start, wantDump, budget-1)
 			log.Debugw("response state", logp.Namespace("cel"), "state", redactor{state: state, cfg: cfg.Redact})
 			if err != nil {
@@ -303,6 +314,7 @@ func (i input) run(env v2.Context, src *source, cursor map[string]interface{}, p
 			}
 			isDegraded = err != nil
 			metrics.celProcessingTime.Update(time.Since(start).Nanoseconds())
+			otelMetrics.AddCELDuration(ctx, time.Since(start))
 			if trace != nil {
 				log.Debugw("final transaction", "transaction.id", trace.TxID())
 			}
@@ -451,7 +463,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.AddEvents(ctx, int64(len(events)))
+			otelMetrics.AddBatch(ctx, 1)
 			// Drop events from state. If we fail during the publication,
 			// we will re-request these events.
 			delete(state, "events")
@@ -528,8 +541,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 {
@@ -542,11 +558,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.
@@ -573,6 +590,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
 }
 
@@ -784,10 +802,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() {
@@ -883,19 +901,46 @@ 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 authClient, trace, nil, nil
+	} else {
+
+		c.Transport = userAgentDecorator{
+			UserAgent: userAgent,
+			Transport: c.Transport,
+		}
 	}
 
-	return c, trace, nil
+	resource := resource.NewWithAttributes(
+		semconv.SchemaURL,
+		semconv.ServiceInstanceID(env.IDWithoutName),
+		semconv.ServiceNameKey.String(cfg.GetPackageName()),
+		semconv.ServiceVersionKey.String(cfg.GetPackageVersion()),
+		attribute.String("agent.version", env.Agent.Version),
+		attribute.String("agent.componentID", env.Agent.ComponentID),
+		attribute.String("agent.id", env.Agent.ID.String()),
+		attribute.String("agent.version", env.Agent.Version),
+		attribute.String("agent.componentID", env.Agent.ComponentID),
+	)
+
+	log.Infof("created cel input resource", resource.String())
+	exporterType := otel.GetExporterTypeFromEnv()
+	exporter, err := otel.NewExporterFactory(log).NewExporter(ctx, exporterType)
+	if err != nil {
+		log.Errorw("failed to get exporter", "error", err)
+	}
+	exportInterval, err := otel.GetCollectionPeriodFromEnvironment(ctx, cfg.Interval)
+	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.Agent.UserAgent, *resource, c.Transport, exporter, exportInterval)
+	c.Transport = otelTransport
+	return c, trace, otelMetrics, nil
 }
 
 func wantClient(cfg config) bool {
@@ -918,7 +963,6 @@ func clientOptions(u *url.URL, keepalive httpcommon.WithKeepaliveSettings, log *
 	case !ok:
 		return []httpcommon.TransportOption{
 			httpcommon.WithLogger(log),
-			httpcommon.WithAPMHTTPInstrumentation(),
 			keepalive,
 		}
 

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

@@ -0,0 +1,71 @@
+# Opentelemetry metrics
+
+The CEL input is currently able to export Open Telemetry metrics.
+The export of OTEL metrics are off by default. The control of OTEL
+exporting is through environment variables.
+
+## Configuration
+OTEL metrics can be sent to an otlp endpoint or to console for testing.
+OTEL metrics endpoints [Open Telemetry Collector](https://www.elastic.co/docs/reference/opentelemetry), the
+[Elastic Cloud Mangaged OTLP Endpoint](https://www.elastic.co/docs/reference/opentelemetry/motlp) in Elastic Cloud
+or the [otlp endpoint in Elastic APM](https://www.elastic.co/docs/solutions/observability/apm/use-opentelemetry-with-apm).
+
+To export OTEL metrics to console set these environment variables.
+```
+unset OTEL_EXPORTER_OTLP_ENDPOINT
+unset OTEL_EXPORTER_OTLP_HEADERS
+export OTEL_METRICS_EXPORTER=console
+```
+To export OTEL metrics to an OTLP endpoint set these environment variables,
+```
+export OTEL_EXPORTER_OTLP_ENDPOINT=<endpoint URL>
+export OTEL_METRICS_EXPORTER=otlp
+
+Authorization headers will depend upon the endpoint being used.
+export OTEL_EXPORTER_OTLP_HEADERS="Authorization=ApiKey <key>"
+```
+
+The console exports in JSON. The default protocol for OTLP is GRPC.
+Filebeat also supports "http/protobuf". It does not support "http/json."
+To use an http/protobuf protocol:
+
+```
+export OTEL_EXPORTER_OTLP_METRICS_PROTOCOL="http/protobuf"
+```
+## Exported metrics
+
+Each CEL input has an associated Open Telemetry Resource associate with it
+
+|name|description|
+|----|---|
+|resource.attributes.service.name | the package name of the integration|
+|resource.attributes.service.version | version of the integration|
+|resource.attributes.agent.id  | id of the agent|
+|resource.attributes.instance.id ||
+
+Exported metrics:
+A program run is single run of the cel program. A periodic run is all the
+program runs for that periodic run.
+
+|name|description|metric type|
+|---|---|---|
+|input.cel.periodic.run.count              | the number of times a periodic run was started.|Int64Counter|
+| input.cel.periodic.program.started       | a histogram of times a program was started in a periodic run.|Int64Histogram|
+| input.cel.periodic.program.success       | a histogram of times a program terminated without an error in a periodic run.|Int64Histogram|
+| input.cel.periodic.batch.generated       | a histogram of the number of batches generated in a periodic run.|Int64Histogram|
+| input.cel.periodic.batch.published       | a histogram of the number of batches successfully published in a periodic run.|Int64Histogram|
+| input.cel.periodic.event.generated       | a histogram of the number of events generated in a periodic run.|Int64Histogram|
+| input.cel.periodic.event.published       | a histogram of the number of events published in a periodic run.|Int64Histogram|
+| input.cel.periodic.run.duration          | a histogram of the total duration of time in seconds spent in a periodic run.|Int64Histogram|
+| input.cel.periodic.cel.duration          | a histogram of the total duration of time in seconds spent processing CEL programs in a periodic run.|Int64Histogram|
+|input.cel.periodic.event.publish.duration | a histogram of the total duration of time in seconds publishing events in a periodic run.|Int64Histogram|
+|input.cel.program.run.started.count      | a count of the number of times a program was started.|Int64Counter|
+| input.cel.program.run.success.count      | a count of the number of times a program terminated without error.|Int64Counter|
+| input.cel.program.batch.count            | a count of the number of batches the program has generated.|Int64Counter|
+| input.cel.program.event.count            | a count of the number of events the program has generated.|Int64Counter|
+| input.cel.program.event.published.count  | a count of the number of events the program has published.|Int64Counter|
+| input.cel.program.batch.published.count  | a count of the number of batched the program has published.|Int64Counter|
+| input.cel.program.run.duration           | a histogram of the total time in seconds spent executing the program.|Int64Histogram|
+| input.cel.program.cel.duration           | a histogram of the time in seconds spent processing the CEL program.|Int64Histogram|
+| input.cel.program.publish.duration       | a histogram of the time in seconds spent publishing in the program.|Int64Histogram|
+