Support for Custom Exporter Authenticators as Extensions

config/configauth/clientauth.go

@@ -1,14 +1,29 @@
+// Copyright  The OpenTelemetry Authors
+//
+// Licensed 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 configauth
 
 import (
 	"fmt"
+	"net/http"
+
+	"google.golang.org/grpc/credentials"
+
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config"
-	"google.golang.org/grpc/credentials"
-	"net/http"
 )
 
-
 // ClientAuth is an Extension that can be used as an authenticator for the configauth.Authentication option.
 // Authenticators are then included as part of OpenTelemetry Collector builds and can be referenced by their
 // names from the Authentication configuration. .
@@ -42,9 +57,8 @@ func GetHTTPClientAuth(extensions map[config.NamedEntity]component.Extension, re
 		if name.Name() == requested {
 			if auth, ok := ext.(HTTPClientAuth); ok {
 				return auth, nil
-			} else {
-				return nil, fmt.Errorf("requested authenticator is not for HTTP clients")
 			}
+			return nil, fmt.Errorf("requested authenticator is not for HTTP clients")
 		}
 	}
 	return nil, fmt.Errorf("failed to resolve authenticator %q: %w", requested, errAuthenticatorNotFound)
@@ -62,9 +76,8 @@ func GetGRPCClientAuth(extensions map[config.NamedEntity]component.Extension, re
 		if name.Name() == requested {
 			if auth, ok := ext.(GRPCClientAuth); ok {
 				return auth, nil
-			} else {
-				return nil, fmt.Errorf("requested authenticator is not for gRPC clients")
 			}
+			return nil, fmt.Errorf("requested authenticator is not for gRPC clients")
 		}
 	}
 	return nil, fmt.Errorf("failed to resolve authenticator %q: %w", requested, errAuthenticatorNotFound)

config/confighttp/confighttp.go

