Support for dynamic kube_cluster resources (#16276)

This PR introduces Auth Server CRUD API for managing the lifecycle of dynamic `kube_cluster` resources.

Currently, `kube_cluster` resources cannot be manipulated without using the API, nor are they being listened to. It will be part of future developments.

# Future work
- The `kubernetes_service` will have to watch these `kube_cluster` resources.
- The discovery service will have to create these resources. 

Resource manipulation via `tctl` will be introduced later and is not part of the scope.

Part of #16135, #13376  
Related to  #12048

api/client/client.go

@@ -2476,6 +2476,63 @@ func (c *Client) DeleteAllApps(ctx context.Context) error {
 	return trail.FromGRPC(err)
 }
 
+// CreateKubernetesCluster creates a new kubernetes cluster resource.
+func (c *Client) CreateKubernetesCluster(ctx context.Context, cluster types.KubeCluster) error {
+	kubeClusterV3, ok := cluster.(*types.KubernetesClusterV3)
+	if !ok {
+		return trace.BadParameter("unsupported kubernetes cluster type %T", cluster)
+	}
+	_, err := c.grpc.CreateKubernetesCluster(ctx, kubeClusterV3, c.callOpts...)
+	return trail.FromGRPC(err)
+}
+
+// UpdateKubernetesCluster updates existing kubernetes cluster resource.
+func (c *Client) UpdateKubernetesCluster(ctx context.Context, cluster types.KubeCluster) error {
+	kubeClusterV3, ok := cluster.(*types.KubernetesClusterV3)
+	if !ok {
+		return trace.BadParameter("unsupported kubernetes cluster type %T", cluster)
+	}
+	_, err := c.grpc.UpdateKubernetesCluster(ctx, kubeClusterV3, c.callOpts...)
+	return trail.FromGRPC(err)
+}
+
+// GetKubernetesCluster returns the specified kubernetes resource.
+func (c *Client) GetKubernetesCluster(ctx context.Context, name string) (types.KubeCluster, error) {
+	if name == "" {
+		return nil, trace.BadParameter("missing kubernetes cluster name")
+	}
+	cluster, err := c.grpc.GetKubernetesCluster(ctx, &types.ResourceRequest{Name: name}, c.callOpts...)
+	if err != nil {
+		return nil, trail.FromGRPC(err)
+	}
+	return cluster, nil
+}
+
+// GetKubernetesClusters returns all kubernetes cluster resources.
+func (c *Client) GetKubernetesClusters(ctx context.Context) ([]types.KubeCluster, error) {
+	items, err := c.grpc.GetKubernetesClusters(ctx, &empty.Empty{}, c.callOpts...)
+	if err != nil {
+		return nil, trail.FromGRPC(err)
+	}
+	clusters := make([]types.KubeCluster, len(items.KubernetesClusters))
+	for i := range items.KubernetesClusters {
+		clusters[i] = items.KubernetesClusters[i]
+	}
+	return clusters, nil
+}
+
+// DeleteKubernetesCluster deletes specified kubernetes cluster resource.
+func (c *Client) DeleteKubernetesCluster(ctx context.Context, name string) error {
+	_, err := c.grpc.DeleteKubernetesCluster(ctx, &types.ResourceRequest{Name: name}, c.callOpts...)
+	return trail.FromGRPC(err)
+}
+
+// DeleteAllKubernetesClusters deletes all kubernetes cluster resources.
+func (c *Client) DeleteAllKubernetesClusters(ctx context.Context) error {
+	_, err := c.grpc.DeleteAllKubernetesClusters(ctx, &empty.Empty{}, c.callOpts...)
+	return trail.FromGRPC(err)
+}
+
 // CreateDatabase creates a new database resource.
 func (c *Client) CreateDatabase(ctx context.Context, database types.Database) error {
 	databaseV3, ok := database.(*types.DatabaseV3)

api/client/proto/joinservice.pb.go

@@ -27,6 +27,7 @@ var _ = math.Inf
 // proto package needs to be updated.
 const _ = proto.GoGoProtoPackageIsVersion3 // please upgrade the proto package
 
+// TODO(nklaassen): Document me.
 type RegisterUsingIAMMethodRequest struct {
 	// RegisterUsingTokenRequest holds registration parameters common to all
 	// join methods.

api/proto/teleport/legacy/client/proto/authservice.proto

@@ -2513,6 +2513,19 @@ service AuthService {
   // DeleteAllDatabases removes all database resources.
   rpc DeleteAllDatabases(google.protobuf.Empty) returns (google.protobuf.Empty);
 
+  // GetKubernetesClusters returns all registered kubernetes clusters.
+  rpc GetKubernetesClusters(google.protobuf.Empty) returns (types.KubernetesClusterV3List);
+  // GetKubernetesCluster returns a kubernetes cluster by name.
+  rpc GetKubernetesCluster(types.ResourceRequest) returns (types.KubernetesClusterV3);
+  // CreateKubernetesCluster creates a new kubernetes cluster resource.
+  rpc CreateKubernetesCluster(types.KubernetesClusterV3) returns (google.protobuf.Empty);
+  // UpdateKubernetesCluster updates existing kubernetes cluster resource.
+  rpc UpdateKubernetesCluster(types.KubernetesClusterV3) returns (google.protobuf.Empty);
+  // DeleteKubernetesCluster removes specified kubernetes cluster resource.
+  rpc DeleteKubernetesCluster(types.ResourceRequest) returns (google.protobuf.Empty);
+  // DeleteAllKubernetesClusters removes all kubernetes cluster resources.
+  rpc DeleteAllKubernetesClusters(google.protobuf.Empty) returns (google.protobuf.Empty);
+
   // GetWindowsDesktopServices returns all registered Windows desktop services.
   rpc GetWindowsDesktopServices(google.protobuf.Empty) returns (GetWindowsDesktopServicesResponse);
   // TODO(zmb3): Document me.

api/proto/teleport/legacy/types/events/events.proto

@@ -2952,6 +2952,9 @@ message OneOf {
     events.UpgradeWindowStartUpdate UpgradeWindowStartUpdate = 92;
     events.AppSessionEnd AppSessionEnd = 93;
     events.SessionRecordingAccess SessionRecordingAccess = 94;
+    events.KubernetesClusterCreate KubernetesClusterCreate = 96;
+    events.KubernetesClusterUpdate KubernetesClusterUpdate = 97;
+    events.KubernetesClusterDelete KubernetesClusterDelete = 98;
   }
 }
 
@@ -3670,3 +3673,90 @@ message SessionRecordingAccess {
     (gogoproto.jsontag) = ""
   ];
 }
+
+// KubeClusterMetadata contains common kubernetes cluster information.
+message KubeClusterMetadata {
+  // KubeLabels are the configured cluster labels.
+  map<string, string> KubeLabels = 1 [
+    (gogoproto.nullable) = false,
+    (gogoproto.jsontag) = "kube_labels,omitempty"
+  ];
+}
+
+// KubernetesClusterCreate is emitted when a new kubernetes cluster resource is created.
+message KubernetesClusterCreate {
+  // Metadata is a common event metadata.
+  Metadata Metadata = 1 [
+    (gogoproto.nullable) = false,
+    (gogoproto.embed) = true,
+    (gogoproto.jsontag) = ""
+  ];
+  // User is a common user event metadata.
+  UserMetadata User = 2 [
+    (gogoproto.nullable) = false,
+    (gogoproto.embed) = true,
+    (gogoproto.jsontag) = ""
+  ];
+  // ResourceMetadata is a common resource event metadata.
+  ResourceMetadata Resource = 3 [
+    (gogoproto.nullable) = false,
+    (gogoproto.embed) = true,
+    (gogoproto.jsontag) = ""
+  ];
+  // KubeClusterMetadata is a common kubernetes resource metadata.
+  KubeClusterMetadata KubeClusterMetadata = 4 [
+    (gogoproto.nullable) = false,
+    (gogoproto.embed) = true,
+    (gogoproto.jsontag) = ""
+  ];
+}
+
+// KubernetesClusterUpdate is emitted when an existing kubernetes cluster resource is updated.
+message KubernetesClusterUpdate {
+  // Metadata is a common event metadata.
+  Metadata Metadata = 1 [
+    (gogoproto.nullable) = false,
+    (gogoproto.embed) = true,
+    (gogoproto.jsontag) = ""
+  ];
+  // User is a common user event metadata.
+  UserMetadata User = 2 [
+    (gogoproto.nullable) = false,
+    (gogoproto.embed) = true,
+    (gogoproto.jsontag) = ""
+  ];
+  // ResourceMetadata is a common resource event metadata.
+  ResourceMetadata Resource = 3 [
+    (gogoproto.nullable) = false,
+    (gogoproto.embed) = true,
+    (gogoproto.jsontag) = ""
+  ];
+  // KubeClusterMetadata is a common kubernetes resource metadata.
+  KubeClusterMetadata KubeClusterMetadata = 4 [
+    (gogoproto.nullable) = false,
+    (gogoproto.embed) = true,
+    (gogoproto.jsontag) = ""
+  ];
+}
+
+// KubernetesClusterDelete is emitted when a kubernetes cluster resource is deleted.
+message KubernetesClusterDelete {
+  // Metadata is a common event metadata.
+  Metadata Metadata = 1 [
+    (gogoproto.nullable) = false,
+    (gogoproto.embed) = true,
+    (gogoproto.jsontag) = ""
+  ];
+  // User is a common user event metadata.
+  UserMetadata User = 2 [
+    (gogoproto.nullable) = false,
+    (gogoproto.embed) = true,
+    (gogoproto.jsontag) = ""
+  ];
+  // ResourceMetadata is a common resource event metadata.
+  ResourceMetadata Resource = 3 [
+    (gogoproto.nullable) = false,
+    (gogoproto.embed) = true,
+    (gogoproto.jsontag) = ""
+  ];
+}

api/proto/teleport/legacy/types/types.proto

@@ -2907,6 +2907,12 @@ message KubernetesClusterSpecV3 {
   ];
 }
 
+// KubernetesClusterV3List represents a list of kubernetes clusters.
+message KubernetesClusterV3List {
+  // KubernetesClusters is a list of kubernetes clusters resources.
+  repeated KubernetesClusterV3 KubernetesClusters = 1;
+}
+
 // KubernetesServerV3 represents a Kubernetes server.
 message KubernetesServerV3 {
   option (gogoproto.goproto_stringer) = false;

api/types/events/oneof.go

@@ -415,6 +415,18 @@ func ToOneOf(in AuditEvent) (*OneOf, error) {
 		out.Event = &OneOf_Unknown{
 			Unknown: e,
 		}
+	case *KubernetesClusterCreate:
+		out.Event = &OneOf_KubernetesClusterCreate{
+			KubernetesClusterCreate: e,
+		}
+	case *KubernetesClusterUpdate:
+		out.Event = &OneOf_KubernetesClusterUpdate{
+			KubernetesClusterUpdate: e,
+		}
+	case *KubernetesClusterDelete:
+		out.Event = &OneOf_KubernetesClusterDelete{
+			KubernetesClusterDelete: e,
+		}
 	default:
 		log.Errorf("Attempted to convert dynamic event of unknown type \"%v\" into protobuf event.", in.GetType())
 		unknown := &Unknown{}

api/types/kubernetes.go

@@ -130,22 +130,22 @@ func (k *KubernetesClusterV3) SetOrigin(origin string) {
 	k.Metadata.SetOrigin(origin)
 }
 
-// GetNamespace returns the app resource namespace.
+// GetNamespace returns the kube resource namespace.
 func (k *KubernetesClusterV3) GetNamespace() string {
 	return k.Metadata.Namespace
 }
 
-// SetExpiry sets the app resource expiration time.
+// SetExpiry sets the kube resource expiration time.
 func (k *KubernetesClusterV3) SetExpiry(expiry time.Time) {
 	k.Metadata.SetExpiry(expiry)
 }
 
-// Expiry returns the app resource expiration time.
+// Expiry returns the kube resource expiration time.
 func (k *KubernetesClusterV3) Expiry() time.Time {
 	return k.Metadata.Expiry()
 }
 
-// GetName returns the app resource name.
+// GetName returns the kube resource name.
 func (k *KubernetesClusterV3) GetName() string {
 	return k.Metadata.Name
 }
@@ -235,6 +235,25 @@ func (k *KubernetesClusterV3) CheckAndSetDefaults() error {
 // KubeClusters represents a list of kube clusters.
 type KubeClusters []KubeCluster
 
+// Find returns kube cluster with the specified name or nil.
+func (s KubeClusters) Find(name string) KubeCluster {
+	for _, cluster := range s {
+		if cluster.GetName() == name {
+			return cluster
+		}
+	}
+	return nil
+}
+
+// ToMap returns these kubernetes clusters as a map keyed by cluster name.
+func (s KubeClusters) ToMap() map[string]KubeCluster {
+	m := make(map[string]KubeCluster)
+	for _, kubeCluster := range s {
+		m[kubeCluster.GetName()] = kubeCluster
+	}
+	return m
+}
+
 // Len returns the slice length.
 func (s KubeClusters) Len() int { return len(s) }
 
@@ -266,8 +285,8 @@ func (s KubeClusters) SortByCustom(sortBy SortBy) error {
 }
 
 // AsResources returns as type resources with labels.
-func (s KubeClusters) AsResources() []ResourceWithLabels {
-	resources := make([]ResourceWithLabels, 0, len(s))
+func (s KubeClusters) AsResources() ResourcesWithLabels {
+	resources := make(ResourcesWithLabels, 0, len(s))
 	for _, cluster := range s {
 		resources = append(resources, ResourceWithLabels(cluster))
 	}

lib/auth/api.go

@@ -280,6 +280,11 @@ type ReadProxyAccessPoint interface {
 	GetWindowsDesktopServices(ctx context.Context) ([]types.WindowsDesktopService, error)
 	// GetWindowsDesktopService returns a windows desktop host by name.
 	GetWindowsDesktopService(ctx context.Context, name string) (types.WindowsDesktopService, error)
+
+	// GetKubernetesClusters returns all kubernetes cluster resources.
+	GetKubernetesClusters(ctx context.Context) ([]types.KubeCluster, error)
+	// GetKubernetesCluster returns the specified kubernetes cluster resource.
+	GetKubernetesCluster(ctx context.Context, name string) (types.KubeCluster, error)
 }
 
 // SnowflakeSessionWatcher is watcher interface used by Snowflake web session watcher.
@@ -453,6 +458,11 @@ type ReadKubernetesAccessPoint interface {
 
 	// GetKubernetesServers returns a list of kubernetes servers registered in the cluster
 	GetKubernetesServers(context.Context) ([]types.KubeServer, error)
+
+	// GetKubernetesClusters returns all kubernetes cluster resources.
+	GetKubernetesClusters(ctx context.Context) ([]types.KubeCluster, error)
+	// GetKubernetesCluster returns the specified kubernetes cluster resource.
+	GetKubernetesCluster(ctx context.Context, name string) (types.KubeCluster, error)
 }
 
 // KubernetesAccessPoint is an API interface implemented by a certificate authority (CA) to be
@@ -848,6 +858,11 @@ type Cache interface {
 
 	// GetInstallers gets all the installer resources.
 	GetInstallers(ctx context.Context) ([]types.Installer, error)
+
+	// GetKubernetesClusters returns all kubernetes cluster resources.
+	GetKubernetesClusters(ctx context.Context) ([]types.KubeCluster, error)
+	// GetKubernetesCluster returns the specified kubernetes cluster resource.
+	GetKubernetesCluster(ctx context.Context, name string) (types.KubeCluster, error)
 }
 
 type NodeWrapper struct {

lib/auth/auth.go

@@ -139,6 +139,9 @@ func NewServer(cfg *InitConfig, opts ...ServerOption) (*Server, error) {
 	if cfg.Databases == nil {
 		cfg.Databases = local.NewDatabasesService(cfg.Backend)
 	}
+	if cfg.Kubernetes == nil {
+		cfg.Kubernetes = local.NewKubernetesService(cfg.Backend)
+	}
 	if cfg.Status == nil {
 		cfg.Status = local.NewStatusService(cfg.Backend)
 	}
@@ -205,6 +208,7 @@ func NewServer(cfg *InitConfig, opts ...ServerOption) (*Server, error) {
 		ClusterConfiguration:  cfg.ClusterConfiguration,
 		Restrictions:          cfg.Restrictions,
 		Apps:                  cfg.Apps,
+		Kubernetes:            cfg.Kubernetes,
 		Databases:             cfg.Databases,
 		IAuditLog:             cfg.AuditLog,
 		Events:                cfg.Events,
@@ -259,6 +263,7 @@ type Services struct {
 	services.ClusterConfiguration
 	services.Restrictions
 	services.Apps
+	services.Kubernetes
 	services.Databases
 	services.WindowsDesktops
 	services.SessionTrackerService
@@ -3186,6 +3191,74 @@ func (a *Server) ListResources(ctx context.Context, req proto.ListResourcesReque
 	return a.Cache.ListResources(ctx, req)
 }
 
+// CreateKubernetesCluster creates a new kubernetes cluster resource.
+func (a *Server) CreateKubernetesCluster(ctx context.Context, kubeCluster types.KubeCluster) error {
+	if err := a.Services.CreateKubernetesCluster(ctx, kubeCluster); err != nil {
+		return trace.Wrap(err)
+	}
+	if err := a.emitter.EmitAuditEvent(ctx, &apievents.KubernetesClusterCreate{
+		Metadata: apievents.Metadata{
+			Type: events.KubernetesClusterCreateEvent,
+			Code: events.KubernetesClusterCreateCode,
+		},
+		UserMetadata: ClientUserMetadata(ctx),
+		ResourceMetadata: apievents.ResourceMetadata{
+			Name:    kubeCluster.GetName(),
+			Expires: kubeCluster.Expiry(),
+		},
+		KubeClusterMetadata: apievents.KubeClusterMetadata{
+			KubeLabels: kubeCluster.GetStaticLabels(),
+		},
+	}); err != nil {
+		log.WithError(err).Warn("Failed to emit kube cluster create event.")
+	}
+	return nil
+}
+
+// UpdateKubernetesCluster updates an existing kubernetes cluster resource.
+func (a *Server) UpdateKubernetesCluster(ctx context.Context, kubeCluster types.KubeCluster) error {
+	if err := a.Kubernetes.UpdateKubernetesCluster(ctx, kubeCluster); err != nil {
+		return trace.Wrap(err)
+	}
+	if err := a.emitter.EmitAuditEvent(ctx, &apievents.KubernetesClusterUpdate{
+		Metadata: apievents.Metadata{
+			Type: events.KubernetesClusterUpdateEvent,
+			Code: events.KubernetesClusterUpdateCode,
+		},
+		UserMetadata: ClientUserMetadata(ctx),
+		ResourceMetadata: apievents.ResourceMetadata{
+			Name:    kubeCluster.GetName(),
+			Expires: kubeCluster.Expiry(),
+		},
+		KubeClusterMetadata: apievents.KubeClusterMetadata{
+			KubeLabels: kubeCluster.GetStaticLabels(),
+		},
+	}); err != nil {
+		log.WithError(err).Warn("Failed to emit kube cluster update event.")
+	}
+	return nil
+}
+
+// DeleteKubernetesCluster deletes a kubernetes cluster resource.
+func (a *Server) DeleteKubernetesCluster(ctx context.Context, name string) error {
+	if err := a.Kubernetes.DeleteKubernetesCluster(ctx, name); err != nil {
+		return trace.Wrap(err)
+	}
+	if err := a.emitter.EmitAuditEvent(ctx, &apievents.KubernetesClusterDelete{
+		Metadata: apievents.Metadata{
+			Type: events.KubernetesClusterDeleteEvent,
+			Code: events.KubernetesClusterDeleteCode,
+		},
+		UserMetadata: ClientUserMetadata(ctx),
+		ResourceMetadata: apievents.ResourceMetadata{
+			Name: name,
+		},
+	}); err != nil {
+		log.WithError(err).Warn("Failed to emit kube cluster delete event.")
+	}
+	return nil
+}
+
 func (a *Server) isMFARequired(ctx context.Context, checker services.AccessChecker, req *proto.IsMFARequiredRequest) (*proto.IsMFARequiredResponse, error) {
 	pref, err := a.GetAuthPreference(ctx)
 	if err != nil {

lib/auth/auth_with_roles.go

@@ -4613,6 +4613,108 @@ func (a *ServerWithRoles) DeleteAllApps(ctx context.Context) error {
 	return nil
 }
 
+// CreateKubernetesCluster creates a new kubernetes cluster resource.
+func (a *ServerWithRoles) CreateKubernetesCluster(ctx context.Context, cluster types.KubeCluster) error {
+	if err := a.action(apidefaults.Namespace, types.KindKubernetesCluster, types.VerbCreate); err != nil {
+		return trace.Wrap(err)
+	}
+	// Don't allow users create clusters they wouldn't have access to (e.g.
+	// non-matching labels).
+	if err := a.checkAccessToKubeCluster(cluster); err != nil {
+		return trace.Wrap(err)
+	}
+	return trace.Wrap(a.authServer.CreateKubernetesCluster(ctx, cluster))
+}
+
+// UpdateKubernetesCluster updates existing kubernetes cluster resource.
+func (a *ServerWithRoles) UpdateKubernetesCluster(ctx context.Context, cluster types.KubeCluster) error {
+	if err := a.action(apidefaults.Namespace, types.KindKubernetesCluster, types.VerbUpdate); err != nil {
+		return trace.Wrap(err)
+	}
+	// Don't allow users update clusters they don't have access to (e.g.
+	// non-matching labels). Make sure to check existing cluster too.
+	existing, err := a.authServer.GetKubernetesCluster(ctx, cluster.GetName())
+	if err != nil {
+		return trace.Wrap(err)
+	}
+	if err := a.checkAccessToKubeCluster(existing); err != nil {
+		return trace.Wrap(err)
+	}
+	if err := a.checkAccessToKubeCluster(cluster); err != nil {
+		return trace.Wrap(err)
+	}
+	return trace.Wrap(a.authServer.UpdateKubernetesCluster(ctx, cluster))
+}
+
+// GetKubernetesCluster returns specified kubernetes cluster resource.
+func (a *ServerWithRoles) GetKubernetesCluster(ctx context.Context, name string) (types.KubeCluster, error) {
+	if err := a.action(apidefaults.Namespace, types.KindKubernetesCluster, types.VerbRead); err != nil {
+		return nil, trace.Wrap(err)
+	}
+	kubeCluster, err := a.authServer.GetKubernetesCluster(ctx, name)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	if err := a.checkAccessToKubeCluster(kubeCluster); err != nil {
+		return nil, trace.Wrap(err)
+	}
+	return kubeCluster, nil
+}
+
+// GetKubernetesClusters returns all kubernetes cluster resources.
+func (a *ServerWithRoles) GetKubernetesClusters(ctx context.Context) (result []types.KubeCluster, err error) {
+	if err := a.action(apidefaults.Namespace, types.KindKubernetesCluster, types.VerbList, types.VerbRead); err != nil {
+		return nil, trace.Wrap(err)
+	}
+	// Filter out kube clusters user doesn't have access to.
+	clusters, err := a.authServer.GetKubernetesClusters(ctx)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	for _, cluster := range clusters {
+		if err := a.checkAccessToKubeCluster(cluster); err == nil {
+			result = append(result, cluster)
+		}
+	}
+	return result, nil
+}
+
+// DeleteKubernetesCluster removes the specified kubernetes cluster resource.
+func (a *ServerWithRoles) DeleteKubernetesCluster(ctx context.Context, name string) error {
+	if err := a.action(apidefaults.Namespace, types.KindKubernetesCluster, types.VerbDelete); err != nil {
+		return trace.Wrap(err)
+	}
+	// Make sure user has access to the kubernetes cluster before deleting.
+	cluster, err := a.authServer.GetKubernetesCluster(ctx, name)
+	if err != nil {
+		return trace.Wrap(err)
+	}
+	if err := a.checkAccessToKubeCluster(cluster); err != nil {
+		return trace.Wrap(err)
+	}
+	return trace.Wrap(a.authServer.DeleteKubernetesCluster(ctx, name))
+}
+
+// DeleteAllKubernetesClusters removes all kubernetes cluster resources.
+func (a *ServerWithRoles) DeleteAllKubernetesClusters(ctx context.Context) error {
+	if err := a.action(apidefaults.Namespace, types.KindKubernetesCluster, types.VerbList, types.VerbDelete); err != nil {
+		return trace.Wrap(err)
+	}
+	// Make sure to only delete kubernetes cluster user has access to.
+	clusters, err := a.authServer.GetKubernetesClusters(ctx)
+	if err != nil {
+		return trace.Wrap(err)
+	}
+	for _, cluster := range clusters {
+		if err := a.checkAccessToKubeCluster(cluster); err == nil {
+			if err := a.authServer.DeleteKubernetesCluster(ctx, cluster.GetName()); err != nil {
+				return trace.Wrap(err)
+			}
+		}
+	}
+	return nil
+}
+
 func (a *ServerWithRoles) checkAccessToNode(node types.Server) error {
 	// For certain built-in roles, continue to allow full access and return
 	// the full set of nodes to not break existing clusters during migration.

lib/auth/clt.go

@@ -1602,6 +1602,7 @@ type ClientI interface {
 	services.Restrictions
 	services.Apps
 	services.Databases
+	services.Kubernetes
 	services.WindowsDesktops
 	WebService
 	services.Status

lib/auth/grpcserver.go

@@ -4369,6 +4369,97 @@ func (g *GRPCServer) AppendDiagnosticTrace(ctx context.Context, in *proto.Append
 	return connectionDiagnosticV1, nil
 }
 
+// GetKubernetesCluster returns the specified kubernetes cluster resource.
+func (g *GRPCServer) GetKubernetesCluster(ctx context.Context, req *types.ResourceRequest) (*types.KubernetesClusterV3, error) {
+	auth, err := g.authenticate(ctx)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	kubeCluster, err := auth.GetKubernetesCluster(ctx, req.Name)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	kubeClusterV3, ok := kubeCluster.(*types.KubernetesClusterV3)
+	if !ok {
+		return nil, trace.BadParameter("unsupported kubernetes cluster type %T", kubeCluster)
+	}
+	return kubeClusterV3, nil
+}
+
+// CreateKubernetesCluster creates a new kubernetes cluster resource.
+func (g *GRPCServer) CreateKubernetesCluster(ctx context.Context, cluster *types.KubernetesClusterV3) (*empty.Empty, error) {
+	auth, err := g.authenticate(ctx)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	cluster.SetOrigin(types.OriginDynamic)
+	if err := auth.CreateKubernetesCluster(ctx, cluster); err != nil {
+		return nil, trace.Wrap(err)
+	}
+	return &empty.Empty{}, nil
+}
+
+// UpdateKubernetesCluster updates existing kubernetes cluster resource.
+func (g *GRPCServer) UpdateKubernetesCluster(ctx context.Context, cluster *types.KubernetesClusterV3) (*empty.Empty, error) {
+	auth, err := g.authenticate(ctx)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	cluster.SetOrigin(types.OriginDynamic)
+	if err := auth.UpdateKubernetesCluster(ctx, cluster); err != nil {
+		return nil, trace.Wrap(err)
+	}
+	return &empty.Empty{}, nil
+}
+
+// GetKubernetesClusters returns all kubernetes cluster resources.
+func (g *GRPCServer) GetKubernetesClusters(ctx context.Context, _ *empty.Empty) (*types.KubernetesClusterV3List, error) {
+	auth, err := g.authenticate(ctx)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	clusters, err := auth.GetKubernetesClusters(ctx)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+
+	kubeClusters := make([]*types.KubernetesClusterV3, 0, len(clusters))
+	for _, cluster := range clusters {
+		clusterV3, ok := cluster.(*types.KubernetesClusterV3)
+		if !ok {
+			return nil, trace.BadParameter("unsupported kube cluster type %T", cluster)
+		}
+		kubeClusters = append(kubeClusters, clusterV3)
+	}
+	return &types.KubernetesClusterV3List{
+		KubernetesClusters: kubeClusters,
+	}, nil
+}
+
+// DeleteKubernetesCluster removes the specified kubernetes cluster.
+func (g *GRPCServer) DeleteKubernetesCluster(ctx context.Context, req *types.ResourceRequest) (*empty.Empty, error) {
+	auth, err := g.authenticate(ctx)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	if err := auth.DeleteKubernetesCluster(ctx, req.Name); err != nil {
+		return nil, trace.Wrap(err)
+	}
+	return &empty.Empty{}, nil
+}
+
+// DeleteAllKubernetesClusters removes all kubernetes cluster.
+func (g *GRPCServer) DeleteAllKubernetesClusters(ctx context.Context, _ *empty.Empty) (*empty.Empty, error) {
+	auth, err := g.authenticate(ctx)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	if err := auth.DeleteAllKubernetesClusters(ctx); err != nil {
+		return nil, trace.Wrap(err)
+	}
+	return &empty.Empty{}, nil
+}
+
 // GRPCServerConfig specifies GRPC server configuration
 type GRPCServerConfig struct {
 	// APIConfig is GRPC server API configuration

lib/auth/init.go

@@ -186,6 +186,9 @@ type InitConfig struct {
 	// TraceClient is used to forward spans to the upstream telemetry collector
 	TraceClient otlptrace.Client
 
+	// Kubernetes is a service that manages kubernetes cluster resources.
+	Kubernetes services.Kubernetes
+
 	// AssertionReplayService is a service that mitigatates SSO assertion replay.
 	*local.AssertionReplayService
 }

lib/auth/permissions.go

@@ -638,6 +638,7 @@ func definitionForBuiltinRole(clusterName string, recConfig types.SessionRecordi
 						types.NewRule(types.KindRole, services.RO()),
 						types.NewRule(types.KindNamespace, services.RO()),
 						types.NewRule(types.KindLock, services.RO()),
+						types.NewRule(types.KindKubernetesCluster, services.RW()),
 					},
 				},
 			})

lib/cache/cache.go

@@ -105,6 +105,7 @@ func ForAuth(cfg Config) Config {
 		{Kind: types.KindWindowsDesktop},
 		{Kind: types.KindKubeServer},
 		{Kind: types.KindInstaller},
+		{Kind: types.KindKubernetesCluster},
 	}
 	cfg.QueueSize = defaults.AuthQueueSize
 	return cfg
@@ -143,6 +144,7 @@ func ForProxy(cfg Config) Config {
 		{Kind: types.KindWindowsDesktop},
 		{Kind: types.KindKubeServer},
 		{Kind: types.KindInstaller},
+		{Kind: types.KindKubernetesCluster},
 	}
 	cfg.QueueSize = defaults.ProxyQueueSize
 	return cfg
@@ -175,6 +177,7 @@ func ForRemoteProxy(cfg Config) Config {
 		{Kind: types.KindDatabase},
 		{Kind: types.KindKubeServer},
 		{Kind: types.KindInstaller},
+		{Kind: types.KindKubernetesCluster},
 	}
 	cfg.QueueSize = defaults.ProxyQueueSize
 	return cfg
@@ -254,6 +257,7 @@ func ForKubernetes(cfg Config) Config {
 		{Kind: types.KindNamespace, Name: apidefaults.Namespace},
 		{Kind: types.KindKubeService},
 		{Kind: types.KindKubeServer},
+		{Kind: types.KindKubernetesCluster},
 	}
 	cfg.QueueSize = defaults.KubernetesQueueSize
 	return cfg
@@ -387,6 +391,7 @@ type Cache struct {
 	presenceCache         services.Presence
 	restrictionsCache     services.Restrictions
 	appsCache             services.Apps
+	kubernetesCache       services.Kubernetes
 	databasesCache        services.Databases
 	appSessionCache       services.AppSession
 	snowflakeSessionCache services.SnowflakeSession
@@ -447,6 +452,7 @@ func (c *Cache) read() (readGuard, error) {
 			presence:         c.presenceCache,
 			restrictions:     c.restrictionsCache,
 			apps:             c.appsCache,
+			kubernetes:       c.kubernetesCache,
 			databases:        c.databasesCache,
 			appSession:       c.appSessionCache,
 			snowflakeSession: c.snowflakeSessionCache,
@@ -467,6 +473,7 @@ func (c *Cache) read() (readGuard, error) {
 		presence:         c.Config.Presence,
 		restrictions:     c.Config.Restrictions,
 		apps:             c.Config.Apps,
+		kubernetes:       c.Config.Kubernetes,
 		databases:        c.Config.Databases,
 		appSession:       c.Config.AppSession,
 		snowflakeSession: c.Config.SnowflakeSession,
@@ -493,6 +500,7 @@ type readGuard struct {
 	snowflakeSession services.SnowflakeSession
 	restrictions     services.Restrictions
 	apps             services.Apps
+	kubernetes       services.Kubernetes
 	databases        services.Databases
 	webSession       types.WebSessionInterface
 	webToken         types.WebTokenInterface
@@ -546,6 +554,8 @@ type Config struct {
 	Restrictions services.Restrictions
 	// Apps is an apps service.
 	Apps services.Apps
+	// Kubernetes is an kubernetes service.
+	Kubernetes services.Kubernetes
 	// Databases is a databases service.
 	Databases services.Databases
 	// SnowflakeSession holds Snowflake sessions.
@@ -701,6 +711,7 @@ func New(config Config) (*Cache, error) {
 		presenceCache:         local.NewPresenceService(config.Backend),
 		restrictionsCache:     local.NewRestrictionsService(config.Backend),
 		appsCache:             local.NewAppService(config.Backend),
+		kubernetesCache:       local.NewKubernetesService(config.Backend),
 		databasesCache:        local.NewDatabasesService(config.Backend),
 		appSessionCache:       local.NewIdentityService(config.Backend),
 		snowflakeSessionCache: local.NewIdentityService(config.Backend),
@@ -1880,6 +1891,32 @@ func (c *Cache) GetApplicationServers(ctx context.Context, namespace string) ([]
 	return rg.presence.GetApplicationServers(ctx, namespace)
 }
 
+// GetKubernetesClusters returns all kubernetes cluster resources.
+func (c *Cache) GetKubernetesClusters(ctx context.Context) ([]types.KubeCluster, error) {
+	ctx, span := c.Tracer.Start(ctx, "cache/GetKubernetesClusters")
+	defer span.End()
+
+	rg, err := c.read()
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	defer rg.Release()
+	return rg.kubernetes.GetKubernetesClusters(ctx)
+}
+
+// GetKubernetesCluster returns the specified kubernetes cluster resource.
+func (c *Cache) GetKubernetesCluster(ctx context.Context, name string) (types.KubeCluster, error) {
+	ctx, span := c.Tracer.Start(ctx, "cache/GetKubernetesCluster")
+	defer span.End()
+
+	rg, err := c.read()
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	defer rg.Release()
+	return rg.kubernetes.GetKubernetesCluster(ctx, name)
+}
+
 // GetApps returns all application resources.
 func (c *Cache) GetApps(ctx context.Context) ([]types.Application, error) {
 	ctx, span := c.Tracer.Start(ctx, "cache/GetApps")

lib/cache/cache_test.go

@@ -76,6 +76,7 @@ type testPack struct {
 	snowflakeSessionS services.SnowflakeSession
 	restrictions      services.Restrictions
 	apps              services.Apps
+	kubernetes        services.Kubernetes
 	databases         services.Databases
 	webSessionS       types.WebSessionInterface
 	webTokenS         types.WebTokenInterface
@@ -190,6 +191,7 @@ func newPackWithoutCache(dir string, opts ...packOption) (*testPack, error) {
 	p.webTokenS = local.NewIdentityService(p.backend).WebTokens()
 	p.restrictions = local.NewRestrictionsService(p.backend)
 	p.apps = local.NewAppService(p.backend)
+	p.kubernetes = local.NewKubernetesService(p.backend)
 	p.databases = local.NewDatabasesService(p.backend)
 	p.windowsDesktops = local.NewWindowsDesktopService(p.backend)
 
@@ -221,6 +223,7 @@ func newPack(dir string, setupConfig func(c Config) Config, opts ...packOption)
 		SnowflakeSession: p.snowflakeSessionS,
 		Restrictions:     p.restrictions,
 		Apps:             p.apps,
+		Kubernetes:       p.kubernetes,
 		Databases:        p.databases,
 		WindowsDesktops:  p.windowsDesktops,
 		MaxRetryPeriod:   200 * time.Millisecond,
@@ -232,8 +235,9 @@ func newPack(dir string, setupConfig func(c Config) Config, opts ...packOption)
 
 	select {
 	case event := <-p.eventsC:
+
 		if event.Type != WatcherStarted {
-			return nil, trace.CompareFailed("%q != %q", event.Type, WatcherStarted)
+			return nil, trace.CompareFailed("%q != %q %s", event.Type, WatcherStarted, event)
 		}
 	case <-time.After(time.Second):
 		return nil, trace.ConnectionProblem(nil, "wait for the watcher to start")
@@ -424,6 +428,7 @@ func TestNodeCAFiltering(t *testing.T) {
 		Presence:        p.cache.presenceCache,
 		Restrictions:    p.cache.restrictionsCache,
 		Apps:            p.cache.appsCache,
+		Kubernetes:      p.cache.kubernetesCache,
 		Databases:       p.cache.databasesCache,
 		AppSession:      p.cache.appSessionCache,
 		WebSession:      p.cache.webSessionCache,
@@ -597,6 +602,7 @@ func TestCompletenessInit(t *testing.T) {
 			WebToken:         p.webTokenS,
 			Restrictions:     p.restrictions,
 			Apps:             p.apps,
+			Kubernetes:       p.kubernetes,
 			Databases:        p.databases,
 			WindowsDesktops:  p.windowsDesktops,
 			MaxRetryPeriod:   200 * time.Millisecond,
@@ -658,6 +664,7 @@ func TestCompletenessReset(t *testing.T) {
 		WebToken:         p.webTokenS,
 		Restrictions:     p.restrictions,
 		Apps:             p.apps,
+		Kubernetes:       p.kubernetes,
 		Databases:        p.databases,
 		WindowsDesktops:  p.windowsDesktops,
 		MaxRetryPeriod:   200 * time.Millisecond,
@@ -831,6 +838,7 @@ func TestListResources_NodesTTLVariant(t *testing.T) {
 		SnowflakeSession: p.snowflakeSessionS,
 		Restrictions:     p.restrictions,
 		Apps:             p.apps,
+		Kubernetes:       p.kubernetes,
 		Databases:        p.databases,
 		WindowsDesktops:  p.windowsDesktops,
 		MaxRetryPeriod:   200 * time.Millisecond,
@@ -902,6 +910,7 @@ func initStrategy(t *testing.T) {
 		WebToken:         p.webTokenS,
 		Restrictions:     p.restrictions,
 		Apps:             p.apps,
+		Kubernetes:       p.kubernetes,
 		Databases:        p.databases,
 		WindowsDesktops:  p.windowsDesktops,
 		MaxRetryPeriod:   200 * time.Millisecond,
@@ -1956,6 +1965,89 @@ func TestAppServers(t *testing.T) {
 	require.Empty(t, out)
 }
 
+// TestKubernetes tests that CRUD operations on kubernetes clusters resources are
+// replicated from the backend to the cache.
+func TestKubernetes(t *testing.T) {
+	t.Parallel()
+	p, err := newPack(t.TempDir(), ForProxy)
+	require.NoError(t, err)
+	t.Cleanup(p.Close)
+
+	ctx := context.Background()
+
+	// Create an cluster.
+	cluster, err := types.NewKubernetesClusterV3(types.Metadata{
+		Name: "foo",
+	}, types.KubernetesClusterSpecV3{})
+	require.NoError(t, err)
+
+	err = p.kubernetes.CreateKubernetesCluster(ctx, cluster)
+	require.NoError(t, err)
+
+	// Check that the cluster is now in the backend.
+	out, err := p.kubernetes.GetKubernetesClusters(ctx)
+	require.NoError(t, err)
+	require.Empty(t, cmp.Diff([]types.KubeCluster{cluster}, out,
+		cmpopts.IgnoreFields(types.Metadata{}, "ID")))
+
+	// Wait until the information has been replicated to the cache.
+	select {
+	case event := <-p.eventsC:
+		require.Equal(t, EventProcessed, event.Type)
+	case <-time.After(time.Second):
+		t.Fatal("timeout waiting for event")
+	}
+
+	// Make sure the cache has a single cluster in it.
+	out, err = p.kubernetes.GetKubernetesClusters(ctx)
+	require.NoError(t, err)
+	require.Empty(t, cmp.Diff([]types.KubeCluster{cluster}, out,
+		cmpopts.IgnoreFields(types.Metadata{}, "ID")))
+
+	// Update the cluster and upsert it into the backend again.
+	cluster.SetExpiry(time.Now().Add(30 * time.Minute).UTC())
+	err = p.kubernetes.UpdateKubernetesCluster(ctx, cluster)
+	require.NoError(t, err)
+
+	// Check that the cluster is in the backend and only one exists (so an
+	// update occurred).
+	out, err = p.kubernetes.GetKubernetesClusters(ctx)
+	require.NoError(t, err)
+	require.Empty(t, cmp.Diff([]types.KubeCluster{cluster}, out,
+		cmpopts.IgnoreFields(types.Metadata{}, "ID")))
+
+	// Check that information has been replicated to the cache.
+	select {
+	case event := <-p.eventsC:
+		require.Equal(t, EventProcessed, event.Type)
+	case <-time.After(time.Second):
+		t.Fatal("timeout waiting for event")
+	}
+
+	// Make sure the cache has a single cluster in it.
+	out, err = p.cache.GetKubernetesClusters(ctx)
+	require.NoError(t, err)
+	require.Empty(t, cmp.Diff([]types.KubeCluster{cluster}, out,
+		cmpopts.IgnoreFields(types.Metadata{}, "ID")))
+
+	// Remove all clusters from the backend.
+	err = p.kubernetes.DeleteAllKubernetesClusters(ctx)
+	require.NoError(t, err)
+
+	// Check that information has been replicated to the cache.
+	select {
+	case event := <-p.eventsC:
+		require.Equal(t, EventProcessed, event.Type)
+	case <-time.After(time.Second):
+		t.Fatal("timeout waiting for event")
+	}
+
+	// Check that the cache is now empty.
+	out, err = p.kubernetes.GetKubernetesClusters(ctx)
+	require.NoError(t, err)
+	require.Equal(t, 0, len(out))
+}
+
 // TestApplicationServers tests that CRUD operations on app servers are
 // replicated from the backend to the cache.
 func TestApplicationServers(t *testing.T) {
@@ -2608,6 +2700,7 @@ func TestCacheWatchKindExistsInEvents(t *testing.T) {
 		types.KindWindowsDesktopService:   &types.WindowsDesktopServiceV3{},
 		types.KindWindowsDesktop:          &types.WindowsDesktopV3{},
 		types.KindInstaller:               &types.InstallerV1{},
+		types.KindKubernetesCluster:       &types.KubernetesClusterV3{},
 	}
 
 	for name, cfg := range cases {

lib/cache/collections.go

@@ -201,6 +201,11 @@ func setupCollections(c *Cache, watches []types.WatchKind) (map[resourceKind]col
 				return nil, trace.BadParameter("missing parameter Databases")
 			}
 			collections[resourceKind] = &database{watch: watch, Cache: c}
+		case types.KindKubernetesCluster:
+			if c.Kubernetes == nil {
+				return nil, trace.BadParameter("missing parameter Kubernetes")
+			}
+			collections[resourceKind] = &kubeCluster{watch: watch, Cache: c}
 		case types.KindNetworkRestrictions:
 			if c.Restrictions == nil {
 				return nil, trace.BadParameter("missing parameter Restrictions")
@@ -2554,3 +2559,74 @@ func (c *windowsDesktops) processEvent(ctx context.Context, event types.Event) e
 func (c *windowsDesktops) watchKind() types.WatchKind {
 	return c.watch
 }
+
+type kubeCluster struct {
+	*Cache
+	watch types.WatchKind
+}
+
+func (s *kubeCluster) erase(ctx context.Context) error {
+	err := s.kubernetesCache.DeleteAllKubernetesClusters(ctx)
+	if err != nil && !trace.IsNotFound(err) {
+		return trace.Wrap(err)
+	}
+	return nil
+}
+
+func (s *kubeCluster) fetch(ctx context.Context) (apply func(ctx context.Context) error, err error) {
+	resources, err := s.Kubernetes.GetKubernetesClusters(ctx)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	return func(ctx context.Context) error {
+		if err := s.erase(ctx); err != nil {
+			return trace.Wrap(err)
+		}
+		for _, resource := range resources {
+			if err := s.kubernetesCache.CreateKubernetesCluster(ctx, resource); err != nil {
+				if !trace.IsAlreadyExists(err) {
+					return trace.Wrap(err)
+				}
+				if err := s.kubernetesCache.UpdateKubernetesCluster(ctx, resource); err != nil {
+					return trace.Wrap(err)
+				}
+			}
+		}
+		return nil
+	}, nil
+}
+
+func (s *kubeCluster) processEvent(ctx context.Context, event types.Event) error {
+	switch event.Type {
+	case types.OpDelete:
+		err := s.kubernetesCache.DeleteKubernetesCluster(ctx, event.Resource.GetName())
+		if err != nil {
+			// Resource could be missing in the cache expired or not created,
+			// if the first consumed event is delete.
+			if !trace.IsNotFound(err) {
+				s.WithError(err).Warn("Failed to delete resource.")
+				return trace.Wrap(err)
+			}
+		}
+	case types.OpPut:
+		resource, ok := event.Resource.(types.KubeCluster)
+		if !ok {
+			return trace.BadParameter("unexpected type %T", event.Resource)
+		}
+		if err := s.kubernetesCache.CreateKubernetesCluster(ctx, resource); err != nil {
+			if !trace.IsAlreadyExists(err) {
+				return trace.Wrap(err)
+			}
+			if err := s.kubernetesCache.UpdateKubernetesCluster(ctx, resource); err != nil {
+				return trace.Wrap(err)
+			}
+		}
+	default:
+		s.Warnf("Skipping unsupported event type %v.", event.Type)
+	}
+	return nil
+}
+
+func (s *kubeCluster) watchKind() types.WatchKind {
+	return s.watch
+}

lib/events/api.go

@@ -499,6 +499,13 @@ const (
 	// request.
 	KubeRequestEvent = "kube.request"
 
+	// KubernetesClusterCreateEvent is emitted when a kubernetes cluster resource is created.
+	KubernetesClusterCreateEvent = "kube.create"
+	// KubernetesClusterUpdateEvent is emitted when a kubernetes cluster resource is updated.
+	KubernetesClusterUpdateEvent = "kube.update"
+	// KubernetesClusterDeleteEvent is emitted when a kubernetes cluster resource is deleted.
+	KubernetesClusterDeleteEvent = "kube.delete"
+
 	// MFADeviceAddEvent is an event type for users adding MFA devices.
 	MFADeviceAddEvent = "mfa.add"
 	// MFADeviceDeleteEvent is an event type for users deleting MFA devices.

lib/events/codes.go

@@ -225,6 +225,13 @@ const (
 	// ExecCode).
 	KubeRequestCode = "T3009I"
 
+	// KubernetesClusterCreateCode is the kube.create event code.
+	KubernetesClusterCreateCode = "T3010I"
+	// KubernetesClusterUpdateCode is the kube.update event code.
+	KubernetesClusterUpdateCode = "T3011I"
+	// KubernetesClusterDeleteCode is the kube.delete event code.
+	KubernetesClusterDeleteCode = "T3012I"
+
 	// The following codes correspond to SFTP file operations.
 	SFTPOpenCode            = "TS001I"
 	SFTPOpenFailureCode     = "TS001E"

lib/service/service.go

@@ -1913,6 +1913,7 @@ func (process *TeleportProcess) newAccessCache(cfg accessCacheConfig) (*cache.Ca
 		Presence:         cfg.services,
 		Restrictions:     cfg.services,
 		Apps:             cfg.services,
+		Kubernetes:       cfg.services,
 		Databases:        cfg.services,
 		AppSession:       cfg.services,
 		SnowflakeSession: cfg.services,

lib/services/kubernetes.go

@@ -17,12 +17,36 @@ limitations under the License.
 package services
 
 import (
+	"context"
+
 	"github.com/gravitational/teleport/api/types"
 	"github.com/gravitational/teleport/lib/utils"
 
 	"github.com/gravitational/trace"
 )
 
+// KubernetesGetter defines interface for fetching kubernetes cluster resources.
+type KubernetesGetter interface {
+	// GetKubernetesClusters returns all kubernetes cluster resources.
+	GetKubernetesClusters(context.Context) ([]types.KubeCluster, error)
+	// GetKubernetesCluster returns the specified kubernetes cluster resource.
+	GetKubernetesCluster(ctx context.Context, name string) (types.KubeCluster, error)
+}
+
+// Kubernetes defines an interface for managing kubernetes clusters resources.
+type Kubernetes interface {
+	// KubernetesGetter provides methods for fetching kubernetes resources.
+	KubernetesGetter
+	// CreateKubernetesCluster creates a new kubernetes cluster resource.
+	CreateKubernetesCluster(context.Context, types.KubeCluster) error
+	// UpdateKubernetesCluster updates an existing kubernetes cluster resource.
+	UpdateKubernetesCluster(context.Context, types.KubeCluster) error
+	// DeleteKubernetesCluster removes the specified kubernetes cluster resource.
+	DeleteKubernetesCluster(ctx context.Context, name string) error
+	// DeleteAllKubernetesClusters removes all kubernetes resources.
+	DeleteAllKubernetesClusters(context.Context) error
+}
+
 // MarshalKubeServer marshals the KubeServer resource to JSON.
 func MarshalKubeServer(kubeServer types.KubeServer, opts ...MarshalOption) ([]byte, error) {
 	if err := kubeServer.CheckAndSetDefaults(); err != nil {
@@ -79,3 +103,60 @@ func UnmarshalKubeServer(data []byte, opts ...MarshalOption) (types.KubeServer,
 	}
 	return nil, trace.BadParameter("unsupported kube server resource version %q", h.Version)
 }
+
+// MarshalKubeCluster marshals the KubeCluster resource to JSON.
+func MarshalKubeCluster(kubeCluster types.KubeCluster, opts ...MarshalOption) ([]byte, error) {
+	if err := kubeCluster.CheckAndSetDefaults(); err != nil {
+		return nil, trace.Wrap(err)
+	}
+
+	cfg, err := CollectOptions(opts)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+
+	switch cluster := kubeCluster.(type) {
+	case *types.KubernetesClusterV3:
+		if !cfg.PreserveResourceID {
+			copy := *cluster
+			copy.SetResourceID(0)
+			cluster = &copy
+		}
+		return utils.FastMarshal(cluster)
+	default:
+		return nil, trace.BadParameter("unsupported kube cluster resource %T", cluster)
+	}
+}
+
+// UnmarshalKubeCluster unmarshals KubeCluster resource from JSON.
+func UnmarshalKubeCluster(data []byte, opts ...MarshalOption) (types.KubeCluster, error) {
+	if len(data) == 0 {
+		return nil, trace.BadParameter("missing kube cluster data")
+	}
+	cfg, err := CollectOptions(opts)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	var h types.ResourceHeader
+	if err := utils.FastUnmarshal(data, &h); err != nil {
+		return nil, trace.Wrap(err)
+	}
+	switch h.Version {
+	case types.V3:
+		var s types.KubernetesClusterV3
+		if err := utils.FastUnmarshal(data, &s); err != nil {
+			return nil, trace.BadParameter(err.Error())
+		}
+		if err := s.CheckAndSetDefaults(); err != nil {
+			return nil, trace.Wrap(err)
+		}
+		if cfg.ID != 0 {
+			s.SetResourceID(cfg.ID)
+		}
+		if !cfg.Expires.IsZero() {
+			s.SetExpiry(cfg.Expires)
+		}
+		return &s, nil
+	}
+	return nil, trace.BadParameter("unsupported kube cluster resource version %q", h.Version)
+}

lib/services/local/events.go

@@ -138,6 +138,8 @@ func (e *EventsService) NewWatcher(ctx context.Context, watch types.Watch) (type
 			parser = newWindowsDesktopsParser()
 		case types.KindInstaller:
 			parser = newInstallerParser()
+		case types.KindKubernetesCluster:
+			parser = newKubeClusterParser()
 		default:
 			return nil, trace.BadParameter("watcher on object kind %q is not supported", kind.Kind)
 		}
@@ -1039,6 +1041,30 @@ func (p *databaseServerParser) parse(event backend.Event) (types.Resource, error
 	}
 }
 
+func newKubeClusterParser() *kubeClusterParser {
+	return &kubeClusterParser{
+		baseParser: newBaseParser(backend.Key(kubernetesPrefix)),
+	}
+}
+
+type kubeClusterParser struct {
+	baseParser
+}
+
+func (p *kubeClusterParser) parse(event backend.Event) (types.Resource, error) {
+	switch event.Type {
+	case types.OpDelete:
+		return resourceHeader(event, types.KindKubernetesCluster, types.V3, 0)
+	case types.OpPut:
+		return services.UnmarshalKubeCluster(event.Item.Value,
+			services.WithResourceID(event.Item.ID),
+			services.WithExpires(event.Item.Expires),
+		)
+	default:
+		return nil, trace.BadParameter("event %v is not supported", event.Type)
+	}
+}
+
 func newAppParser() *appParser {
 	return &appParser{
 		baseParser: newBaseParser(backend.Key(appPrefix)),

lib/services/local/kube.go

@@ -0,0 +1,143 @@
+/*
+Copyright 2021 Gravitational, Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package local
+
+import (
+	"context"
+
+	"github.com/gravitational/teleport/api/types"
+	"github.com/gravitational/teleport/lib/backend"
+	"github.com/gravitational/teleport/lib/services"
+
+	"github.com/gravitational/trace"
+)
+
+// KubernetesService manages kubernetes resources in the backend.
+type KubernetesService struct {
+	backend.Backend
+}
+
+// NewKubernetesService creates a new KubernetesService.
+func NewKubernetesService(backend backend.Backend) *KubernetesService {
+	return &KubernetesService{Backend: backend}
+}
+
+// GetKubernetesClusters returns all kubernetes cluster resources.
+func (s *KubernetesService) GetKubernetesClusters(ctx context.Context) ([]types.KubeCluster, error) {
+	startKey := backend.Key(kubernetesPrefix)
+	result, err := s.GetRange(ctx, startKey, backend.RangeEnd(startKey), backend.NoLimit)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	kubeClusters := make([]types.KubeCluster, len(result.Items))
+	for i, item := range result.Items {
+		cluster, err := services.UnmarshalKubeCluster(item.Value,
+			services.WithResourceID(item.ID), services.WithExpires(item.Expires))
+		if err != nil {
+			return nil, trace.Wrap(err)
+		}
+		kubeClusters[i] = cluster
+	}
+	return kubeClusters, nil
+}
+
+// GetKubernetesCluster returns the specified kubernetes cluster resource.
+func (s *KubernetesService) GetKubernetesCluster(ctx context.Context, name string) (types.KubeCluster, error) {
+	item, err := s.Get(ctx, backend.Key(kubernetesPrefix, name))
+	if err != nil {
+		if trace.IsNotFound(err) {
+			return nil, trace.NotFound("kubernetes cluster %q doesn't exist", name)
+		}
+		return nil, trace.Wrap(err)
+	}
+	cluster, err := services.UnmarshalKubeCluster(item.Value,
+		services.WithResourceID(item.ID), services.WithExpires(item.Expires))
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	return cluster, nil
+}
+
+// CreateKubernetesCluster creates a new kubernetes cluster resource.
+func (s *KubernetesService) CreateKubernetesCluster(ctx context.Context, cluster types.KubeCluster) error {
+	if err := cluster.CheckAndSetDefaults(); err != nil {
+		return trace.Wrap(err)
+	}
+	value, err := services.MarshalKubeCluster(cluster)
+	if err != nil {
+		return trace.Wrap(err)
+	}
+	item := backend.Item{
+		Key:     backend.Key(kubernetesPrefix, cluster.GetName()),
+		Value:   value,
+		Expires: cluster.Expiry(),
+		ID:      cluster.GetResourceID(),
+	}
+	_, err = s.Create(ctx, item)
+	if err != nil {
+		return trace.Wrap(err)
+	}
+	return nil
+}
+
+// UpdateKubernetesCluster updates an existing kubernetes cluster resource.
+func (s *KubernetesService) UpdateKubernetesCluster(ctx context.Context, cluster types.KubeCluster) error {
+	if err := cluster.CheckAndSetDefaults(); err != nil {
+		return trace.Wrap(err)
+	}
+	value, err := services.MarshalKubeCluster(cluster)
+	if err != nil {
+		return trace.Wrap(err)
+	}
+	item := backend.Item{
+		Key:     backend.Key(kubernetesPrefix, cluster.GetName()),
+		Value:   value,
+		Expires: cluster.Expiry(),
+		ID:      cluster.GetResourceID(),
+	}
+	_, err = s.Update(ctx, item)
+	if err != nil {
+		return trace.Wrap(err)
+	}
+	return nil
+}
+
+// DeleteKubernetesCluster removes the specified kubernetes cluster resource.
+func (s *KubernetesService) DeleteKubernetesCluster(ctx context.Context, name string) error {
+	err := s.Delete(ctx, backend.Key(kubernetesPrefix, name))
+	if err != nil {
+		if trace.IsNotFound(err) {
+			return trace.NotFound("kubernetes cluster %q doesn't exist", name)
+		}
+		return trace.Wrap(err)
+	}
+	return nil
+}
+
+// DeleteAllKubernetesClusters removes all kubernetes cluster resources.
+func (s *KubernetesService) DeleteAllKubernetesClusters(ctx context.Context) error {
+	startKey := backend.Key(kubernetesPrefix)
+	err := s.DeleteRange(ctx, startKey, backend.RangeEnd(startKey))
+	if err != nil {
+		return trace.Wrap(err)
+	}
+	return nil
+}
+
+const (
+	kubernetesPrefix = "kubernetes"
+)

lib/services/local/kube_test.go

@@ -0,0 +1,117 @@
+/*
+Copyright 2021 Gravitational, Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package local
+
+import (
+	"context"
+	"testing"
+
+	"github.com/gravitational/teleport/api/types"
+	"github.com/gravitational/teleport/lib/backend/memory"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
+	"github.com/gravitational/trace"
+	"github.com/jonboulle/clockwork"
+	"github.com/stretchr/testify/require"
+)
+
+// TestKubernetesCRUD tests backend operations with kubernetes resources.
+func TestKubernetesCRUD(t *testing.T) {
+	ctx := context.Background()
+
+	backend, err := memory.New(memory.Config{
+		Context: ctx,
+		Clock:   clockwork.NewFakeClock(),
+	})
+	require.NoError(t, err)
+
+	service := NewKubernetesService(backend)
+
+	// Create a couple kube clusters.
+	kubeCluster1, err := types.NewKubernetesClusterV3(types.Metadata{
+		Name: "c1",
+	}, types.KubernetesClusterSpecV3{})
+	require.NoError(t, err)
+	kubeCluster2, err := types.NewKubernetesClusterV3(types.Metadata{
+		Name: "c2",
+	}, types.KubernetesClusterSpecV3{})
+	require.NoError(t, err)
+
+	// Initially we expect no Kubernetess.
+	out, err := service.GetKubernetesClusters(ctx)
+	require.NoError(t, err)
+	require.Equal(t, 0, len(out))
+
+	// Create both Kubernetess.
+	err = service.CreateKubernetesCluster(ctx, kubeCluster1)
+	require.NoError(t, err)
+	err = service.CreateKubernetesCluster(ctx, kubeCluster2)
+	require.NoError(t, err)
+
+	// Fetch all Kubernetess.
+	out, err = service.GetKubernetesClusters(ctx)
+	require.NoError(t, err)
+	require.Empty(t, cmp.Diff([]types.KubeCluster{kubeCluster1, kubeCluster2}, out,
+		cmpopts.IgnoreFields(types.Metadata{}, "ID"),
+	))
+
+	// Fetch a specific Kubernetes.
+	cluster, err := service.GetKubernetesCluster(ctx, kubeCluster2.GetName())
+	require.NoError(t, err)
+	require.Empty(t, cmp.Diff(kubeCluster2, cluster,
+		cmpopts.IgnoreFields(types.Metadata{}, "ID"),
+	))
+
+	// Try to fetch a Kubernetes that doesn't exist.
+	_, err = service.GetKubernetesCluster(ctx, "doesnotexist")
+	require.IsType(t, trace.NotFound(""), err)
+
+	// Try to create the same Kubernetes.
+	err = service.CreateKubernetesCluster(ctx, kubeCluster1)
+	require.IsType(t, trace.AlreadyExists(""), err)
+
+	// Update a Kubernetes.
+	kubeCluster1.Metadata.Description = "description"
+	err = service.UpdateKubernetesCluster(ctx, kubeCluster1)
+	require.NoError(t, err)
+	cluster, err = service.GetKubernetesCluster(ctx, kubeCluster1.GetName())
+	require.NoError(t, err)
+	require.Empty(t, cmp.Diff(kubeCluster1, cluster,
+		cmpopts.IgnoreFields(types.Metadata{}, "ID"),
+	))
+
+	// Delete a Kubernetes.
+	err = service.DeleteKubernetesCluster(ctx, kubeCluster1.GetName())
+	require.NoError(t, err)
+	out, err = service.GetKubernetesClusters(ctx)
+	require.NoError(t, err)
+	require.Empty(t, cmp.Diff([]types.KubeCluster{kubeCluster2}, out,
+		cmpopts.IgnoreFields(types.Metadata{}, "ID"),
+	))
+
+	// Try to delete a Kubernetes that doesn't exist.
+	err = service.DeleteKubernetesCluster(ctx, "doesnotexist")
+	require.IsType(t, trace.NotFound(""), err)
+
+	// Delete all Kubernetess.
+	err = service.DeleteAllKubernetesClusters(ctx)
+	require.NoError(t, err)
+	out, err = service.GetKubernetesClusters(ctx)
+	require.NoError(t, err)
+	require.Len(t, out, 0)
+}

lib/services/resource.go

@@ -153,6 +153,8 @@ func ParseShortcut(in string) (string, error) {
 		return types.KindRemoteCluster, nil
 	case types.KindSemaphore, "semaphores", "sem", "sems":
 		return types.KindSemaphore, nil
+	case types.KindKubernetesCluster, "kube_clusters":
+		return types.KindKubernetesCluster, nil
 	case types.KindKubeService, "kube_services":
 		return types.KindKubeService, nil
 	case types.KindKubeServer, "kube_servers":

lib/services/role.go

@@ -64,6 +64,7 @@ var DefaultImplicitRules = []types.Rule{
 	types.NewRule(types.KindApp, RO()),
 	types.NewRule(types.KindWindowsDesktopService, RO()),
 	types.NewRule(types.KindWindowsDesktop, RO()),
+	types.NewRule(types.KindKubernetesCluster, RO()),
 }
 
 // DefaultCertAuthorityRules provides access the minimal set of resources
@@ -151,6 +152,7 @@ func RoleForUser(u types.User) types.Role {
 				types.NewRule(types.KindLock, RW()),
 				types.NewRule(types.KindToken, RW()),
 				types.NewRule(types.KindConnectionDiagnostic, RW()),
+				types.NewRule(types.KindKubernetesCluster, RW()),
 			},
 			JoinSessions: []*types.SessionJoinPolicy{
 				{
@@ -414,15 +416,14 @@ func applyValueTraitsSlice(inputs []string, traits map[string][]string, fieldNam
 // and traits from identity provider. For example:
 //
 // cluster_labels:
-//   env: ['{{external.groups}}']
+//	env: ['{{external.groups}}']
 //
 // and groups: ['admins', 'devs']
 //
 // will be interpolated to:
 //
 // cluster_labels:
-//   env: ['admins', 'devs']
-//
+//	env: ['admins', 'devs']
 func applyLabelsTraits(inLabels types.Labels, traits map[string][]string) types.Labels {
 	outLabels := make(types.Labels, len(inLabels))
 	// every key will be mapped to the first value
@@ -558,7 +559,6 @@ func MakeRuleSet(rules []types.Rule) RuleSet {
 // Specifying order solves the problem on having multiple rules, e.g. one wildcard
 // rule can override more specific rules with 'where' sections that can have
 // 'actions' lists with side effects that will not be triggered otherwise.
-//
 func (set RuleSet) Match(whereParser predicate.Parser, actionsParser predicate.Parser, resource string, verb string) (bool, error) {
 	// empty set matches nothing
 	if len(set) == 0 {

lib/services/services.go

@@ -31,6 +31,7 @@ type Services interface {
 	Restrictions
 	Apps
 	Databases
+	Kubernetes
 	AppSession
 	SnowflakeSession
 	types.WebSessionsGetter