diff --git a/.github/workflows/e2e-nightly.yaml b/.github/workflows/e2e-nightly.yaml index 9e4d306..183614c 100644 --- a/.github/workflows/e2e-nightly.yaml +++ b/.github/workflows/e2e-nightly.yaml @@ -1,6 +1,8 @@ name: E2E tests (nightly) -# Runs ONLY specs labeled `Nightly` (slow / destructive scaling cases, etc.). +# Two jobs on real Karpenter/NAP GPU nodes: +# - e2e-nightly-tests: Karpenter scale-from-zero specs (eastus). +# - e2e-nightly-scaling-tests: KEDA Scaling specs on H100 (australiaeast). # Triggers: # - schedule: every day at 00:00 Sydney time # - workflow_dispatch: manual run @@ -118,3 +120,84 @@ jobs: run: make e2e-teardown env: RESOURCE_GROUP: ${{ env.RESOURCE_GROUP }} + + # KEDA Scaling specs (1<->2, 0<->1) on real H100 nodes provisioned by + # Karpenter/NAP. Uses its own cluster in australiaeast (H100 quota) and runs + # after the Karpenter job so the shared self-hosted runner is not contended; + # `!cancelled()` keeps it independent of that job's result. + e2e-nightly-scaling-tests: + needs: e2e-nightly-tests + if: ${{ !cancelled() }} + runs-on: [ "self-hosted", "hostname:kaito-e2e-github-runner" ] + environment: e2e-test + # Scaling (240m) + cluster setup/teardown. + timeout-minutes: 330 + permissions: + contents: read + env: + # Short names keep "MC___australiaeast" within 80 chars. + RESOURCE_GROUP: "kaito-ns-rg-${{ github.run_id }}-${{ github.run_attempt }}" + CLUSTER_NAME: "kaito-ns-aks-${{ github.run_id }}-${{ github.run_attempt }}" + ACR_NAME: "kaitons${{ github.run_id }}a${{ github.run_attempt }}acr" + LOCATION: australiaeast + E2E_MODEL_STREAMING: 'true' + E2E_MODEL_STREAMING_NAMESPACE: e2e-scaling + + steps: + - name: Checkout Repository + uses: actions/checkout@v6 + with: + ref: ${{ github.event.inputs.branch || github.ref }} + + - name: E2E base setup + uses: ./.github/actions/e2e-base-setup + with: + resource-group: ${{ env.RESOURCE_GROUP }} + cluster-name: ${{ env.CLUSTER_NAME }} + acr-name: ${{ env.ACR_NAME }} + location: ${{ env.LOCATION }} + node-count: ${{ env.NODE_COUNT }} + node-vm-size: ${{ env.NODE_VM_SIZE }} + node-provisioner: karpenter + enable-node-mocker: false + node-class: azure + provider: ${{ github.event.inputs.provider }} + istio-version: ${{ github.event.inputs.istio_version }} + gateway-api-version: ${{ github.event.inputs.gateway_api_version }} + keda-version: ${{ github.event.inputs.keda_version }} + + - name: Run E2E tests (Scaling — real H100) + run: make test-e2e-scaling + env: + E2E_SCALING_INSTANCE_TYPE: Standard_NC40ads_H100_v5 + + - name: Generate Scaling coverage report + if: always() + run: | + go run ./hack/e2e/report \ + --label-filter 'Scaling' \ + --workflow 'E2E tests (nightly, scaling, real GPU)' \ + --results ginkgo-report-scaling.json \ + --output-md e2e-scaling-coverage-report.md \ + --output-html e2e-scaling-coverage-report.html + echo '## Scaling Nightly (real GPU) Coverage' >> "$GITHUB_STEP_SUMMARY" + cat e2e-scaling-coverage-report.md >> "$GITHUB_STEP_SUMMARY" + + - name: Upload Scaling coverage report + if: always() + uses: actions/upload-artifact@v4 + with: + name: e2e-nightly-scaling-coverage-report + path: | + e2e-scaling-coverage-report.md + e2e-scaling-coverage-report.html + + - name: Dump cluster state + if: failure() + run: make e2e-dump + + - name: Teardown cluster + if: always() + run: make e2e-teardown + env: + RESOURCE_GROUP: ${{ env.RESOURCE_GROUP }} diff --git a/.github/workflows/e2e-upstream-daily.yaml b/.github/workflows/e2e-upstream-daily.yaml index 623bde7..2e9b9cb 100644 --- a/.github/workflows/e2e-upstream-daily.yaml +++ b/.github/workflows/e2e-upstream-daily.yaml @@ -1,7 +1,8 @@ name: E2E tests (daily, upstream) # Runs the regular E2E suite against upstream KEDA and Istio with the GPU node -# mocker. Slow Nightly and Perf specs remain in their dedicated workflows. +# mocker, then the Scaling specs serially on the same cluster. Slow Nightly +# and Perf specs remain in their dedicated workflows. # Triggers: # - schedule: every day at 20:00 Sydney time # - workflow_dispatch: manual run @@ -50,6 +51,8 @@ jobs: e2e-upstream-daily-tests: runs-on: [ "self-hosted", "hostname:kaito-e2e-github-runner" ] environment: e2e-test + # main suite (60m) + Scaling (240m) + cluster setup/teardown. + timeout-minutes: 420 permissions: contents: read @@ -60,6 +63,7 @@ jobs: ref: ${{ github.event.inputs.branch || github.ref }} - name: E2E base setup + id: setup uses: ./.github/actions/e2e-base-setup with: resource-group: ${{ env.RESOURCE_GROUP }} @@ -77,13 +81,13 @@ jobs: - name: Run E2E tests run: make test-e2e env: - E2E_LABEL: "!Nightly && !Perf" + E2E_LABEL: "!Nightly && !Perf && !Scaling" - name: Generate E2E coverage report if: always() run: | go run ./hack/e2e/report \ - --label-filter '!Nightly && !Perf' \ + --label-filter '!Nightly && !Perf && !Scaling' \ --workflow 'E2E tests (daily, upstream)' \ --results ginkgo-report.json \ --output-md e2e-coverage-report.md \ @@ -100,6 +104,33 @@ jobs: e2e-coverage-report.md e2e-coverage-report.html + # Runs even when the main suite failed so a regression there does not + # hide scaling coverage. + - name: Run E2E tests (Scaling) + if: ${{ !cancelled() && steps.setup.outcome == 'success' }} + run: make test-e2e-scaling + + - name: Generate Scaling coverage report + if: ${{ !cancelled() && steps.setup.outcome == 'success' }} + run: | + go run ./hack/e2e/report \ + --label-filter 'Scaling' \ + --workflow 'E2E tests (daily, upstream, scaling)' \ + --results ginkgo-report-scaling.json \ + --output-md e2e-scaling-coverage-report.md \ + --output-html e2e-scaling-coverage-report.html + echo '## Upstream Daily Scaling Coverage' >> "$GITHUB_STEP_SUMMARY" + cat e2e-scaling-coverage-report.md >> "$GITHUB_STEP_SUMMARY" + + - name: Upload Scaling coverage report + if: ${{ !cancelled() && steps.setup.outcome == 'success' }} + uses: actions/upload-artifact@v4 + with: + name: e2e-upstream-scaling-coverage-report + path: | + e2e-scaling-coverage-report.md + e2e-scaling-coverage-report.html + - name: Dump cluster state if: failure() run: make e2e-dump diff --git a/.github/workflows/e2e.yaml b/.github/workflows/e2e.yaml index 56dce22..edef6c2 100644 --- a/.github/workflows/e2e.yaml +++ b/.github/workflows/e2e.yaml @@ -100,16 +100,16 @@ jobs: node-class: ${{ github.event.inputs.node_class }} - name: Run E2E tests - # Slow Nightly and Perf specs run in their dedicated workflows. + # Slow Nightly, Perf, and Scaling specs run in their dedicated workflows. run: make test-e2e env: - E2E_LABEL: "!Nightly && !Perf" + E2E_LABEL: "!Nightly && !Perf && !Scaling" - name: Generate E2E coverage report if: always() run: | go run ./hack/e2e/report \ - --label-filter '!Nightly && !Perf' \ + --label-filter '!Nightly && !Perf && !Scaling' \ --workflow 'E2E tests' \ --results ginkgo-report.json \ --output-md e2e-coverage-report.md \ diff --git a/Makefile b/Makefile index 362c751..100230b 100644 --- a/Makefile +++ b/Makefile @@ -164,6 +164,10 @@ test-e2e: ## Run e2e tests against a live cluster (requires KUBECONFIG). test-e2e-karpenter: ## Run the Karpenter nightly scale-from-zero scenarios with safe swedencentral defaults. $(MAKE) test-e2e E2E_LABEL='Karpenter' E2E_PARALLEL=1 E2E_TIMEOUT=180m +.PHONY: test-e2e-scaling +test-e2e-scaling: ## Run the KEDA Scaling specs (1<->2, 0<->1) serially. On real GPUs set E2E_SCALING_INSTANCE_TYPE (e.g. Standard_NC40ads_H100_v5). + $(MAKE) test-e2e E2E_LABEL='Scaling' E2E_PARALLEL=1 E2E_TIMEOUT=240m E2E_JSON_REPORT=ginkgo-report-scaling.json + .PHONY: test-e2e-perf test-e2e-perf: ## Run prefix-cache perf specs serially. Override E2E_TRACE_FIXTURE and E2E_PERF_TIMEOUT as needed. $(MAKE) test-e2e E2E_LABEL='Perf' E2E_PARALLEL=1 E2E_TIMEOUT=$(E2E_PERF_TIMEOUT) diff --git a/charts/productionstack/Chart.lock b/charts/productionstack/Chart.lock index 803d05f..e1d381e 100644 --- a/charts/productionstack/Chart.lock +++ b/charts/productionstack/Chart.lock @@ -4,7 +4,7 @@ dependencies: version: 1.5.0 - name: keda-kaito-scaler repository: "" - version: 0.7.0 + version: 0.7.1 - name: gpu-node-mocker repository: file://../gpu-node-mocker version: 0.4.3 @@ -17,5 +17,5 @@ dependencies: - name: productionstack-status-reporter repository: "" version: 0.4.3 -digest: sha256:e8025c947a3dcb700bd7ba42849b2df9c216af1304f46d5ee70ce92ab67913fd -generated: "2026-10-15T15:01:28.026668+11:00" +digest: sha256:094d638a9e4fc5b7139701282c617b3aff9edba3851da2f3067ca11d5bb06727 +generated: "2026-10-16T15:14:52.038649-07:00" diff --git a/charts/productionstack/Chart.yaml b/charts/productionstack/Chart.yaml index 81ff58d..2257036 100644 --- a/charts/productionstack/Chart.yaml +++ b/charts/productionstack/Chart.yaml @@ -43,7 +43,7 @@ dependencies: - gateway - routing - name: keda-kaito-scaler - version: 0.7.0 + version: 0.7.1 tags: - autoscaling # In-tree gpu-node-mocker chart. Sourced via a `file://` repository diff --git a/charts/productionstack/charts/keda-kaito-scaler/Chart.yaml b/charts/productionstack/charts/keda-kaito-scaler/Chart.yaml index e6bd2ab..9b5e4dd 100644 --- a/charts/productionstack/charts/keda-kaito-scaler/Chart.yaml +++ b/charts/productionstack/charts/keda-kaito-scaler/Chart.yaml @@ -15,10 +15,10 @@ type: application # This is the chart version. This version number should be incremented each time you make changes # to the chart and its templates, including the app version. # Versions are expected to follow Semantic Versioning (https://semver.org/) -version: 0.7.0 +version: 0.7.1 # This is the version number of the application being deployed. This version number should be # incremented each time you make changes to the application. Versions are not expected to # follow Semantic Versioning. They should reflect the version the application is using. # It is recommended to use it with quotes. -appVersion: "v0.7.0" \ No newline at end of file +appVersion: "v0.7.1" \ No newline at end of file diff --git a/charts/productionstack/charts/keda-kaito-scaler/README.md b/charts/productionstack/charts/keda-kaito-scaler/README.md index 5b4a3d7..5ae1595 100644 --- a/charts/productionstack/charts/keda-kaito-scaler/README.md +++ b/charts/productionstack/charts/keda-kaito-scaler/README.md @@ -126,7 +126,7 @@ Set `namespaceOverride: ""` to fall back to the standard Helm | `log.level` | `3` | klog verbosity (`--v`). Bump to `5` for debug logging. | | `image.registry` | `mcr.microsoft.com` | Container image registry. | | `image.repository` | `oss/v2/kaito/keda-kaito-scaler` | Container image repository. | -| `image.tag` | `v0.7.0` | Container image tag. | +| `image.tag` | `v0.7.1` | Container image tag. | | `image.pullSecrets` | `[]` | `imagePullSecrets` for the pod spec. | | `ports.grpc` | `10450` | gRPC `ExternalScaler` port exposed via the Service. | | `ports.metrics` | `10451` | Prometheus metrics port (`--metrics-port`). Not currently fronted by a Service. | diff --git a/charts/productionstack/charts/keda-kaito-scaler/values.yaml b/charts/productionstack/charts/keda-kaito-scaler/values.yaml index 26473ee..000c68b 100644 --- a/charts/productionstack/charts/keda-kaito-scaler/values.yaml +++ b/charts/productionstack/charts/keda-kaito-scaler/values.yaml @@ -21,7 +21,7 @@ namespaceOverride: "" image: registry: mcr.microsoft.com repository: oss/v2/kaito/keda-kaito-scaler - tag: v0.7.0 + tag: v0.7.1 pullSecrets: [] ports: diff --git a/hack/e2e/scripts/install-components.sh b/hack/e2e/scripts/install-components.sh index 693f74a..e0e5f78 100755 --- a/hack/e2e/scripts/install-components.sh +++ b/hack/e2e/scripts/install-components.sh @@ -127,6 +127,94 @@ prepare_managed_kaito_crds() { KAITO_HELM_ARGS+=(--skip-crds) } +retry_azure_identity_not_found() { + local attempt output exit_code + for attempt in 1 2 3 4 5 6; do + if output=$(az "$@" --only-show-errors 2>&1); then + printf '%s\n' "${output}" + return 0 + else + exit_code=$? + fi + printf '%s\n' "${output}" >&2 + if ! grep -Eq '^ERROR: \((Resource)?NotFound\)' <<< "${output}" || [[ "${attempt}" == 6 ]]; then + return "${exit_code}" + fi + echo "Identity operation not yet available (attempt ${attempt}/6); retrying in $((attempt * 10))s..." >&2 + sleep "$((attempt * 10))" + done +} + +prepare_model_streaming() { + local namespace="${E2E_MODEL_STREAMING_NAMESPACE:?Set E2E_MODEL_STREAMING_NAMESPACE}" + local storage_class="kaito-model-mirror" + local service_account="kaito-model-streamer" + local identity_name="kaito-model-streamer" + local storage_account identity_json identity_client_id identity_principal_id storage_scope oidc_issuer control_plane_id + local subscription="${AZURE_SUBSCRIPTION_ID:?Set AZURE_SUBSCRIPTION_ID}" + storage_account="kaitomm$(printf '%s' "${CLUSTER_NAME}" | openssl dgst -sha256 | awk '{print substr($NF,1,16)}')" + + echo "=== Preparing ModelMirror and ModelStreaming (${namespace}) ===" + echo " Enabling Blob CSI, OIDC issuer, and workload identity" + az aks update --subscription "${subscription}" \ + --resource-group "${RESOURCE_GROUP}" --name "${CLUSTER_NAME}" \ + --enable-blob-driver --enable-oidc-issuer --enable-workload-identity \ + --yes --only-show-errors --output none + echo " Creating model mirror storage account ${storage_account}" + az storage account create --subscription "${subscription}" \ + --resource-group "${RESOURCE_GROUP}" --name "${storage_account}" \ + --location "${LOCATION}" --sku Premium_LRS --kind BlockBlobStorage \ + --https-only true --allow-shared-key-access false --output none + echo " Creating streaming identity ${identity_name}" + identity_json=$(retry_azure_identity_not_found identity create --subscription "${subscription}" \ + --resource-group "${RESOURCE_GROUP}" --name "${identity_name}" --output json) + identity_client_id=$(jq -er '.clientId | select(type == "string" and length > 0)' <<< "${identity_json}") + identity_principal_id=$(jq -er '.principalId | select(type == "string" and length > 0)' <<< "${identity_json}") + storage_scope="/subscriptions/${subscription}/resourceGroups/${RESOURCE_GROUP}/providers/Microsoft.Storage/storageAccounts/${storage_account}" + echo " Granting streaming identity access to model blobs" + az role assignment create --subscription "${subscription}" \ + --assignee-object-id "${identity_principal_id}" --assignee-principal-type ServicePrincipal \ + --role "Storage Blob Data Contributor" --scope "${storage_scope}" --output none + echo " Granting AKS control-plane identity access to the storage account" + control_plane_id=$(az aks show --subscription "${subscription}" \ + --resource-group "${RESOURCE_GROUP}" --name "${CLUSTER_NAME}" --query identity.principalId --output tsv) + az role assignment create --subscription "${subscription}" \ + --assignee-object-id "${control_plane_id}" --assignee-principal-type ServicePrincipal \ + --role "Storage Account Contributor" --scope "${storage_scope}" --output none + oidc_issuer=$(az aks show --subscription "${subscription}" \ + --resource-group "${RESOURCE_GROUP}" --name "${CLUSTER_NAME}" --query oidcIssuerProfile.issuerUrl --output tsv) + echo " Creating streaming federated credential for ${namespace}/${service_account}" + retry_azure_identity_not_found identity federated-credential create --subscription "${subscription}" \ + --resource-group "${RESOURCE_GROUP}" --identity-name "${identity_name}" \ + --name model-streaming --issuer "${oidc_issuer}" \ + --subject "system:serviceaccount:${namespace}:${service_account}" \ + --audiences api://AzureADTokenExchange --output none + + echo " Configuring streaming service account and storage class" + kubectl create namespace "${namespace}" --dry-run=client --output yaml | kubectl apply -f - + kubectl create serviceaccount "${service_account}" --namespace "${namespace}" \ + --dry-run=client --output yaml | kubectl apply -f - + kubectl annotate serviceaccount "${service_account}" --namespace "${namespace}" \ + "azure.workload.identity/client-id=${identity_client_id}" --overwrite + jq -n --arg name "${storage_class}" --arg rg "${RESOURCE_GROUP}" \ + --arg account "${storage_account}" --arg client "${identity_client_id}" '{ + apiVersion: "storage.k8s.io/v1", kind: "StorageClass", metadata: {name: $name}, + provisioner: "blob.csi.azure.com", allowVolumeExpansion: true, reclaimPolicy: "Delete", + parameters: {resourceGroup: $rg, storageAccount: $account, protocol: "fuse2", + skuName: "Premium_LRS", clientID: $client, mountWithWorkloadIdentityToken: "true"}, + mountOptions: ["-o allow_other", "--block-cache", "--block-cache-block-size=16", + "--block-cache-pool-size=8192", "--block-cache-prefetch=11"] + }' | kubectl apply -f - + kubectl rollout status daemonset/csi-blob-node --namespace kube-system --timeout=300s + + KAITO_HELM_ARGS+=( + --set featureGates.ModelMirror=true + --set featureGates.ModelStreaming=true + --set "defaultModelMirrorStorageClass=${storage_class}" + --set "defaultStreamingServiceAccount=${service_account}" + ) +} + install_kaito() { echo "=== Installing KAITO workspace operator (image: nightly-latest) ===" @@ -139,6 +227,9 @@ install_kaito() { KAITO_HELM_ARGS=() run_provider_hook prepare_kaito_crds + if [[ "${E2E_MODEL_STREAMING:-false}" == "true" ]]; then + prepare_model_streaming + fi # Per-model GAIE artifacts are provisioned by charts/modeldeployment; enabling # the gate would render a duplicate set of resources via Flux and conflict. diff --git a/test/e2e/README.md b/test/e2e/README.md index 140c3ab..4cfe146 100644 --- a/test/e2e/README.md +++ b/test/e2e/README.md @@ -50,8 +50,12 @@ make test-e2e # Subset by Ginkgo label E2E_LABEL=Routing make test-e2e E2E_LABEL=PrefixCache make test-e2e +E2E_LABEL=ScaleToZero E2E_PARALLEL=1 make test-e2e E2E_LABEL=Smoke make test-e2e +# Scaling specs (Scaling label: 1<->2 and 0<->1), serial, 240m budget +make test-e2e-scaling + # Override parallelism E2E_PARALLEL=4 make test-e2e ``` @@ -59,7 +63,17 @@ E2E_PARALLEL=4 make test-e2e Labels live in [`utils/ginkgo.go`](utils/ginkgo.go) and fall into three groups: - **Cadence** (when a spec runs): `Smoke` (every PR), `Nightly` (long-running, nightly only). -- **Feature area** (what a spec verifies): `Infra`, `Routing`, `PrefixCache`, `Perf` (prefix-cache load/perf, see [below](#prefix-cache-perf--load-test)), `Auth`, `CORS`, `NetworkPolicy`, `Scaling` (scale-up / scale-down / anti-flapping), `InferenceSet`, `FilterOrder`, `Karpenter`, `Outage` (fail-closed / HA resilience). + `Scaling` specs are excluded from the main runs (`!Nightly && !Perf && !Scaling`) and run + through `make test-e2e-scaling`: on the GPU mocker (`Standard_NV36ads_A10_v5`) in the daily + upstream job, and on real `Standard_NC40ads_H100_v5` nodes (Karpenter/NAP, australiaeast) in + the nightly `e2e-nightly-scaling-tests` job, which sets + `E2E_SCALING_INSTANCE_TYPE=Standard_NC40ads_H100_v5`. That job also sets + `E2E_MODEL_STREAMING=true` and `E2E_MODEL_STREAMING_NAMESPACE=e2e-scaling`: + component setup provisions keyless Blob CSI storage and workload identity and then enables + ModelMirror and ModelStreaming. + on a local-NVMe PVC that a fake node cannot bind. The specs assert only traits common to both + backends (model pods, Nodes, NodeClaims, vLLM metrics), never mocker internals. +- **Feature area** (what a spec verifies): `Infra`, `Routing`, `PrefixCache`, `Perf` (prefix-cache load/perf, see [below](#prefix-cache-perf--load-test)), `Auth`, `CORS`, `NetworkPolicy`, `Scaling` (scale-up / scale-down / anti-flapping), `ScaleToZero` (full `0 -> 1 -> 2 -> 1 -> 0` lifecycle), `InferenceSet`, `FilterOrder`, `Karpenter`, `Outage` (fail-closed / HA resilience). - **Environment** (what kind of cluster a spec needs): `StandardK8sOnly`. `ModelDiscovery` marks the `/v1/models` listing, retrieval, and authentication @@ -82,7 +96,8 @@ E2E_LABEL='!StandardK8sOnly' make test-e2e ``` It currently covers `bbr_outage`, `ext_authz_outage`, `epp_outage`, -`cluster_filter_ha`, `cluster_status`, `control_plane_error`, `scaling`, and the +`cluster_filter_ha`, `cluster_status`, `control_plane_error`, `scaling`, +`scale_to_zero`, and the `Load distribution` Context of `model_routing` (which turns prefix-cache scoring off through `EPPScorerWeights` so identical prompts spread across replicas). It also covers `prefix_cache_routing`, `prefix_cache_perf`, the cross-model diff --git a/test/e2e/cases.go b/test/e2e/cases.go index 15a63fb..8f886ae 100644 --- a/test/e2e/cases.go +++ b/test/e2e/cases.go @@ -18,10 +18,12 @@ package e2e import ( "context" + "os" "time" . "github.com/onsi/ginkgo/v2" //nolint:revive // Ginkgo DSL . "github.com/onsi/gomega" //nolint:revive // Gomega DSL + "k8s.io/utils/ptr" "github.com/kaito-project/production-stack/test/e2e/deploy" "github.com/kaito-project/production-stack/test/e2e/utils" @@ -31,6 +33,19 @@ import ( // ModelDeploymentValues (e.g. EPPScorerWeights). func intPtr(v int) *int { return &v } +// envScalingInstanceType overrides the Scaling cases' SKU; the real-GPU +// nightly job sets it to Standard_NC40ads_H100_v5. +const envScalingInstanceType = "E2E_SCALING_INSTANCE_TYPE" + +// scalingInstanceType defaults to a non-NVMe SKU because, on NVMe SKUs, KAITO +// puts weights on a local-NVMe PVC that a mocker fake node cannot bind. +var scalingInstanceType = func() string { + if v := os.Getenv(envScalingInstanceType); v != "" { + return v + } + return "Standard_NV36ads_A10_v5" +}() + // Inference preset names. These are the underlying KAITO presets and become // the InferenceSet's spec.template.inference.preset.name. They do NOT appear // in the gateway's HTTPRoute matches — those use the deployment Name. @@ -353,27 +368,42 @@ var CaseDeployments = map[string][]deploy.ModelDeploymentValues{ // when the queue drains and clamps to minReplicaCount=1, so the // pool would never return to its starting size and Scale-Down // assertions would fail. - Name: "scaling-phi", - Namespace: "e2e-scaling", - Model: presetPhi, - Replicas: 1, - InstanceType: "Standard_NV36ads_A10_v5", - EnableScaling: true, - MinReplicas: 1, - MaxReplicas: 2, - // Single gauge signal on the vLLM waiting-queue length. The - // up-threshold is deliberately low so the pressure workload - // (scalingPressureConcurrency) pushes the per-replica queue - // above it and scale-up fires; the down-threshold sits below - // it so the sub-threshold workload (scalingSubThresholdConcurrency) - // drains the queue under it and scale-down converges back to - // the baseline. UpThreshold MUST be > DownThreshold (chart guard). + Name: "scaling-phi", + Namespace: "e2e-scaling", + Model: presetPhi, + Replicas: 1, + InstanceType: scalingInstanceType, + EnableScaling: true, + MinReplicas: 1, + MaxReplicas: 2, + ScaleDownStabilizationWindow: ptr.To(int64(300)), + ScaleUpPeriodSeconds: ptr.To(int64(300)), + ScaleDownPeriodSeconds: ptr.To(int64(300)), + // Caps concurrent sequences per pod so a modest workload builds + // a waiting queue even on an H100. The ConfigMap is created by + // the scaling spec before install. + Config: scalingVLLMConfigName, + // KEDA's scale-up gate needs readyReplicas, which KAITO only counts + // after the benchmark; on the mocker the benchmark can never finish. + DisableBenchmark: true, + // The bulk load repeats one prompt, so prefix-cache scoring pins it + // to the warm baseline pod and the scaled-up pod never serves (D1). + EPPScorerWeights: &deploy.EPPScorerWeights{PrefixCache: intPtr(0)}, + // Single gauge signal on the vLLM waiting-queue length. Per-pod + // waiting is capped around max-num-seqs (5): the EPP saturation + // detector (queueDepthThreshold 5) and KAITO's vLLM rate limiter + // (HTTP 429 once waiting > max_num_seqs) stop it growing further. + // The up-threshold therefore sits below that cap so the pressure + // workload (scalingPressureConcurrency) trips scale-up; the + // down-threshold sits below it so the sub-threshold workload + // (scalingSubThresholdConcurrency) lets scale-down converge back + // to the baseline. UpThreshold MUST be > DownThreshold (chart guard). ScalingMetrics: []deploy.ScalingMetric{ { Name: "vllm:num_requests_waiting", Type: "gauge", - UpThreshold: "10", - DownThreshold: "5", + UpThreshold: "3", + DownThreshold: "1", }, }, }, diff --git a/test/e2e/deploy/helm/helm.go b/test/e2e/deploy/helm/helm.go index 197ad48..726cf59 100644 --- a/test/e2e/deploy/helm/helm.go +++ b/test/e2e/deploy/helm/helm.go @@ -387,6 +387,9 @@ func setArgs(v deploy.ModelDeploymentValues) []string { if v.Config != "" { args = append(args, "--set-string", "config="+v.Config) } + if v.DisableBenchmark { + args = append(args, "--set", "enableBenchmark=false") + } if v.EnableScaling { args = append(args, "--set", "enableScaling=true") args = append(args, @@ -399,14 +402,40 @@ func setArgs(v deploy.ModelDeploymentValues) []string { if m.Type != "" { args = append(args, "--set", prefix+"type="+m.Type) } - args = append(args, - "--set", prefix+"upThreshold="+m.UpThreshold, - "--set", prefix+"downThreshold="+m.DownThreshold, - ) + if m.Source != "" { + args = append(args, "--set", prefix+"source="+m.Source) + } + if m.UpThreshold != "" { + args = append(args, "--set", prefix+"upThreshold="+m.UpThreshold) + } + if m.DownThreshold != "" { + args = append(args, "--set", prefix+"downThreshold="+m.DownThreshold) + } + if m.ActivationThreshold != "" { + args = append(args, "--set", prefix+"activationThreshold="+m.ActivationThreshold) + } + if m.DeactivationThreshold != "" { + args = append(args, "--set", prefix+"deactivationThreshold="+m.DeactivationThreshold) + } if m.MetricCacheWindow != "" { args = append(args, "--set", prefix+"metricCacheWindow="+m.MetricCacheWindow) } } + if v.CooldownPeriod > 0 { + args = append(args, "--set", "scaling.cooldownPeriod="+strconv.FormatInt(v.CooldownPeriod, 10)) + } + if v.InitialCooldownPeriod != nil { + args = append(args, "--set", "scaling.initialCooldownPeriod="+strconv.FormatInt(*v.InitialCooldownPeriod, 10)) + } + if v.ScaleDownStabilizationWindow != nil { + args = append(args, "--set", "scaling.hpa.scaleDown.stabilizationWindowSeconds="+strconv.FormatInt(*v.ScaleDownStabilizationWindow, 10)) + } + if v.ScaleUpPeriodSeconds != nil { + args = append(args, "--set", "scaling.hpa.scaleUp.periodSeconds="+strconv.FormatInt(*v.ScaleUpPeriodSeconds, 10)) + } + if v.ScaleDownPeriodSeconds != nil { + args = append(args, "--set", "scaling.hpa.scaleDown.periodSeconds="+strconv.FormatInt(*v.ScaleDownPeriodSeconds, 10)) + } } if v.AutoUpgrade.Enabled { args = append(args, "--set", "autoUpgrade.enabled=true") diff --git a/test/e2e/deploy/helm/helm_test.go b/test/e2e/deploy/helm/helm_test.go index 48b3447..aa38382 100644 --- a/test/e2e/deploy/helm/helm_test.go +++ b/test/e2e/deploy/helm/helm_test.go @@ -26,6 +26,8 @@ import ( "strings" "testing" + "k8s.io/utils/ptr" + "github.com/kaito-project/production-stack/test/e2e/deploy" ) @@ -284,6 +286,9 @@ func TestInstallModelDeploymentMapsValues(t *testing.T) { if !hasArg(args, "epp.scorerWeights.prefixCache=0") { t.Errorf("zero-valued prefixCache weight was dropped: %v", args) } + if _, found := argValue(args, "--set", "enableBenchmark="); found { + t.Errorf("enableBenchmark must inherit the chart default unless DisableBenchmark is set: %v", args) + } // kvCacheUtilization was nil and must fall through to the chart default. if _, found := argValue(args, "--set", "epp.scorerWeights.kvCacheUtilization="); found { t.Errorf("nil kvCacheUtilization must not be rendered: %v", args) @@ -308,6 +313,72 @@ func TestInstallModelDeploymentMapsValues(t *testing.T) { } } +func TestInstallModelDeploymentMapsScaleToZero(t *testing.T) { + d, calls := newTestDeployer(t, nil) + values := deploy.ModelDeploymentValues{ + Name: "phi", Namespace: "e2e-ns", Model: "phi-4-mini-instruct", + Replicas: 1, EnableScaling: true, MinReplicas: 0, MaxReplicas: 1, CooldownPeriod: 60, + DisableBenchmark: true, + InitialCooldownPeriod: ptr.To(int64(0)), + ScaleDownStabilizationWindow: ptr.To(int64(300)), + ScaleUpPeriodSeconds: ptr.To(int64(300)), + ScaleDownPeriodSeconds: ptr.To(int64(300)), + } + + if err := d.InstallModelDeployment(context.Background(), values); err != nil { + t.Fatalf("InstallModelDeployment: %v", err) + } + args := (*calls)[0] + for _, want := range []string{ + "replicas=1", "minReplicas=0", "maxReplicas=1", "scaling.cooldownPeriod=60", "enableBenchmark=false", + "scaling.initialCooldownPeriod=0", "scaling.hpa.scaleDown.stabilizationWindowSeconds=300", + "scaling.hpa.scaleUp.periodSeconds=300", "scaling.hpa.scaleDown.periodSeconds=300", + } { + if !hasArg(args, want) { + t.Errorf("missing --set %q in %v", want, args) + } + } + if hasArg(args, "scaling.metrics[0].name=") { + t.Errorf("chart-default metrics must not be overridden: %v", args) + } +} + +func TestUpgradeModelDeploymentEnablesCapacityWithoutRemovingZeroScaling(t *testing.T) { + d, calls := newTestDeployer(t, nil) + values := deploy.ModelDeploymentValues{ + Name: "phi", Namespace: "e2e-ns", Model: "phi-4-mini-instruct", + Replicas: 1, EnableScaling: true, MinReplicas: 0, MaxReplicas: 1, CooldownPeriod: 60, + InitialCooldownPeriod: ptr.To(int64(60)), + ScalingMetrics: []deploy.ScalingMetric{ + {Name: "vllm:num_requests_waiting", Type: "gauge", Source: "modelpod", UpThreshold: "3", DownThreshold: "1", DeactivationThreshold: "0"}, + {Name: "llm_d_epp_flow_control_queue_size", Type: "gauge", Source: "epp", ActivationThreshold: "10", DeactivationThreshold: "0"}, + {Name: "vllm:num_requests_running", Type: "gauge", Source: "modelpod", DeactivationThreshold: "0"}, + }, + } + for _, maximum := range []int64{1, 2} { + values.MaxReplicas = maximum + if err := d.UpgradeModelDeployment(context.Background(), values); err != nil { + t.Fatalf("UpgradeModelDeployment (maxReplicas=%d): %v", maximum, err) + } + args := (*calls)[len(*calls)-1] + wantMaximum := "maxReplicas=1" + if maximum == 2 { + wantMaximum = "maxReplicas=2" + } + for _, want := range []string{ + wantMaximum, "replicas=1", "minReplicas=0", "enableScaling=true", + "scaling.cooldownPeriod=60", "scaling.initialCooldownPeriod=60", + "scaling.metrics[0].upThreshold=3", "scaling.metrics[0].downThreshold=1", + "scaling.metrics[1].activationThreshold=10", "scaling.metrics[1].deactivationThreshold=0", + "scaling.metrics[2].name=vllm:num_requests_running", "scaling.metrics[2].deactivationThreshold=0", + } { + if !hasArg(args, want) { + t.Errorf("maxReplicas=%d: missing --set %q in %v", maximum, want, args) + } + } + } +} + func TestInstallValidatesBeforeInvokingHelm(t *testing.T) { tests := []struct { name string @@ -324,18 +395,68 @@ func TestInstallValidatesBeforeInvokingHelm(t *testing.T) { values: deploy.ModelDeploymentValues{Name: "phi"}, wantErr: "Model is required", }, - { - name: "scaling without metrics", - values: deploy.ModelDeploymentValues{Name: "phi", Model: "phi", EnableScaling: true}, - wantErr: "at least one ScalingMetric", - }, { name: "metric without thresholds", values: deploy.ModelDeploymentValues{ - Name: "phi", Model: "phi", EnableScaling: true, + Name: "phi", Model: "phi", EnableScaling: true, MinReplicas: 1, MaxReplicas: 2, ScalingMetrics: []deploy.ScalingMetric{{Name: "vllm:num_requests_waiting"}}, }, - wantErr: "UpThreshold is required", + wantErr: "requires UpThreshold/DownThreshold, ActivationThreshold or DeactivationThreshold", + }, + { + name: "positive minReplicas with only zero-edge metrics", + values: deploy.ModelDeploymentValues{ + Name: "phi", Model: "phi", EnableScaling: true, MinReplicas: 1, MaxReplicas: 2, + ScalingMetrics: []deploy.ScalingMetric{{Name: "queue", Source: "epp", ActivationThreshold: "0", DeactivationThreshold: "0"}}, + }, + wantErr: "requires at least one metric with UpThreshold/DownThreshold", + }, + { + name: "cooldown with positive minReplicas", + values: deploy.ModelDeploymentValues{ + Name: "phi", Model: "phi", EnableScaling: true, MinReplicas: 1, MaxReplicas: 2, CooldownPeriod: 60, + }, + wantErr: "CooldownPeriod requires MinReplicas=0", + }, + { + name: "initial cooldown with positive minimum", + values: deploy.ModelDeploymentValues{ + Name: "phi", Model: "phi", EnableScaling: true, MinReplicas: 1, MaxReplicas: 2, + InitialCooldownPeriod: ptr.To(int64(0)), + }, + wantErr: "InitialCooldownPeriod requires MinReplicas=0", + }, + { + name: "invalid stabilization window", + values: deploy.ModelDeploymentValues{ + Name: "phi", Model: "phi", EnableScaling: true, MinReplicas: 1, MaxReplicas: 2, + ScaleDownStabilizationWindow: ptr.To(int64(3601)), + }, + wantErr: "ScaleDownStabilizationWindow must be between 0 and 3600", + }, + { + name: "invalid scale-up policy period", + values: deploy.ModelDeploymentValues{ + Name: "phi", Model: "phi", EnableScaling: true, MinReplicas: 1, MaxReplicas: 2, + ScaleUpPeriodSeconds: ptr.To(int64(0)), + }, + wantErr: "ScaleUpPeriodSeconds must be between 1 and 1800", + }, + { + name: "zero without activation metric", + values: deploy.ModelDeploymentValues{ + Name: "phi", Model: "phi", EnableScaling: true, MaxReplicas: 1, + ScalingMetrics: []deploy.ScalingMetric{{Name: "running", Source: "modelpod", DeactivationThreshold: "0"}}, + }, + wantErr: "requires at least one EPP activation metric", + }, + { + name: "zero without modelpod deactivation metric", + values: deploy.ModelDeploymentValues{ + Name: "phi", Model: "phi", EnableScaling: true, MaxReplicas: 1, + ScalingMetrics: []deploy.ScalingMetric{{Name: "queue", Source: "epp", ActivationThreshold: "0", DeactivationThreshold: "0"}}, + }, + wantErr: "requires at least one modelpod deactivation metric", }, { name: "negative minReplicas", diff --git a/test/e2e/deploy/values.go b/test/e2e/deploy/values.go index 68c1c8c..a89c1f0 100644 --- a/test/e2e/deploy/values.go +++ b/test/e2e/deploy/values.go @@ -147,10 +147,21 @@ type ModelDeploymentValues struct { MaxReplicas int64 // ScalingMetrics is the ordered list of composite scaling signals wired // onto the modeldeployment chart's scaling.metrics[] entries. Only - // used when EnableScaling is true; at least one entry is required in - // that case (the chart rejects an empty metrics list). Each entry's + // used when EnableScaling is true. Empty inherits the chart's default + // scale-to-zero and positive-replica metric policy. Each custom entry's // UpThreshold MUST be strictly greater than its DownThreshold. ScalingMetrics []ScalingMetric + // CooldownPeriod is KEDA's idle cooldown in seconds before parking at + // zero. Only valid with MinReplicas=0; zero keeps the chart default. + CooldownPeriod int64 + // Nil inherits scaler defaults; pointers allow an explicit zero. + InitialCooldownPeriod *int64 + ScaleDownStabilizationWindow *int64 + ScaleUpPeriodSeconds *int64 + ScaleDownPeriodSeconds *int64 + // DisableBenchmark sets enableBenchmark=false so KAITO skips its + // post-load benchmark. Zero value keeps the chart default (enabled). + DisableBenchmark bool // AutoUpgrade opts the InferenceSet into KAITO automatic base image // upgrades, wired onto the modeldeployment chart's autoUpgrade.* values // (rendered as spec.autoUpgrade). Only rendered when Enabled is true. @@ -199,12 +210,21 @@ type ScalingMetric struct { // average) or "histogram" (per-pod windowed average) (metrics entry // `type`). Empty defaults to gauge. Type string + // Source selects the metric endpoint: "modelpod" or "epp". Empty + // preserves the scaler default. + Source string // UpThreshold is the per-replica scale-up threshold (metrics entry // `upthreshold`). Required; MUST be strictly greater than DownThreshold. UpThreshold string // DownThreshold is the per-replica scale-down threshold (metrics entry // `downthreshold`). Required; MUST be strictly less than UpThreshold. DownThreshold string + // ActivationThreshold is the EPP threshold that wakes a zero-replica + // deployment. It requires DeactivationThreshold and Source="epp". + ActivationThreshold string + // DeactivationThreshold is the zero-edge inactive threshold. Model-pod + // metrics may use it without an activation threshold. + DeactivationThreshold string // MetricCacheWindow is the rolling cache window in seconds for histogram // metrics (metrics entry `metriccachewindow`). Optional; ignored for gauge. MetricCacheWindow string @@ -224,28 +244,78 @@ func (v ModelDeploymentValues) Validate() error { if !v.EnableScaling { return nil } + if v.MinReplicas < 0 { + return fmt.Errorf("modeldeployment %q: MinReplicas must not be negative (got %d)", v.Name, v.MinReplicas) + } + if v.MaxReplicas < 1 { + return fmt.Errorf("modeldeployment %q: EnableScaling requires MaxReplicas >= 1 (got %d)", v.Name, v.MaxReplicas) + } + if v.MaxReplicas < v.MinReplicas { + return fmt.Errorf("modeldeployment %q: MaxReplicas (%d) must be >= MinReplicas (%d)", v.Name, v.MaxReplicas, v.MinReplicas) + } + // The chart silently ignores cooldownPeriod unless minReplicas is 0. + if v.MinReplicas > 0 && v.CooldownPeriod > 0 { + return fmt.Errorf("modeldeployment %q: CooldownPeriod requires MinReplicas=0", v.Name) + } + if v.InitialCooldownPeriod != nil { + if v.MinReplicas != 0 { + return fmt.Errorf("modeldeployment %q: InitialCooldownPeriod requires MinReplicas=0", v.Name) + } + if *v.InitialCooldownPeriod < 0 || *v.InitialCooldownPeriod > 2147483647 { + return fmt.Errorf("modeldeployment %q: InitialCooldownPeriod must be between 0 and 2147483647", v.Name) + } + } + if v.ScaleDownStabilizationWindow != nil && (*v.ScaleDownStabilizationWindow < 0 || *v.ScaleDownStabilizationWindow > 3600) { + return fmt.Errorf("modeldeployment %q: ScaleDownStabilizationWindow must be between 0 and 3600", v.Name) + } + for name, period := range map[string]*int64{"ScaleUpPeriodSeconds": v.ScaleUpPeriodSeconds, "ScaleDownPeriodSeconds": v.ScaleDownPeriodSeconds} { + if period != nil && (*period < 1 || *period > 1800) { + return fmt.Errorf("modeldeployment %q: %s must be between 1 and 1800", v.Name, name) + } + } if len(v.ScalingMetrics) == 0 { - return fmt.Errorf("modeldeployment %q: EnableScaling requires at least one ScalingMetric", v.Name) + return nil } + hasActivationMetric := false + hasModelPodDeactivationMetric := false + hasScalingMetric := false for i, m := range v.ScalingMetrics { if m.Name == "" { return fmt.Errorf("modeldeployment %q: ScalingMetrics[%d].Name is required", v.Name, i) } - if m.UpThreshold == "" { - return fmt.Errorf("modeldeployment %q: ScalingMetrics[%d] (%s) UpThreshold is required", v.Name, i, m.Name) + hasUp := m.UpThreshold != "" + hasDown := m.DownThreshold != "" + if hasUp != hasDown { + return fmt.Errorf("modeldeployment %q: ScalingMetrics[%d] (%s) UpThreshold and DownThreshold must be supplied together", v.Name, i, m.Name) + } + if m.ActivationThreshold != "" { + if m.DeactivationThreshold == "" { + return fmt.Errorf("modeldeployment %q: ScalingMetrics[%d] (%s) ActivationThreshold requires DeactivationThreshold", v.Name, i, m.Name) + } + if m.Source != "epp" { + return fmt.Errorf("modeldeployment %q: ScalingMetrics[%d] (%s) ActivationThreshold requires Source=epp", v.Name, i, m.Name) + } + hasActivationMetric = true + } + if !hasUp && m.ActivationThreshold == "" && m.DeactivationThreshold == "" { + return fmt.Errorf("modeldeployment %q: ScalingMetrics[%d] (%s) requires UpThreshold/DownThreshold, ActivationThreshold or DeactivationThreshold", v.Name, i, m.Name) + } + if hasUp { + hasScalingMetric = true } - if m.DownThreshold == "" { - return fmt.Errorf("modeldeployment %q: ScalingMetrics[%d] (%s) DownThreshold is required", v.Name, i, m.Name) + if m.Source == "modelpod" && m.DeactivationThreshold != "" { + hasModelPodDeactivationMetric = true } } - if v.MinReplicas < 0 { - return fmt.Errorf("modeldeployment %q: MinReplicas must not be negative (got %d)", v.Name, v.MinReplicas) + // The chart drops zero-edge-only metrics when MinReplicas > 0. + if v.MinReplicas > 0 && !hasScalingMetric { + return fmt.Errorf("modeldeployment %q: MinReplicas > 0 requires at least one metric with UpThreshold/DownThreshold", v.Name) } - if v.MaxReplicas < 1 { - return fmt.Errorf("modeldeployment %q: EnableScaling requires MaxReplicas >= 1 (got %d)", v.Name, v.MaxReplicas) + if v.MinReplicas == 0 && !hasActivationMetric { + return fmt.Errorf("modeldeployment %q: scale-to-zero requires at least one EPP activation metric", v.Name) } - if v.MaxReplicas < v.MinReplicas { - return fmt.Errorf("modeldeployment %q: MaxReplicas (%d) must be >= MinReplicas (%d)", v.Name, v.MaxReplicas, v.MinReplicas) + if v.MinReplicas == 0 && !hasModelPodDeactivationMetric { + return fmt.Errorf("modeldeployment %q: scale-to-zero requires at least one modelpod deactivation metric", v.Name) } return nil } diff --git a/test/e2e/production-stack-E2E-test-scenarios.md b/test/e2e/production-stack-E2E-test-scenarios.md index 60976cd..29a41fd 100644 --- a/test/e2e/production-stack-E2E-test-scenarios.md +++ b/test/e2e/production-stack-E2E-test-scenarios.md @@ -8,7 +8,8 @@ | **Infrastructure** | Fake Node / Shadow Pod Lifecycle | PR | gpu-node-mocker two-phase pipeline validation | | | InferenceSet / InferencePool Lifecycle | PR | KAITO controller and routing infrastructure health | | | NodeClaim / Fake Node Cleanup | PR | Reverse-path teardown and reconcile idempotency | -| | InferenceSet Scaling (E2E) | Nightly | End-to-end scale up/down driven by KEDA + traffic verification | +| | InferenceSet Scaling (E2E) | Daily (mocker) + Nightly (H100) | End-to-end scale up/down driven by KEDA + traffic verification | +| | InferenceSet Scale To Zero | Daily (mocker) + Nightly (H100) | EPP-driven `1 -> 0 -> 1 -> 0` activation and post-cold-start serving | | **Routing** | Model-Based Routing | PR | BBR → HTTPRoute → EPP → Pod routing correctness | | | Unknown Model / Malformed Request Handling | PR | Catch-all 404 JSON + BBR malformed-body behaviour | | **Performance** | Prefix Cache Aware Routing | PR (core) + Nightly (full) | EPP prefix-cache scorer and KV-cache locality | @@ -186,43 +187,54 @@ activity. ## InferenceSet Scaling — Infra -**CI tier:** Nightly. KEDA polling interval + cooldown windows make the full Scale-Up/Scale-Down loop -too slow (several minutes each) and timing-sensitive for PR CI; Anti-Flapping in particular needs a -full cooldown window to be meaningful. +**CI tier:** `Scaling` label, run serially by `make test-e2e-scaling` — on the GPU mocker +(`Standard_NV36ads_A10_v5`) in the daily upstream job and on real `Standard_NC40ads_H100_v5` nodes +in the nightly `e2e-nightly-scaling-tests` job (`E2E_SCALING_INSTANCE_TYPE`). KEDA polling + cooldown windows (and real GPU provisioning) make the +full loop too slow for PR CI; Anti-Flapping in particular needs a full cooldown window. Verifies the system reacts correctly to **workload-driven** scaling by exercising the full end-to-end -chain: rising request queue → KEDA triggers InferenceSet replica change → gpu-node-mocker provisions or -removes fake nodes and shadow pods → EPP updates its endpoint list → traffic redistributes. +chain: rising request queue → KEDA triggers InferenceSet replica change → Karpenter (or the GPU mocker) +provisions or removes GPU nodes → EPP updates its endpoint list → traffic redistributes. -The KEDA scaling trigger is the annotation-configured metric `vllm:num_requests_waiting` with threshold -`10` (see `scaledobject.kaito.sh/*` annotations on the InferenceSet). Each phase of the flow is verified -individually so failures can be localised. +Assertions use only traits common to the mocker and real GPUs: model pods (selected by +`inferenceset.kaito.sh/created-by`), Nodes, NodeClaims, and `vllm:*` metrics scraped from model pods. +The case caps concurrency with `max-num-seqs: 5` and uses up/down thresholds `3`/`1`, so 40 concurrent +bounded (`max_tokens: 128`) requests saturate one H100 replica. Each phase is verified individually so +failures can be localised. ### Scale-Up — End-to-End -* Record baseline — Capture initial `InferenceSet.spec.replicas` (e.g. 2), per-pod `vllm:request_success_total`, `inference_pool_ready_pods{name}`, and fake-node/shadow-pod counts. Establishes a clean starting point. -* Generate queue pressure — Send concurrent requests at a rate exceeding serving capacity so that `vllm:num_requests_waiting` rises above the KEDA threshold (10) on at least one pod. Verify by scraping shadow pod metrics. This is the trigger condition for scale-up. -* KEDA triggers replica increase — Poll the InferenceSet and verify `.spec.replicas` increases (e.g. 2 → 3) within the KEDA polling interval. Proves the ScaledObject correctly read the vLLM metric and patched the InferenceSet. -* New fake node provisioned — Verify a new fake node appears with the correct labels and `Ready=True` status. Proves gpu-node-mocker Phase 1 (NodeClaimReconciler) handled the new NodeClaim produced by Karpenter. -* New shadow pod running — Verify a new shadow pod reaches `Running` in the model's namespace with both `inference-sim` and `tokenizer` containers ready. Proves gpu-node-mocker Phase 2 (ShadowPodReconciler) completed the shadow-pod creation path. -* Original pod patched to Running — Verify the newly scheduled original pod's status is patched to `Running` with the shadow pod's real CNI IP as `PodIP`. Proves the status-patch pipeline works for dynamically scaled pods, not just initially deployed ones. -* InferencePool endpoint list updated — Verify `inference_pool_ready_pods{name}` increases to match the new replica count. If InferencePool doesn't learn about the new pod, EPP cannot route to it and the new replica is wasted. +* Record baseline — Capture initial `InferenceSet.spec.replicas` (e.g. 1), the Ready model pods, and the NodeClaims backing their nodes. Establishes a clean starting point. +* Generate queue pressure — Send concurrent requests at a rate exceeding serving capacity so that `vllm:num_requests_waiting` rises above the KEDA threshold (3) on at least one model pod. This is the trigger condition for scale-up. +* KEDA triggers replica increase — Poll the InferenceSet and verify `.spec.replicas` increases (e.g. 1 → 2). Proves the ScaledObject correctly read the vLLM metric and patched the InferenceSet. +* New model pod on a new GPU node — Verify a model pod not in the baseline becomes Ready on a new Ready node carrying the case's `node.kubernetes.io/instance-type`, allocatable `nvidia.com/gpu` ≥ 1, and a NodeClaim whose `status.nodeName` is that node. +* Ready model pods match replicas — Verify the Ready model-pod count reaches the new replica count, so EPP has an endpoint for every replica. * New pod receives traffic — Continue sending requests. Scrape `vllm:request_success_total` from the new pod and verify it is > 0 within a reasonable window. The new replica must actually serve traffic, not just exist on paper. (General load-distribution fairness is covered by *Model-Based Routing → Load distribution*; not duplicated here.) ### Scale-Down — End-to-End * Stop traffic and drain queue — Cease all inbound requests. Wait for `vllm:num_requests_waiting` to reach 0 on every pod and `vllm:num_requests_running` to reach 0. The queue must fully drain before KEDA begins scale-down. * KEDA triggers replica decrease — Poll the InferenceSet and verify `.spec.replicas` decreases (e.g. 3 → 2) after the KEDA cooldown period elapses. Proves the scale-down path is not silently stuck. -* Excess shadow pod and fake node cleaned up — Verify the extra shadow pod is deleted from the model's namespace, the corresponding fake node is removed, and the NodeClaim is released. Proves the cleanup pipeline (ShadowPodReconciler → NodeClaim release, including the lease-renewal goroutine stop) works. -* InferencePool endpoint list shrinks — Verify `inference_pool_ready_pods{name}` decreases to match the new replica count. A stale endpoint would cause EPP to route to a non-existent pod and requests would fail. +* Excess pod, NodeClaim, and node cleaned up — Every model pod seen during scale-up that is no longer current is gone, and each node that no longer hosts a model pod has its NodeClaim and then its Node removed. Which replica KAITO removes is nondeterministic, so identity is not asserted. +* Ready model pods shrink — Verify the Ready model-pod count returns to the baseline replica count. * Remaining pods continue to serve traffic — Resume sending requests. Verify all requests succeed (HTTP 200). Scrape `vllm:request_success_total` on the remaining pods to confirm they handle the load. `inference_extension_scheduler_attempts_total{status="failure"}` must not increment. -* No request failures during transition — Send a low-rate stream of requests throughout the scale-down window. Verify `inference_objective_request_error_total` does not increment and no HTTP 5xx responses are observed. Proves the system handles pod removal gracefully without dropping in-flight requests. +* No request failures during transition — Send back-to-back long-response requests with one concurrent worker throughout the scale-down window. Verify successful responses and no HTTP 5xx responses are observed. Proves the system handles pod removal gracefully without dropping in-flight requests. ### Anti-Flapping -* Below-threshold stability — Hold `vllm:num_requests_waiting` strictly below the KEDA threshold for the full polling + cooldown window. Verify `InferenceSet.spec.replicas` does **not** change and no new NodeClaim/fake-node/shadow-pod is created. Guards against false-positive scale-ups from noisy metrics. +* Below-threshold stability — Hold `vllm:num_requests_waiting` strictly below the KEDA threshold for the full polling + cooldown window. Verify `InferenceSet.spec.replicas` does **not** change, no model pod outside the baseline appears, and no new NodeClaim is created. Guards against false-positive scale-ups from noisy metrics. * Cooldown respected — Immediately after a Scale-Down completes, re-apply queue pressure above the threshold. Verify the next Scale-Up does **not** fire until the ScaledObject's cooldown has elapsed and that no intermediate scale oscillations are observed. Flapping is the top production pain point for autoscalers and must be explicitly pinned. +### Scale To Zero — Full Scaling Lifecycle + +**CI tier:** same jobs as the Scaling group (`make test-e2e-scaling`). Run alone with `E2E_LABEL='ScaleToZero && StandardK8sOnly' E2E_PARALLEL=1`. + +Warm-bootstrap one replica (GPU-mocked or real H100), then upgrade the same release to a zero-minimum, one-maximum ScaledObject with EPP flow control. Verify the idle model parks and releases its GPU NodeClaim while EPP remains Ready. Sustained authenticated requests must wake one replica and keep it through provisioning; cold-start delivery is best effort, with eventual success and bounded 5xx responses. The temporary one-replica cap prevents cold-start pressure from triggering capacity scaling before the baseline is recorded. + +The E2E explicitly sets HPA scale-down stabilization and both scaling policy periods to 300 seconds, with initial and ordinary zero cooldowns of 60 seconds. These test overrides do not change production defaults. After successful cold-start traffic, start the single long-request background worker before stopping cold-start pressure. Wait for the backlog to drain, background requests to succeed, and KEDA's active timestamp to advance before requiring a successful chat probe and recording the one-replica baseline. Only after A1 establishes bulk queue pressure does the test raise the maximum to two and verify the ScaledObject reconciles. Unexpected parking during this handoff or capacity scaling fails immediately. Timing budgets read initial cooldown, stabilization, and rate-limit periods independently from the ScaledObject. + +On that same deployment, run the Scale-Up and Scale-Down assertions above: queue pressure, replica increase, new GPU node and NodeClaim, routing to the new pod, queue drain, replica decrease, inventory cleanup, and no 5xx during the two-to-one transition. A single background worker sends back-to-back long-response requests (up to 1,024 tokens each) to keep the remaining replica active without building a waiting queue. Stop that worker, verify the deployment returns to zero, and verify all case-created NodeClaims are released while EPP remains Ready. Anti-flapping runs separately after restoring a one-replica minimum. + ## Unknown model / malformed request handling — Routing **CI tier:** PR (required). Negative-path API contract, fast and deterministic. diff --git a/test/e2e/scaling_spec.go b/test/e2e/scaling_spec.go index af9df10..e9d39a3 100644 --- a/test/e2e/scaling_spec.go +++ b/test/e2e/scaling_spec.go @@ -19,6 +19,8 @@ package e2e import ( "context" "fmt" + "maps" + "slices" "strings" "time" @@ -27,6 +29,7 @@ import ( 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" "github.com/kaito-project/production-stack/test/e2e/deploy" "github.com/kaito-project/production-stack/test/e2e/utils" @@ -40,92 +43,249 @@ import ( // of truth for the model name carried in OpenAI requests and matched by // the gateway. The shared gateway request helpers authenticate both probes // and LoadGenerator traffic using this namespace's backend credentials. +// +// The specs run unchanged on the GPU node mocker and on real GPU nodes, so +// they observe a scale event only through objects both environments share: +// the InferenceSet's model pods, the nodes those pods run on, and the +// Karpenter NodeClaims backing those nodes (see utils/scaling.go). const ( - // Concurrency for queue-pressure workloads. The CaseScaling baseline - // is 1 replica with the gauge up-threshold=10 and the inference - // simulator's max-num-seqs=5, so the single pod has 5 running slots - // and any extra in-flight requests pile into the queue. Setting - // concurrency to 40 leaves ~35 requests waiting on that pod — well - // above the up-threshold — guaranteeing scale-up fires. + // Concurrency for queue-pressure workloads. CaseScaling runs 1 replica + // with max-num-seqs=5 (scalingVLLMConfig), so 5 requests run and the + // rest queue. The EPP saturation detector and KAITO's vLLM rate limiter + // (429 once waiting > max-num-seqs) cap the per-pod queue around 5, + // which still sits above the gauge up-threshold=3, so scale-up fires. scalingPressureConcurrency = 40 - // Concurrency for below-threshold workload. Chosen so the queue settles - // strictly below the gauge down-threshold=5 even with the scheduler's - // load spread, so scale-down converges back to the baseline. + // Concurrency for below-threshold workload. It fits within + // max-num-seqs, so the queue stays strictly below the gauge + // down-threshold=1 and scale-down converges back to the baseline. scalingSubThresholdConcurrency = 4 - // Rate of the background low-rate stream used for "no 5xx during scale-down". - scalingLowRate = 1.0 // requests per second + // Longer, back-to-back generations keep the pool active during scale-down + // without filling the five running slots or building a waiting queue. + scalingBackgroundMaxTokens = 1024 + + // Bounds every completion so request latency on a real GPU stays short + // and the load shape matches the simulator's. + scalingMaxTokens = 128 + + // Real GPU nodes are provisioned on demand: an H100 node takes ~7m to + // join, then the model image is pulled and weights are loaded. + scalingProvisionTimeout = 30 * time.Minute + + // Upper bound for Karpenter to delete a NodeClaim and its node after + // the workload that needed it is gone. + scalingNodeRemovalTimeout = 30 * time.Minute + + // Upper bound for a model pod to terminate after KAITO removes it. + scalingPodRemovalTimeout = 10 * time.Minute - // Margins applied on top of the KEDA-derived Eventually timeouts. - scalingNodeReadyTimeout = 2 * time.Minute - scalingEndpointTimeout = 1 * time.Minute + // Margin applied on top of the KEDA-derived Eventually timeouts. + scalingEndpointTimeout = 1 * time.Minute + + scaleToZeroProvisioningMargin = 2 * time.Minute + scaleToZeroColdLoadConcurrency = 16 + + // scalingVLLMConfigName is the KAITO inference ConfigMap CaseScaling + // references; installScalingCase creates it before the InferenceSet. + scalingVLLMConfigName = "scaling-vllm-config" ) +// scalingVLLMConfig caps concurrent sequences per pod so a modest workload +// builds a waiting queue even on an H100. +var scalingVLLMConfig = map[string]string{ + "inference_config.yaml": "vllm:\n max-num-seqs: 5\n", +} + var _ = Describe("InferenceSet Scaling — Infra", - Ordered, utils.GinkgoLabelScaling, utils.GinkgoLabelNightly, - utils.GinkgoLabelStandardK8sOnly, func() { + Ordered, utils.GinkgoLabelScaling, utils.GinkgoLabelStandardK8sOnly, func() { // Per-case deployment owned by scaling_spec.go (see cases.go). // Resolved as Describe-local values so the table lookup happens // inside the suite (matching the pattern in model_routing_spec.go), // not at package init. caseDeployments := CaseDeployments[CaseScaling] - scalingModel := caseDeployments[0].Name + scalingValues := caseDeployments[0] + scalingModel := scalingValues.Name scalingNamespace := CaseNamespace(CaseScaling) - // gateway routes into the CaseScaling deployment's Gateway. - // Resolved in the outer BeforeAll once per suite run so all nested - // Describes (Scale-Up → Scale-Down, Anti-Flapping) share it. - var gateway deploy.GatewayEndpoint + var ( + // gateway routes into the CaseScaling deployment's Gateway. + // Resolved in the outer BeforeAll once per suite run so all + // nested Describes (Scale-Up → Scale-Down, Anti-Flapping) share it. + gateway deploy.GatewayEndpoint + // NodeClaims that existed before the case was installed; any + // other NodeClaim must be gone after teardown. + nodeClaimsBeforeInstall []string + ) BeforeAll(func() { - gateway = InstallCase(CaseScaling) + ctx := context.Background() + nodeClaimsBeforeInstall = snapshotNodeClaimNames(ctx) + gateway = installScalingCase(ctx) }) AfterAll(func() { UninstallCase(CaseScaling) + if nodeClaimsBeforeInstall != nil { + expectNoNewNodeClaims(context.Background(), nodeClaimsBeforeInstall, scalingNodeRemovalTimeout) + } }) // Baseline state shared by the Ordered Scale-Up → Scale-Down flow. type baseline struct { - replicas int32 - inventory utils.ScalingInventory - keda utils.KEDAParams + replicas int32 + // pods maps each Ready model pod to its node. + pods map[string]string + keda utils.KEDAParams } - Describe("Scale-Up → Scale-Down End-to-End", - Ordered, func() { + Describe("Scale 0 → 1 → 2 → 1 → 0 End-to-End", + Ordered, utils.GinkgoLabelScaleToZero, func() { var ( - ctx context.Context - base baseline - bulkLoad *utils.LoadGenerator - lowLoad *utils.LoadGenerator + ctx context.Context + base baseline + bulkLoad *utils.LoadGenerator + lowLoad *utils.LoadGenerator + coldLoad *utils.LoadGenerator + parked bool + lifecycleValues deploy.ModelDeploymentValues // State carried across Its. - newFakeNode string - newShadowPod string - newNodeClaim string - newLease string - inventoryAfter utils.ScalingInventory + newPod string + // Every model pod seen during the flow → its node, and + // node → backing NodeClaim. Which replica KAITO removes + // on scale-down is not fixed, so F1 derives the victim + // from these. + seenPods map[string]string + claimsByNode map[string]string ) BeforeAll(func() { ctx = context.Background() - clientset, err := utils.GetK8sClientset() + values := scalingValues + values.MinReplicas = 0 + values.MaxReplicas = 1 + values.CooldownPeriod = 60 + initialCooldown := int64(60) + values.InitialCooldownPeriod = &initialCooldown + values.ScalingMetrics = slices.Clone(values.ScalingMetrics) + values.ScalingMetrics[0].Source = "modelpod" + values.ScalingMetrics[0].DeactivationThreshold = "0" + values.ScalingMetrics = append(values.ScalingMetrics, + deploy.ScalingMetric{ + Name: "llm_d_epp_flow_control_queue_size", Type: "gauge", Source: "epp", + ActivationThreshold: "10", DeactivationThreshold: "0", + }, + deploy.ScalingMetric{Name: "vllm:num_requests_running", Type: "gauge", Source: "modelpod", DeactivationThreshold: "0"}, + ) + lifecycleValues = values + Expect(utils.UpgradeModelDeployment(ctx, values)).To(Succeed()) + Eventually(func(g Gomega) { + dynClient, err := utils.GetDynamicClient() + g.Expect(err).NotTo(HaveOccurred()) + scaledObject, err := dynClient.Resource(utils.ScaledObjectGVR).Namespace(scalingNamespace). + Get(ctx, scalingModel, metav1.GetOptions{}) + g.Expect(err).NotTo(HaveOccurred()) + minReplicas, found, err := unstructured.NestedInt64(scaledObject.Object, "spec", "minReplicaCount") + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(found).To(BeTrue()) + g.Expect(minReplicas).To(Equal(int64(0))) + maxReplicas, found, err := unstructured.NestedInt64(scaledObject.Object, "spec", "maxReplicaCount") + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(found).To(BeTrue()) + g.Expect(maxReplicas).To(Equal(int64(1))) + g.Expect(scaledObjectReady(scaledObject.Object)).To(BeTrue()) + }, 2*time.Minute, 5*time.Second).Should(Succeed()) + var err error + base.keda, err = utils.GetKEDAParams(ctx, scalingModel, scalingNamespace) Expect(err).NotTo(HaveOccurred()) - dynClient, err := utils.GetDynamicClient() + }) + + It("parks the warm model and releases its GPU NodeClaim while EPP stays ready", func() { + Eventually(func() (int32, error) { + return utils.GetInferenceSetReplicas(ctx, scalingModel, scalingNamespace) + }, max(base.keda.InitialCooldownPeriod, base.keda.CooldownPeriod)+scaleToZeroProvisioningMargin, 5*time.Second). + Should(Equal(int32(0))) + expectReadyModelPods(ctx, scalingValues, 0, scalingPodRemovalTimeout) + Expect(assertEPPReady(ctx, scalingNamespace, scalingModel)).To(Succeed()) + expectNoNewNodeClaims(ctx, nodeClaimsBeforeInstall, scalingNodeRemovalTimeout) + }) + + It("wakes from sustained EPP demand and serves traffic at one replica", func() { + coldLoad = &utils.LoadGenerator{ + Gateway: gateway, Model: scalingModel, Prompt: "hello", + Concurrency: scaleToZeroColdLoadConcurrency, MaxTokens: scalingMaxTokens, + } + coldLoad.Start(ctx) + Eventually(func() (int32, error) { + return utils.GetInferenceSetReplicas(ctx, scalingModel, scalingNamespace) + }, base.keda.ScaleUpTotalWait+scaleToZeroProvisioningMargin, 5*time.Second).Should(Equal(int32(1))) + Eventually(func(g Gomega) { + replicas, err := utils.GetInferenceSetReplicas(ctx, scalingModel, scalingNamespace) + g.Expect(err).NotTo(HaveOccurred()) + if replicas == 0 { + StopTrying("InferenceSet parked during cold-start provisioning").Now() + } + g.Expect(utils.WaitForInferenceSetReady(ctx, scalingValues, 10*time.Second)).To(Succeed()) + }, scalingProvisionTimeout, 5*time.Second).Should(Succeed()) + Eventually(func() int64 { + return coldLoad.Stats().Success + }, 2*time.Minute, time.Second).Should(BeNumerically(">", 0)) + + By("Starting one continuous background worker before stopping cold-start pressure") + lowLoad = &utils.LoadGenerator{ + Gateway: gateway, + Model: scalingModel, + Prompt: "Write a detailed tutorial about how computers work, covering processors, memory, storage, networking, and operating systems. Explain each topic thoroughly with examples and continue for at least 1500 words.", + Concurrency: 1, + MaxTokens: scalingBackgroundMaxTokens, + } + lowLoad.Start(ctx) + coldLoad.Stop() + handoffStarted := time.Now() + stats := coldLoad.Stats() + GinkgoWriter.Printf("cold-start load through handoff: %+v\n", stats) + Expect(stats.Success).To(BeNumerically(">", 0)) + Expect(stats.Errors5xx).To(BeNumerically("<=", 3*scaleToZeroColdLoadConcurrency), "cold-start load: %+v", stats) + + clientset, err := utils.GetK8sClientset() Expect(err).NotTo(HaveOccurred()) + By("Draining cold-start backlog and verifying KEDA observes the background workload before probing") + Eventually(func(g Gomega) { + replicas, err := utils.GetInferenceSetReplicas(ctx, scalingModel, scalingNamespace) + g.Expect(err).NotTo(HaveOccurred()) + if replicas == 0 { + StopTrying("InferenceSet parked during the cold-to-background load handoff").Now() + } + g.Expect(replicas).To(Equal(int32(1))) + g.Expect(lowLoad.Stats().Success).To(BeNumerically(">", 0)) + waiting, err := utils.ScrapeModelPodMetric(ctx, clientset, scalingValues, "vllm:num_requests_waiting") + g.Expect(err).NotTo(HaveOccurred()) + for pod, count := range waiting { + g.Expect(count).To(BeZero(), "cold-start backlog must drain on pod %s", pod) + } + dynClient, err := utils.GetDynamicClient() + g.Expect(err).NotTo(HaveOccurred()) + so, err := dynClient.Resource(utils.ScaledObjectGVR).Namespace(scalingNamespace). + Get(ctx, scalingModel, metav1.GetOptions{}) + g.Expect(err).NotTo(HaveOccurred()) + lastActive, found, err := unstructured.NestedString(so.Object, "status", "lastActiveTime") + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(found).To(BeTrue(), "KEDA must record active demand") + activeAt, err := time.Parse(time.RFC3339, lastActive) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(activeAt.After(handoffStarted)).To(BeTrue(), + "lastActiveTime=%s must advance after cold-start pressure stops", lastActive) + g.Expect(utils.CheckChatSuccess(ctx, gateway, scalingModel, utils.WithMaxTokens(scalingMaxTokens))).To(Succeed()) + }, 2*time.Minute, 5*time.Second).Should(Succeed()) + }) + + It("records the reactivated baseline before adding capacity", func() { + ctx = context.Background() + var err error - // The shared BeforeSuite installs the CaseScaling - // InferenceSet via the modeldeployment Helm chart at - // Replicas=MinReplicas=1 with EnableScaling=true, so KEDA's - // ScaledObject and the baseline pool are already in - // place by the time we run. We only need to read the - // effective KEDA params and snapshot the baseline - // inventory — no replica patching, no warm-up — see - // cases.go::CaseScaling for the source of truth. By("Reading effective KEDA parameters from the ScaledObject") base.keda, err = utils.GetKEDAParams(ctx, scalingModel, scalingNamespace) Expect(err).NotTo(HaveOccurred(), @@ -133,39 +293,43 @@ var _ = Describe("InferenceSet Scaling — Infra", Expect(base.keda.Threshold).To(BeNumerically(">", 0), "KEDA threshold must be set") GinkgoWriter.Printf( - "KEDA params: threshold=%d polling=%s cooldown=%s upStab=%s downStab=%s "+ + "KEDA params: threshold=%d polling=%s cooldown=%s initialCooldown=%s upPeriod=%s downPeriod=%s upStab=%s downStab=%s "+ "→ scaleUpWait=%s scaleDownWait=%s\n", base.keda.Threshold, base.keda.PollingInterval, base.keda.CooldownPeriod, + base.keda.InitialCooldownPeriod, base.keda.ScaleUpRatePeriod, base.keda.ScaleDownRatePeriod, base.keda.ScaleUpStabilization, base.keda.ScaleDownStabilization, base.keda.ScaleUpTotalWait, base.keda.ScaleDownTotalWait) - By("Capturing baseline replicas and inventory") + By("Capturing baseline replicas and model pod placement") base.replicas, err = utils.GetInferenceSetReplicas(ctx, scalingModel, scalingNamespace) Expect(err).NotTo(HaveOccurred()) - Expect(base.replicas).To(BeNumerically(">=", 1), - "BeforeSuite must have installed %s with replicas>=1", scalingModel) + Expect(base.replicas).To(Equal(int32(1)), + "cold-start activation must leave %s at one replica", scalingModel) GinkgoWriter.Printf("Baseline replicas for %s: %d\n", scalingModel, base.replicas) - base.inventory, err = utils.SnapshotScalingInventory(ctx, clientset, dynClient, scalingNamespace, scalingModel) - Expect(err).NotTo(HaveOccurred()) - Expect(base.inventory.ShadowPodNames).To(HaveLen(int(base.replicas)), - "expected %d shadow pods at baseline, got %v", - base.replicas, base.inventory.ShadowPodNames) - GinkgoWriter.Printf("Baseline inventory: %d fake-nodes, %d shadow-pods, %d NodeClaims, %d leases\n", - len(base.inventory.FakeNodeNames), len(base.inventory.ShadowPodNames), - len(base.inventory.NodeClaimNames), len(base.inventory.LeaseNames)) - - By("Starting low-rate background stream for the scale-down transition window") - lowLoad = &utils.LoadGenerator{ - Gateway: gateway, - Model: scalingModel, - Prompt: "hi", - Rate: scalingLowRate, + base.pods = expectReadyModelPods(ctx, scalingValues, int(base.replicas), scalingEndpointTimeout) + seenPods = maps.Clone(base.pods) + claimsByNode = currentNodeClaimsByNode(ctx) + for pod, node := range base.pods { + Expect(claimsByNode).To(HaveKey(node), + "baseline pod %s runs on node %s, which must be backed by a NodeClaim", pod, node) } - lowLoad.Start(ctx) + GinkgoWriter.Printf("Baseline model pods → nodes: %v\n", base.pods) + + // keda-kaito-scaler only scales up while readyReplicas >= spec.replicas. + By("Waiting for InferenceSet readyReplicas to match the baseline (KEDA readiness gate)") + Eventually(func() (int32, error) { + return utils.GetInferenceSetReadyReplicas(ctx, scalingModel, scalingNamespace) + }, scalingEndpointTimeout, 5*time.Second).Should(BeNumerically(">=", base.replicas), + "InferenceSet %s readyReplicas must reach %d or KEDA cannot scale up; "+ + "check the child Workspace WorkspaceSucceeded condition", scalingModel, base.replicas) + }) AfterAll(func() { + if coldLoad != nil { + coldLoad.Stop() + } if bulkLoad != nil { bulkLoad.Stop() } @@ -176,7 +340,7 @@ var _ = Describe("InferenceSet Scaling — Infra", // replica count, restore it so subsequent suites (or // the Anti-Flapping Describe) start from the same // baseline the chart installed. - if base.replicas <= 0 { + if base.replicas <= 0 || parked { return } By("Restoring baseline replicas") @@ -193,6 +357,7 @@ var _ = Describe("InferenceSet Scaling — Infra", Model: scalingModel, Prompt: "please explain the theory of relativity in as much detail as possible", Concurrency: scalingPressureConcurrency, + MaxTokens: scalingMaxTokens, } bulkLoad.Start(ctx) @@ -202,7 +367,7 @@ var _ = Describe("InferenceSet Scaling — Infra", By(fmt.Sprintf("Waiting for at least one pod's vllm:num_requests_waiting > %d", base.keda.Threshold)) Eventually(func(g Gomega) { - snap, err := utils.ScrapeModelMetric(ctx, clientset, scalingNamespace, scalingModel, "vllm:num_requests_waiting") + snap, err := utils.ScrapeModelPodMetric(ctx, clientset, scalingValues, "vllm:num_requests_waiting") g.Expect(err).NotTo(HaveOccurred()) var maxWaiting float64 for _, v := range snap { @@ -213,178 +378,142 @@ var _ = Describe("InferenceSet Scaling — Infra", g.Expect(maxWaiting).To(BeNumerically(">", float64(base.keda.Threshold)), "expected max num_requests_waiting > %d, saw %v", base.keda.Threshold, snap) }, base.keda.PollingInterval+60*time.Second, 3*time.Second).Should(Succeed()) + + By("Allowing the second replica only after bulk queue pressure is established") + lifecycleValues.MaxReplicas = scalingValues.MaxReplicas + Expect(utils.UpgradeModelDeployment(ctx, lifecycleValues)).To(Succeed()) + Eventually(func(g Gomega) { + replicas, err := utils.GetInferenceSetReplicas(ctx, scalingModel, scalingNamespace) + g.Expect(err).NotTo(HaveOccurred()) + if replicas == 0 { + StopTrying("InferenceSet parked while enabling capacity scaling").Now() + } + dynClient, err := utils.GetDynamicClient() + g.Expect(err).NotTo(HaveOccurred()) + so, err := dynClient.Resource(utils.ScaledObjectGVR).Namespace(scalingNamespace). + Get(ctx, scalingModel, metav1.GetOptions{}) + g.Expect(err).NotTo(HaveOccurred()) + maxReplicas, found, err := unstructured.NestedInt64(so.Object, "spec", "maxReplicaCount") + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(found).To(BeTrue()) + g.Expect(maxReplicas).To(Equal(int64(scalingValues.MaxReplicas))) + g.Expect(scaledObjectReady(so.Object)).To(BeTrue()) + }, 2*time.Minute, 5*time.Second).Should(Succeed()) }) It("A2: KEDA bumps InferenceSet.spec.replicas", func() { expected := base.replicas + 1 By(fmt.Sprintf("Waiting for .spec.replicas to reach %d", expected)) - Eventually(func() (int32, error) { - return utils.GetInferenceSetReplicas(ctx, scalingModel, scalingNamespace) + Eventually(func(g Gomega) { + replicas, err := utils.GetInferenceSetReplicas(ctx, scalingModel, scalingNamespace) + g.Expect(err).NotTo(HaveOccurred()) + if replicas == 0 { + StopTrying("InferenceSet parked while capacity-scaling load was active").Now() + } + g.Expect(replicas).To(BeNumerically(">=", expected)) }, base.keda.ScaleUpTotalWait+60*time.Second, 5*time.Second). - Should(BeNumerically(">=", expected), + Should(Succeed(), "KEDA should bump replicas from %d to >=%d within %s", base.replicas, expected, base.keda.ScaleUpTotalWait+60*time.Second) }) - It("B1: a new Fake Node becomes Ready with the correct labels", func() { + It("B1: a new model pod becomes Ready on a new GPU node backed by a NodeClaim", func() { clientset, err := utils.GetK8sClientset() Expect(err).NotTo(HaveOccurred()) dynClient, err := utils.GetDynamicClient() Expect(err).NotTo(HaveOccurred()) - Eventually(func(g Gomega) { - inv, err := utils.SnapshotScalingInventory(ctx, clientset, dynClient, scalingNamespace, scalingModel) - g.Expect(err).NotTo(HaveOccurred()) - diff := utils.DiffInventory(base.inventory, inv) - g.Expect(diff.AddedFakeNodes).NotTo(BeEmpty(), - "expected at least one new fake node, got inventory=%+v", inv) - // Pick the first new fake node and verify Ready + labels. - newFakeNode = diff.AddedFakeNodes[0] - node, err := clientset.CoreV1().Nodes().Get(ctx, newFakeNode, metav1.GetOptions{}) - g.Expect(err).NotTo(HaveOccurred()) - g.Expect(node.Labels).To(HaveKeyWithValue("kaito.sh/fake-node", "true")) - g.Expect(node.Labels).To(HaveKey("kaito.sh/workspace")) - g.Expect(utils.IsFakeNodeReady(node)).To(BeTrue(), - "new fake node %s must be Ready=True without unreachable taint", newFakeNode) - }, scalingNodeReadyTimeout, 5*time.Second).Should(Succeed()) - GinkgoWriter.Printf("New fake node: %s\n", newFakeNode) - }) - - It("B2: a new Shadow Pod becomes Running with both containers", func() { - clientset, err := utils.GetK8sClientset() - Expect(err).NotTo(HaveOccurred()) - dynClient, err := utils.GetDynamicClient() - Expect(err).NotTo(HaveOccurred()) + baseNodes := map[string]bool{} + for _, node := range base.pods { + baseNodes[node] = true + } + var placements, byNode map[string]string + var newNode string Eventually(func(g Gomega) { - inv, err := utils.SnapshotScalingInventory(ctx, clientset, dynClient, scalingNamespace, scalingModel) - g.Expect(err).NotTo(HaveOccurred()) - diff := utils.DiffInventory(base.inventory, inv) - g.Expect(diff.AddedShadowPods).NotTo(BeEmpty(), - "expected a new shadow pod, got inventory=%+v", inv) - newShadowPod = diff.AddedShadowPods[0] - pod, err := clientset.CoreV1().Pods(scalingNamespace).Get( - ctx, newShadowPod, metav1.GetOptions{}) + pods, err := utils.ListModelPods(ctx, clientset, scalingValues) g.Expect(err).NotTo(HaveOccurred()) - g.Expect(pod.Status.Phase).To(Equal(corev1.PodRunning), - "shadow pod %s should be Running", newShadowPod) - g.Expect(pod.Status.PodIP).NotTo(BeEmpty(), - "shadow pod %s must have a CNI PodIP", newShadowPod) - - // llm-d-inference-sim runs as a regular container and is the - // only container in the shadow pod (the UDS tokenizer sidecar - // was removed — the sim forces its built-in dummy tokenizer). - // Aggregate readiness from both Spec.Containers and - // Spec.InitContainers defensively, then map names to their - // *Status entries. - containers := map[string]bool{} - for _, c := range pod.Spec.Containers { - containers[c.Name] = false - } - for _, c := range pod.Spec.InitContainers { - containers[c.Name] = false - } - for _, cs := range pod.Status.ContainerStatuses { - containers[cs.Name] = cs.Ready - } - for _, cs := range pod.Status.InitContainerStatuses { - containers[cs.Name] = cs.Ready - } - g.Expect(containers).To(HaveKey("llm-d-inference-sim")) - for name, ready := range containers { - g.Expect(ready).To(BeTrue(), - "container %s of shadow pod %s should be ready", name, newShadowPod) + var candidate *corev1.Pod + ready := utils.ReadyPods(pods) + for i := range ready { + if _, isBase := base.pods[ready[i].Name]; !isBase { + candidate = &ready[i] + break + } } - }, scalingNodeReadyTimeout, 5*time.Second).Should(Succeed()) - GinkgoWriter.Printf("New shadow pod: %s\n", newShadowPod) - }) - - It("B3: the original pod is patched with the shadow pod's CNI IP", func() { - clientset, err := utils.GetK8sClientset() - Expect(err).NotTo(HaveOccurred()) - - shadowPod, err := clientset.CoreV1().Pods(scalingNamespace). - Get(ctx, newShadowPod, metav1.GetOptions{}) - Expect(err).NotTo(HaveOccurred()) - // Annotation `kaito.sh/original-pod` is `/` and - // is the canonical reference back to the original pod. - originalRef := shadowPod.Annotations["kaito.sh/original-pod"] - Expect(originalRef).NotTo(BeEmpty(), - "shadow pod %s must carry kaito.sh/original-pod annotation", newShadowPod) - parts := strings.SplitN(originalRef, "/", 2) - Expect(parts).To(HaveLen(2), - "kaito.sh/original-pod should be ns/name, got %q", originalRef) - origNS, origName := parts[0], parts[1] - shadowIP := shadowPod.Status.PodIP - Expect(shadowIP).NotTo(BeEmpty()) - - Eventually(func(g Gomega) { - orig, err := clientset.CoreV1().Pods(origNS). - Get(ctx, origName, metav1.GetOptions{}) + g.Expect(candidate).NotTo(BeNil(), + "expected a Ready model pod beyond baseline %v, have %v", + base.pods, utils.PodPlacements(pods)) + + nodeName := candidate.Spec.NodeName + g.Expect(baseNodes).NotTo(HaveKey(nodeName), + "new pod %s must run on a newly provisioned node, not baseline node %s", + candidate.Name, nodeName) + node, err := clientset.CoreV1().Nodes().Get(ctx, nodeName, metav1.GetOptions{}) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(isNodeReady(node)).To(BeTrue(), "node %s must be Ready", nodeName) + g.Expect(node.Labels).To(HaveKeyWithValue(corev1.LabelInstanceTypeStable, scalingValues.InstanceType), + "node %s must be the requested instance type", nodeName) + gpus := node.Status.Allocatable[corev1.ResourceName("nvidia.com/gpu")] + g.Expect(gpus.Value()).To(BeNumerically(">=", 1), + "node %s must advertise allocatable nvidia.com/gpu", nodeName) + + nodeClaims, err := utils.ListNodeClaims(ctx, dynClient) g.Expect(err).NotTo(HaveOccurred()) - g.Expect(orig.Status.Phase).To(Equal(corev1.PodRunning), - "original pod %s should be Running", origName) - g.Expect(orig.Status.PodIP).To(Equal(shadowIP), - "original pod podIP should match shadow IP") - g.Expect(orig.Annotations).To(HaveKey("kaito.sh/shadow-pod-ref"), - "original pod should carry shadow-pod-ref annotation") - }, scalingEndpointTimeout, 3*time.Second).Should(Succeed()) + byNode = utils.NodeClaimsByNode(nodeClaims) + g.Expect(byNode).To(HaveKey(nodeName), + "node %s must be backed by a NodeClaim", nodeName) + + newPod = candidate.Name + newNode = nodeName + placements = utils.PodPlacements(pods) + }, scalingProvisionTimeout, 15*time.Second).Should(Succeed()) + + maps.Copy(seenPods, placements) + maps.Copy(claimsByNode, byNode) + GinkgoWriter.Printf("New model pod %s on node %s (NodeClaim %s)\n", + newPod, newNode, byNode[newNode]) }) - It("C1: inference_pool_ready_pods grows to match the new replica count", func() { + It("C1: Ready model pods grow to match the new replica count", func() { clientset, err := utils.GetK8sClientset() Expect(err).NotTo(HaveOccurred()) Eventually(func(g Gomega) { current, err := utils.GetInferenceSetReplicas(ctx, scalingModel, scalingNamespace) g.Expect(err).NotTo(HaveOccurred()) - // Count Ready inference-pool pods directly via the Kubernetes + // Count Ready model pods directly via the Kubernetes // API instead of scraping the EPP /metrics endpoint. EPP // v1.3.x enables controller-runtime's metrics auth filter, // which rejects requests routed through the apiserver // pod-proxy (the proxy strips the Authorization header). - // Counting pods with the InferencePool's selector that - // have PodReady=True is semantically equivalent to the - // `inference_pool_ready_pods` gauge that EPP exports. - ready, err := utils.CountInferencePoolReadyPods(ctx, clientset, scalingModel, scalingNamespace) + // Ready pods selected by the InferenceSet are what the + // InferencePool's `inference_pool_ready_pods` gauge counts. + pods, err := utils.ListModelPods(ctx, clientset, scalingValues) g.Expect(err).NotTo(HaveOccurred()) + ready := len(utils.ReadyPods(pods)) g.Expect(ready).To(BeNumerically(">=", int(current)), - "ready inference-pool pods (%d) should match replicas (%d)", ready, current) + "Ready model pods (%d) should match replicas (%d)", ready, current) }, scalingEndpointTimeout+1*time.Minute, 5*time.Second).Should(Succeed()) }) - It("D1: the new shadow pod actually serves traffic", func() { + It("D1: the new model pod actually serves traffic", func() { clientset, err := utils.GetK8sClientset() Expect(err).NotTo(HaveOccurred()) By("Scraping per-pod vllm:request_success_total") var lastSnap utils.PodMetricSnapshot Eventually(func(g Gomega) { - snap, err := utils.ScrapeRequestSuccessTotal(ctx, clientset, scalingNamespace, scalingModel) + snap, err := utils.ScrapeModelPodMetric(ctx, clientset, scalingValues, "vllm:request_success_total") g.Expect(err).NotTo(HaveOccurred()) lastSnap = snap - v, ok := snap[newShadowPod] + v, ok := snap[newPod] g.Expect(ok).To(BeTrue(), - "new shadow pod %s must appear in per-pod metrics", newShadowPod) + "new model pod %s must appear in per-pod metrics", newPod) g.Expect(v).To(BeNumerically(">", 0), - "new shadow pod %s should have vllm:request_success_total > 0", newShadowPod) + "new model pod %s should have vllm:request_success_total > 0", newPod) }, 90*time.Second, 5*time.Second).Should(Succeed()) GinkgoWriter.Printf("Per-pod success counters after scale-up: %+v\n", lastSnap) - - // Capture the post-scale-up inventory for Scale-Down asserts. - dynClient, err := utils.GetDynamicClient() - Expect(err).NotTo(HaveOccurred()) - inventoryAfter, err = utils.SnapshotScalingInventory(ctx, clientset, dynClient, scalingNamespace, scalingModel) - Expect(err).NotTo(HaveOccurred()) - diff := utils.DiffInventory(base.inventory, inventoryAfter) - if len(diff.AddedNodeClaims) > 0 { - newNodeClaim = diff.AddedNodeClaims[0] - } - if len(diff.AddedLeases) > 0 { - newLease = diff.AddedLeases[0] - } - GinkgoWriter.Printf("Scale-up added: fakeNodes=%v shadowPods=%v nodeClaims=%v leases=%v\n", - diff.AddedFakeNodes, diff.AddedShadowPods, diff.AddedNodeClaims, diff.AddedLeases) }) It("E1: queue drains once the bulk load stops", func() { @@ -399,11 +528,8 @@ var _ = Describe("InferenceSet Scaling — Infra", By("Snapshotting pre-stop queue depth") var preStopMaxWaiting float64 Eventually(func(g Gomega) { - snap, err := utils.ScrapeModelMetric(ctx, clientset, scalingNamespace, scalingModel, - "vllm:num_requests_waiting") + snap, err := utils.ScrapeModelPodMetric(ctx, clientset, scalingValues, "vllm:num_requests_waiting") g.Expect(err).NotTo(HaveOccurred()) - g.Expect(snap).NotTo(BeEmpty(), - "expected at least one shadow pod to expose vllm:num_requests_waiting") var m float64 for _, v := range snap { if v > m { @@ -413,7 +539,7 @@ var _ = Describe("InferenceSet Scaling — Infra", // The queue must still be loaded — bulk load is // running at concurrency=40 against ≤2 pods × 5 // running slots. If we don't see a non-zero queue, - // either the simulator isn't reporting the metric + // either the model server isn't reporting the metric // or bulk load isn't actually pressuring the pool; // either way the drain assertion below would be // vacuous. @@ -429,14 +555,11 @@ var _ = Describe("InferenceSet Scaling — Infra", GinkgoWriter.Printf("Bulk load stats: total=%d success=%d 5xx=%d other=%d transportErr=%d\n", stats.Total, stats.Success, stats.Errors5xx, stats.OtherNon2xx, stats.TransportErr) - By("Waiting for vllm:num_requests_waiting == 0 on every shadow pod") - // We don't wait for running==0 because lowLoad keeps 1 req/s running. + By("Waiting for vllm:num_requests_waiting == 0 on every Ready model pod") + // The background worker continues generating while the waiting queue drains. Eventually(func(g Gomega) { - snap, err := utils.ScrapeModelMetric(ctx, clientset, scalingNamespace, scalingModel, - "vllm:num_requests_waiting") + snap, err := utils.ScrapeModelPodMetric(ctx, clientset, scalingValues, "vllm:num_requests_waiting") g.Expect(err).NotTo(HaveOccurred()) - g.Expect(snap).NotTo(BeEmpty(), - "expected at least one shadow pod in metrics snapshot") for pod, v := range snap { g.Expect(v).To(BeNumerically("==", 0), "pod %s still has waiting=%v", pod, v) @@ -452,79 +575,86 @@ var _ = Describe("InferenceSet Scaling — Infra", r, err := utils.GetInferenceSetReplicas(ctx, scalingModel, scalingNamespace) g.Expect(err).NotTo(HaveOccurred()) last = r + if r == 0 { + StopTrying("InferenceSet parked before the background workload was stopped").Now() + } g.Expect(r).To(Equal(base.replicas), "replicas should drop to %d, now %d", base.replicas, r) }, base.keda.ScaleDownTotalWait+2*time.Minute, 15*time.Second).Should(Succeed()) _ = last }) - It("F1: excess shadow-pod, fake-node, NodeClaim, and lease are cleaned up", func() { + It("F1: the removed model pod, its NodeClaim, and its node are cleaned up", func() { clientset, err := utils.GetK8sClientset() Expect(err).NotTo(HaveOccurred()) dynClient, err := utils.GetDynamicClient() Expect(err).NotTo(HaveOccurred()) - Eventually(func(g Gomega) { - // The extra shadow pod must be gone. - _, err := clientset.CoreV1().Pods(scalingNamespace). - Get(ctx, newShadowPod, metav1.GetOptions{}) - g.Expect(apierrors.IsNotFound(err)).To(BeTrue(), - "shadow pod %s should be deleted, got err=%v", newShadowPod, err) - - // The extra fake node must be gone. - _, err = clientset.CoreV1().Nodes().Get(ctx, newFakeNode, metav1.GetOptions{}) - g.Expect(apierrors.IsNotFound(err)).To(BeTrue(), - "fake node %s should be deleted, got err=%v", newFakeNode, err) - - // The lease for the fake node must be GC'd. - if newLease != "" { - _, err = clientset.CoordinationV1().Leases("kube-node-lease"). - Get(ctx, newLease, metav1.GetOptions{}) - g.Expect(apierrors.IsNotFound(err)).To(BeTrue(), - "lease %s should be GC'd, got err=%v", newLease, err) - } + By(fmt.Sprintf("Waiting for %d Ready model pod(s) and no others", base.replicas)) + current := expectReadyModelPods(ctx, scalingValues, int(base.replicas), scalingPodRemovalTimeout) - // The extra NodeClaim must be gone (or at least marked deleting). - if newNodeClaim != "" { - _, err = dynClient.Resource(utils.NodeClaimGVR). - Get(ctx, newNodeClaim, metav1.GetOptions{}) - g.Expect(apierrors.IsNotFound(err)).To(BeTrue(), - "NodeClaim %s should be deleted, got err=%v", newNodeClaim, err) + // KAITO may remove either replica, so the victim is + // whichever recorded pod is no longer current. + hostingNodes := map[string]bool{} + for _, node := range current { + hostingNodes[node] = true + } + var victimPods, victimNodes []string + for _, pod := range slices.Sorted(maps.Keys(seenPods)) { + if _, ok := current[pod]; ok { + continue + } + victimPods = append(victimPods, pod) + if node := seenPods[pod]; node != "" && !hostingNodes[node] && !slices.Contains(victimNodes, node) { + victimNodes = append(victimNodes, node) } - }, 8*time.Minute, 5*time.Second).Should(Succeed()) + } + GinkgoWriter.Printf("Scale-down victims: pods=%v nodes=%v (current %v)\n", victimPods, victimNodes, current) + Expect(victimPods).NotTo(BeEmpty(), + "a model pod recorded during scale-up must be gone; seen=%v current=%v", seenPods, current) + Expect(victimNodes).NotTo(BeEmpty(), + "the removed pod's node must no longer host a model pod; seen=%v current=%v", seenPods, current) + + for _, pod := range victimPods { + By(fmt.Sprintf("Waiting for model pod %s to be deleted", pod)) + Eventually(func() bool { + _, err := clientset.CoreV1().Pods(scalingNamespace).Get(ctx, pod, metav1.GetOptions{}) + return apierrors.IsNotFound(err) + }, scalingPodRemovalTimeout, 5*time.Second).Should(BeTrue(), + "model pod %s should be deleted", pod) + } - By("Verifying baseline inventory is preserved") - inv, err := utils.SnapshotScalingInventory(ctx, clientset, dynClient, scalingNamespace, scalingModel) - Expect(err).NotTo(HaveOccurred()) - Expect(inv.ShadowPodNames).To(HaveLen(len(base.inventory.ShadowPodNames)), - "baseline shadow pods should be preserved") - Expect(inv.FakeNodeNames).To(HaveLen(len(base.inventory.FakeNodeNames)), - "baseline fake nodes should be preserved") - for _, name := range base.inventory.ShadowPodNames { - Expect(inv.ShadowPodNames).To(ContainElement(name), - "baseline shadow pod %s must still exist", name) + for _, node := range victimNodes { + nodeClaim := claimsByNode[node] + Expect(nodeClaim).NotTo(BeEmpty(), "no NodeClaim recorded for node %s", node) + + By(fmt.Sprintf("Waiting for NodeClaim %s to be deleted", nodeClaim)) + Eventually(func() bool { + _, err := dynClient.Resource(utils.NodeClaimGVR).Get(ctx, nodeClaim, metav1.GetOptions{}) + return apierrors.IsNotFound(err) + }, scalingNodeRemovalTimeout, 15*time.Second).Should(BeTrue(), + "NodeClaim %s should be deleted", nodeClaim) + + By(fmt.Sprintf("Waiting for node %s to be deleted", node)) + Eventually(func() bool { + _, err := clientset.CoreV1().Nodes().Get(ctx, node, metav1.GetOptions{}) + return apierrors.IsNotFound(err) + }, scalingNodeRemovalTimeout, 15*time.Second).Should(BeTrue(), + "node %s should be deleted", node) } }) - It("F2: inference_pool_ready_pods shrinks to match the new replica count", func() { - clientset, err := utils.GetK8sClientset() - Expect(err).NotTo(HaveOccurred()) - - Eventually(func(g Gomega) { - // See C1 for why we count Ready pods directly via the - // Kubernetes API instead of scraping EPP /metrics. - ready, err := utils.CountInferencePoolReadyPods(ctx, clientset, scalingModel, scalingNamespace) - g.Expect(err).NotTo(HaveOccurred()) - g.Expect(ready).To(Equal(int(base.replicas)), - "ready inference-pool pods should be %d, got %d", base.replicas, ready) - }, scalingEndpointTimeout, 5*time.Second).Should(Succeed()) + It("F2: Ready model pods shrink to match the new replica count", func() { + // See C1 for why Ready model pods stand in for EPP's + // inference_pool_ready_pods gauge. + expectReadyModelPods(ctx, scalingValues, int(base.replicas), scalingEndpointTimeout) }) It("G1: remaining pods continue to serve traffic after scale-down", func() { clientset, err := utils.GetK8sClientset() Expect(err).NotTo(HaveOccurred()) - beforeSnap, err := utils.ScrapeRequestSuccessTotal(ctx, clientset, scalingNamespace, scalingModel) + beforeSnap, err := utils.ScrapeModelPodMetric(ctx, clientset, scalingValues, "vllm:request_success_total") Expect(err).NotTo(HaveOccurred()) By("Driving a short moderate-rate burst strictly below threshold") @@ -533,6 +663,7 @@ var _ = Describe("InferenceSet Scaling — Infra", Model: scalingModel, Prompt: "hello", Concurrency: scalingSubThresholdConcurrency, + MaxTokens: scalingMaxTokens, } burst.Start(ctx) time.Sleep(15 * time.Second) @@ -546,7 +677,7 @@ var _ = Describe("InferenceSet Scaling — Infra", "no 5xx expected post-scale-down, got stats=%+v", stats) Expect(stats.Success).To(BeNumerically(">", 0)) - afterSnap, err := utils.ScrapeRequestSuccessTotal(ctx, clientset, scalingNamespace, scalingModel) + afterSnap, err := utils.ScrapeModelPodMetric(ctx, clientset, scalingValues, "vllm:request_success_total") Expect(err).NotTo(HaveOccurred()) diff := utils.DiffSnapshots(beforeSnap, afterSnap) var incremented int @@ -559,15 +690,27 @@ var _ = Describe("InferenceSet Scaling — Infra", "at least one remaining pod should serve traffic, diff=%+v", diff) }) - It("G2: low-rate stream saw no 5xx during the scale-down transition", func() { + It("G2: continuous background stream saw no 5xx during the scale-down transition", func() { stats := lowLoad.Stats() - GinkgoWriter.Printf("Low-rate stats across scale-down window: %+v\n", stats) + GinkgoWriter.Printf("Continuous background stats across scale-down window: %+v\n", stats) Expect(stats.Errors5xx).To(BeNumerically("==", 0), - "background low-rate stream must not observe any 5xx, stats=%+v", stats) + "continuous background stream must not observe any 5xx, stats=%+v", stats) // Allow a handful of transport-level reconnects; be strict about server errors. - Expect(stats.Total).To(BeNumerically(">", 0), - "low-rate generator must have issued some requests") + Expect(stats.Success).To(BeNumerically(">", 0), + "background worker must have completed successful requests, stats=%+v", stats) + }) + + It("returns to zero and releases GPU NodeClaims after all work clears", func() { + lowLoad.Stop() + Eventually(func() (int32, error) { + return utils.GetInferenceSetReplicas(ctx, scalingModel, scalingNamespace) + }, base.keda.PollingInterval+base.keda.CooldownPeriod+scaleToZeroProvisioningMargin, 5*time.Second). + Should(Equal(int32(0))) + expectReadyModelPods(ctx, scalingValues, 0, scalingPodRemovalTimeout) + Expect(assertEPPReady(ctx, scalingNamespace, scalingModel)).To(Succeed()) + expectNoNewNodeClaims(ctx, nodeClaimsBeforeInstall, scalingNodeRemovalTimeout) + parked = true }) }, ) @@ -577,27 +720,35 @@ var _ = Describe("InferenceSet Scaling — Infra", var ( ctx context.Context base struct { - replicas int32 - inventory utils.ScalingInventory - keda utils.KEDAParams + replicas int32 + pods map[string]string + nodeClaims []string + keda utils.KEDAParams } ) + BeforeAll(func() { + ctx = context.Background() + Expect(utils.UpgradeModelDeployment(ctx, scalingValues)).To(Succeed()) + Expect(utils.SetInferenceSetReplicas(ctx, scalingModel, scalingNamespace, 1)).To(Succeed()) + Expect(utils.WaitForInferenceSetReady(ctx, scalingValues, scalingProvisionTimeout)).To(Succeed()) + Eventually(func() error { + return utils.CheckChatSuccess(ctx, gateway, scalingModel, utils.WithMaxTokens(scalingMaxTokens)) + }, scalingProvisionTimeout, 10*time.Second).Should(Succeed()) + }) + BeforeEach(func() { ctx = context.Background() - clientset, err := utils.GetK8sClientset() - Expect(err).NotTo(HaveOccurred()) - dynClient, err := utils.GetDynamicClient() - Expect(err).NotTo(HaveOccurred()) + var err error base.keda, err = utils.GetKEDAParams(ctx, scalingModel, scalingNamespace) Expect(err).NotTo(HaveOccurred()) base.replicas, err = utils.GetInferenceSetReplicas(ctx, scalingModel, scalingNamespace) Expect(err).NotTo(HaveOccurred()) - base.inventory, err = utils.SnapshotScalingInventory(ctx, clientset, dynClient, scalingNamespace, scalingModel) - Expect(err).NotTo(HaveOccurred()) - GinkgoWriter.Printf("Anti-flapping baseline: replicas=%d fakeNodes=%d shadowPods=%d\n", - base.replicas, len(base.inventory.FakeNodeNames), len(base.inventory.ShadowPodNames)) + base.pods = expectReadyModelPods(ctx, scalingValues, int(base.replicas), scalingPodRemovalTimeout) + base.nodeClaims = snapshotNodeClaimNames(ctx) + GinkgoWriter.Printf("Anti-flapping baseline: replicas=%d pods=%v nodeClaims=%d\n", + base.replicas, base.pods, len(base.nodeClaims)) }) AfterEach(func() { @@ -606,9 +757,9 @@ var _ = Describe("InferenceSet Scaling — Infra", context.Background(), scalingModel, scalingNamespace, base.replicas); err != nil { GinkgoWriter.Printf("warning: restore replicas: %v\n", err) } - // Passively wait one cooldown so the next case starts fresh. + // Let the HPA rate-limit policies expire before the next case. // (Only pays the full window when a test actually caused a scale event.) - time.Sleep(base.keda.CooldownPeriod) + time.Sleep(max(base.keda.ScaleUpRatePeriod, base.keda.ScaleDownRatePeriod)) }) It("H1: stays at baseline when pressure is held below threshold", func() { @@ -623,46 +774,49 @@ var _ = Describe("InferenceSet Scaling — Infra", Model: scalingModel, Prompt: "hello", Concurrency: scalingSubThresholdConcurrency, + MaxTokens: scalingMaxTokens, } load.Start(ctx) defer load.Stop() - observeWindow := base.keda.PollingInterval + base.keda.CooldownPeriod - // Sample num_requests_waiting every 15s and confirm it stays below threshold. + observeWindow := max(base.keda.ScaleUpTotalWait, base.keda.ScaleDownTotalWait) + // Every 15s confirm that nothing scaled — replicas, model pods, + // and NodeClaims all stay at baseline — and track the queue. sampleEvery := 15 * time.Second deadline := time.Now().Add(observeWindow) var maxSeen float64 for time.Now().Before(deadline) { - snap, err := utils.ScrapeModelMetric(ctx, clientset, scalingNamespace, scalingModel, - "vllm:num_requests_waiting") + snap, err := utils.ScrapeModelPodMetric(ctx, clientset, scalingValues, "vllm:num_requests_waiting") Expect(err).NotTo(HaveOccurred()) for _, v := range snap { if v > maxSeen { maxSeen = v } } + + cur, err := utils.GetInferenceSetReplicas(ctx, scalingModel, scalingNamespace) + Expect(err).NotTo(HaveOccurred()) + Expect(cur).To(Equal(base.replicas), + "replicas must remain at baseline %d, got %d", base.replicas, cur) + + pods, err := utils.ListModelPods(ctx, clientset, scalingValues) + Expect(err).NotTo(HaveOccurred()) + for _, pod := range pods { + Expect(base.pods).To(HaveKey(pod.Name), + "no model pod should be added; baseline=%v got %s", base.pods, pod.Name) + } + + nodeClaims, err := utils.ListNodeClaims(ctx, dynClient) + Expect(err).NotTo(HaveOccurred()) + added := utils.NewNodeClaimNames(base.nodeClaims, nodeClaims) + Expect(added).To(BeEmpty(), "no NodeClaim should be added; got %v", added) + time.Sleep(sampleEvery) } GinkgoWriter.Printf("Max num_requests_waiting during H1: %v (threshold=%d)\n", maxSeen, base.keda.Threshold) Expect(maxSeen).To(BeNumerically("<", float64(base.keda.Threshold)), "H1 must keep load strictly below threshold to be meaningful") - - By("Verifying replicas and inventory did not change") - cur, err := utils.GetInferenceSetReplicas(ctx, scalingModel, scalingNamespace) - Expect(err).NotTo(HaveOccurred()) - Expect(cur).To(Equal(base.replicas), - "replicas must remain at baseline %d, got %d", base.replicas, cur) - - inv, err := utils.SnapshotScalingInventory(ctx, clientset, dynClient, scalingNamespace, scalingModel) - Expect(err).NotTo(HaveOccurred()) - diff := utils.DiffInventory(base.inventory, inv) - Expect(diff.AddedFakeNodes).To(BeEmpty(), - "no fake node should be added; got %v", diff.AddedFakeNodes) - Expect(diff.AddedShadowPods).To(BeEmpty(), - "no shadow pod should be added; got %v", diff.AddedShadowPods) - Expect(diff.AddedNodeClaims).To(BeEmpty(), - "no NodeClaim should be added; got %v", diff.AddedNodeClaims) }) It("H2: HPA scale-up stabilization gates immediate re-scale-up after a scale-down", func() { @@ -671,14 +825,13 @@ var _ = Describe("InferenceSet Scaling — Infra", // - replicas hold at baseline for the full HPA scale-up // stabilization window after tDown (safety / no oscillation), // - replicas DO scale up again once the window has elapsed - // plus one polling interval of slack (liveness / not stuck). + // plus any HPA rate-limit period (liveness / not stuck). // // NOTE on KEDA semantics: spec.cooldownPeriod only gates // scale-to-zero (it kicks in when the last trigger reports // inactive and minReplicaCount is 0). With minReplicaCount>=1, - // the actual anti-oscillation gate after a scale-down is the - // HPA scale-up stabilizationWindowSeconds, which keda-kaito-scaler - // sets via spec.advanced.horizontalPodAutoscalerConfig.behavior. + // HPA stabilization and rate-limit policies are independent gates. + // Liveness also allows the configured scale-up rate-limit period. By("Scale-Up: driving pressure") bulk := &utils.LoadGenerator{ @@ -686,6 +839,7 @@ var _ = Describe("InferenceSet Scaling — Infra", Model: scalingModel, Prompt: "please explain the theory of relativity in as much detail as possible", Concurrency: scalingPressureConcurrency, + MaxTokens: scalingMaxTokens, } bulk.Start(ctx) @@ -712,6 +866,7 @@ var _ = Describe("InferenceSet Scaling — Infra", Model: scalingModel, Prompt: "please explain the theory of relativity in as much detail as possible", Concurrency: scalingPressureConcurrency, + MaxTokens: scalingMaxTokens, } post.Start(ctx) defer post.Stop() @@ -733,8 +888,7 @@ var _ = Describe("InferenceSet Scaling — Infra", // // liveness zone: [antiFlapBoundary + slack, endAt] // replicas MUST eventually reach >= base+1, - // proving the stabilization window was the - // only gate. + // proving scaling resumes after the gates. // // Sample at min(PollingInterval/3, 5s) so the strict zone has // at least 5 sample points even at the default 60s window. @@ -805,9 +959,147 @@ var _ = Describe("InferenceSet Scaling — Infra", } Expect(seenScaleUpInLiveness).To(BeTrue(), "after the anti-flapping window + slack (%s after tDown), replicas should "+ - "scale up again (proves stabilization was the only gate). all samples: %s", + "scale up again (allowing the HPA rate-limit period). all samples: %s", base.keda.ScaleUpStabilization+slack, renderSamples()) }) }) }, ) + +// installScalingCase mirrors InstallCase for CaseScaling, with three +// differences: it creates the KAITO inference ConfigMap the case references, +// it waits up to scalingProvisionTimeout (a fresh GPU node plus model load can +// exceed InferenceSetReadyTimeout), and the warm-up completion is bounded. +func installScalingCase(ctx context.Context) deploy.GatewayEndpoint { + ns := CaseNamespace(CaseScaling) + Expect(ns).NotTo(BeEmpty(), "case %q has no namespace declared in CaseDeployments", CaseScaling) + Expect(utils.EnsureNamespace(ctx, ns)).To(Succeed(), "failed to ensure namespace %s", ns) + + clientset, err := utils.GetK8sClientset() + Expect(err).NotTo(HaveOccurred()) + Expect(utils.EnsureConfigMap(ctx, clientset, ns, scalingVLLMConfigName, scalingVLLMConfig)).To(Succeed()) + + gateway, err := utils.OpenGateway(ctx, ns, CaseGatewayName(CaseScaling)) + Expect(err).NotTo(HaveOccurred(), "failed to open gateway for case %s", CaseScaling) + + utils.GetClusterClient(utils.TestingCluster) + deployments := CaseDeployments[CaseScaling] + for _, d := range deployments { + By(fmt.Sprintf("Installing modeldeployment %s (model=%s) in %s", d.Name, d.Model, d.Namespace)) + Expect(utils.InstallModelDeployment(ctx, d)).To(Succeed(), "failed to install modeldeployment for %s", d.Name) + } + for _, d := range deployments { + By(fmt.Sprintf("Waiting up to %s for modeldeployment %s to become ready", scalingProvisionTimeout, d.Name)) + Expect(utils.WaitForInferenceSetReady(ctx, d, scalingProvisionTimeout)). + To(Succeed(), "modeldeployment %s not ready", d.Name) + + By(fmt.Sprintf("Waiting for gateway routing to be ready for deployment %s", d.Name)) + Eventually(func() error { + utils.ForgetNamespaceAuthHeaders(d.Namespace) + return utils.CheckChatSuccess(ctx, gateway, d.Name, utils.WithMaxTokens(scalingMaxTokens)) + }, scalingProvisionTimeout, 10*time.Second).Should(Succeed(), + "gateway should route to deployment %s successfully", d.Name) + } + return gateway +} + +// expectReadyModelPods waits until the deployment has exactly want model pods, +// all Ready, and returns their pod → node placement. +func expectReadyModelPods(ctx context.Context, values deploy.ModelDeploymentValues, want int, timeout time.Duration) map[string]string { + clientset, err := utils.GetK8sClientset() + Expect(err).NotTo(HaveOccurred()) + var placements map[string]string + Eventually(func(g Gomega) { + pods, err := utils.ListModelPods(ctx, clientset, values) + g.Expect(err).NotTo(HaveOccurred()) + ready := utils.ReadyPods(pods) + g.Expect(pods).To(HaveLen(want), "model pods: %v", utils.PodPlacements(pods)) + g.Expect(ready).To(HaveLen(want), "Ready model pods: %v of %v", + utils.PodPlacements(ready), utils.PodPlacements(pods)) + placements = utils.PodPlacements(ready) + }, timeout, 5*time.Second).Should(Succeed()) + return placements +} + +// currentNodeClaimsByNode returns node name → NodeClaim name for every +// NodeClaim that has registered a node. +func currentNodeClaimsByNode(ctx context.Context) map[string]string { + dynClient, err := utils.GetDynamicClient() + Expect(err).NotTo(HaveOccurred()) + nodeClaims, err := utils.ListNodeClaims(ctx, dynClient) + Expect(err).NotTo(HaveOccurred()) + return utils.NodeClaimsByNode(nodeClaims) +} + +// snapshotNodeClaimNames returns the names of all NodeClaims in the cluster. +func snapshotNodeClaimNames(ctx context.Context) []string { + dynClient, err := utils.GetDynamicClient() + Expect(err).NotTo(HaveOccurred()) + nodeClaims, err := utils.ListNodeClaims(ctx, dynClient) + Expect(err).NotTo(HaveOccurred()) + return utils.NodeClaimNames(nodeClaims) +} + +// expectNoNewNodeClaims waits until every NodeClaim not in before is gone, +// including NodeClaims that never registered a node (e.g. a scale-up cancelled +// before its GPU node joined). It attributes NodeClaims by snapshot rather +// than by name, so it assumes nothing else provisions nodes while the +// Scaling specs run serially. +func expectNoNewNodeClaims(ctx context.Context, before []string, timeout time.Duration) { + dynClient, err := utils.GetDynamicClient() + Expect(err).NotTo(HaveOccurred()) + By(fmt.Sprintf("Waiting up to %s for NodeClaims created by the case to be removed", timeout)) + Eventually(func(g Gomega) { + nodeClaims, err := utils.ListNodeClaims(ctx, dynClient) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(utils.NewNodeClaimNames(before, nodeClaims)).To(BeEmpty(), + "NodeClaims created during the case must be removed") + }, timeout, 30*time.Second).Should(Succeed()) +} + +func isNodeReady(node *corev1.Node) bool { + for _, c := range node.Status.Conditions { + if c.Type == corev1.NodeReady { + return c.Status == corev1.ConditionTrue + } + } + return false +} + +func scaledObjectReady(object map[string]interface{}) bool { + conditions, found, err := unstructured.NestedSlice(object, "status", "conditions") + if err != nil || !found { + return false + } + for _, condition := range conditions { + fields, ok := condition.(map[string]interface{}) + if ok && fields["type"] == "Ready" && fields["status"] == "True" { + return true + } + } + return false +} + +func assertEPPReady(ctx context.Context, namespace, modelName string) error { + clientset, err := utils.GetK8sClientset() + if err != nil { + return err + } + eppName := utils.EPPServiceName(modelName) + deployment, err := clientset.AppsV1().Deployments(namespace).Get(ctx, eppName, metav1.GetOptions{}) + if err != nil { + return err + } + if deployment.Status.ReadyReplicas < 1 { + return fmt.Errorf("EPP Deployment %s/%s has no ready replicas", namespace, eppName) + } + selector := "inferencepool=" + eppName + pods, err := clientset.CoreV1().Pods(namespace).List(ctx, metav1.ListOptions{LabelSelector: selector}) + if err != nil { + return err + } + if len(utils.ReadyPods(pods.Items)) > 0 { + return nil + } + return fmt.Errorf("no Running/Ready EPP pod matches %q; matching pods=%v", selector, pods.Items) +} diff --git a/test/e2e/utils/ginkgo.go b/test/e2e/utils/ginkgo.go index f5404b4..7b2c9ad 100644 --- a/test/e2e/utils/ginkgo.go +++ b/test/e2e/utils/ginkgo.go @@ -36,7 +36,8 @@ var ( // GinkgoLabelInferenceSet marks tests that verify InferenceSet lifecycle and routing setup. GinkgoLabelInferenceSet = g.Label("InferenceSet") - // GinkgoLabelNightly marks tests that are destructive or slow and only run in nightly CI. + // GinkgoLabelNightly marks tests that are destructive or slow. The PR and + // daily-upstream runs exclude it; the Azure nightly run includes it. GinkgoLabelNightly = g.Label("Nightly") // GinkgoLabelPerf marks prefix-cache load/perf specs that replay real @@ -63,10 +64,16 @@ var ( // GinkgoLabelScaling marks tests that verify end-to-end KEDA-driven // scaling of InferenceSet replicas, covering the scale-up and // scale-down phases as well as anti-flapping (no oscillation near the - // threshold or during cooldown). Nightly-only because KEDA polling and - // cooldown windows make these tests take minutes per run. + // threshold or during cooldown). Excluded from the main runs and executed + // serially by `make test-e2e-scaling`: on the GPU mocker in the daily + // upstream job, and on real H100 nodes in a dedicated nightly job, because + // KEDA windows and real GPU provisioning make these specs take tens of + // minutes each. GinkgoLabelScaling = g.Label("Scaling") + // GinkgoLabelScaleToZero marks the full 0 -> 1 -> 2 -> 1 -> 0 lifecycle. + GinkgoLabelScaleToZero = g.Label("ScaleToZero") + // GinkgoLabelFilterOrder marks tests that verify the Envoy HTTP // filter chain execution order on the per-namespace Gateway: // ext_authz → ext_proc.bbr → ext_proc (InferencePool/EPP) → router diff --git a/test/e2e/utils/http.go b/test/e2e/utils/http.go index ea37f1d..0cd1ba8 100644 --- a/test/e2e/utils/http.go +++ b/test/e2e/utils/http.go @@ -142,6 +142,7 @@ type gatewayRequest struct { transportRetry bool originPath bool timeout time.Duration + maxTokens int buildBody func(*gatewayRequest) ([]byte, error) } @@ -160,6 +161,12 @@ func WithPrompt(prompt string) RequestOption { return func(r *gatewayRequest) { r.prompt = prompt } } +// WithMaxTokens bounds the completion length of chat requests so a real model +// answers well inside HTTPTimeout. Zero leaves max_tokens unset. +func WithMaxTokens(n int) RequestOption { + return func(r *gatewayRequest) { r.maxTokens = n } +} + // WithHeader adds one request header. Repeatable. func WithHeader(name, value string) RequestOption { return func(r *gatewayRequest) { @@ -217,8 +224,9 @@ func SendChatContext(ctx context.Context, gateway deploy.GatewayEndpoint, model chatBody := func(request *gatewayRequest) { request.buildBody = func(settings *gatewayRequest) ([]byte, error) { return json.Marshal(ChatCompletionRequest{ - Model: model, - Messages: []ChatMessage{{Role: "user", Content: settings.prompt}}, + Model: model, + Messages: []ChatMessage{{Role: "user", Content: settings.prompt}}, + MaxTokens: settings.maxTokens, }) } } diff --git a/test/e2e/utils/http_test.go b/test/e2e/utils/http_test.go index 9fc683e..a1a4c0c 100644 --- a/test/e2e/utils/http_test.go +++ b/test/e2e/utils/http_test.go @@ -20,6 +20,7 @@ import ( "context" "encoding/json" "errors" + "fmt" "io" "net/http" "net/http/httptest" @@ -464,6 +465,58 @@ func TestLoadGeneratorStopCancelsRequests(t *testing.T) { } } +func TestChatMaxTokensIsSentOnlyWhenSet(t *testing.T) { + received := make(chan map[string]any, 1) + server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + var body map[string]any + if err := json.NewDecoder(request.Body).Decode(&body); err != nil { + t.Error(err) + } + received <- body + })) + defer server.Close() + gateway := &credentialEndpoint{ + fakeEndpoint: fakeEndpoint{urls: []string{server.URL + "/v1"}}, + headers: []deploy.AuthHeader{{Name: "X-API-Key", Value: "key"}}, + } + + for _, test := range []struct { + name string + send func() error + want any + }{ + {name: "chat default", send: func() error { return CheckChatSuccess(context.Background(), gateway, "model") }}, + {name: "chat bounded", send: func() error { + return CheckChatSuccess(context.Background(), gateway, "model", WithMaxTokens(128)) + }, want: float64(128)}, + {name: "load generator default", send: func() error { return runOneLoadRequest(gateway, 0) }}, + {name: "load generator bounded", send: func() error { return runOneLoadRequest(gateway, 64) }, want: float64(64)}, + } { + t.Run(test.name, func(t *testing.T) { + if err := test.send(); err != nil { + t.Fatal(err) + } + body := <-received + got, present := body["max_tokens"] + if test.want == nil && present { + t.Fatalf("max_tokens sent without being requested: %v", got) + } + if test.want != nil && got != test.want { + t.Fatalf("max_tokens=%v, want %v", got, test.want) + } + }) + } +} + +func runOneLoadRequest(gateway deploy.GatewayEndpoint, maxTokens int) error { + load := &LoadGenerator{Gateway: gateway, Model: "model", MaxTokens: maxTokens} + load.sendOnce(context.Background()) + if stats := load.Stats(); stats.Success != 1 { + return fmt.Errorf("load request failed: %+v", stats) + } + return nil +} + type trackedResponseBody struct { io.Reader closes int diff --git a/test/e2e/utils/loadgen.go b/test/e2e/utils/loadgen.go index d47ce0c..bface0e 100644 --- a/test/e2e/utils/loadgen.go +++ b/test/e2e/utils/loadgen.go @@ -52,6 +52,7 @@ type LoadGenerator struct { Prompt string Concurrency int Rate float64 // requests per second; 0 means Concurrency mode + MaxTokens int // max_tokens per request; 0 leaves it unset cancel context.CancelFunc wg sync.WaitGroup @@ -140,7 +141,8 @@ func (lg *LoadGenerator) sendOnce(ctx context.Context) { return } lg.total.Add(1) - resp, err := SendChatContext(ctx, lg.Gateway, lg.Model, WithPrompt(lg.Prompt), WithTransportRetry()) + resp, err := SendChatContext(ctx, lg.Gateway, lg.Model, + WithPrompt(lg.Prompt), WithMaxTokens(lg.MaxTokens), WithTransportRetry()) if err != nil { lg.transportErr.Add(1) return diff --git a/test/e2e/utils/scaling.go b/test/e2e/utils/scaling.go index ee399e3..8cb0751 100644 --- a/test/e2e/utils/scaling.go +++ b/test/e2e/utils/scaling.go @@ -19,11 +19,11 @@ package utils import ( "context" "fmt" + "sort" "strconv" "strings" "time" - coordinationv1 "k8s.io/api/coordination/v1" corev1 "k8s.io/api/core/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -32,6 +32,8 @@ import ( "k8s.io/client-go/dynamic" "k8s.io/client-go/kubernetes" "sigs.k8s.io/yaml" + + "github.com/kaito-project/production-stack/test/e2e/deploy" ) // KEDA default parameters (applied when the ScaledObject leaves them unset). @@ -54,10 +56,13 @@ type KEDAParams struct { Threshold int PollingInterval time.Duration CooldownPeriod time.Duration + InitialCooldownPeriod time.Duration + ScaleUpRatePeriod time.Duration + ScaleDownRatePeriod time.Duration ScaleUpStabilization time.Duration ScaleDownStabilization time.Duration - ScaleUpTotalWait time.Duration // polling + upStabilization + margin - ScaleDownTotalWait time.Duration // polling + downStabilization + margin + ScaleUpTotalWait time.Duration // polling + max(stabilization, rate period) + margin + ScaleDownTotalWait time.Duration // polling + max(stabilization, rate period) + margin } // GetKEDAParams fetches the ScaledObject managed for the given InferenceSet @@ -74,27 +79,7 @@ func GetKEDAParams(ctx context.Context, model, namespace string) (KEDAParams, er return KEDAParams{}, fmt.Errorf("get ScaledObject %s/%s: %w", namespace, model, err) } - p := KEDAParams{ - PollingInterval: DefaultKEDAPollingInterval, - CooldownPeriod: DefaultKEDACooldownPeriod, - ScaleUpStabilization: DefaultKEDAScaleUpStabilize, - ScaleDownStabilization: DefaultKEDAScaleDownStabilize, - } - - if v, found, _ := unstructured.NestedInt64(so.Object, "spec", "pollingInterval"); found { - p.PollingInterval = time.Duration(v) * time.Second - } - if v, found, _ := unstructured.NestedInt64(so.Object, "spec", "cooldownPeriod"); found { - p.CooldownPeriod = time.Duration(v) * time.Second - } - if v, found, _ := unstructured.NestedInt64(so.Object, "spec", "advanced", - "horizontalPodAutoscalerConfig", "behavior", "scaleUp", "stabilizationWindowSeconds"); found { - p.ScaleUpStabilization = time.Duration(v) * time.Second - } - if v, found, _ := unstructured.NestedInt64(so.Object, "spec", "advanced", - "horizontalPodAutoscalerConfig", "behavior", "scaleDown", "stabilizationWindowSeconds"); found { - p.ScaleDownStabilization = time.Duration(v) * time.Second - } + p := scalingTimingParams(so.Object) // Threshold is read from the first trigger's metadata. triggers, found, _ := unstructured.NestedSlice(so.Object, "spec", "triggers") @@ -114,11 +99,6 @@ func GetKEDAParams(ctx context.Context, model, namespace string) (KEDAParams, er } } if p.Threshold == 0 { - // Fallback to the annotation on the InferenceSet. Since - // keda-kaito-scaler v0.6.1 the composite (multi-metric) config is - // carried in a single scaledobject.kaito.sh/metrics YAML list - // annotation; the per-replica scale-up threshold for the first - // metric lives under that list's first entry `upthreshold`. is, err := dynClient.Resource(InferenceSetGVR).Namespace(namespace).Get(ctx, model, metav1.GetOptions{}) if err == nil { if v, ok := is.GetAnnotations()["scaledobject.kaito.sh/metrics"]; ok { @@ -132,11 +112,53 @@ func GetKEDAParams(ctx context.Context, model, namespace string) (KEDAParams, er } } } + return p, nil +} + +func scalingTimingParams(object map[string]interface{}) KEDAParams { + p := KEDAParams{ + PollingInterval: DefaultKEDAPollingInterval, + CooldownPeriod: DefaultKEDACooldownPeriod, + ScaleUpStabilization: DefaultKEDAScaleUpStabilize, + ScaleDownStabilization: DefaultKEDAScaleDownStabilize, + } + + if v, found, _ := unstructured.NestedInt64(object, "spec", "pollingInterval"); found { + p.PollingInterval = time.Duration(v) * time.Second + } + if v, found, _ := unstructured.NestedInt64(object, "spec", "cooldownPeriod"); found { + p.CooldownPeriod = time.Duration(v) * time.Second + } + if v, found, _ := unstructured.NestedInt64(object, "spec", "initialCooldownPeriod"); found { + p.InitialCooldownPeriod = time.Duration(v) * time.Second + } + if v, found, _ := unstructured.NestedInt64(object, "spec", "advanced", + "horizontalPodAutoscalerConfig", "behavior", "scaleUp", "stabilizationWindowSeconds"); found { + p.ScaleUpStabilization = time.Duration(v) * time.Second + } + if v, found, _ := unstructured.NestedInt64(object, "spec", "advanced", + "horizontalPodAutoscalerConfig", "behavior", "scaleDown", "stabilizationWindowSeconds"); found { + p.ScaleDownStabilization = time.Duration(v) * time.Second + } + + for direction, period := range map[string]*time.Duration{ + "scaleUp": &p.ScaleUpRatePeriod, "scaleDown": &p.ScaleDownRatePeriod, + } { + policies, _, _ := unstructured.NestedSlice(object, "spec", "advanced", + "horizontalPodAutoscalerConfig", "behavior", direction, "policies") + for _, policy := range policies { + if entry, ok := policy.(map[string]interface{}); ok { + if seconds, ok := entry["periodSeconds"].(int64); ok { + *period = max(*period, time.Duration(seconds)*time.Second) + } + } + } + } const margin = 15 * time.Second - p.ScaleUpTotalWait = p.PollingInterval + p.ScaleUpStabilization + margin - p.ScaleDownTotalWait = p.PollingInterval + p.ScaleDownStabilization + margin - return p, nil + p.ScaleUpTotalWait = p.PollingInterval + max(p.ScaleUpStabilization, p.ScaleUpRatePeriod) + margin + p.ScaleDownTotalWait = p.PollingInterval + max(p.ScaleDownStabilization, p.ScaleDownRatePeriod) + margin + return p } // GetInferenceSetReplicas returns the current .spec.replicas value of the @@ -157,6 +179,24 @@ func GetInferenceSetReplicas(ctx context.Context, model, namespace string) (int3 return int32(v), nil } +// GetInferenceSetReadyReplicas returns .status.readyReplicas of the named +// InferenceSet; an absent field (omitempty) reads as 0. +func GetInferenceSetReadyReplicas(ctx context.Context, model, namespace string) (int32, error) { + dynClient, err := GetDynamicClient() + if err != nil { + return 0, err + } + is, err := dynClient.Resource(InferenceSetGVR).Namespace(namespace).Get(ctx, model, metav1.GetOptions{}) + if err != nil { + return 0, fmt.Errorf("get InferenceSet %s/%s: %w", namespace, model, err) + } + v, _, err := unstructured.NestedInt64(is.Object, "status", "readyReplicas") + if err != nil { + return 0, fmt.Errorf("InferenceSet %s/%s: read .status.readyReplicas: %w", namespace, model, err) + } + return int32(v), nil +} + // SetInferenceSetReplicas patches the InferenceSet's .spec.replicas directly. // // Reserved for KEDA-managed deployments (EnableScaling), where KEDA owns @@ -183,161 +223,212 @@ func SetInferenceSetReplicas(ctx context.Context, model, namespace string, repli return nil } -// ScalingInventory captures the set of infrastructure objects that participate -// in an InferenceSet scale event. Snapshotting it before and after lets tests -// reason about what was created or removed by the scaling pipeline. -type ScalingInventory struct { - // FakeNodeNames lists Node names where kaito.sh/fake-node=true and the - // node's kaito.sh/workspace label starts with the given model name. - FakeNodeNames []string - // ShadowPodNames lists shadow pod names serving this model. - ShadowPodNames []string - // NodeClaimNames lists NodeClaim names whose kaito.sh/workspace label - // starts with the given model name. - NodeClaimNames []string - // LeaseNames lists Lease names in kube-node-lease matching FakeNodeNames. - LeaseNames []string -} +// The helpers below describe a scale event only through objects that exist +// identically on the GPU node mocker and on real Karpenter-provisioned GPUs: +// the InferenceSet's model pods, the nodes they run on, and the NodeClaims +// whose status.nodeName points at those nodes. On the mocker the model pod's +// IP and Ready condition are patched from its shadow pod, so scraping the +// model pod reaches the simulator. -// SnapshotScalingInventory captures the current state of fake nodes, shadow -// pods, NodeClaims, and node leases attributable to the given model. -// -// Filtering uses the kaito.sh/workspace label propagated from NodeClaim → -// fake Node, and matches any workspace name that starts with model+"-" (KAITO -// names workspaces -). -func SnapshotScalingInventory(ctx context.Context, clientset *kubernetes.Clientset, dynClient dynamic.Interface, namespace, model string) (ScalingInventory, error) { - inv := ScalingInventory{} - wsPrefix := model + "-" - workspaceMatches := func(ws string) bool { - return ws == model || strings.HasPrefix(ws, wsPrefix) - } - - // Fake nodes. - nodes, err := clientset.CoreV1().Nodes().List(ctx, metav1.ListOptions{ - LabelSelector: "kaito.sh/fake-node=true", +// DefaultModelServerPort is the vLLM port KAITO serves inference and /metrics +// on; used when a model pod declares no container port. +const DefaultModelServerPort = 5000 + +// ListModelPods returns the InferenceSet's model pods that are not being +// deleted, sorted by name. +func ListModelPods(ctx context.Context, clientset kubernetes.Interface, values deploy.ModelDeploymentValues) ([]corev1.Pod, error) { + list, err := clientset.CoreV1().Pods(values.Namespace).List(ctx, metav1.ListOptions{ + LabelSelector: values.InferencePodSelector(), }) if err != nil { - return inv, fmt.Errorf("list fake nodes: %w", err) + return nil, fmt.Errorf("list model pods for %s/%s: %w", values.Namespace, values.Name, err) } - for _, n := range nodes.Items { - ws := n.Labels["kaito.sh/workspace"] - if workspaceMatches(ws) { - inv.FakeNodeNames = append(inv.FakeNodeNames, n.Name) + pods := make([]corev1.Pod, 0, len(list.Items)) + for _, pod := range list.Items { + if pod.DeletionTimestamp == nil { + pods = append(pods, pod) } } + sort.Slice(pods, func(i, j int) bool { return pods[i].Name < pods[j].Name }) + return pods, nil +} - // Shadow pods (already model-scoped via GetShadowPodsForModel). - pods, err := GetShadowPodsForModel(ctx, clientset, namespace, model) - if err != nil && !strings.Contains(err.Error(), "no running shadow pods found") { - return inv, err +// IsPodReady reports whether the pod is Running with Ready=True. +func IsPodReady(pod corev1.Pod) bool { + if pod.Status.Phase != corev1.PodRunning { + return false } - for _, p := range pods { - inv.ShadowPodNames = append(inv.ShadowPodNames, p.Name) + for _, c := range pod.Status.Conditions { + if c.Type == corev1.PodReady { + return c.Status == corev1.ConditionTrue + } } + return false +} - // NodeClaims. - ncList, err := dynClient.Resource(NodeClaimGVR).List(ctx, metav1.ListOptions{}) - if err != nil { - return inv, fmt.Errorf("list NodeClaims: %w", err) +// ReadyPods returns the subset of pods that are Ready. +func ReadyPods(pods []corev1.Pod) []corev1.Pod { + var ready []corev1.Pod + for _, pod := range pods { + if IsPodReady(pod) { + ready = append(ready, pod) + } + } + return ready +} + +// PodPlacements maps pod name to the node it is scheduled on (empty when +// unscheduled). +func PodPlacements(pods []corev1.Pod) map[string]string { + placements := make(map[string]string, len(pods)) + for _, pod := range pods { + placements[pod.Name] = pod.Spec.NodeName } - for _, nc := range ncList.Items { - ws := nc.GetLabels()["kaito.sh/workspace"] - if workspaceMatches(ws) { - inv.NodeClaimNames = append(inv.NodeClaimNames, nc.GetName()) + return placements +} + +// ModelServerPort returns the first declared container port of the pod, or +// DefaultModelServerPort when none is declared. +func ModelServerPort(pod corev1.Pod) int { + for _, c := range pod.Spec.Containers { + for _, p := range c.Ports { + if p.ContainerPort > 0 { + return int(p.ContainerPort) + } } } + return DefaultModelServerPort +} - // Leases matching fake node names. - for _, nodeName := range inv.FakeNodeNames { - lease := &coordinationv1.Lease{} - _, err := clientset.CoordinationV1().Leases("kube-node-lease"). - Get(ctx, nodeName, metav1.GetOptions{}) - if err == nil { - inv.LeaseNames = append(inv.LeaseNames, nodeName) +// SumMetricValues sums every sample of metricName whose labels include the +// given subset. vLLM splits some counters across extra labels (e.g. +// vllm:request_success_total by finished_reason), so the first match alone +// would undercount. +func SumMetricValues(metricsText, metricName string, labels map[string]string) (float64, bool) { + var ( + sum float64 + found bool + ) + for _, line := range strings.Split(metricsText, "\n") { + line = strings.TrimSpace(line) + m := metricsLineRegex.FindStringSubmatch(line) + if m == nil || m[1] != metricName || !labelsMatch(m[2], labels) { continue } - if !apierrors.IsNotFound(err) { - return inv, fmt.Errorf("get lease for %s: %w", nodeName, err) + v, err := strconv.ParseFloat(m[3], 64) + if err != nil { + continue } - _ = lease // silence unused + sum += v + found = true } - return inv, nil + return sum, found } -// DiffInventory returns (added, removed) lists for each sub-collection. -type InventoryDiff struct { - AddedFakeNodes, RemovedFakeNodes []string - AddedShadowPods, RemovedShadowPods []string - AddedNodeClaims, RemovedNodeClaims []string - AddedLeases, RemovedLeases []string +// ScrapeModelPodMetric scrapes metricName (labelled model_name=) +// from every Ready model pod via the API server pod proxy and returns +// pod name → value. It fails if there is no Ready pod, or if any Ready pod +// cannot be scraped or does not export the metric, so a broken replica cannot +// hide behind a healthy one. +func ScrapeModelPodMetric(ctx context.Context, clientset *kubernetes.Clientset, values deploy.ModelDeploymentValues, metricName string) (PodMetricSnapshot, error) { + pods, err := ListModelPods(ctx, clientset, values) + if err != nil { + return nil, err + } + ready := ReadyPods(pods) + if len(ready) == 0 { + return nil, fmt.Errorf("no Ready model pods for %s/%s", values.Namespace, values.Name) + } + snapshot := make(PodMetricSnapshot, len(ready)) + var missing []string + for _, pod := range ready { + raw, err := ScrapePodMetrics(ctx, clientset, values.Namespace, pod.Name, ModelServerPort(pod)) + if err != nil { + return nil, err + } + v, found := SumMetricValues(raw, metricName, map[string]string{"model_name": values.Name}) + if !found { + missing = append(missing, pod.Name) + continue + } + snapshot[pod.Name] = v + } + if len(missing) > 0 { + return nil, fmt.Errorf("%s{model_name=%q} missing on Ready model pods %v", metricName, values.Name, missing) + } + return snapshot, nil } -func DiffInventory(before, after ScalingInventory) InventoryDiff { - d := InventoryDiff{} - d.AddedFakeNodes, d.RemovedFakeNodes = diffStringSlices(before.FakeNodeNames, after.FakeNodeNames) - d.AddedShadowPods, d.RemovedShadowPods = diffStringSlices(before.ShadowPodNames, after.ShadowPodNames) - d.AddedNodeClaims, d.RemovedNodeClaims = diffStringSlices(before.NodeClaimNames, after.NodeClaimNames) - d.AddedLeases, d.RemovedLeases = diffStringSlices(before.LeaseNames, after.LeaseNames) - return d +// ListNodeClaims returns all Karpenter NodeClaims in the cluster. +func ListNodeClaims(ctx context.Context, dynClient dynamic.Interface) ([]unstructured.Unstructured, error) { + list, err := dynClient.Resource(NodeClaimGVR).List(ctx, metav1.ListOptions{}) + if err != nil { + return nil, fmt.Errorf("list NodeClaims: %w", err) + } + return list.Items, nil } -func diffStringSlices(before, after []string) (added, removed []string) { - bm := make(map[string]struct{}, len(before)) - for _, s := range before { - bm[s] = struct{}{} - } - am := make(map[string]struct{}, len(after)) - for _, s := range after { - am[s] = struct{}{} +// NodeClaimNames returns the sorted names of the NodeClaims. +func NodeClaimNames(nodeClaims []unstructured.Unstructured) []string { + names := make([]string, 0, len(nodeClaims)) + for _, nc := range nodeClaims { + names = append(names, nc.GetName()) } - for s := range am { - if _, ok := bm[s]; !ok { - added = append(added, s) - } - } - for s := range bm { - if _, ok := am[s]; !ok { - removed = append(removed, s) + sort.Strings(names) + return names +} + +// NodeClaimsByNode maps status.nodeName → NodeClaim name for the NodeClaims +// that have registered a node. Both the mocker and Karpenter set the field. +func NodeClaimsByNode(nodeClaims []unstructured.Unstructured) map[string]string { + byNode := make(map[string]string, len(nodeClaims)) + for _, nc := range nodeClaims { + node, _, _ := unstructured.NestedString(nc.Object, "status", "nodeName") + if node != "" { + byNode[node] = nc.GetName() } } - return added, removed + return byNode } -// IsFakeNodeReady returns true when the Node has Ready=True and no -// unreachable taint. -func IsFakeNodeReady(node *corev1.Node) bool { - for _, t := range node.Spec.Taints { - if t.Key == "node.kubernetes.io/unreachable" { - return false - } +// NewNodeClaimNames returns the sorted names in current that are not in +// before. Callers snapshot before installing a case and use the result to +// find every NodeClaim the case caused, including ones that never registered +// a node. +func NewNodeClaimNames(before []string, current []unstructured.Unstructured) []string { + seen := make(map[string]struct{}, len(before)) + for _, name := range before { + seen[name] = struct{}{} } - for _, c := range node.Status.Conditions { - if c.Type == corev1.NodeReady && c.Status == corev1.ConditionTrue { - return true + var added []string + for _, name := range NodeClaimNames(current) { + if _, ok := seen[name]; !ok { + added = append(added, name) } } - return false + return added } -// CountInferencePoolReadyPods returns the number of Running shadow pods -// (which mirror the InferencePool's inference pods) that are Ready for -// the given modeldeployment in the given namespace. -func CountInferencePoolReadyPods(ctx context.Context, clientset *kubernetes.Clientset, model, namespace string) (int, error) { - pods, err := GetShadowPodsForModel(ctx, clientset, namespace, model) - if err != nil { - if strings.Contains(err.Error(), "no running shadow pods found") { - return 0, nil +// EnsureConfigMap creates the ConfigMap or replaces its data. +func EnsureConfigMap(ctx context.Context, clientset kubernetes.Interface, namespace, name string, data map[string]string) error { + configMaps := clientset.CoreV1().ConfigMaps(namespace) + existing, err := configMaps.Get(ctx, name, metav1.GetOptions{}) + if apierrors.IsNotFound(err) { + _, err = configMaps.Create(ctx, &corev1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: namespace}, + Data: data, + }, metav1.CreateOptions{}) + if err != nil { + return fmt.Errorf("create ConfigMap %s/%s: %w", namespace, name, err) } - return 0, err + return nil } - ready := 0 - for _, p := range pods { - for _, c := range p.Status.Conditions { - if c.Type == corev1.PodReady && c.Status == corev1.ConditionTrue { - ready++ - break - } - } + if err != nil { + return fmt.Errorf("get ConfigMap %s/%s: %w", namespace, name, err) + } + existing.Data = data + if _, err := configMaps.Update(ctx, existing, metav1.UpdateOptions{}); err != nil { + return fmt.Errorf("update ConfigMap %s/%s: %w", namespace, name, err) } - return ready, nil + return nil } diff --git a/test/e2e/utils/scaling_test.go b/test/e2e/utils/scaling_test.go new file mode 100644 index 0000000..d9bc1ca --- /dev/null +++ b/test/e2e/utils/scaling_test.go @@ -0,0 +1,224 @@ +/* +Copyright 2026 The KAITO Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package utils + +import ( + "context" + "reflect" + "testing" + "time" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/client-go/kubernetes/fake" + + "github.com/kaito-project/production-stack/test/e2e/deploy" +) + +func TestScalingTimingParams(t *testing.T) { + so := map[string]interface{}{} + for _, field := range []struct { + path []string + value int64 + }{ + {[]string{"spec", "pollingInterval"}, 15}, + {[]string{"spec", "cooldownPeriod"}, 60}, + {[]string{"spec", "initialCooldownPeriod"}, 120}, + {[]string{"spec", "advanced", "horizontalPodAutoscalerConfig", "behavior", "scaleUp", "stabilizationWindowSeconds"}, 60}, + {[]string{"spec", "advanced", "horizontalPodAutoscalerConfig", "behavior", "scaleDown", "stabilizationWindowSeconds"}, 300}, + } { + if err := unstructured.SetNestedField(so, field.value, field.path...); err != nil { + t.Fatal(err) + } + } + for direction, seconds := range map[string]int64{"scaleUp": 900, "scaleDown": 1800} { + if err := unstructured.SetNestedSlice(so, []interface{}{map[string]interface{}{"periodSeconds": seconds}}, + "spec", "advanced", "horizontalPodAutoscalerConfig", "behavior", direction, "policies"); err != nil { + t.Fatal(err) + } + } + p := scalingTimingParams(so) + if p.InitialCooldownPeriod != 120*time.Second || p.CooldownPeriod != time.Minute { + t.Fatalf("initial and ordinary cooldowns must remain independent: %+v", p) + } + if p.ScaleUpTotalWait != 930*time.Second || p.ScaleDownTotalWait != 1830*time.Second { + t.Fatalf("wait budgets must include the longer rate-limit period: %+v", p) + } + // Explicit zero must not fall back to a nonzero default. + if err := unstructured.SetNestedField(so, int64(0), "spec", "initialCooldownPeriod"); err != nil { + t.Fatal(err) + } + if scalingTimingParams(so).InitialCooldownPeriod != 0 { + t.Fatal("explicit initial cooldown zero was lost") + } +} + +func modelPod(name, node string, phase corev1.PodPhase, ready corev1.ConditionStatus, labels map[string]string) *corev1.Pod { + return &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: "ns", Labels: labels}, + Spec: corev1.PodSpec{NodeName: node}, + Status: corev1.PodStatus{ + Phase: phase, + Conditions: []corev1.PodCondition{{Type: corev1.PodReady, Status: ready}}, + }, + } +} + +func TestListModelPodsSelectsTheInferenceSetAndSkipsTerminatingPods(t *testing.T) { + values := deploy.ModelDeploymentValues{Name: "scaling-phi", Namespace: "ns"} + own := map[string]string{"inferenceset.kaito.sh/created-by": "scaling-phi"} + terminating := modelPod("scaling-phi-c-0", "n3", corev1.PodRunning, corev1.ConditionTrue, own) + now := metav1.Now() + terminating.DeletionTimestamp = &now + terminating.Finalizers = []string{"test"} + clientset := fake.NewSimpleClientset( + modelPod("scaling-phi-b-0", "n2", corev1.PodRunning, corev1.ConditionTrue, own), + modelPod("scaling-phi-a-0", "n1", corev1.PodPending, corev1.ConditionFalse, own), + terminating, + modelPod("other-0", "n1", corev1.PodRunning, corev1.ConditionTrue, + map[string]string{"inferenceset.kaito.sh/created-by": "other"}), + modelPod("shadow", "n9", corev1.PodRunning, corev1.ConditionTrue, + map[string]string{"kaito.sh/shadow-pod-inferenceset": "scaling-phi"}), + ) + + pods, err := ListModelPods(context.Background(), clientset, values) + if err != nil { + t.Fatal(err) + } + var names []string + for _, p := range pods { + names = append(names, p.Name) + } + if want := []string{"scaling-phi-a-0", "scaling-phi-b-0"}; !reflect.DeepEqual(names, want) { + t.Fatalf("pods=%v, want %v", names, want) + } + if got := PodPlacements(pods); !reflect.DeepEqual(got, map[string]string{"scaling-phi-a-0": "n1", "scaling-phi-b-0": "n2"}) { + t.Fatalf("placements=%v", got) + } + ready := ReadyPods(pods) + if len(ready) != 1 || ready[0].Name != "scaling-phi-b-0" { + t.Fatalf("ready=%v", ready) + } +} + +func TestIsPodReady(t *testing.T) { + for _, test := range []struct { + name string + pod *corev1.Pod + ready bool + }{ + {"running and ready", modelPod("p", "", corev1.PodRunning, corev1.ConditionTrue, nil), true}, + {"running not ready", modelPod("p", "", corev1.PodRunning, corev1.ConditionFalse, nil), false}, + {"ready condition but pending", modelPod("p", "", corev1.PodPending, corev1.ConditionTrue, nil), false}, + {"no conditions", &corev1.Pod{Status: corev1.PodStatus{Phase: corev1.PodRunning}}, false}, + } { + if got := IsPodReady(*test.pod); got != test.ready { + t.Errorf("%s: IsPodReady=%v, want %v", test.name, got, test.ready) + } + } +} + +func TestModelServerPort(t *testing.T) { + withPorts := func(containers ...corev1.Container) corev1.Pod { + return corev1.Pod{Spec: corev1.PodSpec{Containers: containers}} + } + for _, test := range []struct { + name string + pod corev1.Pod + want int + }{ + {"no ports", withPorts(corev1.Container{Name: "a"}), DefaultModelServerPort}, + {"first declared port", withPorts(corev1.Container{Ports: []corev1.ContainerPort{{ContainerPort: 5000}, {ContainerPort: 5557}}}), 5000}, + {"port on a later container", withPorts(corev1.Container{Name: "a"}, corev1.Container{Ports: []corev1.ContainerPort{{ContainerPort: 8001}}}), 8001}, + } { + if got := ModelServerPort(test.pod); got != test.want { + t.Errorf("%s: port=%d, want %d", test.name, got, test.want) + } + } +} + +func TestSumMetricValuesAddsEverySeriesForTheModel(t *testing.T) { + text := `# HELP vllm:request_success_total Count of successfully processed requests. +vllm:request_success_total{engine="0",finished_reason="stop",model_name="scaling-phi"} 3.0 +vllm:request_success_total{engine="0",finished_reason="length",model_name="scaling-phi"} 7.0 +vllm:request_success_total{engine="0",finished_reason="stop",model_name="other"} 100.0 +vllm:num_requests_waiting{engine="0",model_name="scaling-phi"} 0.0 +` + labels := map[string]string{"model_name": "scaling-phi"} + if got, found := SumMetricValues(text, "vllm:request_success_total", labels); !found || got != 10 { + t.Fatalf("success=%v found=%v, want 10 true", got, found) + } + if got, found := SumMetricValues(text, "vllm:num_requests_waiting", labels); !found || got != 0 { + t.Fatalf("waiting=%v found=%v, want present zero", got, found) + } + if _, found := SumMetricValues(text, "vllm:num_requests_running", labels); found { + t.Fatal("absent metric reported as found") + } + if _, found := SumMetricValues(text, "vllm:request_success_total", map[string]string{"model_name": "missing"}); found { + t.Fatal("metric for another model reported as found") + } +} + +func nodeClaim(name, nodeName string) unstructured.Unstructured { + nc := unstructured.Unstructured{Object: map[string]interface{}{}} + nc.SetName(name) + if nodeName != "" { + _ = unstructured.SetNestedField(nc.Object, nodeName, "status", "nodeName") + } + return nc +} + +func TestNodeClaimHelpers(t *testing.T) { + claims := []unstructured.Unstructured{ + nodeClaim("e2e-scaling-ws-b-xyz", "node-b"), + nodeClaim("system-surge-1", "node-sys"), + nodeClaim("e2e-scaling-ws-a-abc", ""), + } + if got, want := NodeClaimNames(claims), []string{"e2e-scaling-ws-a-abc", "e2e-scaling-ws-b-xyz", "system-surge-1"}; !reflect.DeepEqual(got, want) { + t.Fatalf("names=%v, want %v", got, want) + } + if got, want := NodeClaimsByNode(claims), map[string]string{"node-b": "e2e-scaling-ws-b-xyz", "node-sys": "system-surge-1"}; !reflect.DeepEqual(got, want) { + t.Fatalf("byNode=%v, want %v (claims without a node must be skipped)", got, want) + } + // Claims that existed before the case are not attributed to it; a + // still-launching claim without a node is. + if got, want := NewNodeClaimNames([]string{"system-surge-1"}, claims), []string{"e2e-scaling-ws-a-abc", "e2e-scaling-ws-b-xyz"}; !reflect.DeepEqual(got, want) { + t.Fatalf("new=%v, want %v", got, want) + } + if got := NewNodeClaimNames(NodeClaimNames(claims), claims); len(got) != 0 { + t.Fatalf("no new claims expected, got %v", got) + } +} + +func TestEnsureConfigMapCreatesThenReplacesData(t *testing.T) { + ctx := context.Background() + clientset := fake.NewSimpleClientset() + if err := EnsureConfigMap(ctx, clientset, "ns", "cfg", map[string]string{"a": "1"}); err != nil { + t.Fatal(err) + } + if err := EnsureConfigMap(ctx, clientset, "ns", "cfg", map[string]string{"b": "2"}); err != nil { + t.Fatal(err) + } + cm, err := clientset.CoreV1().ConfigMaps("ns").Get(ctx, "cfg", metav1.GetOptions{}) + if err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(cm.Data, map[string]string{"b": "2"}) { + t.Fatalf("data=%v, want replaced", cm.Data) + } +}