1Panel/agent/utils/ai_tools/vllm/metrics.go

package vllm

import (
	"bufio"
	"encoding/json"
	"fmt"
	"io"
	"math"
	"sort"
	"strconv"
	"strings"

	"github.com/1Panel-dev/1Panel/agent/app/model"
	"github.com/1Panel-dev/1Panel/agent/utils/re"
)

type Metrics map[string]map[string]float64

type histogram struct {
	Seconds float64            `json:"seconds,omitempty"`
	Sum     float64            `json:"sum"`
	Count   float64            `json:"count"`
	Buckets map[string]float64 `json:"buckets,omitempty"`
}

func Parse(reader io.Reader) (Metrics, error) {
	metrics := Metrics{}
	scanner := bufio.NewScanner(reader)
	scanner.Buffer(make([]byte, 4096), 1024*1024)
	for scanner.Scan() {
		line := strings.TrimSpace(scanner.Text())
		if line == "" || strings.HasPrefix(line, "#") {
			continue
		}
		match := re.GetRegex(re.VLLMMetricSamplePattern).FindStringSubmatch(line)
		if match == nil {
			return nil, fmt.Errorf("invalid vLLM metric sample")
		}
		value, err := strconv.ParseFloat(match[3], 64)
		if err != nil {
			return nil, fmt.Errorf("invalid vLLM metric value: %w", err)
		}
		if !strings.HasPrefix(match[1], "vllm:") {
			continue
		}
		name := strings.TrimPrefix(match[1], "vllm:")
		switch name {
		case "request_prefill_time_seconds_bucket", "request_decode_time_seconds_bucket", "request_queue_time_seconds_bucket":
			continue
		case "num_requests_running", "num_requests_waiting", "kv_cache_usage_perc", "gpu_cache_usage_perc", "prompt_tokens_total", "generation_tokens_total", "request_success_total", "prefix_cache_hits_total", "prefix_cache_queries_total":
		default:
			base := strings.TrimSuffix(strings.TrimSuffix(strings.TrimSuffix(name, "_sum"), "_count"), "_bucket")
			if base == name {
				continue
			}
			switch base {
			case "time_to_first_token_seconds", "time_per_output_token_seconds", "inter_token_latency_seconds", "e2e_request_latency_seconds", "request_prefill_time_seconds", "request_decode_time_seconds", "request_queue_time_seconds", "request_prompt_tokens", "request_generation_tokens":
			default:
				continue
			}
		}
		if math.IsNaN(value) || math.IsInf(value, 0) || value < 0 {
			continue
		}
		if metrics[name] == nil {
			metrics[name] = map[string]float64{}
		}
		labels := re.GetRegex(re.VLLMMetricLabelPattern).FindAllStringSubmatch(match[2], -1)
		parts := make([]string, 0, len(labels))
		for _, label := range labels {
			parts = append(parts, label[1]+"="+label[2])
		}
		sort.Strings(parts)
		metrics[name]["{"+strings.Join(parts, ",")+"}"] = value
	}
	if err := scanner.Err(); err != nil {
		return nil, err
	}
	return metrics, nil
}

