feat: collect Dedicated Inference (vLLM) metrics, gated by gateway capability flag

cmd/agent/main.go

@@ -160,10 +160,25 @@ func main() {
 		slog.Info("gpu collector skipped", "reason", "disabled")
 	}
 
+	// Dedicated Inference is always scheduled: whether this VM actually runs
+	// vLLM is a fact the gateway knows (from the cloud provider API), not
+	// something the agent can detect locally, so the collector starts gated
+	// off and is flipped on by runMetricsEnabledSyncLoop once confirmed.
+	vllmCollector := &collectors.VLLMCollector{
+		Hub:   hub,
+		Probe: probes.VLLMProbe{Endpoint: cfg.DedicatedInference.Endpoint},
+	}
+	scheduled = append(scheduled, collectors.ScheduledCollector{
+		Collector: vllmCollector,
+		Interval:  cfg.DedicatedInference.ScrapeInterval,
+	})
+	slog.Info("dedicated inference collector scheduled", "endpoint", cfg.DedicatedInference.Endpoint, "interval", cfg.DedicatedInference.ScrapeInterval)
+
 	enabledCollectors := map[string]bool{
-		"agent": true,
-		"gpu":   gpuCollectorEnabled,
-		"node":  cfg.Node.Enable,
+		"agent":               true,
+		"gpu":                 gpuCollectorEnabled,
+		"node":                cfg.Node.Enable,
+		"dedicated_inference": true,
 	}
 	scheduled = append(scheduled, collectors.ScheduledCollector{
 		Collector: &collectors.AgentCollector{
@@ -194,7 +209,7 @@ func main() {
 		// an explicit false from the gateway preserves that contract. Starting
 		// disabled is risky: any first-sync failure (network blip, missing field,
 		// gateway rollout) would leave collectors permanently off.
-		go runMetricsEnabledSyncLoop(ctx, hub, mgr, meta.UUID, len(scheduled), metricsConfigSyncInterval)
+		go runMetricsEnabledSyncLoop(ctx, hub, mgr, vllmCollector, meta.UUID, len(scheduled), metricsConfigSyncInterval)
 	}
 
 	managerErrCh := make(chan error, 1)
@@ -281,6 +296,7 @@ func runMetricsEnabledSyncLoop(
 	ctx context.Context,
 	hub *client.HubClient,
 	manager *collectors.Manager,
+	vllmCollector *collectors.VLLMCollector,
 	uuid string,
 	collectorCount int,
 	interval time.Duration,
@@ -291,6 +307,8 @@ func runMetricsEnabledSyncLoop(
 
 	lastEnabled := true
 	hasLastEnabled := false
+	lastDIEnabled := false
+	hasLastDIEnabled := false
 	sync := func() {
 		syncCtx, cancel := context.WithTimeout(ctx, metricsConfigSyncTimeout)
 		metadata, err := hub.GetMetadata(syncCtx, uuid)
@@ -323,6 +341,23 @@ func runMetricsEnabledSyncLoop(
 			lastEnabled = enabled
 			hasLastEnabled = true
 		}
+
+		// DedicatedInference is a fact about the VM, not a toggle: a nil
+		// Capabilities (no enrichment data) means "not a Dedicated Inference
+		// VM", the opposite default direction from MetricsEnabled above.
+		isDI := metadata.Capabilities != nil && metadata.Capabilities.DedicatedInference
+		if vllmCollector != nil {
+			vllmCollector.SetEnabled(isDI)
+		}
+		if !hasLastDIEnabled || lastDIEnabled != isDI {
+			if isDI {
+				slog.Info("dedicated inference confirmed; collector active", "uuid", uuid)
+			} else {
+				slog.Info("dedicated inference not present; collector idle", "uuid", uuid)
+			}
+			lastDIEnabled = isDI
+			hasLastDIEnabled = true
+		}
 	}
 
 	sync()

internal/client/hub.go

@@ -718,6 +718,23 @@ func closeResponseBody(resp *http.Response, operation string) {
 // dropped field silently disable all collectors.
 type VMMetadata struct {
 	MetricsEnabled *bool `json:"metrics_enabled"`
+
+	// Capabilities is only present when the gateway was able to enrich this
+	// VM from Hyperstack (vm.enhanced_metrics). A nil Capabilities means no
+	// enrichment data is available; callers should treat every field within
+	// it as its zero value in that case, not as "default enabled".
+	Capabilities *VMMetadataCapabilities `json:"capabilities"`
+}
+
+// VMMetadataCapabilities holds enrichment-derived facts about what a VM is
+// provisioned for.
+type VMMetadataCapabilities struct {
+	// DedicatedInference reports whether this VM was provisioned as a
+	// Hyperstack Dedicated Inference (vLLM) instance. Unlike MetricsEnabled,
+	// this is a fact about the VM rather than a toggle: when Capabilities is
+	// nil, or this is false, callers should treat it as "not a Dedicated
+	// Inference VM".
+	DedicatedInference bool `json:"dedicated_inference"`
 }
 
 func (h *HubClient) GetMetadata(ctx context.Context, uuid string) (*VMMetadata, error) {

internal/client/hub_test.go

@@ -1004,6 +1004,46 @@ func TestGetMetadata_MissingFieldIsNil(t *testing.T) {
 	}
 }
 
+// TestGetMetadata_CapabilitiesDedicatedInferenceTrue verifies that a nested
+// capabilities.dedicated_inference:true is decoded correctly.
+func TestGetMetadata_CapabilitiesDedicatedInferenceTrue(t *testing.T) {
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		_, _ = fmt.Fprintln(w, `{"capabilities":{"dedicated_inference":true}}`)
+	}))
+	defer server.Close()
+
+	h := NewHubClient(server.URL)
+	meta, err := h.GetMetadata(context.Background(), "test-uuid")
+	if err != nil {
+		t.Fatalf("GetMetadata() error = %v", err)
+	}
+	if meta.Capabilities == nil || !meta.Capabilities.DedicatedInference {
+		t.Fatalf("Capabilities = %v, want DedicatedInference = true", meta.Capabilities)
+	}
+}
+
+// TestGetMetadata_CapabilitiesMissingFieldIsNil verifies that a response
+// without a capabilities object leaves the field nil, distinguishable from an
+// enriched-but-false response. Unlike MetricsEnabled, nil must be treated as
+// "not a Dedicated Inference VM" by callers, not as default-enabled.
+func TestGetMetadata_CapabilitiesMissingFieldIsNil(t *testing.T) {
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		_, _ = fmt.Fprintln(w, `{}`)
+	}))
+	defer server.Close()
+
+	h := NewHubClient(server.URL)
+	meta, err := h.GetMetadata(context.Background(), "test-uuid")
+	if err != nil {
+		t.Fatalf("GetMetadata() error = %v", err)
+	}
+	if meta.Capabilities != nil {
+		t.Fatalf("Capabilities = %v, want nil for absent field", meta.Capabilities)
+	}
+}
+
 // TestGetMetadata_RefreshesHyperstackKeyOnUnauthorized verifies that GetMetadata
 // retries the request after a 401 triggers a successful key refresh
 // (mirrors the same test pattern used for SubmitBatch and Submit).

