cmd/telemetrygen: add HTTP export for logs

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,90 @@
+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://%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.StatusCode, respData)
+	}
+
+	return nil
+}

cmd/telemetrygen/internal/logs/logs.go

@@ -10,57 +10,25 @@ 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()))
+	exporter, err := newExporter(context.Background(), cfg)
 	if err != nil {
 		return err
 	}
-	exporter := &gRPCClientExporter{
-		client: plogotlp.NewGRPCClient(clientConn),
-	}
 
 	if err = Run(cfg, exporter, logger); err != nil {
 		logger.Error("failed to stop the exporter", zap.Error(err))