[Otelsoak] Adds Vercel drain mode for otelsoak

Makefile

@@ -14,7 +14,8 @@ INTEGRATION_MODS := $(shell $(FIND_INTEGRATION_TEST_MODS) | xargs $(TO_MOD_DIR)
 
 GROUP ?= all
 FOR_GROUP_TARGET=for-$(GROUP)-target
-GOTOOLCHAIN ?= go1.25.7+auto
+# Modules such as processor/elasticapmprocessor require go >= 1.26.
+GOTOOLCHAIN ?= go1.26.5+auto
 
 .DEFAULT_GOAL := all
 
@@ -92,7 +93,16 @@ remove-toolchain:
 # Build a collector based on the Elastic components (generate Elastic collector)
 .PHONY: genelasticcol
 genelasticcol:
-	GOTOOLCHAIN=${GOTOOLCHAIN} GOOS=${TARGET_GOOS} GOARCH=${TARGET_GOARCH} $(BUILDER) --config ./distributions/elastic-components/manifest.yaml
+	@set +e; \
+	GOTOOLCHAIN=${GOTOOLCHAIN} GOOS=${TARGET_GOOS} GOARCH=${TARGET_GOARCH} $(BUILDER) \
+		--config ./distributions/elastic-components/manifest.yaml --skip-compilation=true; \
+	status=$$?; \
+	set -e; \
+	if [ ! -f ./_build/go.mod ]; then exit $${status:-1}; fi; \
+	cd ./_build && GOTOOLCHAIN=${GOTOOLCHAIN} $(GOCMD) mod tidy; \
+	cd ..; \
+	GOTOOLCHAIN=${GOTOOLCHAIN} GOOS=${TARGET_GOOS} GOARCH=${TARGET_GOARCH} $(BUILDER) \
+		--config ./distributions/elastic-components/manifest.yaml --skip-get-modules=true
 
 # Validate that the Elastic components collector can run with the example configuration.
 .PHONY: elasticcol-validate
@@ -119,12 +129,39 @@ builddocker:
 .PHONY: otelsoak-validate
 otelsoak-validate: genelasticcol
 	ELASTIC_APM_SERVER_URL=http://localhost:8200 ELASTIC_APM_API_KEY=foobar ./loadgen/cmd/otelsoak/otelsoak validate --config ./loadgen/cmd/otelsoak/config.example.yaml
+	ELASTIC_SERVER_URL=http://localhost:8200 ELASTIC_APM_API_KEY=foobar ./loadgen/cmd/otelsoak/otelsoak validate --config ./loadgen/cmd/otelsoak/config.vercel.example.yaml
 
 # Run otelsoak
 .PHONY: otelsoak-run
 otelsoak-run: genelasticcol
 	./loadgen/cmd/otelsoak/otelsoak --config ./loadgen/cmd/otelsoak/config.example.yaml $(ARGS)
 
+# Run otelsoak against a Vercel Managed Inputs drain endpoint (HTTP NDJSON via httpexporter).
+# Optional: VERCEL_SIGNAL=logs|speed_insights|both (default logs).
+# For both, cats logs.jsonl + speed_insights.jsonl into a temp file at run time.
+VERCEL_SIGNAL ?= logs
+VERCEL_TESTDATA := loadgen/cmd/otelsoak/testdata/vercel
+.PHONY: otelsoak-run-vercel
+otelsoak-run-vercel: genelasticcol
+	@case "$(VERCEL_SIGNAL)" in \
+	  logs|speed_insights) \
+	    ./loadgen/cmd/otelsoak/otelsoak --config ./loadgen/cmd/otelsoak/config.vercel.example.yaml $(ARGS); \
+	    ;; \
+	  both) \
+	    tmp=$$(mktemp); \
+	    cat $(VERCEL_TESTDATA)/logs.jsonl $(VERCEL_TESTDATA)/speed_insights.jsonl > $$tmp; \
+	    ./loadgen/cmd/otelsoak/otelsoak --config ./loadgen/cmd/otelsoak/config.vercel.example.yaml \
+	      --set receivers.loadgen.logs.jsonl_file=$$tmp $(ARGS); \
+	    status=$$?; \
+	    rm -f $$tmp; \
+	    exit $$status; \
+	    ;; \
+	  *) \
+	    echo "VERCEL_SIGNAL must be logs, speed_insights, or both (got: $(VERCEL_SIGNAL))" >&2; \
+	    exit 1; \
+	    ;; \
+	esac
+
 
 # Clones the upstream opentelemetry-collector repository in a temporal .release
 # directory. If the directory already exists,

