cmd/telemetrygen: add HTTP export for logs (open-telemetry#29078)

**Description:**

Closes
open-telemetry#18867

**Testing:**

Ran opentelemetry-collector locally with debug exporter, then used
telemetrygen with `--otlp-http` with and without `--otlp-insecure`.

**Documentation:** None

.chloggen/telemetrygen-logs-http.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: cmd/telemetrygen
+
+# A brief description of the change. Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add support for --otlp-http for telemetrygen logs
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [18867]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

cmd/telemetrygen/config_test.go

@@ -11,8 +11,8 @@ import (
 
 // TestConfig_HTTPPath verifies that the HTTPPath configuration defaults are correctly set for each sub-command.
 func TestConfig_HTTPPath(t *testing.T) {
- t.Run("LogsConfigEmptyDefaultUrlPath", func(t *testing.T) {
- assert.Equal(t, "", logsCfg.HTTPPath)
+ t.Run("LogsConfigValidDefaultUrlPath", func(t *testing.T) {
+ assert.Equal(t, "/v1/logs", logsCfg.HTTPPath)
  })
 
  t.Run("MetricsConfigValidDefaultUrlPath", func(t *testing.T) {

cmd/telemetrygen/internal/logs/config.go

@@ -19,6 +19,9 @@ type Config struct {
 // Flags registers config flags.
 func (c *Config) Flags(fs *pflag.FlagSet) {
  c.CommonFlags(fs)
+
+ fs.StringVar(&c.HTTPPath, "otlp-http-url-path", "/v1/logs", "Which URL path to write to")
+
  fs.IntVar(&c.NumLogs, "logs", 1, "Number of logs to generate in each worker (ignored if duration is provided)")
  fs.StringVar(&c.Body, "body", "the message", "Body of the log")
 }

cmd/telemetrygen/internal/logs/exporter.go

@@ -0,0 +1,93 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package logs
+
+import (
+ "bytes"
+ "context"
+ "fmt"
+ "io"
+ "net/http"
+
+ "go.opentelemetry.io/collector/pdata/plog"
+ "go.opentelemetry.io/collector/pdata/plog/plogotlp"
+ "google.golang.org/grpc"
+ "google.golang.org/grpc/credentials/insecure"
+)
+
+type exporter interface {
+ export(plog.Logs) error
+}
+
+func newExporter(ctx context.Context, cfg *Config) (exporter, error) {
+ if cfg.UseHTTP {
+ return &httpClientExporter{
+ client: http.DefaultClient,
+ cfg: cfg,
+ }, nil
+ }
+
+ if !cfg.Insecure {
+ return nil, fmt.Errorf("'telemetrygen logs' only supports insecure gRPC")
+ }
+ // only support grpc in insecure mode
+ clientConn, err := grpc.DialContext(ctx, cfg.Endpoint(), grpc.WithTransportCredentials(insecure.NewCredentials()))
+ if err != nil {
+ return nil, err
+ }
+ return &gRPCClientExporter{client: plogotlp.NewGRPCClient(clientConn)}, nil
+}
+
+type gRPCClientExporter struct {
+ client plogotlp.GRPCClient
+}
+
+func (e *gRPCClientExporter) export(logs plog.Logs) error {
+ req := plogotlp.NewExportRequestFromLogs(logs)
+ if _, err := e.client.Export(context.Background(), req); err != nil {
+ return err
+ }
+ return nil
+}
+
+type httpClientExporter struct {
+ client *http.Client
+ cfg *Config
+}
+
+func (e *httpClientExporter) export(logs plog.Logs) error {
+ scheme := "https"
+ if e.cfg.Insecure {
+ scheme = "http"
+ }
+ path := e.cfg.HTTPPath
+ url := fmt.Sprintf("%s:https://%s%s", scheme, e.cfg.Endpoint(), path)
+
+ req := plogotlp.NewExportRequestFromLogs(logs)
+ body, err := req.MarshalProto()
+ if err != nil {
+ return fmt.Errorf("failed to marshal logs to protobuf: %w", err)
+ }
+
+ httpReq, err := http.NewRequestWithContext(context.Background(), "POST", url, bytes.NewReader(body))
+ if err != nil {
+ return fmt.Errorf("failed to create logs HTTP request: %w", err)
+ }
+ for k, v := range e.cfg.Headers {
+ httpReq.Header.Set(k, v)
+ }
+ httpReq.Header.Set("Content-Type", "application/x-protobuf")
+ resp, err := e.client.Do(httpReq)
+ if err != nil {
+ return fmt.Errorf("failed to execute logs HTTP request: %w", err)
+ }
+ defer resp.Body.Close()
+ if resp.StatusCode != http.StatusOK {
+ var respData bytes.Buffer
+ _, _ = io.Copy(&respData, resp.Body)
+ return fmt.Errorf("log request failed with status %s (%s)", resp.Status, respData.String())
+ }
+
+ return nil
+}

cmd/telemetrygen/internal/logs/logs.go

@@ -10,59 +10,27 @@ import (
  "sync/atomic"
  "time"
 
- "go.opentelemetry.io/collector/pdata/plog"
- "go.opentelemetry.io/collector/pdata/plog/plogotlp"
  "go.opentelemetry.io/otel/sdk/resource"
  semconv "go.opentelemetry.io/otel/semconv/v1.4.0"
  "go.uber.org/zap"
  "golang.org/x/time/rate"
- "google.golang.org/grpc"
- "google.golang.org/grpc/credentials/insecure"
 
  "github.com/open-telemetry/opentelemetry-collector-contrib/cmd/telemetrygen/internal/common"
 )
 
-type exporter interface {
- export(plog.Logs) error
-}
-
-type gRPCClientExporter struct {
- client plogotlp.GRPCClient
-}
-
-func (e *gRPCClientExporter) export(logs plog.Logs) error {
- req := plogotlp.NewExportRequestFromLogs(logs)
- if _, err := e.client.Export(context.Background(), req); err != nil {
- return err
- }
- return nil
-}
-
 // Start starts the log telemetry generator
 func Start(cfg *Config) error {
  logger, err := common.CreateLogger(cfg.SkipSettingGRPCLogger)
  if err != nil {
  return err
  }
 
- if cfg.UseHTTP {
- return fmt.Errorf("http is not supported by 'telemetrygen logs'")
- }
-
- if !cfg.Insecure {
- return fmt.Errorf("'telemetrygen logs' only supports insecure gRPC")
- }
-
- // only support grpc in insecure mode
- clientConn, err := grpc.DialContext(context.TODO(), cfg.Endpoint(), grpc.WithTransportCredentials(insecure.NewCredentials()))
+ e, err := newExporter(context.Background(), cfg)
  if err != nil {
  return err
  }
- exporter := &gRPCClientExporter{
- client: plogotlp.NewGRPCClient(clientConn),
- }
 
- if err = Run(cfg, exporter, logger); err != nil {
+ if err = Run(cfg, e, logger); err != nil {
  logger.Error("failed to stop the exporter", zap.Error(err))
  return err
  }