mtls metrics service (#7079)

* mtls metrics service

* pr review changes

* errors caused by upstream

* address pr comments

constants.go

@@ -246,6 +246,9 @@ const (
 	// ComponentSAML is a SAML service provider.
 	ComponentSAML = "saml"
 
+	// ComponentMetrics is a metrics server
+	ComponentMetrics = "metrics"
+
 	// DebugEnvVar tells tests to use verbose debug output
 	DebugEnvVar = "DEBUG"
 

lib/config/configuration.go

@@ -210,6 +210,9 @@ func ApplyFileConfig(fc *FileConfig, cfg *service.Config) error {
 	if fc.Databases.Disabled() {
 		cfg.Databases.Enabled = false
 	}
+	if fc.Metrics.Disabled() {
+		cfg.Metrics.Enabled = false
+	}
 	applyString(fc.NodeName, &cfg.Hostname)
 
 	// apply "advertise_ip" setting:
@@ -382,6 +385,11 @@ func ApplyFileConfig(fc *FileConfig, cfg *service.Config) error {
 			return trace.Wrap(err)
 		}
 	}
+	if fc.Metrics.Enabled() {
+		if err := applyMetricsConfig(fc, cfg); err != nil {
+			return trace.Wrap(err)
+		}
+	}
 
 	return nil
 }
@@ -1002,6 +1010,76 @@ func applyAppsConfig(fc *FileConfig, cfg *service.Config) error {
 	return nil
 }
 
+// applyMetricsConfig applies file configuration for the "metrics_service" section.
+func applyMetricsConfig(fc *FileConfig, cfg *service.Config) error {
+	// Metrics is enabled.
+	cfg.Metrics.Enabled = true
+
+	addr, err := utils.ParseHostPortAddr(fc.Metrics.ListenAddress, int(defaults.MetricsListenPort))
+	if err != nil {
+		return trace.Wrap(err)
+	}
+	cfg.Metrics.ListenAddr = addr
+
+	if !fc.Metrics.MTLSEnabled() {
+		return nil
+	}
+
+	cfg.Metrics.MTLS = true
+
+	if len(fc.Metrics.KeyPairs) == 0 {
+		return trace.BadParameter("at least one keypair shoud be provided when mtls is enabled in the metrics config")
+	}
+
+	if len(fc.Metrics.CACerts) == 0 {
+		return trace.BadParameter("at least one CA cert shoud be provided when mtls is enabled in the metrics config")
+	}
+
+	for _, p := range fc.Metrics.KeyPairs {
+		// Check that the certificate exists on disk. This exists to provide the
+		// user a sensible error message.
+		if !utils.FileExists(p.PrivateKey) {
+			return trace.NotFound("metrics service private key does not exist: %s", p.PrivateKey)
+		}
+		if !utils.FileExists(p.Certificate) {
+			return trace.NotFound("metrics service cert does not exist: %s", p.Certificate)
+		}
+
+		certificateChainBytes, err := utils.ReadPath(p.Certificate)
+		if err != nil {
+			return trace.Wrap(err)
+		}
+		certificateChain, err := utils.ReadCertificateChain(certificateChainBytes)
+		if err != nil {
+			return trace.Wrap(err)
+		}
+
+		if !utils.IsSelfSigned(certificateChain) {
+			if err := utils.VerifyCertificateChain(certificateChain); err != nil {
+				return trace.BadParameter("unable to verify the metrics service certificate chain in %v: %s",
+					p.Certificate, utils.UserMessageFromError(err))
+			}
+		}
+
+		cfg.Metrics.KeyPairs = append(cfg.Metrics.KeyPairs, service.KeyPairPath{
+			PrivateKey:  p.PrivateKey,
+			Certificate: p.Certificate,
+		})
+	}
+
+	for _, caCert := range fc.Metrics.CACerts {
+		// Check that the certificate exists on disk. This exists to provide the
+		// user a sensible error message.
+		if !utils.FileExists(caCert) {
+			return trace.NotFound("metrics service ca cert does not exist: %s", caCert)
+		}
+
+		cfg.Metrics.CACerts = append(cfg.Metrics.CACerts, caCert)
+	}
+
+	return nil
+}
+
 // parseAuthorizedKeys parses keys in the authorized_keys format and
 // returns a types.CertAuthority.
 func parseAuthorizedKeys(bytes []byte, allowedLogins []string) (types.CertAuthority, types.Role, error) {

lib/config/configuration_test.go

@@ -308,6 +308,19 @@ func TestConfigReading(t *testing.T) {
 				},
 			},
 		},
