diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml index 87189dd15..0c8559494 100644 --- a/.github/workflows/ci.yaml +++ b/.github/workflows/ci.yaml @@ -1835,7 +1835,7 @@ jobs: # (operator does zero logins and is wait-dominated, so it runs # wide; the api buckets stay within the login limiter's budget). # The arm64 job below uses the same steps and per-bucket settings. - bucket: [operator, api-auth, api-roles, api-rbac, api-agent, api-mods, telemetry, optional-components, restricted-workloads, network-isolation, management-placement, backup-preparation, monitoring] + bucket: [operator, api-auth, api-roles, api-rbac, api-agent, api-mods, telemetry, optional-components, restricted-workloads, wake-hostport, network-isolation, management-placement, backup-preparation, monitoring] arch: [amd64] include: - bucket: monitoring @@ -1850,10 +1850,14 @@ jobs: parallel: 8 test_timeout: 20m job_timeout: 60 # Boots a kind cluster, installs the chart, runs the bucket's suite. + - bucket: wake-hostport + parallel: 1 + test_timeout: 15m + job_timeout: 60 # Boots two-node kind, installs the chart and exercises repeated Hostport sleep/wake handoffs. - bucket: management-placement parallel: 1 test_timeout: 15m - job_timeout: 60 + job_timeout: 60 # Boots management kind, installs the chart and checks placement/disruption behavior. - bucket: backup-preparation parallel: 1 test_timeout: 15m @@ -1920,7 +1924,7 @@ jobs: if: env.RUN_REGEX != '' || env.TAIL_REGEX != '' env: GAMEPLANE_E2E_CNI: ${{ (matrix.bucket == 'network-isolation' || matrix.bucket == 'monitoring') && 'cilium' || 'kindnet' }} - GAMEPLANE_E2E_PROFILE: ${{ (matrix.bucket == 'management-placement' || matrix.bucket == 'backup-preparation') && 'management' || 'default' }} + GAMEPLANE_E2E_PROFILE: ${{ matrix.bucket == 'wake-hostport' && 'two-node' || (matrix.bucket == 'management-placement' || matrix.bucket == 'backup-preparation') && 'management' || 'default' }} run: ./deploy/kind/e2e.sh up "$CLUSTER" e2e "$GAMEPLANE_E2E_PROFILE" - name: install real monitoring fixture if: matrix.bucket == 'monitoring' && env.RUN_REGEX != '' @@ -2751,7 +2755,7 @@ jobs: 'go-e2e-unit': (n) => n === 'go (test/e2e)', 'api-postgres': (n) => n === 'api (postgres)', 'e2e-buckets': (n) => n === 'e2e bucket coverage', - 'e2e-go': (n) => /^e2e (operator|api-auth|api-roles|api-rbac|api-agent|api-mods|telemetry|optional-components|restricted-workloads|network-isolation|management-placement|backup-preparation|monitoring) \//.test(n) && n.includes('/ amd64 '), + 'e2e-go': (n) => /^e2e (operator|api-auth|api-roles|api-rbac|api-agent|api-mods|telemetry|optional-components|restricted-workloads|wake-hostport|network-isolation|management-placement|backup-preparation|monitoring) \//.test(n) && n.includes('/ amd64 '), 'e2e-go-arm64': (n) => /^e2e (operator|api-auth|api-roles|api-rbac|api-agent|api-mods|telemetry|optional-components|restricted-workloads) \//.test(n) && n.includes('/ arm64 '), 'e2e-multicluster': (n) => n.startsWith('e2e multicluster /') && n.includes('/ amd64 '), 'e2e-multicluster-arm64': (n) => n.startsWith('e2e multicluster /') && n.includes('/ arm64 '), @@ -2904,7 +2908,7 @@ jobs: const bucketSet = new Set(); for (const j of apiJobs) { const n = j.name || ''; - const m = n.match(/^e2e (operator|api-auth|api-roles|api-rbac|api-agent|api-mods|telemetry|optional-components|restricted-workloads|network-isolation|management-placement|backup-preparation|monitoring) \//); + const m = n.match(/^e2e (operator|api-auth|api-roles|api-rbac|api-agent|api-mods|telemetry|optional-components|restricted-workloads|wake-hostport|network-isolation|management-placement|backup-preparation|monitoring) \//); if (m) bucketSet.add(m[1]); if (/^e2e multicluster \//.test(n)) bucketSet.add('multicluster'); if (/^e2e upgrade \//.test(n)) bucketSet.add('upgrade'); diff --git a/charts/gameplane/templates/operator.yaml b/charts/gameplane/templates/operator.yaml index 75ccb20fa..874c181e0 100644 --- a/charts/gameplane/templates/operator.yaml +++ b/charts/gameplane/templates/operator.yaml @@ -17,6 +17,10 @@ metadata: name: {{ .Release.Name }}-operator-watch labels: {{- include "gameplane.labels" . | nindent 4 }} rules: + # Direct node identity/readiness checks for durable Hostport wake endpoints. + - apiGroups: [""] + resources: [nodes] + verbs: [get] - apiGroups: ["gameplane.local"] resources: [gameservers, gametemplates, backups, backupschedules, restores, modules, modulesources, networkcaptures] verbs: [get, list, watch] diff --git a/deploy/kind/e2e.sh b/deploy/kind/e2e.sh index 3cf78410c..0830fbf19 100755 --- a/deploy/kind/e2e.sh +++ b/deploy/kind/e2e.sh @@ -6,7 +6,8 @@ # deploy/kind/e2e.sh down [cluster-name] # # Diffs from dev-up (deploy/kind/up.sh): -# - Single-node by default; management profile adds two scheduling workers. +# - Single-node by default; two-node adds one worker for Hostport handoff, +# and management adds two workers for scheduling and disruption checks. # - Skips ingress-nginx (the dashboard isn't exercised here). # - Loads pre-built gameplane-test/{operator,api,agent,sentinel,capture-sidecar,telemetry-receiver,fakeoidc}: # images (and gameplane-test/gameprobe: when present). @@ -33,9 +34,9 @@ kindnet | cilium) ;; *) echo "unsupported E2E CNI: ${CNI}" >&2; exit 2 ;; esac case "${PROFILE}" in -default | postgres | management) ;; +default | postgres | two-node | management) ;; *) - echo "unknown profile '${PROFILE}' (want default, postgres or management)" >&2 + echo "unknown profile '${PROFILE}' (want default, postgres, two-node or management)" >&2 exit 2 ;; esac @@ -201,7 +202,9 @@ up) kind_wait=0s # Nodes become Ready after Cilium is installed below. fi extra_nodes="" - if [[ "${PROFILE}" == "management" ]]; then + if [[ "${PROFILE}" == "two-node" ]]; then + extra_nodes=" - role: worker" + elif [[ "${PROFILE}" == "management" ]]; then extra_nodes=$' - role: worker\n - role: worker' fi kind create cluster --name "${CLUSTER}" --wait "${kind_wait}" --config - < 0 + case intstr.String: + if len(validation.IsValidPercent(value.StrVal)) != 0 { + return fence, false + } + percent, err := strconv.Atoi(strings.TrimSuffix(value.StrVal, "%")) + return fence, err == nil && percent > 0 && percent <= 100 + default: + return fence, false + } + default: + return fence, false + } +} + +func clearHostportStopFence(ss *appsv1.StatefulSet) { + raw := ss.Annotations[hostportStopFenceAnnotation] + if fence, valid := decodeHostportStopFence(raw); valid { + ss.Spec.UpdateStrategy = fence.Strategy + } + // Unknown saved state must not replace the live strategy or obstruct + // required scale-down. Keep that strategy and retire only our own fence. + ss.Spec.Template.Spec.SchedulingGates = slices.DeleteFunc(ss.Spec.Template.Spec.SchedulingGates, func(gate corev1.PodSchedulingGate) bool { return gate.Name == hostportStopSchedulingGate }) + delete(ss.Annotations, hostportStopFenceAnnotation) +} diff --git a/operator/internal/controller/gameserver_hostport_stop_test.go b/operator/internal/controller/gameserver_hostport_stop_test.go new file mode 100644 index 000000000..3f066146c --- /dev/null +++ b/operator/internal/controller/gameserver_hostport_stop_test.go @@ -0,0 +1,412 @@ +package controller + +import ( + "encoding/json" + "fmt" + "reflect" + "slices" + "testing" + "time" + + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/util/intstr" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + gameplanev1alpha1 "github.com/GameplanePanel/gameplane/operator/api/v1alpha1" +) + +func hostportStopFixture(t *testing.T) (*GameServerReconciler, *gameplanev1alpha1.GameServer, *gameplanev1alpha1.GameTemplate, *lifecycleStopper) { + t.Helper() + node := hostportIdentityNode("original-node", corev1.ConditionTrue) + now := metav1.Now() + node.DeletionTimestamp, node.Finalizers = &now, []string{"test/hold"} + r, gs, tmpl := hostportIdentityFixture(t, node, corev1.PodRunning, nil) + grace := int32(60) + gs.Spec.StopGracePeriodSeconds = &grace + if err := r.Update(t.Context(), gs); err != nil { + t.Fatal(err) + } + tmpl.Spec.RCON = &gameplanev1alpha1.RCONSpec{Protocol: "source"} + tmpl.Spec.Capabilities = &gameplanev1alpha1.CapabilitiesSpec{Lifecycle: &gameplanev1alpha1.LifecycleSpec{Stop: []string{"save", "stop"}}} + var ss appsv1.StatefulSet + if err := r.Get(t.Context(), client.ObjectKeyFromObject(gs), &ss); err != nil { + t.Fatal(err) + } + ss.Generation = 1 + ss.Spec.UpdateStrategy = appsv1.StatefulSetUpdateStrategy{Type: appsv1.RollingUpdateStatefulSetStrategyType} + if err := r.Update(t.Context(), &ss); err != nil { + t.Fatal(err) + } + ss.Status.ReadyReplicas = 1 + if err := r.Status().Update(t.Context(), &ss); err != nil { + t.Fatal(err) + } + var pod corev1.Pod + if err := r.Get(t.Context(), client.ObjectKey{Namespace: gs.Namespace, Name: gs.Name + "-0"}, &pod); err != nil { + t.Fatal(err) + } + pod.UID = "original-pod" + pod.OwnerReferences = []metav1.OwnerReference{*metav1.NewControllerRef(&ss, appsv1.SchemeGroupVersion.WithKind("StatefulSet"))} + if err := r.Update(t.Context(), &pod); err != nil { + t.Fatal(err) + } + stopper := &lifecycleStopper{} + r.AgentClient = stopper + return r, gs, tmpl, stopper +} + +func hostportStopSet(t *testing.T, r *GameServerReconciler, gs *gameplanev1alpha1.GameServer) *appsv1.StatefulSet { + t.Helper() + var ss appsv1.StatefulSet + if err := r.Get(t.Context(), client.ObjectKeyFromObject(gs), &ss); err != nil { + t.Fatal(err) + } + return &ss +} + +func TestHostportDeletingOriginalNodeFencesReplacementsBeforeGracefulStop(t *testing.T) { + r, gs, tmpl, stopper := hostportStopFixture(t) + original := hostportStopSet(t, r, gs).Spec.Template.DeepCopy() + for pass := range 3 { + if err := r.Get(t.Context(), client.ObjectKeyFromObject(gs), gs); err != nil { + t.Fatal(err) + } + // replicas=0 models an already-blocked plan. The same mutation gate + // must also protect a stale healthy plan passing replicas=1. + replicas := int32(0) + if pass == 0 { + replicas = 1 + } + if err := r.reconcileStatefulSet(t.Context(), gs, tmpl, nil, &materializedConfig{}, replicas); err != nil { + t.Fatal(err) + } + ss := hostportStopSet(t, r, gs) + if *ss.Spec.Replicas != 1 || ss.Spec.UpdateStrategy.Type != appsv1.OnDeleteStatefulSetStrategyType || !slices.ContainsFunc(ss.Spec.Template.Spec.SchedulingGates, func(g corev1.PodSchedulingGate) bool { return g.Name == hostportStopSchedulingGate }) { + t.Fatal("graceful stop lost original Pod or replacement scheduling fence") + } + withoutGate := ss.Spec.Template.DeepCopy() + withoutGate.Spec.SchedulingGates = nil + if !reflect.DeepEqual(original, withoutGate) { + t.Fatal("endpoint drain changed the running game's template") + } + wantCalls := 1 + if pass == 0 { + wantCalls = 0 + } + if stopper.calls != wantCalls { + t.Fatalf("pass %d: stop calls=%d, want=%d", pass, stopper.calls, wantCalls) + } + if pass == 0 { + ss.Status.ObservedGeneration = ss.Generation + if err := r.Status().Update(t.Context(), ss); err != nil { + t.Fatal(err) + } + } + // Recreate the reconciler: durable gate and grace timestamp must be + // sufficient to continue without another stop command. + r = &GameServerReconciler{Client: r.Client, APIReader: r.APIReader, Scheme: r.Scheme, AgentClient: stopper} + } + if err := r.Get(t.Context(), client.ObjectKeyFromObject(gs), gs); err != nil { + t.Fatal(err) + } + gs.Annotations[stopRequestedAtAnnotation] = time.Now().Add(-2 * time.Minute).UTC().Format(time.RFC3339) + if err := r.Update(t.Context(), gs); err != nil { + t.Fatal(err) + } + if err := r.reconcileStatefulSet(t.Context(), gs, tmpl, nil, &materializedConfig{}, 0); err != nil { + t.Fatal(err) + } + if *hostportStopSet(t, r, gs).Spec.Replicas != 0 { + t.Fatal("grace deadline did not stop original workload") + } +} + +func TestHostportStopFenceRejectsReplacementPodAndRestoresAfterMigration(t *testing.T) { + r, gs, tmpl, _ := hostportStopFixture(t) + originalSet := hostportStopSet(t, r, gs) + originalSet.Spec.Template.Spec.SchedulingGates = []corev1.PodSchedulingGate{{Name: "example.invalid/existing-gate"}} + if err := r.Update(t.Context(), originalSet); err != nil { + t.Fatal(err) + } + if err := r.reconcileStatefulSet(t.Context(), gs, tmpl, nil, &materializedConfig{}, 1); err != nil { + t.Fatal(err) + } + var pod corev1.Pod + if err := r.Get(t.Context(), client.ObjectKey{Namespace: gs.Namespace, Name: gs.Name + "-0"}, &pod); err != nil { + t.Fatal(err) + } + pod.UID = "replacement-pod" + if err := r.Update(t.Context(), &pod); err != nil { + t.Fatal(err) + } + if err := r.reconcileStatefulSet(t.Context(), gs, tmpl, nil, &materializedConfig{}, 1); err != nil { + t.Fatal(err) + } + if *hostportStopSet(t, r, gs).Spec.Replicas != 0 { + t.Fatal("replacement Pod inherited original Pod's save window") + } + // Explicit endpoint migration follows the documented stopped/drained + // procedure. Ordinary rendering removes only our temporary scheduling gate. + if err := r.Delete(t.Context(), &pod); err != nil { + t.Fatal(err) + } + delete(gs.Annotations, hostportNodeAnnotation) + delete(gs.Annotations, hostportNodeUIDAnnotation) + if err := r.Update(t.Context(), gs); err != nil { + t.Fatal(err) + } + if err := r.reconcileStatefulSet(t.Context(), gs, tmpl, nil, &materializedConfig{}, 0); err != nil { + t.Fatal(err) + } + ss := hostportStopSet(t, r, gs) + if ss.Annotations[hostportStopFenceAnnotation] != "" || ss.Spec.UpdateStrategy.Type != appsv1.RollingUpdateStatefulSetStrategyType || !reflect.DeepEqual(ss.Spec.Template.Spec.SchedulingGates, originalSet.Spec.Template.Spec.SchedulingGates) { + t.Fatal("migration retained temporary drain fence") + } +} + +func TestHostportStopWaitsForFenceObservation(t *testing.T) { + r, gs, tmpl, stopper := hostportStopFixture(t) + ss := hostportStopSet(t, r, gs) + ss.Generation = 9 + if err := r.Update(t.Context(), ss); err != nil { + t.Fatal(err) + } + ss.Status.ObservedGeneration = 8 + if err := r.Status().Update(t.Context(), ss); err != nil { + t.Fatal(err) + } + for range 2 { + if err := r.reconcileStatefulSet(t.Context(), gs, tmpl, nil, &materializedConfig{}, 1); err != nil { + t.Fatal(err) + } + } + if stopper.calls != 0 || *hostportStopSet(t, r, gs).Spec.Replicas != 1 { + t.Fatal("stop started before replacement fence was observed") + } + ss = hostportStopSet(t, r, gs) + ss.Status.ObservedGeneration = ss.Generation + if err := r.Status().Update(t.Context(), ss); err != nil { + t.Fatal(err) + } + if err := r.reconcileStatefulSet(t.Context(), gs, tmpl, nil, &materializedConfig{}, 1); err != nil { + t.Fatal(err) + } + if stopper.calls != 1 { + t.Fatal("observed replacement fence did not permit graceful stop") + } +} + +func TestHostportValidEndpointRestoresFencedStrategy(t *testing.T) { + r, gs, tmpl, _ := hostportStopFixture(t) + if err := r.reconcileStatefulSet(t.Context(), gs, tmpl, nil, &materializedConfig{}, 1); err != nil { + t.Fatal(err) + } + // A fresh healthy endpoint read governs cleanup, independently of the + // previous cached deleting-node plan. + r.APIReader = fake.NewClientBuilder().WithScheme(r.Scheme).WithObjects(gs, hostportIdentityNode("original-node", corev1.ConditionTrue)).Build() + if err := r.reconcileStatefulSet(t.Context(), gs, tmpl, nil, &materializedConfig{}, 1); err != nil { + t.Fatal(err) + } + ss := hostportStopSet(t, r, gs) + if ss.Annotations[hostportStopFenceAnnotation] != "" || ss.Spec.UpdateStrategy.Type != appsv1.RollingUpdateStatefulSetStrategyType || slices.ContainsFunc(ss.Spec.Template.Spec.SchedulingGates, func(g corev1.PodSchedulingGate) bool { return g.Name == hostportStopSchedulingGate }) { + t.Fatal("valid endpoint retained temporary stop fence") + } +} + +func TestHostportDeletingNodeDoesNotGrantGraceToUntrustedPod(t *testing.T) { + for _, mode := range []string{"pending", "unowned", "other-node"} { + t.Run(mode, func(t *testing.T) { + r, gs, tmpl, stopper := hostportStopFixture(t) + var pod corev1.Pod + if err := r.Get(t.Context(), client.ObjectKey{Namespace: gs.Namespace, Name: gs.Name + "-0"}, &pod); err != nil { + t.Fatal(err) + } + switch mode { + case "pending": + pod.Status.Phase = corev1.PodPending + case "unowned": + pod.OwnerReferences = nil + case "other-node": + pod.Spec.NodeName = "node-b" + } + if err := r.Update(t.Context(), &pod); err != nil { + t.Fatal(err) + } + if mode == "pending" { + pod.Status.Phase = corev1.PodPending + if err := r.Status().Update(t.Context(), &pod); err != nil { + t.Fatal(err) + } + } + if err := r.reconcileStatefulSet(t.Context(), gs, tmpl, nil, &materializedConfig{}, 1); err != nil { + t.Fatal(err) + } + if *hostportStopSet(t, r, gs).Spec.Replicas != 0 || stopper.calls != 0 { + t.Fatal("untrusted Pod bypassed endpoint identity fence") + } + }) + } +} + +func TestHostportDeletingNodeWaitsForCopyOwnerBeforeSaveStop(t *testing.T) { + for _, recovery := range []bool{false, true} { + t.Run(map[bool]string{false: "backup", true: "recovery"}[recovery], func(t *testing.T) { + r, gs, tmpl, stopper := hostportStopFixture(t) + op := lifecycleOperation{lifecycleIdentity: lifecycleIdentity{Kind: "Backup", Name: "copy", UID: "copy-uid", TargetUID: gs.UID}} + if recovery { + op.Kind, op.Name, op.UID = "Wipe", gs.Name, gs.UID + op.Token = "wipe" + op.Recovery = []lifecycleIdentity{{Kind: "Backup", Name: "old-copy", UID: "old-copy-uid", TargetUID: gs.UID}} + } + gs.Annotations[lifecycleOperationAnnotation] = lifecycleOperationJSON(op) + if err := r.Update(t.Context(), gs); err != nil { + t.Fatal(err) + } + if err := r.reconcileStatefulSet(t.Context(), gs, tmpl, nil, &materializedConfig{}, 0); err != nil { + t.Fatal(err) + } + ss := hostportStopSet(t, r, gs) + ss.Status.ObservedGeneration = ss.Generation + if err := r.Status().Update(t.Context(), ss); err != nil { + t.Fatal(err) + } + if err := r.reconcileStatefulSet(t.Context(), gs, tmpl, nil, &materializedConfig{}, 0); err != nil { + t.Fatal(err) + } + if err := r.Get(t.Context(), client.ObjectKeyFromObject(gs), gs); err != nil { + t.Fatal(err) + } + if stopper.calls != 0 || gs.Annotations[stopRequestedAtAnnotation] != "" || *hostportStopSet(t, r, gs).Spec.Replicas != 1 { + t.Fatal("endpoint stop RPC overlapped copy/save-off ownership") + } + if hostportStopSet(t, r, gs).Annotations[lifecycleOperationAnnotation] != gs.Annotations[lifecycleOperationAnnotation] { + t.Fatal("fenced copy owner was not projected to the workload") + } + delete(gs.Annotations, lifecycleOperationAnnotation) + if err := r.Update(t.Context(), gs); err != nil { + t.Fatal(err) + } + if err := r.reconcileStatefulSet(t.Context(), gs, tmpl, nil, &materializedConfig{}, 0); err != nil { + t.Fatal(err) + } + if stopper.calls != 1 { + t.Fatal("copy cleanup did not permit graceful stop") + } + if _, held := hostportStopSet(t, r, gs).Annotations[lifecycleOperationAnnotation]; held { + t.Fatal("endpoint stop retained a released copy owner") + } + }) + } +} + +func TestHostportMalformedStopFenceCannotGrantGrace(t *testing.T) { + for _, raw := range []string{"malformed fence", "{}", "null"} { + t.Run(raw, func(t *testing.T) { + r, gs, tmpl, stopper := hostportStopFixture(t) + ss := hostportStopSet(t, r, gs) + ss.Annotations = map[string]string{hostportStopFenceAnnotation: raw} + ss.Spec.UpdateStrategy = appsv1.StatefulSetUpdateStrategy{Type: appsv1.OnDeleteStatefulSetStrategyType} + ss.Spec.Template.Spec.SchedulingGates = []corev1.PodSchedulingGate{{Name: hostportStopSchedulingGate}} + if err := r.Update(t.Context(), ss); err != nil { + t.Fatal(err) + } + ss.Status.ObservedGeneration = ss.Generation + if err := r.Status().Update(t.Context(), ss); err != nil { + t.Fatal(err) + } + if err := r.reconcileStatefulSet(t.Context(), gs, tmpl, nil, &materializedConfig{}, 1); err != nil { + t.Fatal(err) + } + if *hostportStopSet(t, r, gs).Spec.Replicas != 0 || stopper.calls != 0 { + t.Fatal("malformed or incomplete fence authorized grace or a stop RPC") + } + }) + } +} + +func TestHostportFenceCleanupRecoversInvalidMetadataWithoutGuessingStrategy(t *testing.T) { + partition, unavailable := int32(3), intstr.FromInt32(1) + saved := appsv1.StatefulSetUpdateStrategy{Type: appsv1.RollingUpdateStatefulSetStrategyType, RollingUpdate: &appsv1.RollingUpdateStatefulSetStrategy{Partition: &partition, MaxUnavailable: &unavailable}} + valid, err := json.Marshal(hostportStopFence{PodUID: "original-pod", NodeUID: "original-node", Strategy: saved}) + if err != nil { + t.Fatal(err) + } + recreate := appsv1.StatefulSetUpdateStrategy{Type: appsv1.RecreateStatefulSetStrategyType} + validRecreate, err := json.Marshal(hostportStopFence{PodUID: "original-pod", NodeUID: "original-node", Strategy: recreate}) + if err != nil { + t.Fatal(err) + } + percentage := saved.DeepCopy() + percent := intstr.FromString("25%") + percentage.RollingUpdate.MaxUnavailable = &percent + validPercentage, err := json.Marshal(hostportStopFence{PodUID: "original-pod", NodeUID: "original-node", Strategy: *percentage}) + if err != nil { + t.Fatal(err) + } + for _, test := range []struct { + name, raw string + restore *appsv1.StatefulSetUpdateStrategy + }{ + {"empty-annotation", "", nil}, + {"missing-annotation", "", nil}, + {"malformed", "malformed fence", nil}, + {"empty-object", "{}", nil}, + {"null", "null", nil}, + {"missing-identities", `{"strategy":{"type":"RollingUpdate"}}`, nil}, + {"empty-strategy", `{"podUID":"original-pod","nodeUID":"original-node","strategy":{}}`, nil}, + {"unknown-strategy", `{"podUID":"original-pod","nodeUID":"original-node","strategy":{"type":"unknown"}}`, nil}, + {"on-delete-with-rolling-fields", `{"podUID":"original-pod","nodeUID":"original-node","strategy":{"type":"OnDelete","rollingUpdate":{}}}`, nil}, + {"negative-partition", `{"podUID":"original-pod","nodeUID":"original-node","strategy":{"type":"RollingUpdate","rollingUpdate":{"partition":-1}}}`, nil}, + {"zero-max-unavailable", `{"podUID":"original-pod","nodeUID":"original-node","strategy":{"type":"RollingUpdate","rollingUpdate":{"maxUnavailable":0}}}`, nil}, + {"invalid-max-unavailable", `{"podUID":"original-pod","nodeUID":"original-node","strategy":{"type":"RollingUpdate","rollingUpdate":{"maxUnavailable":"invalid"}}}`, nil}, + {"excessive-max-unavailable-percent", `{"podUID":"original-pod","nodeUID":"original-node","strategy":{"type":"RollingUpdate","rollingUpdate":{"maxUnavailable":"101%"}}}`, nil}, + {"valid-custom-strategy", string(valid), &saved}, + {"valid-percentage-strategy", string(validPercentage), percentage}, + {"valid-recreate-strategy", string(validRecreate), &recreate}, + } { + for _, hostport := range []bool{false, true} { + for _, replicas := range []int32{0, 1} { + t.Run(fmt.Sprintf("%s/hostport=%t/replicas=%d", test.name, hostport, replicas), func(t *testing.T) { + r, gs, tmpl, _ := hostportStopFixture(t) + if !hostport { + gs.Spec.Networking.Expose = "NodePort" + if err := r.Update(t.Context(), gs); err != nil { + t.Fatal(err) + } + } + r.APIReader = fake.NewClientBuilder().WithScheme(r.Scheme).WithObjects(gs, hostportIdentityNode("original-node", corev1.ConditionTrue)).Build() + ss := hostportStopSet(t, r, gs) + ss.Annotations = map[string]string{"example.invalid/retained": "value"} + if test.name != "missing-annotation" { + ss.Annotations[hostportStopFenceAnnotation] = test.raw + } + current := appsv1.StatefulSetUpdateStrategy{Type: appsv1.OnDeleteStatefulSetStrategyType} + ss.Spec.UpdateStrategy = current + otherGate := corev1.PodSchedulingGate{Name: "example.invalid/retained"} + ss.Spec.Template.Spec.SchedulingGates = []corev1.PodSchedulingGate{otherGate, {Name: hostportStopSchedulingGate}} + if err := r.Update(t.Context(), ss); err != nil { + t.Fatal(err) + } + if err := r.reconcileStatefulSet(t.Context(), gs, tmpl, nil, &materializedConfig{}, replicas); err != nil { + t.Fatal(err) + } + ss = hostportStopSet(t, r, gs) + want := current + if test.restore != nil { + want = *test.restore + } + if !reflect.DeepEqual(ss.Spec.UpdateStrategy, want) || ss.Spec.Replicas == nil || *ss.Spec.Replicas != replicas { + t.Fatalf("cleanup blocked power intent or guessed strategy: %+v", ss.Spec) + } + _, fenceRemains := ss.Annotations[hostportStopFenceAnnotation] + if !reflect.DeepEqual(ss.Spec.Template.Spec.SchedulingGates, []corev1.PodSchedulingGate{otherGate}) || fenceRemains || ss.Annotations["example.invalid/retained"] != "value" { + t.Fatal("cleanup changed unrelated state or retained its fence") + } + }) + } + } + } +} diff --git a/operator/internal/controller/gameserver_hostport_test.go b/operator/internal/controller/gameserver_hostport_test.go new file mode 100644 index 000000000..fc19f5116 --- /dev/null +++ b/operator/internal/controller/gameserver_hostport_test.go @@ -0,0 +1,132 @@ +package controller + +import ( + "testing" + + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + gameplanev1alpha1 "github.com/GameplanePanel/gameplane/operator/api/v1alpha1" +) + +func hostportFixture(t *testing.T) (*GameServerReconciler, *gameplanev1alpha1.GameServer, *gameplanev1alpha1.GameTemplate) { + t.Helper() + s := wipeScheme(t) + if err := appsv1.AddToScheme(s); err != nil { + t.Fatal(err) + } + gs := &gameplanev1alpha1.GameServer{ObjectMeta: metav1.ObjectMeta{Name: "game", Namespace: "games", UID: "game-uid"}, Spec: gameplanev1alpha1.GameServerSpec{Networking: gameplanev1alpha1.GameServerNetworking{Expose: "Hostport"}, Idle: &gameplanev1alpha1.IdleSpec{Enabled: true, WakeOnConnect: true}}} + node := &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: "node-a", UID: "node-uid"}, Status: corev1.NodeStatus{Conditions: []corev1.NodeCondition{{Type: corev1.NodeReady, Status: corev1.ConditionTrue}}}} + zero := int32(0) + ss := &appsv1.StatefulSet{ObjectMeta: metav1.ObjectMeta{Name: gs.Name, Namespace: gs.Namespace}, Spec: appsv1.StatefulSetSpec{Replicas: &zero}} + c := fake.NewClientBuilder().WithScheme(s).WithObjects(gs, node, ss).Build() + return &GameServerReconciler{Client: c, APIReader: c, Scheme: s}, gs, singlePortTemplate(true, "generic") +} + +func TestHostportSleepCapturesNodeAndWaitsForActualPodDeletion(t *testing.T) { + r, gs, tmpl := hostportFixture(t) + pod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "game-0", Namespace: gs.Namespace, Labels: map[string]string{"app.kubernetes.io/name": "gameplane-game", "app.kubernetes.io/instance": gs.Name}, Finalizers: []string{"test/drain"}}, Spec: corev1.PodSpec{NodeName: "node-a"}} + if err := r.Create(t.Context(), pod); err != nil { + t.Fatal(err) + } + if err := r.Delete(t.Context(), pod); err != nil { + t.Fatal(err) + } + plan, err := r.planSentinel(t.Context(), gs, tmpl, idleAsleep) + if err != nil { + t.Fatal(err) + } + if plan.wantSentinel || plan.hostportCondition.Reason != "GameDraining" { + t.Fatalf("plan before actual deletion: %+v", plan) + } + var saved gameplanev1alpha1.GameServer + if err := r.Get(t.Context(), client.ObjectKeyFromObject(gs), &saved); err != nil { + t.Fatal(err) + } + if saved.Annotations[hostportNodeAnnotation] != "node-a" || saved.Annotations[hostportNodeUIDAnnotation] != "node-uid" { + t.Fatalf("durable endpoint not saved: %v", saved.Annotations) + } + if err := r.Get(t.Context(), client.ObjectKeyFromObject(pod), pod); err != nil { + t.Fatal(err) + } + pod.Finalizers = nil + if err := r.Update(t.Context(), pod); err != nil { + t.Fatal(err) + } + plan, err = r.planSentinel(t.Context(), &saved, tmpl, idleAsleep) + if err != nil || !plan.wantSentinel { + t.Fatalf("plan after actual deletion: %+v, %v", plan, err) + } +} + +func TestHostportWakeUsesUncachedPodsAndWaitsForProducerDeletion(t *testing.T) { + r, gs, tmpl := hostportFixture(t) + gs.Annotations = map[string]string{hostportNodeAnnotation: "node-a", hostportNodeUIDAnnotation: "node-uid"} + dep := &appsv1.Deployment{ObjectMeta: metav1.ObjectMeta{Name: gs.Name + "-waker", Namespace: gs.Namespace}} + if err := r.Create(t.Context(), dep); err != nil { + t.Fatal(err) + } + // A cached view with no Pods must not authorize transfer while the live + // API still has a terminating listener, even after its Deployment is gone. + live := fake.NewClientBuilder().WithScheme(r.Scheme).WithObjects(dep, &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "old-waker", Namespace: gs.Namespace, Labels: map[string]string{sentinelLabel: sentinelValue, "app.kubernetes.io/instance": gs.Name}}}, &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: "node-a", UID: "node-uid"}, Status: corev1.NodeStatus{Conditions: []corev1.NodeCondition{{Type: corev1.NodeReady, Status: corev1.ConditionTrue}}}}).Build() + r.APIReader = live + plan, err := r.planSentinel(t.Context(), gs, tmpl, idleAwake) + if err != nil || !plan.blockGame || plan.wantSentinel { + t.Fatalf("producer still live: %+v, %v", plan, err) + } + if err := live.Delete(t.Context(), dep); err != nil { + t.Fatal(err) + } + plan, err = r.planSentinel(t.Context(), gs, tmpl, idleAwake) + if err != nil || !plan.blockGame { + t.Fatalf("live Pod bypassed by cache: %+v, %v", plan, err) + } + if err := live.Delete(t.Context(), &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "old-waker", Namespace: gs.Namespace}}); err != nil { + t.Fatal(err) + } + plan, err = r.planSentinel(t.Context(), gs, tmpl, idleAwake) + if err != nil || plan.blockGame { + t.Fatalf("drained listener still blocks: %+v, %v", plan, err) + } +} + +func TestHostportNodeLossAndUnschedulableListenerStayUnavailable(t *testing.T) { + r, gs, tmpl := hostportFixture(t) + gs.Annotations = map[string]string{hostportNodeAnnotation: "node-a", hostportNodeUIDAnnotation: "node-uid"} + pod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "waker", Namespace: gs.Namespace, Labels: map[string]string{sentinelLabel: sentinelValue, "app.kubernetes.io/instance": gs.Name}}, Status: corev1.PodStatus{Conditions: []corev1.PodCondition{{Type: corev1.PodScheduled, Status: corev1.ConditionFalse, Reason: "Unschedulable", Message: "Insufficient memory"}}}} + if err := r.Create(t.Context(), pod); err != nil { + t.Fatal(err) + } + plan, err := r.planSentinel(t.Context(), gs, tmpl, idleAsleep) + if err != nil || plan.routeToSentinel || plan.hostportCondition.Reason != "EndpointUnschedulable" { + t.Fatalf("full node: %+v, %v", plan, err) + } + if err := r.Delete(t.Context(), &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: "node-a"}}); err != nil { + t.Fatal(err) + } + plan, err = r.planSentinel(t.Context(), gs, tmpl, idleAsleep) + if err != nil || plan.wantSentinel || plan.hostportCondition.Reason != "EndpointNodeUnavailable" { + t.Fatalf("lost node: %+v, %v", plan, err) + } +} + +func TestHostportAffinityConjoinsEveryAlternativeAndPreservesOtherConstraints(t *testing.T) { + gs := &gameplanev1alpha1.GameServer{ObjectMeta: metav1.ObjectMeta{Annotations: map[string]string{hostportNodeAnnotation: "node-a"}}, Spec: gameplanev1alpha1.GameServerSpec{Networking: gameplanev1alpha1.GameServerNetworking{Expose: "Hostport"}, Affinity: &corev1.Affinity{NodeAffinity: &corev1.NodeAffinity{RequiredDuringSchedulingIgnoredDuringExecution: &corev1.NodeSelector{NodeSelectorTerms: []corev1.NodeSelectorTerm{{MatchExpressions: []corev1.NodeSelectorRequirement{{Key: "pool", Operator: corev1.NodeSelectorOpIn, Values: []string{"a"}}}}, {MatchExpressions: []corev1.NodeSelectorRequirement{{Key: "pool", Operator: corev1.NodeSelectorOpIn, Values: []string{"b"}}}}}}}, PodAntiAffinity: &corev1.PodAntiAffinity{}}}} + got := hostportAffinity(gs) + for _, term := range got.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms { + if len(term.MatchFields) != 1 || term.MatchFields[0].Key != "metadata.name" || term.MatchFields[0].Values[0] != "node-a" { + t.Fatalf("unpinned alternative: %+v", term) + } + } + if got.PodAntiAffinity == nil || len(gs.Spec.Affinity.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms[0].MatchFields) != 0 { + t.Fatal("lost constraints or mutated CR") + } + // An explicitly empty selector matches no nodes; pinning must not widen it. + gs.Spec.Affinity.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms = nil + if len(hostportAffinity(gs).NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms) != 0 { + t.Fatal("empty selector widened") + } +} diff --git a/operator/internal/controller/gameserver_sentinel.go b/operator/internal/controller/gameserver_sentinel.go index c2549fc75..cb263fe7e 100644 --- a/operator/internal/controller/gameserver_sentinel.go +++ b/operator/internal/controller/gameserver_sentinel.go @@ -102,8 +102,10 @@ func wakeOnConnectEligible(gs *gameplanev1alpha1.GameServer, tmpl *gameplanev1al // whether it should exist, and whether the game Service should route to it // instead of the game pod. type sentinelPlan struct { - wantSentinel bool - routeToSentinel bool + wantSentinel bool + routeToSentinel bool + blockGame bool + hostportCondition *metav1.Condition } // planSentinel is the single source of truth for the sentinel/game-Service @@ -132,9 +134,10 @@ type sentinelPlan struct { // terminationGracePeriodSeconds plus the sentinel's shutdown drain // (sentinelShutdownDrainTimeout) keep any handed-through session alive // until it ends on its own; only new connections stop reaching it. -// - Hostport: the sentinel and the game pod bind the exact same host -// port, so the sentinel cannot outlive the wake decision — the game -// pod cannot even schedule until the sentinel releases that port. The +// - Hostport: planHostportSentinel records the endpoint node and waits +// for actual game Pod deletion before creating a pinned sentinel. On +// wake it waits for foreground sentinel deletion and actual Pod drain +// before permitting the game to start on that same node. The // sentinel is expected to bounce any clients it is holding with a // protocol-native "waking up, try again shortly" response before the // operator tears it down here; players simply reconnect once the game @@ -142,6 +145,22 @@ type sentinelPlan struct { func (r *GameServerReconciler) planSentinel( ctx context.Context, gs *gameplanev1alpha1.GameServer, tmpl *gameplanev1alpha1.GameTemplate, idle idleState, ) (sentinelPlan, error) { + if gs.Spec.Networking.Expose == "Hostport" { + eligible := wakeOnConnectEligible(gs, tmpl) + if !eligible { + idle = idleAwake + } + plan, err := r.planHostportSentinel(ctx, gs, idle) + if err != nil { + return plan, err + } + // A Hostport server that has never used wake keeps its existing + // status behavior. Still drain a leftover sentinel when disabling it. + if !eligible && gs.Annotations[hostportNodeAnnotation] == "" && !plan.blockGame { + return sentinelPlan{}, nil + } + return plan, nil + } if !wakeOnConnectEligible(gs, tmpl) { return sentinelPlan{}, nil } @@ -157,10 +176,6 @@ func (r *GameServerReconciler) planSentinel( // idle == idleAwake: either a wake was just decided, or the server was // never asleep to begin with — sentinelExists is false in the latter // case, so we return immediately without the extra StatefulSet Get. - if gs.Spec.Networking.Expose == "Hostport" { - // See the Hostport bullet above: no hold-open window, ever. - return sentinelPlan{}, nil - } exists, err := r.sentinelExists(ctx, gs.Namespace, gs.Name) if err != nil || !exists { return sentinelPlan{}, err @@ -206,7 +221,10 @@ func (r *GameServerReconciler) reconcileSentinel( want bool, ) error { if !want { - return r.deleteSentinel(ctx, gs.Namespace, gs.Name) + return r.deleteSentinel(ctx, gs.Namespace, gs.Name, gs.Spec.Networking.Expose == "Hostport") + } + if gs.Spec.Networking.Expose == "Hostport" && gs.Annotations[hostportNodeAnnotation] == "" { + return fmt.Errorf("hostport sentinel requires a persisted endpoint node") } // Create or update the sentinel Deployment. @@ -219,6 +237,44 @@ func (r *GameServerReconciler) reconcileSentinel( _, err := controllerutil.CreateOrUpdate(ctx, r.Client, dep, func() error { replicas := int32(1) + if gs.Spec.Networking.Expose == "Hostport" { + var live gameplanev1alpha1.GameServer + if err := r.hostportReader().Get(ctx, client.ObjectKeyFromObject(gs), &live); err != nil { + return err + } + if live.UID != gs.UID || !live.DeletionTimestamp.IsZero() { + return fmt.Errorf("GameServer changed identity or is deleting before sentinel mutation") + } + if live.Spec.Networking.Expose != "Hostport" || + live.Annotations[hostportNodeAnnotation] != gs.Annotations[hostportNodeAnnotation] || + live.Annotations[hostportNodeUIDAnnotation] != gs.Annotations[hostportNodeUIDAnnotation] { + return fmt.Errorf("hostport endpoint changed before sentinel mutation") + } + if dep.ResourceVersion != "" && !metav1.IsControlledBy(dep, &live) { + return fmt.Errorf("sentinel Deployment is not owned by this GameServer") + } + node, err := r.hostportEndpointNode(ctx, &live) + if err != nil { + return err + } + if node == nil { + // A node can be replaced after the sleep plan. Fence this + // producer before it creates a listener on the reused name. + replicas = 0 + dep.Spec.Replicas = &replicas + if dep.ResourceVersion != "" { + return nil // Stop without rolling the existing pod template. + } + } else if !hostportNodeReady(node) { + // Retain an existing listener's exact desired state while the + // original node recovers. Do not roll its template or start a + // stopped/new producer from a stale healthy-node plan. + if dep.ResourceVersion != "" { + return nil + } + replicas = 0 + } + } dep.Spec.Replicas = &replicas dep.Spec.Selector = &metav1.LabelSelector{ MatchLabels: map[string]string{ @@ -280,6 +336,16 @@ func (r *GameServerReconciler) reconcileSentinel( dep.Spec.Template.Spec.TerminationGracePeriodSeconds = &graceSeconds dep.Spec.Template.Spec.ServiceAccountName = sentinelSAName(gs) + if hostPort { + dep.Spec.Template.Spec.NodeSelector = gs.Spec.NodeSelector + dep.Spec.Template.Spec.Tolerations = gs.Spec.Tolerations + dep.Spec.Template.Spec.Affinity = hostportAffinity(gs) + dep.Spec.Strategy = appsv1.DeploymentStrategy{Type: appsv1.RecreateDeploymentStrategyType} + } else { + dep.Spec.Template.Spec.NodeSelector = nil + dep.Spec.Template.Spec.Tolerations = nil + dep.Spec.Template.Spec.Affinity = nil + } dep.Spec.Template.Spec.Containers = []corev1.Container{{ Name: "sentinel", Image: image, @@ -410,12 +476,15 @@ func (r *GameServerReconciler) sentinelIsReady(ctx context.Context, namespace, g } // deleteSentinel removes the sentinel Deployment if it exists. -func (r *GameServerReconciler) deleteSentinel(ctx context.Context, namespace, gsName string) error { +func (r *GameServerReconciler) deleteSentinel(ctx context.Context, namespace, gsName string, foreground ...bool) error { dep, exists, err := r.getSentinel(ctx, namespace, gsName) if err != nil || !exists { return err } policy := metav1.DeletePropagationBackground + if len(foreground) != 0 && foreground[0] { + policy = metav1.DeletePropagationForeground + } return client.IgnoreNotFound(r.Delete(ctx, dep, &client.DeleteOptions{PropagationPolicy: &policy})) } diff --git a/operator/internal/controller/gameserver_sentinel_envtest_test.go b/operator/internal/controller/gameserver_sentinel_envtest_test.go index e805d58c8..bf79d3c8a 100644 --- a/operator/internal/controller/gameserver_sentinel_envtest_test.go +++ b/operator/internal/controller/gameserver_sentinel_envtest_test.go @@ -12,6 +12,7 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/types" "k8s.io/client-go/util/retry" + "sigs.k8s.io/controller-runtime/pkg/client" gameplanev1alpha1 "github.com/GameplanePanel/gameplane/operator/api/v1alpha1" ) @@ -181,8 +182,20 @@ func TestSentinelReconciliation(t *testing.T) { } deleteCleanup(t, tmpl) + node := &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: uniqueName("sentinel-node")}} + if err := k8sClient.Create(ctx, node); err != nil { + t.Fatalf("create node: %v", err) + } + deleteCleanup(t, node) + node.Status.Conditions = []corev1.NodeCondition{{Type: corev1.NodeReady, Status: corev1.ConditionTrue}} + if err := k8sClient.Status().Update(ctx, node); err != nil { + t.Fatalf("mark node Ready: %v", err) + } + gs := &gameplanev1alpha1.GameServer{ - ObjectMeta: metav1.ObjectMeta{Name: "test-sentinel-hostport", Namespace: ns}, + ObjectMeta: metav1.ObjectMeta{Name: "test-sentinel-hostport", Namespace: ns, Annotations: map[string]string{ + hostportNodeAnnotation: node.Name, hostportNodeUIDAnnotation: string(node.UID), + }}, Spec: gameplanev1alpha1.GameServerSpec{ TemplateRef: gameplanev1alpha1.GameTemplateRef{Name: tmpl.Name}, Idle: armedIdleSpec(), @@ -207,6 +220,9 @@ func TestSentinelReconciliation(t *testing.T) { if len(dep.Spec.Template.Spec.Containers) == 0 { t.Fatal("no containers in sentinel deployment") } + if dep.Spec.Replicas == nil || *dep.Spec.Replicas != 1 { + t.Fatal("valid endpoint did not start the sentinel") + } container := dep.Spec.Template.Spec.Containers[0] if len(container.Ports) == 0 { t.Fatal("no ports in sentinel container") @@ -550,8 +566,40 @@ func TestGameServer_WakeOnConnectHostportNoHoldOpen(t *testing.T) { return ssReplicas(t, ns, "wakehp") == 1, "statefulset not scaled up yet" }) setSSStatus(t, ns, "wakehp", 1, 1) + node := &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: uniqueName("hostport-node")}} + if err := k8sClient.Create(ctx, node); err != nil { + t.Fatal(err) + } + deleteCleanup(t, node) + node.Status.Conditions = []corev1.NodeCondition{{Type: corev1.NodeReady, Status: corev1.ConditionTrue}} + if err := k8sClient.Status().Update(ctx, node); err != nil { + t.Fatal(err) + } + gamePod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "wakehp-0", Namespace: ns, Labels: map[string]string{"app.kubernetes.io/name": "gameplane-game", "app.kubernetes.io/instance": gs.Name}}, Spec: corev1.PodSpec{NodeName: node.Name, Containers: []corev1.Container{{Name: "game", Image: "busybox"}}}} + if err := k8sClient.Create(ctx, gamePod); err != nil { + t.Fatal(err) + } setAsleep(t, ns, "wakehp") + eventually(t, func() (bool, string) { return ssReplicas(t, ns, gs.Name) == 0, "game not scaled down" }) + consistently(t, time.Second, func() (bool, string) { + return !deploymentExists(t, ns, "wakehp-waker"), "sentinel overlaps the live game Pod" + }) + zero := int64(0) + if err := k8sClient.Delete(ctx, gamePod, &client.DeleteOptions{GracePeriodSeconds: &zero}); err != nil { + t.Fatal(err) + } + if err := retry.RetryOnConflict(retry.DefaultRetry, func() error { + var ss appsv1.StatefulSet + if err := k8sClient.Get(ctx, types.NamespacedName{Namespace: ns, Name: gs.Name}, &ss); err != nil { + return err + } + ss.Status.Replicas, ss.Status.ReadyReplicas = 0, 0 + ss.Status.ObservedGeneration = ss.Generation + return k8sClient.Status().Update(ctx, &ss) + }); err != nil { + t.Fatal(err) + } eventually(t, func() (bool, string) { return deploymentExists(t, ns, "wakehp-waker"), "sentinel deployment not created" }) @@ -565,6 +613,23 @@ func TestGameServer_WakeOnConnectHostportNoHoldOpen(t *testing.T) { clearAsleep(t, ns, "wakehp") eventually(t, func() (bool, string) { - return !deploymentExists(t, ns, "wakehp-waker"), "sentinel not deleted immediately on Hostport wake" + var dep appsv1.Deployment + err := k8sClient.Get(ctx, types.NamespacedName{Namespace: ns, Name: "wakehp-waker"}, &dep) + return err == nil && !dep.DeletionTimestamp.IsZero(), "sentinel foreground deletion not requested" + }) + consistently(t, time.Second, func() (bool, string) { + return ssReplicas(t, ns, gs.Name) == 0, "game restarted before sentinel producer disappeared" + }) + // envtest has no garbage collector: finish foreground deletion explicitly. + var dep appsv1.Deployment + if err := k8sClient.Get(ctx, types.NamespacedName{Namespace: ns, Name: "wakehp-waker"}, &dep); err != nil { + t.Fatal(err) + } + dep.Finalizers = nil + if err := k8sClient.Update(ctx, &dep); err != nil { + t.Fatal(err) + } + eventually(t, func() (bool, string) { + return ssReplicas(t, ns, gs.Name) == 1, "game did not resume after sentinel deletion" }) } diff --git a/operator/internal/controller/gameserver_status.go b/operator/internal/controller/gameserver_status.go index dcc1454ec..361697ffa 100644 --- a/operator/internal/controller/gameserver_status.go +++ b/operator/internal/controller/gameserver_status.go @@ -43,6 +43,7 @@ func (r *GameServerReconciler) reconcileStatus( ctx context.Context, gs *gameplanev1alpha1.GameServer, idle idleState, idleStatus *gameplanev1alpha1.IdleStatus, idleScheduleErr error, tunnelPlan tunnelPlan, tmpl *gameplanev1alpha1.GameTemplate, svcEvents []corev1.Event, conflictingServer string, + hostportConditions ...*metav1.Condition, ) (time.Duration, error) { // base captures the object as fetched so we can issue a JSON merge // patch of only the fields this reconciler owns. The agent sidecar @@ -194,6 +195,11 @@ func (r *GameServerReconciler) reconcileStatus( // Compute tunnel conditions. gs.Status.Conditions = computeTunnelConditions(gs, tunnelPlan, tunnelDep) + if len(hostportConditions) != 0 && hostportConditions[0] != nil { + gs.Status.Conditions = applyHostportCondition(gs.Status.Conditions, hostportConditions[0]) + } else { + gs.Status.Conditions = removeCondition(gs.Status.Conditions, hostportWakeCondition) + } // Validate and merge playit tunnel endpoints if present. var playitErr string @@ -319,6 +325,19 @@ func (r *GameServerReconciler) reconcileStatus( return 15 * time.Second, nil } +// applyHostportCondition preserves the more specific failure already computed +// for the game while still exposing the independent endpoint diagnostic. +func applyHostportCondition(conditions []metav1.Condition, hostport *metav1.Condition) []metav1.Condition { + conditions = upsertCondition(conditions, *hostport) + ready := meta.FindStatusCondition(conditions, "Ready") + if hostport.Status != metav1.ConditionTrue && (ready == nil || ready.Status != metav1.ConditionFalse) { + condition := *hostport + condition.Type = "Ready" + conditions = upsertCondition(conditions, condition) + } + return conditions +} + func derivePhase( gs *gameplanev1alpha1.GameServer, ssExists, ssReady, hbFresh bool, idle idleState, ) gameplanev1alpha1.GameServerPhase { diff --git a/operator/specs.md b/operator/specs.md index 9b3bd3a34..780c969c3 100644 --- a/operator/specs.md +++ b/operator/specs.md @@ -86,6 +86,68 @@ start/stop and expiry remain reconciler responsibilities; the gateway cannot bypass them or operate an arbitrary pod. Existing game resources and storage are not recreated to enable remote parity. +### Hostport wake handoff + +Hostport wake-on-connect keeps a durable endpoint node name and Node UID in +`gameplane.local/hostport-node` / `gameplane.local/hostport-node-uid`, captured from +the live game Pod before the first sleep. The current-node dashboard annotation +is independent. Sleep waits for the zero-replica StatefulSet generation to be +observed and for actual game Pod disappearance through the uncached API reader. +Wake foreground-deletes the sentinel and waits for its Deployment and all Pods +to disappear before scheduling the game. Both successor templates use required +node affinity for the recorded endpoint, combined with the game's scheduling +constraints; the sentinel carries its tolerations. Node gets are read-only RBAC. +`HostportWakeReady` reports listener readiness, each drain and unavailable or +unschedulable endpoints without blocking unrelated reconciliation. There is no +automatic endpoint migration. Healthy Hostport listeners rely on watches; the +five-second backstop runs only while `HostportWakeReady` is not True. Endpoint +diagnostics never replace an existing `Ready=False` game failure. Pending-listener +messages omit the endpoint node name until it has been recorded. + +A missing or replaced endpoint Node forces the game StatefulSet to zero replicas +even if a replacement game Pod already exists. The stop-only path projects the +live GameServer's explicit lifecycle annotation verbatim, or clears a stale one, +so admitted rollouts can observe drainage without losing the endpoint fence. +A deleting Node with the original +UID permits only its owned original running game Pod to finish a bounded graceful +stop. Before issuing the stop, the operator durably switches the StatefulSet to +OnDelete and gates scheduling of replacement Pods, waiting for that generation to +be observed. The running Pod is not rolled. The save deadline/readiness check then +scales to zero; a Backup or inherited copy recovery retains its cleanup precedence +before any save/stop RPC. Pending, foreign or replacement Pods receive no save window. The +temporary gate and previous update strategy are restored during valid endpoint +reconciliation or the documented explicit migration. Cleanup validates saved +identities and strategy fields, preserving valid custom RollingUpdate and Recreate +strategies. Malformed or incomplete records retain the current update strategy +while removing only the operator's stop gate/annotation; they cannot block a +requested scale-down. The live Node +identity is checked again immediately before workload mutation, including +backup replica preservation and restore startup; existing container and scheduling +settings are otherwise retained. The sentinel Deployment also rechecks the live GameServer endpoint +and Node UID before mutation; a stale sleep plan cannot start a listener on a +replacement node. A transient NotReady condition on the same Node UID preserves an +existing running game or sleeping sentinel. Sentinel planning requires a running, +non-terminating Pod on that endpoint, and mutation preserves its Deployment without +rolling the template or raising replicas while the node is NotReady. Pending or +terminating Pods do not authorize a wake or a new sleep listener. + +After restoring node capacity/readiness, reconciliation resumes automatically. +For intentional migration, suspend the server, wait for both game and sentinel +Pods to disappear, clear both durable endpoint annotations, update scheduling, +storage placement and the client-facing node address, then resume the game before +sleeping it again. Do not clear the annotations while a listener remains active. +An installation upgraded while already asleep without a recorded game node must +first start its game to establish placement. Hostport clients reconnect after +handoff; the historical "briefly Pending, self-heals" sequence in the archived +[July design](../docs/archive/superpowers/specs/2026-07-24-wake-on-connect-design.md#sequencing-and-port-handover) +is superseded by this specification. Service-backed wake retains its +established-session drain behavior. Auxiliary resource defaults are defined separately. + +The sentinel starts the join deadline before its Kubernetes wake PATCH, whose +separate five-second maximum fits inside that budget. Concurrent callers share +the in-flight PATCH result. UDP reads enqueue coalesced requests to one bounded +worker, and established TCP sessions use the longer session/drain context. + ## Template revision admission - Admission and the final pin update reject removing or retyping a populated diff --git a/sentinel/main.go b/sentinel/main.go index edc563b4d..954962e98 100644 --- a/sentinel/main.go +++ b/sentinel/main.go @@ -72,7 +72,8 @@ const ( // defaultPollInterval is how often waitForUpstream retries dialing the // game-direct Service while holding a connection open. - defaultPollInterval = 250 * time.Millisecond + defaultPollInterval = 250 * time.Millisecond + defaultWakePatchTimeout = 5 * time.Second // defaultShutdownDrainTimeout bounds how long the sentinel keeps // in-flight sessions alive after it is told to shut down (see @@ -354,44 +355,80 @@ type wakeRequester interface { // waker patches the GameServer's idle-wake-requested annotation to signal // the operator that this server should wake. // -// It coalesces bursts: while a patch is in flight, or within minInterval of -// the last one landing, RequestWake is a cheap no-op instead of issuing -// another apiserver call. This is what keeps a port scan, or a pile of +// It coalesces bursts: callers share the result of an in-flight patch, and +// successful patches suppress another call for minInterval. This keeps a port scan, or a pile of // players reconnecting at once, from hammering the apiserver with patches — // the operator only needs to see the annotation change once to act on it. type waker struct { - res dynamic.ResourceInterface - name string - minInterval time.Duration + res dynamic.ResourceInterface + name string + minInterval time.Duration + patchTimeout time.Duration mu sync.Mutex - inFlight bool + inFlight *wakeAttempt lastPatch time.Time } +type wakeAttempt struct { + done chan struct{} + err error + callerCanceled bool +} + func newWaker(res dynamic.ResourceInterface, name string, minInterval time.Duration) *waker { - return &waker{res: res, name: name, minInterval: minInterval} + return &waker{res: res, name: name, minInterval: minInterval, patchTimeout: defaultWakePatchTimeout} } // RequestWake stamps a fresh token onto the GameServer's // idle-wake-requested annotation via a JSON merge patch (rather than a full // object update), so any other annotations already on the object survive. -func (w *waker) RequestWake(ctx context.Context) error { - w.mu.Lock() - now := time.Now() - if w.inFlight || (!w.lastPatch.IsZero() && now.Sub(w.lastPatch) < w.minInterval) { +func (w *waker) RequestWake(ctx context.Context) (result error) { + var attempt *wakeAttempt + for { + if err := ctx.Err(); err != nil { + return err + } + w.mu.Lock() + now := time.Now() + if shared := w.inFlight; shared != nil { + w.mu.Unlock() + select { + case <-ctx.Done(): + return ctx.Err() + case <-shared.done: + // A live caller still owes a wake when the first caller's + // context canceled the shared PATCH. Compete for a new shared + // attempt within this caller's own remaining budget. + if shared.callerCanceled { + continue + } + return shared.err + } + } + if !w.lastPatch.IsZero() && now.Sub(w.lastPatch) < w.minInterval { + w.mu.Unlock() + return nil + } + attempt = &wakeAttempt{done: make(chan struct{})} + w.inFlight = attempt w.mu.Unlock() - return nil + break } - w.inFlight = true - w.mu.Unlock() defer func() { w.mu.Lock() - w.inFlight = false - w.lastPatch = time.Now() + attempt.err = result + attempt.callerCanceled = result != nil && ctx.Err() != nil + w.inFlight = nil + if result == nil { + w.lastPatch = time.Now() + } + close(attempt.done) w.mu.Unlock() }() + patchCtx, cancel := context.WithTimeout(ctx, w.patchTimeout) + defer cancel() patch, err := json.Marshal(map[string]any{ "metadata": map[string]any{ @@ -404,7 +441,7 @@ func (w *waker) RequestWake(ctx context.Context) error { return fmt.Errorf("encode wake patch: %w", err) } - if _, err := w.res.Patch(ctx, w.name, types.MergePatchType, patch, metav1.PatchOptions{}); err != nil { + if _, err := w.res.Patch(patchCtx, w.name, types.MergePatchType, patch, metav1.PatchOptions{}); err != nil { return fmt.Errorf("patch %s annotation: %w", idleWakeRequestedAnnotation, err) } return nil @@ -714,17 +751,20 @@ type joinParams struct { // bidirectionally; on timeout, the connection is bounced (if a bounce // function is set) and closed. func handleJoin(ctx context.Context, conn net.Conn, br *bufio.Reader, w wakeRequester, p joinParams) { - if err := w.RequestWake(ctx); err != nil { + joinCtx, cancelJoin := context.WithTimeout(ctx, p.deadline) + defer cancelJoin() + if err := w.RequestWake(joinCtx); err != nil { log.Printf("request wake: %v", err) } - upstream, err := waitForUpstream(ctx, p.upstreamAddr, p.deadline, p.pollInterval) + upstream, err := waitForUpstream(joinCtx, p.upstreamAddr, p.deadline, p.pollInterval) if err != nil { if p.bounce != nil { p.bounce(conn) } return } + cancelJoin() defer func() { if err := upstream.Close(); err != nil && !errors.Is(err, net.ErrClosed) { log.Printf("close upstream connection: %v", err) @@ -842,6 +882,11 @@ func serveUDP(ctx context.Context, port PortConfig, w wakeRequester, cfg Config, } return } + defer func() { _ = pc.Close() }() + serveUDPPackets(ctx, pc, port, w, cfg, errCh) +} + +func serveUDPPackets(ctx context.Context, pc net.PacketConn, port PortConfig, w wakeRequester, cfg Config, errCh chan<- error) { // Belt-and-suspenders shutdown: Close() unblocks a ReadFrom that's // already parked waiting for a packet, and the read deadline below @@ -858,6 +903,11 @@ func serveUDP(ctx context.Context, port PortConfig, w wakeRequester, cfg Config, sweepCtx, cancelSweep := context.WithCancel(ctx) defer cancelSweep() go heuristic.sweepLoop(sweepCtx, cfg.UDPWindow) + wakeWorker := newUDPWakeWorker(sweepCtx, w) + defer func() { + cancelSweep() + <-wakeWorker.done + }() buf := make([]byte, 2048) for { @@ -882,9 +932,7 @@ func serveUDP(ctx context.Context, port PortConfig, w wakeRequester, cfg Config, } if heuristic.shouldWake(addr.String(), time.Now()) { - if err := w.RequestWake(ctx); err != nil { - log.Printf("request wake: %v", err) - } + wakeWorker.request() } } } diff --git a/sentinel/specs.md b/sentinel/specs.md index e39166621..d74e9ade8 100644 --- a/sentinel/specs.md +++ b/sentinel/specs.md @@ -59,12 +59,12 @@ Single executable module; no subdirectories or packages. - `none`: Operator still advertises the port (container spec, Service), but the sentinel doesn't listen on it (port scan sink). - **Startup validation:** At startup, all configured `wakeProtocol` values (except `generic` and `none`) are validated against the gameproto registry. Unknown protocol names cause a fatal startup error listing available registered protocols. -- **`WAKE_DEADLINE`** (optional, default `25s`): How long to hold a join connection waiting for the game pod to come up. Must be less than the game client's own timeout (Minecraft ~30s, Terraria ~30s) or the client gives up first. +- **`WAKE_DEADLINE`** (optional, default `25s`): Total join budget, starting before the wake PATCH and including upstream dialing. The PATCH is bounded by the lesser of five seconds and the remaining join budget. Must be less than the game client's own timeout (Minecraft ~30s, Terraria ~30s) or the client gives up first. After a successful upstream connection, proxying uses the session/drain context and outlives this budget. - **`UDP_PACKET_THRESHOLD`** (optional, default `3`): Number of packets from one source within the window before triggering a wake. - **`UDP_PACKET_WINDOW`** (optional, default `10s`): Time window for counting UDP packets. - **`UDP_MAX_SOURCES`** (optional, default `4096`): Maximum number of distinct UDP sources (keyed by `ip:port`, not just IP) to track. Older sources are evicted when this limit is reached. - **`MAX_CONNECTIONS`** (optional, default `256`): Maximum concurrent TCP connection handlers. Excess connections queue in the OS accept backlog. -- **`WAKE_PATCH_INTERVAL`** (optional, default `2s`): Minimum time between GameServer annotation patches. Bursts of joining clients within this window are coalesced into one apiserver call. +- **`WAKE_PATCH_INTERVAL`** (optional, default `2s`): Minimum time between successful GameServer annotation patches. Concurrent callers share an in-flight PATCH and can cancel their own waits. If the first caller's context cancels that PATCH, live waiters retry through one new shared attempt within their own remaining budgets. API failures and the independent five-second PATCH timeout are returned to all waiters without an automatic retry. - **`SHUTDOWN_DRAIN_TIMEOUT`** (optional, default `4h`, must be `>= 0`): How long in-flight sessions are kept alive after SIGTERM before they are force-closed. `0s` closes them immediately. The operator always sets it: `4h0m0s` in the ClusterIP/NodePort/LoadBalancer expose modes (with `terminationGracePeriodSeconds` = drain + 30s, so the kubelet does not SIGKILL the pod mid-drain), and `0s` in Hostport mode (grace 30s), where the terminating sentinel would otherwise keep holding the host port the game pod needs. ### Kubernetes RBAC @@ -246,7 +246,7 @@ Each accepted TCP connection runs as a *session* under its own context, separate 5. **No data duplication or loss.** The complete original stream is: `Consumed + remaining(bufio.Reader) + new-data-from-client`. Both classifiers and proxying preserve this boundary. -6. **Annotation patch coalesces bursts.** Bursts of players reconnecting at once don't hammer the apiserver; RequestWake is a no-op if a patch is already in flight or within `wakePatchInterval` of the last one. +6. **Annotation patch coalesces bursts.** Callers awaiting an in-flight PATCH observe that attempt's completion and error, unless their own context expires first. Only successful attempts start the `wakePatchInterval` cooldown; failures release the shared attempt so a later request can retry. One worker per UDP listener accepts a one-entry coalescing queue; packet reception and bounded source tracking continue while Kubernetes is stalled. Listener shutdown cancels and joins that worker. 7. **UDP sources are evicted fairly.** The least-recently-active source is evicted (LRU), not a random or oldest source, so brief bursts of scanners don't unfairly starve legitimate players trying to rejoin. @@ -293,6 +293,7 @@ Each accepted TCP connection runs as a *session* under its own context, separate - **Shutdown drain (F-179):** `TestRunDrainsProxiedSessionAfterContextCancel` (replaces `TestProxyBidirectionalStopsOnContextCancel`, which asserted the old cut-on-cancel behaviour) verifies that after cancel new connections are refused while a handed-through session keeps working in both directions and `run` returns only once it ends; `TestRunForceClosesSessionsAfterDrainTimeout` verifies the drain is bounded; `TestLoadConfigShutdownDrainTimeout` covers `SHUTDOWN_DRAIN_TIMEOUT` parsing. - **Listener failure (F-180):** `TestRunSurfacesUDPListenFailureWhileTCPHealthy` verifies a UDP bind failure is returned promptly while a TCP listener is still healthy. - **Waker (annotation patching):** `TestWakerRequestWakeSetsAnnotation`, `TestWakerRequestWakePreservesOtherAnnotations`, `TestWakerRequestWakeCoalescesBursts` verify merge patch isolation, burst coalescing. +- **Wake availability:** `wake_budget_test.go` covers a hanging HTTP API, shared failure and retry, the total join deadline, worker cancellation/coalescing, successful gameplay beyond the join deadline, and receiving a 101-datagram burst while wake is stalled. CI logs elapsed receive time; this fixture does not establish a production capacity ceiling. - **Registry dispatch:** `TestRegistryProtocolDispatchMinecraftJoinWakes`, `TestRegistryProtocolDispatchMinecraftStatusPingNoWake`, `TestRegistryProtocolDispatchTerrariaJoinWakes`, `TestRegistryProtocolDispatchUnknownNoWake`, `TestRegistryProtocolReplayConsumedBytes` verify registry lookup, Classifier.Classify invocation, protocol-agnostic join/status/unknown handling, and consumed-byte replay. - **Startup validation:** `TestParsePortsConfigRejectsUnknownProtocol` verifies that unknown protocol names cause a fatal startup error with a clear message listing available protocols. - **Generic handlers:** `TestHandleGenericWakesOnConnect`, `TestHandleGenericClosesSilentlyOnDeadline` verify generic TCP (always join), generic UDP bounce (no message). diff --git a/sentinel/udp_wake.go b/sentinel/udp_wake.go new file mode 100644 index 000000000..3b8a9b29a --- /dev/null +++ b/sentinel/udp_wake.go @@ -0,0 +1,41 @@ +package main + +import ( + "context" + "log" +) + +type udpWakeWorker struct { + requests chan struct{} + done chan struct{} +} + +func newUDPWakeWorker(ctx context.Context, w wakeRequester) *udpWakeWorker { + worker := &udpWakeWorker{requests: make(chan struct{}, 1), done: make(chan struct{})} + go func() { + defer close(worker.done) + for { + select { + case <-ctx.Done(): + return + case <-worker.requests: + if ctx.Err() != nil { + return + } + wakeCtx, cancel := context.WithTimeout(ctx, defaultWakePatchTimeout) + if err := w.RequestWake(wakeCtx); err != nil { + log.Printf("request UDP wake: %v", err) + } + cancel() + } + } + }() + return worker +} + +func (w *udpWakeWorker) request() { + select { + case w.requests <- struct{}{}: + default: + } +} diff --git a/sentinel/wake_budget_test.go b/sentinel/wake_budget_test.go new file mode 100644 index 000000000..25e67a424 --- /dev/null +++ b/sentinel/wake_budget_test.go @@ -0,0 +1,349 @@ +package main + +import ( + "bufio" + "context" + "errors" + "io" + "net" + "net/http" + "net/http/httptest" + "sync" + "sync/atomic" + "testing" + "time" + + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/client-go/dynamic" + "k8s.io/client-go/rest" + ktesting "k8s.io/client-go/testing" +) + +func TestWakeSharedFailureAndImmediateRetry(t *testing.T) { + w, client := newFakeGameServerWaker(t, "alpha", "games", time.Minute, nil) + started, release := make(chan struct{}), make(chan struct{}) + var patches atomic.Int32 + failure := errors.New("API unavailable") + client.PrependReactor("patch", "gameservers", func(ktesting.Action) (bool, runtime.Object, error) { + if patches.Add(1) == 1 { + close(started) + <-release + return true, nil, failure + } + return false, nil, nil + }) + first, second := make(chan error, 1), make(chan error, 1) + go func() { first <- w.RequestWake(t.Context()) }() + <-started + waiting := &observedWaitContext{Context: t.Context(), observed: make(chan struct{})} + go func() { second <- w.RequestWake(waiting) }() + select { + case err := <-second: + close(release) + t.Fatalf("in-flight waiter returned before wake completed: %v", err) + case <-waiting.observed: + case <-time.After(time.Second): + close(release) + t.Fatal("second wake request never reached its wait") + } + close(release) + for _, result := range []<-chan error{first, second} { + select { + case err := <-result: + if !errors.Is(err, failure) { + t.Fatalf("got %v, want shared failure", err) + } + case <-time.After(time.Second): + t.Fatal("wake waiter did not finish") + } + } + if err := w.RequestWake(t.Context()); err != nil { + t.Fatalf("retry: %v", err) + } + if patches.Load() != 2 { + t.Fatalf("got %d patches, want failed attempt and retry", patches.Load()) + } +} + +func TestWakeLiveWaiterRetriesCanceledOwner(t *testing.T) { + started := make(chan struct{}) + var patches atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + _, _ = io.Copy(io.Discard, req.Body) + if patches.Add(1) == 1 { + close(started) + <-req.Context().Done() + return + } + w.Header().Set("Content-Type", "application/json") + _, _ = io.WriteString(w, `{"apiVersion":"gameplane.local/v1alpha1","kind":"GameServer","metadata":{"name":"alpha"}}`) + })) + defer server.Close() + client, err := dynamic.NewForConfig(&rest.Config{Host: server.URL}) + if err != nil { + t.Fatal(err) + } + w := newWaker(client.Resource(gameServerGVR).Namespace("games"), "alpha", time.Minute) + ownerCtx, cancelOwner := context.WithCancel(t.Context()) + defer cancelOwner() + ownerDone := make(chan error, 1) + go func() { ownerDone <- w.RequestWake(ownerCtx) }() + select { + case <-started: + case <-time.After(time.Second): + t.Fatal("owning wake did not reach the API") + } + waiting := &observedWaitContext{Context: t.Context(), observed: make(chan struct{})} + waiterDone := make(chan error, 1) + go func() { waiterDone <- w.RequestWake(waiting) }() + select { + case <-waiting.observed: + case <-time.After(time.Second): + t.Fatal("live caller did not join the shared wake") + } + // Canceling a different waiter must release only that caller, without + // canceling the owning PATCH or the live caller's shared wait. + canceledCtx, cancelWaiter := context.WithCancel(t.Context()) + defer cancelWaiter() + canceled := &observedWaitContext{Context: canceledCtx, observed: make(chan struct{})} + canceledDone := make(chan error, 1) + go func() { canceledDone <- w.RequestWake(canceled) }() + select { + case <-canceled.observed: + case <-time.After(time.Second): + t.Fatal("cancelable caller did not join the shared wake") + } + cancelWaiter() + select { + case err := <-canceledDone: + if !errors.Is(err, context.Canceled) { + t.Fatalf("canceled waiter: %v", err) + } + case <-time.After(time.Second): + t.Fatal("canceled waiter remained blocked") + } + if patches.Load() != 1 { + t.Fatal("canceling a waiter started another PATCH") + } + cancelOwner() + select { + case err := <-ownerDone: + if !errors.Is(err, context.Canceled) { + t.Fatalf("owning caller: %v", err) + } + case <-time.After(time.Second): + t.Fatal("owning caller remained blocked after cancellation") + } + select { + case err := <-waiterDone: + if err != nil { + t.Fatalf("live waiter did not complete its own wake: %v", err) + } + case <-time.After(time.Second): + t.Fatal("live waiter did not retry the canceled shared wake") + } + if err := w.RequestWake(t.Context()); err != nil { + t.Fatalf("coalesced successful wake: %v", err) + } + if patches.Load() != 2 { + t.Fatalf("got %d PATCHes, want canceled owner and one successful retry", patches.Load()) + } +} + +// Done is evaluated when RequestWake waits on the shared result. This handshake +// avoids relying on the scheduler running the second goroutine within a sleep. +type observedWaitContext struct { + context.Context + observed chan struct{} + once sync.Once +} + +func (c *observedWaitContext) Done() <-chan struct{} { + c.once.Do(func() { close(c.observed) }) + return c.Context.Done() +} + +func TestWakePatchHasIndependentTimeout(t *testing.T) { + release := make(chan struct{}) + server := httptest.NewServer(http.HandlerFunc(func(_ http.ResponseWriter, req *http.Request) { + _, _ = io.Copy(io.Discard, req.Body) + select { + case <-req.Context().Done(): + case <-release: + } + })) + defer server.Close() + defer close(release) + client, err := dynamic.NewForConfig(&rest.Config{Host: server.URL}) + if err != nil { + t.Fatal(err) + } + w := newWaker(client.Resource(gameServerGVR).Namespace("games"), "alpha", time.Minute) + w.patchTimeout = 25 * time.Millisecond + started := time.Now() + if err := w.RequestWake(t.Context()); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("got %v, want deadline exceeded", err) + } + if time.Since(started) > time.Second { + t.Fatal("wake PATCH exceeded its own timeout") + } +} + +type contextBoundWake struct{ started chan struct{} } + +func (w *contextBoundWake) RequestWake(ctx context.Context) error { + close(w.started) + <-ctx.Done() + return ctx.Err() +} + +func TestJoinBudgetIncludesWakeRequest(t *testing.T) { + server, client := net.Pipe() + defer server.Close() + defer client.Close() + wake := &contextBoundWake{started: make(chan struct{})} + done := make(chan struct{}) + go func() { + defer close(done) + handleJoin(t.Context(), server, bufio.NewReader(server), wake, joinParams{upstreamAddr: "127.0.0.1:1", deadline: 25 * time.Millisecond, pollInterval: time.Millisecond}) + }() + <-wake.started + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("join deadline did not bound wake request") + } +} + +func TestUDPWakeWorkerCoalescesAndStops(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + wake := &contextBoundWake{started: make(chan struct{})} + worker := newUDPWakeWorker(ctx, wake) + worker.request() + <-wake.started + for range 1000 { + worker.request() + } + if len(worker.requests) > 1 { + t.Fatal("unbounded UDP wake backlog") + } + cancel() + select { + case <-worker.done: + case <-time.After(time.Second): + t.Fatal("UDP wake worker did not stop") + } +} + +func TestSuccessfulGameplayOutlivesJoinBudget(t *testing.T) { + upstream, err := (&net.ListenConfig{}).Listen(t.Context(), "tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer upstream.Close() + accepted := make(chan struct{}) + go func() { + conn, acceptErr := upstream.Accept() + if acceptErr != nil { + return + } + defer conn.Close() + close(accepted) + _, _ = io.Copy(conn, conn) + }() + server, player := tcpPipe(t) + defer player.Close() + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + done := make(chan struct{}) + go func() { + defer close(done) + defer server.Close() + handleJoin(ctx, server, bufio.NewReader(server), newCountingWaker(nil), joinParams{upstreamAddr: upstream.Addr().String(), deadline: 100 * time.Millisecond, pollInterval: time.Millisecond}) + }() + select { + case <-accepted: + case <-time.After(time.Second): + t.Fatal("join did not reach game") + } + // The session must retain the drain context after the short join context + // is cancelled, including after its original deadline has elapsed. + time.Sleep(200 * time.Millisecond) + _ = player.SetDeadline(time.Now().Add(time.Second)) + if _, err := player.Write([]byte("still playing")); err != nil { + t.Fatal(err) + } + got := make([]byte, len("still playing")) + if _, err := io.ReadFull(player, got); err != nil || string(got) != "still playing" { + t.Fatalf("gameplay after join budget: %q, %v", got, err) + } + cancel() + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("session did not stop on drain cancellation") + } +} + +type observedPacketConn struct { + net.PacketConn + reads chan struct{} +} + +func (c observedPacketConn) ReadFrom(p []byte) (int, net.Addr, error) { + n, addr, err := c.PacketConn.ReadFrom(p) + if err == nil { + c.reads <- struct{}{} + } + return n, addr, err +} + +func TestUDPReceiveContinuesWhileWakeIsStalled(t *testing.T) { + listener, err := (&net.ListenConfig{}).ListenPacket(t.Context(), "udp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer listener.Close() + client, err := (&net.Dialer{}).DialContext(t.Context(), "udp", listener.LocalAddr().String()) + if err != nil { + t.Fatal(err) + } + defer client.Close() + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + wake := &contextBoundWake{started: make(chan struct{})} + reads := make(chan struct{}, 128) + done := make(chan struct{}) + go func() { + defer close(done) + serveUDPPackets(ctx, observedPacketConn{listener, reads}, PortConfig{}, wake, Config{UDPThreshold: 1, UDPWindow: time.Minute, UDPMaxSources: 10}, make(chan error, 1)) + }() + if _, err := client.Write([]byte("wake")); err != nil { + t.Fatal(err) + } + select { + case <-wake.started: + case <-time.After(time.Second): + t.Fatal("wake did not start") + } + started := time.Now() + for range 100 { + if _, err := client.Write([]byte("game packet")); err != nil { + t.Fatal(err) + } + } + for range 101 { + select { + case <-reads: + case <-time.After(time.Second): + t.Fatal("UDP receive blocked behind stalled wake PATCH") + } + } + t.Logf("received 101 datagrams while wake blocked in %s", time.Since(started)) + cancel() + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("UDP listener/worker did not stop") + } +} diff --git a/test/e2e/buckets.sh b/test/e2e/buckets.sh index f23e4edcf..988c6d8b4 100755 --- a/test/e2e/buckets.sh +++ b/test/e2e/buckets.sh @@ -424,6 +424,11 @@ EOF # Tests that exist in the suite but deliberately run in NO bucket. Every # entry needs a reason; `verify` fails on any unlisted stray so additions # here are a conscious, reviewed act. Currently empty. +bucket_wake_hostport() { cat <<'EOF' +TestGameServer_HostportWakeStableEndpoint +EOF +} + bucket_management_placement() { cat <<'EOF' TestManagementPlacementAndDisruption EOF @@ -446,7 +451,7 @@ bucket_network_isolation() { } bucket_names() { - printf '%s\n' operator api-auth api-roles api-rbac api-agent api-mods ratelimit bot-fast bot-heavy multicluster upgrade postgres-availability telemetry capture-overhead optional-components restricted-workloads network-isolation management-placement backup-preparation monitoring + printf '%s\n' operator api-auth api-roles api-rbac api-agent api-mods ratelimit bot-fast bot-heavy multicluster upgrade postgres-availability telemetry capture-overhead optional-components restricted-workloads wake-hostport network-isolation management-placement backup-preparation monitoring } list_bucket() { @@ -466,6 +471,7 @@ list_bucket() { telemetry) bucket_telemetry ;; capture-overhead) bucket_capture_overhead ;; optional-components) bucket_optional_components ;; + wake-hostport) bucket_wake_hostport ;; network-isolation) bucket_network_isolation ;; restricted-workloads) bucket_restricted_workloads ;; management-placement) bucket_management_placement ;; diff --git a/test/e2e/ci_selection_e2e_test.go b/test/e2e/ci_selection_e2e_test.go index 411a8ea2e..b37fee629 100644 --- a/test/e2e/ci_selection_e2e_test.go +++ b/test/e2e/ci_selection_e2e_test.go @@ -22,7 +22,7 @@ import ( // ciSelectionBuckets mirrors E2E_BUCKETS in hack/ci_scope.py (the e2e-go matrix). var ciSelectionBuckets = []string{ "operator", "api-auth", "api-roles", "api-rbac", "api-agent", "api-mods", - "telemetry", "optional-components", "restricted-workloads", "network-isolation", "management-placement", "backup-preparation", "monitoring", + "telemetry", "optional-components", "restricted-workloads", "wake-hostport", "network-isolation", "management-placement", "backup-preparation", "monitoring", } // ciSelection is the decoded output of one `ci_scope.py e2e` run. diff --git a/test/e2e/hostport_wake_e2e_test.go b/test/e2e/hostport_wake_e2e_test.go new file mode 100644 index 000000000..4cef3451b --- /dev/null +++ b/test/e2e/hostport_wake_e2e_test.go @@ -0,0 +1,241 @@ +//go:build e2e + +package e2e + +import ( + "context" + "encoding/json" + "fmt" + "io" + "net" + "net/http" + "strings" + "testing" + "time" + + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/types" +) + +// This bucket owns its two-node cluster. It deliberately retains terminating +// Pods and changes node scheduling; it must not run with other game tests. +func TestGameServer_HostportWakeStableEndpoint(t *testing.T) { + ctx := t.Context() + // The chart owns this namespace and scopes operator Secret access to it. + const ns, port = "gameplane-games", 12345 + name := fmt.Sprintf("hostport-wake-%d", time.Now().UnixNano()) + tmpl := name + "-template" + nodes, err := envInstance.K8s.CoreV1().Nodes().List(ctx, metav1.ListOptions{}) + if err != nil { + t.Fatal(err) + } + if len(nodes.Items) < 2 { + t.Fatal("wake-hostport requires the two-node kind profile") + } + var endpoint corev1.Node + for _, node := range nodes.Items { + if _, control := node.Labels["node-role.kubernetes.io/control-plane"]; !control { + endpoint = node + break + } + } + if endpoint.Name == "" { + t.Fatal("no worker node") + } + for _, node := range nodes.Items { + patch := []byte(`{"metadata":{"labels":{"gameplane.local/wake-pool":"test"}}}`) + if _, err := envInstance.K8s.CoreV1().Nodes().Patch(ctx, node.Name, types.MergePatchType, patch, metav1.PatchOptions{}); err != nil { + t.Fatal(err) + } + } + // The game and sentinel both need this dedicated-node toleration. + taints := append(append([]corev1.Taint{}, endpoint.Spec.Taints...), corev1.Taint{Key: "gameplane.local/dedicated", Value: "games", Effect: corev1.TaintEffectNoSchedule}) + patchHostportNode(t, endpoint.Name, map[string]any{"taints": taints}) + t.Cleanup(func() { + patchHostportNode(t, endpoint.Name, map[string]any{"unschedulable": false, "taints": endpoint.Spec.Taints}) + }) + address := "" + for _, addr := range endpoint.Status.Addresses { + if addr.Type == corev1.NodeInternalIP { + address = net.JoinHostPort(addr.Address, fmt.Sprint(port)) + break + } + } + if address == "" { + t.Fatal("worker has no InternalIP") + } + + applyBusyboxTemplate(t, tmpl) + // During deletion the shell keeps httpd alive for 12 seconds. This makes + // the old port genuinely occupied while the sleep handoff is observed. + templatePatch := map[string]any{"spec": map[string]any{ + "command": []string{"sh", "-c", "mkdir -p /tmp/site; echo endpoint-ok >/tmp/site/index.html; httpd -p 12345 -h /tmp/site; trap 'sleep 12; exit 0' TERM; while :; do sleep 1; done"}, + "ports": []any{map[string]any{"name": "game", "containerPort": port, "advertise": true, "protocol": "TCP", "wakeProtocol": "generic"}}, + }} + data, err := json.Marshal(templatePatch) + if err != nil { + t.Fatal(err) + } + if _, err := envInstance.Dyn.Resource(gameTemplateGVR).Patch(ctx, tmpl, types.MergePatchType, data, metav1.PatchOptions{}); err != nil { + t.Fatal(err) + } + gs := &unstructured.Unstructured{Object: map[string]any{ + "apiVersion": "gameplane.local/v1alpha1", "kind": "GameServer", + "metadata": map[string]any{"name": name, "namespace": ns}, + "spec": map[string]any{"templateRef": map[string]any{"name": tmpl}, "networking": map[string]any{"expose": "Hostport"}, + "nodeSelector": map[string]any{"kubernetes.io/hostname": endpoint.Labels["kubernetes.io/hostname"]}, + "tolerations": []any{map[string]any{"key": "gameplane.local/dedicated", "operator": "Equal", "value": "games", "effect": "NoSchedule"}, map[string]any{"key": "node-role.kubernetes.io/control-plane", "operator": "Exists", "effect": "NoSchedule"}}, + "idle": map[string]any{"enabled": true, "afterMinutes": int64(60), "wakeOnConnect": true}}, + }} + if _, err := envInstance.Dyn.Resource(gameServerGVR).Namespace(ns).Create(ctx, gs, metav1.CreateOptions{}); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + _ = envInstance.Dyn.Resource(gameServerGVR).Namespace(ns).Delete(context.Background(), name, metav1.DeleteOptions{}) + }) + waitPVCBound(t, ns, name+"-data", 90*time.Second) + httpClient := &http.Client{Timeout: 2 * time.Second, Transport: &http.Transport{DisableKeepAlives: true}} + defer httpClient.CloseIdleConnections() + gameReachable := func() (bool, string) { + request, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://"+address, nil) + if err != nil { + return false, err.Error() + } + resp, err := httpClient.Do(request) + if err != nil { + return false, err.Error() + } + defer resp.Body.Close() + body, err := io.ReadAll(resp.Body) + return err == nil && strings.Contains(string(body), "endpoint-ok"), string(body) + } + for cycle := range 2 { + envInstance.Eventually(t, 2*time.Minute, gameReachable) + pod, err := envInstance.K8s.CoreV1().Pods(ns).Get(ctx, name+"-0", metav1.GetOptions{}) + if err != nil || pod.Spec.NodeName != endpoint.Name { + t.Fatalf("game moved: pod=%v err=%v", pod, err) + } + patchHostportServer(t, ns, name, map[string]any{"metadata": map[string]any{"annotations": map[string]any{"gameplane.local/idle-asleep-since": time.Now().UTC().Format(time.RFC3339)}}}) + envInstance.Eventually(t, 20*time.Second, func() (bool, string) { + pod, err := envInstance.K8s.CoreV1().Pods(ns).Get(ctx, name+"-0", metav1.GetOptions{}) + return err == nil && !pod.DeletionTimestamp.IsZero(), fmt.Sprint(err) + }) + if ok, msg := gameReachable(); !ok { + t.Fatalf("fixture did not retain occupied old port during termination: %s", msg) + } + if _, err := envInstance.K8s.AppsV1().Deployments(ns).Get(ctx, name+"-waker", metav1.GetOptions{}); !apierrors.IsNotFound(err) { + t.Fatalf("sentinel created before old game released port: %v", err) + } + assertHostportCondition(t, ns, name, "GameDraining", false) + if cycle == 0 { + // Once the endpoint is recorded, broaden the worker pool to both + // nodes. Only the persisted required affinity can retain identity. + patchHostportServer(t, ns, name, map[string]any{"spec": map[string]any{"nodeSelector": map[string]any{"kubernetes.io/hostname": nil, "gameplane.local/wake-pool": "test"}}}) + patchHostportNode(t, endpoint.Name, map[string]any{"unschedulable": true}) + envInstance.Eventually(t, 90*time.Second, func() (bool, string) { return hostportConditionIs(t, ns, name, "EndpointUnschedulable", false) }) + // The alternate node is available; the listener must stay Pending + // on the pinned endpoint instead of silently moving there. + assertHostportPodsOnNode(t, ns, name, endpoint.Name) + patchHostportNode(t, endpoint.Name, map[string]any{"unschedulable": false}) + } + envInstance.Eventually(t, 90*time.Second, func() (bool, string) { return hostportConditionIs(t, ns, name, "ListenerReady", true) }) + assertHostportPodsOnNode(t, ns, name, endpoint.Name) + wakers, err := envInstance.K8s.CoreV1().Pods(ns).List(ctx, metav1.ListOptions{LabelSelector: "app.kubernetes.io/name=gameplane-waker,app.kubernetes.io/instance=" + name}) + if err != nil || len(wakers.Items) != 1 { + t.Fatalf("sentinel Pods: %v, %v", wakers, err) + } + waker := wakers.Items[0] + finalizer := []byte(`{"metadata":{"finalizers":["test.gameplane.local/hold-handoff"]}}`) + if _, err := envInstance.K8s.CoreV1().Pods(ns).Patch(ctx, waker.Name, types.MergePatchType, finalizer, metav1.PatchOptions{}); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + _, _ = envInstance.K8s.CoreV1().Pods(ns).Patch(context.Background(), waker.Name, types.MergePatchType, []byte(`{"metadata":{"finalizers":[]}}`), metav1.PatchOptions{}) + }) + // A connection to the same node address must reach the sentinel and + // wake the game. Hostport clients reconnect after this intentional bounce. + conn, err := (&net.Dialer{Timeout: 3 * time.Second}).DialContext(ctx, "tcp", address) + if err != nil { + t.Fatalf("sleeping endpoint unreachable: %v", err) + } + _ = conn.Close() + envInstance.Eventually(t, 30*time.Second, func() (bool, string) { return hostportConditionIs(t, ns, name, "SentinelDraining", false) }) + envInstance.Consistently(t, 3*time.Second, 200*time.Millisecond, func() (bool, string) { + _, err := envInstance.K8s.CoreV1().Pods(ns).Get(ctx, name+"-0", metav1.GetOptions{}) + return apierrors.IsNotFound(err), "game scheduled before terminating sentinel Pod disappeared" + }) + if _, err := envInstance.K8s.CoreV1().Pods(ns).Patch(ctx, waker.Name, types.MergePatchType, []byte(`{"metadata":{"finalizers":[]}}`), metav1.PatchOptions{}); err != nil { + t.Fatal(err) + } + envInstance.Eventually(t, 2*time.Minute, gameReachable) + assertHostportPodsOnNode(t, ns, name, endpoint.Name) + t.Logf("cycle %d kept reachable endpoint %s on %s", cycle+1, address, endpoint.Name) + } + // Real node deletion (including replacement under the same name) must + // report unavailable instead of migrating to the otherwise eligible node. + patchHostportServer(t, ns, name, map[string]any{"metadata": map[string]any{"annotations": map[string]any{"gameplane.local/idle-asleep-since": time.Now().UTC().Format(time.RFC3339)}}}) + if err := envInstance.K8s.CoreV1().Nodes().Delete(ctx, endpoint.Name, metav1.DeleteOptions{}); err != nil { + t.Fatal(err) + } + envInstance.Eventually(t, 45*time.Second, func() (bool, string) { return hostportConditionIs(t, ns, name, "EndpointNodeUnavailable", false) }) + assertHostportPodsOnNode(t, ns, name, endpoint.Name) +} + +func patchHostportServer(t *testing.T, ns, name string, patch map[string]any) { + t.Helper() + data, err := json.Marshal(patch) + if err != nil { + t.Fatal(err) + } + if _, err := envInstance.Dyn.Resource(gameServerGVR).Namespace(ns).Patch(context.Background(), name, types.MergePatchType, data, metav1.PatchOptions{}); err != nil { + t.Fatal(err) + } +} + +func patchHostportNode(t *testing.T, name string, spec map[string]any) { + t.Helper() + data, err := json.Marshal(map[string]any{"spec": spec}) + if err != nil { + t.Fatal(err) + } + if _, err := envInstance.K8s.CoreV1().Nodes().Patch(context.Background(), name, types.MergePatchType, data, metav1.PatchOptions{}); err != nil && !apierrors.IsNotFound(err) { + t.Error(err) + } +} + +func hostportConditionIs(t *testing.T, ns, name, reason string, ready bool) (bool, string) { + t.Helper() + gs, err := envInstance.Dyn.Resource(gameServerGVR).Namespace(ns).Get(t.Context(), name, metav1.GetOptions{}) + if err != nil { + return false, err.Error() + } + conditions, _, _ := unstructured.NestedSlice(gs.Object, "status", "conditions") + for _, item := range conditions { + condition, ok := item.(map[string]any) + if ok && condition["type"] == "HostportWakeReady" { + return condition["reason"] == reason && (condition["status"] == "True") == ready, fmt.Sprint(condition) + } + } + return false, "HostportWakeReady not yet present" +} + +func assertHostportCondition(t *testing.T, ns, name, reason string, ready bool) { + t.Helper() + envInstance.Eventually(t, 15*time.Second, func() (bool, string) { return hostportConditionIs(t, ns, name, reason, ready) }) +} + +func assertHostportPodsOnNode(t *testing.T, ns, name, node string) { + t.Helper() + pods, err := envInstance.K8s.CoreV1().Pods(ns).List(t.Context(), metav1.ListOptions{LabelSelector: "app.kubernetes.io/instance=" + name}) + if err != nil { + t.Fatal(err) + } + for _, pod := range pods.Items { + if pod.Spec.NodeName != "" && pod.Spec.NodeName != node { + t.Fatalf("listener %s moved to %s; endpoint remains %s", pod.Name, pod.Spec.NodeName, node) + } + } +} diff --git a/test/e2e/specs.md b/test/e2e/specs.md index 8e4bd861b..bc82ab948 100644 --- a/test/e2e/specs.md +++ b/test/e2e/specs.md @@ -61,6 +61,19 @@ complete pinned render inputs and source-template preservation for CSI restore. - Verify the installed operator identity can read Secrets only in its configured namespaces, can mutate game-namespace Secrets, and cannot read or write unrelated namespace Secrets (`TestHelmInstall_OperatorSecretNamespaceBoundary`, operator bucket). - Enforce, via `buckets.sh`, that every test in the suite is assigned to exactly one CI bucket so nothing silently drops out of coverage. +### Hostport wake acceptance + +The dedicated `wake-hostport` bucket runs on the `two-node` kind profile. It uses +a real Hostport HTTP listener and dedicated-node taint, broadens the scheduling +pool after recording the endpoint, and exercises two sleep/wake cycles at the +same node address. It observes the old game still serving during termination, +holds a terminating sentinel Pod to prove the reverse handoff waits, checks a +cordoned endpoint remains unavailable rather than relocating, and deletes the +node to exercise the explicit unavailable/replaced-node condition. This bucket +runs alone because it changes node scheduling. Its GameServer and template names +are unique per run; it uses the chart-owned game namespace so the operator's +configured Secret scope and RBAC apply. Runtime results come from CI. + ### Remote gateway parity acceptance The multicluster bucket runs `TestMultiCluster_GatewayParity` against two real @@ -120,10 +133,12 @@ The golangci-lint job mirrors this: it must be invoked with `--build-tags=e2e` o ## The bucket contract — the module's most important invariant -The `network-isolation` and `management-placement` audit buckets are registered in the change-based selector, -with explicit component mapping and amd64-only selection-summary rows. Each -matrix leg skips cluster setup only when its selected test pattern is empty; -full-scope and affected-component runs must include the affected buckets. +The `wake-hostport`, `network-isolation`, `monitoring`, `management-placement` +and `backup-preparation` audit buckets are registered in the change-based +selector with explicit component mapping and amd64-only selection-summary +rows. Each matrix leg skips cluster setup only when its selected test pattern +is empty; full-scope and affected-component runs must include the affected +buckets. `buckets.sh` is the single source of truth for how CI's e2e suite is sharded into parallel jobs (`operator`, `api-auth`, `api-roles`, `api-rbac`, `api-agent`, `api-mods`, `ratelimit`, `bot-fast`, `bot-heavy`, `multicluster`, `upgrade`). Each bucket is a flat list of **exact Go function names** (e.g. `TestGameServer_MinecraftJavaBot_Joined`), and `buckets.sh regex ` turns that list into an **anchored** `^(A|B|...)$` regex passed to `go test -run`. @@ -144,7 +159,7 @@ This is a name-matching contract, not a structural one: ## Conventions (already normative via `CLAUDE.md`, restated here for the module's own record) -- **`t.Parallel()`**: tests call it unless shared infrastructure requires serialization. The `ratelimit` bucket deliberately drains the shared login limiter; gateway parity temporarily changes the shared remote Helm release and trust; `restricted-workloads` upgrades the shared release and namespace admission policy; `postgres-availability` (`TestUpgrade_PostgresAvailability`) upgrades the shared release and interrupts PostgreSQL. These are documented exceptions. +- **`t.Parallel()`**: tests call it unless shared infrastructure requires serialization. The `ratelimit` bucket deliberately drains the shared login limiter; gateway parity temporarily changes the shared remote Helm release and trust; `wake-hostport` changes shared node scheduling and deletes the endpoint node; `restricted-workloads` upgrades the shared release and namespace admission policy; `postgres-availability` (`TestUpgrade_PostgresAvailability`) upgrades the shared release and interrupts PostgreSQL. These are documented exceptions. - **Per-test unique resource names**: CRs, namespaces, and users a test creates get a name unique to that test run (commonly time-based, e.g. `CreateUser`'s `fmt.Sprintf("%s-%d", prefix, time.Now().UnixNano())`) so parallel tests never collide on a shared cluster. - **`ociPushMu`**: a package-level mutex guarding tests that push through the shared, fixed-name `oras-push` Job (`module_e2e_test.go`, `module_verify_e2e_test.go`). Because the Job's name and namespace are fixed fixtures, two tests racing to apply/wait on it would stomp each other; `ociPushMu.Lock()`/`Unlock()` serializes them. - **`ensureResticRepo(t)`**: guards tests that run a backup against the shared restic repository fixture (`backup_e2e_test.go`, `backupschedule_e2e_test.go`, `failure_paths_e2e_test.go`) — restic repo initialization is not safe to race. diff --git a/test/e2e/suite-components.json b/test/e2e/suite-components.json index 9de35541b..6b20b3468 100644 --- a/test/e2e/suite-components.json +++ b/test/e2e/suite-components.json @@ -16,6 +16,7 @@ "test/e2e/suite-components.json" ], "suites": [ + { "tests": "^TestGameServer_HostportWakeStableEndpoint$", "components": ["operator", "api", "agent", "sentinel", "gameproto"] }, { "tests": "^TestNetworkIsolation_", "components": ["operator", "api", "agent"] }, { "tests": "^TestMonitoring_DiscoveryPerformanceAndPolicy$", "components": ["operator", "api", "agent", "tunnel"] }, { "tests": "^TestRestrictedWorkloads_", "components": ["operator", "api", "agent", "capture-sidecar"] },