@@ -16,14 +16,14 @@ package confighttp
 
 import (
 	"crypto/tls"
-	"fmt"
-	"go.opentelemetry.io/collector/component"
-	"go.opentelemetry.io/collector/config"
-	"go.opentelemetry.io/collector/config/configauth"
 	"net"
 	"net/http"
 	"time"
 
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/config"
+	"go.opentelemetry.io/collector/config/configauth"
+
 	"github.com/rs/cors"
 
 	"go.opentelemetry.io/collector/config/configtls"
@@ -58,8 +58,31 @@ type HTTPClientSettings struct {
 	Auth *configauth.Authentication `mapstructure:"auth,omitempty"`
 }
 
+type toClientOptions struct {
+	extensionsMap map[config.NamedEntity]component.Extension
+}
+
+// ToClientOption is an option to add/change additional configuration on top of HTTPClientSettings
+// to construct HTTPClientSettings.ToClient().
+type ToClientOption func(options *toClientOptions)
+
+// WithExtensionsConfiguration is passed to HTTPClientSettings.ToClient() in order to
+// extend HTTPClientSettings with the configuration provided through extensions
+// e.g Authenticator configuration via extensions
+func WithExtensionsConfiguration(ext map[config.NamedEntity]component.Extension) ToClientOption {
+	return func(opts *toClientOptions) {
+		opts.extensionsMap = ext
+	}
+}
+
 // ToClient creates an HTTP client.
-func (hcs *HTTPClientSettings) ToClient() (*http.Client, error) {
+func (hcs *HTTPClientSettings) ToClient(opts ...ToClientOption) (*http.Client, error) {
+	// apply client options
+	clientOptions := &toClientOptions{}
+	for _, opt := range opts {
+		opt(clientOptions)
+	}
+
 	tlsCfg, err := hcs.TLSSetting.LoadTLSConfig()
 	if err != nil {
 		return nil, err
@@ -83,6 +106,17 @@ func (hcs *HTTPClientSettings) ToClient() (*http.Client, error) {
 		}
 	}
 
+	if clientOptions.extensionsMap != nil && hcs.Auth != nil {
+		httpCustomAuthRoundTripper, aerr := configauth.GetHTTPClientAuth(clientOptions.extensionsMap, hcs.Auth.AuthenticatorName)
+		if aerr != nil {
+			return nil, aerr
+		}
+		clientTransport, err = httpCustomAuthRoundTripper.RoundTripper(clientTransport)
+		if err != nil {
+			return nil, err
+		}
+	}
+
 	if hcs.CustomRoundTripper != nil {
 		clientTransport, err = hcs.CustomRoundTripper(clientTransport)
 		if err != nil {
@@ -96,19 +130,6 @@ func (hcs *HTTPClientSettings) ToClient() (*http.Client, error) {
 	}, nil
 }
 
-// AuthRoundTripper intercepts base transport and returns a new authenticated transport.
-func (hcs *HTTPClientSettings) AuthRoundTripper(clientBaseTransport http.RoundTripper, ext map[config.NamedEntity]component.Extension) (http.RoundTripper, error) {
-	if hcs.Auth != nil {
-		return nil, fmt.Errorf("configuration for auth absent")
-	}
-
-	httpCustomAuthRoundTripper, err := configauth.GetHTTPClientAuth(ext, hcs.Auth.AuthenticatorName)
-	if err != nil {
-		return nil, err
-	}
-	return httpCustomAuthRoundTripper.RoundTripper(clientBaseTransport)
-}
-
 // Custom RoundTripper that add headers
 type headerRoundTripper struct {
 	transport http.RoundTripper

config/confighttp/confighttp_test.go

@@ -15,6 +15,7 @@
 package confighttp
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"io/ioutil"
@@ -25,6 +26,10 @@ import (
 	"testing"
 	"time"
 
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/config"
+	"go.opentelemetry.io/collector/config/configauth"
+
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
@@ -120,6 +125,118 @@ func TestHTTPClientSettingsError(t *testing.T) {
 	}
 }
 
+type mockHTTPClientAuth struct {
+	forceError bool
+	err        error
+}
+
+func (m *mockHTTPClientAuth) Start(_ context.Context, _ component.Host) error {
+	return nil
+}
+
+func (m *mockHTTPClientAuth) Shutdown(_ context.Context) error {
+	return nil
+}
+
+type customRoundTripper struct {
+}
+
+var _ http.RoundTripper = (*customRoundTripper)(nil)
+
+func (c *customRoundTripper) RoundTrip(request *http.Request) (*http.Response, error) {
+	return nil, nil
+}
+
+func (m *mockHTTPClientAuth) RoundTripper(base http.RoundTripper) (http.RoundTripper, error) {
+	if m.forceError {
+		return nil, m.err
+	}
+	return &customRoundTripper{}, nil
+}
+
+var _ configauth.HTTPClientAuth = (*mockHTTPClientAuth)(nil)
+
+func TestHTTPClientSettingWithAuthConfig(t *testing.T) {
+	tests := []struct {
+		name         string
+		shouldErr    bool
+		settings     HTTPClientSettings
+		extensionMap map[config.NamedEntity]component.Extension
+	}{
+		{
+			name: "no_auth_extension_enabled",
+			settings: HTTPClientSettings{
+				Endpoint: "localhost:1234",
+				Auth:     nil,
+			},
+			shouldErr: false,
+			extensionMap: map[config.NamedEntity]component.Extension{
+				&config.ExtensionSettings{
+					NameVal: "mock",
+					TypeVal: "mock",
+				}: &mockHTTPClientAuth{},
+			},
+		},
+		{
+			name: "with_auth_configuration_has_no_extension",
+			settings: HTTPClientSettings{
+				Endpoint: "localhost:1234",
+				Auth:     &configauth.Authentication{AuthenticatorName: "dummy"},
+			},
+			shouldErr: true,
+			extensionMap: map[config.NamedEntity]component.Extension{
+				&config.ExtensionSettings{
+					NameVal: "mock",
+					TypeVal: "mock",
+				}: &mockHTTPClientAuth{},
+			},
+		},
+		{
+			name: "with_auth_configuration_has_extension",
+			settings: HTTPClientSettings{
+				Endpoint: "localhost:1234",
+				Auth:     &configauth.Authentication{AuthenticatorName: "mock"},
+			},
+			shouldErr: false,
+			extensionMap: map[config.NamedEntity]component.Extension{
+				&config.ExtensionSettings{
+					NameVal: "mock",
+					TypeVal: "mock",
+				}: &mockHTTPClientAuth{},
+			},
+		},
+		{
+			name: "with_auth_configuration_has_err_extension",
+			settings: HTTPClientSettings{
+				Endpoint: "localhost:1234",
+				Auth:     &configauth.Authentication{AuthenticatorName: "mock"},
+			},
+			shouldErr: true,
+			extensionMap: map[config.NamedEntity]component.Extension{
+				&config.ExtensionSettings{
+					NameVal: "mock",
+					TypeVal: "mock",
+				}: &mockHTTPClientAuth{forceError: true, err: errors.New("dummy extension")},
+			},
+		},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			client, err := test.settings.ToClient(WithExtensionsConfiguration(test.extensionMap))
+			if test.shouldErr {
+				assert.Error(t, err)
+				return
+			}
+			assert.NoError(t, err)
+			assert.NotNil(t, client)
+			if test.settings.Auth != nil {
+				_, ok := client.Transport.(*customRoundTripper)
+				assert.True(t, ok)
+			}
+		})
+	}
+}
+
 func TestHTTPServerSettingsError(t *testing.T) {
 	tests := []struct {
 		settings HTTPServerSettings

exporter/otlphttpexporter/otlp.go

@@ -19,7 +19,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"go.opentelemetry.io/collector/component"
 	"io"
 	"io/ioutil"
 	"net/http"
@@ -28,16 +27,19 @@ import (
 	"strings"
 	"time"
 
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/config/configgrpc"
+	"go.opentelemetry.io/collector/config/confighttp"
+	"go.opentelemetry.io/collector/internal/middleware"
+
 	"go.uber.org/zap"
 	"google.golang.org/genproto/googleapis/rpc/status"
 	"google.golang.org/protobuf/proto"
 
 	"go.opentelemetry.io/collector/config"
-	"go.opentelemetry.io/collector/config/configgrpc"
 	"go.opentelemetry.io/collector/consumer/consumererror"
 	"go.opentelemetry.io/collector/consumer/pdata"
 	"go.opentelemetry.io/collector/exporter/exporterhelper"
-	"go.opentelemetry.io/collector/internal/middleware"
 )
 
 type exporterImp struct {
@@ -66,41 +68,29 @@ func newExporter(cfg config.Exporter, logger *zap.Logger) (*exporterImp, error)
 		}
 	}
 
-	client, err := oCfg.HTTPClientSettings.ToClient()
-	if err != nil {
-		return nil, err
-	}
-
-	if oCfg.Compression != "" {
-		if strings.ToLower(oCfg.Compression) == configgrpc.CompressionGzip {
-			client.Transport = middleware.NewCompressRoundTripper(client.Transport)
-		} else {
-			return nil, fmt.Errorf("unsupported compression type %q", oCfg.Compression)
-		}
-	}
-
+	// client construction is deferred to start
 	return &exporterImp{
 		config: oCfg,
-		client: client,
 		logger: logger,
 	}, nil
 }
 
-// start overrides defaults base exporters start (no op) to attach auth transport to the http client.
-// This is the only place we get hold of Extensions which are required to construct auth round tripper.
+// start actually creates the HTTP client. 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 *exporterImp) start(_ context.Context, host component.Host) error {
-	// nothing to do if auth is not specified
-	if e.config.Auth == nil {
-		return nil
+	client, err := e.config.HTTPClientSettings.ToClient(confighttp.WithExtensionsConfiguration(host.GetExtensions()))
+	if err != nil {
+		return err
 	}
 
-	if e.config.HTTPClientSettings.Auth != nil {
-		transport, err := e.config.HTTPClientSettings.AuthRoundTripper(e.client.Transport, host.GetExtensions())
-		if err != nil {
-			return err
+	if e.config.Compression != "" {
+		if strings.ToLower(e.config.Compression) == configgrpc.CompressionGzip {
+			client.Transport = middleware.NewCompressRoundTripper(client.Transport)
+		} else {
+			return fmt.Errorf("unsupported compression type %q", e.config.Compression)
 		}
-		e.client.Transport = transport
 	}
+	e.client = client
 	return nil
 }