Implement kubernetes_service registration and sratup

The new service now starts, registers (locally or via a join token) and
heartbeats its presence to the auth server.

This service can handle k8s requests (like a proxy) but not to remote
teleport clusters. Proxies will be responsible for routing those.
The client (tsh) will not yet go to this service, until proxy routing is
implemented. I manually tweaked server addres in kubeconfig to test it.

You can also run `tctl get kube_service` to list all registered
instances. The self-reported info is currently limited - only listening
address is set.

Author: Andrew Lytvynov

Diff for: constants.go

@@ -104,8 +104,8 @@ const (
 	// ComponentLabel is a component label name used in reporting
 	ComponentLabel = "component"
 
-	// ComponentKube is a kubernetes proxy
-	ComponentKube = "proxy:kube"
+	// ComponentProxyKube is a kubernetes proxy
+	ComponentProxyKube = "proxy:kube"
 
 	// ComponentAuth is the cluster CA node (auth server API)
 	ComponentAuth = "auth"
@@ -219,6 +219,9 @@ const (
 	// ComponentCgroup is the cgroup package.
 	ComponentCgroup = "cgroups"
 
+	// ComponentKube is an Kubernetes API gateway.
+	ComponentKube = "kubernetes"
+
 	// DebugEnvVar tells tests to use verbose debug output
 	DebugEnvVar = "DEBUG"
 

Diff for: lib/auth/api.go

@@ -41,6 +41,10 @@ type Announcer interface {
 	// for the specified duration with second resolution if it's >= 1 second
 	UpsertAuthServer(s services.Server) error
 
+	// UpsertKubeService registers kubernetes presence, permanently if ttl is 0
+	// or for the specified duration with second resolution if it's >= 1 second
+	UpsertKubeService(s services.Server) error
+
 	// NewKeepAliver returns a new instance of keep aliver
 	NewKeepAliver(ctx context.Context) (services.KeepAliver, error)
 }
@@ -96,6 +100,9 @@ type ReadAccessPoint interface {
 
 	// GetTunnelConnections returns tunnel connections for a given cluster
 	GetTunnelConnections(clusterName string, opts ...services.MarshalOption) ([]services.TunnelConnection, error)
+
+	// GetKubeServices returns a list of kubernetes services registered in the cluster
+	GetKubeServices() ([]services.Server, error)
 }
 
 // AccessPoint is an API interface implemented by a certificate authority (CA)
@@ -237,6 +244,11 @@ func (w *Wrapper) DeleteSemaphore(ctx context.Context, filter services.Semaphore
 	return w.NoCache.DeleteSemaphore(ctx, filter)
 }
 
+// UpsertKubeService is part of auth.AccessPoint implementation
+func (w *Wrapper) UpsertKubeService(s services.Server) error {
+	return w.NoCache.UpsertKubeService(s)
+}
+
 // NewCachingAcessPoint returns new caching access point using
 // access point policy
 type NewCachingAccessPoint func(clt ClientI, cacheName []string) (AccessPoint, error)

Diff for: lib/auth/apiserver.go

@@ -140,6 +140,8 @@ func NewAPIServer(config *APIConfig) http.Handler {
 	srv.DELETE("/:version/tunnelconnections/:cluster/:conn", srv.withAuth(srv.deleteTunnelConnection))
 	srv.DELETE("/:version/tunnelconnections/:cluster", srv.withAuth(srv.deleteTunnelConnections))
 	srv.DELETE("/:version/tunnelconnections", srv.withAuth(srv.deleteAllTunnelConnections))
+	srv.POST("/:version/kube_services", srv.withAuth(srv.upsertKubeService))
+	srv.GET("/:version/kube_services", srv.withAuth(srv.getKubeServices))
 
 	// Server Credentials
 	srv.POST("/:version/server/credentials", srv.withAuth(srv.generateServerKeys))
@@ -366,6 +368,12 @@ func (s *APIServer) upsertServer(auth ClientI, role teleport.Role, w http.Respon
 		if err := auth.UpsertProxy(server); err != nil {
 			return nil, trace.Wrap(err)
 		}
+	case teleport.RoleKube:
+		if err := auth.UpsertKubeService(server); err != nil {
+			return nil, trace.Wrap(err)
+		}
+	default:
+		return nil, trace.BadParameter("unknown server role %q", role)
 	}
 	return message("ok"), nil
 }
@@ -2481,6 +2489,18 @@ func (s *APIServer) getServerID(r *http.Request) (string, error) {
 	return strings.TrimSuffix(role.Username, "."+clusterName), nil
 }
 
+func (s *APIServer) upsertKubeService(auth ClientI, w http.ResponseWriter, r *http.Request, p httprouter.Params, version string) (interface{}, error) {
+	return s.upsertServer(auth, teleport.RoleKube, w, r, p, version)
+}
+
+func (s *APIServer) getKubeServices(auth ClientI, w http.ResponseWriter, r *http.Request, p httprouter.Params, version string) (interface{}, error) {
+	servers, err := auth.GetKubeServices()
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	return marshalServers(servers, version)
+}
+
 func message(msg string) map[string]interface{} {
 	return map[string]interface{}{"message": msg}
 }

Diff for: lib/auth/auth_with_roles.go

@@ -1969,6 +1969,26 @@ func (a *ServerWithRoles) WaitForDelivery(context.Context) error {
 	return nil
 }
 
+func (a *ServerWithRoles) UpsertKubeService(s services.Server) error {
+	if err := a.action(defaults.Namespace, services.KindKubeService, services.VerbCreate); err != nil {
+		return trace.Wrap(err)
+	}
+	if err := a.action(defaults.Namespace, services.KindKubeService, services.VerbUpdate); err != nil {
+		return trace.Wrap(err)
+	}
+	return a.authServer.UpsertKubeService(s)
+}
+
+func (a *ServerWithRoles) GetKubeServices() ([]services.Server, error) {
+	if err := a.action(defaults.Namespace, services.KindKubeService, services.VerbList); err != nil {
+		return nil, trace.Wrap(err)
+	}
+	if err := a.action(defaults.Namespace, services.KindKubeService, services.VerbRead); err != nil {
+		return nil, trace.Wrap(err)
+	}
+	return a.authServer.GetKubeServices()
+}
+
 // NewAdminAuthServer returns auth server authorized as admin,
 // used for auth server cached access
 func NewAdminAuthServer(authServer *Server, sessions session.Service, alog events.IAuditLog) (ClientI, error) {

Diff for: lib/auth/clt.go

@@ -2920,6 +2920,42 @@ func (c *Client) DeleteSemaphore(ctx context.Context, filter services.SemaphoreF
 	return nil
 }
 
+// UpsertKubeService is used by kubernetes services to report their presence
+// to other auth servers in form of hearbeat expiring after ttl period.
+func (c *Client) UpsertKubeService(s services.Server) error {
+	data, err := services.GetServerMarshaler().MarshalServer(s)
+	if err != nil {
+		return trace.Wrap(err)
+	}
+	args := &upsertServerRawReq{
+		Server: data,
+	}
+	_, err = c.PostJSON(c.Endpoint("kube_services"), args)
+	return trace.Wrap(err)
+}
+
+// GetKubeServices returns the list of kubernetes services registered in the
+// cluster.
+func (c *Client) GetKubeServices() ([]services.Server, error) {
+	out, err := c.Get(c.Endpoint("kube_services"), url.Values{})
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	var items []json.RawMessage
+	if err := json.Unmarshal(out.Bytes(), &items); err != nil {
+		return nil, trace.Wrap(err)
+	}
+	re := make([]services.Server, len(items))
+	for i, raw := range items {
+		server, err := services.GetServerMarshaler().UnmarshalServer(raw, services.KindKubeService, services.SkipValidation())
+		if err != nil {
+			return nil, trace.Wrap(err)
+		}
+		re[i] = server
+	}
+	return re, nil
+}
+
 // WebService implements features used by Web UI clients
 type WebService interface {
 	// GetWebSessionInfo checks if a web sesion is valid, returns session id in case if

Diff for: lib/auth/permissions.go

@@ -416,6 +416,23 @@ func GetCheckerForBuiltinRole(clusterName string, clusterConfig services.Cluster
 					Rules:      []services.Rule{},
 				},
 			})
+	case teleport.RoleKube:
+		return services.FromSpec(
+			role.String(),
+			services.RoleSpecV3{
+				Allow: services.RoleConditions{
+					Namespaces: []string{services.Wildcard},
+					Rules: []services.Rule{
+						services.NewRule(services.KindKubeService, services.RW()),
+						services.NewRule(services.KindEvent, services.RW()),
+						services.NewRule(services.KindCertAuthority, services.ReadNoSecrets()),
+						services.NewRule(services.KindClusterConfig, services.RO()),
+						services.NewRule(services.KindUser, services.RO()),
+						services.NewRule(services.KindRole, services.RO()),
+						services.NewRule(services.KindNamespace, services.RO()),
+					},
+				},
+			})
 	}
 
 	return nil, trace.NotFound("%v is not reconginzed", role.String())

Diff for: lib/cache/cache.go

@@ -90,6 +90,20 @@ func ForNode(cfg Config) Config {
 	return cfg
 }
 
+// ForKubernetes sets up watch configuration for a kubernetes service.
+func ForKubernetes(cfg Config) Config {
+	cfg.Watches = []services.WatchKind{
+		{Kind: services.KindCertAuthority, LoadSecrets: false},
+		{Kind: services.KindClusterName},
+		{Kind: services.KindClusterConfig},
+		{Kind: services.KindUser},
+		{Kind: services.KindRole},
+		{Kind: services.KindNamespace, Name: defaults.Namespace},
+	}
+	cfg.QueueSize = defaults.KubernetesQueueSize
+	return cfg
+}
+
 // SetupConfigFn is a function that sets up configuration
 // for cache
 type SetupConfigFn func(c Config) Config
@@ -661,3 +675,8 @@ func (c *Cache) GetTunnelConnections(clusterName string, opts ...services.Marsha
 func (c *Cache) GetAllTunnelConnections(opts ...services.MarshalOption) (conns []services.TunnelConnection, err error) {
 	return c.presenceCache.GetAllTunnelConnections(opts...)
 }
+
+// GetKubeServices is a part of auth.AccessPoint implementation
+func (c *Cache) GetKubeServices() ([]services.Server, error) {
+	return c.presenceCache.GetKubeServices()
+}

Diff for: lib/config/configuration.go

@@ -160,16 +160,16 @@ func ApplyFileConfig(fc *FileConfig, cfg *service.Config) error {
 		return nil
 	}
 	// merge file-based config with defaults in 'cfg'
-	if fc.Auth.Disabled() {
+	if fc.Auth.Disabled(true) {
 		cfg.Auth.Enabled = false
 	}
-	if fc.SSH.Disabled() {
+	if fc.SSH.Disabled(true) {
 		cfg.SSH.Enabled = false
 	}
-	if fc.Proxy.Disabled() {
+	if fc.Proxy.Disabled(true) {
 		cfg.Proxy.Enabled = false
 	}
-	if fc.Kube.Enabled() {
+	if fc.Kube.Enabled(false) {
 		cfg.Kube.Enabled = true
 	}
 	applyString(fc.NodeName, &cfg.Hostname)
@@ -323,25 +323,25 @@ func ApplyFileConfig(fc *FileConfig, cfg *service.Config) error {
 
 	// Apply configuration for "auth_service", "proxy_service", and
 	// "ssh_service" if it's enabled.
-	if fc.Auth.Enabled() {
+	if fc.Auth.Enabled(true) {
 		err = applyAuthConfig(fc, cfg)
 		if err != nil {
 			return trace.Wrap(err)
 		}
 	}
-	if fc.Proxy.Enabled() {
+	if fc.Proxy.Enabled(true) {
 		err = applyProxyConfig(fc, cfg)
 		if err != nil {
 			return trace.Wrap(err)
 		}
 	}
-	if fc.SSH.Enabled() {
+	if fc.SSH.Enabled(true) {
 		err = applySSHConfig(fc, cfg)
 		if err != nil {
 			return trace.Wrap(err)
 		}
 	}
-	if fc.Kube.Enabled() {
+	if fc.Kube.Enabled(false) {
 		if err := applyKubeConfig(fc, cfg); err != nil {
 			return trace.Wrap(err)
 		}
@@ -540,7 +540,7 @@ func applyProxyConfig(fc *FileConfig, cfg *service.Config) error {
 
 	// apply kubernetes proxy config, by default kube proxy is disabled
 	if fc.Proxy.Kube.Configured() {
-		cfg.Proxy.Kube.Enabled = fc.Proxy.Kube.Enabled()
+		cfg.Proxy.Kube.Enabled = fc.Proxy.Kube.Enabled(false)
 	}
 	if fc.Proxy.Kube.KubeconfigFile != "" {
 		cfg.Proxy.Kube.KubeconfigPath = fc.Proxy.Kube.KubeconfigFile
@@ -653,8 +653,8 @@ func applySSHConfig(fc *FileConfig, cfg *service.Config) error {
 
 // applyKubeConfig applies file configuration for the "kubernetes_service" section.
 func applyKubeConfig(fc *FileConfig, cfg *service.Config) error {
-	if fc.Proxy.ListenAddress != "" {
-		addr, err := utils.ParseHostPortAddr(fc.Proxy.ListenAddress, int(defaults.SSHProxyListenPort))
+	if fc.Kube.ListenAddress != "" {
+		addr, err := utils.ParseHostPortAddr(fc.Kube.ListenAddress, int(defaults.SSHProxyListenPort))
 		if err != nil {
 			return trace.Wrap(err)
 		}