usage_lib.py 8.69 KB
Newer Older
1
# SPDX-License-Identifier: Apache-2.0
2
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3

yhu422's avatar
yhu422 committed
4
5
6
7
8
9
10
11
12
import datetime
import json
import logging
import os
import platform
import time
from enum import Enum
from pathlib import Path
from threading import Thread
13
from typing import Any, Optional, Union
yhu422's avatar
yhu422 committed
14
15
16
17
18
19
20
from uuid import uuid4

import cpuinfo
import psutil
import requests
import torch

21
import vllm.envs as envs
22
from vllm.connections import global_http_connection
23
from vllm.utils import cuda_device_count_stateless, cuda_get_device_properties
24
from vllm.version import __version__ as VLLM_VERSION
25
26

_config_home = envs.VLLM_CONFIG_ROOT
27
28
_USAGE_STATS_JSON_PATH = os.path.join(_config_home, "usage_stats.json")
_USAGE_STATS_DO_NOT_TRACK_PATH = os.path.join(_config_home, "do_not_track")
yhu422's avatar
yhu422 committed
29
_USAGE_STATS_ENABLED = None
30
_USAGE_STATS_SERVER = envs.VLLM_USAGE_STATS_SERVER
yhu422's avatar
yhu422 committed
31

32
_GLOBAL_RUNTIME_DATA = dict[str, Union[str, int, bool]]()
33

34
35
36
37
38
39
40
41
42
43
44
_USAGE_ENV_VARS_TO_COLLECT = [
    "VLLM_USE_MODELSCOPE",
    "VLLM_USE_TRITON_FLASH_ATTN",
    "VLLM_ATTENTION_BACKEND",
    "VLLM_USE_FLASHINFER_SAMPLER",
    "VLLM_PP_LAYER_PARTITION",
    "VLLM_USE_TRITON_AWQ",
    "VLLM_USE_V1",
    "VLLM_ENABLE_V1_MULTIPROCESSING",
]

45
46
47
48
49

def set_runtime_usage_data(key: str, value: Union[str, int, bool]) -> None:
    """Set global usage data that will be sent with every usage heartbeat."""
    _GLOBAL_RUNTIME_DATA[key] = value

yhu422's avatar
yhu422 committed
50
51
52
53
54

def is_usage_stats_enabled():
    """Determine whether or not we can send usage stats to the server.
    The logic is as follows:
    - By default, it should be enabled.
55
56
    - Three environment variables can disable it:
        - VLLM_DO_NOT_TRACK=1
yhu422's avatar
yhu422 committed
57
58
59
60
61
62
63
        - DO_NOT_TRACK=1
        - VLLM_NO_USAGE_STATS=1
    - A file in the home directory can disable it if it exists:
        - $HOME/.config/vllm/do_not_track
    """
    global _USAGE_STATS_ENABLED
    if _USAGE_STATS_ENABLED is None:
64
65
        do_not_track = envs.VLLM_DO_NOT_TRACK
        no_usage_stats = envs.VLLM_NO_USAGE_STATS
yhu422's avatar
yhu422 committed
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
        do_not_track_file = os.path.exists(_USAGE_STATS_DO_NOT_TRACK_PATH)

        _USAGE_STATS_ENABLED = not (do_not_track or no_usage_stats
                                    or do_not_track_file)
    return _USAGE_STATS_ENABLED


def _get_current_timestamp_ns() -> int:
    return int(datetime.datetime.now(datetime.timezone.utc).timestamp() * 1e9)


def _detect_cloud_provider() -> str:
    # Try detecting through vendor file
    vendor_files = [
        "/sys/class/dmi/id/product_version", "/sys/class/dmi/id/bios_vendor",
        "/sys/class/dmi/id/product_name",
        "/sys/class/dmi/id/chassis_asset_tag", "/sys/class/dmi/id/sys_vendor"
    ]
    # Mapping of identifiable strings to cloud providers
    cloud_identifiers = {
        "amazon": "AWS",
        "microsoft corporation": "AZURE",
        "google": "GCP",
        "oraclecloud": "OCI",
    }

    for vendor_file in vendor_files:
        path = Path(vendor_file)
        if path.is_file():
            file_content = path.read_text().lower()
            for identifier, provider in cloud_identifiers.items():
                if identifier in file_content:
                    return provider

    # Try detecting through environment variables
    env_to_cloud_provider = {
        "RUNPOD_DC_ID": "RUNPOD",
    }
    for env_var, provider in env_to_cloud_provider.items():
        if os.environ.get(env_var):
            return provider

    return "UNKNOWN"


class UsageContext(str, Enum):
    UNKNOWN_CONTEXT = "UNKNOWN_CONTEXT"
    LLM_CLASS = "LLM_CLASS"
    API_SERVER = "API_SERVER"
    OPENAI_API_SERVER = "OPENAI_API_SERVER"
116
    OPENAI_BATCH_RUNNER = "OPENAI_BATCH_RUNNER"
yhu422's avatar
yhu422 committed
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
    ENGINE_CONTEXT = "ENGINE_CONTEXT"


class UsageMessage:
    """Collect platform information and send it to the usage stats server."""

    def __init__(self) -> None:
        # NOTE: vLLM's server _only_ support flat KV pair.
        # Do not use nested fields.

        self.uuid = str(uuid4())

        # Environment Information
        self.provider: Optional[str] = None
        self.num_cpu: Optional[int] = None
        self.cpu_type: Optional[str] = None
        self.cpu_family_model_stepping: Optional[str] = None
        self.total_memory: Optional[int] = None
        self.architecture: Optional[str] = None
        self.platform: Optional[str] = None