internal/collectors/manager.go

@@ -54,6 +54,8 @@ func (m *Manager) Run(ctx context.Context) error {
 			collectorName = "gpu"
 		case *AgentCollector:
 			collectorName = "agent"
+		case *VLLMCollector:
+			collectorName = "dedicated_inference"
 		}
 		go func() {
 			slog.Debug("collector starting", "collector", collectorName, "interval", interval.String())

internal/collectors/vllm.go

@@ -0,0 +1,71 @@
+package collectors
+
+import (
+	"context"
+	"errors"
+	"log/slog"
+	"sync/atomic"
+	"time"
+
+	"github.com/NexGenCloud/hyperstack-agent/internal/client"
+	"github.com/NexGenCloud/hyperstack-agent/internal/metrics"
+	"github.com/NexGenCloud/hyperstack-agent/internal/probes"
+	"github.com/NexGenCloud/hyperstack-agent/internal/system"
+)
+
+// VLLMCollector scrapes a local vLLM (Hyperstack Dedicated Inference) server.
+// Unlike GPU/Node, whether this VM should scrape at all is decided by the
+// gateway at runtime (not local hardware detection), so it carries its own
+// gate defaulting to disabled until the gateway confirms this is a
+// Dedicated Inference VM.
+type VLLMCollector struct {
+	Hub   *client.HubClient
+	Probe probes.Probe
+
+	gate atomic.Bool
+}
+
+func (v *VLLMCollector) SetEnabled(enabled bool) { v.gate.Store(enabled) }
+
+func (v *VLLMCollector) IsEnabled() bool { return v.gate.Load() }
+
+func (v *VLLMCollector) Run(ctx context.Context) error {
+	if v.Probe == nil {
+		return errors.New("vllm collector missing probe implementation")
+	}
+
+	if !v.IsEnabled() {
+		slog.Debug("dedicated inference collector skipped", "reason", "gate disabled")
+		return nil
+	}
+
+	start := time.Now()
+	slog.Debug("dedicated inference probe: begin")
+	samples, err := v.Probe.Collect(ctx)
+	recordDuration := time.Since(start)
+	system.SetLastScrapeMs(recordDuration.Milliseconds())
+	if err != nil {
+		system.RecordProbeFailure(v.Probe.Name(), err)
+		slog.Warn("dedicated inference probe failed", "error", err)
+		return err
+	}
+	filtered := metrics.FilterSamples(samples)
+	if len(filtered) == 0 {
+		system.RecordProbeFailure(v.Probe.Name(), errors.New("no dedicated inference samples collected"))
+		return nil
+	}
+	system.RecordProbeSuccess(v.Probe.Name())
+	slog.Debug("dedicated inference probe: samples", "count", len(filtered), "duration_ms", recordDuration.Milliseconds())
+
+	if v.Hub != nil {
+		now := time.Now().UTC()
+		measures := metrics.SamplesToMeasures(filtered, now, v.Probe.Name())
+		v.Hub.EnqueueMetrics(measures)
+		slog.Info(
+			"metrics enqueued",
+			"collector", v.Probe.Name(),
+			"samples", len(filtered),
+		)
+	}
+	return nil
+}

internal/collectors/vllm_test.go

@@ -0,0 +1,74 @@
+package collectors
+
+import (
+	"context"
+	"errors"
+	"sync/atomic"
+	"testing"
+
+	"github.com/NexGenCloud/hyperstack-agent/internal/metrics"
+)
+
+type stubProbe struct {
+	name    string
+	calls   atomic.Int64
+	samples []metrics.Sample
+	err     error
+}
+
+func (p *stubProbe) Name() string { return p.name }
+
+func (p *stubProbe) Collect(context.Context) ([]metrics.Sample, error) {
+	p.calls.Add(1)
+	if p.err != nil {
+		return nil, p.err
+	}
+	return p.samples, nil
+}
+
+func TestVLLMCollectorSkipsProbeWhenGateDisabled(t *testing.T) {
+	probe := &stubProbe{name: "dedicated_inference"}
+	c := &VLLMCollector{Probe: probe}
+
+	if err := c.Run(context.Background()); err != nil {
+		t.Fatalf("Run() error = %v", err)
+	}
+	if got := probe.calls.Load(); got != 0 {
+		t.Fatalf("probe calls = %d, want 0 when gate disabled", got)
+	}
+}
+
+func TestVLLMCollectorRunsProbeWhenGateEnabled(t *testing.T) {
+	probe := &stubProbe{
+		name: "dedicated_inference",
+		samples: []metrics.Sample{
+			{Name: "hyperstack_dedicated_inference_num_requests_running", Value: "1"},
+		},
+	}
+	c := &VLLMCollector{Probe: probe}
+	c.SetEnabled(true)
+
+	if err := c.Run(context.Background()); err != nil {
+		t.Fatalf("Run() error = %v", err)
+	}
+	if got := probe.calls.Load(); got != 1 {
+		t.Fatalf("probe calls = %d, want 1 when gate enabled", got)
+	}
+}
+
+func TestVLLMCollectorPropagatesProbeError(t *testing.T) {
+	probe := &stubProbe{name: "dedicated_inference", err: errors.New("scrape failed")}
+	c := &VLLMCollector{Probe: probe}
+	c.SetEnabled(true)
+
+	if err := c.Run(context.Background()); err == nil {
+		t.Fatal("expected error to propagate from probe, got nil")
+	}
+}
+
+func TestVLLMCollectorGateDefaultsDisabled(t *testing.T) {
+	c := &VLLMCollector{}
+	if c.IsEnabled() {
+		t.Fatal("IsEnabled() = true, want false by default")
+	}
+}

internal/config/config.go

@@ -25,13 +25,22 @@ type ExporterConfig struct {
 	ScrapeInterval time.Duration
 }
 
+// DedicatedInferenceConfig has no Enable field: whether to actually scrape is
+// decided at runtime by the gateway (VMMetadata.DedicatedInference), not by
+// local config, since the agent has no local way to detect it.
+type DedicatedInferenceConfig struct {
+	Endpoint       string
+	ScrapeInterval time.Duration
+}
+
 // Config holds environment-driven settings for the agent.
 type Config struct {
 	Hub             HubConfig
 	DefaultInterval time.Duration
 
-	Node ExporterConfig
-	GPU  ExporterConfig
+	Node               ExporterConfig
+	GPU                ExporterConfig
+	DedicatedInference DedicatedInferenceConfig
 }
 
 // Load reads configuration from environment variables with sensible defaults.
@@ -51,6 +60,10 @@ func Load() Config {
 			Enable:         getEnvBool("HYPERSTACK_ENABLE_GPU", true),
 			ScrapeInterval: 30 * time.Second,
 		},
+		DedicatedInference: DedicatedInferenceConfig{
+			Endpoint:       getEnv("HYPERSTACK_DEDICATED_INFERENCE_URL", "http://127.0.0.1:8000/metrics"),
+			ScrapeInterval: 15 * time.Second,
+		},
 	}
 }
 

