metrics.py 23.1 KB
Newer Older
1
2
from typing import TYPE_CHECKING
from typing import Counter as CollectionsCounter
3
from typing import Dict, List, Optional, Union
4
5

import numpy as np
6
import prometheus_client
7

8
9
from vllm.engine.metrics_types import (StatLoggerBase, Stats,
                                       SupportsMetricsInfo)
10
from vllm.executor.ray_utils import ray
11
from vllm.logger import init_logger
12

13
14
15
16
17
if ray is not None:
    from ray.util import metrics as ray_metrics
else:
    ray_metrics = None

18
19
20
if TYPE_CHECKING:
    from vllm.spec_decode.metrics import SpecDecodeWorkerMetrics

21
22
logger = init_logger(__name__)

23
prometheus_client.disable_created_metrics()
24
25
26
27

# The begin-* and end* here are used by the documentation generator
# to extract the metrics definitions.

28

29
# begin-metrics-definitions
30
class Metrics:
31
32
33
34
35
36
    """
    vLLM uses a multiprocessing-based frontend for the OpenAI server.
    This means that we need to run prometheus_client in multiprocessing mode
    See https://prometheus.github.io/client_python/multiprocess/ for more
    details on limitations.
    """
37
    labelname_finish_reason = "finished_reason"
38
39
40
    _gauge_cls = prometheus_client.Gauge
    _counter_cls = prometheus_client.Counter
    _histogram_cls = prometheus_client.Histogram
41

42
    def __init__(self, labelnames: List[str], max_model_len: int):
43
        # Unregister any existing vLLM collectors (for CI/CD)
44
        self._unregister_vllm_metrics()
45
46

        # System stats
47
        #   Scheduler State
48
        self.gauge_scheduler_running = self._gauge_cls(
49
50
            name="vllm:num_requests_running",
            documentation="Number of requests currently running on GPU.",
51
52
            labelnames=labelnames,
            multiprocess_mode="sum")
53
        self.gauge_scheduler_waiting = self._gauge_cls(
54
55
            name="vllm:num_requests_waiting",
            documentation="Number of requests waiting to be processed.",
56
57
            labelnames=labelnames,
            multiprocess_mode="sum")
58
        self.gauge_scheduler_swapped = self._gauge_cls(
59
60
            name="vllm:num_requests_swapped",
            documentation="Number of requests swapped to CPU.",
61
62
            labelnames=labelnames,
            multiprocess_mode="sum")
63
        #   KV Cache Usage in %
64
        self.gauge_gpu_cache_usage = self._gauge_cls(
65
66
            name="vllm:gpu_cache_usage_perc",
            documentation="GPU KV-cache usage. 1 means 100 percent usage.",
67
68
            labelnames=labelnames,
            multiprocess_mode="sum")
69
        self.gauge_cpu_cache_usage = self._gauge_cls(
70
71
            name="vllm:cpu_cache_usage_perc",
            documentation="CPU KV-cache usage. 1 means 100 percent usage.",
72
73
            labelnames=labelnames,
            multiprocess_mode="sum")
74
75
76
77
78
79
80
81
82
83
84
        #   Prefix caching block hit rate
        self.gauge_cpu_prefix_cache_hit_rate = self._gauge_cls(
            name="vllm:cpu_prefix_cache_hit_rate",
            documentation="CPU prefix cache block hit rate.",
            labelnames=labelnames,
            multiprocess_mode="sum")
        self.gauge_gpu_prefix_cache_hit_rate = self._gauge_cls(
            name="vllm:gpu_prefix_cache_hit_rate",
            documentation="GPU prefix cache block hit rate.",
            labelnames=labelnames,
            multiprocess_mode="sum")
85

86
        # Iteration stats
87
        self.counter_num_preemption = self._counter_cls(
88
89
90
            name="vllm:num_preemptions_total",
            documentation="Cumulative number of preemption from the engine.",
            labelnames=labelnames)
91
        self.counter_prompt_tokens = self._counter_cls(
92
93
94
            name="vllm:prompt_tokens_total",
            documentation="Number of prefill tokens processed.",
            labelnames=labelnames)
95
        self.counter_generation_tokens = self._counter_cls(
96
97
98
            name="vllm:generation_tokens_total",
            documentation="Number of generation tokens processed.",
            labelnames=labelnames)
99
        self.histogram_time_to_first_token = self._histogram_cls(
100
101
102
103
104
105
106
            name="vllm:time_to_first_token_seconds",
            documentation="Histogram of time to first token in seconds.",
            labelnames=labelnames,
            buckets=[
                0.001, 0.005, 0.01, 0.02, 0.04, 0.06, 0.08, 0.1, 0.25, 0.5,
                0.75, 1.0, 2.5, 5.0, 7.5, 10.0
            ])
107
        self.histogram_time_per_output_token = self._histogram_cls(
108
109
110
111
112
113
114
            name="vllm:time_per_output_token_seconds",
            documentation="Histogram of time per output token in seconds.",
            labelnames=labelnames,
            buckets=[
                0.01, 0.025, 0.05, 0.075, 0.1, 0.15, 0.2, 0.3, 0.4, 0.5, 0.75,
                1.0, 2.5
            ])
115
116
117

        # Request stats
        #   Latency
118
        self.histogram_e2e_time_request = self._histogram_cls(
119
120
121
122
            name="vllm:e2e_request_latency_seconds",
            documentation="Histogram of end to end request latency in seconds.",
            labelnames=labelnames,
            buckets=[1.0, 2.5, 5.0, 10.0, 15.0, 20.0, 30.0, 40.0, 50.0, 60.0])
123
        #   Metadata
124
        self.histogram_num_prompt_tokens_request = self._histogram_cls(
125
126
127
128
129
            name="vllm:request_prompt_tokens",
            documentation="Number of prefill tokens processed.",
            labelnames=labelnames,
            buckets=build_1_2_5_buckets(max_model_len),
        )
130
        self.histogram_num_generation_tokens_request = \
131
            self._histogram_cls(
132
133
134
135
136
                name="vllm:request_generation_tokens",
                documentation="Number of generation tokens processed.",
                labelnames=labelnames,
                buckets=build_1_2_5_buckets(max_model_len),
            )
137
        self.histogram_n_request = self._histogram_cls(
138
139
140
141
142
            name="vllm:request_params_n",
            documentation="Histogram of the n request parameter.",
            labelnames=labelnames,
            buckets=[1, 2, 5, 10, 20],
        )
143
        self.counter_request_success = self._counter_cls(
144
            name="vllm:request_success_total",
145
146
            documentation="Count of successfully processed requests.",
            labelnames=labelnames + [Metrics.labelname_finish_reason])
147

148
        # Speculatie decoding stats
149
        self.gauge_spec_decode_draft_acceptance_rate = self._gauge_cls(
150
151
            name="vllm:spec_decode_draft_acceptance_rate",
            documentation="Speulative token acceptance rate.",
152
153
            labelnames=labelnames,
            multiprocess_mode="sum")
154
        self.gauge_spec_decode_efficiency = self._gauge_cls(
155
156
            name="vllm:spec_decode_efficiency",
            documentation="Speculative decoding system efficiency.",
157
158
            labelnames=labelnames,
            multiprocess_mode="sum")
159
160
161
162
163
        self.counter_spec_decode_num_accepted_tokens = (self._counter_cls(
            name="vllm:spec_decode_num_accepted_tokens_total",
            documentation="Number of accepted tokens.",
            labelnames=labelnames))
        self.counter_spec_decode_num_draft_tokens = self._counter_cls(
164
165
166
            name="vllm:spec_decode_num_draft_tokens_total",
            documentation="Number of draft tokens.",
            labelnames=labelnames)
167
168
169
170
        self.counter_spec_decode_num_emitted_tokens = (self._counter_cls(
            name="vllm:spec_decode_num_emitted_tokens_total",
            documentation="Number of emitted tokens.",
            labelnames=labelnames))
171

