[extension/bearertokenauthextension] support reading tokens from file

extension/bearertokenauthextension/README.md

@@ -13,19 +13,26 @@ The authenticator type has to be set to `bearertokenauth`.
 
 ## Configuration
 
-The following is the only setting and is required:
+One of the following two options is required. If a token **and** a tokenfile are specified, the token is **ignored**:
 
 - `token`: static authorization token that needs to be sent on every gRPC client call as metadata.
   This token is prepended by "Bearer " before being sent as a value of "authorization" key in
   RPC metadata.
-
-  **Note**: bearertokenauth requires transport layer security enabled on the exporter.
+
+- `filename`: filename of file that contains a authorization token that needs to be sent on every
+  gRPC client call as metadata.
+  This token is prepended by "Bearer " before being sent as a value of "authorization" key in
+  RPC metadata.
+
+
+**Note**: bearertokenauth requires transport layer security enabled on the exporter.
 
 
 ```yaml
 extensions:
   bearertokenauth:
     token: "somerandomtoken"
+    filename: "file-containing.token"
 
 receivers:
   hostmetrics:
@@ -58,4 +65,4 @@ service:
 
 
 [beta]:https://github.com/open-telemetry/opentelemetry-collector#beta
-[contrib]:https://github.com/open-telemetry/opentelemetry-collector-releases/tree/main/distributions/otelcol-contrib
+[contrib]:https://github.com/open-telemetry/opentelemetry-collector-releases/tree/main/distributions/otelcol-contrib

extension/bearertokenauthextension/bearertokenauth.go

@@ -18,7 +18,10 @@ import (
 	"context"
 	"fmt"
 	"net/http"
+	"os"
+	"sync"
 
+	"github.com/fsnotify/fsnotify"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config/configauth"
 	"go.uber.org/zap"
@@ -44,26 +47,114 @@ func (c *PerRPCAuth) RequireTransportSecurity() bool {
 
 // BearerTokenAuth is an implementation of configauth.GRPCClientAuthenticator. It embeds a static authorization "bearer" token in every rpc call.
 type BearerTokenAuth struct {
-	tokenString string
-	logger      *zap.Logger
+	muTokenString sync.RWMutex
+	tokenString   string
+
+	shutdownCH chan struct{}
+
+	filename string
+	logger   *zap.Logger
 }
 
 var _ configauth.ClientAuthenticator = (*BearerTokenAuth)(nil)
 
 func newBearerTokenAuth(cfg *Config, logger *zap.Logger) *BearerTokenAuth {
+	if cfg.Filename != "" && cfg.BearerToken != "" {
+		logger.Warn("a filename is specified. Configured token is ignored!")
+	}
 	return &BearerTokenAuth{
 		tokenString: cfg.BearerToken,
+		filename:    cfg.Filename,
 		logger:      logger,
 	}
 }
 
-// Start of BearerTokenAuth does nothing and returns nil
+// Start of BearerTokenAuth does nothing and returns nil if no filename
+// is specified. Otherwise a routine is started to monitor the file containing
+// the token to be transferred.
 func (b *BearerTokenAuth) Start(ctx context.Context, host component.Host) error {
-	return nil
+	if b.filename == "" {
+		return nil
+	}
+
+	if b.shutdownCH != nil {
+		return fmt.Errorf("bearerToken file monitoring is already running")
+	}
+
+	// Read file once
+	b.refreshToken()
+
+	b.shutdownCH = make(chan struct{})
+
+	watcher, err := fsnotify.NewWatcher()
+	if err != nil {
+		return err
+	}
+	// start file watcher
+	go b.startWatcher(ctx, watcher)
+
+	return watcher.Add(b.filename)
+}
+
+func (b *BearerTokenAuth) startWatcher(ctx context.Context, watcher *fsnotify.Watcher) {
+	defer watcher.Close()
+	for {
+		select {
+		case _, ok := <-b.shutdownCH:
+			_ = ok
+			return
+		case <-ctx.Done():
+			return
+		case event, ok := <-watcher.Events:
+			if !ok {
+				continue
+			}
+			// NOTE: k8s configmaps uses symlinks, we need this workaround.
+			// original configmap file is removed.
+			// SEE: https://martensson.io/go-fsnotify-and-kubernetes-configmaps/
+			if event.Op == fsnotify.Remove || event.Op == fsnotify.Chmod {
+				// remove the watcher since the file is removed
+				if err := watcher.Remove(event.Name); err != nil {
+					b.logger.Error(err.Error())
+				}
+				// add a new watcher pointing to the new symlink/file
+				if err := watcher.Add(b.filename); err != nil {
+					b.logger.Error(err.Error())
+				}
+				b.refreshToken()
+			}
+			// also allow normal files to be modified and reloaded.
+			if event.Op == fsnotify.Write {
+				b.refreshToken()
+			}
+		}
+	}
+}
+
+func (b *BearerTokenAuth) refreshToken() {
+	b.logger.Info("refresh token", zap.String("filename", b.filename))
+	token, err := os.ReadFile(b.filename)
+	if err != nil {
+		b.logger.Error(err.Error())
+		return
+	}
+	b.muTokenString.Lock()
+	b.tokenString = string(token)
+	b.muTokenString.Unlock()
 }
 
 // Shutdown of BearerTokenAuth does nothing and returns nil
 func (b *BearerTokenAuth) Shutdown(ctx context.Context) error {
+	if b.filename == "" {
+		return nil
+	}
+
+	if b.shutdownCH == nil {
+		return fmt.Errorf("bearerToken file monitoring is not running")
+	}
+	b.shutdownCH <- struct{}{}
+	close(b.shutdownCH)
+	b.shutdownCH = nil
 	return nil
 }
 
@@ -75,7 +166,10 @@ func (b *BearerTokenAuth) PerRPCCredentials() (credentials.PerRPCCredentials, er
 }
 
 func (b *BearerTokenAuth) bearerToken() string {
-	return fmt.Sprintf("Bearer %s", b.tokenString)
+	b.muTokenString.RLock()
+	token := fmt.Sprintf("Bearer %s", b.tokenString)
+	b.muTokenString.RUnlock()
+	return token
 }
 
 // RoundTripper is not implemented by BearerTokenAuth

extension/bearertokenauthextension/bearertokenauth_test.go

@@ -18,10 +18,13 @@ import (
 	"context"
 	"fmt"
 	"net/http"
+	"os"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"go.opentelemetry.io/collector/component/componenttest"
+	"go.uber.org/zap/zaptest"
 )
 
 func TestPerRPCAuth(t *testing.T) {
@@ -104,3 +107,52 @@ func TestBearerAuthenticator(t *testing.T) {
 	assert.Equal(t, expectedHeaders, resp.Header)
 	assert.Nil(t, bauth.Shutdown(context.Background()))
 }
+
+func TestBearerStartWatchStop(t *testing.T) {
+	cfg := createDefaultConfig().(*Config)
+	cfg.Filename = "test.token"
+
+	bauth := newBearerTokenAuth(cfg, zaptest.NewLogger(t))
+	assert.NotNil(t, bauth)
+
+	assert.Nil(t, bauth.Start(context.Background(), componenttest.NewNopHost()))
+	assert.Error(t, bauth.Start(context.Background(), componenttest.NewNopHost()))
+
+	credential, err := bauth.PerRPCCredentials()
+	assert.NoError(t, err)
+	assert.NotNil(t, credential)
+
+	token, err := os.ReadFile(bauth.filename)
+	assert.NoError(t, err)
+
+	tokenStr := fmt.Sprintf("Bearer %s", token)
+	md, err := credential.GetRequestMetadata(context.Background())
+	expectedMd := map[string]string{
+		"authorization": tokenStr,
+	}
+	assert.Equal(t, md, expectedMd)
+	assert.NoError(t, err)
+	assert.True(t, credential.RequireTransportSecurity())
+
+	// change file content once
+	assert.Nil(t, os.WriteFile(bauth.filename, []byte(fmt.Sprintf("%stest", token)), 0600))
+	time.Sleep(5 * time.Second)
+	credential, _ = bauth.PerRPCCredentials()
+	md, err = credential.GetRequestMetadata(context.Background())
+	expectedMd["authorization"] = tokenStr + "test"
+	assert.Equal(t, md, expectedMd)
+	assert.NoError(t, err)
+
+	// change file content back
+	assert.Nil(t, os.WriteFile(bauth.filename, token, 0600))
+	time.Sleep(5 * time.Second)
+	credential, _ = bauth.PerRPCCredentials()
+	md, err = credential.GetRequestMetadata(context.Background())
+	expectedMd["authorization"] = tokenStr
+	time.Sleep(5 * time.Second)
+	assert.Equal(t, md, expectedMd)
+	assert.NoError(t, err)
+
+	assert.Nil(t, bauth.Shutdown(context.Background()))
+	assert.Nil(t, bauth.shutdownCH)
+}

extension/bearertokenauthextension/config.go

@@ -26,14 +26,17 @@ type Config struct {
 
 	// BearerToken specifies the bearer token to use for every RPC.
 	BearerToken string `mapstructure:"token,omitempty"`
+
+	// Filename points to a file that contains the bearer token to use for every RPC.
+	Filename string `mapstructure:"filename,omitempty"`
 }
 
 var _ config.Extension = (*Config)(nil)
 var errNoTokenProvided = errors.New("no bearer token provided")
 
 // Validate checks if the extension configuration is valid
 func (cfg *Config) Validate() error {
-	if cfg.BearerToken == "" {
+	if cfg.BearerToken == "" && cfg.Filename == "" {
 		return errNoTokenProvided
 	}
 	return nil

extension/bearertokenauthextension/go.mod

@@ -3,15 +3,16 @@ module github.com/open-telemetry/opentelemetry-collector-contrib/extension/beare
 go 1.18
 
 require (
+	github.com/fsnotify/fsnotify v1.5.4
 	github.com/stretchr/testify v1.8.0
 	go.opentelemetry.io/collector v0.60.1-0.20220916163348-84621e483dfb
 	go.uber.org/zap v1.23.0
 	google.golang.org/grpc v1.49.0
 )
 
 require (
+	github.com/benbjohnson/clock v1.3.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
-	github.com/fsnotify/fsnotify v1.5.4 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang/protobuf v1.5.2 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect

extension/bearertokenauthextension/test.token

@@ -0,0 +1 @@
+eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...+file+

unreleased/bearertokenauthextension_support_token_from_file.yaml

@@ -0,0 +1,16 @@
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: bearertokenauthextension
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: support reading tokens from file
+
+# One or more tracking issues related to the change
+issues: [14325]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext: