x-pack/filebeat/input/cel: add http metrics collection

CHANGELOG.next.asciidoc

@@ -265,6 +265,7 @@ https:/elastic/beats/compare/v8.8.1\...main[Check the HEAD diff]
 - Ensure all responses sent by HTTP Endpoint are HTML-escaped. {pull}39329[39329]
 - Update CEL mito extensions to v1.11.0 to improve type checking. {pull}39460[39460]
 - Improve logging of request and response with request trace logging in error conditions. {pull}39455[39455]
+- Add HTTP metrics to CEL input. {issue}39501[39501] {pull}39503[39503]
 
 *Auditbeat*
 

x-pack/filebeat/docs/inputs/input-cel.asciidoc

@@ -765,15 +765,36 @@ observe the activity of the input.
 
 [options="header"]
 |=======
-| Metric | Description
-| `resource` | URL or path of the input resource.
-| `cel_executions` | Number times the CEL program has been executed.
-| `batches_received_total` | Number of event arrays received.
-| `events_received_total` | Number of events received.
-| `batches_published_total` | Number of event arrays published.
-| `events_published_total` | Number of events published.
-| `cel_processing_time` | Histogram of the elapsed successful CEL program processing times in nanoseconds.
-| `batch_processing_time` | Histogram of the elapsed successful batch processing times in nanoseconds (time of receipt to time of ACK for non-empty batches).
+| Metric | Description
+| `resource` | URL or path of the input resource.
+| `cel_executions` | Number times the CEL program has been executed.
+| `batches_received_total` | Number of event arrays received.
+| `events_received_total` | Number of events received.
+| `batches_published_total` | Number of event arrays published.
+| `events_published_total` | Number of events published.
+| `cel_processing_time` | Histogram of the elapsed successful CEL program processing times in nanoseconds.
+| `batch_processing_time` | Histogram of the elapsed successful batch processing times in nanoseconds (time of receipt to time of ACK for non-empty batches).
+| `http_request_total` | Total number of processed requests.
+| `http_request_errors_total` | Total number of request errors.
+| `http_request_delete_total` | Total number of `DELETE` requests.
+| `http_request_get_total` | Total number of `GET` requests.
+| `http_request_head_total` | Total number of `HEAD` requests.
+| `http_request_options_total` | Total number of `OPTIONS` requests.
+| `http_request_patch_total` | Total number of `PATCH` requests.
+| `http_request_post_total` | Total number of `POST` requests.
+| `http_request_put_total` | Total number of `PUT` requests.
+| `http_request_body_bytes_total` | Total of the requests body size.
+| `http_request_body_bytes` | Histogram of the requests body size.
+| `http_response_total` | Total number of responses received.
+| `http_response_errors_total` | Total number of response errors.
+| `http_response_1xx_total` | Total number of `1xx` responses.
+| `http_response_2xx_total` | Total number of `2xx` responses.
+| `http_response_3xx_total` | Total number of `3xx` responses.
+| `http_response_4xx_total` | Total number of `4xx` responses.
+| `http_response_5xx_total` | Total number of `5xx` responses.
+| `http_response_body_bytes_total` | Total of the responses body size.
+| `http_response_body_bytes` | Histogram of the responses body size.
+| `http_round_trip_time` | Histogram of the round trip time.
 |=======
 
 ==== Developer tools

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

@@ -42,6 +42,7 @@
  "github.com/elastic/beats/v7/libbeat/monitoring/inputmon"
  "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/elastic-agent-libs/logp"
  "github.com/elastic/elastic-agent-libs/mapstr"
  "github.com/elastic/elastic-agent-libs/monitoring"
@@ -122,7 +123,7 @@
  cfg := src.cfg
  log := env.Logger.With("input_url", cfg.Resource.URL)
 
- metrics := newInputMetrics(env.ID)
+ metrics, reg := newInputMetrics(env.ID)
  defer metrics.Close()
 
  ctx := ctxtool.FromCanceller(env.Cancelation)
@@ -132,7 +133,7 @@
  cfg.Resource.Tracer.Filename = strings.ReplaceAll(cfg.Resource.Tracer.Filename, "*", id)
  }
 
- client, trace, err := newClient(ctx, cfg, log)
+ client, trace, err := newClient(ctx, cfg, log, reg)
  if err != nil {
  return err
  }
@@ -686,7 +687,7 @@
  return limit, true
 }
 
-func newClient(ctx context.Context, cfg config, log *logp.Logger) (*http.Client, *httplog.LoggingRoundTripper, error) {
+func newClient(ctx context.Context, cfg config, log *logp.Logger, reg *monitoring.Registry) (*http.Client, *httplog.LoggingRoundTripper, error) {
  if !wantClient(cfg) {
  return nil, nil, nil
  }
@@ -729,6 +730,10 @@
  c.Transport = trace
  }
 
+ if reg != nil {
+ c.Transport = httpmon.NewMetricsRoundTripper(c.Transport, reg)
+ }
+
  c.CheckRedirect = checkRedirect(cfg.Resource, log)
 
  if cfg.Resource.Retry.getMaxAttempts() > 1 {
@@ -1070,7 +1075,7 @@
  batchProcessingTime metrics.Sample // histogram of the elapsed successful batch processing times in nanoseconds (time of receipt to time of ACK for non-empty batches).
 }
 
-func newInputMetrics(id string) *inputMetrics {
+func newInputMetrics(id string) (*inputMetrics, *monitoring.Registry) {
  reg, unreg := inputmon.NewInputRegistry(inputName, id, nil)
  out := &inputMetrics{
  unregister: unreg,
@@ -1088,7 +1093,7 @@
  _ = adapter.NewGoMetrics(reg, "batch_processing_time", adapter.Accept).
  Register("histogram", metrics.NewHistogram(out.batchProcessingTime))
 
- return out
+ return out, reg
 }
 
 func (m *inputMetrics) Close() {
@@ -1161,7 +1166,7 @@
 // walkMap walks to all ends of the provided path in m and applies fn to the
 // final element of each walk. Nested arrays are not handled.
 func walkMap(m mapstr.M, path string, fn func(parent mapstr.M, key string)) {
  key, rest, more := strings.Cut(path, ".")
  v, ok := m[key]
  if !ok {
  return