diff --git a/api/v1alpha/instance_types.go b/api/v1alpha/instance_types.go index 643c5a12..b280cb66 100644 --- a/api/v1alpha/instance_types.go +++ b/api/v1alpha/instance_types.go @@ -449,6 +449,13 @@ type InstanceStatus struct { // // +kubebuilder:validation:Optional Controller *InstanceControllerStatus `json:"controller,omitempty"` + + // Suspended, when true, indicates that the instance's process should be stopped + // without releasing its placement, disk attachments, or quota allocation. + // The provider controller stops the running container/VM. + // + // +kubebuilder:validation:Optional + Suspended bool `json:"suspended,omitempty"` } type InstanceControllerStatus struct { @@ -587,6 +594,15 @@ const ( // InstanceAvailableReasonAvailable indicates that the instance is available InstanceAvailableReasonAvailable = "Available" + // InstanceReadyReasonSuspended indicates the instance is intentionally + // stopped due to project suspension. Its placement, disk, and quota + // allocation are retained; the process will restart from disk on reinstatement. + InstanceReadyReasonSuspended = "Suspended" + + // InstanceAvailableReasonSuspended indicates the instance is suspended + // and is not currently serving traffic. + InstanceAvailableReasonSuspended = "Suspended" + // InstanceProgrammedReasonPendingProgramming indicates that the instance has not been programmed InstanceProgrammedReasonPendingProgramming = "PendingProgramming" diff --git a/api/v1alpha/workloaddeployment_types.go b/api/v1alpha/workloaddeployment_types.go index 7da6bf45..c1ec37f0 100644 --- a/api/v1alpha/workloaddeployment_types.go +++ b/api/v1alpha/workloaddeployment_types.go @@ -71,6 +71,12 @@ type WorkloadDeploymentStatus struct { // // +kubebuilder:validation:Optional ObservedGeneration int64 `json:"observedGeneration,omitempty"` + + // Suspended, when true, requests that all instances managed by this deployment + // be stopped without releasing their placement, disk attachments, or quota allocation. + // + // +kubebuilder:validation:Optional + Suspended bool `json:"suspended,omitempty"` } const ( diff --git a/cmd/main.go b/cmd/main.go index b94d03e6..24c8bb74 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -576,6 +576,12 @@ func initializeClusterDiscovery( Teardowns: []consumerprovider.Teardown{ controller.NewComputeTeardown(quotaClientManager, federationClient, scheme), }, + Suspends: []consumerprovider.Suspend{ + controller.NewComputeSuspend(providerMgr.GetEventRecorder("compute-suspension")), + }, + Resumes: []consumerprovider.Resume{ + controller.NewComputeResume(providerMgr.GetEventRecorder("compute-suspension")), + }, }) if err != nil { return nil, nil, "", nil, fmt.Errorf("unable to create consumer provider: %w", err) diff --git a/config/base/crd/bases/compute.datumapis.com_instances.yaml b/config/base/crd/bases/compute.datumapis.com_instances.yaml index cfbc039c..cabe9d9f 100644 --- a/config/base/crd/bases/compute.datumapis.com_instances.yaml +++ b/config/base/crd/bases/compute.datumapis.com_instances.yaml @@ -1036,6 +1036,12 @@ spec: type: object type: object type: array + suspended: + description: |- + Suspended, when true, indicates that the instance's process should be stopped + without releasing its placement, disk attachments, or quota allocation. + The provider controller stops the running container/VM. + type: boolean type: object type: object served: true diff --git a/config/base/crd/bases/compute.datumapis.com_workloaddeployments.yaml b/config/base/crd/bases/compute.datumapis.com_workloaddeployments.yaml index 777d3324..f7f9fd22 100644 --- a/config/base/crd/bases/compute.datumapis.com_workloaddeployments.yaml +++ b/config/base/crd/bases/compute.datumapis.com_workloaddeployments.yaml @@ -1173,6 +1173,11 @@ spec: description: The number of instances created format: int32 type: integer + suspended: + description: |- + Suspended, when true, requests that all instances managed by this deployment + be stopped without releasing their placement, disk attachments, or quota allocation. + type: boolean updatedReplicas: description: |- The number of instances updated to the latest template revision, i.e. diff --git a/config/components/controller_rbac/role.yaml b/config/components/controller_rbac/role.yaml index e0c6beb5..d3f92436 100644 --- a/config/components/controller_rbac/role.yaml +++ b/config/components/controller_rbac/role.yaml @@ -96,12 +96,3 @@ rules: - get - list - watch -- apiGroups: - - services.miloapis.com - resources: - - serviceconsumers - - services - verbs: - - get - - list - - watch diff --git a/go.mod b/go.mod index 6bf99677..332997ca 100644 --- a/go.mod +++ b/go.mod @@ -11,7 +11,7 @@ require ( github.com/prometheus/client_golang v1.23.2 github.com/stretchr/testify v1.11.1 go.datum.net/network-services-operator v0.21.10-0.20260528021428-b0f2347f5359 - go.miloapis.com/milo v0.30.3 + go.miloapis.com/milo v0.32.0 golang.org/x/crypto v0.53.0 golang.org/x/sync v0.21.0 google.golang.org/protobuf v1.36.11 @@ -68,7 +68,7 @@ require ( github.com/spf13/pflag v1.0.9 // indirect github.com/stoewer/go-strcase v1.3.1 // indirect github.com/x448/float16 v0.8.4 // indirect - go.miloapis.com/service-catalog v0.3.2-0.20260716164540-4e9a0970caaa + go.miloapis.com/service-catalog v0.4.0 go.opentelemetry.io/auto/sdk v1.2.1 // indirect go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.62.0 // indirect go.opentelemetry.io/otel v1.43.0 // indirect diff --git a/go.sum b/go.sum index 7a4b6f33..b29f6393 100644 --- a/go.sum +++ b/go.sum @@ -155,10 +155,10 @@ github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM= github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg= go.datum.net/network-services-operator v0.21.10-0.20260528021428-b0f2347f5359 h1:P3dePA6cCXKimZzE6d7Xxpj2rz54BxOHI8K8ic7VQ+c= go.datum.net/network-services-operator v0.21.10-0.20260528021428-b0f2347f5359/go.mod h1:Nr0PsCodkTW31vWVxR9dhAP9w0y+WHUYeyrcRnchcIE= -go.miloapis.com/milo v0.30.3 h1:gnk3Vsn6ahgfO9zIXlZksqJOqOydbJuXpSqvdFBw7Lo= -go.miloapis.com/milo v0.30.3/go.mod h1:GKK3afjCwshfZfvhjNe1wp/H45z4m7x5oG/8xbSgU1M= -go.miloapis.com/service-catalog v0.3.2-0.20260716164540-4e9a0970caaa h1:Sn+/EvOO3GW9LReL7ksV0Qw/OWAEAxXIQyekcK/B8Yw= -go.miloapis.com/service-catalog v0.3.2-0.20260716164540-4e9a0970caaa/go.mod h1:Kkg6Tfpbc3wQaZizfd8sp3ap3uv9WYXEoZW/R4RBy3M= +go.miloapis.com/milo v0.32.0 h1:TkNIQu/37d+SEquLJ5+GmdisSl+K2RT7eEC4idg6RIs= +go.miloapis.com/milo v0.32.0/go.mod h1:GKK3afjCwshfZfvhjNe1wp/H45z4m7x5oG/8xbSgU1M= +go.miloapis.com/service-catalog v0.4.0 h1:LvO1WCHMCoFokpS5igWMP8kyqly9gUFQmQj5IGhwuKs= +go.miloapis.com/service-catalog v0.4.0/go.mod h1:1OfIYkdWH0lpbUH1d0Dc4A1yCtyNpcaKh50H6faG45A= go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.62.0 h1:Hf9xI/XLML9ElpiHVDNwvqI0hIFlzV8dgIr35kV1kRU= diff --git a/internal/controller/instance_controller.go b/internal/controller/instance_controller.go index fd497152..2bc40bf2 100644 --- a/internal/controller/instance_controller.go +++ b/internal/controller/instance_controller.go @@ -289,6 +289,14 @@ func (r *InstanceReconciler) Reconcile(ctx context.Context, req mcreconcile.Requ return ctrl.Result{}, r.reconcileDeletion(ctx, cl.GetClient(), req.ClusterName, &instance) } + // Honor project suspension: stop all normal provisioning and mark the + // instance unavailable. Placement, quota claim, and disk attachments are + // intentionally left intact so the instance can restart from disk when + // Status.Suspended is cleared on reinstatement. + if instance.Status.Suspended { + return ctrl.Result{}, r.reconcileSuspendedState(ctx, cl.GetClient(), &instance) + } + if !controllerutil.ContainsFinalizer(&instance, instanceQuotaFinalizer) { controllerutil.AddFinalizer(&instance, instanceQuotaFinalizer) if err := cl.GetClient().Update(ctx, &instance); err != nil { @@ -947,6 +955,43 @@ func (r *InstanceReconciler) emitEvent(obj *computev1alpha.Instance, eventType, r.recorder.Eventf(obj, nil, eventType, reason, action, "%s", message) } +// reconcileSuspendedState is called when instance.Status.Suspended is true. It +// sets Ready=False/Suspended and Available=False/Suspended without touching the +// instance's placement, quota claim, or disk attachments. Scheduling gates are +// left in place; quota is not re-evaluated. The only work done here is a status +// update (if anything changed) and a write-back to the upstream hub so the +// management plane reflects the suspended state. +func (r *InstanceReconciler) reconcileSuspendedState( + ctx context.Context, + cl client.Client, + instance *computev1alpha.Instance, +) error { + changed := apimeta.SetStatusCondition(&instance.Status.Conditions, metav1.Condition{ + Type: computev1alpha.InstanceReady, + Status: metav1.ConditionFalse, + Reason: computev1alpha.InstanceReadyReasonSuspended, + Message: "Instance is suspended due to project suspension.", + ObservedGeneration: instance.Generation, + }) + changed = apimeta.SetStatusCondition(&instance.Status.Conditions, metav1.Condition{ + Type: computev1alpha.InstanceAvailable, + Status: metav1.ConditionFalse, + Reason: computev1alpha.InstanceAvailableReasonSuspended, + Message: "Instance is suspended and not serving traffic.", + ObservedGeneration: instance.Generation, + }) || changed + + if changed { + if err := cl.Status().Update(ctx, instance); err != nil { + return fmt.Errorf("updating instance status for suspension: %w", err) + } + } + + // Write suspended status back to the federation hub so the management + // plane aggregates the correct per-instance state. + return r.writeBackToUpstream(ctx, instance) +} + // reconcileDeletion handles quota-claim cleanup when an Instance is being // deleted. It removes the quota finalizer once the ResourceClaim is gone. func (r *InstanceReconciler) reconcileDeletion(ctx context.Context, cl client.Client, clusterName multicluster.ClusterName, instance *computev1alpha.Instance) error { diff --git a/internal/controller/suspend_hooks.go b/internal/controller/suspend_hooks.go new file mode 100644 index 00000000..506aef32 --- /dev/null +++ b/internal/controller/suspend_hooks.go @@ -0,0 +1,154 @@ +// SPDX-License-Identifier: AGPL-3.0-only + +package controller + +import ( + "context" + "fmt" + + corev1 "k8s.io/api/core/v1" + "k8s.io/client-go/tools/events" + "sigs.k8s.io/controller-runtime/pkg/client" + + computev1alpha "go.datum.net/compute/api/v1alpha" +) + +const ( + // eventReasonProjectPaused is emitted on each Instance after it is + // successfully suspended due to a project suspension signal. + eventReasonProjectPaused = "ProjectPaused" + + // eventReasonProjectResumed is emitted on each Instance after it is + // successfully reinstated following a project reinstatement signal. + eventReasonProjectResumed = "ProjectResumed" + + // eventReasonPauseFailed is emitted on an Instance when the controller + // fails to suspend or resume it. The error is still returned so the + // provider's backoff retry fires; no teardown or deletion is attempted. + eventReasonPauseFailed = "PauseFailed" + + eventActionSuspending = "Suspending" + eventActionResuming = "Resuming" +) + +// ComputeSuspend implements consumer.Suspend. It sets spec.suspended=true on +// every Instance owned by the suspended consumer project, scoped by the +// services.miloapis.com/service-name label. It is idempotent: instances that +// are already suspended are skipped. +// +// On success it emits a ProjectPaused event on each affected Instance. +// On any patch failure it emits a PauseFailed Warning event on that Instance +// and returns the error immediately so the service-catalog provider retries +// with backoff. No teardown or deletion is ever attempted. +type ComputeSuspend struct { + recorder events.EventRecorder +} + +// NewComputeSuspend creates a ComputeSuspend. recorder is used to emit +// ProjectPaused / PauseFailed events on affected Instance objects. +func NewComputeSuspend(recorder events.EventRecorder) *ComputeSuspend { + return &ComputeSuspend{recorder: recorder} +} + +// SuspendConsumer implements consumer.Suspend. +func (cs *ComputeSuspend) SuspendConsumer( + ctx context.Context, + consumerProject string, + consumerClient client.Client, + serviceNames []string, +) error { + for _, svcName := range serviceNames { + // Suspend all WorkloadDeployments + var wds computev1alpha.WorkloadDeploymentList + if err := consumerClient.List(ctx, &wds, + client.MatchingLabels{labelServiceName: svcName}, + ); err != nil { + return fmt.Errorf("listing workloaddeployments for service %q: %w", svcName, err) + } + for i := range wds.Items { + wd := &wds.Items[i] + if wd.Status.Suspended { + continue // already suspended — idempotent + } + base := wd.DeepCopy() + wd.Status.Suspended = true + if err := consumerClient.Status().Patch(ctx, wd, client.MergeFrom(base)); err != nil { + if cs.recorder != nil { + cs.recorder.Eventf(wd, nil, corev1.EventTypeWarning, + eventReasonPauseFailed, eventActionSuspending, + "Failed to suspend workloaddeployment %s/%s for project %s: %v", + wd.Namespace, wd.Name, consumerProject, err) + } + return fmt.Errorf("patching workloaddeployment %s/%s status to suspended: %w", + wd.Namespace, wd.Name, err) + } + if cs.recorder != nil { + cs.recorder.Eventf(wd, nil, corev1.EventTypeNormal, + eventReasonProjectPaused, eventActionSuspending, + "WorkloadDeployment suspended due to project suspension of %s", consumerProject) + } + } + } + return nil +} + +// ComputeResume implements consumer.Resume. It clears spec.suspended on every +// Instance owned by the reinstated consumer project, scoped by the +// services.miloapis.com/service-name label. It is idempotent: instances that +// are already active are skipped. +// +// On success it emits a ProjectResumed event on each affected Instance. +// On any patch failure it emits a PauseFailed Warning event on that Instance +// and returns the error immediately so the service-catalog provider retries +// with backoff. No teardown or deletion is ever attempted. +type ComputeResume struct { + recorder events.EventRecorder +} + +// NewComputeResume creates a ComputeResume. recorder is used to emit +// ProjectResumed / PauseFailed events on affected Instance objects. +func NewComputeResume(recorder events.EventRecorder) *ComputeResume { + return &ComputeResume{recorder: recorder} +} + +// ResumeConsumer implements consumer.Resume. +func (cr *ComputeResume) ResumeConsumer( + ctx context.Context, + consumerProject string, + consumerClient client.Client, + serviceNames []string, +) error { + for _, svcName := range serviceNames { + // Resume all WorkloadDeployments + var wds computev1alpha.WorkloadDeploymentList + if err := consumerClient.List(ctx, &wds, + client.MatchingLabels{labelServiceName: svcName}, + ); err != nil { + return fmt.Errorf("listing workloaddeployments for service %q: %w", svcName, err) + } + for i := range wds.Items { + wd := &wds.Items[i] + if !wd.Status.Suspended { + continue // already active — idempotent + } + base := wd.DeepCopy() + wd.Status.Suspended = false + if err := consumerClient.Status().Patch(ctx, wd, client.MergeFrom(base)); err != nil { + if cr.recorder != nil { + cr.recorder.Eventf(wd, nil, corev1.EventTypeWarning, + eventReasonPauseFailed, eventActionResuming, + "Failed to resume workloaddeployment %s/%s for project %s: %v", + wd.Namespace, wd.Name, consumerProject, err) + } + return fmt.Errorf("patching workloaddeployment %s/%s status to resumed: %w", + wd.Namespace, wd.Name, err) + } + if cr.recorder != nil { + cr.recorder.Eventf(wd, nil, corev1.EventTypeNormal, + eventReasonProjectResumed, eventActionResuming, + "WorkloadDeployment resumed after project reinstatement of %s", consumerProject) + } + } + } + return nil +} diff --git a/internal/controller/suspend_hooks_test.go b/internal/controller/suspend_hooks_test.go new file mode 100644 index 00000000..bad4c9ad --- /dev/null +++ b/internal/controller/suspend_hooks_test.go @@ -0,0 +1,104 @@ +// SPDX-License-Identifier: AGPL-3.0-only + +package controller_test + +import ( + "testing" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + clientgoscheme "k8s.io/client-go/kubernetes/scheme" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + computev1alpha "go.datum.net/compute/api/v1alpha" + "go.datum.net/compute/internal/controller" + networkingv1alpha "go.datum.net/network-services-operator/api/v1alpha" + servicesv1alpha1 "go.miloapis.com/service-catalog/api/v1alpha1" + consumerprovider "go.miloapis.com/service-catalog/pkg/multicluster-runtime/consumer" +) + +func TestConformanceSuspendResume(t *testing.T) { + scheme := runtime.NewScheme() + if err := clientgoscheme.AddToScheme(scheme); err != nil { + t.Fatalf("adding clientgo scheme: %v", err) + } + if err := computev1alpha.AddToScheme(scheme); err != nil { + t.Fatalf("adding computev1alpha scheme: %v", err) + } + if err := networkingv1alpha.AddToScheme(scheme); err != nil { + t.Fatalf("adding networkingv1alpha scheme: %v", err) + } + if err := servicesv1alpha1.AddToScheme(scheme); err != nil { + t.Fatalf("adding servicesv1alpha1 scheme: %v", err) + } + + const ( + serviceName = "compute.miloapis.com" + consumerName = "sc-testconsumer" + consumerProject = "test-project" + instanceNS = "default" + instanceName = "test-instance" + ) + + // A pre-existing WorkloadDeployment in the consumer project, labelled so the hooks + // can find it. + deployment := &computev1alpha.WorkloadDeployment{ + ObjectMeta: metav1.ObjectMeta{ + Name: instanceName, // Keep name for simplicity + Namespace: instanceNS, + Labels: map[string]string{ + "services.miloapis.com/service-name": serviceName, + }, + }, + Spec: computev1alpha.WorkloadDeploymentSpec{ + Template: computev1alpha.InstanceTemplateSpec{ + Spec: computev1alpha.InstanceSpec{ + Runtime: computev1alpha.InstanceRuntimeSpec{ + Resources: computev1alpha.InstanceRuntimeResources{}, + }, + NetworkInterfaces: []computev1alpha.InstanceNetworkInterface{}, + }, + }, + }, + } + + // ServiceConsumer in Active phase on the provider side. + sc := &servicesv1alpha1.ServiceConsumer{ + ObjectMeta: metav1.ObjectMeta{ + Name: consumerName, + }, + Spec: servicesv1alpha1.ServiceConsumerSpec{ + ConsumerProjectRef: servicesv1alpha1.ConsumerProjectRef{ + Name: consumerProject, + }, + }, + Status: servicesv1alpha1.ServiceConsumerStatus{ + Phase: servicesv1alpha1.ConsumerPhaseActive, + }, + } + + consumerClient := fake.NewClientBuilder(). + WithScheme(scheme). + WithObjects(deployment). + WithStatusSubresource(deployment). + Build() + + providerClient := fake.NewClientBuilder(). + WithScheme(scheme). + WithObjects(sc). + WithStatusSubresource(sc). + Build() + + consumerprovider.ConformanceSuspendResume( + t, + providerClient, + consumerClient, + consumerName, + consumerProject, + []string{serviceName}, + controller.NewComputeSuspend(nil), + controller.NewComputeResume(nil), + // No retained objects: the WorkloadDeployment itself is what the hooks operate on + // (status.suspended flips), so it is not a bystander that must stay unchanged. + ) +} diff --git a/internal/controller/workloaddeployment_controller.go b/internal/controller/workloaddeployment_controller.go index 5e3de448..059ce361 100644 --- a/internal/controller/workloaddeployment_controller.go +++ b/internal/controller/workloaddeployment_controller.go @@ -256,6 +256,15 @@ func (r *WorkloadDeploymentReconciler) reconcileInstanceGates( ) (currentReplicas, updatedReplicas, readyReplicas, quotaBlockedReplicas, referencedDataBlockedReplicas int, err error) { templateHash := instancecontrol.ComputeHash(deployment.Spec.Template) for _, instance := range instances { + // Propagate suspension state from deployment to instance. + if instance.Status.Suspended != deployment.Status.Suspended { + base := instance.DeepCopy() + instance.Status.Suspended = deployment.Status.Suspended + if err := c.Status().Patch(ctx, &instance, client.MergeFrom(base)); err != nil { + return 0, 0, 0, 0, 0, fmt.Errorf("failed propagating suspension state to instance %s: %w", instance.Name, err) + } + } + if apimeta.IsStatusConditionPresentAndEqual(instance.Status.Conditions, computev1alpha.InstanceQuotaGranted, metav1.ConditionFalse) { quotaBlockedReplicas++ }