[chore] update testbed to embed jaeger exporter (#28880)

Rather than importing a deprecated module, this embeds the contents of
that module in the testbed. Part of #28647

Signed-off-by: Alex Boten <aboten@lightstep.com>

cmd/oteltestbedcol/builder-config.yaml

@@ -18,7 +18,6 @@ exporters:
   - gomod: go.opentelemetry.io/collector/exporter/otlpexporter v0.88.1-0.20231026220224-6405e152a2d9
   - gomod: go.opentelemetry.io/collector/exporter/otlphttpexporter v0.88.1-0.20231026220224-6405e152a2d9
   - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/exporter/carbonexporter v0.88.0
-  - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/exporter/jaegerexporter v0.85.0
   - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/exporter/opencensusexporter v0.88.0
   - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/exporter/prometheusexporter v0.88.0
   - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/exporter/sapmexporter v0.88.0

cmd/oteltestbedcol/go.mod

@@ -6,7 +6,6 @@ go 1.20
 
 require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/exporter/carbonexporter v0.88.0
-	github.com/open-telemetry/opentelemetry-collector-contrib/exporter/jaegerexporter v0.85.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/exporter/opencensusexporter v0.88.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/exporter/prometheusexporter v0.88.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/exporter/sapmexporter v0.88.0

testbed/datasenders/jaeger.go

@@ -5,15 +5,28 @@ package datasenders // import "github.com/open-telemetry/opentelemetry-collector
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"sync"
+	"time"
 
+	jaegerproto "github.com/jaegertracing/jaeger/proto-gen/api_v2"
+	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/component/componenttest"
+	"go.opentelemetry.io/collector/config/configgrpc"
 	"go.opentelemetry.io/collector/config/configtls"
 	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/consumer/consumererror"
+	"go.opentelemetry.io/collector/exporter"
+	"go.opentelemetry.io/collector/exporter/exporterhelper"
 	"go.opentelemetry.io/collector/exporter/exportertest"
+	"go.opentelemetry.io/collector/pdata/ptrace"
 	"go.uber.org/zap"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/connectivity"
+	"google.golang.org/grpc/metadata"
 
-	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/jaegerexporter"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/jaeger"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/testbed/testbed"
 )
 
@@ -35,20 +48,10 @@ func NewJaegerGRPCDataSender(host string, port int) testbed.TraceDataSender {
 }
 
 func (je *jaegerGRPCDataSender) Start() error {
-	factory := jaegerexporter.NewFactory()
-	cfg := factory.CreateDefaultConfig().(*jaegerexporter.Config)
-	// Disable retries, we should push data and if error just log it.
-	cfg.RetrySettings.Enabled = false
-	// Disable sending queue, we should push data from the caller goroutine.
-	cfg.QueueSettings.Enabled = false
-	cfg.Endpoint = je.GetEndpoint().String()
-	cfg.TLSSetting = configtls.TLSClientSetting{
-		Insecure: true,
-	}
 	params := exportertest.NewNopCreateSettings()
 	params.Logger = zap.L()
 
-	exp, err := factory.CreateTracesExporter(context.Background(), params, cfg)
+	exp, err := je.newTracesExporter(params)
 	if err != nil {
 		return err
 	}
@@ -68,3 +71,123 @@ func (je *jaegerGRPCDataSender) GenConfigYAMLStr() string {
 func (je *jaegerGRPCDataSender) ProtocolName() string {
 	return "jaeger"
 }
+
+// Config defines configuration for Jaeger gRPC exporter.
+type jaegerConfig struct {
+	exporterhelper.TimeoutSettings `mapstructure:",squash"` // squash ensures fields are correctly decoded in embedded struct.
+	exporterhelper.QueueSettings   `mapstructure:"sending_queue"`
+	exporterhelper.RetrySettings   `mapstructure:"retry_on_failure"`
+
+	configgrpc.GRPCClientSettings `mapstructure:",squash"` // squash ensures fields are correctly decoded in embedded struct.
+}
+
+var _ component.Config = (*jaegerConfig)(nil)
+
+// Validate checks if the exporter configuration is valid
+func (cfg *jaegerConfig) Validate() error {
+	if cfg.Endpoint == "" {
+		return errors.New("must have a non-empty \"endpoint\"")
+	}
+	return nil
+}
+
+// newTracesExporter returns a new Jaeger gRPC exporter.
+// The exporter name is the name to be used in the observability of the exporter.
+// The collectorEndpoint should be of the form "hostname:14250" (a gRPC target).
+func (je *jaegerGRPCDataSender) newTracesExporter(set exporter.CreateSettings) (exporter.Traces, error) {
+	cfg := jaegerConfig{}
+	cfg.Endpoint = je.GetEndpoint().String()
+	cfg.TLSSetting = configtls.TLSClientSetting{
+		Insecure: true,
+	}
+
+	s := &protoGRPCSender{
+		name:                      set.ID.String(),
+		settings:                  set.TelemetrySettings,
+		metadata:                  metadata.New(nil),
+		waitForReady:              cfg.WaitForReady,
+		connStateReporterInterval: time.Second,
+		stopCh:                    make(chan struct{}),
+		clientSettings:            &cfg.GRPCClientSettings,
+	}
+
+	return exporterhelper.NewTracesExporter(
+		context.TODO(), set, cfg, s.pushTraces,
+		exporterhelper.WithCapabilities(consumer.Capabilities{MutatesData: false}),
+		exporterhelper.WithStart(s.start),
+		exporterhelper.WithShutdown(s.shutdown),
+	)
+}
+
+// protoGRPCSender forwards spans encoded in the jaeger proto
+// format, to a grpc server.
+type protoGRPCSender struct {
+	name         string
+	settings     component.TelemetrySettings
+	client       jaegerproto.CollectorServiceClient
+	metadata     metadata.MD
+	waitForReady bool
+
+	conn                      stateReporter
+	connStateReporterInterval time.Duration
+
+	stopCh         chan struct{}
+	stopped        bool
+	stopLock       sync.Mutex
+	clientSettings *configgrpc.GRPCClientSettings
+}
+
+type stateReporter interface {
+	GetState() connectivity.State
+}
+
+func (s *protoGRPCSender) pushTraces(
+	ctx context.Context,
+	td ptrace.Traces,
+) error {
+
+	batches, err := jaeger.ProtoFromTraces(td)
+	if err != nil {
+		return consumererror.NewPermanent(fmt.Errorf("failed to push trace data via Jaeger exporter: %w", err))
+	}
+
+	if s.metadata.Len() > 0 {
+		ctx = metadata.NewOutgoingContext(ctx, s.metadata)
+	}
+
+	for _, batch := range batches {
+		_, err = s.client.PostSpans(
+			ctx,
+			&jaegerproto.PostSpansRequest{Batch: *batch}, grpc.WaitForReady(s.waitForReady))
+
+		if err != nil {
+			s.settings.Logger.Debug("failed to push trace data to Jaeger", zap.Error(err))
+			return fmt.Errorf("failed to push trace data via Jaeger exporter: %w", err)
+		}
+	}
+
+	return nil
+}
+
+func (s *protoGRPCSender) shutdown(context.Context) error {
+	s.stopLock.Lock()
+	s.stopped = true
+	s.stopLock.Unlock()
+	close(s.stopCh)
+	return nil
+}
+
+func (s *protoGRPCSender) start(ctx context.Context, host component.Host) error {
+	if s.clientSettings == nil {
+		return fmt.Errorf("client settings not found")
+	}
+	conn, err := s.clientSettings.ToClientConn(ctx, host, s.settings)
+	if err != nil {
+		return err
+	}
+
+	s.client = jaegerproto.NewCollectorServiceClient(conn)
+	s.conn = conn
+
+	return nil
+}

testbed/go.mod

@@ -4,8 +4,8 @@ go 1.20
 
 require (
 	github.com/fluent/fluent-logger-golang v1.9.0
+	github.com/jaegertracing/jaeger v1.48.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/exporter/carbonexporter v0.88.0
-	github.com/open-telemetry/opentelemetry-collector-contrib/exporter/jaegerexporter v0.85.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/exporter/opencensusexporter v0.88.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/exporter/prometheusexporter v0.88.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/exporter/sapmexporter v0.88.0
@@ -17,6 +17,7 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/splunk v0.88.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/golden v0.0.0-00010101000000-000000000000
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza v0.88.0
+	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/jaeger v0.88.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/carbonreceiver v0.88.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/datadogreceiver v0.88.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/jaegerreceiver v0.88.0
@@ -153,7 +154,6 @@ require (
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/influxdata/go-syslog/v3 v3.0.1-0.20210608084020-ac565dc76ba6 // indirect
 	github.com/ionos-cloud/sdk-go/v6 v6.1.8 // indirect
-	github.com/jaegertracing/jaeger v1.48.0 // indirect
 	github.com/jmespath/go-jmespath v0.4.0 // indirect
 	github.com/josharian/intern v1.0.0 // indirect
 	github.com/jpillora/backoff v1.0.0 // indirect
@@ -187,7 +187,6 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/experimentalmetricmetadata v0.88.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatautil v0.88.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/resourcetotelemetry v0.88.0 // indirect
-	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/jaeger v0.88.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/opencensus v0.88.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/prometheus v0.88.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/signalfx v0.88.0 // indirect