internal/config/config_test.go

@@ -31,6 +31,24 @@ func TestLoad_DefaultInterval(t *testing.T) {
 	}
 }
 
+func TestLoad_DedicatedInferenceDefaults(t *testing.T) {
+	cfg := Load()
+	if cfg.DedicatedInference.Endpoint != "http://127.0.0.1:8000/metrics" {
+		t.Fatalf("cfg.DedicatedInference.Endpoint = %q, want default 127.0.0.1:8000/metrics", cfg.DedicatedInference.Endpoint)
+	}
+	if cfg.DedicatedInference.ScrapeInterval.String() != "15s" {
+		t.Fatalf("cfg.DedicatedInference.ScrapeInterval = %v, want 15s", cfg.DedicatedInference.ScrapeInterval)
+	}
+}
+
+func TestLoad_DedicatedInferenceEndpointOverride(t *testing.T) {
+	t.Setenv("HYPERSTACK_DEDICATED_INFERENCE_URL", "http://localhost:9000/metrics")
+	cfg := Load()
+	if cfg.DedicatedInference.Endpoint != "http://localhost:9000/metrics" {
+		t.Fatalf("cfg.DedicatedInference.Endpoint = %q, want override", cfg.DedicatedInference.Endpoint)
+	}
+}
+
 func TestSecurityWarningsWarnsForNonLoopbackHTTPHub(t *testing.T) {
 	cfg := Config{Hub: HubConfig{URL: "http://gateway.example.com"}}
 

internal/metrics/filter.go

@@ -33,6 +33,8 @@ var ImportantPrefixes = []string{
 	"nvidia_gpu_",
 	"nvidia_smi_",
 	"gpu_",
+	// Dedicated Inference (vLLM)
+	"hyperstack_dedicated_inference_",
 }
 
 // FilterSamples applies the important prefix filter and returns the subset of samples.