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: | diff --git a/deploy/gpustack-operator/chart/templates/worker/serviceaccount.yaml b/deploy/gpustack-operator/chart/templates/worker/serviceaccount.yaml index e524dfa30..614133a25 100644 --- a/deploy/gpustack-operator/chart/templates/worker/serviceaccount.yaml +++ b/deploy/gpustack-operator/chart/templates/worker/serviceaccount.yaml @@ -11,6 +11,17 @@ 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, 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 metadata: diff --git a/docs/README.md b/docs/README.md index c260d3c11..9a71e04ee 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) @@ -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/internals.md b/docs/contribute/internals.md index 6592a4307..1e077a0ac 100644 --- a/docs/contribute/internals.md +++ b/docs/contribute/internals.md @@ -92,6 +92,50 @@ 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. + +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 +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/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/pkg/systemkuberess/endpoint_slice.go b/pkg/systemkuberess/endpoint_slice.go new file mode 100644 index 000000000..52279fd0f --- /dev/null +++ b/pkg/systemkuberess/endpoint_slice.go @@ -0,0 +1,436 @@ +package systemkuberess + +import ( + "context" + "errors" + "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" +) + +// 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 +// 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. +// +// 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 { + sliceCli := cli.DiscoveryV1().EndpointSlices(nsName) + + actual, err := sliceCli.Get(ctx, desired.GetName(), meta.GetOptions{}) + switch { + case err == nil: + err = checkRoutingSliceIdentity(actual, svc) + if err != nil { + return err + } + if actual.AddressType != desired.AddressType { + 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. + default: + return fmt.Errorf("get routing endpoint slice %q: %w", desired.GetName(), err) + } + + kubemeta.ControlOn(desired, svc, core.SchemeGroupVersion.WithKind("Service")) + + // 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, &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")) + + // 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 + } + return aligned, false, nil + } +} + +// 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 +// 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) + 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) + } + + for i := range slices.Items { + slice := &slices.Items[i] + if slice.GetName() == keepName { + continue + } + + if !isObsoleteRoutingSlice(slice, svc, ep) { + // Not owned by this installation, keep it. + continue + } + + 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 b1a727ca0..bce9f5969 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, desiredSlice.AddressType) + 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,192 @@ 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)+1) + 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, + }, + }) + } + + // 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, + addressType discovery.AddressType, +) (*core.Service, error) { + expected := newRoutingService(nsName, svcName, port, addressType) + svcCompareFn := func(actual *core.Service) bool { + 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 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.Name == expectedPort.Name && + ap.Port == expectedPort.Port && + (ap.Protocol == "" || ap.Protocol == expectedPort.Protocol) + }) + } + + 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, addressType discovery.AddressType) *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, + Type: core.ServiceTypeClusterIP, + IPFamilies: []core.IPFamily{routingIPFamily(addressType)}, Ports: []core.ServicePort{ { - Name: "https", - Port: port, + Name: routingPortName, + Port: port, + Protocol: core.ProtocolTCP, }, }, }, } - 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 - }) +} + +// 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 +} - eSvc, err = kubeclientset.Create(ctx, svcCli, eSvc, - kubeclientset.WithRecreateIfDuplicated(svcCompareFn)) +// 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 + } + + 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..be9a8db71 --- /dev/null +++ b/pkg/systemkuberess/service_test.go @@ -0,0 +1,865 @@ +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 + // 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. + 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: "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}, + 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", + }, + { + // 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}, + 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_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") + + 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, + slice: true, + address: "fd00::1", + port: testPort, + addressType: discovery.AddressTypeIPv6, + 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 +// 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{}) + 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]) + 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) + } + default: + 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 +} + +// 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{ + 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)}, + }, + } +} 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