137
        self.cuda_runtime: Optional[str] = None
yhu422's avatar
yhu422 committed
138
139
140
        self.gpu_count: Optional[int] = None
        self.gpu_type: Optional[str] = None
        self.gpu_memory_per_device: Optional[int] = None
141
        self.env_var_json: Optional[str] = None
yhu422's avatar
yhu422 committed
142
143
144
145
146
147
148
149
150
151
152
153
154

        # vLLM Information
        self.model_architecture: Optional[str] = None
        self.vllm_version: Optional[str] = None
        self.context: Optional[str] = None

        # Metadata
        self.log_time: Optional[int] = None
        self.source: Optional[str] = None

    def report_usage(self,
                     model_architecture: str,
                     usage_context: UsageContext,
155
                     extra_kvs: Optional[dict[str, Any]] = None) -> None:
yhu422's avatar
yhu422 committed
156
157
158
159
160
161
162
        t = Thread(target=self._report_usage_worker,
                   args=(model_architecture, usage_context, extra_kvs or {}),
                   daemon=True)
        t.start()

    def _report_usage_worker(self, model_architecture: str,
                             usage_context: UsageContext,
163
                             extra_kvs: dict[str, Any]) -> None:
yhu422's avatar
yhu422 committed
164
        self._report_usage_once(model_architecture, usage_context, extra_kvs)
Ning Xie's avatar
Ning Xie committed
165
        self._report_continuous_usage()
yhu422's avatar
yhu422 committed
166
167
168

    def _report_usage_once(self, model_architecture: str,
                           usage_context: UsageContext,
169
                           extra_kvs: dict[str, Any]) -> None:
yhu422's avatar
yhu422 committed
170
        # Platform information
171
        from vllm.platforms import current_platform
172
        if current_platform.is_cuda_alike():
173
174
175
            self.gpu_count = cuda_device_count_stateless()
            self.gpu_type, self.gpu_memory_per_device = (
                cuda_get_device_properties(0, ("name", "total_memory")))
176
177
        if current_platform.is_cuda():
            self.cuda_runtime = torch.version.cuda
178
179
180
181
182
183
184
185
186
        if current_platform.is_tpu():
            try:
                import torch_xla
                self.gpu_count = torch_xla.runtime.world_size()
                self.gpu_type = torch_xla.tpu.get_tpu_type()
                self.gpu_memory_per_device = (
                    torch_xla.core.xla_model.get_memory_info()["bytes_limit"])
            except Exception:
                pass
yhu422's avatar
yhu422 committed
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
        self.provider = _detect_cloud_provider()
        self.architecture = platform.machine()
        self.platform = platform.platform()
        self.total_memory = psutil.virtual_memory().total

        info = cpuinfo.get_cpu_info()
        self.num_cpu = info.get("count", None)
        self.cpu_type = info.get("brand_raw", "")
        self.cpu_family_model_stepping = ",".join([
            str(info.get("family", "")),
            str(info.get("model", "")),
            str(info.get("stepping", ""))
        ])

        # vLLM information
        self.context = usage_context.value
203
        self.vllm_version = VLLM_VERSION
yhu422's avatar
yhu422 committed
204
205
        self.model_architecture = model_architecture

206
207
208
209
210
211
        # Environment variables
        self.env_var_json = json.dumps({
            env_var: getattr(envs, env_var)
            for env_var in _USAGE_ENV_VARS_TO_COLLECT
        })

yhu422's avatar
yhu422 committed
212
213
        # Metadata
        self.log_time = _get_current_timestamp_ns()
214
        self.source = envs.VLLM_USAGE_SOURCE
yhu422's avatar
yhu422 committed
215
216
217
218
219
220
221
222

        data = vars(self)
        if extra_kvs:
            data.update(extra_kvs)

        self._write_to_file(data)
        self._send_to_server(data)

Ning Xie's avatar
Ning Xie committed
223
    def _report_continuous_usage(self):
yhu422's avatar
yhu422 committed
224
225
226
227
228
229
230
        """Report usage every 10 minutes.

        This helps us to collect more data points for uptime of vLLM usages.
        This function can also help send over performance metrics over time.
        """
        while True:
            time.sleep(600)
231
232
233
234
235
            data = {
                "uuid": self.uuid,
                "log_time": _get_current_timestamp_ns(),
            }
            data.update(_GLOBAL_RUNTIME_DATA)
yhu422's avatar
yhu422 committed
236
237
238
239

            self._write_to_file(data)
            self._send_to_server(data)

240
    def _send_to_server(self, data: dict[str, Any]) -> None:
yhu422's avatar
yhu422 committed
241
        try:
242
243
            global_http_client = global_http_connection.get_sync_client()
            global_http_client.post(_USAGE_STATS_SERVER, json=data)
yhu422's avatar
yhu422 committed
244
245
246
247
        except requests.exceptions.RequestException:
            # silently ignore unless we are using debug log
            logging.debug("Failed to send usage data to server")

248
    def _write_to_file(self, data: dict[str, Any]) -> None:
yhu422's avatar
yhu422 committed
249
250
251
252
253
254
255
256
        os.makedirs(os.path.dirname(_USAGE_STATS_JSON_PATH), exist_ok=True)
        Path(_USAGE_STATS_JSON_PATH).touch(exist_ok=True)
        with open(_USAGE_STATS_JSON_PATH, "a") as f:
            json.dump(data, f)
            f.write("\n")


usage_message = UsageMessage()