distributions/elastic-components/manifest.yaml

@@ -67,6 +67,7 @@ exporters:
   - gomod: go.opentelemetry.io/collector/exporter/debugexporter v0.157.0
   - gomod: go.opentelemetry.io/collector/exporter/otlpexporter v0.157.0
   - gomod: go.opentelemetry.io/collector/exporter/otlphttpexporter v0.157.0
+  - gomod: github.com/elastic/opentelemetry-collector-components/internal/exporter/httpexporter v0.1.0
 
 providers:
   - gomod: go.opentelemetry.io/collector/confmap/provider/envprovider v1.63.0
@@ -94,5 +95,6 @@ replaces:
   - github.com/elastic/opentelemetry-collector-components/receiver/integrationreceiver => ../receiver/integrationreceiver
   - github.com/elastic/opentelemetry-collector-components/receiver/entityanalyticsreceiver => ../receiver/entityanalyticsreceiver
   - github.com/elastic/opentelemetry-collector-components/receiver/akamaisiemreceiver => ../receiver/akamaisiemreceiver
+  - github.com/elastic/opentelemetry-collector-components/internal/exporter/httpexporter => ../internal/exporter/httpexporter
   - github.com/elastic/opentelemetry-collector-components/processor/elastictraceprocessor => ../processor/elastictraceprocessor
   - github.com/elastic/opentelemetry-collector-components/internal/elasticattr => ../internal/elasticattr

internal/exporter/httpexporter/Makefile

@@ -0,0 +1 @@
+include ../../../Makefile.Common

internal/exporter/httpexporter/README.md

@@ -0,0 +1,47 @@
+# HTTP body exporter
+
+<!-- status autogenerated section -->
+| Status        |           |
+| ------------- |-----------|
+| Stability     | [development]: logs   |
+| Distributions | [] |
+
+[development]: https://github.com/open-telemetry/opentelemetry-collector/blob/main/docs/component-stability.md#development
+<!-- end autogenerated section -->
+
+The HTTP body exporter POSTs log record bodies as the raw HTTP request body.
+Bodies are joined with newlines (NDJSON-friendly). It does **not** send OTLP.
+
+This is intended for load generation against Managed Input drain endpoints
+(for example Vercel) when used with [`loadgenreceiver`](../../receiver/loadgenreceiver) both in otelsoak / otelbench.
+
+## Configuration
+
+```yaml
+exporters:
+  http:
+    endpoint: https://example.ingest.us-central1.gcp.qa.elastic.cloud/inputs/vercel/_default_
+    headers:
+      - name: Authorization
+        value: "ApiKey ${env:ELASTIC_APM_API_KEY}"
+    tls:
+      insecure_skip_verify: true
+    timeout: 60s
+```
+
+| Field | Description |
+| --- | --- |
+| `endpoint` | Full URL to POST (required). |
+| `headers` | Extra HTTP headers (e.g. `Authorization`). |
+| `tls` | Standard collector TLS client settings. |
+| `timeout` | HTTP client timeout (default `30s`). |
+| `retry_on_failure` | Standard exporter retry settings. |
+| `sending_queue` | Standard exporter queue settings. |
+
+`Content-Type` defaults to `application/json` when not set in `headers`.
+
+## Sample otelsoak pipeline (Vercel drain)
+
+Put each drain NDJSON line in a loadgen OTLP-JSONL log body (see
+`loadgen/cmd/otelsoak/testdata/vercel/`). Select logs, speed-insights, or both
+with `VERCEL_SIGNAL` in [`config.vercel.example.yaml`](../../loadgen/cmd/otelsoak/config.vercel.example.yaml).

internal/exporter/httpexporter/config.go

