Init session uploader in kubernetes service

It's started in all other services that upload sessions (app/proxy/ssh),
but was missing here. Because of this, the session storage directory for
async uploads wasn't created on disk and caused interactive sessions to
fail.

lib/kube/proxy/forwarder.go

@@ -115,19 +115,19 @@ type ForwarderConfig struct {
 // CheckAndSetDefaults checks and sets default values
 func (f *ForwarderConfig) CheckAndSetDefaults() error {
 	if f.AuthClient == nil {
-		return trace.BadParameter("missing parameter Client")
+		return trace.BadParameter("missing parameter AuthClient")
 	}
 	if f.CachingAuthClient == nil {
-		return trace.BadParameter("missing parameter AccessPoint")
+		return trace.BadParameter("missing parameter CachingAuthClient")
 	}
 	if f.Authz == nil {
-		return trace.BadParameter("missing parameter Auth")
+		return trace.BadParameter("missing parameter Authz")
 	}
 	if f.StreamEmitter == nil {
 		return trace.BadParameter("missing parameter StreamEmitter")
 	}
 	if f.ClusterName == "" {
-		return trace.BadParameter("missing parameter LocalCluster")
+		return trace.BadParameter("missing parameter ClusterName")
 	}
 	if f.Keygen == nil {
 		return trace.BadParameter("missing parameter Keygen")
@@ -1476,14 +1476,14 @@ func (f *Forwarder) getOrCreateRequestContext(key string) (context.Context, cont
 }
 
 func (f *Forwarder) getOrRequestClientCreds(ctx authContext) (*tls.Config, error) {
-	c := f.getClientCred(ctx)
+	c := f.getClientCreds(ctx)
 	if c == nil {
 		return f.serializedRequestClientCreds(ctx)
 	}
 	return c, nil
 }
 
-func (f *Forwarder) getClientCred(ctx authContext) *tls.Config {
+func (f *Forwarder) getClientCreds(ctx authContext) *tls.Config {
 	f.mu.Lock()
 	defer f.mu.Unlock()
 	creds, ok := f.clientCredentials.Get(ctx.key())
@@ -1497,7 +1497,7 @@ func (f *Forwarder) getClientCred(ctx authContext) *tls.Config {
 	return c
 }
 
-func (f *Forwarder) saveClientCred(ctx authContext, c *tls.Config) error {
+func (f *Forwarder) saveClientCreds(ctx authContext, c *tls.Config) error {
 	f.mu.Lock()
 	defer f.mu.Unlock()
 	return f.clientCredentials.Set(ctx.key(), c, ctx.sessionTTL)
@@ -1525,14 +1525,14 @@ func (f *Forwarder) serializedRequestClientCreds(authContext authContext) (*tls.
 		if err != nil {
 			return nil, trace.Wrap(err)
 		}
-		return c, f.saveClientCred(authContext, c)
+		return c, f.saveClientCreds(authContext, c)
 	}
 	// cancel == nil means that another request is in progress, so simply wait until
 	// it finishes or fails
 	f.log.Debugf("Another request is in progress for %v, waiting until it gets completed.", authContext)
 	select {
 	case <-ctx.Done():
-		c := f.getClientCred(authContext)
+		c := f.getClientCreds(authContext)
 		if c == nil {
 			return nil, trace.BadParameter("failed to request ephemeral certificate, try again")
 		}

lib/service/kubernetes.go

@@ -81,6 +81,12 @@ func (process *TeleportProcess) initKubernetesService(log *logrus.Entry, conn *C
 		return trace.Wrap(err)
 	}
 
+	// Start uploader that will scan a path on disk and upload completed
+	// sessions to the Auth Server.
+	if err := process.initUploaderService(accessPoint, conn.Client); err != nil {
+		return trace.Wrap(err)
+	}
+
 	// This service can run in 2 modes:
 	// 1. Reachable (by the proxy) - registers with auth server directly and
 	//    creates a local listener to accept proxy conns.
@@ -198,22 +204,22 @@ func (process *TeleportProcess) initKubernetesService(log *logrus.Entry, conn *C
 
 	kubeServer, err := kubeproxy.NewTLSServer(kubeproxy.TLSServerConfig{
 		ForwarderConfig: kubeproxy.ForwarderConfig{
-			Namespace:       defaults.Namespace,
-			Keygen:          cfg.Keygen,
-			ClusterName:     conn.ServerIdentity.Cert.Extensions[utils.CertExtensionAuthority],
-			Authz:            authorizer,
-			AuthClient:          conn.Client,
-			StreamEmitter:   streamEmitter,
-			DataDir:         cfg.DataDir,
-			CachingAuthClient:     accessPoint,
-			ServerID:        cfg.HostUUID,
-			Context:         process.ExitContext(),
-			KubeconfigPath:  cfg.Kube.KubeconfigPath,
-			KubeClusterName: cfg.Kube.KubeClusterName,
-			NewKubeService:  true,
-			Component:       teleport.ComponentKube,
-			StaticLabels:    cfg.Kube.StaticLabels,
-			DynamicLabels:   dynLabels,
+			Namespace:         defaults.Namespace,
+			Keygen:            cfg.Keygen,
+			ClusterName:       conn.ServerIdentity.Cert.Extensions[utils.CertExtensionAuthority],
+			Authz:             authorizer,
+			AuthClient:        conn.Client,
+			StreamEmitter:     streamEmitter,
+			DataDir:           cfg.DataDir,
+			CachingAuthClient: accessPoint,
+			ServerID:          cfg.HostUUID,
+			Context:           process.ExitContext(),
+			KubeconfigPath:    cfg.Kube.KubeconfigPath,
+			KubeClusterName:   cfg.Kube.KubeClusterName,
+			NewKubeService:    true,
+			Component:         teleport.ComponentKube,
+			StaticLabels:      cfg.Kube.StaticLabels,
+			DynamicLabels:     dynLabels,
 		},
 		TLS:           tlsConfig,
 		AccessPoint:   accessPoint,

lib/service/service.go

@@ -2548,19 +2548,19 @@ func (process *TeleportProcess) initProxyEndpoint(conn *Connector) error {
 		component := teleport.Component(teleport.ComponentProxy, teleport.ComponentProxyKube)
 		kubeServer, err = kubeproxy.NewTLSServer(kubeproxy.TLSServerConfig{
 			ForwarderConfig: kubeproxy.ForwarderConfig{
-				Namespace:       defaults.Namespace,
-				Keygen:          cfg.Keygen,
-				ClusterName:     conn.ServerIdentity.Cert.Extensions[utils.CertExtensionAuthority],
-				ReverseTunnelSrv:          tsrv,
-				Authz:            authorizer,
-				AuthClient:          conn.Client,
-				StreamEmitter:   streamEmitter,
-				DataDir:         cfg.DataDir,
-				CachingAuthClient:     accessPoint,
-				ServerID:        cfg.HostUUID,
-				ClusterOverride: cfg.Proxy.Kube.ClusterOverride,
-				KubeconfigPath:  cfg.Proxy.Kube.KubeconfigPath,
-				Component:       component,
+				Namespace:         defaults.Namespace,
+				Keygen:            cfg.Keygen,
+				ClusterName:       conn.ServerIdentity.Cert.Extensions[utils.CertExtensionAuthority],
+				ReverseTunnelSrv:  tsrv,
+				Authz:             authorizer,
+				AuthClient:        conn.Client,
+				StreamEmitter:     streamEmitter,
+				DataDir:           cfg.DataDir,
+				CachingAuthClient: accessPoint,
+				ServerID:          cfg.HostUUID,
+				ClusterOverride:   cfg.Proxy.Kube.ClusterOverride,
+				KubeconfigPath:    cfg.Proxy.Kube.KubeconfigPath,
+				Component:         component,
 			},
 			TLS:           tlsConfig,
 			LimiterConfig: cfg.Proxy.Limiter,