172
        # Deprecated in favor of vllm:prompt_tokens_total
173
        self.gauge_avg_prompt_throughput = self._gauge_cls(
174
175
176
            name="vllm:avg_prompt_throughput_toks_per_s",
            documentation="Average prefill throughput in tokens/s.",
            labelnames=labelnames,
177
            multiprocess_mode="sum",
178
        )
179
        # Deprecated in favor of vllm:generation_tokens_total
180
        self.gauge_avg_generation_throughput = self._gauge_cls(
181
182
183
            name="vllm:avg_generation_throughput_toks_per_s",
            documentation="Average generation throughput in tokens/s.",
            labelnames=labelnames,
184
            multiprocess_mode="sum",
185
186
        )

187
188

# end-metrics-definitions
189

190
    def _unregister_vllm_metrics(self) -> None:
191
        for collector in list(prometheus_client.REGISTRY._collector_to_names):
192
            if hasattr(collector, "_name") and "vllm" in collector._name:
193
194
195
196
197
198
199
200
201
202
                prometheus_client.REGISTRY.unregister(collector)


class _RayGaugeWrapper:
    """Wraps around ray.util.metrics.Gauge to provide same API as
    prometheus_client.Gauge"""

    def __init__(self,
                 name: str,
                 documentation: str = "",
203
204
205
                 labelnames: Optional[List[str]] = None,
                 multiprocess_mode: str = ""):
        del multiprocess_mode
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
        labelnames_tuple = tuple(labelnames) if labelnames else None
        self._gauge = ray_metrics.Gauge(name=name,
                                        description=documentation,
                                        tag_keys=labelnames_tuple)

    def labels(self, **labels):
        self._gauge.set_default_tags(labels)
        return self

    def set(self, value: Union[int, float]):
        return self._gauge.set(value)


class _RayCounterWrapper:
    """Wraps around ray.util.metrics.Counter to provide same API as
    prometheus_client.Counter"""

    def __init__(self,
                 name: str,
                 documentation: str = "",
                 labelnames: Optional[List[str]] = None):
        labelnames_tuple = tuple(labelnames) if labelnames else None
        self._counter = ray_metrics.Counter(name=name,
                                            description=documentation,
                                            tag_keys=labelnames_tuple)

    def labels(self, **labels):
        self._counter.set_default_tags(labels)
        return self

    def inc(self, value: Union[int, float] = 1.0):
        if value == 0:
            return
        return self._counter.inc(value)


class _RayHistogramWrapper:
    """Wraps around ray.util.metrics.Histogram to provide same API as
    prometheus_client.Histogram"""

    def __init__(self,
                 name: str,
                 documentation: str = "",
                 labelnames: Optional[List[str]] = None,
                 buckets: Optional[List[float]] = None):
        labelnames_tuple = tuple(labelnames) if labelnames else None
        self._histogram = ray_metrics.Histogram(name=name,
                                                description=documentation,
                                                tag_keys=labelnames_tuple,
                                                boundaries=buckets)

    def labels(self, **labels):
        self._histogram.set_default_tags(labels)
        return self

    def observe(self, value: Union[int, float]):
        return self._histogram.observe(value)
263
264
265
266
267
268
269


class RayMetrics(Metrics):
    """
    RayMetrics is used by RayPrometheusStatLogger to log to Ray metrics.
    Provides the same metrics as Metrics but uses Ray's util.metrics library.
    """
270
271
272
    _gauge_cls = _RayGaugeWrapper
    _counter_cls = _RayCounterWrapper
    _histogram_cls = _RayHistogramWrapper
273
274
275
276
277
278
279
280
281
282

    def __init__(self, labelnames: List[str], max_model_len: int):
        if ray_metrics is None:
            raise ImportError("RayMetrics requires Ray to be installed.")
        super().__init__(labelnames, max_model_len)

    def _unregister_vllm_metrics(self) -> None:
        # No-op on purpose
        pass

283

