Updated configgrpc ToDialOptions and confighttp ToClient apis to take…

… extensions configuration map (#3340)

This PR is a port of the configfrpc's ToDialOptions() and confighttp ToClient() from PR #3128  in a piece meal fashion. 

The following are the changes
- Refactored configgrpc.PerRPCAuth as extension implementing configauth.GrpcClientAuthenticator
- Plugged in extensions configuration to all the grpc based clients in the core (OTLP, OpenCensus, Jaeger, JaegerReceiver)
- Plugged in extensions configuration to all the HTTP based clients in the core (Zipkin, OTLPHTTP)

Link to tracking Issue:
 #3282 #3276

Testing:
Unit tests, [manual test described (for only oidc)](#3128 (comment))

config/configgrpc/configgrpc.go

@@ -94,12 +94,12 @@ type GRPCClientSettings struct {
  // The headers associated with gRPC requests.
  Headers map[string]string `mapstructure:"headers"`
 
- // PerRPCAuth parameter configures the client to send authentication data on a per-RPC basis.
- PerRPCAuth *PerRPCAuthConfig `mapstructure:"per_rpc_auth"`
-
  // Sets the balancer in grpclb_policy to discover the servers. Default is pick_first.
  // https://github.com/grpc/grpc-go/blob/master/examples/features/load_balancing/README.md
  BalancerName string `mapstructure:"balancer_name"`
+
+ // Auth configuration for outgoing RPCs.
+ Auth *configauth.Authentication `mapstructure:"auth,omitempty"`
 }
 
 // KeepaliveServerConfig is the configuration for keepalive.
@@ -108,15 +108,6 @@ type KeepaliveServerConfig struct {
  EnforcementPolicy *KeepaliveEnforcementPolicy `mapstructure:"enforcement_policy,omitempty"`
 }
 
-// PerRPCAuthConfig specifies how the Per-RPC authentication data should be obtained.
-type PerRPCAuthConfig struct {
- // AuthType represents the authentication type to use. Currently, only 'bearer' is supported.
- AuthType string `mapstructure:"type,omitempty"`
-
- // BearerToken specifies the bearer token to use for every RPC.
- BearerToken string `mapstructure:"bearer_token,omitempty"`
-}
-
 // KeepaliveServerParameters allow configuration of the keepalive.ServerParameters.
 // The same default values as keepalive.ServerParameters are applicable and get applied by the server.
 // See https://godoc.org/google.golang.org/grpc/keepalive#ServerParameters for details.
@@ -168,7 +159,7 @@ type GRPCServerSettings struct {
 }
 
 // ToDialOptions maps configgrpc.GRPCClientSettings to a slice of dial options for gRPC.
-func (gcs *GRPCClientSettings) ToDialOptions() ([]grpc.DialOption, error) {
+func (gcs *GRPCClientSettings) ToDialOptions(ext map[config.ComponentID]component.Extension) ([]grpc.DialOption, error) {
  var opts []grpc.DialOption
  if gcs.Compression != "" {
  if compressionKey := GetGRPCCompressionKey(gcs.Compression); compressionKey != CompressionUnsupported {
@@ -205,14 +196,26 @@ func (gcs *GRPCClientSettings) ToDialOptions() ([]grpc.DialOption, error) {
  opts = append(opts, keepAliveOption)
  }
 
- if gcs.PerRPCAuth != nil {
- if strings.EqualFold(gcs.PerRPCAuth.AuthType, PerRPCAuthTypeBearer) {
- sToken := gcs.PerRPCAuth.BearerToken
- token := BearerToken(sToken)
- opts = append(opts, grpc.WithPerRPCCredentials(token))
- } else {
- return nil, fmt.Errorf("unsupported per-RPC auth type %q", gcs.PerRPCAuth.AuthType)
+ if gcs.Auth != nil {
+ if ext == nil {
+ return nil, fmt.Errorf("no extensions configuration available")
+ }
+
+ componentID, cperr := config.NewIDFromString(gcs.Auth.AuthenticatorName)
+ if cperr != nil {
+ return nil, cperr
+ }
+
+ grpcAuthenticator, cerr := configauth.GetGRPCClientAuthenticator(ext, componentID)
+ if cerr != nil {
+ return nil, cerr
+ }
+
+ perRPCCredentials, perr := grpcAuthenticator.PerRPCCredentials()
+ if perr != nil {
+ return nil, err
  }
+ opts = append(opts, grpc.WithPerRPCCredentials(perRPCCredentials))
  }
 
  if gcs.BalancerName != "" {

config/configgrpc/configgrpc_test.go

@@ -41,7 +41,7 @@ func TestDefaultGrpcClientSettings(t *testing.T) {
  Insecure: true,
  },
  }
- opts, err := gcs.ToDialOptions()
+ opts, err := gcs.ToDialOptions(map[config.ComponentID]component.Extension{})
  assert.NoError(t, err)
  assert.Len(t, opts, 1)
 }
@@ -64,12 +64,17 @@ func TestAllGrpcClientSettings(t *testing.T) {
  ReadBufferSize: 1024,
  WriteBufferSize: 1024,
  WaitForReady: true,
- PerRPCAuth: nil,
  BalancerName: "round_robin",
+ Auth: &configauth.Authentication{AuthenticatorName: "testauth"},
  }
- opts, err := gcs.ToDialOptions()
+
+ ext := map[config.ComponentID]component.Extension{
+ config.NewID("testauth"): &configauth.MockClientAuthenticator{},
+ }
+
+ opts, err := gcs.ToDialOptions(ext)
  assert.NoError(t, err)
- assert.Len(t, opts, 6)
+ assert.Len(t, opts, 7)
 }
 
 func TestDefaultGrpcServerSettings(t *testing.T) {
@@ -137,6 +142,7 @@ func TestGRPCClientSettingsError(t *testing.T) {
  tests := []struct {
  settings GRPCClientSettings
  err string
+ ext map[config.ComponentID]component.Extension
  }{
  {
  err: "^failed to load TLS config: failed to load CA CertPool: failed to load CA /doesnt/exist:",
@@ -192,10 +198,36 @@ func TestGRPCClientSettingsError(t *testing.T) {
  BalancerName: "test",
  },
  },
+ {
+ err: "idStr must have non empty type",
+ settings: GRPCClientSettings{
+ Endpoint: "localhost:1234",
+ Auth: &configauth.Authentication{},
+ },
+ ext: map[config.ComponentID]component.Extension{
+ config.NewID("mock"): &configauth.MockClientAuthenticator{},
+ },
+ },
+ {
+ err: "failed to resolve authenticator \"doesntexist\": authenticator not found",
+ settings: GRPCClientSettings{
+ Endpoint: "localhost:1234",
+ Auth: &configauth.Authentication{AuthenticatorName: "doesntexist"},
+ },
+ ext: map[config.ComponentID]component.Extension{},
+ },
+ {
+ err: "no extensions configuration available",
+ settings: GRPCClientSettings{
+ Endpoint: "localhost:1234",
+ Auth: &configauth.Authentication{AuthenticatorName: "doesntexist"},
+ },
+ ext: nil,
+ },
  }
  for _, test := range tests {
  t.Run(test.err, func(t *testing.T) {
- opts, err := test.settings.ToDialOptions()
+ opts, err := test.settings.ToDialOptions(test.ext)
  assert.Nil(t, opts)
  assert.Error(t, err)
  assert.Regexp(t, test.err, err)
@@ -210,9 +242,8 @@ func TestUseSecure(t *testing.T) {
  Compression: "",
  TLSSetting: configtls.TLSClientSetting{},
  Keepalive: nil,
- PerRPCAuth: nil,
  }
- dialOpts, err := gcs.ToDialOptions()
+ dialOpts, err := gcs.ToDialOptions(map[config.ComponentID]component.Extension{})
  assert.NoError(t, err)
  assert.Equal(t, len(dialOpts), 1)
 }
@@ -431,7 +462,7 @@ func TestHttpReception(t *testing.T) {
  Endpoint: ln.Addr().String(),
  TLSSetting: *tt.tlsClientCreds,
  }
- clientOpts, errClient := gcs.ToDialOptions()
+ clientOpts, errClient := gcs.ToDialOptions(map[config.ComponentID]component.Extension{})
  assert.NoError(t, errClient)
  grpcClientConn, errDial := grpc.Dial(gcs.Endpoint, clientOpts...)
  assert.NoError(t, errDial)
@@ -478,7 +509,7 @@ func TestReceiveOnUnixDomainSocket(t *testing.T) {
  Insecure: true,
  },
  }
- clientOpts, errClient := gcs.ToDialOptions()
+ clientOpts, errClient := gcs.ToDialOptions(map[config.ComponentID]component.Extension{})
  assert.NoError(t, errClient)
  grpcClientConn, errDial := grpc.Dial(gcs.Endpoint, clientOpts...)
  assert.NoError(t, errDial)
@@ -497,36 +528,6 @@ func (gts *grpcTraceServer) Export(context.Context, *otelcol.ExportTraceServiceR
  return &otelcol.ExportTraceServiceResponse{}, nil
 }
 
-func TestWithPerRPCAuthBearerToken(t *testing.T) {
- // prepare
- // test
- gcs := &GRPCClientSettings{
- PerRPCAuth: &PerRPCAuthConfig{
- AuthType: "bearer",
- BearerToken: "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...",
- },
- }
- dialOpts, err := gcs.ToDialOptions()
-
- // verify
- assert.NoError(t, err)
- assert.Len(t, dialOpts, 2) // WithInsecure and WithPerRPCCredentials
-}
-
-func TestWithPerRPCAuthInvalidAuthType(t *testing.T) {
- // test
- gcs := &GRPCClientSettings{
- PerRPCAuth: &PerRPCAuthConfig{
- AuthType: "non-existing",
- },
- }
- dialOpts, err := gcs.ToDialOptions()
-
- // verify
- assert.Error(t, err)
- assert.Nil(t, dialOpts)
-}
-
 // tempSocketName provides a temporary Unix socket name for testing.
 func tempSocketName(t *testing.T) string {
  tmpfile, err := ioutil.TempFile("", "sock")

config/confighttp/confighttp.go

@@ -16,12 +16,16 @@ package confighttp
 
 import (
  "crypto/tls"
+ "fmt"
  "net"
  "net/http"
  "time"
 
  "github.com/rs/cors"
 
+ "go.opentelemetry.io/collector/component"
+ "go.opentelemetry.io/collector/config"
+ "go.opentelemetry.io/collector/config/configauth"
  "go.opentelemetry.io/collector/config/configtls"
  "go.opentelemetry.io/collector/internal/middleware"
 )
@@ -49,10 +53,13 @@ type HTTPClientSettings struct {
 
  // Custom Round Tripper to allow for individual components to intercept HTTP requests
  CustomRoundTripper func(next http.RoundTripper) (http.RoundTripper, error)
+
+ // Auth configuration for outgoing HTTP calls.
+ Auth *configauth.Authentication `mapstructure:"auth,omitempty"`
 }
 
 // ToClient creates an HTTP client.
-func (hcs *HTTPClientSettings) ToClient() (*http.Client, error) {
+func (hcs *HTTPClientSettings) ToClient(ext map[config.ComponentID]component.Extension) (*http.Client, error) {
  tlsCfg, err := hcs.TLSSetting.LoadTLSConfig()
  if err != nil {
  return nil, err
@@ -76,6 +83,27 @@ func (hcs *HTTPClientSettings) ToClient() (*http.Client, error) {
  }
  }
 
+ if hcs.Auth != nil {
+ if ext == nil {
+ return nil, fmt.Errorf("extensions configuration not found")
+ }
+
+ componentID, cperr := config.NewIDFromString(hcs.Auth.AuthenticatorName)
+ if cperr != nil {
+ return nil, cperr
+ }
+
+ httpCustomAuthRoundTripper, aerr := configauth.GetHTTPClientAuthenticator(ext, componentID)
+ 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 {
@@ -89,7 +117,7 @@ func (hcs *HTTPClientSettings) ToClient() (*http.Client, error) {
  }, nil
 }
 
-// Custom RoundTripper that add headers.
+// Custom RoundTripper that adds headers.
 type headerRoundTripper struct {
  transport http.RoundTripper
  headers map[string]string