diff --git a/Makefile b/Makefile index 66d06c3c..052cba80 100644 --- a/Makefile +++ b/Makefile @@ -64,6 +64,24 @@ foreman-chart-crds: manifests ## Sync foreman.llmkube.dev CRDs to the foreman ch synced=$$((synced+1)); \ done; echo "Synced $$synced foreman CRD(s)" +.PHONY: federation-chart-crds +federation-chart-crds: manifests ## Sync federation.llmkube.dev CRDs into the main llmkube chart. +# Federation runs IN the main llmkube operator (config-gated by --federation-role), +# not a separate operator, so its CRD ships in the main chart alongside the +# inference CRDs. Mirrors scripts/sync-crds.sh: same destination and the same +# crds.install + crds.keep resource-policy wrapping the inference CRDs get. + @mkdir -p charts/llmkube/templates/crds + @synced=0; for src in config/crd/bases/federation.llmkube.dev_*.yaml; do \ + [ -e "$$src" ] || { echo "no federation CRDs in config/crd/bases (did make manifests run?)"; exit 1; }; \ + base=$$(basename $$src); short=$${base#federation.llmkube.dev_}; \ + echo "Syncing $$base -> $$short"; \ + { echo '{{- if .Values.crds.install }}'; \ + awk '/controller-gen.kubebuilder.io\/version:/{print; print " {{- if .Values.crds.keep }}"; print " helm.sh/resource-policy: keep"; print " {{- end }}"; next}1' "$$src"; \ + echo '{{- end }}'; \ + } > charts/llmkube/templates/crds/$$short; \ + synced=$$((synced+1)); \ + done; echo "Synced $$synced federation CRD(s)" + .PHONY: check-helm-rbac check-helm-rbac: manifests ## Verify the Helm charts' RBAC covers every kubebuilder-generated rule (#379). @./scripts/check-helm-rbac.sh diff --git a/PROJECT b/PROJECT index de8699ea..5b4db40b 100644 --- a/PROJECT +++ b/PROJECT @@ -26,4 +26,13 @@ resources: kind: InferenceService path: github.com/defilantech/llmkube/api/v1alpha1 version: v1alpha1 +- api: + crdVersion: v1 + namespaced: false + controller: true + domain: llmkube.dev + group: federation + kind: FederatedCluster + path: github.com/defilantech/llmkube/api/federation/v1alpha1 + version: v1alpha1 version: "3" diff --git a/api/federation/v1alpha1/federatedcluster_types.go b/api/federation/v1alpha1/federatedcluster_types.go new file mode 100644 index 00000000..6b587c13 --- /dev/null +++ b/api/federation/v1alpha1/federatedcluster_types.go @@ -0,0 +1,127 @@ +/* +Copyright 2025. + +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 v1alpha1 + +import ( + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +// Phase values for FederatedClusterStatus.Phase. Datacenter-owned: computed +// by the datacenter controller from LastHeartbeatTime staleness relative to +// a multiple of HeartbeatIntervalSeconds. +const ( + // FederatedClusterConnected means a heartbeat was received within the + // expected interval. + FederatedClusterConnected = "Connected" + + // FederatedClusterStale means no heartbeat was received for 3x the + // expected interval. + FederatedClusterStale = "Stale" + + // FederatedClusterUnreachable means no heartbeat was received for 10x + // the expected interval. + FederatedClusterUnreachable = "Unreachable" +) + +// FederatedClusterSpec is admin-owned. displayName and dataResidencyTier are +// set at registration; the edge and the datacenter controller only write status. +type FederatedClusterSpec struct { + // DisplayName is a human label for the site. + // +optional + DisplayName string `json:"displayName,omitempty"` + + // DataResidencyTier is recorded now and enforced later by the federation + // router (#1237). It is a free-form tier label (for example "eu", "floor-3"). + // +optional + DataResidencyTier string `json:"dataResidencyTier,omitempty"` + + // HeartbeatIntervalSeconds is how often the edge is expected to push status. + // The datacenter derives staleness thresholds from it (3x Stale, 10x Unreachable). + // +kubebuilder:default=30 + // +kubebuilder:validation:Minimum=5 + HeartbeatIntervalSeconds int32 `json:"heartbeatIntervalSeconds,omitempty"` +} + +// ClusterCapacity is edge-written node/GPU capacity. +type ClusterCapacity struct { + Nodes int32 `json:"nodes"` + GPUsTotal int32 `json:"gpusTotal"` + GPUsAllocatable int32 `json:"gpusAllocatable"` +} + +// ClusterInferenceSummary is edge-written inference health. +type ClusterInferenceSummary struct { + ServicesReady int32 `json:"servicesReady"` + ServicesFailed int32 `json:"servicesFailed"` + ServicesTotal int32 `json:"servicesTotal"` + Models int32 `json:"models"` +} + +// FederatedClusterStatus has two writers. The edge writes everything except +// Phase, over an RBAC-scoped status-subresource client. The datacenter +// controller writes ONLY Phase, from LastHeartbeatTime staleness. +type FederatedClusterStatus struct { + // Phase is datacenter-owned: Connected, Stale, or Unreachable. + // +optional + Phase string `json:"phase,omitempty"` + + // LastHeartbeatTime is edge-owned: set to now on every successful push. + // +optional + LastHeartbeatTime *metav1.Time `json:"lastHeartbeatTime,omitempty"` + + // ObservedVersion is edge-owned: the LLMKube/operator version on the site. + // +optional + ObservedVersion string `json:"observedVersion,omitempty"` + + // Capacity is edge-owned node/GPU capacity. + // +optional + Capacity *ClusterCapacity `json:"capacity,omitempty"` + + // Inference is edge-owned inference health. + // +optional + Inference *ClusterInferenceSummary `json:"inference,omitempty"` +} + +// +kubebuilder:object:root=true +// +kubebuilder:subresource:status +// +kubebuilder:resource:scope=Cluster,shortName=fedcluster;fc +// +kubebuilder:printcolumn:name="Tier",type=string,JSONPath=`.spec.dataResidencyTier` +// +kubebuilder:printcolumn:name="Phase",type=string,JSONPath=`.status.phase` +// +kubebuilder:printcolumn:name="Last Heartbeat",type=date,JSONPath=`.status.lastHeartbeatTime` +// +kubebuilder:printcolumn:name="GPUs",type=string,JSONPath=`.status.capacity.gpusAllocatable` + +// FederatedCluster registers one edge site on the datacenter cluster. +type FederatedCluster struct { + metav1.TypeMeta `json:",inline"` + metav1.ObjectMeta `json:"metadata,omitempty"` + + Spec FederatedClusterSpec `json:"spec,omitempty"` + Status FederatedClusterStatus `json:"status,omitempty"` +} + +// +kubebuilder:object:root=true + +// FederatedClusterList contains a list of FederatedCluster. +type FederatedClusterList struct { + metav1.TypeMeta `json:",inline"` + metav1.ListMeta `json:"metadata,omitempty"` + Items []FederatedCluster `json:"items"` +} + +func init() { + SchemeBuilder.Register(&FederatedCluster{}, &FederatedClusterList{}) +} diff --git a/api/federation/v1alpha1/groupversion_info.go b/api/federation/v1alpha1/groupversion_info.go new file mode 100644 index 00000000..842587f7 --- /dev/null +++ b/api/federation/v1alpha1/groupversion_info.go @@ -0,0 +1,47 @@ +/* +Copyright 2025. + +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 v1alpha1 contains API Schema definitions for the federation +// v1alpha1 API group. Federation registers remote edge-site clusters on a +// datacenter cluster (FederatedCluster) so fleet-wide inference status and +// routing can be reasoned about from one place. Installing LLMKube alone +// does not install or require any of these types. +// +// +kubebuilder:object:generate=true +// +groupName=federation.llmkube.dev +package v1alpha1 + +import ( + "k8s.io/apimachinery/pkg/runtime/schema" + "sigs.k8s.io/controller-runtime/pkg/scheme" +) + +var ( + // GroupVersion is group version used to register these objects. + GroupVersion = schema.GroupVersion{Group: "federation.llmkube.dev", Version: "v1alpha1"} + + // SchemeBuilder is used to add go types to the GroupVersionKind scheme. + // scheme.Builder was deprecated in controller-runtime v0.24 (the rationale + // is that api packages should have minimal deps, and scheme.Builder pulls + // controller-runtime into api/). It still works; migrating to the + // runtime.NewSchemeBuilder pattern is a project-wide refactor that touches + // both API groups and is tracked as its own follow-up. Suppress the + // deprecation here only. + SchemeBuilder = &scheme.Builder{GroupVersion: GroupVersion} //nolint:staticcheck // SA1019: see comment above + + // AddToScheme adds the types in this group-version to the given scheme. + AddToScheme = SchemeBuilder.AddToScheme +) diff --git a/api/federation/v1alpha1/zz_generated.deepcopy.go b/api/federation/v1alpha1/zz_generated.deepcopy.go new file mode 100644 index 00000000..2c5233fb --- /dev/null +++ b/api/federation/v1alpha1/zz_generated.deepcopy.go @@ -0,0 +1,158 @@ +//go:build !ignore_autogenerated + +/* +Copyright 2026. + +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. +*/ + +// Code generated by controller-gen. DO NOT EDIT. + +package v1alpha1 + +import ( + runtime "k8s.io/apimachinery/pkg/runtime" +) + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *ClusterCapacity) DeepCopyInto(out *ClusterCapacity) { + *out = *in +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ClusterCapacity. +func (in *ClusterCapacity) DeepCopy() *ClusterCapacity { + if in == nil { + return nil + } + out := new(ClusterCapacity) + in.DeepCopyInto(out) + return out +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *ClusterInferenceSummary) DeepCopyInto(out *ClusterInferenceSummary) { + *out = *in +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ClusterInferenceSummary. +func (in *ClusterInferenceSummary) DeepCopy() *ClusterInferenceSummary { + if in == nil { + return nil + } + out := new(ClusterInferenceSummary) + in.DeepCopyInto(out) + return out +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *FederatedCluster) DeepCopyInto(out *FederatedCluster) { + *out = *in + out.TypeMeta = in.TypeMeta + in.ObjectMeta.DeepCopyInto(&out.ObjectMeta) + out.Spec = in.Spec + in.Status.DeepCopyInto(&out.Status) +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new FederatedCluster. +func (in *FederatedCluster) DeepCopy() *FederatedCluster { + if in == nil { + return nil + } + out := new(FederatedCluster) + in.DeepCopyInto(out) + return out +} + +// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object. +func (in *FederatedCluster) DeepCopyObject() runtime.Object { + if c := in.DeepCopy(); c != nil { + return c + } + return nil +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *FederatedClusterList) DeepCopyInto(out *FederatedClusterList) { + *out = *in + out.TypeMeta = in.TypeMeta + in.ListMeta.DeepCopyInto(&out.ListMeta) + if in.Items != nil { + in, out := &in.Items, &out.Items + *out = make([]FederatedCluster, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new FederatedClusterList. +func (in *FederatedClusterList) DeepCopy() *FederatedClusterList { + if in == nil { + return nil + } + out := new(FederatedClusterList) + in.DeepCopyInto(out) + return out +} + +// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object. +func (in *FederatedClusterList) DeepCopyObject() runtime.Object { + if c := in.DeepCopy(); c != nil { + return c + } + return nil +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *FederatedClusterSpec) DeepCopyInto(out *FederatedClusterSpec) { + *out = *in +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new FederatedClusterSpec. +func (in *FederatedClusterSpec) DeepCopy() *FederatedClusterSpec { + if in == nil { + return nil + } + out := new(FederatedClusterSpec) + in.DeepCopyInto(out) + return out +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *FederatedClusterStatus) DeepCopyInto(out *FederatedClusterStatus) { + *out = *in + if in.LastHeartbeatTime != nil { + in, out := &in.LastHeartbeatTime, &out.LastHeartbeatTime + *out = (*in).DeepCopy() + } + if in.Capacity != nil { + in, out := &in.Capacity, &out.Capacity + *out = new(ClusterCapacity) + **out = **in + } + if in.Inference != nil { + in, out := &in.Inference, &out.Inference + *out = new(ClusterInferenceSummary) + **out = **in + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new FederatedClusterStatus. +func (in *FederatedClusterStatus) DeepCopy() *FederatedClusterStatus { + if in == nil { + return nil + } + out := new(FederatedClusterStatus) + in.DeepCopyInto(out) + return out +} diff --git a/charts/llmkube/templates/clusterrole.yaml b/charts/llmkube/templates/clusterrole.yaml index 4ef96f7b..47fbdb8e 100644 --- a/charts/llmkube/templates/clusterrole.yaml +++ b/charts/llmkube/templates/clusterrole.yaml @@ -46,6 +46,22 @@ rules: - get - list - watch +- apiGroups: + - federation.llmkube.dev + resources: + - federatedclusters + verbs: + - get + - list + - watch +- apiGroups: + - federation.llmkube.dev + resources: + - federatedclusters/status + verbs: + - get + - patch + - update - apiGroups: - "" resources: diff --git a/charts/llmkube/templates/crds/federatedclusters.yaml b/charts/llmkube/templates/crds/federatedclusters.yaml new file mode 100644 index 00000000..14b06339 --- /dev/null +++ b/charts/llmkube/templates/crds/federatedclusters.yaml @@ -0,0 +1,143 @@ +{{- if .Values.crds.install }} +--- +apiVersion: apiextensions.k8s.io/v1 +kind: CustomResourceDefinition +metadata: + annotations: + controller-gen.kubebuilder.io/version: v0.19.0 + {{- if .Values.crds.keep }} + helm.sh/resource-policy: keep + {{- end }} + name: federatedclusters.federation.llmkube.dev +spec: + group: federation.llmkube.dev + names: + kind: FederatedCluster + listKind: FederatedClusterList + plural: federatedclusters + shortNames: + - fedcluster + - fc + singular: federatedcluster + scope: Cluster + versions: + - additionalPrinterColumns: + - jsonPath: .spec.dataResidencyTier + name: Tier + type: string + - jsonPath: .status.phase + name: Phase + type: string + - jsonPath: .status.lastHeartbeatTime + name: Last Heartbeat + type: date + - jsonPath: .status.capacity.gpusAllocatable + name: GPUs + type: string + name: v1alpha1 + schema: + openAPIV3Schema: + description: FederatedCluster registers one edge site on the datacenter cluster. + properties: + apiVersion: + description: |- + APIVersion defines the versioned schema of this representation of an object. + Servers should convert recognized schemas to the latest internal value, and + may reject unrecognized values. + More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources + type: string + kind: + description: |- + Kind is a string value representing the REST resource this object represents. + Servers may infer this from the endpoint the client submits requests to. + Cannot be updated. + In CamelCase. + More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds + type: string + metadata: + type: object + spec: + description: |- + FederatedClusterSpec is admin-owned. displayName and dataResidencyTier are + set at registration; the edge and the datacenter controller only write status. + properties: + dataResidencyTier: + description: |- + DataResidencyTier is recorded now and enforced later by the federation + router (#1237). It is a free-form tier label (for example "eu", "floor-3"). + type: string + displayName: + description: DisplayName is a human label for the site. + type: string + heartbeatIntervalSeconds: + default: 30 + description: |- + HeartbeatIntervalSeconds is how often the edge is expected to push status. + The datacenter derives staleness thresholds from it (3x Stale, 10x Unreachable). + format: int32 + minimum: 5 + type: integer + type: object + status: + description: |- + FederatedClusterStatus has two writers. The edge writes everything except + Phase, over an RBAC-scoped status-subresource client. The datacenter + controller writes ONLY Phase, from LastHeartbeatTime staleness. + properties: + capacity: + description: Capacity is edge-owned node/GPU capacity. + properties: + gpusAllocatable: + format: int32 + type: integer + gpusTotal: + format: int32 + type: integer + nodes: + format: int32 + type: integer + required: + - gpusAllocatable + - gpusTotal + - nodes + type: object + inference: + description: Inference is edge-owned inference health. + properties: + models: + format: int32 + type: integer + servicesFailed: + format: int32 + type: integer + servicesReady: + format: int32 + type: integer + servicesTotal: + format: int32 + type: integer + required: + - models + - servicesFailed + - servicesReady + - servicesTotal + type: object + lastHeartbeatTime: + description: 'LastHeartbeatTime is edge-owned: set to now on every + successful push.' + format: date-time + type: string + observedVersion: + description: 'ObservedVersion is edge-owned: the LLMKube/operator + version on the site.' + type: string + phase: + description: 'Phase is datacenter-owned: Connected, Stale, or Unreachable.' + type: string + type: object + type: object + served: true + storage: true + subresources: + status: {} +{{- end }} diff --git a/cmd/main.go b/cmd/main.go index eacd0747..789cbc3b 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -39,13 +39,16 @@ import ( "k8s.io/apimachinery/pkg/runtime" utilruntime "k8s.io/apimachinery/pkg/util/runtime" clientgoscheme "k8s.io/client-go/kubernetes/scheme" + "k8s.io/client-go/tools/clientcmd" ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/healthz" "sigs.k8s.io/controller-runtime/pkg/log/zap" "sigs.k8s.io/controller-runtime/pkg/metrics/filters" metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server" "sigs.k8s.io/controller-runtime/pkg/webhook" + federationv1alpha1 "github.com/defilantech/llmkube/api/federation/v1alpha1" inferencev1alpha1 "github.com/defilantech/llmkube/api/v1alpha1" "github.com/defilantech/llmkube/internal/controller" _ "github.com/defilantech/llmkube/internal/metrics" // Register custom Prometheus metrics @@ -55,12 +58,17 @@ import ( var ( scheme = runtime.NewScheme() setupLog = ctrl.Log.WithName("setup") + // Version is the operator build version (set during build via ldflags, + // same convention as cmd/metal-agent). Recorded as + // status.observedVersion by FederationEdgeReconciler in --federation-role=edge. + Version = "dev" ) func init() { utilruntime.Must(clientgoscheme.AddToScheme(scheme)) utilruntime.Must(inferencev1alpha1.AddToScheme(scheme)) + utilruntime.Must(federationv1alpha1.AddToScheme(scheme)) // +kubebuilder:scaffold:scheme } @@ -186,6 +194,20 @@ func main() { flag.BoolVar(&enablePyrraSLO, "enable-pyrra-slo", false, "Enable rendering Pyrra ServiceLevelObjective resources for InferenceServices with spec.slo. "+ "Requires the Pyrra CRD in the cluster (https://github.com/pyrra-dev/pyrra).") + var federationRole string + var federationDatacenterKubeconfig string + var federationClusterName string + flag.StringVar(&federationRole, "federation-role", controller.FederationRoleHub, + "Federation role for this operator instance: \"hub\" (default) reconciles status.phase for every "+ + "registered edge from its own (datacenter) cluster; \"edge\" instead pushes this cluster's "+ + "observed inference state to a remote datacenter's FederatedCluster status subresource and "+ + "does not run the hub controller. A single operator instance is exactly one role.") + flag.StringVar(&federationDatacenterKubeconfig, "federation-datacenter-kubeconfig", "", + "Path to a kubeconfig for the datacenter cluster, scoped (RBAC) to get/patch this edge's own "+ + "FederatedCluster/status object. Required when --federation-role=edge; ignored in hub mode.") + flag.StringVar(&federationClusterName, "federation-cluster-name", "", + "Name of this edge's FederatedCluster object on the datacenter cluster. Required when "+ + "--federation-role=edge; ignored in hub mode.") flag.BoolVar(&enableLeaderElection, "leader-elect", false, "Enable leader election for controller manager. "+ "Enabling this will ensure there is only one active controller manager.") @@ -254,6 +276,47 @@ func main() { os.Exit(1) } + // Validate --federation-role up front: an unrecognized value is a + // startup error, not a silent fallback to hub, since that would run the + // wrong controller for wherever this operator instance actually lives. + if federationRole != controller.FederationRoleHub && federationRole != controller.FederationRoleEdge { + setupLog.Error(fmt.Errorf("invalid value %q", federationRole), + "invalid --federation-role: must be \"hub\" or \"edge\"") + os.Exit(1) + } + + // In edge mode, build a SEPARATE client for the datacenter cluster from + // the scoped kubeconfig. This client is used ONLY to patch this edge's + // own FederatedCluster/status object on the datacenter; the manager's + // own client (mgr.GetClient(), wired below) stays scoped to the local + // (edge) cluster for reading Models/InferenceServices/Nodes. The two + // are never the same client, so an edge operator cannot accidentally + // read or write anything else on the datacenter. + var federationDatacenterClient client.Client + if federationRole == controller.FederationRoleEdge { + if federationDatacenterKubeconfig == "" { + setupLog.Error(fmt.Errorf("missing --federation-datacenter-kubeconfig"), + "--federation-datacenter-kubeconfig is required when --federation-role=edge") + os.Exit(1) + } + if federationClusterName == "" { + setupLog.Error(fmt.Errorf("missing --federation-cluster-name"), + "--federation-cluster-name is required when --federation-role=edge") + os.Exit(1) + } + datacenterCfg, cfgErr := clientcmd.BuildConfigFromFlags("", federationDatacenterKubeconfig) + if cfgErr != nil { + setupLog.Error(cfgErr, "unable to load --federation-datacenter-kubeconfig", + "path", federationDatacenterKubeconfig) + os.Exit(1) + } + federationDatacenterClient, err = client.New(datacenterCfg, client.Options{Scheme: scheme}) + if err != nil { + setupLog.Error(err, "unable to build datacenter client from --federation-datacenter-kubeconfig") + os.Exit(1) + } + } + // Initialize OpenTelemetry tracing (noop if OTEL_EXPORTER_OTLP_ENDPOINT not set) shutdownTracer := initTracer(context.Background()) defer shutdownTracer() @@ -448,6 +511,37 @@ func main() { setupLog.Error(err, "unable to create controller", "controller", "GPUQuota") os.Exit(1) } + // Federation: exactly one of the two sides runs per operator instance, + // gated by --federation-role. + // - hub (default): FederatedClusterReconciler reconciles status.phase + // from edge-written heartbeat staleness, on the datacenter cluster. + // - edge: FederationEdgeReconciler instead runs as a manager.Runnable + // that pushes this cluster's observed inference state to the + // datacenter's FederatedCluster status subresource over the + // separately-scoped federationDatacenterClient built above. The hub + // controller does NOT run in edge mode, since an edge has no local + // FederatedCluster objects of its own to reconcile phase for. + switch federationRole { + case controller.FederationRoleEdge: + if err := mgr.Add(&controller.FederationEdgeReconciler{ + LocalClient: mgr.GetClient(), + DatacenterClient: federationDatacenterClient, + ClusterName: federationClusterName, + Version: Version, + Log: ctrl.Log.WithName("federation-edge"), + }); err != nil { + setupLog.Error(err, "unable to create runnable", "runnable", "FederationEdge") + os.Exit(1) + } + default: // controller.FederationRoleHub + if err := (&controller.FederatedClusterReconciler{ + Client: mgr.GetClient(), + Scheme: mgr.GetScheme(), + }).SetupWithManager(mgr); err != nil { + setupLog.Error(err, "unable to create controller", "controller", "FederatedCluster") + os.Exit(1) + } + } // Register the ModelRouter validating webhook ONLY when the serving cert is // actually present in the configured cert dir. The webhook server crashes the diff --git a/config/crd/bases/federation.llmkube.dev_federatedclusters.yaml b/config/crd/bases/federation.llmkube.dev_federatedclusters.yaml new file mode 100644 index 00000000..b71fc4f9 --- /dev/null +++ b/config/crd/bases/federation.llmkube.dev_federatedclusters.yaml @@ -0,0 +1,138 @@ +--- +apiVersion: apiextensions.k8s.io/v1 +kind: CustomResourceDefinition +metadata: + annotations: + controller-gen.kubebuilder.io/version: v0.19.0 + name: federatedclusters.federation.llmkube.dev +spec: + group: federation.llmkube.dev + names: + kind: FederatedCluster + listKind: FederatedClusterList + plural: federatedclusters + shortNames: + - fedcluster + - fc + singular: federatedcluster + scope: Cluster + versions: + - additionalPrinterColumns: + - jsonPath: .spec.dataResidencyTier + name: Tier + type: string + - jsonPath: .status.phase + name: Phase + type: string + - jsonPath: .status.lastHeartbeatTime + name: Last Heartbeat + type: date + - jsonPath: .status.capacity.gpusAllocatable + name: GPUs + type: string + name: v1alpha1 + schema: + openAPIV3Schema: + description: FederatedCluster registers one edge site on the datacenter cluster. + properties: + apiVersion: + description: |- + APIVersion defines the versioned schema of this representation of an object. + Servers should convert recognized schemas to the latest internal value, and + may reject unrecognized values. + More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources + type: string + kind: + description: |- + Kind is a string value representing the REST resource this object represents. + Servers may infer this from the endpoint the client submits requests to. + Cannot be updated. + In CamelCase. + More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds + type: string + metadata: + type: object + spec: + description: |- + FederatedClusterSpec is admin-owned. displayName and dataResidencyTier are + set at registration; the edge and the datacenter controller only write status. + properties: + dataResidencyTier: + description: |- + DataResidencyTier is recorded now and enforced later by the federation + router (#1237). It is a free-form tier label (for example "eu", "floor-3"). + type: string + displayName: + description: DisplayName is a human label for the site. + type: string + heartbeatIntervalSeconds: + default: 30 + description: |- + HeartbeatIntervalSeconds is how often the edge is expected to push status. + The datacenter derives staleness thresholds from it (3x Stale, 10x Unreachable). + format: int32 + minimum: 5 + type: integer + type: object + status: + description: |- + FederatedClusterStatus has two writers. The edge writes everything except + Phase, over an RBAC-scoped status-subresource client. The datacenter + controller writes ONLY Phase, from LastHeartbeatTime staleness. + properties: + capacity: + description: Capacity is edge-owned node/GPU capacity. + properties: + gpusAllocatable: + format: int32 + type: integer + gpusTotal: + format: int32 + type: integer + nodes: + format: int32 + type: integer + required: + - gpusAllocatable + - gpusTotal + - nodes + type: object + inference: + description: Inference is edge-owned inference health. + properties: + models: + format: int32 + type: integer + servicesFailed: + format: int32 + type: integer + servicesReady: + format: int32 + type: integer + servicesTotal: + format: int32 + type: integer + required: + - models + - servicesFailed + - servicesReady + - servicesTotal + type: object + lastHeartbeatTime: + description: 'LastHeartbeatTime is edge-owned: set to now on every + successful push.' + format: date-time + type: string + observedVersion: + description: 'ObservedVersion is edge-owned: the LLMKube/operator + version on the site.' + type: string + phase: + description: 'Phase is datacenter-owned: Connected, Stale, or Unreachable.' + type: string + type: object + type: object + served: true + storage: true + subresources: + status: {} diff --git a/config/rbac/role.yaml b/config/rbac/role.yaml index 9881aa6b..56b3fb93 100644 --- a/config/rbac/role.yaml +++ b/config/rbac/role.yaml @@ -102,6 +102,22 @@ rules: - get - list - watch +- apiGroups: + - federation.llmkube.dev + resources: + - federatedclusters + verbs: + - get + - list + - watch +- apiGroups: + - federation.llmkube.dev + resources: + - federatedclusters/status + verbs: + - get + - patch + - update - apiGroups: - foreman.llmkube.dev resources: diff --git a/config/samples/federation_v1alpha1_federatedcluster.yaml b/config/samples/federation_v1alpha1_federatedcluster.yaml new file mode 100644 index 00000000..c5849e02 --- /dev/null +++ b/config/samples/federation_v1alpha1_federatedcluster.yaml @@ -0,0 +1,31 @@ +apiVersion: federation.llmkube.dev/v1alpha1 +kind: FederatedCluster +metadata: + name: acme-datacenter-1 +spec: + # DisplayName is a human-readable label for this edge site. + displayName: "ACME Datacenter (Floor 3)" + + # DataResidencyTier is a free-form tier label (e.g., "eu", "floor-3", "us-west") + # that is recorded now and enforced later by the federation router (issue #1237). + # It is not enforced by the federation status slice. + dataResidencyTier: "floor-3" + + # HeartbeatIntervalSeconds is how often the edge operator is expected to push + # status updates to the datacenter. The datacenter controller derives staleness + # thresholds from this: 3x the interval marks the site as Stale, 10x marks it as + # Unreachable. Minimum is 5 seconds; defaults to 30 seconds. + heartbeatIntervalSeconds: 30 + +# Status is edge-written by the federation-edge controller and datacenter-written +# by the federation hub controller. Status is not set by hand in this sample. +# +# An edge site's operator (running with --federation-role=edge) writes: +# - lastHeartbeatTime (set to now on every successful push) +# - observedVersion (the LLMKube operator version on the site) +# - capacity (node and GPU count) +# - inference (inference service health summary) +# +# The datacenter cluster's controller (running with --federation-role=hub, the default) +# writes: +# - phase (Connected, Stale, or Unreachable, computed from lastHeartbeatTime staleness) diff --git a/docs/federation/registry-and-status.md b/docs/federation/registry-and-status.md new file mode 100644 index 00000000..d6b32969 --- /dev/null +++ b/docs/federation/registry-and-status.md @@ -0,0 +1,167 @@ +# Federation Registry and Status Rollup + +LLMKube federation lets a datacenter cluster track fleet-wide inference status and capacity from remote edge sites without granting those sites more than minimal, per-site scoped credentials. This guide covers how to register an edge site, configure the edge operator, and read the fleet-wide status. + +## Architecture Overview: Spoke Model + +Federation uses a spoke model where edge sites initiate outbound connections to the datacenter: + +- **Edge initiates**: Edge operators push heartbeats and inference status to the datacenter cluster, never accepting inbound connections. +- **Outbound only**: All communication flows from edge to datacenter. No datacenter-to-edge control plane traffic. +- **NAT and firewall friendly**: Works across network boundaries since the edge cluster always dials outbound. + +The datacenter cluster reconciles each edge site's health status (Connected, Stale, Unreachable) based on heartbeat staleness. Edge sites carry a narrowly scoped credential that can ONLY update their own FederatedCluster's status subresource. + +## Registering an Edge Site + +Registration happens on the datacenter cluster using the `llmkube fleet register` command. This creates: + +1. A FederatedCluster object on the datacenter (identity for the site) +2. A least-privilege ServiceAccount, ClusterRole, and ClusterRoleBinding on the datacenter (scoped to the site's own FederatedCluster status subresource) +3. A long-lived bearer token for that ServiceAccount +4. A kubeconfig snippet and operator flags for the site admin to install on the edge + +### Register a Site + +Run this on the datacenter cluster: + +```bash +llmkube fleet register \ + --name acme-floor-3 \ + --residency floor-3 \ + --heartbeat-interval 30 \ + --datacenter-endpoint https://datacenter.example.com:6443 +``` + +Flag meanings: + +- `--name` (required): Unique identifier for this edge site's FederatedCluster object. Used as the token's resource scope. +- `--residency`: Free-form data residency tier label (e.g., "eu", "us-west", "floor-3"). Recorded now; enforced later by the federation router (issue #1237). +- `--heartbeat-interval`: Expected edge heartbeat interval in seconds. Staleness thresholds are derived from this (3x=Stale, 10x=Unreachable). Defaults to 30 seconds. +- `--datacenter-endpoint`: Datacenter API server URL to embed in the edge kubeconfig. If omitted, falls back to the current kubeconfig context's server. +- `--namespace`: Kubernetes namespace on the datacenter to create the per-site ServiceAccount. Defaults to llmkube-system. + +The command outputs a kubeconfig snippet and operator flags: + +``` +# Edge site "acme-floor-3" registered on the datacenter as FederatedCluster/acme-floor-3. +# Save the block below as a kubeconfig file on the edge site (for example, +# federation-datacenter.kubeconfig), then start the edge operator with the +# flags shown after it. + +apiVersion: v1 +kind: Config +clusters: +- cluster: + server: https://datacenter.example.com:6443 + certificate-authority-data: + name: datacenter +contexts: +- context: + cluster: datacenter + user: fedcluster-acme-floor-3 + name: datacenter +current-context: datacenter +users: +- name: fedcluster-acme-floor-3 + user: + token: + +# ServiceAccount: llmkube-system/fedcluster-acme-floor-3 +# On the edge site's operator, set: +# --federation-role=edge --federation-cluster-name=acme-floor-3 --federation-datacenter-kubeconfig= +``` + +### Configure the Edge Operator + +On the edge site's cluster, start the operator with: + +```bash +llmkube-controller-manager \ + --federation-role=edge \ + --federation-cluster-name=acme-floor-3 \ + --federation-datacenter-kubeconfig=/path/to/federation-datacenter.kubeconfig \ + # ... other flags unchanged +``` + +Operator flags: + +- `--federation-role=edge`: Switch from hub (datacenter, default) to edge mode. In edge mode, the operator runs only the FederationEdgeReconciler and skips the FederatedClusterReconciler. +- `--federation-cluster-name`: Name of this edge site's FederatedCluster object on the datacenter (matches the registration --name). +- `--federation-datacenter-kubeconfig`: Path to the kubeconfig file that was printed by `llmkube fleet register`. This file contains the narrowly scoped token and datacenter API endpoint. + +In edge mode, the operator reads local Models, InferenceServices, and Nodes, computes the fleet's inference health and capacity, and pushes updates to the datacenter every heartbeat interval. + +## Reading Fleet Status + +Once edge sites are registered and running, view fleet-wide status on the datacenter cluster: + +```bash +llmkube fleet status +``` + +Example output: + +``` +NAME TIER PHASE LAST HEARTBEAT GPUS SERVICES +acme-floor-3 floor-3 Connected 2 minutes ago 6/8 3/0/5 +prod-us-west us-west Connected 1 minute ago 16/16 10/0/15 +backup-eu eu Stale 8 minutes ago -/- -/-/- + +Fleet-wide: 3 site(s) (2 Connected, 1 Stale, 0 Unreachable) +GPUs: 22/24 allocatable/total +Services: 13/0/20 ready/failed/total +``` + +Column meanings: + +- **NAME**: FederatedCluster name (site identity). +- **TIER**: DataResidencyTier label (or "-" if unset). +- **PHASE**: Connection status: + - Connected: heartbeat received within the interval. + - Stale: no heartbeat for 3x the interval. + - Unreachable: no heartbeat for 10x the interval. +- **LAST HEARTBEAT**: How long ago the last status push was received. +- **GPUS**: Allocatable/total GPU count on the site. +- **SERVICES**: Ready/failed/total inference service count. + +The footer aggregates across all sites: + +- Count of sites in each phase. +- Total GPU allocatable/total capacity. +- Total inference service counts (ready/failed/total). + +## Least-Privilege Token Scope + +The bearer token minted for each edge site is scoped to the following permissions (enforced by a ClusterRole with resourceNames): + +- **Get** the FederatedCluster object by its own name (no list, no watch, no delete). +- **Get**, **update**, **patch** the FederatedCluster's status subresource by name (no other verbs, no other resources). + +This means an edge site cannot: + +- Read or modify any other site's FederatedCluster. +- Read or write any other Kubernetes resource on the datacenter. +- List, watch, or delete its own FederatedCluster (only read and update status). + +This least-privilege scope is enforced by Kubernetes RBAC at the datacenter cluster, independent of the edge operator's behavior. + +## Data Residency: Status-Only Slice + +This federation slice (status rollup and registry) is a status-only system. The dataResidencyTier field is recorded in the FederatedCluster spec but NOT enforced by the federation components: + +- **Recorded**: Every FederatedCluster stores its tier label for audit and inventory. +- **Not enforced**: Requests are not routed or filtered by tier yet. +- **Future enforcement**: The federation router (issue #1237) will enforce tier constraints in a later phase, selecting edge sites based on residency requirements in request routes. + +If you are planning a deployment that requires data residency constraints, set the tier now so the router has the information when it is implemented later. + +## Token Rotation and Renewal + +The tokens minted by `llmkube fleet register` are long-lived (10-year expiration) bearer credentials embedded in the edge kubeconfig. Token rotation is out of scope for this slice. In a future phase, automation may be added to rotate tokens while keeping the same identity and permissions scoped to the site. + +## Next Steps + +- Review the sample [FederatedCluster manifest](../../config/samples/federation_v1alpha1_federatedcluster.yaml). +- Check the [federation CRD API reference](../api/federation.md). +- See [federation architecture](./architecture.md) for detailed design rationale. diff --git a/internal/controller/federatedcluster_controller.go b/internal/controller/federatedcluster_controller.go new file mode 100644 index 00000000..af7644ea --- /dev/null +++ b/internal/controller/federatedcluster_controller.go @@ -0,0 +1,91 @@ +/* +Copyright 2025. + +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 controller + +import ( + "context" + "time" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + + federationv1alpha1 "github.com/defilantech/llmkube/api/federation/v1alpha1" +) + +// FederatedClusterReconciler owns ONLY status.phase, derived from the edge-written +// lastHeartbeatTime. It never writes to edge clusters. +type FederatedClusterReconciler struct { + client.Client + Scheme *runtime.Scheme +} + +// phaseForHeartbeat derives status.phase from staleness of the last edge +// heartbeat relative to a multiple of the expected interval. A nil heartbeat +// (never reported) is Unreachable. +func phaseForHeartbeat(last *metav1.Time, intervalSeconds int32, now time.Time) string { + if last == nil { + return federationv1alpha1.FederatedClusterUnreachable + } + if intervalSeconds <= 0 { + intervalSeconds = 30 + } + age := now.Sub(last.Time) + iv := time.Duration(intervalSeconds) * time.Second + switch { + case age <= 3*iv: + return federationv1alpha1.FederatedClusterConnected + case age <= 10*iv: + return federationv1alpha1.FederatedClusterStale + default: + return federationv1alpha1.FederatedClusterUnreachable + } +} + +// +kubebuilder:rbac:groups=federation.llmkube.dev,resources=federatedclusters,verbs=get;list;watch +// +kubebuilder:rbac:groups=federation.llmkube.dev,resources=federatedclusters/status,verbs=get;update;patch + +func (r *FederatedClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) { + fc := &federationv1alpha1.FederatedCluster{} + if err := r.Get(ctx, req.NamespacedName, fc); err != nil { + return ctrl.Result{}, client.IgnoreNotFound(err) + } + + want := phaseForHeartbeat(fc.Status.LastHeartbeatTime, fc.Spec.HeartbeatIntervalSeconds, time.Now()) + if fc.Status.Phase != want { + fc.Status.Phase = want + if err := r.Status().Update(ctx, fc); err != nil { + return ctrl.Result{}, err + } + } + + // Requeue so phase decays even without an edge write. Requeue at the + // interval so a missed heartbeat is reflected within one interval. + iv := fc.Spec.HeartbeatIntervalSeconds + if iv <= 0 { + iv = 30 + } + return ctrl.Result{RequeueAfter: time.Duration(iv) * time.Second}, nil +} + +func (r *FederatedClusterReconciler) SetupWithManager(mgr ctrl.Manager) error { + return ctrl.NewControllerManagedBy(mgr). + For(&federationv1alpha1.FederatedCluster{}). + Named("federatedcluster"). + Complete(r) +} diff --git a/internal/controller/federatedcluster_controller_test.go b/internal/controller/federatedcluster_controller_test.go new file mode 100644 index 00000000..929a5c3d --- /dev/null +++ b/internal/controller/federatedcluster_controller_test.go @@ -0,0 +1,137 @@ +/* +Copyright 2025. + +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 controller + +import ( + "context" + "testing" + "time" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/reconcile" + + federationv1alpha1 "github.com/defilantech/llmkube/api/federation/v1alpha1" +) + +func TestPhaseForHeartbeat(t *testing.T) { + now := time.Unix(1_000_000, 0) + iv := int32(30) + cases := []struct { + name string + age time.Duration + want string + }{ + {"fresh", 10 * time.Second, "Connected"}, + {"just under 3x", 89 * time.Second, "Connected"}, + {"at stale edge", 4 * time.Minute, "Stale"}, + {"just under 10x", 299 * time.Second, "Stale"}, + {"unreachable", 6 * time.Minute, "Unreachable"}, + } + for _, c := range cases { + last := metav1.NewTime(now.Add(-c.age)) + if got := phaseForHeartbeat(&last, iv, now); got != c.want { + t.Errorf("%s: got %q want %q", c.name, got, c.want) + } + } + if got := phaseForHeartbeat(nil, iv, now); got != "Unreachable" { + t.Errorf("nil heartbeat: got %q want Unreachable", got) + } +} + +var _ = Describe("FederatedClusterReconciler phase transitions", func() { + var ( + reconciler *FederatedClusterReconciler + ctx context.Context + ) + + const name = "fc-phase-test" + const intervalSeconds = int32(30) + + BeforeEach(func() { + ctx = context.Background() + reconciler = &FederatedClusterReconciler{ + Client: k8sClient, + Scheme: k8sClient.Scheme(), + } + + fc := &federationv1alpha1.FederatedCluster{} + err := k8sClient.Get(ctx, types.NamespacedName{Name: name}, fc) + if err != nil && errors.IsNotFound(err) { + fc = &federationv1alpha1.FederatedCluster{ + ObjectMeta: metav1.ObjectMeta{Name: name}, + Spec: federationv1alpha1.FederatedClusterSpec{ + HeartbeatIntervalSeconds: intervalSeconds, + }, + } + Expect(k8sClient.Create(ctx, fc)).To(Succeed()) + } + }) + + AfterEach(func() { + fc := &federationv1alpha1.FederatedCluster{} + if err := k8sClient.Get(ctx, types.NamespacedName{Name: name}, fc); err == nil { + _ = k8sClient.Delete(ctx, fc) + } + }) + + setHeartbeat := func(age time.Duration) { + fc := &federationv1alpha1.FederatedCluster{} + Expect(k8sClient.Get(ctx, types.NamespacedName{Name: name}, fc)).To(Succeed()) + ts := metav1.NewTime(time.Now().Add(-age)) + fc.Status.LastHeartbeatTime = &ts + Expect(k8sClient.Status().Update(ctx, fc)).To(Succeed()) + } + + reconcileAndGet := func() (reconcile.Result, *federationv1alpha1.FederatedCluster) { + result, err := reconciler.Reconcile(ctx, reconcile.Request{ + NamespacedName: types.NamespacedName{Name: name}, + }) + Expect(err).NotTo(HaveOccurred()) + + fc := &federationv1alpha1.FederatedCluster{} + Expect(k8sClient.Get(ctx, types.NamespacedName{Name: name}, fc)).To(Succeed()) + return result, fc + } + + It("sets Connected for a fresh heartbeat and requeues", func() { + setHeartbeat(5 * time.Second) + + result, fc := reconcileAndGet() + Expect(fc.Status.Phase).To(Equal(federationv1alpha1.FederatedClusterConnected)) + Expect(result.RequeueAfter).To(BeNumerically(">", 0)) + }) + + It("sets Stale once the heartbeat is older than 3x the interval", func() { + setHeartbeat(5 * time.Duration(intervalSeconds) * time.Second) + + result, fc := reconcileAndGet() + Expect(fc.Status.Phase).To(Equal(federationv1alpha1.FederatedClusterStale)) + Expect(result.RequeueAfter).To(BeNumerically(">", 0)) + }) + + It("sets Unreachable once the heartbeat is older than 10x the interval", func() { + setHeartbeat(20 * time.Duration(intervalSeconds) * time.Second) + + result, fc := reconcileAndGet() + Expect(fc.Status.Phase).To(Equal(federationv1alpha1.FederatedClusterUnreachable)) + Expect(result.RequeueAfter).To(BeNumerically(">", 0)) + }) +}) diff --git a/internal/controller/federation_edge_controller.go b/internal/controller/federation_edge_controller.go new file mode 100644 index 00000000..c06269e7 --- /dev/null +++ b/internal/controller/federation_edge_controller.go @@ -0,0 +1,253 @@ +/* +Copyright 2025. + +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 controller + +import ( + "context" + "time" + + "github.com/go-logr/logr" + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/client" + + federationv1alpha1 "github.com/defilantech/llmkube/api/federation/v1alpha1" + inferencev1alpha1 "github.com/defilantech/llmkube/api/v1alpha1" +) + +// defaultFederationEdgeInterval is the push cadence used when +// FederationEdgeReconciler.Interval is left at its zero value. +const defaultFederationEdgeInterval = 30 * time.Second + +// Federation role values for cmd/main.go's --federation-role flag. Hub (the +// default) runs the datacenter-side FederatedClusterReconciler, which owns +// status.phase for every registered edge. Edge runs FederationEdgeReconciler +// instead: it never runs both roles in the same process, since only one +// side of a given FederatedCluster object should ever be reconciled by a +// given operator instance. +const ( + FederationRoleHub = "hub" + FederationRoleEdge = "edge" +) + +// gpuResourceNames are the extended resources summed across node +// Capacity/Allocatable for FederatedCluster GPU accounting. Reuses the +// vendor resource-name variables declared in gpu_resources.go (used there +// for per-InferenceService scheduling) rather than redeclaring them, so a +// new vendor added there is picked up here for free. +var gpuResourceNames = []corev1.ResourceName{ + nvidiaGPUResourceName, + amdGPUResourceName, + intelGPUResourceNameI915, + intelGPUResourceNameXE, + vulkanDRIResourceName, +} + +// FederationEdgeReconciler is a manager.Runnable (not a reconcile.Reconciler: +// there is no local CR driving this side of federation) that periodically +// observes local inference state and pushes it to the datacenter's +// FederatedCluster status subresource. It is spoke-initiated and outbound +// only: +// +// - LocalClient reads Models, InferenceServices, and Nodes on THIS +// (edge) cluster. +// - DatacenterClient writes ONLY the status subresource of the +// ClusterName FederatedCluster object on the datacenter cluster, using +// a separate, RBAC-scoped client built from +// --federation-datacenter-kubeconfig (a scoped token, distinct from +// the local manager client). Nothing else is ever written on the +// datacenter. +// +// Two-writer discipline: buildStatusSummary never populates Phase, and +// tick's status patch leaves fc.Status.Phase exactly as read from the +// datacenter, so the computed merge patch carries no "phase" key at all. +// Phase stays the datacenter FederatedClusterReconciler's exclusively (see +// federatedcluster_controller.go). +type FederationEdgeReconciler struct { + // LocalClient reads Models/InferenceServices/Nodes on the edge cluster. + LocalClient client.Client + // DatacenterClient writes status on the datacenter cluster's + // FederatedCluster/ClusterName object. Built from a separate, + // RBAC-scoped kubeconfig (--federation-datacenter-kubeconfig). + DatacenterClient client.Client + // ClusterName is this edge's FederatedCluster object name on the + // datacenter (--federation-cluster-name). + ClusterName string + // Version is recorded as status.observedVersion on every push. + Version string + // Interval is the push cadence; defaultFederationEdgeInterval when unset. + Interval time.Duration + // Log is used for the log-and-retry-next-tick failure path; discarded + // when unset. + Log logr.Logger +} + +// NeedLeaderElection makes the edge push leader-election-aware: when the +// operator runs with --leader-elect=true, only the elected replica pushes, +// avoiding redundant writes to the datacenter from every replica. A +// standalone deployment (or leader election disabled) pushes regardless, +// same as pkg/foreman/audit.Reaper. +func (r *FederationEdgeReconciler) NeedLeaderElection() bool { return true } + +// +kubebuilder:rbac:groups=inference.llmkube.dev,resources=models,verbs=get;list;watch +// +kubebuilder:rbac:groups=inference.llmkube.dev,resources=inferenceservices,verbs=get;list;watch +// +kubebuilder:rbac:groups="",resources=nodes,verbs=get;list;watch + +// Start runs the periodic observe-and-push cycle until ctx is cancelled, +// then returns nil. It ticks once immediately (so a freshly started edge +// reports in without waiting a full interval) and then on Interval. Every +// failure mode inside tick is logged and swallowed, never propagated as a +// returned error, so a single bad tick (local list error, datacenter +// unreachable, FederatedCluster not yet registered) never crashes the +// operator; the next tick simply tries again. +func (r *FederationEdgeReconciler) Start(ctx context.Context) error { + interval := r.Interval + if interval <= 0 { + interval = defaultFederationEdgeInterval + } + log := r.Log + if log.IsZero() { + log = logr.Discard() + } + + r.tick(ctx, log) + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return nil + case <-ticker.C: + r.tick(ctx, log) + } + } +} + +// tick performs one observe-and-push cycle: list local state, build the +// summary, then patch it onto the datacenter FederatedCluster status. Every +// error is logged and swallowed here so the caller (Start's loop) never +// sees an error to propagate. +func (r *FederationEdgeReconciler) tick(ctx context.Context, log logr.Logger) { + var models inferencev1alpha1.ModelList + if err := r.LocalClient.List(ctx, &models); err != nil { + log.Error(err, "federation edge: list local Models, retrying next tick") + return + } + var isvcs inferencev1alpha1.InferenceServiceList + if err := r.LocalClient.List(ctx, &isvcs); err != nil { + log.Error(err, "federation edge: list local InferenceServices, retrying next tick") + return + } + var nodes corev1.NodeList + if err := r.LocalClient.List(ctx, &nodes); err != nil { + log.Error(err, "federation edge: list local Nodes, retrying next tick") + return + } + + summary := buildStatusSummary(models.Items, isvcs.Items, nodes.Items, r.Version) + + fc := &federationv1alpha1.FederatedCluster{} + if err := r.DatacenterClient.Get(ctx, types.NamespacedName{Name: r.ClusterName}, fc); err != nil { + if apierrors.IsNotFound(err) { + log.Info("federation edge: FederatedCluster not registered on datacenter yet, retrying next tick", + "clusterName", r.ClusterName) + return + } + log.Error(err, "federation edge: get FederatedCluster on datacenter (unreachable?), retrying next tick", + "clusterName", r.ClusterName) + return + } + + // Two-writer discipline: patch is diffed against this DeepCopy. Every + // edge-owned field below is assigned; fc.Status.Phase is left exactly + // as read, so it never appears in the computed merge patch and the + // datacenter's phase can never be clobbered by this push. + patch := client.MergeFrom(fc.DeepCopy()) + now := metav1.Now() + fc.Status.LastHeartbeatTime = &now + fc.Status.ObservedVersion = summary.ObservedVersion + fc.Status.Capacity = summary.Capacity + fc.Status.Inference = summary.Inference + + if err := r.DatacenterClient.Status().Patch(ctx, fc, patch); err != nil { + log.Error(err, "federation edge: patch FederatedCluster status on datacenter (unreachable?), retrying next tick", + "clusterName", r.ClusterName) + return + } + + log.V(1).Info("federation edge: pushed status to datacenter", + "clusterName", r.ClusterName, + "servicesReady", summary.Inference.ServicesReady, + "servicesTotal", summary.Inference.ServicesTotal, + "nodes", summary.Capacity.Nodes) +} + +// buildStatusSummary is a pure function deriving the edge-owned +// FederatedClusterStatus fields (ObservedVersion, Capacity, Inference) from +// locally observed state. Phase and LastHeartbeatTime are intentionally left +// zero: Phase is datacenter-owned (federatedcluster_controller.go) and +// LastHeartbeatTime is stamped by the caller (tick) at push time, so this +// function stays deterministic and easy to table-test. +func buildStatusSummary( + models []inferencev1alpha1.Model, + isvcs []inferencev1alpha1.InferenceService, + nodes []corev1.Node, + version string, +) federationv1alpha1.FederatedClusterStatus { + inference := federationv1alpha1.ClusterInferenceSummary{ + Models: int32(len(models)), //nolint:gosec // G115: fleet size is far below int32 range + ServicesTotal: int32(len(isvcs)), //nolint:gosec // G115: fleet size is far below int32 range + } + for i := range isvcs { + switch isvcs[i].Status.Phase { + case PhaseReady: + inference.ServicesReady++ + case PhaseFailed: + inference.ServicesFailed++ + } + } + + capacity := federationv1alpha1.ClusterCapacity{ + Nodes: int32(len(nodes)), //nolint:gosec // G115: fleet size is far below int32 range + } + for i := range nodes { + capacity.GPUsTotal += sumGPUQuantity(nodes[i].Status.Capacity) + capacity.GPUsAllocatable += sumGPUQuantity(nodes[i].Status.Allocatable) + } + + return federationv1alpha1.FederatedClusterStatus{ + ObservedVersion: version, + Capacity: &capacity, + Inference: &inference, + } +} + +// sumGPUQuantity sums every known GPU extended resource (gpuResourceNames) +// present in a node's ResourceList (either Status.Capacity or +// Status.Allocatable). GPU counts are always whole units, so Value() (not +// MilliValue) is the correct accessor. +func sumGPUQuantity(rl corev1.ResourceList) int32 { + var total int64 + for _, name := range gpuResourceNames { + if q, ok := rl[name]; ok { + total += q.Value() + } + } + return int32(total) +} diff --git a/internal/controller/federation_edge_controller_test.go b/internal/controller/federation_edge_controller_test.go new file mode 100644 index 00000000..7958e75f --- /dev/null +++ b/internal/controller/federation_edge_controller_test.go @@ -0,0 +1,237 @@ +/* +Copyright 2025. + +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 controller + +import ( + "context" + "testing" + "time" + + "github.com/go-logr/logr" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/api/resource" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + + federationv1alpha1 "github.com/defilantech/llmkube/api/federation/v1alpha1" + inferencev1alpha1 "github.com/defilantech/llmkube/api/v1alpha1" +) + +func gpuNode(name string, gpus int64) corev1.Node { + qty := *resource.NewQuantity(gpus, resource.DecimalSI) + return corev1.Node{ + ObjectMeta: metav1.ObjectMeta{Name: name}, + Status: corev1.NodeStatus{ + Capacity: corev1.ResourceList{ + nvidiaGPUResourceName: qty, + }, + Allocatable: corev1.ResourceList{ + nvidiaGPUResourceName: qty, + }, + }, + } +} + +// multiVendorGPUNode advertises two different GPU resource names on the same +// node (e.g. an NVIDIA card plus an AMD card), so buildStatusSummary's +// ranging over the full gpuResourceNames list is actually exercised, not +// just the nvidia key. +func multiVendorGPUNode(name string, nvidia, amd int64) corev1.Node { + rl := corev1.ResourceList{ + nvidiaGPUResourceName: *resource.NewQuantity(nvidia, resource.DecimalSI), + amdGPUResourceName: *resource.NewQuantity(amd, resource.DecimalSI), + } + return corev1.Node{ + ObjectMeta: metav1.ObjectMeta{Name: name}, + Status: corev1.NodeStatus{ + Capacity: rl, + Allocatable: rl, + }, + } +} + +func TestBuildStatusSummary(t *testing.T) { + models := []inferencev1alpha1.Model{ + {ObjectMeta: metav1.ObjectMeta{Name: "model-a"}}, + {ObjectMeta: metav1.ObjectMeta{Name: "model-b"}}, + } + isvcs := []inferencev1alpha1.InferenceService{ + { + ObjectMeta: metav1.ObjectMeta{Name: "isvc-a"}, + Status: inferencev1alpha1.InferenceServiceStatus{Phase: PhaseReady}, + }, + { + ObjectMeta: metav1.ObjectMeta{Name: "isvc-b"}, + Status: inferencev1alpha1.InferenceServiceStatus{Phase: PhaseReady}, + }, + { + ObjectMeta: metav1.ObjectMeta{Name: "isvc-c"}, + Status: inferencev1alpha1.InferenceServiceStatus{Phase: PhaseFailed}, + }, + } + nodes := []corev1.Node{ + gpuNode("node-a", 2), + // node-b carries two vendors (nvidia 4 + amd 2) so the multi-vendor + // summation path is proven, not just the nvidia key. + multiVendorGPUNode("node-b", 4, 2), + } + + got := buildStatusSummary(models, isvcs, nodes, "0.9.10") + + if got.Phase != "" { + t.Errorf("Phase = %q, want empty (datacenter-owned)", got.Phase) + } + if got.LastHeartbeatTime != nil { + t.Errorf("LastHeartbeatTime = %v, want nil (stamped by the caller at push time)", got.LastHeartbeatTime) + } + if got.ObservedVersion != "0.9.10" { + t.Errorf("ObservedVersion = %q, want %q", got.ObservedVersion, "0.9.10") + } + if got.Inference == nil { + t.Fatal("Inference summary is nil") + } + if got.Inference.Models != 2 { + t.Errorf("Models = %d, want 2", got.Inference.Models) + } + if got.Inference.ServicesReady != 2 { + t.Errorf("ServicesReady = %d, want 2", got.Inference.ServicesReady) + } + if got.Inference.ServicesFailed != 1 { + t.Errorf("ServicesFailed = %d, want 1", got.Inference.ServicesFailed) + } + if got.Inference.ServicesTotal != 3 { + t.Errorf("ServicesTotal = %d, want 3", got.Inference.ServicesTotal) + } + if got.Capacity == nil { + t.Fatal("Capacity is nil") + } + if got.Capacity.Nodes != 2 { + t.Errorf("Nodes = %d, want 2", got.Capacity.Nodes) + } + // node-a nvidia 2 + node-b (nvidia 4 + amd 2) = 8 across both vendors. + if got.Capacity.GPUsTotal != 8 { + t.Errorf("GPUsTotal = %d, want 8", got.Capacity.GPUsTotal) + } + if got.Capacity.GPUsAllocatable != 8 { + t.Errorf("GPUsAllocatable = %d, want 8", got.Capacity.GPUsAllocatable) + } +} + +func TestBuildStatusSummaryEmpty(t *testing.T) { + got := buildStatusSummary(nil, nil, nil, "") + if got.Phase != "" { + t.Errorf("Phase = %q, want empty", got.Phase) + } + if got.Inference.ServicesTotal != 0 || got.Inference.Models != 0 { + t.Errorf("expected zeroed inference summary, got %+v", got.Inference) + } + if got.Capacity.Nodes != 0 || got.Capacity.GPUsTotal != 0 { + t.Errorf("expected zeroed capacity, got %+v", got.Capacity) + } +} + +// --- envtest push spec: a single tick against an envtest apiserver acting +// as both the local cluster (reads) and the datacenter (status write). --- + +var _ = Describe("FederationEdgeReconciler", func() { + const clusterName = "edge-fed-test" + + var ctx context.Context + + BeforeEach(func() { + ctx = context.Background() + + fc := &federationv1alpha1.FederatedCluster{ + ObjectMeta: metav1.ObjectMeta{Name: clusterName}, + Spec: federationv1alpha1.FederatedClusterSpec{HeartbeatIntervalSeconds: 30}, + } + Expect(k8sClient.Create(ctx, fc)).To(Succeed()) + + model := &inferencev1alpha1.Model{ + ObjectMeta: metav1.ObjectMeta{Name: "edge-model", Namespace: "default"}, + Spec: inferencev1alpha1.ModelSpec{Source: "/models/edge.gguf"}, + } + Expect(k8sClient.Create(ctx, model)).To(Succeed()) + + replicas := int32(1) + isvcReady := &inferencev1alpha1.InferenceService{ + ObjectMeta: metav1.ObjectMeta{Name: "edge-isvc-ready", Namespace: "default"}, + Spec: inferencev1alpha1.InferenceServiceSpec{ModelRef: model.Name, Replicas: &replicas}, + } + Expect(k8sClient.Create(ctx, isvcReady)).To(Succeed()) + isvcReady.Status.Phase = PhaseReady + Expect(k8sClient.Status().Update(ctx, isvcReady)).To(Succeed()) + + isvcFailed := &inferencev1alpha1.InferenceService{ + ObjectMeta: metav1.ObjectMeta{Name: "edge-isvc-failed", Namespace: "default"}, + Spec: inferencev1alpha1.InferenceServiceSpec{ModelRef: model.Name, Replicas: &replicas}, + } + Expect(k8sClient.Create(ctx, isvcFailed)).To(Succeed()) + isvcFailed.Status.Phase = PhaseFailed + Expect(k8sClient.Status().Update(ctx, isvcFailed)).To(Succeed()) + }) + + AfterEach(func() { + fc := &federationv1alpha1.FederatedCluster{} + if err := k8sClient.Get(ctx, types.NamespacedName{Name: clusterName}, fc); err == nil { + Expect(k8sClient.Delete(ctx, fc)).To(Succeed()) + } + for _, name := range []string{"edge-isvc-ready", "edge-isvc-failed"} { + isvc := &inferencev1alpha1.InferenceService{} + if err := k8sClient.Get(ctx, types.NamespacedName{Name: name, Namespace: "default"}, isvc); err == nil { + Expect(k8sClient.Delete(ctx, isvc)).To(Succeed()) + } + } + model := &inferencev1alpha1.Model{} + if err := k8sClient.Get(ctx, types.NamespacedName{Name: "edge-model", Namespace: "default"}, model); err == nil { + Expect(k8sClient.Delete(ctx, model)).To(Succeed()) + } + }) + + It("pushes the observed summary to the datacenter FederatedCluster status without touching Phase", func() { + // Seed a datacenter-owned Phase before the push, to prove the edge + // patch leaves it alone. + fc := &federationv1alpha1.FederatedCluster{} + Expect(k8sClient.Get(ctx, types.NamespacedName{Name: clusterName}, fc)).To(Succeed()) + fc.Status.Phase = federationv1alpha1.FederatedClusterConnected + Expect(k8sClient.Status().Update(ctx, fc)).To(Succeed()) + + reconciler := &FederationEdgeReconciler{ + LocalClient: k8sClient, + DatacenterClient: k8sClient, + ClusterName: clusterName, + Version: "0.9.10-test", + Interval: time.Hour, + } + reconciler.tick(ctx, logr.Discard()) + + got := &federationv1alpha1.FederatedCluster{} + Expect(k8sClient.Get(ctx, types.NamespacedName{Name: clusterName}, got)).To(Succeed()) + + Expect(got.Status.Phase).To(Equal(federationv1alpha1.FederatedClusterConnected), "edge push must never clobber the datacenter-owned Phase") + Expect(got.Status.LastHeartbeatTime).NotTo(BeNil()) + Expect(got.Status.ObservedVersion).To(Equal("0.9.10-test")) + Expect(got.Status.Inference).NotTo(BeNil()) + Expect(got.Status.Inference.ServicesReady).To(Equal(int32(1))) + Expect(got.Status.Inference.ServicesFailed).To(Equal(int32(1))) + Expect(got.Status.Inference.ServicesTotal).To(Equal(int32(2))) + Expect(got.Status.Inference.Models).To(Equal(int32(1))) + Expect(got.Status.Capacity).NotTo(BeNil()) + }) +}) diff --git a/internal/controller/suite_test.go b/internal/controller/suite_test.go index 3f731ccf..a43c4251 100644 --- a/internal/controller/suite_test.go +++ b/internal/controller/suite_test.go @@ -32,6 +32,7 @@ import ( logf "sigs.k8s.io/controller-runtime/pkg/log" "sigs.k8s.io/controller-runtime/pkg/log/zap" + federationv1alpha1 "github.com/defilantech/llmkube/api/federation/v1alpha1" inferencev1alpha1 "github.com/defilantech/llmkube/api/v1alpha1" // +kubebuilder:scaffold:imports ) @@ -62,6 +63,9 @@ var _ = BeforeSuite(func() { err = inferencev1alpha1.AddToScheme(scheme.Scheme) Expect(err).NotTo(HaveOccurred()) + err = federationv1alpha1.AddToScheme(scheme.Scheme) + Expect(err).NotTo(HaveOccurred()) + // +kubebuilder:scaffold:scheme By("bootstrapping test environment") diff --git a/pkg/cli/fleet.go b/pkg/cli/fleet.go new file mode 100644 index 00000000..4c7fb736 --- /dev/null +++ b/pkg/cli/fleet.go @@ -0,0 +1,546 @@ +/* +Copyright 2025. + +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 cli + +import ( + "context" + "encoding/base64" + "fmt" + "io" + "sort" + "strings" + "text/tabwriter" + "time" + + "github.com/spf13/cobra" + authenticationv1 "k8s.io/api/authentication/v1" + corev1 "k8s.io/api/core/v1" + rbacv1 "k8s.io/api/rbac/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/kubernetes/scheme" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/config" + + federationv1alpha1 "github.com/defilantech/llmkube/api/federation/v1alpha1" +) + +const ( + // fleetServiceAccountPrefix names the per-site ServiceAccount, ClusterRole, + // and ClusterRoleBinding created for a FederatedCluster (fedcluster-). + fleetServiceAccountPrefix = "fedcluster-" + + // defaultFleetNamespace is where the per-site ServiceAccount and its RBAC + // live on the datacenter cluster. Matches the operator's own namespace + // convention (see llmkube-controller-manager in cache.go/delete.go). + defaultFleetNamespace = "llmkube-system" + + // defaultHeartbeatIntervalSeconds mirrors the CRD's + // +kubebuilder:default=30 on HeartbeatIntervalSeconds. + defaultHeartbeatIntervalSeconds = int32(30) + + // fleetTokenExpirationSeconds is long-lived (10 years): this token is a + // bootstrap credential embedded in a kubeconfig file carried to a remote + // edge site, not a short-lived pod-projected token, so it must keep + // working across the whole registration's lifetime. Token rotation is + // out of scope for this command (see status/rotation follow-up). + fleetTokenExpirationSeconds = int64(10 * 365 * 24 * 3600) + + // datacenterEndpointPlaceholder fills the edge-config snippet's server + // field when neither --datacenter-endpoint nor the caller's own + // kubeconfig host is available. + datacenterEndpointPlaceholder = "" + + // fleetStatusUnknown fills a table cell when a site has never pushed the + // corresponding status field (nil Capacity/Inference, empty tier/phase). + fleetStatusUnknown = "-" + + // fleetStatusNeverHeartbeat is the LAST HEARTBEAT cell for a site whose + // LastHeartbeatTime is nil, i.e. it has never pushed status at all. + fleetStatusNeverHeartbeat = "never" +) + +// mintFleetToken mints a bearer token for a ServiceAccount via the +// TokenRequest API (Kubernetes 1.24+; no more auto-created SA token Secrets). +// It is a package-level seam: client-go's fake ServiceAccounts().CreateToken +// echoes the request object back unchanged instead of synthesizing a token +// value, so tests replace this var rather than exercising the fake +// TokenRequest subresource. +// +// It returns the apiserver's actual Status.ExpirationTimestamp alongside the +// token, not just the ExpirationSeconds that was requested: a datacenter +// cluster running with --service-account-max-token-expiration can cap the +// real lifetime far below the ~10y requested here, and the caller needs the +// real value to warn the site admin rather than silently printing a token +// that expires much sooner than the edge-config snippet implies. +var mintFleetToken = func( + ctx context.Context, kube kubernetes.Interface, namespace, name string, +) (string, metav1.Time, error) { + expirationSeconds := fleetTokenExpirationSeconds + tr, err := kube.CoreV1().ServiceAccounts(namespace).CreateToken(ctx, name, &authenticationv1.TokenRequest{ + Spec: authenticationv1.TokenRequestSpec{ + ExpirationSeconds: &expirationSeconds, + }, + }, metav1.CreateOptions{}) + if err != nil { + return "", metav1.Time{}, fmt.Errorf("mint token for service account %s/%s: %w", namespace, name, err) + } + return tr.Status.Token, tr.Status.ExpirationTimestamp, nil +} + +// fleetRegisterInput is the pure input to fleetRegister, decoupled from +// cobra flags and REST config so it is trivial to unit test. +type fleetRegisterInput struct { + // Name is the FederatedCluster's name and the identity this site's + // scoped token is restricted to (via ResourceNames in the ClusterRole). + Name string + // ResidencyTier is FederatedClusterSpec.DataResidencyTier. + ResidencyTier string + // HeartbeatIntervalSeconds is FederatedClusterSpec.HeartbeatIntervalSeconds. + // Defaulted to defaultHeartbeatIntervalSeconds when <= 0. + HeartbeatIntervalSeconds int32 + // Namespace is where the per-site ServiceAccount lives on the datacenter + // cluster. Defaulted to defaultFleetNamespace when empty. + Namespace string + // DatacenterEndpoint is the datacenter API server URL embedded in the + // returned edge-config snippet. Falls back to + // datacenterEndpointPlaceholder when empty. + DatacenterEndpoint string + // CACertData is the datacenter API server's CA certificate (PEM), used + // to embed certificate-authority-data in the edge-config snippet + // instead of insecure-skip-tls-verify. Optional. + CACertData []byte +} + +type fleetRegisterOptions struct { + name string + residency string + heartbeatInterval int32 + datacenterEndpoint string + namespace string +} + +// NewFleetCommand is the `llmkube fleet` parent command: registering and +// (a later task) inspecting the edge sites participating in federation. +func NewFleetCommand() *cobra.Command { + cmd := &cobra.Command{ + Use: "fleet", + Short: "Manage federated edge sites", + Long: `Register and inspect edge sites participating in LLMKube federation. + +Federation lets a datacenter cluster track fleet-wide inference status and +capacity for remote edge sites (FederatedCluster objects), without those +sites ever being granted more than a narrow, per-site scoped credential.`, + } + cmd.AddCommand(newFleetRegisterCommand()) + cmd.AddCommand(newFleetStatusCommand()) + return cmd +} + +func newFleetRegisterCommand() *cobra.Command { + opts := &fleetRegisterOptions{} + + cmd := &cobra.Command{ + Use: "register", + Short: "Register an edge site as a FederatedCluster", + Long: `Register a new edge site on the datacenter cluster. + +This creates a FederatedCluster object plus a least-privilege ServiceAccount, +ClusterRole, and ClusterRoleBinding scoped to ONLY that one FederatedCluster's +status subresource. It then mints a token for that ServiceAccount and prints +an edge-config snippet (kubeconfig + operator flags) for the site admin to +install on the edge cluster's operator. + +Run this against the DATACENTER cluster's kubeconfig context.`, + RunE: func(cmd *cobra.Command, args []string) error { + return runFleetRegister(cmd, opts) + }, + } + + cmd.Flags().StringVar(&opts.name, "name", "", + "Unique name for the edge site's FederatedCluster (required)") + cmd.Flags().StringVar(&opts.residency, "residency", "", + "Data residency tier label for the site (e.g. \"eu\", \"floor-3\")") + cmd.Flags().Int32Var(&opts.heartbeatInterval, "heartbeat-interval", defaultHeartbeatIntervalSeconds, + "Expected edge heartbeat interval, in seconds") + cmd.Flags().StringVar(&opts.datacenterEndpoint, "datacenter-endpoint", "", + "Datacenter API server endpoint for the edge site to use "+ + "(e.g. https://datacenter.example.com:6443). Defaults to the "+ + "current kubeconfig context's server if not set.") + cmd.Flags().StringVar(&opts.namespace, "namespace", defaultFleetNamespace, + "Namespace on the datacenter cluster to create the per-site ServiceAccount in") + + if err := cmd.MarkFlagRequired("name"); err != nil { + // Only fails if the flag name doesn't exist, which would be a + // programmer error caught immediately by any test that constructs + // this command. + panic(err) + } + + return cmd +} + +func runFleetRegister(cmd *cobra.Command, opts *fleetRegisterOptions) error { + ctx := cmd.Context() + if ctx == nil { + ctx = context.Background() + } + + cfg, err := config.GetConfig() + if err != nil { + return fmt.Errorf("failed to get kubeconfig: %w", err) + } + + if err := federationv1alpha1.AddToScheme(scheme.Scheme); err != nil { + return fmt.Errorf("failed to add scheme: %w", err) + } + + k8sClient, err := client.New(cfg, client.Options{Scheme: scheme.Scheme}) + if err != nil { + return fmt.Errorf("failed to create client: %w", err) + } + + kube, err := kubernetes.NewForConfig(cfg) + if err != nil { + return fmt.Errorf("failed to create clientset: %w", err) + } + + endpoint := opts.datacenterEndpoint + if endpoint == "" { + // Fall back to the host of the kubeconfig context this command is + // itself running against, since `register` is run directly on the + // datacenter cluster. + endpoint = cfg.Host + } + + edgeConfig, err := fleetRegister(ctx, k8sClient, kube, fleetRegisterInput{ + Name: opts.name, + ResidencyTier: opts.residency, + HeartbeatIntervalSeconds: opts.heartbeatInterval, + Namespace: opts.namespace, + DatacenterEndpoint: endpoint, + CACertData: cfg.CAData, + }) + if err != nil { + return err + } + + if _, err := fmt.Fprintln(cmd.OutOrStdout(), edgeConfig); err != nil { + return fmt.Errorf("write edge config: %w", err) + } + return nil +} + +// fleetRegister creates the FederatedCluster plus a least-privilege +// ServiceAccount/ClusterRole/ClusterRoleBinding scoped to ONLY that one +// FederatedCluster's status subresource, mints a token for the ServiceAccount, +// and returns an edge-config snippet for the site admin. It takes no +// dependency on cobra or REST config, so it is testable with fake clients. +func fleetRegister( + ctx context.Context, + c client.Client, + kube kubernetes.Interface, + in fleetRegisterInput, +) (string, error) { + if in.Name == "" { + return "", fmt.Errorf("fleet register: name is required") + } + + namespace := in.Namespace + if namespace == "" { + namespace = defaultFleetNamespace + } + interval := in.HeartbeatIntervalSeconds + if interval <= 0 { + interval = defaultHeartbeatIntervalSeconds + } + saName := fleetServiceAccountPrefix + in.Name + + fc := &federationv1alpha1.FederatedCluster{ + ObjectMeta: metav1.ObjectMeta{Name: in.Name}, + Spec: federationv1alpha1.FederatedClusterSpec{ + DisplayName: in.Name, + DataResidencyTier: in.ResidencyTier, + HeartbeatIntervalSeconds: interval, + }, + } + if err := c.Create(ctx, fc); err != nil { + return "", fmt.Errorf("create FederatedCluster %q: %w", in.Name, err) + } + + sa := &corev1.ServiceAccount{ + ObjectMeta: metav1.ObjectMeta{Name: saName, Namespace: namespace}, + } + if err := c.Create(ctx, sa); err != nil { + return "", fmt.Errorf("create ServiceAccount %q: %w", saName, err) + } + + // Least privilege: this site's token can ONLY get the FederatedCluster + // object by its own name, and get/update/patch ITS OWN status + // subresource. No list, no watch, no delete, no other resource, no + // cluster-wide (unscoped) access. + clusterRole := &rbacv1.ClusterRole{ + ObjectMeta: metav1.ObjectMeta{Name: saName}, + Rules: []rbacv1.PolicyRule{ + { + APIGroups: []string{federationv1alpha1.GroupVersion.Group}, + Resources: []string{"federatedclusters/status"}, + Verbs: []string{"get", "update", "patch"}, + ResourceNames: []string{in.Name}, + }, + { + APIGroups: []string{federationv1alpha1.GroupVersion.Group}, + Resources: []string{"federatedclusters"}, + Verbs: []string{"get"}, + ResourceNames: []string{in.Name}, + }, + }, + } + if err := c.Create(ctx, clusterRole); err != nil { + return "", fmt.Errorf("create ClusterRole %q: %w", saName, err) + } + + clusterRoleBinding := &rbacv1.ClusterRoleBinding{ + ObjectMeta: metav1.ObjectMeta{Name: saName}, + RoleRef: rbacv1.RoleRef{ + APIGroup: rbacv1.GroupName, + Kind: "ClusterRole", + Name: saName, + }, + Subjects: []rbacv1.Subject{ + {Kind: rbacv1.ServiceAccountKind, Name: saName, Namespace: namespace}, + }, + } + if err := c.Create(ctx, clusterRoleBinding); err != nil { + return "", fmt.Errorf("create ClusterRoleBinding %q: %w", saName, err) + } + + token, expiresAt, err := mintFleetToken(ctx, kube, namespace, saName) + if err != nil { + return "", err + } + + return buildEdgeConfigSnippet(in.Name, in.DatacenterEndpoint, in.CACertData, saName, namespace, token, expiresAt), nil +} + +// buildEdgeConfigSnippet formats the kubeconfig + operator flags a site +// admin needs to bring an edge site's operator up in --federation-role=edge +// mode, pointed at this one newly-minted, narrowly-scoped credential. +// expiresAt is the apiserver's actual token expiration, which is printed and, +// when it falls well short of the ~10y requested (fleetTokenExpirationSeconds), +// flagged with a rotation warning: some datacenter clusters cap token +// lifetime server-side, and the token minted here is a bootstrap credential +// carried to a remote edge site, not something the site admin would +// otherwise think to check. +func buildEdgeConfigSnippet( + name, endpoint string, caData []byte, saName, namespace, token string, expiresAt metav1.Time, +) string { + server := endpoint + if server == "" { + server = datacenterEndpointPlaceholder + } + + var b strings.Builder + fmt.Fprintf(&b, "# Edge site %q registered on the datacenter as FederatedCluster/%s.\n", name, name) + fmt.Fprintf(&b, "# Save the block below as a kubeconfig file on the edge site (for example,\n") + fmt.Fprintf(&b, "# federation-datacenter.kubeconfig), then start the edge operator with the\n") + fmt.Fprintf(&b, "# flags shown after it.\n\n") + + fmt.Fprintf(&b, "apiVersion: v1\n") + fmt.Fprintf(&b, "kind: Config\n") + fmt.Fprintf(&b, "clusters:\n") + fmt.Fprintf(&b, "- cluster:\n") + fmt.Fprintf(&b, " server: %s\n", server) + if len(caData) > 0 { + fmt.Fprintf(&b, " certificate-authority-data: %s\n", base64.StdEncoding.EncodeToString(caData)) + } else { + fmt.Fprintf(&b, " insecure-skip-tls-verify: true # TODO: replace with certificate-authority-data\n") + } + fmt.Fprintf(&b, " name: datacenter\n") + fmt.Fprintf(&b, "contexts:\n") + fmt.Fprintf(&b, "- context:\n") + fmt.Fprintf(&b, " cluster: datacenter\n") + fmt.Fprintf(&b, " user: %s\n", saName) + fmt.Fprintf(&b, " name: datacenter\n") + fmt.Fprintf(&b, "current-context: datacenter\n") + fmt.Fprintf(&b, "users:\n") + fmt.Fprintf(&b, "- name: %s\n", saName) + fmt.Fprintf(&b, " user:\n") + fmt.Fprintf(&b, " token: %s\n\n", token) + + fmt.Fprintf(&b, "# ServiceAccount: %s/%s\n", namespace, saName) + fmt.Fprintf(&b, "# token expires: %s\n", expiresAt.UTC().Format(time.RFC3339)) + if requested := time.Duration(fleetTokenExpirationSeconds) * time.Second; time.Until(expiresAt.Time) < requested/2 { + fmt.Fprintf(&b, "# WARNING: the datacenter cluster capped this token's lifetime well below\n") + fmt.Fprintf(&b, "# the ~10y requested (likely --service-account-max-token-expiration on the\n") + fmt.Fprintf(&b, "# apiserver). Rotate this credential (re-run `llmkube fleet register`) before\n") + fmt.Fprintf(&b, "# it expires.\n") + } + fmt.Fprintf(&b, "# On the edge site's operator, set:\n") + fmt.Fprintf(&b, "# --federation-role=edge --federation-cluster-name=%s "+ + "--federation-datacenter-kubeconfig=\n", name) + + return b.String() +} + +func newFleetStatusCommand() *cobra.Command { + cmd := &cobra.Command{ + Use: "status", + Short: "Show per-site and fleet-wide federation health", + Long: `List every edge site registered as a FederatedCluster, with its +connection phase, GPU capacity, and inference service health, followed by a +fleet-wide summary. + +Run this against the DATACENTER cluster's kubeconfig context.`, + RunE: func(cmd *cobra.Command, args []string) error { + return runFleetStatus(cmd) + }, + } + return cmd +} + +func runFleetStatus(cmd *cobra.Command) error { + ctx := cmd.Context() + if ctx == nil { + ctx = context.Background() + } + + cfg, err := config.GetConfig() + if err != nil { + return fmt.Errorf("failed to get kubeconfig: %w", err) + } + + if err := federationv1alpha1.AddToScheme(scheme.Scheme); err != nil { + return fmt.Errorf("failed to add scheme: %w", err) + } + + k8sClient, err := client.New(cfg, client.Options{Scheme: scheme.Scheme}) + if err != nil { + return fmt.Errorf("failed to create client: %w", err) + } + + return fleetStatus(ctx, k8sClient, cmd.OutOrStdout()) +} + +// fleetStatus lists all FederatedClusters, sorts them by name for +// deterministic output, and writes a per-site table (NAME, TIER, PHASE, LAST +// HEARTBEAT, GPUS, SERVICES) followed by a fleet-wide summary footer that +// aggregates GPU capacity, service health, and a count of sites by phase. +// +// A site that has never pushed status (nil Capacity/Inference) renders as +// dashes in its row rather than panicking, and contributes zero to the +// fleet-wide sums. +func fleetStatus(ctx context.Context, c client.Client, w io.Writer) error { + list := &federationv1alpha1.FederatedClusterList{} + if err := c.List(ctx, list); err != nil { + return fmt.Errorf("failed to list FederatedClusters: %w", err) + } + + sites := list.Items + sort.Slice(sites, func(i, j int) bool { return sites[i].Name < sites[j].Name }) + + tw := tabwriter.NewWriter(w, 0, 0, 2, ' ', 0) + if _, err := fmt.Fprintln(tw, "NAME\tTIER\tPHASE\tLAST HEARTBEAT\tGPUS\tSERVICES"); err != nil { + return fmt.Errorf("failed to write header: %w", err) + } + + var totals fleetStatusTotals + for _, site := range sites { + if err := writeFleetStatusRow(tw, site, &totals); err != nil { + return err + } + } + + if err := tw.Flush(); err != nil { + return fmt.Errorf("failed to flush table: %w", err) + } + + return writeFleetStatusSummary(w, len(sites), totals) +} + +// fleetStatusTotals accumulates the fleet-wide sums and per-phase site counts +// across every FederatedCluster while the per-site table is rendered. +type fleetStatusTotals struct { + gpusAllocatable int32 + gpusTotal int32 + servicesReady int32 + servicesFailed int32 + servicesTotal int32 + phaseCounts map[string]int +} + +func writeFleetStatusRow(tw io.Writer, site federationv1alpha1.FederatedCluster, totals *fleetStatusTotals) error { + tier := site.Spec.DataResidencyTier + if tier == "" { + tier = fleetStatusUnknown + } + phase := site.Status.Phase + if phase == "" { + phase = fleetStatusUnknown + } + if totals.phaseCounts == nil { + totals.phaseCounts = map[string]int{} + } + totals.phaseCounts[phase]++ + + heartbeat := fleetStatusNeverHeartbeat + if site.Status.LastHeartbeatTime != nil { + heartbeat = formatAge(site.Status.LastHeartbeatTime.Time) + } + + gpus := fleetStatusUnknown + "/" + fleetStatusUnknown + if capacity := site.Status.Capacity; capacity != nil { + gpus = fmt.Sprintf("%d/%d", capacity.GPUsAllocatable, capacity.GPUsTotal) + totals.gpusAllocatable += capacity.GPUsAllocatable + totals.gpusTotal += capacity.GPUsTotal + } + + services := strings.Join([]string{fleetStatusUnknown, fleetStatusUnknown, fleetStatusUnknown}, "/") + if inference := site.Status.Inference; inference != nil { + services = fmt.Sprintf("%d/%d/%d", inference.ServicesReady, inference.ServicesFailed, inference.ServicesTotal) + totals.servicesReady += inference.ServicesReady + totals.servicesFailed += inference.ServicesFailed + totals.servicesTotal += inference.ServicesTotal + } + + if _, err := fmt.Fprintf(tw, "%s\t%s\t%s\t%s\t%s\t%s\n", + site.Name, tier, phase, heartbeat, gpus, services, + ); err != nil { + return fmt.Errorf("failed to write row for site %q: %w", site.Name, err) + } + return nil +} + +func writeFleetStatusSummary(w io.Writer, siteCount int, totals fleetStatusTotals) error { + if _, err := fmt.Fprintf(w, "\nFleet-wide: %d site(s) (%d %s, %d %s, %d %s)\n", + siteCount, + totals.phaseCounts[federationv1alpha1.FederatedClusterConnected], federationv1alpha1.FederatedClusterConnected, + totals.phaseCounts[federationv1alpha1.FederatedClusterStale], federationv1alpha1.FederatedClusterStale, + totals.phaseCounts[federationv1alpha1.FederatedClusterUnreachable], federationv1alpha1.FederatedClusterUnreachable, + ); err != nil { + return fmt.Errorf("failed to write fleet-wide summary: %w", err) + } + if _, err := fmt.Fprintf(w, "GPUs: %d/%d allocatable/total\n", totals.gpusAllocatable, totals.gpusTotal); err != nil { + return fmt.Errorf("failed to write GPU summary: %w", err) + } + if _, err := fmt.Fprintf(w, "Services: %d/%d/%d ready/failed/total\n", + totals.servicesReady, totals.servicesFailed, totals.servicesTotal, + ); err != nil { + return fmt.Errorf("failed to write service summary: %w", err) + } + return nil +} diff --git a/pkg/cli/fleet_rbac_403_test.go b/pkg/cli/fleet_rbac_403_test.go new file mode 100644 index 00000000..df992f9e --- /dev/null +++ b/pkg/cli/fleet_rbac_403_test.go @@ -0,0 +1,215 @@ +/* +Copyright 2025. + +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 cli + +import ( + "context" + "os" + "path/filepath" + "testing" + + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/rest" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/envtest" + + federationv1alpha1 "github.com/defilantech/llmkube/api/federation/v1alpha1" +) + +// TestFleetRegisterScopedTokenRBACEnforcement is a RUNTIME proof of the +// design's least-privilege acceptance criterion ("the scoped token provably +// cannot write another cluster's status"), complementing +// TestFleetRegisterCreatesFederatedClusterAndScopedRBAC's proof BY +// CONSTRUCTION (which asserts the ClusterRole is exactly the two +// resourceNames-scoped rules, using a fake client that never evaluates +// RBAC). This test authenticates as the actual minted ServiceAccount token +// against a real, RBAC-enforcing apiserver and asserts on the resulting +// Allow/Forbid decisions. +// +// It runs its own *envtest.Environment, entirely separate from +// internal/controller/suite_test.go, so that suite's admin-access assumption +// is never touched by anything here. +// +// Why a naive token client would wrongly succeed without care: envtest's +// apiserver already defaults --authorization-mode=RBAC (see +// sigs.k8s.io/controller-runtime/pkg/internal/testing/controlplane/ +// apiserver.go, defaultArgs()); RBAC is not something this test has to turn +// on. The actual trap is that the *rest.Config envtest.Start() returns +// authenticates via a client certificate whose group is "system:masters", +// which bypasses RBAC entirely (kube-apiserver treats system:masters as +// superuser). So the load-bearing step here is scopedRestConfig: build a +// SEPARATE config carrying ONLY the minted ServiceAccount bearer token, with +// every client-cert field stripped so the request can't ride the admin +// identity in through mutual TLS. +func TestFleetRegisterScopedTokenRBACEnforcement(t *testing.T) { + binDir := envtestBinaryDir(t) + + testScheme := fleetTestScheme(t) + + env := &envtest.Environment{ + CRDDirectoryPaths: []string{filepath.Join("..", "..", "config", "crd", "bases")}, + ErrorIfCRDPathMissing: true, + BinaryAssetsDirectory: binDir, + } + + cfg, err := env.Start() + if err != nil { + t.Fatalf("start envtest environment: %v", err) + } + t.Cleanup(func() { + if err := env.Stop(); err != nil { + t.Errorf("stop envtest environment: %v", err) + } + }) + + adminClient, err := client.New(cfg, client.Options{Scheme: testScheme}) + if err != nil { + t.Fatalf("build admin client: %v", err) + } + kube, err := kubernetes.NewForConfig(cfg) + if err != nil { + t.Fatalf("build clientset: %v", err) + } + + ctx := context.Background() + + ns := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: defaultFleetNamespace}} + if err := adminClient.Create(ctx, ns); err != nil { + t.Fatalf("create namespace %q: %v", defaultFleetNamespace, err) + } + + // Register edge-a and edge-b through the REAL production path + // (fleetRegister, unmodified), so the ClusterRole this test exercises is + // the exact one fleet.go ships, not a hand-rolled lookalike. + if _, err := fleetRegister(ctx, adminClient, kube, fleetRegisterInput{Name: "edge-a"}); err != nil { + t.Fatalf("fleetRegister(edge-a): %v", err) + } + // edge-b only needs to exist as a FederatedCluster that edge-a's token + // must NOT be able to touch; its own RBAC/token are irrelevant here. + if _, err := fleetRegister(ctx, adminClient, kube, fleetRegisterInput{Name: "edge-b"}); err != nil { + t.Fatalf("fleetRegister(edge-b): %v", err) + } + + // Mint a fresh token for edge-a's ServiceAccount via the same production + // seam fleetRegister just used (mintFleetToken, unstubbed here), rather + // than parsing one out of the printed edge-config snippet. + token, _, err := mintFleetToken(ctx, kube, defaultFleetNamespace, fleetServiceAccountPrefix+"edge-a") + if err != nil { + t.Fatalf("mint token for edge-a's ServiceAccount: %v", err) + } + + scopedClient, err := client.New(scopedRestConfig(cfg, token), client.Options{Scheme: testScheme}) + if err != nil { + t.Fatalf("build scoped client: %v", err) + } + + // --- (1) OWN status patch: allowed. --- + edgeA := &federationv1alpha1.FederatedCluster{} + if err := scopedClient.Get(ctx, types.NamespacedName{Name: "edge-a"}, edgeA); err != nil { + t.Fatalf("scoped client get edge-a (allowed by the `get` rule on federatedclusters): %v", err) + } + patchOwnStatus := client.MergeFrom(edgeA.DeepCopy()) + edgeA.Status.ObservedVersion = "rbac-403-test" + if err := scopedClient.Status().Patch(ctx, edgeA, patchOwnStatus); err != nil { + t.Errorf("scoped client patch OWN (edge-a) status: want success, got error: %v", err) + } + + // --- (2) OTHER cluster's status patch: forbidden. --- + edgeB := &federationv1alpha1.FederatedCluster{} + if err := adminClient.Get(ctx, types.NamespacedName{Name: "edge-b"}, edgeB); err != nil { + t.Fatalf("admin client get edge-b: %v", err) + } + patchOtherStatus := client.MergeFrom(edgeB.DeepCopy()) + edgeB.Status.ObservedVersion = "rbac-403-test" + err = scopedClient.Status().Patch(ctx, edgeB, patchOtherStatus) + assertForbidden(t, err, "scoped client patch OTHER (edge-b) status") + + // --- (3) list federatedclusters: forbidden (no list verb granted). --- + list := &federationv1alpha1.FederatedClusterList{} + err = scopedClient.List(ctx, list) + assertForbidden(t, err, "scoped client list FederatedClusters") + + // --- (4) OWN spec patch: forbidden (only the status subresource is + // get/update/patch-able; the base resource's only granted verb is get). --- + edgeASpec := &federationv1alpha1.FederatedCluster{} + if err := adminClient.Get(ctx, types.NamespacedName{Name: "edge-a"}, edgeASpec); err != nil { + t.Fatalf("admin client get edge-a for spec patch: %v", err) + } + patchOwnSpec := client.MergeFrom(edgeASpec.DeepCopy()) + edgeASpec.Spec.DisplayName = "renamed-by-rbac-403-test" + err = scopedClient.Patch(ctx, edgeASpec, patchOwnSpec) + assertForbidden(t, err, "scoped client patch OWN (edge-a) spec") +} + +// assertForbidden fails the test unless err is a genuine RBAC-authorizer +// 403, i.e. apierrors.IsForbidden(err). A nil error, or any other error +// shape (NotFound, connection failure, etc.), means the assertion didn't +// actually prove what it claims to prove. +func assertForbidden(t *testing.T, err error, what string) { + t.Helper() + if err == nil { + t.Errorf("%s: want Forbidden, got success", what) + return + } + if !apierrors.IsForbidden(err) { + t.Errorf("%s: want apierrors.IsForbidden, got %v (%T)", what, err, err) + } +} + +// scopedRestConfig returns a copy of cfg (envtest's admin config) +// re-authenticated as ONLY the given bearer token. It strips every +// client-certificate and other identity field: if a client cert were left +// in place, the request would authenticate via mutual TLS as the cert's +// identity (system:masters, for envtest's admin config) rather than the +// token, silently defeating the entire point of this test by letting the +// "scoped" client ride in on the admin identity. +func scopedRestConfig(cfg *rest.Config, token string) *rest.Config { + scoped := rest.CopyConfig(cfg) + scoped.BearerToken = token + scoped.BearerTokenFile = "" + scoped.Username = "" + scoped.Password = "" + scoped.CertData = nil + scoped.CertFile = "" + scoped.KeyData = nil + scoped.KeyFile = "" + scoped.AuthProvider = nil + scoped.ExecProvider = nil + return scoped +} + +// envtestBinaryDir locates the envtest KUBEBUILDER_ASSETS directory: the env +// var `make test` exports, or (for ad hoc/IDE runs, mirroring +// internal/controller/suite_test.go's getFirstFoundEnvTestBinaryDir) the +// first entry under ../../bin/k8s. Skips (rather than fails) when neither is +// present, so `go test ./pkg/cli/...` run without `make envtest` first +// doesn't hard-fail the whole package. +func envtestBinaryDir(t *testing.T) string { + t.Helper() + if dir := os.Getenv("KUBEBUILDER_ASSETS"); dir != "" { + return dir + } + matches, err := filepath.Glob(filepath.Join("..", "..", "bin", "k8s", "*")) + if err != nil || len(matches) == 0 { + t.Skip("no envtest binaries found: run `make envtest` first (see internal/controller/suite_test.go)") + } + return matches[0] +} diff --git a/pkg/cli/fleet_test.go b/pkg/cli/fleet_test.go new file mode 100644 index 00000000..afa042c7 --- /dev/null +++ b/pkg/cli/fleet_test.go @@ -0,0 +1,464 @@ +/* +Copyright 2025. + +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 cli + +import ( + "bytes" + "context" + "sort" + "strings" + "testing" + "time" + + "github.com/spf13/cobra" + corev1 "k8s.io/api/core/v1" + rbacv1 "k8s.io/api/rbac/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/kubernetes" + clientgoscheme "k8s.io/client-go/kubernetes/scheme" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + federationv1alpha1 "github.com/defilantech/llmkube/api/federation/v1alpha1" +) + +// fleetTestScheme builds a scheme with core, rbac, and federation types +// registered, matching what the real CLI client wires up in runFleetRegister. +func fleetTestScheme(t *testing.T) *runtime.Scheme { + t.Helper() + s := runtime.NewScheme() + if err := clientgoscheme.AddToScheme(s); err != nil { + t.Fatalf("add client-go scheme: %v", err) + } + if err := federationv1alpha1.AddToScheme(s); err != nil { + t.Fatalf("add federation scheme: %v", err) + } + return s +} + +// stubToken swaps mintFleetToken for a fake that returns token without +// touching a real (or fake) TokenRequest subresource. client-go's fake +// clientset echoes the TokenRequest object back unchanged instead of +// synthesizing a token value, so it can't be used to assert on the minted +// token itself; mintFleetToken is a seam for exactly this reason. +// +// The stubbed expiration matches what a real apiserver would return when it +// honors the requested ~10y ExpirationSeconds in full, so tests that don't +// care about token expiry (most of them) don't trip the capped-token +// rotation warning. Tests that DO care use stubTokenWithExpiry directly. +func stubToken(t *testing.T, token string) { + t.Helper() + stubTokenWithExpiry(t, token, metav1.NewTime(time.Now().Add(time.Duration(fleetTokenExpirationSeconds)*time.Second))) +} + +// stubTokenWithExpiry is stubToken plus an explicit ExpirationTimestamp, for +// tests asserting on the printed token-expiry line and capped-lifetime +// warning in the edge-config snippet. +func stubTokenWithExpiry(t *testing.T, token string, expiresAt metav1.Time) { + t.Helper() + prev := mintFleetToken + mintFleetToken = func(_ context.Context, _ kubernetes.Interface, _, _ string) (string, metav1.Time, error) { + return token, expiresAt, nil + } + t.Cleanup(func() { mintFleetToken = prev }) +} + +func TestFleetRegisterCreatesFederatedClusterAndScopedRBAC(t *testing.T) { + stubToken(t, "test-token-123") + + c := fake.NewClientBuilder().WithScheme(fleetTestScheme(t)).Build() + ctx := context.Background() + + edgeConfig, err := fleetRegister(ctx, c, nil, fleetRegisterInput{ + Name: "edge-a", + ResidencyTier: "floor-3", + HeartbeatIntervalSeconds: 45, + DatacenterEndpoint: "https://dc.example.com:6443", + Namespace: "llmkube-system", + }) + if err != nil { + t.Fatalf("fleetRegister: %v", err) + } + + // FederatedCluster: spec tier + interval. + fc := &federationv1alpha1.FederatedCluster{} + if err := c.Get(ctx, types.NamespacedName{Name: "edge-a"}, fc); err != nil { + t.Fatalf("get FederatedCluster: %v", err) + } + if fc.Spec.DataResidencyTier != "floor-3" { + t.Errorf("DataResidencyTier = %q, want %q", fc.Spec.DataResidencyTier, "floor-3") + } + if fc.Spec.HeartbeatIntervalSeconds != 45 { + t.Errorf("HeartbeatIntervalSeconds = %d, want 45", fc.Spec.HeartbeatIntervalSeconds) + } + + // ServiceAccount fedcluster-edge-a in the target namespace. + sa := &corev1.ServiceAccount{} + if err := c.Get(ctx, types.NamespacedName{Name: "fedcluster-edge-a", Namespace: "llmkube-system"}, sa); err != nil { + t.Fatalf("get ServiceAccount: %v", err) + } + + // ClusterRole: EXACTLY the least-privilege rules. No list/watch/delete, + // no other resources, no cluster-wide (unscoped) access. + cr := &rbacv1.ClusterRole{} + if err := c.Get(ctx, types.NamespacedName{Name: "fedcluster-edge-a"}, cr); err != nil { + t.Fatalf("get ClusterRole: %v", err) + } + if len(cr.Rules) != 2 { + t.Fatalf("ClusterRole has %d rules, want exactly 2: %+v", len(cr.Rules), cr.Rules) + } + + statusRule, listGetRule := findRules(t, cr.Rules) + + assertStringSlice(t, "status rule APIGroups", statusRule.APIGroups, []string{"federation.llmkube.dev"}) + assertStringSlice(t, "status rule Resources", statusRule.Resources, []string{"federatedclusters/status"}) + assertStringSlice(t, "status rule Verbs", statusRule.Verbs, []string{"get", "update", "patch"}) + assertStringSlice(t, "status rule ResourceNames", statusRule.ResourceNames, []string{"edge-a"}) + + assertStringSlice(t, "get rule APIGroups", listGetRule.APIGroups, []string{"federation.llmkube.dev"}) + assertStringSlice(t, "get rule Resources", listGetRule.Resources, []string{"federatedclusters"}) + assertStringSlice(t, "get rule Verbs", listGetRule.Verbs, []string{"get"}) + assertStringSlice(t, "get rule ResourceNames", listGetRule.ResourceNames, []string{"edge-a"}) + + // ClusterRoleBinding: binds the ClusterRole to the ServiceAccount. + crb := &rbacv1.ClusterRoleBinding{} + if err := c.Get(ctx, types.NamespacedName{Name: "fedcluster-edge-a"}, crb); err != nil { + t.Fatalf("get ClusterRoleBinding: %v", err) + } + if crb.RoleRef.Kind != "ClusterRole" || crb.RoleRef.Name != "fedcluster-edge-a" { + t.Errorf("RoleRef = %+v, want ClusterRole/fedcluster-edge-a", crb.RoleRef) + } + if len(crb.Subjects) != 1 { + t.Fatalf("Subjects = %+v, want exactly 1", crb.Subjects) + } + subj := crb.Subjects[0] + if subj.Kind != rbacv1.ServiceAccountKind || subj.Name != "fedcluster-edge-a" || subj.Namespace != "llmkube-system" { + t.Errorf("Subject = %+v, want ServiceAccount fedcluster-edge-a/llmkube-system", subj) + } + + // Returned edge-config snippet: cluster name, datacenter endpoint, and + // the operator flags a site admin sets. + for _, want := range []string{ + "edge-a", + "https://dc.example.com:6443", + "test-token-123", + "--federation-role=edge", + "--federation-cluster-name=edge-a", + "--federation-datacenter-kubeconfig=", + } { + if !strings.Contains(edgeConfig, want) { + t.Errorf("edge config snippet missing %q:\n%s", want, edgeConfig) + } + } +} + +// findRules splits a 2-rule ClusterRole into the federatedclusters/status +// rule and the federatedclusters rule, regardless of order, and fails the +// test if the rules don't match that shape. +func findRules(t *testing.T, rules []rbacv1.PolicyRule) (statusRule, getRule rbacv1.PolicyRule) { + t.Helper() + var foundStatus, foundGet bool + for _, r := range rules { + switch { + case len(r.Resources) == 1 && r.Resources[0] == "federatedclusters/status": + statusRule, foundStatus = r, true + case len(r.Resources) == 1 && r.Resources[0] == "federatedclusters": + getRule, foundGet = r, true + default: + t.Fatalf("unexpected rule with broader/unknown resources: %+v", r) + } + } + if !foundStatus || !foundGet { + t.Fatalf("rules missing expected shape: %+v", rules) + } + return statusRule, getRule +} + +func assertStringSlice(t *testing.T, label string, got, want []string) { + t.Helper() + gotSorted := append([]string(nil), got...) + wantSorted := append([]string(nil), want...) + sort.Strings(gotSorted) + sort.Strings(wantSorted) + if len(gotSorted) != len(wantSorted) { + t.Errorf("%s = %v, want %v", label, got, want) + return + } + for i := range gotSorted { + if gotSorted[i] != wantSorted[i] { + t.Errorf("%s = %v, want %v", label, got, want) + return + } + } +} + +func TestFleetRegisterDefaultsNamespaceAndInterval(t *testing.T) { + stubToken(t, "tok") + c := fake.NewClientBuilder().WithScheme(fleetTestScheme(t)).Build() + ctx := context.Background() + + if _, err := fleetRegister(ctx, c, nil, fleetRegisterInput{Name: "edge-b"}); err != nil { + t.Fatalf("fleetRegister: %v", err) + } + + fc := &federationv1alpha1.FederatedCluster{} + if err := c.Get(ctx, types.NamespacedName{Name: "edge-b"}, fc); err != nil { + t.Fatalf("get FederatedCluster: %v", err) + } + if fc.Spec.HeartbeatIntervalSeconds != defaultHeartbeatIntervalSeconds { + t.Errorf("HeartbeatIntervalSeconds = %d, want default %d", + fc.Spec.HeartbeatIntervalSeconds, defaultHeartbeatIntervalSeconds) + } + + sa := &corev1.ServiceAccount{} + saKey := types.NamespacedName{Name: "fedcluster-edge-b", Namespace: defaultFleetNamespace} + if err := c.Get(ctx, saKey, sa); err != nil { + t.Fatalf("get ServiceAccount in default namespace %q: %v", defaultFleetNamespace, err) + } +} + +func TestFleetRegisterRequiresName(t *testing.T) { + c := fake.NewClientBuilder().WithScheme(fleetTestScheme(t)).Build() + if _, err := fleetRegister(context.Background(), c, nil, fleetRegisterInput{}); err == nil { + t.Fatal("expected error for empty name") + } +} + +// TestFleetRegisterSurfacesTokenExpiry seeds mintFleetToken to return the +// apiserver honoring the requested ~10y ExpirationSeconds in full, and +// asserts the real expiry is printed in the edge-config snippet with no +// capped-lifetime warning (the apiserver didn't cap anything). +func TestFleetRegisterSurfacesTokenExpiry(t *testing.T) { + uncapped := metav1.NewTime(time.Now().Add(time.Duration(fleetTokenExpirationSeconds) * time.Second)) + stubTokenWithExpiry(t, "uncapped-token", uncapped) + + c := fake.NewClientBuilder().WithScheme(fleetTestScheme(t)).Build() + edgeConfig, err := fleetRegister(context.Background(), c, nil, fleetRegisterInput{Name: "edge-e"}) + if err != nil { + t.Fatalf("fleetRegister: %v", err) + } + + wantLine := "# token expires: " + uncapped.UTC().Format(time.RFC3339) + if !strings.Contains(edgeConfig, wantLine) { + t.Errorf("edge config snippet missing %q:\n%s", wantLine, edgeConfig) + } + if strings.Contains(edgeConfig, "WARNING") { + t.Errorf("edge config snippet should not warn when the apiserver honored the requested expiration:\n%s", edgeConfig) + } +} + +// TestFleetRegisterWarnsOnCappedTokenExpiry seeds mintFleetToken to return an +// ExpirationTimestamp far short of the ~10y requested (simulating a +// datacenter apiserver running with --service-account-max-token-expiration), +// and asserts both the real (capped) expiry and a rotation warning appear in +// the edge-config snippet. +func TestFleetRegisterWarnsOnCappedTokenExpiry(t *testing.T) { + capped := metav1.NewTime(time.Now().Add(24 * time.Hour)) + stubTokenWithExpiry(t, "capped-token", capped) + + c := fake.NewClientBuilder().WithScheme(fleetTestScheme(t)).Build() + edgeConfig, err := fleetRegister(context.Background(), c, nil, fleetRegisterInput{Name: "edge-f"}) + if err != nil { + t.Fatalf("fleetRegister: %v", err) + } + + wantLine := "# token expires: " + capped.UTC().Format(time.RFC3339) + if !strings.Contains(edgeConfig, wantLine) { + t.Errorf("edge config snippet missing %q:\n%s", wantLine, edgeConfig) + } + if !strings.Contains(edgeConfig, "WARNING") || !strings.Contains(edgeConfig, "capped") { + t.Errorf("edge config snippet should warn that the cluster capped the token lifetime:\n%s", edgeConfig) + } +} + +func TestNewFleetCommandWiring(t *testing.T) { + cmd := NewFleetCommand() + if cmd.Name() != "fleet" { + t.Fatalf("Name() = %q, want fleet", cmd.Name()) + } + + var register *cobra.Command + for _, sub := range cmd.Commands() { + if sub.Name() == "register" { + register = sub + break + } + } + if register == nil { + t.Fatal("fleet command is missing the register subcommand") + } + + flag := register.Flags().Lookup("heartbeat-interval") + if flag == nil { + t.Fatal("register command is missing --heartbeat-interval flag") + } + if flag.DefValue != "30" { + t.Errorf("--heartbeat-interval default = %q, want 30", flag.DefValue) + } + + for _, name := range []string{"name", "residency", "datacenter-endpoint"} { + if register.Flags().Lookup(name) == nil { + t.Errorf("register command is missing --%s flag", name) + } + } +} + +func TestNewFleetCommandHasStatusSubcommand(t *testing.T) { + cmd := NewFleetCommand() + + var status *cobra.Command + for _, sub := range cmd.Commands() { + if sub.Name() == "status" { + status = sub + break + } + } + if status == nil { + t.Fatal("fleet command is missing the status subcommand") + } +} + +// TestFleetStatusRendersPerSiteAndFleetWideTable seeds three FederatedClusters +// covering the three phases (one Connected with real capacity/inference, one +// Stale, one Unreachable with never-pushed nil Capacity/Inference), and +// asserts fleetStatus renders a deterministic (sorted by name) per-site table +// plus a fleet-wide footer that aggregates GPUs, services, and phase counts. +// The Unreachable/nil-Capacity site is the regression case for the nil-guard: +// it must render zeros/dashes, not panic. +func TestFleetStatusRendersPerSiteAndFleetWideTable(t *testing.T) { + now := metav1.NewTime(time.Now().Add(-90 * time.Second)) + stale := metav1.NewTime(time.Now().Add(-10 * time.Minute)) + + sites := []federationv1alpha1.FederatedCluster{ + { + ObjectMeta: metav1.ObjectMeta{Name: "site-a"}, + Spec: federationv1alpha1.FederatedClusterSpec{ + DataResidencyTier: "eu", + }, + Status: federationv1alpha1.FederatedClusterStatus{ + Phase: federationv1alpha1.FederatedClusterConnected, + LastHeartbeatTime: &now, + Capacity: &federationv1alpha1.ClusterCapacity{ + Nodes: 3, + GPUsTotal: 8, + GPUsAllocatable: 6, + }, + Inference: &federationv1alpha1.ClusterInferenceSummary{ + ServicesReady: 10, + ServicesFailed: 1, + ServicesTotal: 12, + Models: 4, + }, + }, + }, + { + ObjectMeta: metav1.ObjectMeta{Name: "site-b"}, + Spec: federationv1alpha1.FederatedClusterSpec{ + DataResidencyTier: "floor-2", + }, + Status: federationv1alpha1.FederatedClusterStatus{ + Phase: federationv1alpha1.FederatedClusterStale, + LastHeartbeatTime: &stale, + Capacity: &federationv1alpha1.ClusterCapacity{ + Nodes: 2, + GPUsTotal: 4, + GPUsAllocatable: 2, + }, + Inference: &federationv1alpha1.ClusterInferenceSummary{ + ServicesReady: 3, + ServicesFailed: 2, + ServicesTotal: 5, + Models: 2, + }, + }, + }, + { + // Deliberately named so sort-by-name puts it first, to prove the + // output is sorted rather than insertion-ordered. + ObjectMeta: metav1.ObjectMeta{Name: "aaa-never-reported"}, + Spec: federationv1alpha1.FederatedClusterSpec{ + DataResidencyTier: "edge-1", + }, + Status: federationv1alpha1.FederatedClusterStatus{ + Phase: federationv1alpha1.FederatedClusterUnreachable, + // LastHeartbeatTime, Capacity, and Inference are all nil: + // this site has never pushed status. + }, + }, + } + + c := fake.NewClientBuilder().WithScheme(fleetTestScheme(t)).WithObjects( + &sites[0], &sites[1], &sites[2], + ).Build() + + var buf bytes.Buffer + if err := fleetStatus(context.Background(), c, &buf); err != nil { + t.Fatalf("fleetStatus: %v (should never panic or error on nil Capacity/Inference)", err) + } + out := buf.String() + + lines := strings.Split(out, "\n") + if len(lines) < 4 { + t.Fatalf("expected header + 3 site rows, got %d lines:\n%s", len(lines), out) + } + + // Sorted by name: "aaa-never-reported" < "site-a" < "site-b". + if !strings.HasPrefix(strings.TrimSpace(lines[0]), "NAME") { + t.Errorf("line 0 = %q, want header starting with NAME", lines[0]) + } + if got := lines[1]; !(strings.Contains(got, "aaa-never-reported") && strings.Contains(got, "Unreachable")) { + t.Errorf("line 1 = %q, want the never-reported/Unreachable site sorted first", got) + } + if got := lines[2]; !strings.Contains(got, "site-a") { + t.Errorf("line 2 = %q, want site-a second", got) + } + if got := lines[3]; !strings.Contains(got, "site-b") { + t.Errorf("line 3 = %q, want site-b third", got) + } + + // Per-site fields. + for _, want := range []string{ + // aaa-never-reported: Unreachable, edge-1 tier, nil Capacity/Inference + // must render as zeros/dashes, never panic and never blank. + "edge-1", "Unreachable", "never", + // site-a: Connected, eu tier, real capacity (6 allocatable/8 total) + // and inference (10 ready/1 failed/12 total). + "site-a", "eu", "Connected", "6/8", "10/1/12", + // site-b: Stale, floor-2 tier, capacity (2/4) and inference (3/2/5). + "site-b", "floor-2", "Stale", "2/4", "3/2/5", + } { + if !strings.Contains(out, want) { + t.Errorf("output missing %q:\n%s", want, out) + } + } + + // Fleet-wide footer: summed GPUs (6+2+0=8 allocatable, 8+4+0=12 total), + // summed services (10+3+0=13 ready, 1+2+0=3 failed, 12+5+0=17 total), + // and per-phase site counts (1 Connected, 1 Stale, 1 Unreachable). + for _, want := range []string{ + "8/12", + "13/3/17", + "1 Connected", + "1 Stale", + "1 Unreachable", + } { + if !strings.Contains(out, want) { + t.Errorf("footer missing %q:\n%s", want, out) + } + } +} diff --git a/pkg/cli/root.go b/pkg/cli/root.go index 6bebceda..11c4d30d 100644 --- a/pkg/cli/root.go +++ b/pkg/cli/root.go @@ -52,6 +52,7 @@ built-in observability, SLO enforcement, and edge-native capabilities.`, cmd.AddCommand(NewLicenseCommand()) cmd.AddCommand(NewAuditCommand()) cmd.AddCommand(NewForemanCommand()) + cmd.AddCommand(NewFleetCommand()) return cmd }