284
def build_1_2_5_buckets(max_value: int) -> List[int]:
285
286
287
288
289
290
291
292
293
294
    """
    Builds a list of buckets with increasing powers of 10 multiplied by 
    mantissa values (1, 2, 5) until the value exceeds the specified maximum.

    Example:
    >>> build_1_2_5_buckets(100)
    [1, 2, 5, 10, 20, 50, 100]
    """
    mantissa_lst = [1, 2, 5]
    exponent = 0
295
    buckets: List[int] = []
296
297
298
299
300
301
302
303
304
305
    while True:
        for m in mantissa_lst:
            value = m * 10**exponent
            if value <= max_value:
                buckets.append(value)
            else:
                return buckets
        exponent += 1


306
307
308
309
310
311
312
313
314
def local_interval_elapsed(now: float, last_log: float,
                           local_interval: float) -> bool:
    elapsed_time = now - last_log
    return elapsed_time > local_interval


def get_throughput(tracked_stats: List[int], now: float,
                   last_log: float) -> float:
    return float(np.sum(tracked_stats) / (now - last_log))
315
316


317
318
319
320
321
322
323
324
325
326
327
class LoggingStatLogger(StatLoggerBase):
    """LoggingStatLogger is used in LLMEngine to log to Stdout."""

    def log(self, stats: Stats) -> None:
        """Called by LLMEngine.
           Logs to Stdout every self.local_interval seconds."""

        # Save tracked stats for token counters.
        self.num_prompt_tokens.append(stats.num_prompt_tokens_iter)
        self.num_generation_tokens.append(stats.num_generation_tokens_iter)

328
329
330
        # Update spec decode metrics
        self.maybe_update_spec_decode_metrics(stats)

331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
        # Log locally every local_interval seconds.
        if local_interval_elapsed(stats.now, self.last_local_log,
                                  self.local_interval):
            # Compute summary metrics for tracked stats (and log them
            # to promethus if applicable).
            prompt_throughput = get_throughput(self.num_prompt_tokens,
                                               now=stats.now,
                                               last_log=self.last_local_log)
            generation_throughput = get_throughput(
                self.num_generation_tokens,
                now=stats.now,
                last_log=self.last_local_log)

            # Log to stdout.
            logger.info(
                "Avg prompt throughput: %.1f tokens/s, "
                "Avg generation throughput: %.1f tokens/s, "
                "Running: %d reqs, Swapped: %d reqs, "
                "Pending: %d reqs, GPU KV cache usage: %.1f%%, "
                "CPU KV cache usage: %.1f%%.",
                prompt_throughput,
                generation_throughput,
                stats.num_running_sys,
                stats.num_swapped_sys,
                stats.num_waiting_sys,
                stats.gpu_cache_usage_sys * 100,
                stats.cpu_cache_usage_sys * 100,
            )
359
360
361
362
363
364
365
            if (stats.cpu_prefix_cache_hit_rate >= 0
                    or stats.gpu_prefix_cache_hit_rate >= 0):
                logger.info(
                    "Prefix cache hit rate: GPU: %.2f%%, CPU: %.2f%%",
                    stats.gpu_prefix_cache_hit_rate * 100,
                    stats.cpu_prefix_cache_hit_rate * 100,
                )
366
367
368
369
370
            if self.spec_decode_metrics is not None:
                logger.info(
                    self._format_spec_decode_metrics_str(
                        self.spec_decode_metrics))

371
372
373
374
            # Reset tracked stats for next interval.
            self.num_prompt_tokens = []
            self.num_generation_tokens = []
            self.last_local_log = stats.now
375
            self.spec_decode_metrics = None
376
377
378
379
380
381
382
383
384

    def _format_spec_decode_metrics_str(
            self, metrics: "SpecDecodeWorkerMetrics") -> str:

        return ("Speculative metrics: "
                f"Draft acceptance rate: {metrics.draft_acceptance_rate:.3f}, "
                f"System efficiency: {metrics.system_efficiency:.3f}, "
                f"Number of speculative tokens: {metrics.num_spec_tokens}, "
                f"Number of accepted tokens: {metrics.accepted_tokens}, "
385
386
                f"Number of draft tokens: {metrics.draft_tokens}, "
                f"Number of emitted tokens: {metrics.emitted_tokens}.")
