From 6046c7432003ed07f2a445d97dd9ded8ef2c9870 Mon Sep 17 00:00:00 2001 From: Aakash Dantre Date: Tue, 3 Nov 2026 13:23:25 +0000 Subject: [PATCH] test(otel): add SGLang solution integration tests Adds the SGLang suite and SGLang workloads to the LLM serving root, so vLLM and SGLang are tested side by side under KServe and as plain Deployments. - Deploy an SGLang InferenceService and a plain SGLang Deployment (no declared port, service.name annotation) from SGLang's CPU image, and send them load - Add test/otel/solutions/sglang: metrics, scope and labels, InferenceService and annotation attribution, the 30000 default port, one series per pod, sglang:* only, counters increase - Add TestModelServerRouting: each of the four servers reaches only its own engine pipeline, with positive controls - Run the sglang suite after test_dir, and require the solution in the cloned chart --- terraform/eks/daemon/otel-llm-serving/main.tf | 150 ++++++++++- .../eks/daemon/otel-llm-serving/variables.tf | 9 +- test/otel/solutions/sglang/metrics_test.go | 34 +++ test/otel/solutions/sglang/setup_test.go | 104 ++++++++ test/otel/solutions/sglang/sglang_test.go | 244 ++++++++++++++++++ 5 files changed, 533 insertions(+), 8 deletions(-) create mode 100644 test/otel/solutions/sglang/metrics_test.go create mode 100644 test/otel/solutions/sglang/setup_test.go create mode 100644 test/otel/solutions/sglang/sglang_test.go diff --git a/terraform/eks/daemon/otel-llm-serving/main.tf b/terraform/eks/daemon/otel-llm-serving/main.tf index 7720eaa7..8037f27a 100644 --- a/terraform/eks/daemon/otel-llm-serving/main.tf +++ b/terraform/eks/daemon/otel-llm-serving/main.tf @@ -9,11 +9,13 @@ # - its transformer (hello-world) -> knative dataplane only, no aws.service.type # - a plain vLLM Deployment -> vllm (found by image name) # - a plain Knative Service (hello-ksvc) -> knative dataplane only, no aws.service.type +# - an SGLang InferenceService -> sglang (shared KServe job), knative dataplane +# - a plain SGLang Deployment -> sglang (found by image name) # - an annotated vLLM mock (custom image) -> negative control: not scraped # - an annotated non-vLLM exporter -> negative control: not scraped # - a load generator, so request metrics have data # then installs the chart with otelContainerInsights enabled (solutions default -# on) and runs ./test/otel/solutions/{vllm,kserve,knative}. +# on) and runs ./test/otel/solutions/{vllm,sglang,kserve,knative}. module "common" { source = "../../../common" @@ -27,9 +29,10 @@ module "basic_components" { } locals { - aws_eks = "aws eks --region ${var.region}" - namespace = "llm" - isvc = "vllm-isvc" + aws_eks = "aws eks --region ${var.region}" + namespace = "llm" + isvc = "vllm-isvc" + sglang_isvc = "sglang-isvc" } resource "aws_eks_cluster" "this" { @@ -371,6 +374,132 @@ resource "kubernetes_deployment_v1" "vllm_standalone" { timeouts { create = "30m" } } +# SGLang in a KServe predictor. It carries the same KServe labels as the vLLM +# predictor; the shared kserve-model-server job scrapes both, and only the +# sglang pipeline keeps its sglang:* series. +resource "null_resource" "sglang_inference_service" { + depends_on = [helm_release.kserve, kubernetes_namespace_v1.llm] + provisioner "local-exec" { + command = <<-EOT + set -e + cat <<'EOF' | kubectl apply -f - + apiVersion: serving.kserve.io/v1beta1 + kind: InferenceService + metadata: + name: ${local.sglang_isvc} + namespace: ${local.namespace} + annotations: + serving.knative.dev/progress-deadline: "1800s" + spec: + predictor: + minReplicas: 1 + maxReplicas: 1 + containers: + - name: kserve-container + image: ${var.sglang_cpu_image} + command: ["python3", "-m", "sglang.launch_server"] + args: ["--model-path", "${var.vllm_model}", "--served-model-name", "${local.sglang_isvc}", "--device", "cpu", + "--host", "0.0.0.0", "--port", "8080", "--enable-metrics", + "--mem-fraction-static", "0.85", "--context-length", "2048"] + ports: + - { containerPort: 8080, protocol: TCP } + # SGLang's /health takes about a second, the probe default timeout, + # so the queue-proxy would never become ready without a longer one. + readinessProbe: + httpGet: { path: /health, port: 8080 } + periodSeconds: 10 + timeoutSeconds: 5 + failureThreshold: 120 + resources: + requests: { cpu: "2", memory: 6Gi } + limits: { cpu: "4", memory: 10Gi } + volumeMounts: + - { name: dshm, mountPath: /dev/shm } + volumes: + - name: dshm + emptyDir: { medium: Memory, sizeLimit: 2Gi } + EOF + kubectl -n ${local.namespace} wait --for=condition=Ready inferenceservice/${local.sglang_isvc} --timeout=1800s + EOT + } + provisioner "local-exec" { + when = destroy + command = "kubectl -n llm delete inferenceservice sglang-isvc --timeout=120s 2>/dev/null || true" + } +} + +# A plain SGLang Deployment, found by its image. It declares no port, so the +# chart falls back to sglang.launch_server's 30000, and it sets the OTel +# service.name annotation, which wins over the workload name. +resource "kubernetes_deployment_v1" "sglang_standalone" { + depends_on = [kubernetes_namespace_v1.llm] + metadata { + name = "sglang-standalone" + namespace = local.namespace + } + spec { + replicas = 1 + selector { match_labels = { app = "sglang-standalone" } } + template { + metadata { + labels = { app = "sglang-standalone" } + annotations = { "resource.opentelemetry.io/service.name" = "sglang-chat" } + } + spec { + container { + name = "sglang" + image = var.sglang_cpu_image + command = ["python3", "-m", "sglang.launch_server"] + # On CPU the memory fraction is of host memory; the default leaves the + # KV cache no room. + args = ["--model-path", var.vllm_model, "--served-model-name", "sglang-chat", "--device", "cpu", "--host", "0.0.0.0", + "--enable-metrics", "--mem-fraction-static", "0.85", "--context-length", "2048"] + resources { + requests = { cpu = "2", memory = "6Gi" } + limits = { memory = "10Gi" } + } + readiness_probe { + http_get { + path = "/health" + port = 30000 + } + period_seconds = 10 + timeout_seconds = 5 + } + volume_mount { + name = "dshm" + mount_path = "/dev/shm" + } + } + volume { + name = "dshm" + empty_dir { + medium = "Memory" + size_limit = "2Gi" + } + } + } + } + } + wait_for_rollout = false + timeouts { create = "30m" } +} + +resource "kubernetes_service_v1" "sglang_standalone" { + depends_on = [kubernetes_namespace_v1.llm] + metadata { + name = "sglang-standalone" + namespace = local.namespace + } + spec { + selector = { app = "sglang-standalone" } + port { + port = 30000 + target_port = 30000 + } + } +} + # Static metrics served by busybox httpd from a ConfigMap. resource "kubernetes_config_map_v1" "mock_metrics" { depends_on = [kubernetes_namespace_v1.llm] @@ -448,7 +577,8 @@ resource "kubernetes_deployment_v1" "mock" { # Requests through the InferenceService (Knative queue-proxy) and directly to # the standalone server, so counters and histograms on both have data. resource "kubernetes_deployment_v1" "loadgen" { - depends_on = [null_resource.inference_service, kubernetes_deployment_v1.vllm_standalone, null_resource.plain_knative_service] + depends_on = [null_resource.inference_service, kubernetes_deployment_v1.vllm_standalone, null_resource.plain_knative_service, + null_resource.sglang_inference_service, kubernetes_deployment_v1.sglang_standalone] metadata { name = "loadgen" namespace = local.namespace @@ -469,6 +599,12 @@ resource "kubernetes_deployment_v1" "loadgen" { http://vllm-standalone.${local.namespace}.svc.cluster.local:8000/v1/completions; do curl -s -o /dev/null -m 60 -H 'Content-Type: application/json' -d "$body" "$url" done + curl -s -o /dev/null -m 60 -H 'Content-Type: application/json' \ + -d '{"model":"${local.sglang_isvc}","prompt":"Hello","max_tokens":8}' \ + http://${local.sglang_isvc}-predictor.${local.namespace}.svc.cluster.local/v1/completions + curl -s -o /dev/null -m 60 -H 'Content-Type: application/json' \ + -d '{"model":"sglang-chat","prompt":"Hello","max_tokens":8}' \ + http://sglang-standalone.${local.namespace}.svc.cluster.local:30000/v1/completions # Request metrics for the transformer and plain Knative revisions. for url in http://${local.isvc}-transformer.${local.namespace}.svc.cluster.local/ \ http://hello-ksvc.${local.namespace}.svc.cluster.local/; do @@ -510,7 +646,7 @@ data "external" "clone_helm_chart" { # Charts before the vLLM, KServe and Knative solutions would install fine and # fail every suite; stop here with the reason instead. values=./helm-charts/charts/amazon-cloudwatch-observability/values.yaml - for s in vllm kserve knative; do + for s in vllm sglang kserve knative; do if ! grep -qE "^ +$s:" "$values"; then echo "${var.helm_chart_repo}@${var.helm_chart_branch} has no '$s' solution in values.yaml" >&2 exit 1 @@ -603,7 +739,7 @@ resource "null_resource" "validator" { # same cluster, as terraform/eks/daemon/otel does for Karpenter and KEDA. # Every suite runs even if an earlier one fails; the run fails at the end. failed="" - for suite in ${var.test_dir} ./test/otel/solutions/kserve ./test/otel/solutions/knative; do + for suite in ${var.test_dir} ./test/otel/solutions/sglang ./test/otel/solutions/kserve ./test/otel/solutions/knative; do echo "Running OTEL solutions tests ($suite)..." go test -tags integration -timeout 1h -v $suite/... \ -eksClusterName=${aws_eks_cluster.this.name} \ diff --git a/terraform/eks/daemon/otel-llm-serving/variables.tf b/terraform/eks/daemon/otel-llm-serving/variables.tf index 442f3a19..22d15792 100644 --- a/terraform/eks/daemon/otel-llm-serving/variables.tf +++ b/terraform/eks/daemon/otel-llm-serving/variables.tf @@ -6,7 +6,7 @@ variable "region" { default = "us-west-2" } -# Run first; ./test/otel/solutions/{kserve,knative} always follow. +# Run first; ./test/otel/solutions/{sglang,kserve,knative} always follow. variable "test_dir" { type = string default = "./test/otel/solutions/vllm" @@ -81,6 +81,13 @@ variable "vllm_cpu_image" { default = "public.ecr.aws/q9t5s3a7/vllm-cpu-release-repo:v0.29.0" } +# SGLang's CPU build. It needs AVX-512 (Intel Xeon, which m5 has); on CPUs +# without it, such as AMD EPYC Zen 2, it exits with an illegal instruction. +variable "sglang_cpu_image" { + type = string + default = "lmsysorg/sglang:v0.5.23-xeon" +} + variable "vllm_model" { type = string default = "Qwen/Qwen2.5-0.5B-Instruct" diff --git a/test/otel/solutions/sglang/metrics_test.go b/test/otel/solutions/sglang/metrics_test.go new file mode 100644 index 00000000..967ab188 --- /dev/null +++ b/test/otel/solutions/sglang/metrics_test.go @@ -0,0 +1,34 @@ +//go:build integration + +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: MIT + +package sglang + +import "github.com/aws/amazon-cloudwatch-agent-test/util/otelmetrics" + +// sglangMetrics are SGLang engine metrics expected from every SGLang server the +// chart finds, once it has served requests. The chart keeps sglang:* and drops +// *_created and summaries. Families SGLang creates only for streaming requests +// (inter-token latency, cached tokens) are not listed. +var sglangMetrics = []otelmetrics.MetricDefinition{ + // Scheduler state + {Name: "sglang:num_running_reqs", MetricType: "gauge", Scope: otelmetrics.ScopePod, ExpectedLabels: []string{"model_name", "engine_type"}}, + {Name: "sglang:num_queue_reqs", MetricType: "gauge", Scope: otelmetrics.ScopePod, ExpectedLabels: []string{"model_name", "engine_type"}}, + {Name: "sglang:token_usage", MetricType: "gauge", Scope: otelmetrics.ScopePod, ExpectedLabels: []string{"model_name", "engine_type"}}, + {Name: "sglang:gen_throughput", MetricType: "gauge", Scope: otelmetrics.ScopePod, ExpectedLabels: []string{"model_name", "engine_type"}}, + {Name: "sglang:cache_hit_rate", MetricType: "gauge", Scope: otelmetrics.ScopePod, ExpectedLabels: []string{"model_name", "engine_type"}}, + + // Tokens and requests + {Name: "sglang:prompt_tokens_total", MetricType: "counter", Scope: otelmetrics.ScopePod, ExpectedLabels: []string{"model_name"}}, + {Name: "sglang:generation_tokens_total", MetricType: "counter", Scope: otelmetrics.ScopePod, ExpectedLabels: []string{"model_name"}}, + {Name: "sglang:num_requests_total", MetricType: "counter", Scope: otelmetrics.ScopePod, ExpectedLabels: []string{"model_name"}}, + + // Latency + {Name: "sglang:e2e_request_latency_seconds", MetricType: "histogram", Scope: otelmetrics.ScopePod, ExpectedLabels: []string{"model_name"}, Unit: "s"}, + {Name: "sglang:time_to_first_token_seconds", MetricType: "histogram", Scope: otelmetrics.ScopePod, ExpectedLabels: []string{"model_name"}, Unit: "s"}, + {Name: "sglang:queue_time_seconds", MetricType: "histogram", Scope: otelmetrics.ScopePod, ExpectedLabels: []string{"model_name"}, Unit: "s"}, + + // API server + {Name: "sglang:http_requests_total", MetricType: "counter", Scope: otelmetrics.ScopePod, ExpectedLabels: []string{"endpoint", "method"}}, +} diff --git a/test/otel/solutions/sglang/setup_test.go b/test/otel/solutions/sglang/setup_test.go new file mode 100644 index 00000000..73d663bc --- /dev/null +++ b/test/otel/solutions/sglang/setup_test.go @@ -0,0 +1,104 @@ +//go:build integration + +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: MIT + +package sglang + +import ( + "context" + "flag" + "fmt" + "os" + "testing" + "time" + + awsconfig "github.com/aws/aws-sdk-go-v2/config" + "github.com/aws/aws-sdk-go-v2/service/sts" + + "github.com/aws/amazon-cloudwatch-agent-test/environment" + "github.com/aws/amazon-cloudwatch-agent-test/util/otelmetrics" +) + +var ( + cfg otelmetrics.TestConfig + client *otelmetrics.OtelMetricsClient + queryCache *otelmetrics.QueryCache +) + +// Instrumentation scope and pipeline set by the chart's SGLang solution, and the +// vLLM pipeline that shares the KServe model-server job with it. +const ( + scopeSGLang = "github.com/sgl-project/sglang" + pipelineSGLang = "sglang" + pipelineVLLM = "vllm" +) + +// Workloads deployed by terraform/eks/daemon/otel-llm-serving. +const ( + namespace = "llm" + inferenceService = "sglang-isvc" + standalonePrefix = "sglang-standalone-" + standaloneWorkload = "sglang-standalone" + annotatedServiceName = "sglang-chat" + vllmPredictorPrefix = "vllm-isvc-predictor-" + vllmStandalonePrefix = "vllm-standalone-" +) + +func TestMain(m *testing.M) { + environment.RegisterEnvironmentMetaDataFlags() + flag.Parse() + env := environment.GetEnvironmentMetaData() + + region := env.Region + if region == "" { + region = os.Getenv("AWS_REGION") + } + if region == "" { + fmt.Fprintf(os.Stderr, "Region not set\n") + os.Exit(1) + } + + clusterName := env.EKSClusterName + if clusterName == "" { + clusterName = os.Getenv("CLUSTER_NAME") + } + if clusterName == "" { + fmt.Fprintf(os.Stderr, "Cluster name not set\n") + os.Exit(1) + } + + ctx := context.Background() + awsCfg, err := awsconfig.LoadDefaultConfig(ctx, awsconfig.WithRegion(region)) + if err != nil { + fmt.Fprintf(os.Stderr, "AWS config error: %v\n", err) + os.Exit(1) + } + stsClient := sts.NewFromConfig(awsCfg) + identity, err := stsClient.GetCallerIdentity(ctx, &sts.GetCallerIdentityInput{}) + if err != nil { + fmt.Fprintf(os.Stderr, "STS GetCallerIdentity error: %v\n", err) + os.Exit(1) + } + + cfg = otelmetrics.TestConfig{ + Region: region, + Endpoint: fmt.Sprintf("https://monitoring.%s.amazonaws.com", region), + Timeout: 30 * time.Second, + MaxRetries: 3, + ClusterName: clusterName, + AccountID: *identity.Account, + SigningService: "monitoring", + } + + client, err = otelmetrics.NewClient(ctx, cfg) + if err != nil { + fmt.Fprintf(os.Stderr, "Client error: %v\n", err) + os.Exit(1) + } + + // The suite only uses GetWithFilter, so no source registry or host types. + queryCache = otelmetrics.NewQueryCache(client, cfg.ClusterName) + + os.Exit(m.Run()) +} diff --git a/test/otel/solutions/sglang/sglang_test.go b/test/otel/solutions/sglang/sglang_test.go new file mode 100644 index 00000000..2cdae4a8 --- /dev/null +++ b/test/otel/solutions/sglang/sglang_test.go @@ -0,0 +1,244 @@ +//go:build integration + +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: MIT + +package sglang + +import ( + "context" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/aws/amazon-cloudwatch-agent-test/util/otelmetrics" +) + +// The two SGLang servers and how the chart finds each. +var servers = []struct { + name string + podPattern string +}{ + // KServe predictor: the shared kserve-model-server job, by KServe's labels. + {name: "kserve", podPattern: inferenceService + "-predictor-.*"}, + // Plain sglang.launch_server: the sglang job, by image name. + {name: "image", podPattern: standalonePrefix + ".*"}, +} + +// recent is the window the routing tests list published metric names over. +const recent = 15 * time.Minute + +func podFilter(pattern string) map[string]string { + return map[string]string{"~@resource.k8s.pod.name": pattern} +} + +func pipelineFilter(podPattern, pipeline string) map[string]string { + return map[string]string{"~@resource.k8s.pod.name": podPattern, otelmetrics.PipelineFilter: pipeline} +} + +// --------------------------------------------------------------------------- +// TestSGLangMetricsExist — every expected metric arrives from both servers. +// --------------------------------------------------------------------------- + +func TestSGLangMetricsExist(t *testing.T) { + t.Parallel() + for _, srv := range servers { + srv := srv + for _, md := range sglangMetrics { + md := md + t.Run(srv.name+"/"+md.Name, func(t *testing.T) { + t.Parallel() + results, err := queryCache.GetWithFilter(context.Background(), md.Name, podFilter(srv.podPattern)) + require.NoError(t, err, "querying %s", md.Name) + require.NotEmpty(t, results, "%s not available from the %s server", md.Name, srv.name) + }) + } + } +} + +// --------------------------------------------------------------------------- +// TestSGLangInstrumentation — scope name and pipeline on every series. +// --------------------------------------------------------------------------- + +func TestSGLangInstrumentation(t *testing.T) { + t.Parallel() + for _, md := range sglangMetrics { + md := md + t.Run(md.Name, func(t *testing.T) { + t.Parallel() + results, err := queryCache.GetWithFilter(context.Background(), md.Name, map[string]string{"@resource.k8s.namespace.name": namespace}) + require.NoError(t, err, "querying %s", md.Name) + require.NotEmpty(t, results, "%s not available", md.Name) + for _, r := range results { + require.Equal(t, scopeSGLang, r.Labels.Instrumentation["@name"], "%s instrumentation name", md.Name) + require.Equal(t, pipelineSGLang, r.Labels.Instrumentation["cloudwatch.pipeline"], "%s pipeline", md.Name) + } + }) + } +} + +// --------------------------------------------------------------------------- +// TestSGLangExpectedLabels — datapoint labels from SGLang survive the pipeline. +// --------------------------------------------------------------------------- + +func TestSGLangExpectedLabels(t *testing.T) { + t.Parallel() + for _, md := range sglangMetrics { + md := md + t.Run(md.Name, func(t *testing.T) { + t.Parallel() + results, err := queryCache.GetWithFilter(context.Background(), md.Name, podFilter(inferenceService+"-predictor-.*")) + require.NoError(t, err, "querying %s", md.Name) + require.NotEmpty(t, results, "%s not available", md.Name) + for _, r := range results { + for _, label := range md.ExpectedLabels { + _, ok := r.Labels.Datapoint[label] + require.True(t, ok, "%s missing expected label '%s'", md.Name, label) + } + } + }) + } +} + +// --------------------------------------------------------------------------- +// TestSGLangKServeAttribution — a KServe predictor is named after its +// InferenceService, as for vLLM, and the workload is the revision's Deployment. +// --------------------------------------------------------------------------- + +func TestSGLangKServeAttribution(t *testing.T) { + t.Parallel() + results, err := queryCache.GetWithFilter(context.Background(), "sglang:num_running_reqs", podFilter(inferenceService+"-predictor-.*")) + require.NoError(t, err) + require.NotEmpty(t, results, "no SGLang series from the InferenceService") + for _, r := range results { + res := r.Labels.Resource + workload := res["k8s.workload.name"] + require.True(t, strings.HasPrefix(workload, inferenceService+"-predictor-") && strings.HasSuffix(workload, "-deployment"), + "k8s.workload.name %q should be the revision Deployment", workload) + require.Equal(t, inferenceService, res["service.name"], "service.name") + require.Equal(t, namespace, res["service.namespace"], "service.namespace") + require.Equal(t, "Deployment", res["k8s.workload.type"], "k8s.workload.type") + require.Equal(t, inferenceService, res[otelmetrics.PodLabelKServeInferenceService], "%s", otelmetrics.PodLabelKServeInferenceService) + require.Equal(t, "predictor", res["k8s.pod.label.component"], "k8s.pod.label.component") + // Set by the agent's Prometheus receiver from the scrape target. + require.Equal(t, "kserve-container", res["k8s.container.name"], "k8s.container.name") + require.Equal(t, "ai_inference", res["aws.service.type"], "aws.service.type") + require.Equal(t, cfg.ClusterName, res["k8s.cluster.name"], "k8s.cluster.name") + otelmetrics.RequireCloudIdentity(t, cfg, "sglang:num_running_reqs", res) + otelmetrics.RequireNoAttribute(t, r, "cw.tmp.service_name_annotation") + otelmetrics.RequireNoAttribute(t, r, "otel_service_name_annotation") + } +} + +// --------------------------------------------------------------------------- +// TestSGLangImageDetected — a plain Deployment declaring no port is scraped on +// sglang.launch_server's 30000, and its resource.opentelemetry.io/service.name +// annotation wins over the workload name. +// --------------------------------------------------------------------------- + +func TestSGLangImageDetected(t *testing.T) { + t.Parallel() + results, err := queryCache.GetWithFilter(context.Background(), "sglang:num_running_reqs", podFilter(standalonePrefix+".*")) + require.NoError(t, err) + require.NotEmpty(t, results, "no SGLang series from the image-detected server") + for _, r := range results { + res := r.Labels.Resource + require.Equal(t, annotatedServiceName, res["service.name"], "service.name from the annotation") + require.Equal(t, standaloneWorkload, res["k8s.workload.name"], "k8s.workload.name") + require.Equal(t, "30000", res["server.port"], "server.port (the default for an undeclared port)") + require.Equal(t, "ai_inference", res["aws.service.type"], "aws.service.type") + _, hasISVC := res[otelmetrics.PodLabelKServeInferenceService] + require.False(t, hasISVC, "InferenceService label on a non-KServe pod") + otelmetrics.RequireNoAttribute(t, r, "otel_service_name_annotation") + otelmetrics.RequireCloudIdentity(t, cfg, "sglang:num_running_reqs", res) + } +} + +// --------------------------------------------------------------------------- +// TestSGLangNoDuplicateSeries — one series per pod, from the model container. +// --------------------------------------------------------------------------- + +func TestSGLangNoDuplicateSeries(t *testing.T) { + t.Parallel() + for _, srv := range servers { + srv := srv + t.Run(srv.name, func(t *testing.T) { + t.Parallel() + results, err := queryCache.GetWithFilter(context.Background(), "sglang:num_running_reqs", podFilter(srv.podPattern)) + require.NoError(t, err) + require.NotEmpty(t, results, "no sglang:num_running_reqs from the %s server", srv.name) + perPod := map[string]int{} + for _, r := range results { + container := r.Labels.Resource["k8s.container.name"] + require.NotEmpty(t, container, "k8s.container.name") + require.NotEqual(t, "queue-proxy", container, "queue-proxy scraped as an SGLang server") + perPod[r.Labels.Resource["k8s.pod.name"]]++ + } + for pod, n := range perPod { + require.Equal(t, 1, n, "%s has %d sglang:num_running_reqs series", pod, n) + } + }) + } +} + +// --------------------------------------------------------------------------- +// TestModelServerRouting — the vLLM and SGLang predictors carry the same KServe +// labels and are scraped by the same job; each engine's series reach only its +// own pipeline. Positive controls: each pod has series in its own pipeline. +// --------------------------------------------------------------------------- + +func TestModelServerRouting(t *testing.T) { + t.Parallel() + for _, c := range []struct { + name, podPattern, own, other string + }{ + {"sglang-predictor", inferenceService + "-predictor-.*", pipelineSGLang, pipelineVLLM}, + {"sglang-image", standalonePrefix + ".*", pipelineSGLang, pipelineVLLM}, + {"vllm-predictor", vllmPredictorPrefix + ".*", pipelineVLLM, pipelineSGLang}, + {"vllm-image", vllmStandalonePrefix + ".*", pipelineVLLM, pipelineSGLang}, + } { + c := c + t.Run(c.name, func(t *testing.T) { + t.Parallel() + require.NotEmpty(t, otelmetrics.PublishedMetricNames(t, queryCache, pipelineFilter(c.podPattern, c.own), recent), + "positive control: nothing from %s pods in the %s pipeline", c.podPattern, c.own) + require.Empty(t, otelmetrics.PublishedMetricNames(t, queryCache, pipelineFilter(c.podPattern, c.other), recent), + "%s pods reached the %s pipeline", c.podPattern, c.other) + }) + } +} + +// --------------------------------------------------------------------------- +// TestSGLangOnlyEngineFamilies — the sglang pipeline publishes only sglang:*. +// --------------------------------------------------------------------------- + +func TestSGLangOnlyEngineFamilies(t *testing.T) { + t.Parallel() + names := otelmetrics.PublishedMetricNames(t, queryCache, map[string]string{otelmetrics.PipelineFilter: pipelineSGLang}, recent) + require.NotEmpty(t, names, "nothing published by the sglang pipeline") + for _, n := range names { + require.True(t, strings.HasPrefix(n, "sglang:"), "%s published by the sglang pipeline", n) + } +} + +// --------------------------------------------------------------------------- +// TestSGLangCountersIncrease — the load generator sends requests to both +// servers, so their request counters must move, not just exist. +// --------------------------------------------------------------------------- + +func TestSGLangCountersIncrease(t *testing.T) { + t.Parallel() + for _, srv := range servers { + srv := srv + for _, name := range []string{"sglang:num_requests_total", "sglang:prompt_tokens_total"} { + name := name + t.Run(srv.name+"/"+name, func(t *testing.T) { + t.Parallel() + inc := otelmetrics.Increase(t, queryCache, name, pipelineFilter(srv.podPattern, pipelineSGLang), "10m") + require.Greater(t, inc, 0.0, "%s did not increase over 10m on the %s server", name, srv.name) + }) + } + } +}