@@ -0,0 +1,58 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package httpexporter // import "github.com/elastic/opentelemetry-collector-components/internal/exporter/httpexporter"
+
+import (
+	"errors"
+	"fmt"
+	"net/url"
+
+	"go.opentelemetry.io/collector/config/confighttp"
+	"go.opentelemetry.io/collector/config/configoptional"
+	"go.opentelemetry.io/collector/config/configretry"
+	"go.opentelemetry.io/collector/exporter/exporterhelper"
+)
+
+// Config configures the HTTP body exporter.
+type Config struct {
+	// ClientConfig holds standard HTTP client settings (endpoint, headers, TLS, timeout).
+	confighttp.ClientConfig `mapstructure:",squash"`
+
+	// RetryConfig defines retry configuration for failed exports.
+	RetryConfig configretry.BackOffConfig `mapstructure:"retry_on_failure"`
+
+	// QueueConfig defines optional sending queue settings.
+	QueueConfig configoptional.Optional[exporterhelper.QueueBatchConfig] `mapstructure:"sending_queue"`
+}
+
+func (cfg *Config) Validate() error {
+	if cfg.Endpoint == "" {
+		return errors.New("endpoint is required")
+	}
+	u, err := url.Parse(cfg.Endpoint)
+	if err != nil {
+		return fmt.Errorf("endpoint must be a valid URL: %w", err)
+	}
+	if u.Scheme != "http" && u.Scheme != "https" {
+		return fmt.Errorf("endpoint scheme must be http or https, got %q", u.Scheme)
+	}
+	if u.Host == "" {
+		return errors.New("endpoint must include a host")
+	}
+	return nil
+}

internal/exporter/httpexporter/doc.go

@@ -0,0 +1,23 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//go:generate mdatagen metadata.yaml
+
+// Package httpexporter posts log record bodies as an HTTP request body.
+// It is intended for load generation against raw HTTP ingest endpoints
+// such as Managed Input drains (e.g. Vercel NDJSON), not OTLP.
+package httpexporter // import "github.com/elastic/opentelemetry-collector-components/internal/exporter/httpexporter"

internal/exporter/httpexporter/exporter.go

@@ -0,0 +1,117 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package httpexporter // import "github.com/elastic/opentelemetry-collector-components/internal/exporter/httpexporter"
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+	"strings"
+
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/exporter"
+	"go.opentelemetry.io/collector/pdata/plog"
+	"go.uber.org/zap"
+)
+
+const defaultContentType = "application/json"
+
+type httpExporter struct {
+	config     *Config
+	logger     *zap.Logger
+	settings   component.TelemetrySettings
+	httpClient *http.Client
+}
+
+func newExporter(cfg *Config, set exporter.Settings) (*httpExporter, error) {
+	if err := cfg.Validate(); err != nil {
+		return nil, err
+	}
+	return &httpExporter{
+		config:   cfg,
+		logger:   set.Logger,
+		settings: set.TelemetrySettings,
+	}, nil
+}
+
+func (e *httpExporter) start(ctx context.Context, host component.Host) error {
+	client, err := e.config.ToClient(ctx, host.GetExtensions(), e.settings)
+	if err != nil {
+		return fmt.Errorf("failed to create HTTP client: %w", err)
+	}
+	e.httpClient = client
+	return nil
+}
+
+func (e *httpExporter) pushLogs(ctx context.Context, ld plog.Logs) error {
+	body := encodeLogBodies(ld)
+	if len(body) == 0 {
+		return nil
+	}
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, e.config.Endpoint, bytes.NewReader(body))
+	if err != nil {
+		return fmt.Errorf("failed to create request: %w", err)
+	}
+
+	if req.Header.Get("Content-Type") == "" {
+		req.Header.Set("Content-Type", defaultContentType)
+	}
+
+	resp, err := e.httpClient.Do(req)
+	if err != nil {
+		return fmt.Errorf("failed to POST to %s: %w", e.config.Endpoint, err)
+	}
+	defer func() {
+		// Drain body so connections can be reused.
+		_, _ = io.Copy(io.Discard, resp.Body)
+		if err := resp.Body.Close(); err != nil {
+			e.logger.Warn("failed to close response body", zap.Error(err))
+		}
+	}()
+
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		return fmt.Errorf("POST %s returned status %d", e.config.Endpoint, resp.StatusCode)
+	}
+	return nil
+}
+
+// encodeLogBodies joins each log record body as a line (NDJSON-friendly).
+func encodeLogBodies(ld plog.Logs) []byte {
+	var b strings.Builder
+	rls := ld.ResourceLogs()
+	for i := 0; i < rls.Len(); i++ {
+		sls := rls.At(i).ScopeLogs()
+		for j := 0; j < sls.Len(); j++ {
+			records := sls.At(j).LogRecords()
+			for k := 0; k < records.Len(); k++ {
+				line := records.At(k).Body().AsString()
+				if line == "" {
+					continue
+				}
+				if b.Len() > 0 {
+					b.WriteByte('\n')
+				}
+				b.WriteString(line)
+			}
+		}
+	}
+	return []byte(b.String())
+}