387

388
389
390
    def info(self, type: str, obj: SupportsMetricsInfo) -> None:
        raise NotImplementedError

391
392
393
394

class PrometheusStatLogger(StatLoggerBase):
    """PrometheusStatLogger is used LLMEngine to log to Promethus."""
    _metrics_cls = Metrics
395
    _gauge_cls = prometheus_client.Gauge
396
397
398
399

    def __init__(self, local_interval: float, labels: Dict[str, str],
                 max_model_len: int) -> None:
        super().__init__(local_interval)
400
401
        # Prometheus metrics
        self.labels = labels
402
403
        self.metrics = self._metrics_cls(labelnames=list(labels.keys()),
                                         max_model_len=max_model_len)
404

405
406
407
    def _log_gauge(self, gauge, data: Union[int, float]) -> None:
        # Convenience function for logging to gauge.
        gauge.labels(**self.labels).set(data)
408

409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
    def _log_counter(self, counter, data: Union[int, float]) -> None:
        # Convenience function for logging to counter.
        counter.labels(**self.labels).inc(data)

    def _log_counter_labels(self, counter, data: CollectionsCounter,
                            label_key: str) -> None:
        # Convenience function for collection counter of labels.
        for label, count in data.items():
            counter.labels(**{**self.labels, label_key: label}).inc(count)

    def _log_histogram(self, histogram, data: Union[List[int],
                                                    List[float]]) -> None:
        # Convenience function for logging list to histogram.
        for datum in data:
            histogram.labels(**self.labels).observe(datum)
424
425

    def _log_prometheus(self, stats: Stats) -> None:
426
427
428
429
430
431
432
433
434
435
436
        # System state data
        self._log_gauge(self.metrics.gauge_scheduler_running,
                        stats.num_running_sys)
        self._log_gauge(self.metrics.gauge_scheduler_swapped,
                        stats.num_swapped_sys)
        self._log_gauge(self.metrics.gauge_scheduler_waiting,
                        stats.num_waiting_sys)
        self._log_gauge(self.metrics.gauge_gpu_cache_usage,
                        stats.gpu_cache_usage_sys)
        self._log_gauge(self.metrics.gauge_cpu_cache_usage,
                        stats.cpu_cache_usage_sys)
437
438
439
440
        self._log_gauge(self.metrics.gauge_cpu_prefix_cache_hit_rate,
                        stats.cpu_prefix_cache_hit_rate)
        self._log_gauge(self.metrics.gauge_gpu_prefix_cache_hit_rate,
                        stats.gpu_prefix_cache_hit_rate)
441
442

        # Iteration level data
443
444
        self._log_counter(self.metrics.counter_num_preemption,
                          stats.num_preemption_iter)
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
        self._log_counter(self.metrics.counter_prompt_tokens,
                          stats.num_prompt_tokens_iter)
        self._log_counter(self.metrics.counter_generation_tokens,
                          stats.num_generation_tokens_iter)
        self._log_histogram(self.metrics.histogram_time_to_first_token,
                            stats.time_to_first_tokens_iter)
        self._log_histogram(self.metrics.histogram_time_per_output_token,
                            stats.time_per_output_tokens_iter)

        # Request level data
        # Latency
        self._log_histogram(self.metrics.histogram_e2e_time_request,
                            stats.time_e2e_requests)
        # Metadata
        finished_reason_counter = CollectionsCounter(
            stats.finished_reason_requests)
        self._log_counter_labels(self.metrics.counter_request_success,
                                 finished_reason_counter,
                                 Metrics.labelname_finish_reason)
        self._log_histogram(self.metrics.histogram_num_prompt_tokens_request,
                            stats.num_prompt_tokens_requests)
        self._log_histogram(
            self.metrics.histogram_num_generation_tokens_request,
            stats.num_generation_tokens_requests)
        self._log_histogram(self.metrics.histogram_n_request, stats.n_requests)