func Calculate(current, previous Metrics, elapsed float64) (model.MonitorVLLM, error) {
	result := model.MonitorVLLM{}
	for name, target := range map[string]**float64{
		"num_requests_running": &result.Running,
		"num_requests_waiting": &result.Waiting,
	} {
		if len(current[name]) == 0 {
			continue
		}
		value := 0.0
		for _, sample := range current[name] {
			value += sample
		}
		*target = &value
	}
	cache := current["kv_cache_usage_perc"]
	if len(cache) == 0 {
		cache = current["gpu_cache_usage_perc"]
	}
	if len(cache) > 0 {
		value := 0.0
		for _, sample := range cache {
			value += sample
		}
		value = value / float64(len(cache)) * 100
		result.CacheUsage = &value
	}
	if elapsed <= 0 {
		return result, nil
	}
	for name, target := range map[string]**float64{
		"prompt_tokens_total":     &result.PromptThroughput,
		"generation_tokens_total": &result.GenerationThroughput,
		"request_success_total":   &result.RequestThroughput,
	} {
		if value := counterDelta(current[name], previous[name]); value != nil {
			rate := *value / elapsed
			*target = &rate
		}
	}
	histograms := map[string]histogram{}
	for _, name := range []string{"time_to_first_token_seconds", "inter_token_latency_seconds", "e2e_request_latency_seconds", "request_prefill_time_seconds", "request_decode_time_seconds", "request_queue_time_seconds", "request_prompt_tokens", "request_generation_tokens"} {
		source := name
		if name == "inter_token_latency_seconds" && len(current[name+"_count"]) == 0 {
			source = "time_per_output_token_seconds"
		}
		sum := counterDelta(current[source+"_sum"], previous[source+"_sum"])
		count := counterDelta(current[source+"_count"], previous[source+"_count"])
		if sum == nil || count == nil {
			continue
		}
		item := histogram{Sum: *sum, Count: *count}
		switch name {
		case "request_prefill_time_seconds", "request_decode_time_seconds", "request_queue_time_seconds":
		default:
			item.Buckets = bucketDeltas(current[source+"_bucket"], previous[source+"_bucket"], *count)
		}
		if name == "request_prompt_tokens" || name == "request_generation_tokens" {
			item.Seconds = elapsed
		}
		histograms[name] = item
	}
	applyHistograms(&result, histograms, true)
	if err := calculateRequestMetrics(current, previous, elapsed, &result); err != nil {
		return model.MonitorVLLM{}, err
	}
	encoded, err := json.Marshal(histograms)
	if err != nil {
		return model.MonitorVLLM{}, err
	}
	result.HistogramDeltas = string(encoded)
	return result, nil
}

func applyHistograms(result *model.MonitorVLLM, histograms map[string]histogram, includeMeans bool) {
	for name, target := range map[string]*map[string]float64{
		"request_prompt_tokens":     &result.PromptDistribution,
		"request_generation_tokens": &result.GenerationDistribution,
	} {
		item := histograms[name]
		if item.Seconds <= 0 || len(item.Buckets) == 0 {
			continue
		}
		result.IntervalSeconds = math.Max(result.IntervalSeconds, item.Seconds)
		*target = make(map[string]float64, len(item.Buckets))
		for bound, count := range item.Buckets {
			(*target)[bound] = count / item.Seconds
		}
	}
	if !includeMeans {
		return
	}
	for _, item := range []struct {
		name      string
		mean      **float64
		quantiles []**float64
	}{
		{"time_to_first_token_seconds", &result.TimeToFirstToken, []**float64{&result.TimeToFirstTokenP50, &result.TimeToFirstTokenP90, &result.TimeToFirstTokenP95, &result.TimeToFirstTokenP99}},
		{"inter_token_latency_seconds", &result.TimePerOutputToken, []**float64{&result.TimePerOutputTokenP50, &result.TimePerOutputTokenP90, &result.TimePerOutputTokenP95, &result.TimePerOutputTokenP99}},
		{"e2e_request_latency_seconds", &result.RequestLatency, []**float64{&result.RequestLatencyP50, &result.RequestLatencyP90, &result.RequestLatencyP95, &result.RequestLatencyP99}},
		{"request_prefill_time_seconds", &result.PrefillTime, nil},
		{"request_decode_time_seconds", &result.DecodeTime, nil},
		{"request_queue_time_seconds", &result.QueueTime, nil},
		{"request_generation_tokens", &result.AverageGenerationTokens, nil},
	} {
		histogram, ok := histograms[item.name]
		if !ok || histogram.Count <= 0 {
			continue
		}
		mean := histogram.Sum / histogram.Count
		*item.mean = &mean
		if len(item.quantiles) == 0 {
			continue
		}
		quantiles := histogramQuantiles(histogram.Buckets)
		for i, target := range item.quantiles {
			*target = quantiles[i]
		}
	}
}

