reload token file for InClusterConfig every 5 minutes

Kubernetes-commit: 287f6a564fb8c264f281056011f4a66f197b18f4

rest/config.go

@@ -30,7 +30,6 @@ import (
 	"time"
 
 	"github.com/golang/glog"
-
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -314,17 +313,23 @@ func DefaultKubernetesUserAgent() string {
 // running inside a pod running on kubernetes. It will return ErrNotInCluster
 // if called from a process not running in a kubernetes environment.
 func InClusterConfig() (*Config, error) {
+	const (
+		tokenFile  = "/var/run/secrets/kubernetes.io/serviceaccount/token"
+		rootCAFile = "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt"
+	)
 	host, port := os.Getenv("KUBERNETES_SERVICE_HOST"), os.Getenv("KUBERNETES_SERVICE_PORT")
 	if len(host) == 0 || len(port) == 0 {
 		return nil, ErrNotInCluster
 	}
 
-	token, err := ioutil.ReadFile("/var/run/secrets/kubernetes.io/serviceaccount/token")
-	if err != nil {
+	ts := newCachedPathTokenSource(tokenFile)
+
+	if _, err := ts.Token(); err != nil {
 		return nil, err
 	}
+
 	tlsClientConfig := TLSClientConfig{}
-	rootCAFile := "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt"
+
 	if _, err := certutil.NewPool(rootCAFile); err != nil {
 		glog.Errorf("Expected to load root CA config from %s, but got err: %v", rootCAFile, err)
 	} else {
@@ -334,8 +339,8 @@ func InClusterConfig() (*Config, error) {
 	return &Config{
 		// TODO: switch to using cluster DNS.
 		Host:            "https://" + net.JoinHostPort(host, port),
-		BearerToken:     string(token),
 		TLSClientConfig: tlsClientConfig,
+		WrapTransport:   TokenSourceWrapTransport(ts),
 	}, nil
 }
 

rest/token_source.go

@@ -0,0 +1,138 @@
+/*
+Copyright 2018 The Kubernetes 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 rest
+
+import (
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/golang/glog"
+	"golang.org/x/oauth2"
+)
+
+// TokenSourceWrapTransport returns a WrapTransport that injects bearer tokens
+// authentication from an oauth2.TokenSource.
+func TokenSourceWrapTransport(ts oauth2.TokenSource) func(http.RoundTripper) http.RoundTripper {
+	return func(rt http.RoundTripper) http.RoundTripper {
+		return &tokenSourceTransport{
+			base: rt,
+			ort: &oauth2.Transport{
+				Source: ts,
+				Base:   rt,
+			},
+		}
+	}
+}
+
+func newCachedPathTokenSource(path string) oauth2.TokenSource {
+	return &cachingTokenSource{
+		now:    time.Now,
+		leeway: 1 * time.Minute,
+		base: &fileTokenSource{
+			path: path,
+			// This period was picked because it is half of the minimum validity
+			// duration for a token provisioned by they TokenRequest API. This is
+			// unsophisticated and should induce rotation at a frequency that should
+			// work with the token volume source.
+			period: 5 * time.Minute,
+		},
+	}
+}
+
+type tokenSourceTransport struct {
+	base http.RoundTripper
+	ort  http.RoundTripper
+}
+
+func (tst *tokenSourceTransport) RoundTrip(req *http.Request) (*http.Response, error) {
+	// This is to allow --token to override other bearer token providers.
+	if req.Header.Get("Authorization") != "" {
+		return tst.base.RoundTrip(req)
+	}
+	return tst.ort.RoundTrip(req)
+}
+
+type fileTokenSource struct {
+	path   string
+	period time.Duration
+}
+
+var _ = oauth2.TokenSource(&fileTokenSource{})
+
+func (ts *fileTokenSource) Token() (*oauth2.Token, error) {
+	tokb, err := ioutil.ReadFile(ts.path)
+	if err != nil {
+		return nil, fmt.Errorf("failed to read token file %q: %v", ts.path, err)
+	}
+	tok := strings.TrimSpace(string(tokb))
+	if len(tok) == 0 {
+		return nil, fmt.Errorf("read empty token from file %q", ts.path)
+	}
+
+	return &oauth2.Token{
+		AccessToken: tok,
+		Expiry:      time.Now().Add(ts.period),
+	}, nil
+}
+
+type cachingTokenSource struct {
+	base   oauth2.TokenSource
+	leeway time.Duration
+
+	sync.RWMutex
+	tok *oauth2.Token
+
+	// for testing
+	now func() time.Time
+}
+
+var _ = oauth2.TokenSource(&cachingTokenSource{})
+
+func (ts *cachingTokenSource) Token() (*oauth2.Token, error) {
+	now := ts.now()
+	// fast path
+	ts.RLock()
+	tok := ts.tok
+	ts.RUnlock()
+
+	if tok != nil && tok.Expiry.Add(-1*ts.leeway).After(now) {
+		return tok, nil
+	}
+
+	// slow path
+	ts.Lock()
+	defer ts.Unlock()
+	if tok := ts.tok; tok != nil && tok.Expiry.Add(-1*ts.leeway).After(now) {
+		return tok, nil
+	}
+
+	tok, err := ts.base.Token()
+	if err != nil {
+		if ts.tok == nil {
+			return nil, err
+		}
+		glog.Errorf("Unable to rotate token: %v", err)
+		return ts.tok, nil
+	}
+
+	ts.tok = tok
+	return tok, nil
+}

rest/token_source_test.go

@@ -0,0 +1,156 @@
+/*
+Copyright 2018 The Kubernetes 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 rest
+
+import (
+	"fmt"
+	"reflect"
+	"sync"
+	"testing"
+	"time"
+
+	"golang.org/x/oauth2"
+)
+
+type testTokenSource struct {
+	calls int
+	tok   *oauth2.Token
+	err   error
+}
+
+func (ts *testTokenSource) Token() (*oauth2.Token, error) {
+	ts.calls++
+	return ts.tok, ts.err
+}
+
+func TestCachingTokenSource(t *testing.T) {
+	start := time.Now()
+	tokA := &oauth2.Token{
+		AccessToken: "a",
+		Expiry:      start.Add(10 * time.Minute),
+	}
+	tokB := &oauth2.Token{
+		AccessToken: "b",
+		Expiry:      start.Add(20 * time.Minute),
+	}
+	tests := []struct {
+		name string
+
+		tok   *oauth2.Token
+		tsTok *oauth2.Token
+		tsErr error
+		wait  time.Duration
+
+		wantTok     *oauth2.Token
+		wantErr     bool
+		wantTSCalls int
+	}{
+		{
+			name:    "valid token returned from cache",
+			tok:     tokA,
+			wantTok: tokA,
+		},
+		{
+			name:    "valid token returned from cache 1 minute before scheduled refresh",
+			tok:     tokA,
+			wait:    8 * time.Minute,
+			wantTok: tokA,
+		},
+		{
+			name:        "new token created when cache is empty",
+			tsTok:       tokA,
+			wantTok:     tokA,
+			wantTSCalls: 1,
+		},
+		{
+			name:        "new token created 1 minute after scheduled refresh",
+			tok:         tokA,
+			tsTok:       tokB,
+			wait:        10 * time.Minute,
+			wantTok:     tokB,
+			wantTSCalls: 1,
+		},
+		{
+			name:        "error on create token returns error",
+			tsErr:       fmt.Errorf("error"),
+			wantErr:     true,
+			wantTSCalls: 1,
+		},
+	}
+	for _, c := range tests {
+		t.Run(c.name, func(t *testing.T) {
+			tts := &testTokenSource{
+				tok: c.tsTok,
+				err: c.tsErr,
+			}
+
+			ts := &cachingTokenSource{
+				base:   tts,
+				tok:    c.tok,
+				leeway: 1 * time.Minute,
+				now:    func() time.Time { return start.Add(c.wait) },
+			}
+
+			gotTok, gotErr := ts.Token()
+			if got, want := gotTok, c.wantTok; !reflect.DeepEqual(got, want) {
+				t.Errorf("unexpected token:\n\tgot:\t%#v\n\twant:\t%#v", got, want)
+			}
+			if got, want := tts.calls, c.wantTSCalls; got != want {
+				t.Errorf("unexpected number of Token() calls: got %d, want %d", got, want)
+			}
+			if gotErr == nil && c.wantErr {
+				t.Errorf("wanted error but got none")
+			}
+			if gotErr != nil && !c.wantErr {
+				t.Errorf("unexpected error: %v", gotErr)
+			}
+		})
+	}
+}
+
+func TestCachingTokenSourceRace(t *testing.T) {
+	for i := 0; i < 100; i++ {
+		tts := &testTokenSource{
+			tok: &oauth2.Token{
+				AccessToken: "a",
+				Expiry:      time.Now().Add(1000 * time.Hour),
+			},
+		}
+
+		ts := &cachingTokenSource{
+			now:    time.Now,
+			base:   tts,
+			leeway: 1 * time.Minute,
+		}
+
+		var wg sync.WaitGroup
+		wg.Add(100)
+
+		for i := 0; i < 100; i++ {
+			go func() {
+				defer wg.Done()
+				if _, err := ts.Token(); err != nil {
+					t.Fatalf("err: %v", err)
+				}
+			}()
+		}
+		wg.Wait()
+		if tts.calls != 1 {
+			t.Errorf("expected one call to Token() but saw: %d", tts.calls)
+		}
+	}
+}