Refactor otlp exporter to upcoming auth changes

exporter/otlpexporter/config_test.go

@@ -36,6 +36,7 @@ func TestLoadConfig(t *testing.T) {
 
 	factory := NewFactory()
 	factories.Exporters[typeStr] = factory
+
 	cfg, err := configtest.LoadConfigFile(t, path.Join(".", "testdata", "config.yaml"), factories)
 
 	require.NoError(t, err)
@@ -81,12 +82,12 @@ func TestLoadConfig(t *testing.T) {
 					PermitWithoutStream: true,
 					Timeout:             30 * time.Second,
 				},
-				WriteBufferSize: 512 * 1024,
 				PerRPCAuth: &configgrpc.PerRPCAuthConfig{
 					AuthType:    "bearer",
 					BearerToken: "some-token",
 				},
-				BalancerName: "round_robin",
+				WriteBufferSize: 512 * 1024,
+				BalancerName:    "round_robin",
 			},
 		})
 }

exporter/otlpexporter/factory.go

@@ -71,6 +71,7 @@ func createTracesExporter(
 		exporterhelper.WithTimeout(oCfg.TimeoutSettings),
 		exporterhelper.WithRetry(oCfg.RetrySettings),
 		exporterhelper.WithQueue(oCfg.QueueSettings),
+		exporterhelper.WithStart(oce.start),
 		exporterhelper.WithShutdown(oce.shutdown))
 }
 
@@ -92,6 +93,7 @@ func createMetricsExporter(
 		exporterhelper.WithTimeout(oCfg.TimeoutSettings),
 		exporterhelper.WithRetry(oCfg.RetrySettings),
 		exporterhelper.WithQueue(oCfg.QueueSettings),
+		exporterhelper.WithStart(oce.start),
 		exporterhelper.WithShutdown(oce.shutdown),
 	)
 }
@@ -114,6 +116,7 @@ func createLogsExporter(
 		exporterhelper.WithTimeout(oCfg.TimeoutSettings),
 		exporterhelper.WithRetry(oCfg.RetrySettings),
 		exporterhelper.WithQueue(oCfg.QueueSettings),
+		exporterhelper.WithStart(oce.start),
 		exporterhelper.WithShutdown(oce.shutdown),
 	)
 }

exporter/otlpexporter/factory_test.go

@@ -24,6 +24,7 @@ import (
 	"go.uber.org/zap"
 
 	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/config"
 	"go.opentelemetry.io/collector/config/configcheck"
 	"go.opentelemetry.io/collector/config/configgrpc"
@@ -57,11 +58,11 @@ func TestCreateMetricsExporter(t *testing.T) {
 
 func TestCreateTracesExporter(t *testing.T) {
 	endpoint := testutil.GetAvailableLocalAddress(t)
-
 	tests := []struct {
-		name     string
-		config   Config
-		mustFail bool
+		name             string
+		config           Config
+		mustFailOnCreate bool
+		mustFailOnStart  bool
 	}{
 		{
 			name: "NoEndpoint",
@@ -71,7 +72,7 @@ func TestCreateTracesExporter(t *testing.T) {
 					Endpoint: "",
 				},
 			},
-			mustFail: true,
+			mustFailOnCreate: true,
 		},
 		{
 			name: "UseSecure",
@@ -140,7 +141,7 @@ func TestCreateTracesExporter(t *testing.T) {
 					Compression: "unknown compression",
 				},
 			},
-			mustFail: true,
+			mustFailOnStart: true,
 		},
 		{
 			name: "CaCert",
@@ -169,7 +170,7 @@ func TestCreateTracesExporter(t *testing.T) {
 					},
 				},
 			},
-			mustFail: true,
+			mustFailOnStart: true,
 		},
 	}
 
@@ -178,19 +179,23 @@ func TestCreateTracesExporter(t *testing.T) {
 			factory := NewFactory()
 			creationParams := component.ExporterCreateParams{Logger: zap.NewNop()}
 			consumer, err := factory.CreateTracesExporter(context.Background(), creationParams, &tt.config)
-
-			if tt.mustFail {
+			if tt.mustFailOnCreate {
 				assert.NotNil(t, err)
-			} else {
-				assert.NoError(t, err)
-				assert.NotNil(t, consumer)
-
-				err = consumer.Shutdown(context.Background())
-				if err != nil {
-					// Since the endpoint of OTLP exporter doesn't actually exist,
-					// exporter may already stop because it cannot connect.
-					assert.Equal(t, err.Error(), "rpc error: code = Canceled desc = grpc: the client connection is closing")
-				}
+				return
+			}
+			assert.NoError(t, err)
+			assert.NotNil(t, consumer)
+			err = consumer.Start(context.Background(), componenttest.NewNopHost())
+			if tt.mustFailOnStart {
+				assert.Error(t, err)
+				return
+			}
+			assert.NoError(t, err)
+			err = consumer.Shutdown(context.Background())
+			if err != nil {
+				// Since the endpoint of OTLP exporter doesn't actually exist,
+				// exporter may already stop because it cannot connect.
+				assert.Equal(t, err.Error(), "rpc error: code = Canceled desc = grpc: the client connection is closing")
 			}
 		})
 	}

exporter/otlpexporter/otlp.go

@@ -26,6 +26,7 @@ import (
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/status"
 
+	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config"
 	"go.opentelemetry.io/collector/consumer/consumererror"
 	"go.opentelemetry.io/collector/consumer/pdata"
@@ -50,12 +51,17 @@ func newExporter(cfg config.Exporter) (*exporter, error) {
 
 	e := &exporter{}
 	e.config = oCfg
-	w, err := newGrpcSender(oCfg)
+	return e, nil
+}
+
+// start actually creates the gRPC connection. The client construction is deferred till this point as this
+// is the only place we get hold of Extensions which are required to construct auth round tripper.
+func (e *exporter) start(_ context.Context, _ component.Host) (err error) {
+	e.w, err = newGrpcSender(e.config)
 	if err != nil {
-		return nil, err
+		return err
 	}
-	e.w = w
-	return e, nil
+	return nil
 }
 
 func (e *exporter) shutdown(context.Context) error {

exporter/otlpexporter/otlp_test.go

@@ -188,12 +188,12 @@ func TestSendTraces(t *testing.T) {
 	exp, err := factory.CreateTracesExporter(context.Background(), creationParams, cfg)
 	require.NoError(t, err)
 	require.NotNil(t, exp)
+
 	defer func() {
 		assert.NoError(t, exp.Shutdown(context.Background()))
 	}()
 
 	host := componenttest.NewNopHost()
-
 	assert.NoError(t, exp.Start(context.Background(), host))
 
 	// Ensure that initially there is no data in the receiver.