add support for dynamic token and certificates reloading

Author: vadimalekseev

app/vlagent/kubernetescollector/client.go

@@ -2,8 +2,6 @@ package kubernetescollector
 
 import (
 	"context"
-	"crypto/tls"
-	"crypto/x509"
 	"encoding/json"
 	"fmt"
 	"io"
@@ -13,14 +11,12 @@ import (
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httputil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
 )
 
 type kubeAPIConfig struct {
-	Server        string
-	BearerToken   string
-	ClientCert    []byte
-	ClientCertKey []byte
-	GetCACert     func() (*x509.CertPool, error)
+	server string
+	ac     *promauth.Config
 }
 
 type kubeAPIClient struct {
@@ -31,34 +27,17 @@ type kubeAPIClient struct {
 }
 
 func newKubeAPIClient(cfg *kubeAPIConfig) (*kubeAPIClient, error) {
-	certPool, err := cfg.GetCACert()
-	if err != nil {
-		return nil, fmt.Errorf("cannot get CA certificate: %w", err)
-	}
-
-	var clientCerts []tls.Certificate
-	if len(cfg.ClientCert) != 0 {
-		cc, err := tls.X509KeyPair(cfg.ClientCert, cfg.ClientCertKey)
-		if err != nil {
-			return nil, fmt.Errorf("cannot load client certificate: %w", err)
-		}
-		clientCerts = append(clientCerts, cc)
-	}
-
 	tr := httputil.NewTransport(false, "vlagent_kubernetescollector")
 	tr.IdleConnTimeout = time.Minute
 	tr.TLSHandshakeTimeout = time.Second * 15
-	tr.TLSClientConfig.RootCAs = certPool
-	tr.TLSClientConfig.Certificates = clientCerts
 
-	// todo: ca cert can be updated, so we need to reload it periodically
 	c := &http.Client{
-		Transport: tr,
+		Transport: cfg.ac.NewRoundTripper(tr),
 	}
 
-	apiURL, err := url.Parse(cfg.Server)
+	apiURL, err := url.Parse(cfg.server)
 	if err != nil {
-		return nil, fmt.Errorf("cannot parse server URL %q: %w", cfg.Server, err)
+		return nil, fmt.Errorf("cannot parse server URL %q: %w", cfg.server, err)
 	}
 
 	return &kubeAPIClient{
@@ -98,7 +77,7 @@ func (c *kubeAPIClient) watchNodePods(ctx context.Context, nodeName, resourceVer
 	}
 
 	req := c.mustCreateRequest(ctx, http.MethodGet, "/api/v1/pods", args)
-	resp, err := c.c.Do(req)
+	resp, err := c.sendRequest(req)
 	if err != nil {
 		return podWatchStream{}, fmt.Errorf("cannot do %q GET request: %w", req.URL.String(), err)
 	}
@@ -223,7 +202,7 @@ func (c *kubeAPIClient) getNodePods(ctx context.Context, nodeName string) (podLi
 	}
 
 	req := c.mustCreateRequest(ctx, http.MethodGet, "/api/v1/pods", args)
-	resp, err := c.c.Do(req)
+	resp, err := c.sendRequest(req)
 	if err != nil {
 		return podList{}, fmt.Errorf("cannot do %q GET request: %w", req.URL.String(), err)
 	}
@@ -250,7 +229,7 @@ func (c *kubeAPIClient) getNodePods(ctx context.Context, nodeName string) (podLi
 // See https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.26/#read-pod-v1-core
 func (c *kubeAPIClient) getPod(ctx context.Context, namespace, podName string) (pod, error) {
 	req := c.mustCreateRequest(ctx, http.MethodGet, "/api/v1/namespaces/"+namespace+"/pods/"+podName, nil)
-	resp, err := c.c.Do(req)
+	resp, err := c.sendRequest(req)
 	if err != nil {
 		return pod{}, fmt.Errorf("cannot do /pods/<podName> request: %w", err)
 	}
@@ -290,7 +269,7 @@ type nodeList struct {
 // See https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.26/#list-node-v1-core
 func (c *kubeAPIClient) getNodes(ctx context.Context) ([]string, error) {
 	req := c.mustCreateRequest(ctx, http.MethodGet, "/api/v1/nodes", nil)
-	resp, err := c.c.Do(req)
+	resp, err := c.sendRequest(req)
 	if err != nil {
 		return nil, fmt.Errorf("cannot do %q GET request: %w", req.URL.String(), err)
 	}
@@ -321,7 +300,7 @@ func (c *kubeAPIClient) getNodes(ctx context.Context) ([]string, error) {
 // See https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.26/#read-node-v1-core
 func (c *kubeAPIClient) getNodeByName(ctx context.Context, nodeName string) (node, error) {
 	req := c.mustCreateRequest(ctx, http.MethodGet, "/api/v1/nodes/"+nodeName, nil)
-	resp, err := c.c.Do(req)
+	resp, err := c.sendRequest(req)
 	if err != nil {
 		return node{}, fmt.Errorf("cannot do %q GET request: %w", req.URL.String(), err)
 	}
@@ -346,14 +325,23 @@ func (c *kubeAPIClient) getNodeByName(ctx context.Context, nodeName string) (nod
 func (c *kubeAPIClient) mustCreateRequest(ctx context.Context, method, urlPath string, args url.Values) *http.Request {
 	req, err := http.NewRequestWithContext(ctx, method, "/", nil)
 	if err != nil {
-		logger.Panicf("BUG: cannot create request: %w", err)
+		logger.Panicf("BUG: cannot create request: %s", err)
 	}
 	u := *c.apiURL
 	req.URL = &u
 	req.URL.Path = urlPath
 	req.URL.RawQuery = args.Encode()
-	if c.config.BearerToken != "" {
-		req.Header.Set("Authorization", "Bearer "+c.config.BearerToken)
-	}
 	return req
 }
+
+func (c *kubeAPIClient) sendRequest(req *http.Request) (*http.Response, error) {
+	if err := c.config.ac.SetHeaders(req, true); err != nil {
+		return nil, err
+	}
+
+	resp, err := c.c.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	return resp, nil
+}

app/vlagent/kubernetescollector/client_config.go

@@ -1,13 +1,13 @@
 package kubernetescollector
 
 import (
-	"crypto/x509"
 	"encoding/base64"
 	"fmt"
 	"net"
 	"os"
 	"path/filepath"
 
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
 	"gopkg.in/yaml.v2"
 )
 
@@ -36,26 +36,27 @@ func loadInClusterConfig() (*kubeAPIConfig, error) {
 		return nil, fmt.Errorf("KUBERNETES_SERVICE_HOST/KUBERNETES_SERVICE_PORT environment variables are not set")
 	}
 
-	token, err := os.ReadFile("/var/run/secrets/kubernetes.io/serviceaccount/token")
+	const bearerTokenFile = "/var/run/secrets/kubernetes.io/serviceaccount/token"
+	// Verify that vlagent is running in a Kubernetes cluster.
+	if _, err := os.Stat(bearerTokenFile); err != nil {
+		return nil, err
+	}
+
+	opts := &promauth.Options{
+		BearerTokenFile: bearerTokenFile,
+		TLSConfig: &promauth.TLSConfig{
+			CAFile: "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt",
+		},
+	}
+	ac, err := opts.NewConfig()
 	if err != nil {
-		return nil, fmt.Errorf("cannot read in-cluster token: %w", err)
+		return nil, fmt.Errorf("cannot initialize in-cluster auth config: %w", err)
 	}
 
+	server := "https://" + net.JoinHostPort(host, port)
 	return &kubeAPIConfig{
-		Server:      "https://" + net.JoinHostPort(host, port),
-		BearerToken: string(token),
-		GetCACert: func() (*x509.CertPool, error) {
-			certs, err := os.ReadFile("/var/run/secrets/kubernetes.io/serviceaccount/ca.crt")
-			if err != nil {
-				return nil, fmt.Errorf("cannot read root CA: %w", err)
-			}
-
-			roots := x509.NewCertPool()
-			if !roots.AppendCertsFromPEM(certs) {
-				return nil, fmt.Errorf("cannot parse PEM encoded certificates")
-			}
-			return roots, nil
-		},
+		server: server,
+		ac:     ac,
 	}, nil
 }
 
@@ -136,7 +137,7 @@ func loadLocalConfig() (*kubeAPIConfig, error) {
 
 	rawConfig, err := os.ReadFile(configPath)
 	if err != nil {
-		return nil, fmt.Errorf("cannot read %q: %w", configPath, err)
+		return nil, err
 	}
 
 	var cfg kubeConfig
@@ -154,64 +155,54 @@ func loadLocalConfig() (*kubeAPIConfig, error) {
 		return nil, fmt.Errorf("cannot find cluster %q in %q", cctx.Context.Cluster, configPath)
 	}
 
-	var ca []byte
+	tlsCfg := promauth.TLSConfig{}
+
 	if cl.Cluster.CertificateAuthority != "" {
-		ca, err = os.ReadFile(cl.Cluster.CertificateAuthority)
-		if err != nil {
-			return nil, fmt.Errorf("cannot read cluster certificate authority: %w", err)
-		}
+		tlsCfg.CAFile = cl.Cluster.CertificateAuthority
 	} else if cl.Cluster.CertificateAuthorityData != "" {
-		ca, err = base64.StdEncoding.AppendDecode(nil, []byte(cl.Cluster.CertificateAuthorityData))
+		ca, err := base64.StdEncoding.DecodeString(cl.Cluster.CertificateAuthorityData)
 		if err != nil {
-			return nil, fmt.Errorf("cannot decode base64 encoded CA certificate data: %w", err)
+			return nil, fmt.Errorf("cannot decode base64 encoded CA certificate data from file %q: %w", configPath, err)
 		}
+		tlsCfg.CA = string(ca)
 	}
 
 	u, ok := cfg.findUser(cctx.Context.User)
 	if !ok {
-		return nil, fmt.Errorf("cannot find user %q in %q", cctx.Context.User, configPath)
+		return nil, fmt.Errorf("cannot find current user %q in %q", cctx.Context.User, configPath)
 	}
 
-	var clientCert []byte
 	if u.User.ClientCertificate != "" {
-		clientCert, err = os.ReadFile(u.User.ClientCertificate)
-		if err != nil {
-			return nil, fmt.Errorf("cannot read client certificate from %q: %w", u.User.ClientCertificate, err)
-		}
+		tlsCfg.CertFile = u.User.ClientCertificate
 	} else if u.User.ClientCertificateData != "" {
-		clientCert, err = base64.StdEncoding.AppendDecode(nil, []byte(u.User.ClientCertificateData))
+		clientCert, err := base64.StdEncoding.DecodeString(u.User.ClientCertificateData)
 		if err != nil {
-			return nil, fmt.Errorf("cannot decode base64 encoded client certificate data: %w", err)
+			return nil, fmt.Errorf("cannot decode base64 encoded client certificate data from file %q: %w", configPath, err)
 		}
+		tlsCfg.Cert = string(clientCert)
 	}
 
-	var clientCertKey []byte
 	if u.User.ClientKey != "" {
-		clientCertKey, err = os.ReadFile(u.User.ClientKey)
-		if err != nil {
-			return nil, fmt.Errorf("cannot read client key from %q: %w", u.User.ClientKey, err)
-		}
+		tlsCfg.KeyFile = u.User.ClientKey
 	} else if u.User.ClientKeyData != "" {
-		clientCertKey, err = base64.StdEncoding.AppendDecode(nil, []byte(u.User.ClientKeyData))
+		clientCertKey, err := base64.StdEncoding.DecodeString(u.User.ClientKeyData)
 		if err != nil {
-			return nil, fmt.Errorf("cannot decode base64 encoded client certificate key data: %w", err)
+			return nil, fmt.Errorf("cannot decode base64 encoded client certificate key data from file %q: %w", configPath, err)
 		}
+		tlsCfg.Key = string(clientCertKey)
+	}
+
+	opts := &promauth.Options{
+		BearerToken: u.User.Token,
+		TLSConfig:   &tlsCfg,
+	}
+	ac, err := opts.NewConfig()
+	if err != nil {
+		return nil, fmt.Errorf("cannot initialize local auth config from file %q: %w", configPath, err)
 	}
 
 	return &kubeAPIConfig{
-		Server:        cl.Cluster.Server,
-		BearerToken:   u.User.Token,
-		ClientCert:    clientCert,
-		ClientCertKey: clientCertKey,
-		GetCACert: func() (*x509.CertPool, error) {
-			if len(ca) == 0 {
-				return nil, nil
-			}
-			roots := x509.NewCertPool()
-			if !roots.AppendCertsFromPEM(ca) {
-				return nil, fmt.Errorf("cannot parse root CA for %q cluster from %q for user %q; no certs fetched", cl, configPath, cctx.Context.User)
-			}
-			return roots, nil
-		},
+		server: cl.Cluster.Server,
+		ac:     ac,
 	}, nil
 }

docs/victorialogs/CHANGELOG.md

@@ -21,6 +21,8 @@ according to the following docs:
 
 ## tip
 
+* FEATURE: [Kubernetes Collector](https://docs.victoriametrics.com/victorialogs/vlagent/#collect-kubernetes-pod-logs): add support for dynamic token and certificates reloading. Previously, vlagent only read credentials at startup, which led to authentication errors after token rotation. See [#995](https://github.com/VictoriaMetrics/VictoriaLogs/issues/995). 
+
 ## [v1.43.1](https://github.com/VictoriaMetrics/VictoriaLogs/releases/tag/v1.43.1)
 
 Released at 2025-12-26