feat: support separate client TLS certificates

client/v2/client.go

@@ -54,6 +54,18 @@ type HTTPConfig struct {
 	// DialContext specifies the dial function for creating unencrypted TCP connections.
 	// If DialContext is nil then the transport dials using package net.
 	DialContext func(ctx context.Context, network, addr string) (net.Conn, error)
+
+	// DialTLSContext specifies the dial function for creating TLS connections for
+	// non-proxied HTTPS requests. The returned connection is assumed to have
+	// already completed its TLS handshake. It allows the TLS configuration to be
+	// resolved per connection instead of being fixed when the client is created.
+	//
+	// When DialTLSContext is set, TLSConfig is only used for proxied requests,
+	// which tunnel through the proxy and so do not use this function. Set both to
+	// keep proxied requests working.
+	//
+	// If DialTLSContext is nil then DialContext and TLSConfig are used.
+	DialTLSContext func(ctx context.Context, network, addr string) (net.Conn, error)
 }
 
 // BatchPointsConfig is the config data needed to create an instance of the BatchPoints struct.
@@ -125,8 +137,9 @@ func NewHTTPClient(conf HTTPConfig) (HTTPClient, error) {
 		TLSClientConfig: &tls.Config{
 			InsecureSkipVerify: conf.InsecureSkipVerify,
 		},
-		Proxy:       conf.Proxy,
-		DialContext: conf.DialContext,
+		Proxy:          conf.Proxy,
+		DialContext:    conf.DialContext,
+		DialTLSContext: conf.DialTLSContext,
 	}
 	if conf.TLSConfig != nil {
 		// A supplied TLSConfig is used as-is, including its own

cmd/influxd/run/server.go

@@ -23,6 +23,7 @@ import (
 	"github.com/influxdata/influxdb/logger"
 	"github.com/influxdata/influxdb/models"
 	"github.com/influxdata/influxdb/monitor"
+	"github.com/influxdata/influxdb/pkg/tlsconfig"
 	prometheus2 "github.com/influxdata/influxdb/prometheus"
 	"github.com/influxdata/influxdb/query"
 	"github.com/influxdata/influxdb/query/control"
@@ -80,6 +81,10 @@ type Server struct {
 	err     chan error
 	closing chan struct{}
 
+	// certMonitor is the TLS certificate monitor all TLSConfigManager objects
+	// in influxd should use.
+	certMonitor *tlsconfig.TLSCertMonitor
+
 	BindAddress string
 	Listener    net.Listener
 
@@ -197,6 +202,8 @@ func NewServer(c *Config, buildInfo *BuildInfo) (*Server, error) {
 
 		reportingDisabled: c.ReportingDisabled,
 
+		certMonitor: tlsconfig.NewTLSCertMonitor(),
+
 		httpAPIAddr: c.HTTPD.BindAddress,
 		httpUseTLS:  c.HTTPD.HTTPSEnabled,
 
@@ -217,7 +224,7 @@ func NewServer(c *Config, buildInfo *BuildInfo) (*Server, error) {
 	s.TSDBStore.EngineOptions.IndexVersion = c.Data.Index
 
 	// Create the Subscriber service
-	s.Subscriber = subscriber.NewService(c.Subscriber)
+	s.Subscriber = subscriber.NewService(c.Subscriber, s.certMonitor)
 
 	// Initialize points writer.
 	s.PointsWriter = coordinator.NewPointsWriter()
@@ -308,11 +315,11 @@ func (s *Server) appendRetentionPolicyService(c retention.Config) {
 	s.Services = append(s.Services, srv)
 }
 
-func (s *Server) appendHTTPDService(c httpd.Config) error {
+func (s *Server) appendHTTPDService(c httpd.Config, certMonitor *tlsconfig.TLSCertMonitor) error {
 	if !c.Enabled {
 		return nil
 	}
-	srv := httpd.NewService(c)
+	srv := httpd.NewService(c, certMonitor)
 	srv.Handler.MetaClient = s.MetaClient
 	authorizer := meta.NewQueryAuthorizer(s.MetaClient)
 	srv.Handler.QueryAuthorizer = authorizer
@@ -361,11 +368,11 @@ func (s *Server) appendCollectdService(c collectd.Config) {
 	s.Services = append(s.Services, srv)
 }
 
-func (s *Server) appendOpenTSDBService(c opentsdb.Config) error {
+func (s *Server) appendOpenTSDBService(c opentsdb.Config, certMonitor *tlsconfig.TLSCertMonitor) error {
 	if !c.Enabled {
 		return nil
 	}
-	srv, err := opentsdb.NewService(c)
+	srv, err := opentsdb.NewService(c, certMonitor)
 	if err != nil {
 		return err
 	}
@@ -437,6 +444,13 @@ func (s *Server) Open() error {
 		return err
 	}
 
+	// Configure and start certificate monitor. Do this early so we don't block
+	// on any of its channels.
+	s.certMonitor.SetLogger(s.Logger)
+	if err := s.certMonitor.Open(); err != nil {
+		return fmt.Errorf("error starting TLS certificate monitor: %w", err)
+	}
+
 	// Open shared TCP connection.
 	ln, err := net.Listen("tcp", s.BindAddress)
 	if err != nil {
@@ -453,7 +467,7 @@ func (s *Server) Open() error {
 	s.appendPrecreatorService(s.config.Precreator)
 	s.appendSnapshotterService()
 	s.appendContinuousQueryService(s.config.ContinuousQuery)
-	if err := s.appendHTTPDService(s.config.HTTPD); err != nil {
+	if err := s.appendHTTPDService(s.config.HTTPD, s.certMonitor); err != nil {
 		return err
 	}
 	s.appendRetentionPolicyService(s.config.Retention)
@@ -466,7 +480,7 @@ func (s *Server) Open() error {
 		s.appendCollectdService(i)
 	}
 	for _, i := range s.config.OpenTSDBInputs {
-		if err := s.appendOpenTSDBService(i); err != nil {
+		if err := s.appendOpenTSDBService(i, s.certMonitor); err != nil {
 			return err
 		}
 	}
@@ -542,6 +556,9 @@ func (s *Server) Open() error {
 func (s *Server) Close() error {
 	s.stopProfile()
 
+	// Close the TLS certificate monitor.
+	s.certMonitor.Close()
+
 	// Close the listener first to stop any new connections
 	if s.Listener != nil {
 		s.Listener.Close()
@@ -616,7 +633,7 @@ func (s *Server) ApplyReloadedConfig(config *Config, log *zap.Logger) error {
 	// Reload the subscriber's client TLS certificate at its currently
 	// configured path so rotated certificates are picked up.
 	if s.Subscriber != nil {
-		if af, err := s.Subscriber.PrepareReloadTLSCertificates(); err == nil {
+		if af, err := s.Subscriber.PrepareReloadTLSCertificates(config.Subscriber); err == nil {
 			applyFuncs = append(applyFuncs, af)
 		} else {
 			log.Error("error reloading subscriber service TLS certificate, no new configuration applied", zap.Error(err))

pkg/testing/helper/helpers.go

@@ -15,6 +15,7 @@ import (
 // can be replaced by:
 // defer CheckedClose(t, c)()
 func CheckedClose(t require.TestingT, c io.Closer) func() {
+	require.NotNil(t, c, "CheckedClose cannot be used with a nil io.Closer")
 	return func() {
 		if h, ok := t.(interface{ Helper() }); ok {
 			h.Helper()