+		Metrics: Metrics{
+			Service: Service{
+				ListenAddress: "tcp://metrics",
+				EnabledFlag:   "yes",
+			},
+			KeyPairs: []KeyPair{
+				KeyPair{
+					PrivateKey:  "/etc/teleport/proxy.key",
+					Certificate: "/etc/teleport/proxy.crt",
+				},
+			},
+			CACerts: []string{"/etc/teleport/ca.crt"},
+		},
 	}, cmp.AllowUnexported(Service{})))
 	require.True(t, conf.Auth.Configured())
 	require.True(t, conf.Auth.Enabled())
@@ -321,6 +334,8 @@ func TestConfigReading(t *testing.T) {
 	require.True(t, conf.Apps.Enabled())
 	require.True(t, conf.Databases.Configured())
 	require.True(t, conf.Databases.Enabled())
+	require.True(t, conf.Metrics.Configured())
+	require.True(t, conf.Metrics.Enabled())
 
 	// good config from file
 	conf, err = ReadFromFile(testConfigs.configFileStatic)
@@ -593,6 +608,7 @@ func TestApplyConfigNoneEnabled(t *testing.T) {
 	require.Empty(t, cfg.SSH.PublicAddrs)
 	require.False(t, cfg.Apps.Enabled)
 	require.False(t, cfg.Databases.Enabled)
+	require.False(t, cfg.Metrics.Enabled)
 	require.Empty(t, cfg.Proxy.PostgresPublicAddrs)
 	require.Empty(t, cfg.Proxy.MySQLPublicAddrs)
 }
@@ -968,6 +984,17 @@ func makeConfigFixture() string {
 		},
 	}
 
+	// Metrics service.
+	conf.Metrics.EnabledFlag = "yes"
+	conf.Metrics.ListenAddress = "tcp://metrics"
+	conf.Metrics.CACerts = []string{"/etc/teleport/ca.crt"}
+	conf.Metrics.KeyPairs = []KeyPair{
+		KeyPair{
+			PrivateKey:  "/etc/teleport/proxy.key",
+			Certificate: "/etc/teleport/proxy.crt",
+		},
+	}
+
 	return conf.DebugDumpToYAML()
 }
 

lib/config/fileconf.go

@@ -72,8 +72,12 @@ type FileConfig struct {
 	Apps Apps `yaml:"app_service,omitempty"`
 
 	// Databases is the "db_service" section in Teleport configuration file
-	// that defined database access configuration.
+	// that defines database access configuration.
 	Databases Databases `yaml:"db_service,omitempty"`
+
+	// Metrics is the "metrics_service" section in Teleport configuration file
+	// that defines the metrics service configuration
+	Metrics Metrics `yaml:"metrics_service,omitempty"`
 }
 
 // ReadFromFile reads Teleport configuration from a file. Currently only YAML
@@ -1139,3 +1143,22 @@ func (o *OIDCConnector) Parse() (types.OIDCConnector, error) {
 	}
 	return v2, nil
 }
+
+// Metrics is a `metrics_service` section of the config file:
+type Metrics struct {
+	// Service is a generic service configuration section
+	Service `yaml:",inline"`
+
+	// KeyPairs is a list of x509 serving key pairs used for securing the metrics endpoint with mTLS.
+	// mTLS will be enabled for the service if both 'keypairs' and 'ca_certs' fields are set.
+	KeyPairs []KeyPair `yaml:"keypairs,omitempty"`
+
+	// CACerts is a list of prometheus CA certificates to validate clients against.
+	// mTLS will be enabled for the service if both 'keypairs' and 'ca_certs' fields are set.
+	CACerts []string `yaml:"ca_certs,omitempty"`
+}
+
+// MTLSEnabled returns whether mtls is enabled or not in the metrics service config.
+func (m *Metrics) MTLSEnabled() bool {
+	return len(m.KeyPairs) > 0 && len(m.CACerts) > 0
+}

lib/defaults/defaults.go

