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

elastic · May 9, 2024 · 3afaead · 3afaead
1 parent 239deef
commit 3afaead
Show file tree

Hide file tree

Showing 3 changed files with 32 additions and 5 deletions.
diff --git a/CHANGELOG.next.asciidoc b/CHANGELOG.next.asciidoc
@@ -265,6 +265,7 @@ https://github.com/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}[]
 
 *Auditbeat*
 

diff --git a/x-pack/filebeat/docs/inputs/input-cel.asciidoc b/x-pack/filebeat/docs/inputs/input-cel.asciidoc
@@ -774,6 +774,27 @@ observe the activity of the input.
 | `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

diff --git a/x-pack/filebeat/input/cel/input.go b/x-pack/filebeat/input/cel/input.go
@@ -42,6 +42,7 @@ import (
 	"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 @@ func (i input) run(env v2.Context, src *source, cursor map[string]interface{}, p
 	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 @@ 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)
+	client, trace, err := newClient(ctx, cfg, log, reg)
 	if err != nil {
 		return err
 	}
@@ -686,7 +687,7 @@ func getLimit(which string, rateLimit map[string]interface{}, log *logp.Logger)
 	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 @@ func newClient(ctx context.Context, cfg config, log *logp.Logger) (*http.Client,
 		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 @@ type inputMetrics struct {
 	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 @@ func newInputMetrics(id string) *inputMetrics {
 	_ = adapter.NewGoMetrics(reg, "batch_processing_time", adapter.Accept).
 		Register("histogram", metrics.NewHistogram(out.batchProcessingTime))
 
-	return out
+	return out, reg
 }
 
 func (m *inputMetrics) Close() {