From 08f1078c0bf30cd72aeb5c3b6b2bb5732bcda493 Mon Sep 17 00:00:00 2001 From: thxCode Date: Mon, 26 Oct 2026 05:10:19 +0800 Subject: [PATCH 1/5] feat(worker): reconcile direct local routing EndpointSlices Signed-off-by: thxCode --- .../templates/worker/serviceaccount.yaml | 8 + docs/README.md | 2 +- docs/contribute/internals.md | 35 + pkg/systemkuberess/endpoint_slice.go | 224 ++++++ pkg/systemkuberess/service.go | 217 ++++-- pkg/systemkuberess/service_test.go | 641 ++++++++++++++++++ 6 files changed, 1072 insertions(+), 55 deletions(-) create mode 100644 pkg/systemkuberess/endpoint_slice.go create mode 100644 pkg/systemkuberess/service_test.go diff --git a/deploy/gpustack-operator/chart/templates/worker/serviceaccount.yaml b/deploy/gpustack-operator/chart/templates/worker/serviceaccount.yaml index e524dfa30..ff935f243 100644 --- a/deploy/gpustack-operator/chart/templates/worker/serviceaccount.yaml +++ b/deploy/gpustack-operator/chart/templates/worker/serviceaccount.yaml @@ -11,6 +11,14 @@ metadata: {{- toYaml . | nindent 4 }} {{- end }} --- +{{- /* + The worker is bound to cluster-admin, so this binding already covers the routing + resources it manages: the service and the legacy endpoints of the system routing, + and the endpoint slices of the discovery.k8s.io group it writes when it runs + outside a nearby cluster. The grant those operations need is get, list, create, + update and delete on endpointslices, plus the same set on services and endpoints. + A narrower binding has to keep all of them, or the worker stops routing. +*/}} apiVersion: rbac.authorization.k8s.io/v1 kind: ClusterRoleBinding metadata: diff --git a/docs/README.md b/docs/README.md index c260d3c11..4d51e6bbb 100644 --- a/docs/README.md +++ b/docs/README.md @@ -131,7 +131,7 @@ A skill link names a procedure; the entry's other fields carry shared facts. - Use for: which version a manifest or call targets. User-facing resources are `worker.gpustack.ai/v1`; `worker.gpustack.ai/v1alpha1` is internal controller storage behind the public API, not a version users switch between. The layers: public types, the handlers that proxy them onto internal storage, code generation, and the RBAC that grants access. - Aliases: `worker.gpustack.ai/v1`, `v1alpha1`, `CRD`, `aggregated apiserver` - Guides: the version table in [Development](contribute/development.md); a public surface example with its grants in [Model Artifact API](modules/model-delivery/views.md). -- Code: public types in [api/worker/v1](../api/worker/v1/) and internal storage in [api/worker/v1alpha1](../api/worker/v1alpha1/) · handlers and proxies in [pkg/worker/extensionapis](../pkg/worker/extensionapis/) over the [pkg/extensionapi](../pkg/extensionapi/) plumbing · generators in [gen/api](../gen/api/) · worker RBAC in [deploy/gpustack-operator/chart/templates/worker/serviceaccount.yaml](../deploy/gpustack-operator/chart/templates/worker/serviceaccount.yaml) +- Code: public types in [api/worker/v1](../api/worker/v1/) and internal storage in [api/worker/v1alpha1](../api/worker/v1alpha1/) · handlers and proxies in [pkg/worker/extensionapis](../pkg/worker/extensionapis/) over the [pkg/extensionapi](../pkg/extensionapi/) plumbing · the routing objects the extension API services are fronted by in [pkg/systemkuberess](../pkg/systemkuberess/) · generators in [gen/api](../gen/api/) · worker RBAC in [deploy/gpustack-operator/chart/templates/worker/serviceaccount.yaml](../deploy/gpustack-operator/chart/templates/worker/serviceaccount.yaml) - Related: [Development](#development) (`make generate` regenerates the API surface) · [GPU Instances](#gpu-instances) (Instance and InstanceType are the main public resources) - Skills: [gpustack-operator-generate](../.agents/skills/gpustack-operator-generate/SKILL.md) (regenerate after type or webhook edits) diff --git a/docs/contribute/internals.md b/docs/contribute/internals.md index 6592a4307..f755b8c21 100644 --- a/docs/contribute/internals.md +++ b/docs/contribute/internals.md @@ -92,6 +92,41 @@ last resort, not a pattern to copy; reach for it only where idempotence is out o > **Why** — Helm's release storage is a compare-and-create, not a mutex, and two Helm actions on one > release can leave it pending where no later attempt gets past. +### System routing resources + +A worker outside a nearby cluster has no pod for the chart's `Service` to select, so it installs the +routing objects itself (`pkg/systemkuberess`), in the system namespace: + +- a selectorless `Service` with a `https` port; +- legacy `Endpoints` of the same name, holding the address of the system; +- an `EndpointSlice` in `discovery.k8s.io/v1`, holding the same address and port. + +Three environment variables configure the routing at worker startup: + +| Variable | Default | Purpose | +|---|---|---| +| `KUBERNETES_POD_NAMESPACE` | `gpustack-system` | Namespace of the routing `Service`, `Endpoints`, `EndpointSlice`, and extension `APIService` backends. | +| `KUBERNETES_SERVICE_NAME` | `gpustack-operator-worker` | Name of the routing `Service` and legacy `Endpoints` (`-direct` for the `EndpointSlice`). | +| `KUBERNETES_POD_IP` | primary host interface IP | Reachable worker IP written to the routing `Endpoints` and `EndpointSlice`. | + +With its `--enable-aggregator-routing` flag, the kube-apiserver resolves the backend of an extension +API service from the `EndpointSlice`; without it, the backend is the cluster IP of the `Service`. + +The slice therefore carries the `kubernetes.io/service-name` label of the `Service`, a port named +after the `Service` port, and a ready endpoint on the address of the system. Its address type +follows the family of that address, and the endpoint slice controller leaves it alone because its +`endpointslice.kubernetes.io/managed-by` label names this operator. + +The legacy `Endpoints` stay for the aggregators that read them, which are the ones older than +Kubernetes 1.34. They carry the `endpointslice.kubernetes.io/skip-mirror` label, so the mirroring +controller neither mirrors them nor leaves a second slice behind that the aggregator could pick +instead. + +The reconcile replaces the address and the port it owns instead of appending them, and removes the +slices it owns and no longer needs: the mirror of the legacy `Endpoints`, and a slice an earlier +version left under another name. It removes a slice only when the owner reference names the routing +`Service` or the legacy `Endpoints` of this worker, so a slice of another service stays. + ## Worker gateway `pkg/workergateway/service` folds many clusters' `InstanceType`s into one fleet-wide diff --git a/pkg/systemkuberess/endpoint_slice.go b/pkg/systemkuberess/endpoint_slice.go new file mode 100644 index 000000000..3a96a7616 --- /dev/null +++ b/pkg/systemkuberess/endpoint_slice.go @@ -0,0 +1,224 @@ +package systemkuberess + +import ( + "context" + "fmt" + "net" + + core "k8s.io/api/core/v1" + discovery "k8s.io/api/discovery/v1" + apiequality "k8s.io/apimachinery/pkg/api/equality" + kerrors "k8s.io/apimachinery/pkg/api/errors" + meta "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/utils/ptr" + + "gpustack.ai/gpustack/pkg/kubeclients/kubernetes" + "gpustack.ai/gpustack/pkg/kubeclientset" + "gpustack.ai/gpustack/pkg/kubemeta" +) + +const ( + // routingManagedBy is the manager of the endpoint slice of the system routing. + // + // It is a value of its own, and not the one of the endpoint slice controller or + // the one of the endpoint slice mirroring controller, so that neither of them + // takes the slice over, and so that the slice of this operator can be told from + // a mirrored one. + routingManagedBy = "gpustack-operator-worker" + + // routingEndpointSliceNameSuffix is the suffix of the name of the endpoint slice + // of the system routing. + // + // The slice is named apart from the service it serves, because a mirrored slice + // of the legacy endpoints is named after the endpoints with a short random + // suffix, and the two must never be the same object. + routingEndpointSliceNameSuffix = "-direct" +) + +// newRoutingEndpointSlice builds the desired endpoint slice of the system routing. +// +// The aggregator reads the slices that carry the name of the service, and takes the +// first one whose port is named after the service port and whose endpoint is ready. +// The slice therefore names the service, names the port of the service and carries a +// ready endpoint on the address of the system. +func newRoutingEndpointSlice(nsName, svcName, address string, port int32) (*discovery.EndpointSlice, error) { + addressType, err := routingAddressType(address) + if err != nil { + return nil, err + } + + ready, serving, terminating := true, true, false + + return &discovery.EndpointSlice{ + ObjectMeta: meta.ObjectMeta{ + Namespace: nsName, + Name: routingEndpointSliceName(svcName), + Labels: map[string]string{ + discovery.LabelServiceName: svcName, + discovery.LabelManagedBy: routingManagedBy, + routingLabel: routingLabelValue, + }, + }, + AddressType: addressType, + Endpoints: []discovery.Endpoint{ + { + Addresses: []string{address}, + Conditions: discovery.EndpointConditions{ + Ready: &ready, + Serving: &serving, + Terminating: &terminating, + }, + }, + }, + Ports: []discovery.EndpointPort{ + { + Name: ptr.To(routingPortName), + Port: ptr.To(port), + Protocol: ptr.To(core.ProtocolTCP), + }, + }, + }, nil +} + +// routingEndpointSliceName returns the name of the endpoint slice of the system routing. +func routingEndpointSliceName(svcName string) string { + return svcName + routingEndpointSliceNameSuffix +} + +// routingAddressType returns the address type of an endpoint slice that can carry the +// given address. +// +// The aggregator prefers the slices of the primary address family of the service, and +// an endpoint slice only accepts the addresses of its own type, so the type has to +// follow the address. +func routingAddressType(address string) (discovery.AddressType, error) { + ip := net.ParseIP(address) + if ip == nil { + return "", fmt.Errorf("routing address %q is not an IP address", address) + } + if ip.To4() != nil { + return discovery.AddressTypeIPv4, nil + } + return discovery.AddressTypeIPv6, nil +} + +// installRoutingEndpointSlice creates or updates the endpoint slice of the system +// routing. +// +// The address type of an endpoint slice cannot be changed, so a slice that was made for +// the other address family is replaced instead of updated. +func installRoutingEndpointSlice(ctx context.Context, cli kubernetes.Interface, nsName string, + desired *discovery.EndpointSlice, svc *core.Service, +) error { + sliceCli := cli.DiscoveryV1().EndpointSlices(nsName) + + actual, err := sliceCli.Get(ctx, desired.GetName(), meta.GetOptions{}) + switch { + case err == nil: + if actual.GetLabels()[discovery.LabelManagedBy] != routingManagedBy { + return fmt.Errorf("routing endpoint slice %q is managed by %q, not by this operator", + desired.GetName(), actual.GetLabels()[discovery.LabelManagedBy]) + } + if actual.AddressType != desired.AddressType { + err = sliceCli.Delete(ctx, desired.GetName(), meta.DeleteOptions{ + PropagationPolicy: ptr.To(meta.DeletePropagationBackground), + }) + if err != nil && !kerrors.IsNotFound(err) { + return fmt.Errorf("delete routing endpoint slice %q: %w", desired.GetName(), err) + } + } + case kerrors.IsNotFound(err): + // The slice does not exist yet, it is created below. + default: + return fmt.Errorf("get routing endpoint slice %q: %w", desired.GetName(), err) + } + + kubemeta.ControlOn(desired, svc, core.SchemeGroupVersion.WithKind("Service")) + + _, err = kubeclientset.Create(ctx, sliceCli, desired, + kubeclientset.WithUpdateIfExisted(routingEndpointSliceAlign(desired, svc))) + if err != nil { + return fmt.Errorf("install routing endpoint slice %q: %w", desired.GetName(), err) + } + + return nil +} + +// routingEndpointSliceAlign returns the align function of the endpoint slice of the +// system routing. +// +// The endpoints and the ports are replaced, not appended, so that a system that moved +// to another address or another port does not keep the old one. +func routingEndpointSliceAlign(desired *discovery.EndpointSlice, svc *core.Service) kubeclientset.AlignWithFn[*discovery.EndpointSlice] { + return func(actual *discovery.EndpointSlice) (*discovery.EndpointSlice, bool, error) { + if actual.AddressType != desired.AddressType { + return nil, false, fmt.Errorf("routing endpoint slice %q has address type %q, want %q", + actual.GetName(), actual.AddressType, desired.AddressType) + } + + aligned := actual.DeepCopy() + aligned.SetLabels(mergeLabels(actual.GetLabels(), desired.GetLabels())) + kubemeta.ControlOn(aligned, svc, core.SchemeGroupVersion.WithKind("Service")) + aligned.Ports = desired.Ports + aligned.Endpoints = desired.Endpoints + + if apiequality.Semantic.DeepEqual(actual, aligned) { + return actual, true, nil + } + return aligned, false, nil + } +} + +// cleanupObsoleteRoutingSlices removes the endpoint slices this installation owns and no +// longer needs: the slice an earlier version of this operator created under another name, +// and the slice the endpoint slice mirroring controller mirrored from the legacy +// endpoints. +// +// Only the slices whose owner reference names the service or the legacy endpoints of +// this installation are removed. A slice that carries the name of the routing service +// but belongs to somebody else is left alone, because a label is not a proof of +// ownership. +func cleanupObsoleteRoutingSlices(ctx context.Context, cli kubernetes.Interface, nsName string, + svc *core.Service, ep *core.Endpoints, keepName string, +) error { + if svc == nil || ep == nil { + // Without the owners there is no proof of ownership, so nothing is removed. + return nil + } + + sliceCli := cli.DiscoveryV1().EndpointSlices(nsName) + slices, err := sliceCli.List(ctx, meta.ListOptions{}) + if err != nil { + return fmt.Errorf("list routing endpoint slices: %w", err) + } + + for i := range slices.Items { + slice := &slices.Items[i] + if slice.GetName() == keepName { + continue + } + + switch { + case kubemeta.IsControlledBy(slice, ep): + // A mirror of the legacy endpoints of this installation. It carries + // the same address as the direct slice, but the aggregator may pick + // either of the two, so the direct slice replaces it. + case slice.GetLabels()[discovery.LabelManagedBy] == routingManagedBy && + kubemeta.IsControlledBy(slice, svc): + // A direct slice of this installation under a name this version no + // longer uses. + default: + // Not owned by this installation, keep it. + continue + } + + err = sliceCli.Delete(ctx, slice.GetName(), meta.DeleteOptions{ + PropagationPolicy: ptr.To(meta.DeletePropagationBackground), + }) + if err != nil && !kerrors.IsNotFound(err) { + return fmt.Errorf("delete obsolete routing endpoint slice %q: %w", slice.GetName(), err) + } + } + + return nil +} diff --git a/pkg/systemkuberess/service.go b/pkg/systemkuberess/service.go index b1a727ca0..2969e3140 100644 --- a/pkg/systemkuberess/service.go +++ b/pkg/systemkuberess/service.go @@ -5,19 +5,82 @@ import ( "fmt" "slices" + authz "k8s.io/api/authorization/v1" core "k8s.io/api/core/v1" + discovery "k8s.io/api/discovery/v1" + apiequality "k8s.io/apimachinery/pkg/api/equality" meta "k8s.io/apimachinery/pkg/apis/meta/v1" "gpustack.ai/gpustack/pkg/kubeclients/kubernetes" "gpustack.ai/gpustack/pkg/kubeclientset" "gpustack.ai/gpustack/pkg/kubeclientset/review" + "gpustack.ai/gpustack/pkg/kubereviewself" "gpustack.ai/gpustack/pkg/system" ) +const ( + // routingLabel marks the resources that carry the system routing service, so that + // they can be told apart from the resources of the cluster itself. + routingLabel = "gpustack.ai/fake-routing" + + // routingLabelValue is the value of the routingLabel. + routingLabelValue = "true" + + // routingPortName is the name of the routing port. The service, the legacy + // endpoints and the endpoint slice all name it the same, because the aggregator + // matches the port of a slice against the name of the service port it serves. + routingPortName = "https" +) + // InstallFakeSystemRoutingService creates the fake routing service/endpoint for system. // // The service points to the SelfIP of the system. func InstallFakeSystemRoutingService(ctx context.Context, cli kubernetes.Interface, nsName, svcName string, port int32) error { + return installFakeSystemRoutingService(ctx, cli, nsName, svcName, system.PrimaryIP.Get(), port) +} + +// installFakeSystemRoutingService creates the fake routing service/endpoint for system. +// +// The address is the address of the system the service routes to. It is a parameter, and +// not read from the system package, so that the desired resources can be built and +// checked without the process wide state they normally come from. +func installFakeSystemRoutingService(ctx context.Context, cli kubernetes.Interface, nsName, svcName, address string, port int32) error { + err := checkRoutingAccess(ctx, cli, nsName) + if err != nil { + return err + } + + // The address decides the family of the endpoint slice, so it is checked before + // anything is written. + desiredSlice, err := newRoutingEndpointSlice(nsName, svcName, address, port) + if err != nil { + return err + } + + svc, err := installRoutingService(ctx, cli, nsName, svcName, port) + if err != nil { + return err + } + + ep, err := installRoutingEndpoints(ctx, cli, nsName, svcName, address, port) + if err != nil { + return err + } + + err = installRoutingEndpointSlice(ctx, cli, nsName, desiredSlice, svc) + if err != nil { + return err + } + + return cleanupObsoleteRoutingSlices(ctx, cli, nsName, svc, ep, desiredSlice.GetName()) +} + +// checkRoutingAccess reviews the operations the routing reconcile performs. +// +// The service and the legacy endpoints keep the operations they have always needed. The +// endpoint slice is reviewed on its own, because the operations of the other two do not +// cover the list the cleanup of the obsolete slices reads. +func checkRoutingAccess(ctx context.Context, cli kubernetes.Interface, nsName string) error { err := review.CanDoCreate(ctx, cli.AuthorizationV1().SelfSubjectAccessReviews(), review.Simples{ @@ -39,98 +102,144 @@ func InstallFakeSystemRoutingService(ctx context.Context, cli kubernetes.Interfa return err } - svcCli := cli.CoreV1().Services(nsName) - eSvc := &core.Service{ + verbs := []string{"get", "list", "create", "update", "delete"} + reviews := make(kubereviewself.Reviews, 0, len(verbs)) + for _, verb := range verbs { + reviews = append(reviews, kubereviewself.Review{ + ResourceAttributes: &authz.ResourceAttributes{ + Verb: verb, + Group: discovery.SchemeGroupVersion.Group, + Version: discovery.SchemeGroupVersion.Version, + Resource: "endpointslices", + Namespace: nsName, + }, + }) + } + + return kubereviewself.CanDo(ctx, cli.AuthorizationV1().SelfSubjectAccessReviews(), reviews) +} + +// installRoutingService creates the selectorless service the extension api services are +// fronted by, and returns it as the cluster holds it. +func installRoutingService(ctx context.Context, cli kubernetes.Interface, nsName, svcName string, port int32) (*core.Service, error) { + expected := newRoutingService(nsName, svcName, port) + svcCompareFn := func(actual *core.Service) bool { + return actual.Spec.Type == expected.Spec.Type && + actual.Spec.ClusterIP == expected.Spec.ClusterIP && + slices.ContainsFunc(actual.Spec.Ports, func(ap core.ServicePort) bool { + return ap.Port == expected.Spec.Ports[0].Port + }) + } + + actual, err := kubeclientset.Create(ctx, cli.CoreV1().Services(nsName), expected, + kubeclientset.WithRecreateIfDuplicated(svcCompareFn)) + if err != nil { + return nil, fmt.Errorf("install fake routing service %q: %w", expected.GetName(), err) + } + + return actual, nil +} + +// newRoutingService builds the desired service of the system routing. +// +// The service has no selector, so no controller picks its endpoints for it, and the +// endpoint slice below is the only thing that routes to the system. +func newRoutingService(nsName, svcName string, port int32) *core.Service { + return &core.Service{ ObjectMeta: meta.ObjectMeta{ - Name: svcName, + Namespace: nsName, + Name: svcName, Labels: map[string]string{ - "gpustack.ai/fake-routing": "true", + routingLabel: routingLabelValue, }, }, Spec: core.ServiceSpec{ Type: core.ServiceTypeClusterIP, Ports: []core.ServicePort{ { - Name: "https", + Name: routingPortName, Port: port, }, }, }, } - svcCompareFn := func(aSvc *core.Service) bool { - return aSvc.Spec.Type == eSvc.Spec.Type && - aSvc.Spec.ClusterIP == eSvc.Spec.ClusterIP && - slices.ContainsFunc(aSvc.Spec.Ports, func(ap core.ServicePort) bool { - return ap.Port == eSvc.Spec.Ports[0].Port - }) +} + +// installRoutingEndpoints creates the legacy endpoints of the system routing, and returns +// them as the cluster holds them. +// +// The legacy endpoints are kept for the aggregators that still read them, which are the +// ones older than the Kubernetes release that moved the aggregator over to endpoint +// slices. The skip mirror label keeps the mirroring controller from making a second +// slice out of them, because the aggregator may then pick either of the two. +func installRoutingEndpoints(ctx context.Context, cli kubernetes.Interface, nsName, svcName, address string, port int32) (*core.Endpoints, error) { + expected := newRoutingEndpoints(nsName, svcName, address, port) + epAlignFn := func(actual *core.Endpoints) (*core.Endpoints, bool, error) { + aligned := actual.DeepCopy() + aligned.SetLabels(mergeLabels(actual.GetLabels(), expected.GetLabels())) + // The address and the port of the legacy endpoints are replaced, not + // appended, so that a system that moved to another address does not keep + // the old one. + aligned.Subsets = expected.Subsets + + if apiequality.Semantic.DeepEqual(actual.GetLabels(), aligned.GetLabels()) && + apiequality.Semantic.DeepEqual(actual.Subsets, aligned.Subsets) { + return actual, true, nil + } + return aligned, false, nil } - eSvc, err = kubeclientset.Create(ctx, svcCli, eSvc, - kubeclientset.WithRecreateIfDuplicated(svcCompareFn)) + actual, err := kubeclientset.Create(ctx, cli.CoreV1().Endpoints(nsName), expected, + kubeclientset.WithUpdateIfExisted(epAlignFn)) if err != nil { - return fmt.Errorf("install fake rounting service %q: %w", eSvc.GetName(), err) + return nil, fmt.Errorf("install fake routing endpoints %q: %w", expected.GetName(), err) } - epCli := cli.CoreV1().Endpoints(nsName) - eEp := &core.Endpoints{ + return actual, nil +} + +// newRoutingEndpoints builds the desired legacy endpoints of the system routing. +func newRoutingEndpoints(nsName, svcName, address string, port int32) *core.Endpoints { + return &core.Endpoints{ ObjectMeta: meta.ObjectMeta{ - Name: eSvc.GetName(), + Namespace: nsName, + Name: svcName, Labels: map[string]string{ - "gpustack.ai/fake-routing": "true", + routingLabel: routingLabelValue, + discovery.LabelSkipMirror: "true", }, }, Subsets: []core.EndpointSubset{ { Addresses: []core.EndpointAddress{ { - IP: system.PrimaryIP.Get(), + IP: address, }, }, Ports: []core.EndpointPort{ { - Name: "https", + Name: routingPortName, Port: port, }, }, }, }, } - epAlignFn := func(aEp *core.Endpoints) (*core.Endpoints, bool, error) { - var found bool - for i := range aEp.Subsets { - for j := range aEp.Subsets[i].Addresses { - if aEp.Subsets[i].Addresses[j].IP == eEp.Subsets[0].Addresses[0].IP { - found = true - break - } - } - if found { - found = false - for j := range aEp.Subsets[i].Ports { - if aEp.Subsets[i].Ports[j].Port == eEp.Subsets[0].Ports[0].Port { - found = true - break - } - } - } - if found { - break - } - } - if found { - return nil, true, nil - } +} - // Append the existing subsets. - aEp.Subsets = append(aEp.Subsets, eEp.Subsets...) - return aEp, false, nil +// mergeLabels returns the labels of an object with the wanted ones applied on top, so +// that the labels another actor added are kept and the wanted ones always win. +func mergeLabels(actual, wanted map[string]string) map[string]string { + if len(actual) == 0 && len(wanted) == 0 { + return nil } - _, err = kubeclientset.Create(ctx, epCli, eEp, - kubeclientset.WithUpdateIfExisted(epAlignFn)) - if err != nil { - return fmt.Errorf("install fake routing service %q: %w", eEp.GetName(), err) + merged := make(map[string]string, len(actual)+len(wanted)) + for k, v := range actual { + merged[k] = v } - - return nil + for k, v := range wanted { + merged[k] = v + } + return merged } diff --git a/pkg/systemkuberess/service_test.go b/pkg/systemkuberess/service_test.go new file mode 100644 index 000000000..dd35c8ce6 --- /dev/null +++ b/pkg/systemkuberess/service_test.go @@ -0,0 +1,641 @@ +package systemkuberess + +import ( + "context" + "fmt" + "sync/atomic" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + authz "k8s.io/api/authorization/v1" + core "k8s.io/api/core/v1" + discovery "k8s.io/api/discovery/v1" + kerrors "k8s.io/apimachinery/pkg/api/errors" + meta "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + ktesting "k8s.io/client-go/testing" + "k8s.io/utils/ptr" + + kubefake "gpustack.ai/gpustack/pkg/kubeclients/kubernetes/fake" + "gpustack.ai/gpustack/pkg/kubemeta" +) + +const ( + testNamespace = "gpustack-system" + testService = "gpustack-operator-worker" + testPort = int32(8443) +) + +// wantRouting is the state the cluster must hold after the reconcile. +type wantRouting struct { + // service is true when the routing service must exist. + service bool + // endpoints is true when the legacy endpoints must exist. + endpoints bool + // slice is true when the endpoint slice must exist. + slice bool + // address is the address the routing must point at. + address string + // port is the routing port. + port int32 + // addressType is the address type the endpoint slice must carry. + addressType discovery.AddressType + // sliceNames are the names of the endpoint slices the cluster must hold. + sliceNames []string +} + +// routingInput is a routing the reconcile is asked for. +type routingInput struct { + address string + port int32 +} + +// Test_installFakeSystemRoutingService drives the routing reconcile over the states a +// cluster can be in, and asserts the state it leaves behind. +func Test_installFakeSystemRoutingService(t *testing.T) { + testCases := []struct { + name string + // objects are the resources the cluster holds before the reconcile. + objects []runtime.Object + // input is the routing the reconcile is asked for. + input routingInput + // second, when set, is a routing the reconcile is asked for after the + // first one, so that a change of address or port is covered. + second *routingInput + // deny denies the access reviews it answers for. + deny func(attrs *authz.ResourceAttributes) bool + // reactors are the reactors of the fake, applied after the built-in ones. + reactors func(cli *kubefake.Clientset) + // want is the state the cluster must hold afterwards. + want wantRouting + // wantErr is the error the reconcile must report, empty when it must + // succeed. + wantErr string + }{ + { + name: "installs the routing resources", + input: routingInput{address: "192.168.50.10", port: testPort}, + want: wantRouting{ + service: true, + endpoints: true, + slice: true, + address: "192.168.50.10", + port: testPort, + addressType: discovery.AddressTypeIPv4, + sliceNames: []string{"gpustack-operator-worker-direct"}, + }, + }, + { + name: "installs the routing resources of an IPv6 system", + input: routingInput{address: "fd00::1", port: testPort}, + objects: nil, + want: wantRouting{ + service: true, + endpoints: true, + slice: true, + address: "fd00::1", + port: testPort, + addressType: discovery.AddressTypeIPv6, + sliceNames: []string{"gpustack-operator-worker-direct"}, + }, + }, + { + name: "converges when installed again", + input: routingInput{address: "192.168.50.10", port: testPort}, + second: &routingInput{ + address: "192.168.50.10", + port: testPort, + }, + want: wantRouting{ + service: true, + endpoints: true, + slice: true, + address: "192.168.50.10", + port: testPort, + addressType: discovery.AddressTypeIPv4, + sliceNames: []string{"gpustack-operator-worker-direct"}, + }, + }, + { + name: "replaces a stale address", + input: routingInput{address: "192.168.50.20", port: testPort}, + second: &routingInput{address: "192.168.50.10", port: testPort}, + objects: []runtime.Object{staleSlice("192.168.50.99", testPort)}, + want: wantRouting{ + service: true, + endpoints: true, + slice: true, + address: "192.168.50.10", + port: testPort, + addressType: discovery.AddressTypeIPv4, + sliceNames: []string{"gpustack-operator-worker-direct"}, + }, + }, + { + name: "replaces a stale port", + input: routingInput{address: "192.168.50.10", port: 9443}, + second: &routingInput{address: "192.168.50.10", port: testPort}, + want: wantRouting{ + service: true, + endpoints: true, + slice: true, + address: "192.168.50.10", + port: testPort, + addressType: discovery.AddressTypeIPv4, + sliceNames: []string{"gpustack-operator-worker-direct"}, + }, + }, + { + name: "replaces the slice of the other address family", + input: routingInput{address: "fd00::1", port: testPort}, + second: &routingInput{address: "fd00::1", port: testPort}, + objects: []runtime.Object{ + staleSlice("192.168.50.10", testPort), + }, + want: wantRouting{ + service: true, + endpoints: true, + slice: true, + address: "fd00::1", + port: testPort, + addressType: discovery.AddressTypeIPv6, + sliceNames: []string{"gpustack-operator-worker-direct"}, + }, + }, + { + name: "removes the mirrored slice of the legacy endpoints", + input: routingInput{address: "192.168.50.10", port: testPort}, + objects: []runtime.Object{ + legacyEndpoints("192.168.50.10", testPort), + mirroredSlice("192.168.50.10", testPort), + }, + want: wantRouting{ + service: true, + endpoints: true, + slice: true, + address: "192.168.50.10", + port: testPort, + addressType: discovery.AddressTypeIPv4, + sliceNames: []string{"gpustack-operator-worker-direct"}, + }, + }, + { + name: "removes the slice of an earlier version of this operator", + input: routingInput{address: "192.168.50.10", port: testPort}, + objects: []runtime.Object{ + obsoleteSlice("192.168.50.10", testPort), + }, + want: wantRouting{ + service: true, + endpoints: true, + slice: true, + address: "192.168.50.10", + port: testPort, + addressType: discovery.AddressTypeIPv4, + sliceNames: []string{"gpustack-operator-worker-direct"}, + }, + }, + { + name: "keeps the slices of others", + input: routingInput{address: "192.168.50.10", port: testPort}, + objects: []runtime.Object{ + foreignSlice(), + }, + want: wantRouting{ + service: true, + endpoints: true, + slice: true, + address: "192.168.50.10", + port: testPort, + addressType: discovery.AddressTypeIPv4, + sliceNames: []string{ + "gpustack-operator-worker-direct", + "other-service-abcde", + }, + }, + }, + { + name: "keeps the endpoints of others", + input: routingInput{address: "192.168.50.10", port: testPort}, + objects: []runtime.Object{ + &core.Endpoints{ + ObjectMeta: meta.ObjectMeta{ + Namespace: testNamespace, + Name: "other-service", + }, + Subsets: []core.EndpointSubset{ + { + Addresses: []core.EndpointAddress{{IP: "10.0.0.1"}}, + Ports: []core.EndpointPort{{Name: "https", Port: 443}}, + }, + }, + }, + }, + want: wantRouting{ + service: true, + endpoints: true, + slice: true, + address: "192.168.50.10", + port: testPort, + addressType: discovery.AddressTypeIPv4, + sliceNames: []string{"gpustack-operator-worker-direct"}, + }, + }, + { + name: "reports a denied access review", + input: routingInput{address: "192.168.50.10", port: testPort}, + deny: func(attrs *authz.ResourceAttributes) bool { + return attrs.Resource == "endpointslices" && attrs.Verb == "list" + }, + want: wantRouting{ + sliceNames: nil, + }, + wantErr: "denied", + }, + { + name: "reports a failed endpoint slice creation", + input: routingInput{address: "192.168.50.10", port: testPort}, + reactors: func(cli *kubefake.Clientset) { + cli.PrependReactor("create", "endpointslices", func(action ktesting.Action) (bool, runtime.Object, error) { + return true, nil, kerrors.NewInternalError(fmt.Errorf("endpoint slice storage is broken")) + }) + }, + want: wantRouting{ + service: true, + endpoints: true, + slice: false, + address: "192.168.50.10", + port: testPort, + addressType: discovery.AddressTypeIPv4, + sliceNames: nil, + }, + wantErr: "endpoint slice storage is broken", + }, + { + name: "retries a lost write", + input: routingInput{address: "192.168.50.10", port: testPort}, + second: &routingInput{ + address: "192.168.50.11", + port: testPort, + }, + reactors: func(cli *kubefake.Clientset) { + var lost atomic.Bool + cli.PrependReactor("update", "endpointslices", func(action ktesting.Action) (bool, runtime.Object, error) { + if lost.CompareAndSwap(false, true) { + return true, nil, kerrors.NewConflict( + discovery.Resource("endpointslices"), testService, fmt.Errorf("the object was modified"), + ) + } + return false, nil, nil + }) + }, + want: wantRouting{ + service: true, + endpoints: true, + slice: true, + address: "192.168.50.11", + port: testPort, + addressType: discovery.AddressTypeIPv4, + sliceNames: []string{"gpustack-operator-worker-direct"}, + }, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + cli := newFakeClient(tc.deny, tc.objects...) + if tc.reactors != nil { + tc.reactors(cli) + } + + err := installFakeSystemRoutingService(t.Context(), cli, testNamespace, testService, + tc.input.address, tc.input.port) + if tc.wantErr != "" { + require.Error(t, err) + assert.Contains(t, err.Error(), tc.wantErr) + } else { + require.NoError(t, err) + } + + if tc.second != nil { + err = installFakeSystemRoutingService(t.Context(), cli, testNamespace, testService, + tc.second.address, tc.second.port) + if tc.wantErr != "" { + require.Error(t, err) + assert.Contains(t, err.Error(), tc.wantErr) + } else { + require.NoError(t, err) + } + } + + checkRouting(t, cli, tc.want) + }) + } +} + +// Test_installFakeSystemRoutingServiceCancelled checks that a canceled call is reported +// as it is, and that a context that is already done stops the reconcile from trying +// again. +// +// The fake client does not read the context itself, so the cancellation is carried by +// the error the call answers with, the way a client answers a canceled call. +func Test_installFakeSystemRoutingServiceCancelled(t *testing.T) { + t.Run("reports the cancellation of a call", func(t *testing.T) { + cli := newFakeClient(nil) + cli.PrependReactor("create", "endpointslices", func(action ktesting.Action) (bool, runtime.Object, error) { + return true, nil, context.Canceled + }) + + err := installFakeSystemRoutingService(t.Context(), cli, testNamespace, testService, + "192.168.50.10", testPort) + require.Error(t, err) + assert.ErrorIs(t, err, context.Canceled) + + checkRouting(t, cli, wantRouting{ + service: true, + endpoints: true, + slice: false, + address: "192.168.50.10", + port: testPort, + }) + }) + + t.Run("stops trying again when the context is done", func(t *testing.T) { + cli := newFakeClient(nil, staleSlice("192.168.50.99", testPort)) + + var attempts atomic.Int64 + cli.PrependReactor("update", "endpointslices", func(action ktesting.Action) (bool, runtime.Object, error) { + attempts.Add(1) + return true, nil, kerrors.NewConflict( + discovery.Resource("endpointslices"), testService, fmt.Errorf("the object was modified"), + ) + }) + + ctx, cancel := context.WithCancel(t.Context()) + cancel() + + err := installFakeSystemRoutingService(ctx, cli, testNamespace, testService, "192.168.50.10", testPort) + require.Error(t, err) + assert.Contains(t, err.Error(), "the object was modified") + assert.Equal(t, int64(1), attempts.Load(), "the write must not be tried again") + }) +} + +// checkRouting asserts the observable state of the routing resources. +func checkRouting(t *testing.T, cli *kubefake.Clientset, want wantRouting) { + t.Helper() + + svc, err := cli.CoreV1().Services(testNamespace).Get(t.Context(), testService, meta.GetOptions{}) + if want.service { + require.NoError(t, err) + assert.Equal(t, core.ServiceTypeClusterIP, svc.Spec.Type) + assert.Nil(t, svc.Spec.Selector, "the routing service must have no selector") + require.Len(t, svc.Spec.Ports, 1) + assert.Equal(t, routingPortName, svc.Spec.Ports[0].Name) + assert.Equal(t, want.port, svc.Spec.Ports[0].Port) + } else { + assert.True(t, kerrors.IsNotFound(err), "the routing service must not exist") + } + + ep, err := cli.CoreV1().Endpoints(testNamespace).Get(t.Context(), testService, meta.GetOptions{}) + if want.endpoints { + require.NoError(t, err) + assert.Equal(t, "true", ep.GetLabels()[discovery.LabelSkipMirror], + "the legacy endpoints must not be mirrored") + require.Len(t, ep.Subsets, 1) + assert.Equal(t, []core.EndpointAddress{{IP: want.address}}, ep.Subsets[0].Addresses) + require.Len(t, ep.Subsets[0].Ports, 1) + assert.Equal(t, routingPortName, ep.Subsets[0].Ports[0].Name) + assert.Equal(t, want.port, ep.Subsets[0].Ports[0].Port) + } else { + assert.True(t, kerrors.IsNotFound(err), "the legacy endpoints must not exist") + } + + slice, err := cli.DiscoveryV1().EndpointSlices(testNamespace). + Get(t.Context(), routingEndpointSliceName(testService), meta.GetOptions{}) + if want.slice { + require.NoError(t, err) + assert.Equal(t, testService, slice.GetLabels()[discovery.LabelServiceName]) + assert.Equal(t, routingManagedBy, slice.GetLabels()[discovery.LabelManagedBy]) + assert.Equal(t, want.addressType, slice.AddressType) + require.Len(t, slice.Endpoints, 1) + assert.Equal(t, []string{want.address}, slice.Endpoints[0].Addresses) + require.NotNil(t, slice.Endpoints[0].Conditions.Ready) + assert.True(t, *slice.Endpoints[0].Conditions.Ready) + require.Len(t, slice.Ports, 1) + require.NotNil(t, slice.Ports[0].Name) + assert.Equal(t, routingPortName, *slice.Ports[0].Name) + require.NotNil(t, slice.Ports[0].Port) + assert.Equal(t, want.port, *slice.Ports[0].Port) + + owner := kubemeta.GetControllerOfNoCopy(slice) + require.NotNil(t, owner, "the endpoint slice must be owned by the routing service") + if want.service { + assert.Equal(t, svc.GetUID(), owner.UID) + } + } else { + assert.True(t, kerrors.IsNotFound(err), "the endpoint slice must not exist") + } + + slices, err := cli.DiscoveryV1().EndpointSlices(testNamespace).List(t.Context(), meta.ListOptions{}) + require.NoError(t, err) + names := make([]string, 0, len(slices.Items)) + for i := range slices.Items { + names = append(names, slices.Items[i].GetName()) + } + assert.ElementsMatch(t, want.sliceNames, names) +} + +// newFakeClient returns a fake clientset holding the given objects. +// +// Every object it creates gets a uid, so that the owner references of the fixtures can +// name it, and every access review is allowed unless the deny function denies it. +func newFakeClient(deny func(attrs *authz.ResourceAttributes) bool, objects ...runtime.Object) *kubefake.Clientset { + for _, obj := range objects { + if m, ok := obj.(meta.Object); ok && m.GetUID() == "" { + m.SetUID(types.UID(fakeKind(m) + "/" + m.GetName())) + } + } + + cli := kubefake.NewSimpleClientset(objects...) + + cli.PrependReactor("create", "*", func(action ktesting.Action) (bool, runtime.Object, error) { + create, ok := action.(ktesting.CreateAction) + if !ok { + return false, nil, nil + } + obj, ok := create.GetObject().(meta.Object) + if !ok || obj.GetUID() != "" { + return false, nil, nil + } + obj.SetUID(types.UID(fakeKind(obj) + "/" + obj.GetName())) + return false, nil, nil + }) + + cli.PrependReactor("create", "selfsubjectaccessreviews", func(action ktesting.Action) (bool, runtime.Object, error) { + create, ok := action.(ktesting.CreateAction) + if !ok { + return false, nil, nil + } + sar, ok := create.GetObject().(*authz.SelfSubjectAccessReview) + if !ok { + return false, nil, nil + } + allowed := true + if deny != nil && sar.Spec.ResourceAttributes != nil { + allowed = !deny(sar.Spec.ResourceAttributes) + } + sar.Status = authz.SubjectAccessReviewStatus{Allowed: allowed} + // The access reviews are answered here, and not stored, the way the api + // server answers them. + return true, sar, nil + }) + + return cli +} + +// fakeKind returns the kind of an object, for the uids of the fake. +func fakeKind(obj meta.Object) string { + switch obj.(type) { + case *core.Service: + return "Service" + case *core.Endpoints: + return "Endpoints" + case *discovery.EndpointSlice: + return "EndpointSlice" + default: + return "Object" + } +} + +// legacyEndpoints builds the legacy endpoints of the routing service, as an earlier +// version of this operator left them. +func legacyEndpoints(address string, port int32) *core.Endpoints { + return &core.Endpoints{ + ObjectMeta: meta.ObjectMeta{ + Namespace: testNamespace, + Name: testService, + Labels: map[string]string{ + routingLabel: routingLabelValue, + }, + }, + Subsets: []core.EndpointSubset{ + { + Addresses: []core.EndpointAddress{{IP: address}}, + Ports: []core.EndpointPort{{Name: routingPortName, Port: port}}, + }, + }, + } +} + +// staleSlice builds the endpoint slice of the routing service, as an earlier address +// left it. +func staleSlice(address string, port int32) *discovery.EndpointSlice { + return &discovery.EndpointSlice{ + ObjectMeta: meta.ObjectMeta{ + Namespace: testNamespace, + Name: routingEndpointSliceName(testService), + Labels: map[string]string{ + discovery.LabelServiceName: testService, + discovery.LabelManagedBy: routingManagedBy, + routingLabel: routingLabelValue, + }, + OwnerReferences: []meta.OwnerReference{ + { + APIVersion: "v1", + Kind: "Service", + Name: testService, + UID: types.UID("Service/" + testService), + Controller: ptr.To(true), + }, + }, + }, + AddressType: discovery.AddressTypeIPv4, + Endpoints: []discovery.Endpoint{ + { + Addresses: []string{address}, + Conditions: discovery.EndpointConditions{Ready: ptr.To(true)}, + }, + }, + Ports: []discovery.EndpointPort{ + {Name: ptr.To(routingPortName), Port: ptr.To(port), Protocol: ptr.To(core.ProtocolTCP)}, + }, + } +} + +// mirroredSlice builds the endpoint slice the mirroring controller mirrors from the +// legacy endpoints of the routing service. +func mirroredSlice(address string, port int32) *discovery.EndpointSlice { + return &discovery.EndpointSlice{ + ObjectMeta: meta.ObjectMeta{ + Namespace: testNamespace, + Name: testService + "-abcde", + Labels: map[string]string{ + discovery.LabelServiceName: testService, + discovery.LabelManagedBy: "endpointslicemirroring-controller.k8s.io", + }, + OwnerReferences: []meta.OwnerReference{ + { + APIVersion: "v1", + Kind: "Endpoints", + Name: testService, + UID: types.UID("Endpoints/" + testService), + Controller: ptr.To(true), + }, + }, + }, + AddressType: discovery.AddressTypeIPv4, + Endpoints: []discovery.Endpoint{ + { + Addresses: []string{address}, + Conditions: discovery.EndpointConditions{Ready: ptr.To(true)}, + }, + }, + Ports: []discovery.EndpointPort{ + {Name: ptr.To(routingPortName), Port: ptr.To(port), Protocol: ptr.To(core.ProtocolTCP)}, + }, + } +} + +// obsoleteSlice builds the endpoint slice an earlier version of this operator created +// under a name it no longer uses. +func obsoleteSlice(address string, port int32) *discovery.EndpointSlice { + slice := staleSlice(address, port) + slice.SetName(testService + "-legacy") + return slice +} + +// foreignSlice builds the endpoint slice of another service. +func foreignSlice() *discovery.EndpointSlice { + return &discovery.EndpointSlice{ + ObjectMeta: meta.ObjectMeta{ + Namespace: testNamespace, + Name: "other-service-abcde", + Labels: map[string]string{ + discovery.LabelServiceName: "other-service", + discovery.LabelManagedBy: "endpointslicemirroring-controller.k8s.io", + }, + OwnerReferences: []meta.OwnerReference{ + { + APIVersion: "v1", + Kind: "Endpoints", + Name: "other-service", + UID: types.UID("Endpoints/other-service"), + Controller: ptr.To(true), + }, + }, + }, + AddressType: discovery.AddressTypeIPv4, + Endpoints: []discovery.Endpoint{ + { + Addresses: []string{"10.0.0.1"}, + Conditions: discovery.EndpointConditions{Ready: ptr.To(true)}, + }, + }, + Ports: []discovery.EndpointPort{ + {Name: ptr.To(routingPortName), Port: ptr.To(int32(443)), Protocol: ptr.To(core.ProtocolTCP)}, + }, + } +} From c59136d732c867954f73154fecdce252f1a0df16 Mon Sep 17 00:00:00 2001 From: thxCode Date: Mon, 26 Oct 2026 12:20:01 +0800 Subject: [PATCH 2/5] fix(worker): harden local routing EndpointSlice ownership and retry guards Signed-off-by: thxCode --- .../templates/worker/serviceaccount.yaml | 7 +- docs/contribute/internals.md | 9 + pkg/systemkuberess/endpoint_slice.go | 273 +++++++++-- pkg/systemkuberess/endpoint_slice_test.go | 444 ++++++++++++++++++ pkg/systemkuberess/service.go | 48 +- pkg/systemkuberess/service_test.go | 195 +++++++- 6 files changed, 933 insertions(+), 43 deletions(-) create mode 100644 pkg/systemkuberess/endpoint_slice_test.go diff --git a/deploy/gpustack-operator/chart/templates/worker/serviceaccount.yaml b/deploy/gpustack-operator/chart/templates/worker/serviceaccount.yaml index ff935f243..614133a25 100644 --- a/deploy/gpustack-operator/chart/templates/worker/serviceaccount.yaml +++ b/deploy/gpustack-operator/chart/templates/worker/serviceaccount.yaml @@ -16,8 +16,11 @@ metadata: resources it manages: the service and the legacy endpoints of the system routing, and the endpoint slices of the discovery.k8s.io group it writes when it runs outside a nearby cluster. The grant those operations need is get, list, create, - update and delete on endpointslices, plus the same set on services and endpoints. - A narrower binding has to keep all of them, or the worker stops routing. + update and delete on endpointslices, and get, create, update and delete on + services and endpoints. The endpoint slice carries the routing service as its + owner and blocks its deletion, so the grant needs update on the finalizers + subresource of services as well. A narrower binding has to keep all of them, or + the worker stops routing. */}} apiVersion: rbac.authorization.k8s.io/v1 kind: ClusterRoleBinding diff --git a/docs/contribute/internals.md b/docs/contribute/internals.md index f755b8c21..956c7cb54 100644 --- a/docs/contribute/internals.md +++ b/docs/contribute/internals.md @@ -127,6 +127,15 @@ slices it owns and no longer needs: the mirror of the legacy `Endpoints`, and a version left under another name. It removes a slice only when the owner reference names the routing `Service` or the legacy `Endpoints` of this worker, so a slice of another service stays. +The owner is recognized by the name it carries, and not by its uid, because the routing `Service` and +the legacy `Endpoints` are recreated whenever their spec changes. A slice the resource of the past +owned still belongs to the worker, so the worker adopts it and rewrites its owner reference; refusing +it would leave the worker unable to repair the routing it owns. + +One worker owns the routing of a cluster at a time. The namespace and the routing service name are the +same in every installation, so two workers pointing at one cluster write the same slice, and the last +write wins. A slice the worker does not own is refused rather than taken over. + ## Worker gateway `pkg/workergateway/service` folds many clusters' `InstanceType`s into one fleet-wide diff --git a/pkg/systemkuberess/endpoint_slice.go b/pkg/systemkuberess/endpoint_slice.go index 3a96a7616..b992b9c49 100644 --- a/pkg/systemkuberess/endpoint_slice.go +++ b/pkg/systemkuberess/endpoint_slice.go @@ -2,6 +2,7 @@ package systemkuberess import ( "context" + "errors" "fmt" "net" @@ -35,6 +36,22 @@ const ( routingEndpointSliceNameSuffix = "-direct" ) +// addressTypeMismatchError reports that the endpoint slice in the cluster carries another +// address type than the routing wants. +// +// The address type of an endpoint slice cannot be changed, so such a slice is replaced +// rather than updated. +type addressTypeMismatchError struct { + name string + actual discovery.AddressType + expected discovery.AddressType +} + +func (e *addressTypeMismatchError) Error() string { + return fmt.Sprintf("routing endpoint slice %q has address type %q, want %q", + e.name, e.actual, e.expected) +} + // newRoutingEndpointSlice builds the desired endpoint slice of the system routing. // // The aggregator reads the slices that carry the name of the service, and takes the @@ -107,6 +124,11 @@ func routingAddressType(address string) (discovery.AddressType, error) { // // The address type of an endpoint slice cannot be changed, so a slice that was made for // the other address family is replaced instead of updated. +// +// One worker owns the routing of a cluster at a time. The namespace and the routing +// service name are the same in every installation, so two workers pointing at one +// cluster write the same slice, and the last write wins. A slice this worker does not +// own is refused rather than taken over. func installRoutingEndpointSlice(ctx context.Context, cli kubernetes.Interface, nsName string, desired *discovery.EndpointSlice, svc *core.Service, ) error { @@ -115,17 +137,17 @@ func installRoutingEndpointSlice(ctx context.Context, cli kubernetes.Interface, actual, err := sliceCli.Get(ctx, desired.GetName(), meta.GetOptions{}) switch { case err == nil: - if actual.GetLabels()[discovery.LabelManagedBy] != routingManagedBy { - return fmt.Errorf("routing endpoint slice %q is managed by %q, not by this operator", - desired.GetName(), actual.GetLabels()[discovery.LabelManagedBy]) + err = checkRoutingSliceIdentity(actual, svc) + if err != nil { + return err } if actual.AddressType != desired.AddressType { - err = sliceCli.Delete(ctx, desired.GetName(), meta.DeleteOptions{ - PropagationPolicy: ptr.To(meta.DeletePropagationBackground), - }) - if err != nil && !kerrors.IsNotFound(err) { - return fmt.Errorf("delete routing endpoint slice %q: %w", desired.GetName(), err) + err = deleteRoutingSlice(ctx, cli, nsName, actual) + if err != nil && !kerrors.IsConflict(err) { + return err } + // A conflict means another actor replaced the slice in between. The + // write below reads the slice that is there now and decides on it. } case kerrors.IsNotFound(err): // The slice does not exist yet, it is created below. @@ -135,32 +157,160 @@ func installRoutingEndpointSlice(ctx context.Context, cli kubernetes.Interface, kubemeta.ControlOn(desired, svc, core.SchemeGroupVersion.WithKind("Service")) - _, err = kubeclientset.Create(ctx, sliceCli, desired, - kubeclientset.WithUpdateIfExisted(routingEndpointSliceAlign(desired, svc))) - if err != nil { - return fmt.Errorf("install routing endpoint slice %q: %w", desired.GetName(), err) + // The read the create uses can still show the slice of the other address family + // after it was deleted, so a family mismatch is retried with the slice gone. + for attempt := 0; ; attempt++ { + _, err = kubeclientset.Create(ctx, sliceCli, desired, + kubeclientset.WithUpdateIfExisted(routingEndpointSliceAlign(desired, svc))) + if err == nil { + return nil + } + + var mismatch *addressTypeMismatchError + if !errors.As(err, &mismatch) || attempt >= 2 { + return fmt.Errorf("install routing endpoint slice %q: %w", desired.GetName(), err) + } + + stale, getErr := sliceCli.Get(ctx, desired.GetName(), meta.GetOptions{}) + if getErr != nil { + if kerrors.IsNotFound(getErr) { + continue + } + return fmt.Errorf("get routing endpoint slice %q: %w", desired.GetName(), getErr) + } + + // The slice may have been replaced between the read that found the + // mismatch and this one, so its identity is checked again before + // anything is deleted. The uid precondition of the deletion binds it to + // the slice that was read here, and that slice may be one another actor + // owns. + err = checkRoutingSliceIdentity(stale, svc) + if err != nil { + return err + } + + if stale.AddressType == desired.AddressType { + // The slice carries the wanted address family now, so the + // reconcile converges on it without deleting it. + continue + } + + err = deleteRoutingSlice(ctx, cli, nsName, stale) + if err != nil { + if !kerrors.IsConflict(err) { + return err + } + // Another actor replaced the slice in between. The write below reads + // the slice that is there now and decides on it. + continue + } + } +} + +// checkRoutingSliceIdentity refuses an endpoint slice this worker does not own. +// +// A slice belongs to this worker when it carries the managed-by label of this operator +// and the routing service controls it, or when it carries the label and has no owner at +// all, which is the state a slice of an earlier version of this operator leaves behind. +// Any other slice is refused, and the refusal names its owner. +func checkRoutingSliceIdentity(slice *discovery.EndpointSlice, svc *core.Service) error { + if slice.GetLabels()[discovery.LabelManagedBy] != routingManagedBy { + return fmt.Errorf("routing endpoint slice %q is managed by %q, not by this operator", + slice.GetName(), slice.GetLabels()[discovery.LabelManagedBy]) + } + + owner := kubemeta.GetControllerOfNoCopy(slice) + switch { + case owner == nil: + // A slice of an earlier version of this operator, which left it without an owner. + case routingServiceOwns(slice, svc): + // The routing service controls the slice. The uid of the owner may be that + // of the service this installation recreated, which the slice outlived; the + // slice is adopted, and its owner reference is rewritten with the service + // of now. + default: + return fmt.Errorf("routing endpoint slice %q is controlled by %s %q with uid %q, not by the routing service", + slice.GetName(), owner.Kind, owner.Name, owner.UID) } return nil } +// routingServiceOwns reports whether the routing service controls the slice. +// +// The owner is recognized by the name of the service it names, and not by its uid, +// because the routing service is recreated whenever its spec changes. The slice that +// the service of the past owned still belongs to this installation, and one service +// carries the name in the namespace, so the name proves the owner as well as the uid +// does, and it survives the recreation of the service. +func routingServiceOwns(slice *discovery.EndpointSlice, svc *core.Service) bool { + if svc == nil { + return false + } + + return controlsRoutingOwner(slice, "Service", svc.GetName()) +} + +// routingEndpointsOwns reports whether the legacy endpoints control the slice. +// +// The owner is recognized by the name here for the same reason it is for the routing +// service: the legacy endpoints are recreated whenever they are deleted, and the +// mirror of the endpoints of the past still belongs to this installation. +func routingEndpointsOwns(slice *discovery.EndpointSlice, ep *core.Endpoints) bool { + if ep == nil { + return false + } + + return controlsRoutingOwner(slice, "Endpoints", ep.GetName()) +} + +// controlsRoutingOwner reports whether the controller owner reference of the slice names +// the resource of the given kind and name. +func controlsRoutingOwner(slice *discovery.EndpointSlice, kind, name string) bool { + owner := kubemeta.GetControllerOfNoCopy(slice) + if owner == nil { + return false + } + + return owner.APIVersion == core.SchemeGroupVersion.String() && + owner.Kind == kind && + owner.Name == name +} + // routingEndpointSliceAlign returns the align function of the endpoint slice of the // system routing. // // The endpoints and the ports are replaced, not appended, so that a system that moved // to another address or another port does not keep the old one. +// +// The identity of the slice is checked on the object that is written, and not only on +// the one that was read before, so that a slice that was swapped in between the two is +// not rewritten. func routingEndpointSliceAlign(desired *discovery.EndpointSlice, svc *core.Service) kubeclientset.AlignWithFn[*discovery.EndpointSlice] { return func(actual *discovery.EndpointSlice) (*discovery.EndpointSlice, bool, error) { + err := checkRoutingSliceIdentity(actual, svc) + if err != nil { + return nil, false, err + } + if actual.AddressType != desired.AddressType { - return nil, false, fmt.Errorf("routing endpoint slice %q has address type %q, want %q", - actual.GetName(), actual.AddressType, desired.AddressType) + return nil, false, &addressTypeMismatchError{ + name: actual.GetName(), + actual: actual.AddressType, + expected: desired.AddressType, + } } aligned := actual.DeepCopy() aligned.SetLabels(mergeLabels(actual.GetLabels(), desired.GetLabels())) kubemeta.ControlOn(aligned, svc, core.SchemeGroupVersion.WithKind("Service")) - aligned.Ports = desired.Ports - aligned.Endpoints = desired.Endpoints + + // The endpoints and the ports are copied, and not shared with the desired + // object, so that a change to the object that is written cannot reach the + // object the next attempt writes. + wanted := desired.DeepCopy() + aligned.Ports = wanted.Ports + aligned.Endpoints = wanted.Endpoints if apiequality.Semantic.DeepEqual(actual, aligned) { return actual, true, nil @@ -169,6 +319,26 @@ func routingEndpointSliceAlign(desired *discovery.EndpointSlice, svc *core.Servi } } +// deleteRoutingSlice deletes an endpoint slice of the system routing. +// +// The deletion is bound to the uid of the slice that was read, so a slice that another +// actor replaced in between is not destroyed. The api server answers a conflict in +// that case, and the slice that is there then is read again by the caller. +func deleteRoutingSlice(ctx context.Context, cli kubernetes.Interface, nsName string, slice *discovery.EndpointSlice) error { + err := cli.DiscoveryV1().EndpointSlices(nsName).Delete(ctx, slice.GetName(), meta.DeleteOptions{ + PropagationPolicy: ptr.To(meta.DeletePropagationBackground), + Preconditions: &meta.Preconditions{UID: ptr.To(slice.GetUID())}, + }) + if err != nil { + if kerrors.IsNotFound(err) { + // The slice is gone, which is the state the delete aimed at. + return nil + } + return fmt.Errorf("delete routing endpoint slice %q: %w", slice.GetName(), err) + } + return nil +} + // cleanupObsoleteRoutingSlices removes the endpoint slices this installation owns and no // longer needs: the slice an earlier version of this operator created under another name, // and the slice the endpoint slice mirroring controller mirrored from the legacy @@ -198,27 +368,68 @@ func cleanupObsoleteRoutingSlices(ctx context.Context, cli kubernetes.Interface, continue } - switch { - case kubemeta.IsControlledBy(slice, ep): - // A mirror of the legacy endpoints of this installation. It carries - // the same address as the direct slice, but the aggregator may pick - // either of the two, so the direct slice replaces it. - case slice.GetLabels()[discovery.LabelManagedBy] == routingManagedBy && - kubemeta.IsControlledBy(slice, svc): - // A direct slice of this installation under a name this version no - // longer uses. - default: + if !isObsoleteRoutingSlice(slice, svc, ep) { // Not owned by this installation, keep it. continue } - err = sliceCli.Delete(ctx, slice.GetName(), meta.DeleteOptions{ - PropagationPolicy: ptr.To(meta.DeletePropagationBackground), - }) - if err != nil && !kerrors.IsNotFound(err) { - return fmt.Errorf("delete obsolete routing endpoint slice %q: %w", slice.GetName(), err) + err = deleteObsoleteRoutingSlice(ctx, cli, nsName, slice, svc, ep, keepName) + if err != nil { + return err } } return nil } + +// isObsoleteRoutingSlice reports whether the slice belongs to this installation and is no +// longer needed. +func isObsoleteRoutingSlice(slice *discovery.EndpointSlice, svc *core.Service, ep *core.Endpoints) bool { + switch { + case routingEndpointsOwns(slice, ep): + // A mirror of the legacy endpoints of this installation. It carries the + // same address as the direct slice, but the aggregator may pick either + // of the two, so the direct slice replaces it. + case slice.GetLabels()[discovery.LabelManagedBy] == routingManagedBy && + routingServiceOwns(slice, svc): + // A direct slice of this installation under a name this version no + // longer uses. + default: + return false + } + + return true +} + +// deleteObsoleteRoutingSlice removes the obsolete slice, and reads the slice that is +// there when another actor replaced it in between. +// +// The delete carries the uid of the slice that was read, so the slice that another +// actor put in its place is not destroyed. The slice that is there then is read again, +// and it is removed when it is ours and obsolete as well. +func deleteObsoleteRoutingSlice(ctx context.Context, cli kubernetes.Interface, nsName string, + slice *discovery.EndpointSlice, svc *core.Service, ep *core.Endpoints, keepName string, +) error { + err := deleteRoutingSlice(ctx, cli, nsName, slice) + if err == nil || !kerrors.IsConflict(err) { + return err + } + + current, err := cli.DiscoveryV1().EndpointSlices(nsName). + Get(ctx, slice.GetName(), meta.GetOptions{}) + if err != nil { + if kerrors.IsNotFound(err) { + // The slice is gone, which is the state the delete aimed at. + return nil + } + return fmt.Errorf("get routing endpoint slice %q: %w", slice.GetName(), err) + } + + if current.GetName() == keepName || !isObsoleteRoutingSlice(current, svc, ep) { + // The slice that is there now is kept, and it is not ours or it is the + // one this installation uses. + return nil + } + + return deleteRoutingSlice(ctx, cli, nsName, current) +} diff --git a/pkg/systemkuberess/endpoint_slice_test.go b/pkg/systemkuberess/endpoint_slice_test.go new file mode 100644 index 000000000..b74902042 --- /dev/null +++ b/pkg/systemkuberess/endpoint_slice_test.go @@ -0,0 +1,444 @@ +package systemkuberess + +import ( + "fmt" + "sync/atomic" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + core "k8s.io/api/core/v1" + discovery "k8s.io/api/discovery/v1" + kerrors "k8s.io/apimachinery/pkg/api/errors" + meta "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/types" + ktesting "k8s.io/client-go/testing" + "k8s.io/utils/ptr" + + "gpustack.ai/gpustack/pkg/kubemeta" +) + +// sliceResource is the resource of the endpoint slices in the tracker of the fake. +var sliceResource = schema.GroupVersionResource{ + Group: "discovery.k8s.io", + Version: "v1", + Resource: "endpointslices", +} + +// serviceResource is the resource of the services in the tracker of the fake. +var serviceResource = schema.GroupVersionResource{ + Group: "", + Version: "v1", + Resource: "services", +} + +// endpointsResource is the resource of the legacy endpoints in the tracker of the fake. +var endpointsResource = schema.GroupVersionResource{ + Group: "", + Version: "v1", + Resource: "endpoints", +} + +// alignSliceReadOrdinal is the ordinal of the slice read that the write of the slice +// does. +// +// The install reads the slice itself, and the write of the slice reads it again through +// kubeclientset. The second read is the one another actor can put its own slice in +// place before, so the reactors that inject a replacement key on it. +const alignSliceReadOrdinal = 2 + +// Test_installFakeSystemRoutingServiceRefusesTheSliceReplacedBeforeTheWrite checks +// that the ownership of the slice is decided on the object the write reads. +// +// The install reads the slice, and the write of the slice reads it again. Another actor +// can replace it in between the two reads, and the slice it puts in place must be +// refused and left alone. +func Test_installFakeSystemRoutingServiceRefusesTheSliceReplacedBeforeTheWrite(t *testing.T) { + cli := newFakeClient(nil, staleSlice("192.168.50.99", testPort)) + + // The second read is the one the write of the slice does, so the slice of + // another owner is put in place right before it. + foreign := foreignOwnedSlice("192.168.50.99", testPort) + foreign.SetUID(types.UID("EndpointSlice/another-owner")) + var reads atomic.Int64 + cli.PrependReactor("get", "endpointslices", func(action ktesting.Action) (bool, runtime.Object, error) { + if reads.Add(1) != alignSliceReadOrdinal { + return false, nil, nil + } + return true, foreign, nil + }) + + var writes atomic.Int64 + cli.PrependReactor("update", "endpointslices", func(action ktesting.Action) (bool, runtime.Object, error) { + writes.Add(1) + return false, nil, nil + }) + + err := installFakeSystemRoutingService(t.Context(), cli, testNamespace, testService, + "192.168.50.10", testPort) + require.Error(t, err) + assert.Contains(t, err.Error(), "is controlled by") + assert.Equal(t, int64(0), writes.Load(), "the slice of another owner must not be rewritten") + + // The cluster must still hold the slice it held before. + slice, err := cli.DiscoveryV1().EndpointSlices(testNamespace). + Get(t.Context(), routingEndpointSliceName(testService), meta.GetOptions{}) + require.NoError(t, err) + assert.Equal(t, []string{"192.168.50.99"}, slice.Endpoints[0].Addresses) +} + +// Test_installFakeSystemRoutingServiceRefusesTheSliceReplacedBeforeTheRetry checks +// that the ownership of the slice is decided again when the write is retried. +// +// The write of the slice reads it and finds the address type of the other family, and +// the retry reads it again. Another actor can replace the slice in between the two +// reads, and the slice it puts in place must be refused and left alone. +func Test_installFakeSystemRoutingServiceRefusesTheSliceReplacedBeforeTheRetry(t *testing.T) { + testCases := []struct { + name string + // replace makes the slice another actor puts in place. + replace func(slice *discovery.EndpointSlice) + // wantErr is the part of the error the reconcile must report. + wantErr string + }{ + { + name: "the managed by label changes", + replace: func(slice *discovery.EndpointSlice) { + slice.SetLabels(map[string]string{ + discovery.LabelServiceName: testService, + discovery.LabelManagedBy: "another-operator", + routingLabel: routingLabelValue, + }) + }, + wantErr: "is managed by", + }, + { + name: "the controller names another service", + replace: func(slice *discovery.EndpointSlice) { + // The fixture of another owner names another service, which is + // what this case puts in place. + slice.SetOwnerReferences( + foreignOwnedSlice("192.168.50.99", testPort).GetOwnerReferences(), + ) + }, + wantErr: "is controlled by", + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + cli := newFakeClient(nil, staleSlice("192.168.50.99", testPort)) + + // The slice of the other actor is put in place while the write of + // the slice reads it, so that the retry reads the new one. + var reads atomic.Int64 + cli.PrependReactor("get", "endpointslices", func(action ktesting.Action) (bool, runtime.Object, error) { + if reads.Add(1) != alignSliceReadOrdinal { + return false, nil, nil + } + + replaced := staleSlice("192.168.50.99", testPort) + replaced.SetUID(types.UID("EndpointSlice/another-owner")) + tc.replace(replaced) + require.NoError(t, cli.Tracker().Create(sliceResource, replaced, testNamespace)) + + // The write of the slice still reads the slice as it was + // before the other actor put its own in place. + return true, staleSlice("192.168.50.99", testPort), nil + }) + + var deleted []types.UID + cli.PrependReactor("delete", "endpointslices", func(action ktesting.Action) (bool, runtime.Object, error) { + del, ok := action.(ktesting.DeleteAction) + if ok && del.GetDeleteOptions().Preconditions != nil && del.GetDeleteOptions().Preconditions.UID != nil { + deleted = append(deleted, *del.GetDeleteOptions().Preconditions.UID) + } + return false, nil, nil + }) + + err := installFakeSystemRoutingService(t.Context(), cli, testNamespace, testService, + "fd00::1", testPort) + require.Error(t, err) + assert.Contains(t, err.Error(), tc.wantErr) + + // The slice of the other actor must be left alone. + slice, err := cli.DiscoveryV1().EndpointSlices(testNamespace). + Get(t.Context(), routingEndpointSliceName(testService), meta.GetOptions{}) + require.NoError(t, err) + assert.Equal(t, types.UID("EndpointSlice/another-owner"), slice.GetUID()) + + for _, uid := range deleted { + assert.NotEqual(t, types.UID("EndpointSlice/another-owner"), uid, + "the slice of the other actor must not be deleted") + } + }) + } +} + +// ipv6Slice builds the endpoint slice of this install for the other address family. +func ipv6Slice(address string, port int32) *discovery.EndpointSlice { + slice := staleSlice(address, port) + slice.AddressType = discovery.AddressTypeIPv6 + return slice +} + +// Test_installFakeSystemRoutingServiceAdoptsTheSliceOfARecreatedService checks that +// the slice of a routing service that was recreated is adopted rather than refused. +// +// The routing service is recreated whenever its spec changes, and the slice that the +// service of the past owned outlives it. The slice belongs to this installation, so +// the reconcile takes it over and rewrites its owner reference with the service of +// now. Refusing it would leave the worker unable to repair the routing it owns. +func Test_installFakeSystemRoutingServiceAdoptsTheSliceOfARecreatedService(t *testing.T) { + cli := newFakeClient(nil, staleSlice("192.168.50.99", testPort)) + + // The routing service was recreated, so it carries another uid than the one + // the slice names. Its spec matches the wanted one, so the reconcile keeps it. + recreated := &core.Service{ + ObjectMeta: meta.ObjectMeta{ + Namespace: testNamespace, + Name: testService, + UID: types.UID("Service/" + testService + "-2"), + }, + Spec: core.ServiceSpec{ + Type: core.ServiceTypeClusterIP, + Ports: []core.ServicePort{{Name: routingPortName, Port: testPort}}, + }, + } + require.NoError(t, cli.Tracker().Create(serviceResource, recreated, testNamespace)) + + require.NoError(t, installFakeSystemRoutingService(t.Context(), cli, testNamespace, testService, + "192.168.50.10", testPort)) + + slice, err := cli.DiscoveryV1().EndpointSlices(testNamespace). + Get(t.Context(), routingEndpointSliceName(testService), meta.GetOptions{}) + require.NoError(t, err) + + owner := kubemeta.GetControllerOfNoCopy(slice) + require.NotNil(t, owner, "the slice must be owned by the routing service") + assert.Equal(t, types.UID("Service/"+testService+"-2"), owner.UID, + "the owner reference must name the service of now") + + require.Len(t, slice.Endpoints, 1) + assert.Equal(t, []string{"192.168.50.10"}, slice.Endpoints[0].Addresses, + "the adopted slice must carry the wanted address") +} + +// Test_installFakeSystemRoutingServiceRemovesTheMirrorOfRecreatedEndpoints checks that +// the mirror of the legacy endpoints is removed even when the endpoints were recreated. +// +// The legacy endpoints are recreated whenever they are deleted, and the mirror of the +// endpoints of the past outlives them. The mirror belongs to this installation, so the +// reconcile removes it, for the same reason it removes the slice of a recreated service. +func Test_installFakeSystemRoutingServiceRemovesTheMirrorOfRecreatedEndpoints(t *testing.T) { + cli := newFakeClient(nil, legacyEndpoints("192.168.50.10", testPort), mirroredSlice("192.168.50.10", testPort)) + + // The legacy endpoints were recreated, so they carry another uid than the one + // the mirror names. + recreated := &core.Endpoints{ + ObjectMeta: meta.ObjectMeta{ + Namespace: testNamespace, + Name: testService, + UID: types.UID("Endpoints/" + testService + "-2"), + }, + } + require.NoError(t, cli.Tracker().Update(endpointsResource, recreated, testNamespace)) + + require.NoError(t, installFakeSystemRoutingService(t.Context(), cli, testNamespace, testService, + "192.168.50.10", testPort)) + + slices, err := cli.DiscoveryV1().EndpointSlices(testNamespace).List(t.Context(), meta.ListOptions{}) + require.NoError(t, err) + names := make([]string, 0, len(slices.Items)) + for i := range slices.Items { + names = append(names, slices.Items[i].GetName()) + } + assert.ElementsMatch(t, []string{routingEndpointSliceName(testService)}, names, + "the mirror of the legacy endpoints must be removed") +} + +// Test_routingEndpointSliceAlignKeepsTheDesiredStateApart checks that the object the +// align writes does not share its endpoints and ports with the desired object. +// +// The desired object is reused by the attempts of a write, so a change to the object +// that is written must not reach the object the next attempt writes. +func Test_routingEndpointSliceAlignKeepsTheDesiredStateApart(t *testing.T) { + desired, err := newRoutingEndpointSlice(testNamespace, testService, "192.168.50.10", testPort) + require.NoError(t, err) + + svc := &core.Service{ + ObjectMeta: meta.ObjectMeta{ + Namespace: testNamespace, + Name: testService, + UID: types.UID("Service/" + testService), + }, + } + + aligned, skip, err := routingEndpointSliceAlign(desired, svc)(staleSlice("192.168.50.10", testPort).DeepCopy()) + require.NoError(t, err) + require.False(t, skip) + + aligned.Endpoints[0].Addresses[0] = "203.0.113.7" + aligned.Ports[0].Port = ptr.To(int32(1)) + + assert.Equal(t, []string{"192.168.50.10"}, desired.Endpoints[0].Addresses, + "the desired endpoints must not be shared with the written object") + assert.Equal(t, testPort, *desired.Ports[0].Port, + "the desired ports must not be shared with the written object") +} + +// Test_installFakeSystemRoutingServiceConvergesOnTheSliceOfTheWantedFamily checks +// that the retry of the write converges on the slice that carries the wanted address +// family, and does not delete it. +// +// The slice of the other address family is written again with the wanted one while +// the write of the slice reads it, and the retry reads the new one. +func Test_installFakeSystemRoutingServiceConvergesOnTheSliceOfTheWantedFamily(t *testing.T) { + cli := newFakeClient(nil, staleSlice("192.168.50.99", testPort)) + + // The slice of this install is written again for the other address family + // while the write of the slice reads the one of the family it replaces. + wanted := ipv6Slice("192.168.50.99", testPort) + wanted.SetUID(types.UID("EndpointSlice/gpustack-operator-worker-direct-2")) + + var reads atomic.Int64 + cli.PrependReactor("get", "endpointslices", func(action ktesting.Action) (bool, runtime.Object, error) { + if reads.Add(1) != alignSliceReadOrdinal { + return false, nil, nil + } + + require.NoError(t, cli.Tracker().Create(sliceResource, wanted, testNamespace)) + return true, staleSlice("192.168.50.99", testPort), nil + }) + + var deleted []types.UID + cli.PrependReactor("delete", "endpointslices", func(action ktesting.Action) (bool, runtime.Object, error) { + del, ok := action.(ktesting.DeleteAction) + if ok && del.GetDeleteOptions().Preconditions != nil && del.GetDeleteOptions().Preconditions.UID != nil { + deleted = append(deleted, *del.GetDeleteOptions().Preconditions.UID) + } + return false, nil, nil + }) + + err := installFakeSystemRoutingService(t.Context(), cli, testNamespace, testService, + "fd00::1", testPort) + require.NoError(t, err) + + // The slice of the wanted family must not be deleted. + for _, uid := range deleted { + assert.NotEqual(t, types.UID("EndpointSlice/gpustack-operator-worker-direct-2"), uid, + "the slice of the wanted address family must not be deleted") + } + + slice, err := cli.DiscoveryV1().EndpointSlices(testNamespace). + Get(t.Context(), routingEndpointSliceName(testService), meta.GetOptions{}) + require.NoError(t, err) + assert.Equal(t, discovery.AddressTypeIPv6, slice.AddressType) + require.Len(t, slice.Endpoints, 1) + assert.Equal(t, []string{"fd00::1"}, slice.Endpoints[0].Addresses) +} + +// Test_installFakeSystemRoutingServiceBindsTheFamilyReplacementDeleteToTheSlice checks +// that the deletion of the slice of the other address family is bound to the slice that +// was read. +// +// A slice that was replaced before the deletion reached it must not be destroyed, and +// the reconcile must not report that it converged while the cluster holds a slice of +// the other address family. +func Test_installFakeSystemRoutingServiceBindsTheFamilyReplacementDeleteToTheSlice(t *testing.T) { + cli := newFakeClient(nil, staleSlice("192.168.50.99", testPort)) + + var deleted []types.UID + cli.PrependReactor("delete", "endpointslices", func(action ktesting.Action) (bool, runtime.Object, error) { + del, ok := action.(ktesting.DeleteAction) + if ok && del.GetDeleteOptions().Preconditions != nil && del.GetDeleteOptions().Preconditions.UID != nil { + deleted = append(deleted, *del.GetDeleteOptions().Preconditions.UID) + } + // The slice was replaced before the deletion reached it. + return true, nil, kerrors.NewConflict(discovery.Resource("endpointslices"), + routingEndpointSliceName(testService), fmt.Errorf("the uid precondition does not hold")) + }) + + err := installFakeSystemRoutingService(t.Context(), cli, testNamespace, testService, + "fd00::1", testPort) + require.Error(t, err, "a deletion that never succeeds must not be reported as a convergence") + assert.Contains(t, err.Error(), "has address type") + + // Every deletion is bound to the slice that was read. + require.NotEmpty(t, deleted) + for _, uid := range deleted { + assert.Equal(t, types.UID("EndpointSlice/"+routingEndpointSliceName(testService)), uid) + } + + // The cluster still holds the slice of the other address family, and the + // reconcile has said so. + slice, err := cli.DiscoveryV1().EndpointSlices(testNamespace). + Get(t.Context(), routingEndpointSliceName(testService), meta.GetOptions{}) + require.NoError(t, err) + assert.Equal(t, discovery.AddressTypeIPv4, slice.AddressType) +} + +// Test_installFakeSystemRoutingServiceBindsTheRetryDeleteToTheSlice checks that the +// deletion of the retry of the write is bound to the slice the retry read. +// +// The slice the retry reads is not the slice the install read, so the deletion must +// carry the uid of the one it read, and the reconcile must not report that it converged +// while the cluster holds a slice of the other address family. +func Test_installFakeSystemRoutingServiceBindsTheRetryDeleteToTheSlice(t *testing.T) { + cli := newFakeClient(nil, staleSlice("192.168.50.99", testPort)) + + // The slice of this install is written again under a uid of its own, the + // way another worker of an earlier version of the operator writes it. + rewritten := staleSlice("192.168.50.99", testPort) + rewritten.SetUID(types.UID("EndpointSlice/gpustack-operator-worker-direct-2")) + + var reads atomic.Int64 + cli.PrependReactor("get", "endpointslices", func(action ktesting.Action) (bool, runtime.Object, error) { + if reads.Add(1) != alignSliceReadOrdinal { + return false, nil, nil + } + + // The write of the slice reads the slice of the other address family + // as it was before it was deleted, and the slice of this install is + // written again in the meantime. + require.NoError(t, cli.Tracker().Create(sliceResource, rewritten, testNamespace)) + return true, staleSlice("192.168.50.99", testPort), nil + }) + + var deleted []types.UID + var deletes atomic.Int64 + cli.PrependReactor("delete", "endpointslices", func(action ktesting.Action) (bool, runtime.Object, error) { + del, ok := action.(ktesting.DeleteAction) + if ok && del.GetDeleteOptions().Preconditions != nil && del.GetDeleteOptions().Preconditions.UID != nil { + deleted = append(deleted, *del.GetDeleteOptions().Preconditions.UID) + } + if deletes.Add(1) == 1 { + // The deletion of the install reaches the slice it read. + return false, nil, nil + } + // The slice was replaced again before the deletion of the retry reached + // it. + return true, nil, kerrors.NewConflict(discovery.Resource("endpointslices"), + routingEndpointSliceName(testService), fmt.Errorf("the uid precondition does not hold")) + }) + + err := installFakeSystemRoutingService(t.Context(), cli, testNamespace, testService, + "fd00::1", testPort) + require.Error(t, err, "a deletion that never succeeds must not be reported as a convergence") + assert.Contains(t, err.Error(), "has address type") + + // The first deletion is bound to the slice the install read, and the ones + // of the retry are bound to the slice the retry read. + require.GreaterOrEqual(t, len(deleted), 2) + assert.Equal(t, types.UID("EndpointSlice/gpustack-operator-worker-direct"), deleted[0]) + assert.Equal(t, types.UID("EndpointSlice/gpustack-operator-worker-direct-2"), deleted[1]) + + // The cluster still holds the slice of the other address family, and the + // reconcile has said so. + slice, err := cli.DiscoveryV1().EndpointSlices(testNamespace). + Get(t.Context(), routingEndpointSliceName(testService), meta.GetOptions{}) + require.NoError(t, err) + assert.Equal(t, discovery.AddressTypeIPv4, slice.AddressType) +} diff --git a/pkg/systemkuberess/service.go b/pkg/systemkuberess/service.go index 2969e3140..0caf5f6cf 100644 --- a/pkg/systemkuberess/service.go +++ b/pkg/systemkuberess/service.go @@ -57,7 +57,7 @@ func installFakeSystemRoutingService(ctx context.Context, cli kubernetes.Interfa return err } - svc, err := installRoutingService(ctx, cli, nsName, svcName, port) + svc, err := installRoutingService(ctx, cli, nsName, svcName, port, desiredSlice.AddressType) if err != nil { return err } @@ -103,7 +103,7 @@ func checkRoutingAccess(ctx context.Context, cli kubernetes.Interface, nsName st } verbs := []string{"get", "list", "create", "update", "delete"} - reviews := make(kubereviewself.Reviews, 0, len(verbs)) + reviews := make(kubereviewself.Reviews, 0, len(verbs)+1) for _, verb := range verbs { reviews = append(reviews, kubereviewself.Review{ ResourceAttributes: &authz.ResourceAttributes{ @@ -116,19 +116,51 @@ func checkRoutingAccess(ctx context.Context, cli kubernetes.Interface, nsName st }) } + // The owner reference of the slice blocks the deletion of the routing service, + // so the api server asks for the permission to update the finalizers of the + // service before it accepts the slice. + reviews = append(reviews, kubereviewself.Review{ + ResourceAttributes: &authz.ResourceAttributes{ + Verb: "update", + Group: core.SchemeGroupVersion.Group, + Version: core.SchemeGroupVersion.Version, + Resource: "services", + Subresource: "finalizers", + Namespace: nsName, + }, + }) + return kubereviewself.CanDo(ctx, cli.AuthorizationV1().SelfSubjectAccessReviews(), reviews) } // installRoutingService creates the selectorless service the extension api services are // fronted by, and returns it as the cluster holds it. -func installRoutingService(ctx context.Context, cli kubernetes.Interface, nsName, svcName string, port int32) (*core.Service, error) { +func installRoutingService(ctx context.Context, cli kubernetes.Interface, nsName, svcName string, port int32, + addressType discovery.AddressType, +) (*core.Service, error) { expected := newRoutingService(nsName, svcName, port) svcCompareFn := func(actual *core.Service) bool { - return actual.Spec.Type == expected.Spec.Type && - actual.Spec.ClusterIP == expected.Spec.ClusterIP && - slices.ContainsFunc(actual.Spec.Ports, func(ap core.ServicePort) bool { - return ap.Port == expected.Spec.Ports[0].Port - }) + if actual.Spec.Type != expected.Spec.Type { + return false + } + // The cluster ip is compared only when the wanted service asks for one. + // The operator does not manage it, and the api server allocates one for + // the service of the cluster, so comparing it against the empty ip of the + // wanted service would delete and recreate the service on every start. + // The recreation takes the endpoint slice of this installation with it, + // because the slice is a dependent of the service that blocks its + // deletion, and it changes the address of the routing as well. + if expected.Spec.ClusterIP != "" && actual.Spec.ClusterIP != expected.Spec.ClusterIP { + return false + } + if actual.Spec.ClusterIP != "" && actual.Spec.ClusterIP != core.ClusterIPNone { + if actualType, err := routingAddressType(actual.Spec.ClusterIP); err == nil && actualType != addressType { + return false + } + } + return slices.ContainsFunc(actual.Spec.Ports, func(ap core.ServicePort) bool { + return ap.Port == expected.Spec.Ports[0].Port + }) } actual, err := kubeclientset.Create(ctx, cli.CoreV1().Services(nsName), expected, diff --git a/pkg/systemkuberess/service_test.go b/pkg/systemkuberess/service_test.go index dd35c8ce6..5d06dcbef 100644 --- a/pkg/systemkuberess/service_test.go +++ b/pkg/systemkuberess/service_test.go @@ -36,6 +36,9 @@ type wantRouting struct { endpoints bool // slice is true when the endpoint slice must exist. slice bool + // foreignSlice is true when the endpoint slice exists and belongs to somebody + // else, so the reconcile must have left it alone. + foreignSlice bool // address is the address the routing must point at. address string // port is the routing port. @@ -216,6 +219,39 @@ func Test_installFakeSystemRoutingService(t *testing.T) { }, }, }, + { + name: "adopts a slice without an owner", + input: routingInput{address: "192.168.50.10", port: testPort}, + objects: []runtime.Object{ + ownerlessSlice("192.168.50.10", testPort), + }, + want: wantRouting{ + service: true, + endpoints: true, + slice: true, + address: "192.168.50.10", + port: testPort, + addressType: discovery.AddressTypeIPv4, + sliceNames: []string{"gpustack-operator-worker-direct"}, + }, + }, + { + name: "refuses a slice of another owner", + input: routingInput{address: "192.168.50.10", port: testPort}, + objects: []runtime.Object{ + foreignOwnedSlice("192.168.50.10", testPort), + }, + want: wantRouting{ + service: true, + endpoints: true, + foreignSlice: true, + address: "192.168.50.10", + port: testPort, + addressType: discovery.AddressTypeIPv4, + sliceNames: []string{"gpustack-operator-worker-direct"}, + }, + wantErr: "is controlled by", + }, { name: "keeps the endpoints of others", input: routingInput{address: "192.168.50.10", port: testPort}, @@ -254,6 +290,23 @@ func Test_installFakeSystemRoutingService(t *testing.T) { }, wantErr: "denied", }, + { + // The owner reference of the slice blocks the deletion of the routing + // service, so the api server asks for the permission to update the + // finalizers of the service before it accepts the slice. A binding + // without that permission stops the worker before it writes anything. + name: "reports a denied finalizers access review", + input: routingInput{address: "192.168.50.10", port: testPort}, + deny: func(attrs *authz.ResourceAttributes) bool { + return attrs.Resource == "services" && + attrs.Subresource == "finalizers" && + attrs.Verb == "update" + }, + want: wantRouting{ + sliceNames: nil, + }, + wantErr: "denied", + }, { name: "reports a failed endpoint slice creation", input: routingInput{address: "192.168.50.10", port: testPort}, @@ -335,6 +388,111 @@ func Test_installFakeSystemRoutingService(t *testing.T) { } } +// Test_installFakeSystemRoutingServiceConvergesWithoutWrites checks that an install +// that finds the routing it wants writes nothing at all. +// +// The state after a repeated install is the same either way, so only the writes tell +// the two apart. +func Test_installFakeSystemRoutingServiceConvergesWithoutWrites(t *testing.T) { + cli := newFakeClient(nil) + + require.NoError(t, installFakeSystemRoutingService(t.Context(), cli, testNamespace, testService, + "192.168.50.10", testPort)) + + var writes atomic.Int64 + count := func(action ktesting.Action) (bool, runtime.Object, error) { + if action.GetResource().Resource != "selfsubjectaccessreviews" { + writes.Add(1) + } + return false, nil, nil + } + cli.PrependReactor("create", "*", count) + cli.PrependReactor("update", "*", count) + cli.PrependReactor("delete", "*", count) + + require.NoError(t, installFakeSystemRoutingService(t.Context(), cli, testNamespace, testService, + "192.168.50.10", testPort)) + assert.Equal(t, int64(0), writes.Load(), "a converged install must not write") +} + +// Test_installFakeSystemRoutingServiceKeepsTheRoutingServiceOfTheCluster checks that +// the routing service of the cluster is kept. +// +// The api server allocates a cluster ip for the service, and the operator does not +// manage it. Comparing the cluster ip of the wanted service, which asks for none, with +// the cluster ip of the service of the cluster would delete and recreate the service on +// every start. The recreation takes the endpoint slice of this installation with it, +// because the slice is a dependent of the service that blocks its deletion, and it +// changes the address of the routing as well. +func Test_installFakeSystemRoutingServiceKeepsTheRoutingServiceOfTheCluster(t *testing.T) { + allocated := &core.Service{ + ObjectMeta: meta.ObjectMeta{ + Namespace: testNamespace, + Name: testService, + }, + Spec: core.ServiceSpec{ + Type: core.ServiceTypeClusterIP, + ClusterIP: "10.96.0.42", + Ports: []core.ServicePort{ + {Name: routingPortName, Port: testPort}, + }, + }, + } + cli := newFakeClient(nil, allocated) + + var deleted atomic.Int64 + cli.PrependReactor("delete", "services", func(action ktesting.Action) (bool, runtime.Object, error) { + deleted.Add(1) + return false, nil, nil + }) + + require.NoError(t, installFakeSystemRoutingService(t.Context(), cli, testNamespace, testService, + "192.168.50.10", testPort)) + + assert.Equal(t, int64(0), deleted.Load(), "the routing service of the cluster must be kept") + + svc, err := cli.CoreV1().Services(testNamespace).Get(t.Context(), testService, meta.GetOptions{}) + require.NoError(t, err) + assert.Equal(t, "10.96.0.42", svc.Spec.ClusterIP, + "the cluster ip of the service of the cluster must be kept") + + checkRouting(t, cli, wantRouting{ + service: true, + endpoints: true, + slice: true, + address: "192.168.50.10", + port: testPort, + addressType: discovery.AddressTypeIPv4, + sliceNames: []string{"gpustack-operator-worker-direct"}, + }) + + t.Run("recreates the routing service when the cluster ip family differs from the routing address", func(t *testing.T) { + cli := newFakeClient(nil, allocated.DeepCopy()) + + var deleted atomic.Int64 + cli.PrependReactor("delete", "services", func(action ktesting.Action) (bool, runtime.Object, error) { + deleted.Add(1) + return false, nil, nil + }) + + require.NoError(t, installFakeSystemRoutingService(t.Context(), cli, testNamespace, testService, + "fd00::1", testPort)) + + assert.Equal(t, int64(1), deleted.Load(), + "the routing service must be recreated when its cluster ip family differs from the routing address") + + checkRouting(t, cli, wantRouting{ + service: true, + endpoints: true, + slice: true, + address: "fd00::1", + port: testPort, + addressType: discovery.AddressTypeIPv6, + sliceNames: []string{"gpustack-operator-worker-direct"}, + }) + }) +} + // Test_installFakeSystemRoutingServiceCancelled checks that a canceled call is reported // as it is, and that a context that is already done stops the reconcile from trying // again. @@ -415,7 +573,16 @@ func checkRouting(t *testing.T, cli *kubefake.Clientset, want wantRouting) { slice, err := cli.DiscoveryV1().EndpointSlices(testNamespace). Get(t.Context(), routingEndpointSliceName(testService), meta.GetOptions{}) - if want.slice { + switch { + case want.foreignSlice: + require.NoError(t, err) + owner := kubemeta.GetControllerOfNoCopy(slice) + require.NotNil(t, owner, "the slice of another owner must keep its owner") + if want.service { + assert.NotEqual(t, svc.GetUID(), owner.UID, + "the reconcile must not take the slice of another owner over") + } + case want.slice: require.NoError(t, err) assert.Equal(t, testService, slice.GetLabels()[discovery.LabelServiceName]) assert.Equal(t, routingManagedBy, slice.GetLabels()[discovery.LabelManagedBy]) @@ -435,7 +602,7 @@ func checkRouting(t *testing.T, cli *kubefake.Clientset, want wantRouting) { if want.service { assert.Equal(t, svc.GetUID(), owner.UID) } - } else { + default: assert.True(t, kerrors.IsNotFound(err), "the endpoint slice must not exist") } @@ -607,6 +774,30 @@ func obsoleteSlice(address string, port int32) *discovery.EndpointSlice { return slice } +// ownerlessSlice builds the endpoint slice an earlier version of this operator left +// without an owner. +func ownerlessSlice(address string, port int32) *discovery.EndpointSlice { + slice := staleSlice(address, port) + slice.SetOwnerReferences(nil) + return slice +} + +// foreignOwnedSlice builds the endpoint slice of another owner, carrying the labels of +// this operator. +func foreignOwnedSlice(address string, port int32) *discovery.EndpointSlice { + slice := staleSlice(address, port) + slice.SetOwnerReferences([]meta.OwnerReference{ + { + APIVersion: "v1", + Kind: "Service", + Name: "other-service", + UID: types.UID("Service/other-service"), + Controller: ptr.To(true), + }, + }) + return slice +} + // foreignSlice builds the endpoint slice of another service. func foreignSlice() *discovery.EndpointSlice { return &discovery.EndpointSlice{ From 4fa1f6b798d9176ab33ce4ecebd8c1d4ffe06ac0 Mon Sep 17 00:00:00 2001 From: thxCode Date: Mon, 26 Oct 2026 13:14:43 +0800 Subject: [PATCH 3/5] ci: align the helm pin of the workflows with the tooling The chart and site workflows installed helm v3.21.0. The tooling moves to v3.21.4 in the same series, so the workflows move with it: a workflow that renders or packages a chart with another helm than the repository develops against can produce an artifact the local gates never saw. The Go version needs no change: every workflow already reads the version from a go.mod (the root one, source/go.mod in the site checkout and gpustack.ai/gpustack/go.mod in the api checkout), so the 1.27.3 of the root go.mod is what they resolve. Signed-off-by: thxCode --- .github/workflows/chart.yml | 2 +- .github/workflows/site.yml | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/chart.yml b/.github/workflows/chart.yml index 19ab18152..b0524601d 100644 --- a/.github/workflows/chart.yml +++ b/.github/workflows/chart.yml @@ -191,7 +191,7 @@ jobs: - name: Setup Helm uses: azure/setup-helm@v4 with: - version: v3.21.0 + version: v3.21.4 - name: Setup Toolbox Cache uses: actions/cache@v4 with: diff --git a/.github/workflows/site.yml b/.github/workflows/site.yml index 9ac0b84c4..3ec7fb367 100644 --- a/.github/workflows/site.yml +++ b/.github/workflows/site.yml @@ -101,7 +101,7 @@ jobs: if: steps.ref.outputs.kind != 'development' uses: azure/setup-helm@v4 with: - version: v3.21.0 + version: v3.21.4 - name: Prepare release chart if: steps.ref.outputs.kind != 'development' run: | From 82f65666ad26e4451e5936e8e578cc8c8b8c6239 Mon Sep 17 00:00:00 2001 From: thxCode Date: Mon, 26 Oct 2026 14:09:40 +0800 Subject: [PATCH 4/5] docs(contribute): rename kind setup guide to local cluster setup Signed-off-by: thxCode --- docs/README.md | 4 ++-- docs/contribute/development.md | 2 +- docs/contribute/{kind.md => local-cluster-setup.md} | 2 +- site/data/navigation.yaml | 2 +- 4 files changed, 5 insertions(+), 5 deletions(-) rename docs/contribute/{kind.md => local-cluster-setup.md} (99%) diff --git a/docs/README.md b/docs/README.md index 4d51e6bbb..9a71e04ee 100644 --- a/docs/README.md +++ b/docs/README.md @@ -151,7 +151,7 @@ A skill link names a procedure; the entry's other fields carry shared facts. - ID: `development` - Use for: build, lint, test and codegen commands; commit and PR conventions; contributor invariants; release steps; site publication and translations. - Aliases: `make`, `lint`, `codegen`, `vendored dependencies`, `staging`, `Hugo`, `GitHub Pages` -- Guides: commands and targets in [Development](contribute/development.md); a local test cluster in [kind Setup](contribute/kind.md); startup invariants in [Internals](contribute/internals.md); orientation in [Architecture](getting-started/architecture.md). +- Guides: commands and targets in [Development](contribute/development.md); a local test cluster in [Local Cluster Setup](contribute/local-cluster-setup.md); startup invariants in [Internals](contribute/internals.md); orientation in [Architecture](getting-started/architecture.md). - Code: [Makefile](../Makefile) and the scripts behind it in [hack](../hack/) · site publication in [Site workflow](../.github/workflows/site.yml) and [publisher](../hack/site/publish.py) · patched modules in [staging](../staging/) · generators in [gen](../gen/) · upgrade compatibility checks in [upgrade-compat.sh](../hack/check/upgrade-compat.sh) · the chart test's kubeconfig isolation in [helm.sh](../hack/lib/helm.sh) and [helm-chart-test-selftest.sh](../hack/check/helm-chart-test-selftest.sh) - Related: [API Changes](#api-changes) (what `make generate` owns) · [Installation](#installation) (chart targets and vendoring) - Skills: [gpustack-operator-overview](../.agents/skills/gpustack-operator-overview/SKILL.md) (codebase tour) · [gpustack-operator-docs](../.agents/skills/gpustack-operator-docs/SKILL.md) (documentation changes) · [gpustack-operator-code-review](../.agents/skills/gpustack-operator-code-review/SKILL.md) (PR review criteria) · [gpustack-operator-issue-pr](../.agents/skills/gpustack-operator-issue-pr/SKILL.md) (issue and PR titles) · [gpustack-operator-release](../.agents/skills/gpustack-operator-release/SKILL.md) (cut a release) · [gpustack-operator-generate](../.agents/skills/gpustack-operator-generate/SKILL.md) (API codegen) @@ -291,7 +291,7 @@ A skill link names a procedure; the entry's other fields carry shared facts. | Page | Description | |---|---| | [Development](contribute/development.md) | Build, generate and lint the project. | -| [kind Setup](contribute/kind.md) | Install kind and provide a cluster for development tests. | +| [Local Cluster Setup](contribute/local-cluster-setup.md) | Install kind and provide a cluster for development tests. | | [Internals](contribute/internals.md) | Review startup and naming constraints. | ## Conventions diff --git a/docs/contribute/development.md b/docs/contribute/development.md index e57b74c17..7d6d414b1 100644 --- a/docs/contribute/development.md +++ b/docs/contribute/development.md @@ -17,7 +17,7 @@ before changing the operator's behavior. ## Commands For cluster-backed tests, provide a kind cluster and its kubeconfig/context. Follow -[kind Setup](kind.md) for manual installation and connection checks. +[Local Cluster Setup](local-cluster-setup.md) for manual installation and connection checks. `make [args]` runs `hack/.sh` and forwards `args` to it. All builds use `CGO_ENABLED=1`, `GODEBUG=gotypesalias=0` and the build tags `goccy netgo`. diff --git a/docs/contribute/kind.md b/docs/contribute/local-cluster-setup.md similarity index 99% rename from docs/contribute/kind.md rename to docs/contribute/local-cluster-setup.md index e39caada3..ae85ee840 100644 --- a/docs/contribute/kind.md +++ b/docs/contribute/local-cluster-setup.md @@ -1,4 +1,4 @@ -# kind Setup +# Local Cluster Setup Use [kind](https://kind.sigs.k8s.io/docs/user/quick-start/) to provide a local Kubernetes cluster for development and integration tests. Keep its kubeconfig separate from other clusters. diff --git a/site/data/navigation.yaml b/site/data/navigation.yaml index b16c97d4b..1af005ed7 100644 --- a/site/data/navigation.yaml +++ b/site/data/navigation.yaml @@ -188,5 +188,5 @@ labels: /docs/reference/instance-type-unit-resources/: Resource Presets /docs/reference/instance-metrics/: Instance Metrics /docs/contribute/development/: Development - /docs/contribute/kind/: kind Setup + /docs/contribute/local-cluster-setup/: Local Cluster Setup /docs/contribute/internals/: Internals From b7d1b74864158dd2302dc15dfe54f97a83db2114 Mon Sep 17 00:00:00 2001 From: thxCode Date: Mon, 26 Oct 2026 14:30:08 +0800 Subject: [PATCH 5/5] fix(worker): set routing Service IPFamilies and scope EndpointSlice cleanup Signed-off-by: thxCode --- docs/contribute/internals.md | 8 +++---- pkg/systemkuberess/endpoint_slice.go | 3 ++- pkg/systemkuberess/service.go | 28 ++++++++++++++++++----- pkg/systemkuberess/service_test.go | 33 ++++++++++++++++++++++++++++ 4 files changed, 61 insertions(+), 11 deletions(-) diff --git a/docs/contribute/internals.md b/docs/contribute/internals.md index 956c7cb54..1e077a0ac 100644 --- a/docs/contribute/internals.md +++ b/docs/contribute/internals.md @@ -127,10 +127,10 @@ slices it owns and no longer needs: the mirror of the legacy `Endpoints`, and a version left under another name. It removes a slice only when the owner reference names the routing `Service` or the legacy `Endpoints` of this worker, so a slice of another service stays. -The owner is recognized by the name it carries, and not by its uid, because the routing `Service` and -the legacy `Endpoints` are recreated whenever their spec changes. A slice the resource of the past -owned still belongs to the worker, so the worker adopts it and rewrites its owner reference; refusing -it would leave the worker unable to repair the routing it owns. +The owner is recognized by the name it carries, and not by its uid, because the routing `Service` is +recreated whenever its spec changes and the legacy `Endpoints` are recreated whenever they are deleted. +A slice the resource of the past owned still belongs to the worker, so the worker adopts it and rewrites +its owner reference; refusing it would leave the worker unable to repair the routing it owns. One worker owns the routing of a cluster at a time. The namespace and the routing service name are the same in every installation, so two workers pointing at one cluster write the same slice, and the last diff --git a/pkg/systemkuberess/endpoint_slice.go b/pkg/systemkuberess/endpoint_slice.go index b992b9c49..52279fd0f 100644 --- a/pkg/systemkuberess/endpoint_slice.go +++ b/pkg/systemkuberess/endpoint_slice.go @@ -357,7 +357,8 @@ func cleanupObsoleteRoutingSlices(ctx context.Context, cli kubernetes.Interface, } sliceCli := cli.DiscoveryV1().EndpointSlices(nsName) - slices, err := sliceCli.List(ctx, meta.ListOptions{}) + selector := discovery.LabelServiceName + "=" + svc.GetName() + slices, err := sliceCli.List(ctx, meta.ListOptions{LabelSelector: selector}) if err != nil { return fmt.Errorf("list routing endpoint slices: %w", err) } diff --git a/pkg/systemkuberess/service.go b/pkg/systemkuberess/service.go index 0caf5f6cf..bce9f5969 100644 --- a/pkg/systemkuberess/service.go +++ b/pkg/systemkuberess/service.go @@ -138,7 +138,7 @@ func checkRoutingAccess(ctx context.Context, cli kubernetes.Interface, nsName st func installRoutingService(ctx context.Context, cli kubernetes.Interface, nsName, svcName string, port int32, addressType discovery.AddressType, ) (*core.Service, error) { - expected := newRoutingService(nsName, svcName, port) + expected := newRoutingService(nsName, svcName, port, addressType) svcCompareFn := func(actual *core.Service) bool { if actual.Spec.Type != expected.Spec.Type { return false @@ -153,13 +153,19 @@ func installRoutingService(ctx context.Context, cli kubernetes.Interface, nsName if expected.Spec.ClusterIP != "" && actual.Spec.ClusterIP != expected.Spec.ClusterIP { return false } + if len(actual.Spec.IPFamilies) > 0 && actual.Spec.IPFamilies[0] != expected.Spec.IPFamilies[0] { + return false + } if actual.Spec.ClusterIP != "" && actual.Spec.ClusterIP != core.ClusterIPNone { if actualType, err := routingAddressType(actual.Spec.ClusterIP); err == nil && actualType != addressType { return false } } + expectedPort := expected.Spec.Ports[0] return slices.ContainsFunc(actual.Spec.Ports, func(ap core.ServicePort) bool { - return ap.Port == expected.Spec.Ports[0].Port + return ap.Name == expectedPort.Name && + ap.Port == expectedPort.Port && + (ap.Protocol == "" || ap.Protocol == expectedPort.Protocol) }) } @@ -176,7 +182,7 @@ func installRoutingService(ctx context.Context, cli kubernetes.Interface, nsName // // The service has no selector, so no controller picks its endpoints for it, and the // endpoint slice below is the only thing that routes to the system. -func newRoutingService(nsName, svcName string, port int32) *core.Service { +func newRoutingService(nsName, svcName string, port int32, addressType discovery.AddressType) *core.Service { return &core.Service{ ObjectMeta: meta.ObjectMeta{ Namespace: nsName, @@ -186,17 +192,27 @@ func newRoutingService(nsName, svcName string, port int32) *core.Service { }, }, Spec: core.ServiceSpec{ - Type: core.ServiceTypeClusterIP, + Type: core.ServiceTypeClusterIP, + IPFamilies: []core.IPFamily{routingIPFamily(addressType)}, Ports: []core.ServicePort{ { - Name: routingPortName, - Port: port, + Name: routingPortName, + Port: port, + Protocol: core.ProtocolTCP, }, }, }, } } +// routingIPFamily returns the service IP family matching the endpoint slice address type. +func routingIPFamily(addressType discovery.AddressType) core.IPFamily { + if addressType == discovery.AddressTypeIPv6 { + return core.IPv6Protocol + } + return core.IPv4Protocol +} + // installRoutingEndpoints creates the legacy endpoints of the system routing, and returns // them as the cluster holds them. // diff --git a/pkg/systemkuberess/service_test.go b/pkg/systemkuberess/service_test.go index 5d06dcbef..be9a8db71 100644 --- a/pkg/systemkuberess/service_test.go +++ b/pkg/systemkuberess/service_test.go @@ -481,6 +481,11 @@ func Test_installFakeSystemRoutingServiceKeepsTheRoutingServiceOfTheCluster(t *t assert.Equal(t, int64(1), deleted.Load(), "the routing service must be recreated when its cluster ip family differs from the routing address") + svc, err := cli.CoreV1().Services(testNamespace).Get(t.Context(), testService, meta.GetOptions{}) + require.NoError(t, err) + assert.Equal(t, []core.IPFamily{core.IPv6Protocol}, svc.Spec.IPFamilies, + "the recreated routing service must request the IPv6 family") + checkRouting(t, cli, wantRouting{ service: true, endpoints: true, @@ -491,6 +496,34 @@ func Test_installFakeSystemRoutingServiceKeepsTheRoutingServiceOfTheCluster(t *t sliceNames: []string{"gpustack-operator-worker-direct"}, }) }) + + t.Run("recreates the routing service when the port name differs", func(t *testing.T) { + renamedPort := allocated.DeepCopy() + renamedPort.Spec.Ports[0].Name = "http" + cli := newFakeClient(nil, renamedPort) + + var deleted atomic.Int64 + cli.PrependReactor("delete", "services", func(action ktesting.Action) (bool, runtime.Object, error) { + deleted.Add(1) + return false, nil, nil + }) + + require.NoError(t, installFakeSystemRoutingService(t.Context(), cli, testNamespace, testService, + "192.168.50.10", testPort)) + + assert.Equal(t, int64(1), deleted.Load(), + "the routing service must be recreated when its port name differs") + + checkRouting(t, cli, wantRouting{ + service: true, + endpoints: true, + slice: true, + address: "192.168.50.10", + port: testPort, + addressType: discovery.AddressTypeIPv4, + sliceNames: []string{"gpustack-operator-worker-direct"}, + }) + }) } // Test_installFakeSystemRoutingServiceCancelled checks that a canceled call is reported