[Otelsoak] Adds Vercel drain mode for otelsoak

Makefile

@@ -119,12 +119,19 @@ 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).
+.PHONY: otelsoak-run-vercel
+otelsoak-run-vercel: genelasticcol
+	./loadgen/cmd/otelsoak/otelsoak --config ./loadgen/cmd/otelsoak/config.vercel.example.yaml $(ARGS)
+
 
 # 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,48 @@
+# 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)
+
+Use loadgenreceiver embedded Vercel presets (`vercel_logs`,
+`vercel_speed_insights`, `vercel_both`) via `VERCEL_SIGNAL` in
+[`config.vercel.example.yaml`](../../loadgen/cmd/otelsoak/config.vercel.example.yaml).
+Fixtures live under [`receiver/loadgenreceiver/testdata/vercel/`](../../receiver/loadgenreceiver/testdata/vercel/).

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,121 @@
+// 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"
+
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/consumer/consumererror"
+	"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 {
+	buf, empty := encodeLogBodies(ld)
+	if empty {
+		return nil
+	}
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, e.config.Endpoint, bytes.NewReader(buf.Bytes()))
+	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 nil
+	}
+
+	err = fmt.Errorf("POST %s returned status %d", e.config.Endpoint, resp.StatusCode)
+	// 5xx and 429 are typically transient; other 4xx (auth, bad request) are not.
+	if resp.StatusCode >= 500 || resp.StatusCode == http.StatusTooManyRequests {
+		return err
+	}
+	return consumererror.NewPermanent(err)
+}
+
+// encodeLogBodies joins each log record body as NDJSON (one JSON text per line,
+// each terminated by '\n', including after the last record).
+func encodeLogBodies(ld plog.Logs) (buf bytes.Buffer, empty bool) {
+	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
+				}
+				buf.WriteString(line)
+				buf.WriteByte('\n')
+			}
+		}
+	}
+	return buf, buf.Len() == 0
+}