@@ -63,6 +63,9 @@ const (
 	// MySQLListenPort is the default listen port for MySQL proxy.
 	MySQLListenPort = 3036
 
+	// MetricsListenPort is the default listen port for the metrics service.
+	MetricsListenPort = 3081
+
 	// Default DB to use for persisting state. Another options is "etcd"
 	BackendType = "bolt"
 
@@ -575,6 +578,11 @@ func ReverseTunnelListenAddr() *utils.NetAddr {
 	return makeAddr(BindIP, SSHProxyTunnelListenPort)
 }
 
+// MetricsServiceListenAddr returns the default listening address for the metrics service
+func MetricsServiceListenAddr() *utils.NetAddr {
+	return makeAddr(BindIP, MetricsListenPort)
+}
+
 func makeAddr(host string, port int16) *utils.NetAddr {
 	addrSpec := fmt.Sprintf("tcp://%s:%d", host, port)
 	retval, err := utils.ParseAddr(addrSpec)

lib/service/cfg.go

@@ -105,6 +105,9 @@ type Config struct {
 	// Databases defines database proxy service configuration.
 	Databases DatabasesConfig
 
+	// Metrics defines the metrics service configuration.
+	Metrics MetricsConfig
+
 	// Keygen points to a key generator implementation
 	Keygen sshca.Authority
 
@@ -323,7 +326,7 @@ type ProxyConfig struct {
 	// Enabled turns proxy role on or off for this process
 	Enabled bool
 
-	//DisableTLS is enabled if we don't want self-signed certs
+	// DisableTLS is enabled if we don't want self signed certs
 	DisableTLS bool
 
 	// DisableWebInterface allows to turn off serving the Web UI interface
@@ -770,6 +773,29 @@ func (a App) Check() error {
 	return nil
 }
 
+// MetricsConfig specifies configuration for the metrics service
+type MetricsConfig struct {
+	// Enabled turns the metrics service role on or off for this process
+	Enabled bool
+
+	// ListenAddr is the address to listen on for incoming metrics requests.
+	// Optional.
+	ListenAddr *utils.NetAddr
+
+	// MTLS turns mTLS on the metrics service on or off
+	MTLS bool
+
+	// KeyPairs are the key and certificate pairs that the metrics service will
+	// use for mTLS.
+	// Used in conjunction with MTLS = true
+	KeyPairs []KeyPairPath
+
+	// CACerts are prometheus ca certs
+	// use for mTLS.
+	// Used in conjunction with MTLS = true
+	CACerts []string
+}
+
 // Rewrite is a list of rewriting rules to apply to requests and responses.
 type Rewrite struct {
 	// Redirect is a list of hosts that should be rewritten to the public address.
@@ -905,6 +931,9 @@ func ApplyDefaults(cfg *Config) {
 
 	// Databases proxy service is disabled by default.
 	cfg.Databases.Enabled = false
+
+	// Metrics service defaults.
+	cfg.Metrics.Enabled = false
 }
 
 // ApplyFIPSDefaults updates default configuration to be FedRAMP/FIPS 140-2

lib/service/listeners.go

@@ -39,6 +39,7 @@ var (
 	listenerProxyWeb          = listenerType(teleport.Component(teleport.ComponentProxy, "web"))
 	listenerProxyTunnel       = listenerType(teleport.Component(teleport.ComponentProxy, "tunnel"))
 	listenerProxyMySQL        = listenerType(teleport.Component(teleport.ComponentProxy, "mysql"))
+	listenerMetrics           = listenerType(teleport.ComponentMetrics)
 )
 
 // AuthSSHAddr returns auth server SSH endpoint, if configured and started.

lib/service/service.go

@@ -22,6 +22,7 @@ import (
 	"context"
 	"crypto/rand"
 	"crypto/tls"
+	"crypto/x509"
 	"encoding/hex"
 	"fmt"
 	"io"
@@ -159,6 +160,10 @@ const (
 	// is ready to start accepting connections.
 	DatabasesReady = "DatabasesReady"
 
+	// MetricsReady is generated when the Teleport metrics service is ready to
+	// start accepting connections.
+	MetricsReady = "MetricsReady"
+
 	// TeleportExitEvent is generated when the Teleport process begins closing
 	// all listening sockets and exiting.
 	TeleportExitEvent = "TeleportExit"
@@ -722,6 +727,9 @@ func NewTeleport(cfg *Config) (*TeleportProcess, error) {
 	if cfg.Apps.Enabled {
 		eventMapping.In = append(eventMapping.In, AppsReady)
 	}
+	if cfg.Metrics.Enabled {
+		eventMapping.In = append(eventMapping.In, MetricsReady)
+	}
 	process.RegisterEventMapping(eventMapping)
 
 	if cfg.Auth.Enabled {
@@ -773,6 +781,13 @@ func NewTeleport(cfg *Config) (*TeleportProcess, error) {
 		warnOnErr(process.closeImportedDescriptors(teleport.ComponentDatabase), process.log)
 	}
 
+	if cfg.Metrics.Enabled {
+		process.initMetricsService()
+		serviceStarted = true
+	} else {
+		warnOnErr(process.closeImportedDescriptors(teleport.ComponentMetrics), process.log)
+	}
+
 	process.RegisterFunc("common.rotate", process.periodicSyncRotationState)
 
 	if !serviceStarted {
@@ -2031,11 +2046,101 @@ func (process *TeleportProcess) initUploaderService(accessPoint auth.AccessPoint
 	return nil
 }
 
+// initMetricsService starts the metrics service currently serving metrics for
+// prometheus consumption
+func (process *TeleportProcess) initMetricsService() error {
+	mux := http.NewServeMux()
+	mux.Handle("/metrics", promhttp.Handler())
+
+	log := process.log.WithFields(logrus.Fields{
+		trace.Component: teleport.Component(teleport.ComponentMetrics, process.id),
+	})
+
+	listener, err := process.importOrCreateListener(listenerMetrics, process.Config.Metrics.ListenAddr.Addr)
+	if err != nil {
+		return trace.Wrap(err)
+	}
+	warnOnErr(process.closeImportedDescriptors(teleport.ComponentMetrics), log)
+
+	tlsConfig := &tls.Config{}
+	if process.Config.Metrics.MTLS {
+		for _, pair := range process.Config.Metrics.KeyPairs {
+			certificate, err := tls.LoadX509KeyPair(pair.Certificate, pair.PrivateKey)
+			if err != nil {
+				return trace.Wrap(err, "failed to read keypair: %+v", err)
+			}
+			tlsConfig.Certificates = append(tlsConfig.Certificates, certificate)
+		}
+
+		if len(tlsConfig.Certificates) == 0 {
+			return trace.BadParameter("no keypairs were provided for the metrics service with mtls enabled")
+		}
+
+		pool := x509.NewCertPool()
+		for _, caCertPath := range process.Config.Metrics.CACerts {
+			caCert, err := ioutil.ReadFile(caCertPath)
+			if err != nil {
+				return trace.Wrap(err, "failed to read prometheus CA certificate %+v", caCertPath)
+			}
+
+			if !pool.AppendCertsFromPEM(caCert) {
+				return trace.BadParameter("failed to parse prometheus CA certificate: %+v", caCertPath)
+			}
+		}
+
+		if len(pool.Subjects()) == 0 {
+			return trace.BadParameter("no prometheus ca certs were provided for the metrics service with mtls enabled")
+		}
+
+		tlsConfig.ClientAuth = tls.RequireAndVerifyClientCert
+		tlsConfig.ClientCAs = pool
+		tlsConfig.BuildNameToCertificate()
+
+		listener = tls.NewListener(listener, tlsConfig)
+	}
+
+	server := &http.Server{
+		Handler:           mux,
+		ReadHeaderTimeout: defaults.ReadHeadersTimeout,
+		ErrorLog:          utils.NewStdlogger(log.Error, teleport.ComponentMetrics),
+		TLSConfig:         tlsConfig,
+	}
+
+	log.Infof("Starting metrics service on %v.", process.Config.Metrics.ListenAddr.Addr)
+
+	process.RegisterFunc("metrics.service", func() error {
+		err := server.Serve(listener)
+		if err != nil && err != http.ErrServerClosed {
+			log.Warningf("Metrics server exited with error: %v.", err)
+		}
+		return nil
+	})
+
+	process.OnExit("metrics.shutdown", func(payload interface{}) {
+		if payload == nil {
+			log.Infof("Shutting down immediately.")
+			warnOnErr(server.Close(), log)
+		} else {
+			log.Infof("Shutting down gracefully.")
+			ctx := payloadContext(payload, log)
+			warnOnErr(server.Shutdown(ctx), log)
+		}
+		log.Infof("Exited.")
+	})
+	return nil
+}
+
 // initDiagnosticService starts diagnostic service currently serving healthz
 // and prometheus endpoints
 func (process *TeleportProcess) initDiagnosticService() error {
 	mux := http.NewServeMux()
-	mux.Handle("/metrics", promhttp.Handler())
+
+	// support legacy metrics collection in the diagnostic service.
+	// metrics will otherwise be served by the metrics service if it's enabled
+	// in the config.
+	if !process.Config.Metrics.Enabled {
+		mux.Handle("/metrics", promhttp.Handler())
+	}
 
 	if process.Config.Debug {
 		process.log.Infof("Adding diagnostic debugging handlers. To connect with profiler, use `go tool pprof %v`.", process.Config.DiagnosticAddr.Addr)