func counterDelta(current, previous map[string]float64) *float64 {
	if len(current) == 0 || len(current) != len(previous) {
		return nil
	}
	value := 0.0
	for label, sample := range current {
		before, ok := previous[label]
		if !ok || sample < before {
			return nil
		}
		value += sample - before
	}
	if math.IsInf(value, 0) || math.IsNaN(value) {
		return nil
	}
	return &value
}

func bucketDeltas(current, previous map[string]float64, count float64) map[string]float64 {
	if len(current) == 0 || len(current) != len(previous) {
		return nil
	}
	groups := map[string]map[string]float64{}
	for labels, sample := range current {
		before, ok := previous[labels]
		if !ok || sample < before {
			return nil
		}
		bound := ""
		var identity []string
		for _, label := range re.GetRegex(re.VLLMMetricLabelPattern).FindAllStringSubmatch(labels, -1) {
			if label[1] == "le" {
				raw, err := strconv.Unquote(label[2])
				if err != nil {
					return nil
				}
				upper, err := strconv.ParseFloat(raw, 64)
				if err != nil || math.IsNaN(upper) || upper < 0 {
					return nil
				}
				bound = strconv.FormatFloat(upper, 'g', -1, 64)
			} else {
				identity = append(identity, label[1]+"="+label[2])
			}
		}
		if bound == "" {
			return nil
		}
		group := strings.Join(identity, ",")
		if groups[group] == nil {
			groups[group] = map[string]float64{}
		}
		if _, exists := groups[group][bound]; exists {
			return nil
		}
		groups[group][bound] = sample - before
	}
	result := map[string]float64{}
	for _, buckets := range groups {
		if _, ok := buckets["+Inf"]; !ok || len(buckets) < 2 {
			return nil
		}
		if len(result) > 0 && len(result) != len(buckets) {
			return nil
		}
		for bound := range result {
			if _, ok := buckets[bound]; !ok {
				return nil
			}
		}
		bounds := make([]float64, 0, len(buckets))
		for bound := range buckets {
			value, _ := strconv.ParseFloat(bound, 64)
			bounds = append(bounds, value)
		}
		sort.Float64s(bounds)
		previousCount := 0.0
		for _, bound := range bounds {
			key := strconv.FormatFloat(bound, 'g', -1, 64)
			value := buckets[key]
			if value < previousCount {
				if previousCount-value > 1e-12*(previousCount+value) {
					return nil
				}
				value = previousCount
			}
			result[key] += value
			previousCount = value
		}
	}
	if math.Abs(result["+Inf"]-count) > 1e-12*(result["+Inf"]+count) {
		return nil
	}
	return result
}

func histogramQuantiles(buckets map[string]float64) [4]*float64 {
	var quantiles [4]*float64
	if len(buckets) < 2 || buckets["+Inf"] <= 0 {
		return quantiles
	}
	bounds := make([]float64, 0, len(buckets)-1)
	for bound := range buckets {
		value, err := strconv.ParseFloat(bound, 64)
		if err != nil || math.IsNaN(value) || value < 0 {
			return quantiles
		}
		if !math.IsInf(value, 1) {
			bounds = append(bounds, value)
		}
	}
	if len(bounds) == 0 {
		return quantiles
	}
	sort.Float64s(bounds)
	for i, q := range [...]float64{0.5, 0.9, 0.95, 0.99} {
		rank := q * buckets["+Inf"]
		lower, before := 0.0, 0.0
		value := bounds[len(bounds)-1]
		for _, upper := range bounds {
			count := buckets[strconv.FormatFloat(upper, 'g', -1, 64)]
			if count >= rank && count > before {
				value = lower + (upper-lower)*(rank-before)/(count-before)
				break
			}
			lower, before = upper, count
		}
		quantiles[i] = &value
	}
	return quantiles
}