471
472
473
    def _log_prometheus_interval(self, prompt_throughput: float,
                                 generation_throughput: float) -> None:
        # Logs metrics to prometheus that are computed every logging_interval.
474
475
476
477
478
479
        # Support legacy gauge metrics that make throughput calculations on
        # the vLLM side. Moving forward, we should use counters like
        # counter_prompt_tokens, counter_generation_tokens
        # Which log raw data and calculate summaries using rate() on the
        # grafana/prometheus side. See
        # https://github.com/vllm-project/vllm/pull/2316#discussion_r1464204666
480
481
482
483
        self.metrics.gauge_avg_prompt_throughput.labels(
            **self.labels).set(prompt_throughput)
        self.metrics.gauge_avg_generation_throughput.labels(
            **self.labels).set(generation_throughput)
484

485
486
    def log(self, stats: Stats):
        """Logs to prometheus and tracked stats every iteration."""
487
488
489
490
        # Log to prometheus.
        self._log_prometheus(stats)

        # Save tracked stats for token counters.
491
492
        self.num_prompt_tokens.append(stats.num_prompt_tokens_iter)
        self.num_generation_tokens.append(stats.num_generation_tokens_iter)
493

494
495
496
        # Update spec decode metrics
        self.maybe_update_spec_decode_metrics(stats)

497
        # Log locally every local_interval seconds.
498
499
        if local_interval_elapsed(stats.now, self.last_local_log,
                                  self.local_interval):
500
501
            # Compute summary metrics for tracked stats (and log them
            # to promethus if applicable).
502
503
504
505
506
507
508
509
            prompt_throughput = get_throughput(self.num_prompt_tokens,
                                               now=stats.now,
                                               last_log=self.last_local_log)
            generation_throughput = get_throughput(
                self.num_generation_tokens,
                now=stats.now,
                last_log=self.last_local_log)

510
511
512
            self._log_prometheus_interval(
                prompt_throughput=prompt_throughput,
                generation_throughput=generation_throughput)
513

514
            if self.spec_decode_metrics is not None:
515
516
                self._log_gauge(
                    self.metrics.gauge_spec_decode_draft_acceptance_rate,
517
                    self.spec_decode_metrics.draft_acceptance_rate)
518
                self._log_gauge(self.metrics.gauge_spec_decode_efficiency,
519
                                self.spec_decode_metrics.system_efficiency)
520
521
                self._log_counter(
                    self.metrics.counter_spec_decode_num_accepted_tokens,
522
                    self.spec_decode_metrics.accepted_tokens)
523
524
                self._log_counter(
                    self.metrics.counter_spec_decode_num_draft_tokens,
525
                    self.spec_decode_metrics.draft_tokens)
526
527
                self._log_counter(
                    self.metrics.counter_spec_decode_num_emitted_tokens,
528
529
530
531
532
533
534
                    self.spec_decode_metrics.emitted_tokens)

            # Reset tracked stats for next interval.
            self.num_prompt_tokens = []
            self.num_generation_tokens = []
            self.last_local_log = stats.now
            self.spec_decode_metrics = None
535

536
537
538
539
540
541
542
543
544
545
546
547
548
    def info(self, type: str, obj: SupportsMetricsInfo) -> None:
        # Info type metrics are syntactic sugar for a gauge permanently set to 1
        # Since prometheus multiprocessing mode does not support Info, emulate
        # info here with a gauge.
        if type == "cache_config":
            metrics_info = obj.metrics_info()
            info_gauge = self._gauge_cls(
                name="vllm:cache_config_info",
                documentation="Information of the LLMEngine CacheConfig",
                labelnames=metrics_info.keys(),
                multiprocess_mode="mostrecent")
            info_gauge.labels(**metrics_info).set(1)

549

550
551
class RayPrometheusStatLogger(PrometheusStatLogger):
    """RayPrometheusStatLogger uses Ray metrics instead."""
552
    _metrics_cls = RayMetrics
553
554
555

    def info(self, type: str, obj: SupportsMetricsInfo) -> None:
        return None