wehub-resource-sync
2099 lines
75 KiB
Python
2099 lines
75 KiB
Python
import asyncio
|
|
import datetime
|
|
import json
|
|
import logging
|
|
import os
|
|
import socket
|
|
import sys
|
|
import traceback
|
|
from collections import defaultdict, namedtuple
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from typing import List, Optional, Tuple
|
|
|
|
import requests
|
|
from grpc.aio import ServicerContext
|
|
from opencensus.stats import stats as stats_module
|
|
from opentelemetry.proto.collector.metrics.v1 import (
|
|
metrics_service_pb2,
|
|
metrics_service_pb2_grpc,
|
|
)
|
|
from opentelemetry.proto.metrics.v1.metrics_pb2 import Metric
|
|
from prometheus_client.core import REGISTRY
|
|
from prometheus_client.parser import text_string_to_metric_families
|
|
|
|
import ray
|
|
import ray._private.prometheus_exporter as prometheus_exporter
|
|
import ray.dashboard.modules.reporter.reporter_consts as reporter_consts
|
|
import ray.dashboard.utils as dashboard_utils
|
|
from ray._common.network_utils import get_localhost_ip, is_localhost
|
|
from ray._common.utils import (
|
|
get_or_create_event_loop,
|
|
)
|
|
from ray._private import utils
|
|
from ray._private.metrics_agent import Gauge, MetricsAgent, Record
|
|
from ray._private.ray_constants import (
|
|
DEBUG_AUTOSCALING_STATUS,
|
|
RAY_ENABLE_OPEN_TELEMETRY,
|
|
env_integer,
|
|
)
|
|
from ray._private.telemetry.open_telemetry_metric_recorder import (
|
|
OpenTelemetryMetricRecorder,
|
|
)
|
|
from ray._private.utils import get_system_memory
|
|
from ray._raylet import (
|
|
GCS_PID_KEY,
|
|
METRICS_EXPORT_PORT_NAME,
|
|
RayletClient,
|
|
WorkerID,
|
|
persist_port,
|
|
)
|
|
from ray.autoscaler.v2.sdk import get_cluster_status
|
|
from ray.autoscaler.v2.utils import _count_by, is_autoscaler_v2
|
|
from ray.core.generated import reporter_pb2, reporter_pb2_grpc
|
|
from ray.dashboard import k8s_utils
|
|
from ray.dashboard.consts import (
|
|
CLUSTER_TAG_KEYS,
|
|
COMPONENT_GPU_TAG_KEYS,
|
|
COMPONENT_METRICS_TAG_KEYS,
|
|
GCS_RPC_TIMEOUT_SECONDS,
|
|
GPU_TAG_KEYS,
|
|
NODE_TAG_KEYS,
|
|
TPU_TAG_KEYS,
|
|
)
|
|
from ray.dashboard.modules.reporter.gpu_profile_manager import GpuProfilingManager
|
|
from ray.dashboard.modules.reporter.gpu_providers import (
|
|
GpuMetricProvider,
|
|
GpuUtilizationInfo,
|
|
TpuUtilizationInfo,
|
|
)
|
|
from ray.dashboard.modules.reporter.jax_profile_manager import JaxProfilingManager
|
|
from ray.dashboard.modules.reporter.profile_manager import (
|
|
CpuProfilingManager,
|
|
MemoryProfilingManager,
|
|
)
|
|
from ray.dashboard.modules.reporter.reporter_models import (
|
|
StatsPayload,
|
|
)
|
|
from ray.exceptions import (
|
|
GetTimeoutError,
|
|
RpcError,
|
|
)
|
|
|
|
import psutil
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
enable_tpu_usage_check = True
|
|
|
|
# Are we in a K8s pod?
|
|
IN_KUBERNETES_POD = "KUBERNETES_SERVICE_HOST" in os.environ
|
|
# Flag to enable showing disk usage when running in a K8s pod,
|
|
# disk usage defined as the result of running psutil.disk_usage("/")
|
|
# in the Ray container.
|
|
ENABLE_K8S_DISK_USAGE = os.environ.get("RAY_DASHBOARD_ENABLE_K8S_DISK_USAGE") == "1"
|
|
# Try to determine if we're in a container.
|
|
IN_CONTAINER = os.path.exists("/sys/fs/cgroup")
|
|
# Using existence of /sys/fs/cgroup as the criterion is consistent with
|
|
# Ray's existing resource logic, see e.g. ray._private.utils.get_num_cpus().
|
|
|
|
# NOTE: Executor in this head is intentionally constrained to just 1 thread by
|
|
# default to limit its concurrency, therefore reducing potential for
|
|
# GIL contention
|
|
RAY_DASHBOARD_REPORTER_AGENT_TPE_MAX_WORKERS = env_integer(
|
|
"RAY_DASHBOARD_REPORTER_AGENT_TPE_MAX_WORKERS", 1
|
|
)
|
|
|
|
# TPU device plugin metric address should be in the format "{HOST_IP}:2112"
|
|
TPU_DEVICE_PLUGIN_ADDR = os.environ.get("TPU_DEVICE_PLUGIN_ADDR", None)
|
|
|
|
|
|
def recursive_asdict(o):
|
|
if isinstance(o, tuple) and hasattr(o, "_asdict"):
|
|
return recursive_asdict(o._asdict())
|
|
|
|
if isinstance(o, (tuple, list)):
|
|
L = []
|
|
for k in o:
|
|
L.append(recursive_asdict(k))
|
|
return L
|
|
|
|
if isinstance(o, dict):
|
|
D = {k: recursive_asdict(v) for k, v in o.items()}
|
|
return D
|
|
|
|
return o
|
|
|
|
|
|
def jsonify_asdict(o) -> str:
|
|
return json.dumps(dashboard_utils.to_google_style(recursive_asdict(o)))
|
|
|
|
|
|
# A list of gauges to record and export metrics.
|
|
METRICS_GAUGES = {
|
|
# CPU metrics
|
|
"node_cpu_utilization": Gauge(
|
|
"node_cpu_utilization",
|
|
"Total CPU usage on a ray node",
|
|
"percentage",
|
|
NODE_TAG_KEYS,
|
|
),
|
|
"node_cpu_count": Gauge(
|
|
"node_cpu_count",
|
|
"Total CPUs available on a ray node",
|
|
"cores",
|
|
NODE_TAG_KEYS,
|
|
),
|
|
# Memory metrics
|
|
"node_mem_used": Gauge(
|
|
"node_mem_used",
|
|
"Memory usage on a ray node",
|
|
"bytes",
|
|
NODE_TAG_KEYS,
|
|
),
|
|
"node_mem_available": Gauge(
|
|
"node_mem_available",
|
|
"Memory available on a ray node",
|
|
"bytes",
|
|
NODE_TAG_KEYS,
|
|
),
|
|
"node_mem_total": Gauge(
|
|
"node_mem_total",
|
|
"Total memory on a ray node",
|
|
"bytes",
|
|
NODE_TAG_KEYS,
|
|
),
|
|
"node_mem_shared_bytes": Gauge(
|
|
"node_mem_shared_bytes",
|
|
"Total shared memory usage on a ray node",
|
|
"bytes",
|
|
NODE_TAG_KEYS,
|
|
),
|
|
"node_mem_used_host": Gauge(
|
|
"node_mem_used_host",
|
|
"Host memory usage on a ray node",
|
|
"bytes",
|
|
NODE_TAG_KEYS,
|
|
),
|
|
"node_mem_total_host": Gauge(
|
|
"node_mem_total_host",
|
|
"Total host memory on a ray node",
|
|
"bytes",
|
|
NODE_TAG_KEYS,
|
|
),
|
|
"node_cgroup_mem_used": Gauge(
|
|
"node_cgroup_mem_used",
|
|
"Container memory usage on a ray node",
|
|
"bytes",
|
|
NODE_TAG_KEYS,
|
|
),
|
|
"node_cgroup_mem_total": Gauge(
|
|
"node_cgroup_mem_total",
|
|
"Container memory limit on a ray node",
|
|
"bytes",
|
|
NODE_TAG_KEYS,
|
|
),
|
|
# GPU metrics
|
|
"node_gpus_available": Gauge(
|
|
"node_gpus_available",
|
|
"Total GPUs available on a ray node",
|
|
"percentage",
|
|
GPU_TAG_KEYS,
|
|
),
|
|
"node_gpus_utilization": Gauge(
|
|
"node_gpus_utilization",
|
|
"Total GPUs usage on a ray node",
|
|
"percentage",
|
|
GPU_TAG_KEYS,
|
|
),
|
|
"node_gram_used": Gauge(
|
|
"node_gram_used",
|
|
"Total GPU RAM usage on a ray node",
|
|
"bytes",
|
|
GPU_TAG_KEYS,
|
|
),
|
|
"node_gram_available": Gauge(
|
|
"node_gram_available",
|
|
"Total GPU RAM available on a ray node",
|
|
"bytes",
|
|
GPU_TAG_KEYS,
|
|
),
|
|
"node_gpu_power_milliwatts": Gauge(
|
|
"node_gpu_power_milliwatts",
|
|
"Current GPU power draw in milliwatts",
|
|
"milliwatts",
|
|
GPU_TAG_KEYS,
|
|
),
|
|
"node_gpu_temperature_celsius": Gauge(
|
|
"node_gpu_temperature_celsius",
|
|
"Current GPU temperature in Celsius",
|
|
"celsius",
|
|
GPU_TAG_KEYS,
|
|
),
|
|
# TPU metrics
|
|
"tpu_tensorcore_utilization": Gauge(
|
|
"tpu_tensorcore_utilization",
|
|
"Percentage TPU tensorcore utilization on a ray node, value should be between 0 and 100",
|
|
"percentage",
|
|
TPU_TAG_KEYS,
|
|
),
|
|
"tpu_memory_bandwidth_utilization": Gauge(
|
|
"tpu_memory_bandwidth_utilization",
|
|
"Percentage TPU memory bandwidth utilization on a ray node, value should be between 0 and 100",
|
|
"percentage",
|
|
TPU_TAG_KEYS,
|
|
),
|
|
"tpu_duty_cycle": Gauge(
|
|
"tpu_duty_cycle",
|
|
"Percentage of time during which the TPU was actively processing, value should be between 0 and 100",
|
|
"percentage",
|
|
TPU_TAG_KEYS,
|
|
),
|
|
"tpu_memory_used": Gauge(
|
|
"tpu_memory_used",
|
|
"Total memory used by the accelerator in bytes",
|
|
"bytes",
|
|
TPU_TAG_KEYS,
|
|
),
|
|
"tpu_memory_total": Gauge(
|
|
"tpu_memory_total",
|
|
"Total memory allocatable by the accelerator in bytes",
|
|
"bytes",
|
|
TPU_TAG_KEYS,
|
|
),
|
|
# Disk I/O metrics
|
|
"node_disk_io_read": Gauge(
|
|
"node_disk_io_read",
|
|
"Total read from disk",
|
|
"bytes",
|
|
NODE_TAG_KEYS,
|
|
),
|
|
"node_disk_io_write": Gauge(
|
|
"node_disk_io_write",
|
|
"Total written to disk",
|
|
"bytes",
|
|
NODE_TAG_KEYS,
|
|
),
|
|
"node_disk_io_read_count": Gauge(
|
|
"node_disk_io_read_count",
|
|
"Total read ops from disk",
|
|
"io",
|
|
NODE_TAG_KEYS,
|
|
),
|
|
"node_disk_io_write_count": Gauge(
|
|
"node_disk_io_write_count",
|
|
"Total write ops to disk",
|
|
"io",
|
|
NODE_TAG_KEYS,
|
|
),
|
|
"node_disk_io_read_speed": Gauge(
|
|
"node_disk_io_read_speed",
|
|
"Disk read speed",
|
|
"bytes/sec",
|
|
NODE_TAG_KEYS,
|
|
),
|
|
"node_disk_io_write_speed": Gauge(
|
|
"node_disk_io_write_speed",
|
|
"Disk write speed",
|
|
"bytes/sec",
|
|
NODE_TAG_KEYS,
|
|
),
|
|
"node_disk_read_iops": Gauge(
|
|
"node_disk_read_iops",
|
|
"Disk read iops",
|
|
"iops",
|
|
NODE_TAG_KEYS,
|
|
),
|
|
"node_disk_write_iops": Gauge(
|
|
"node_disk_write_iops",
|
|
"Disk write iops",
|
|
"iops",
|
|
NODE_TAG_KEYS,
|
|
),
|
|
# Disk usage metrics
|
|
"node_disk_usage": Gauge(
|
|
"node_disk_usage",
|
|
"Total disk usage (bytes) on a ray node",
|
|
"bytes",
|
|
NODE_TAG_KEYS,
|
|
),
|
|
"node_disk_free": Gauge(
|
|
"node_disk_free",
|
|
"Total disk free (bytes) on a ray node",
|
|
"bytes",
|
|
NODE_TAG_KEYS,
|
|
),
|
|
"node_disk_utilization_percentage": Gauge(
|
|
"node_disk_utilization_percentage",
|
|
"Total disk utilization (percentage) on a ray node",
|
|
"percentage",
|
|
NODE_TAG_KEYS,
|
|
),
|
|
# Network metrics
|
|
"node_network_sent": Gauge(
|
|
"node_network_sent",
|
|
"Total network sent",
|
|
"bytes",
|
|
NODE_TAG_KEYS,
|
|
),
|
|
"node_network_received": Gauge(
|
|
"node_network_received",
|
|
"Total network received",
|
|
"bytes",
|
|
NODE_TAG_KEYS,
|
|
),
|
|
"node_network_send_speed": Gauge(
|
|
"node_network_send_speed",
|
|
"Network send speed",
|
|
"bytes/sec",
|
|
NODE_TAG_KEYS,
|
|
),
|
|
"node_network_receive_speed": Gauge(
|
|
"node_network_receive_speed",
|
|
"Network receive speed",
|
|
"bytes/sec",
|
|
NODE_TAG_KEYS,
|
|
),
|
|
# Component metrics
|
|
"component_cpu_percentage": Gauge(
|
|
"component_cpu_percentage",
|
|
"Total CPU usage of the components on a node.",
|
|
"percentage",
|
|
COMPONENT_METRICS_TAG_KEYS,
|
|
),
|
|
"component_shared_bytes": Gauge(
|
|
"component_shared_bytes",
|
|
"SHM usage of all components of the node. "
|
|
"It is equivalent to the top command's SHR column.",
|
|
"bytes",
|
|
COMPONENT_METRICS_TAG_KEYS,
|
|
),
|
|
"component_rss_mb": Gauge(
|
|
"component_rss_mb",
|
|
"RSS usage of all components on the node.",
|
|
"MB",
|
|
COMPONENT_METRICS_TAG_KEYS,
|
|
),
|
|
"component_rss_bytes": Gauge(
|
|
"component_rss_bytes",
|
|
"RSS usage of all components on the node.",
|
|
"bytes",
|
|
COMPONENT_METRICS_TAG_KEYS,
|
|
),
|
|
"component_uss_mb": Gauge(
|
|
"component_uss_mb",
|
|
"USS usage of all components on the node.",
|
|
"MB",
|
|
COMPONENT_METRICS_TAG_KEYS,
|
|
),
|
|
"component_uss_bytes": Gauge(
|
|
"component_uss_bytes",
|
|
"USS usage of all components on the node.",
|
|
"bytes",
|
|
COMPONENT_METRICS_TAG_KEYS,
|
|
),
|
|
"component_num_fds": Gauge(
|
|
"component_num_fds",
|
|
"Number of open fds of all components on the node (Not available on Windows).",
|
|
"count",
|
|
COMPONENT_METRICS_TAG_KEYS,
|
|
),
|
|
# Cluster metrics
|
|
"cluster_active_nodes": Gauge(
|
|
"cluster_active_nodes",
|
|
"Active nodes on the cluster",
|
|
"count",
|
|
CLUSTER_TAG_KEYS,
|
|
),
|
|
# cluster_idle_nodes is only available for v2 autoscaler
|
|
"cluster_idle_nodes": Gauge(
|
|
"cluster_idle_nodes",
|
|
"Idle nodes on the cluster",
|
|
"count",
|
|
CLUSTER_TAG_KEYS,
|
|
),
|
|
"cluster_failed_nodes": Gauge(
|
|
"cluster_failed_nodes",
|
|
"Failed nodes on the cluster",
|
|
"count",
|
|
CLUSTER_TAG_KEYS,
|
|
),
|
|
"cluster_pending_nodes": Gauge(
|
|
"cluster_pending_nodes",
|
|
"Pending nodes on the cluster",
|
|
"count",
|
|
CLUSTER_TAG_KEYS,
|
|
),
|
|
"component_gpu_percentage": Gauge(
|
|
"component_gpu_percentage",
|
|
"GPU usage of all components on the node.",
|
|
"percentage",
|
|
COMPONENT_GPU_TAG_KEYS,
|
|
),
|
|
"component_gpu_memory_mb": Gauge(
|
|
"component_gpu_memory_mb",
|
|
"GPU memory usage of all components on the node.",
|
|
"MB",
|
|
COMPONENT_GPU_TAG_KEYS,
|
|
),
|
|
}
|
|
|
|
PSUTIL_PROCESS_ATTRS = (
|
|
[
|
|
"pid",
|
|
"create_time",
|
|
"cpu_percent",
|
|
"cpu_times",
|
|
"cmdline",
|
|
"memory_info",
|
|
]
|
|
+ (["num_fds"] if sys.platform != "win32" else [])
|
|
# Only collect memory_full_info in Mac OS X
|
|
+ (["memory_full_info"] if sys.platform == "darwin" else [])
|
|
)
|
|
|
|
|
|
class ReporterAgent(
|
|
dashboard_utils.DashboardAgentModule,
|
|
reporter_pb2_grpc.ReporterServiceServicer,
|
|
metrics_service_pb2_grpc.MetricsServiceServicer,
|
|
):
|
|
"""A monitor process for monitoring Ray nodes.
|
|
|
|
Attributes:
|
|
dashboard_agent: The DashboardAgent object contains global config
|
|
raylet_client: The RayletClient object to access raylet server
|
|
"""
|
|
|
|
def __init__(self, dashboard_agent, raylet_client=None):
|
|
"""Initialize the reporter object."""
|
|
super().__init__(dashboard_agent)
|
|
|
|
if IN_KUBERNETES_POD or IN_CONTAINER:
|
|
# psutil does not give a meaningful logical cpu count when in a K8s pod, or
|
|
# in a container in general.
|
|
# Use ray._private.utils for this instead.
|
|
logical_cpu_count = utils.get_num_cpus(override_docker_cpu_warning=True)
|
|
# (Override the docker warning to avoid dashboard log spam.)
|
|
|
|
# The dashboard expects a physical CPU count as well.
|
|
# This is not always meaningful in a container, but we will go ahead
|
|
# and give the dashboard what it wants using psutil.
|
|
physical_cpu_count = psutil.cpu_count(logical=False)
|
|
else:
|
|
logical_cpu_count = psutil.cpu_count()
|
|
physical_cpu_count = psutil.cpu_count(logical=False)
|
|
self._cpu_counts = (logical_cpu_count, physical_cpu_count)
|
|
self._gcs_client = dashboard_agent.gcs_client
|
|
self._ip = dashboard_agent.ip
|
|
self._log_dir = dashboard_agent.log_dir
|
|
self._is_head_node = dashboard_agent.is_head
|
|
self._hostname = socket.gethostname()
|
|
# (pid, created_time) -> psutil.Process
|
|
self._workers = {}
|
|
# psutil.Process of the parent.
|
|
self._raylet_proc = None
|
|
# psutil.Process of the current process.
|
|
self._agent_proc = None
|
|
# The last reported worker proc names (e.g., ray::*).
|
|
self._latest_worker_proc_names = set()
|
|
self._latest_gpu_worker_proc_names = set()
|
|
self._network_stats_hist = [(0, (0.0, 0.0))] # time, (sent, recv)
|
|
self._disk_io_stats_hist = [
|
|
(0, (0.0, 0.0, 0, 0))
|
|
] # time, (bytes read, bytes written, read ops, write ops)
|
|
self._metrics_collection_disabled = dashboard_agent.metrics_collection_disabled
|
|
self._metrics_agent = None
|
|
self._open_telemetry_metric_recorder = None
|
|
self._session_name = dashboard_agent.session_name
|
|
if not self._metrics_collection_disabled:
|
|
stats_exporter = prometheus_exporter.new_stats_exporter(
|
|
prometheus_exporter.Options(
|
|
namespace="ray",
|
|
port=dashboard_agent.metrics_export_port,
|
|
address=get_localhost_ip() if is_localhost(self._ip) else "",
|
|
)
|
|
)
|
|
dashboard_agent.metrics_export_port = stats_exporter.port
|
|
|
|
self._metrics_agent = MetricsAgent(
|
|
stats_module.stats.view_manager,
|
|
stats_module.stats.stats_recorder,
|
|
stats_exporter,
|
|
)
|
|
self._open_telemetry_metric_recorder = OpenTelemetryMetricRecorder()
|
|
if self._metrics_agent.proxy_exporter_collector:
|
|
# proxy_exporter_collector is None
|
|
# if Prometheus server is not started.
|
|
REGISTRY.register(self._metrics_agent.proxy_exporter_collector)
|
|
|
|
# Metrics collection is disabled, write -1 to indicate the port is not in use.
|
|
persist_port(
|
|
dashboard_agent.session_dir,
|
|
self._dashboard_agent.node_id,
|
|
METRICS_EXPORT_PORT_NAME,
|
|
(
|
|
dashboard_agent.metrics_export_port
|
|
if not self._metrics_collection_disabled
|
|
else -1
|
|
),
|
|
)
|
|
self._key = (
|
|
f"{reporter_consts.REPORTER_PREFIX}" f"{self._dashboard_agent.node_id}"
|
|
)
|
|
|
|
self._executor = ThreadPoolExecutor(
|
|
max_workers=RAY_DASHBOARD_REPORTER_AGENT_TPE_MAX_WORKERS,
|
|
thread_name_prefix="reporter_agent_executor",
|
|
)
|
|
self._gcs_pid = None
|
|
self._gcs_proc = None
|
|
|
|
self._gpu_profiling_manager = GpuProfilingManager(
|
|
profile_dir_path=self._log_dir, ip_address=self._ip
|
|
)
|
|
self._gpu_profiling_manager.start_monitoring_daemon()
|
|
|
|
self._jax_profiling_manager = JaxProfilingManager(
|
|
profile_dir_path=self._log_dir
|
|
)
|
|
|
|
# Create GPU metric provider instance
|
|
self._gpu_metric_provider = GpuMetricProvider()
|
|
|
|
if raylet_client:
|
|
self._raylet_client = raylet_client
|
|
else:
|
|
self._raylet_client = RayletClient(
|
|
ip_address=self._ip, port=self._dashboard_agent.node_manager_port
|
|
)
|
|
|
|
async def GetTraceback(self, request, context):
|
|
pid = request.pid
|
|
native = request.native
|
|
subprocesses = request.subprocesses
|
|
p = CpuProfilingManager(self._log_dir)
|
|
success, output = await p.trace_dump(
|
|
pid, native=native, subprocesses=subprocesses
|
|
)
|
|
return reporter_pb2.GetTracebackReply(output=output, success=success)
|
|
|
|
async def CpuProfiling(self, request, context):
|
|
pid = request.pid
|
|
duration = request.duration
|
|
format = request.format
|
|
native = request.native
|
|
idle = request.idle
|
|
subprocesses = request.subprocesses
|
|
p = CpuProfilingManager(self._log_dir)
|
|
success, output = await p.cpu_profile(
|
|
pid,
|
|
format=format,
|
|
duration=duration,
|
|
native=native,
|
|
idle=idle,
|
|
subprocesses=subprocesses,
|
|
)
|
|
return reporter_pb2.CpuProfilingReply(output=output, success=success)
|
|
|
|
async def GpuProfiling(self, request, context):
|
|
pid = request.pid
|
|
num_iterations = request.num_iterations
|
|
success, output = await self._gpu_profiling_manager.gpu_profile(
|
|
pid=pid, num_iterations=num_iterations
|
|
)
|
|
return reporter_pb2.GpuProfilingReply(success=success, output=output)
|
|
|
|
async def JaxProfiling(self, request, context):
|
|
pid = request.pid
|
|
port = request.port
|
|
duration = request.duration if request.HasField("duration") else 5
|
|
success, output = await self._jax_profiling_manager.jax_profile(
|
|
pid=pid, port=port, duration_s=duration
|
|
)
|
|
return reporter_pb2.JaxProfilingReply(success=success, output=output)
|
|
|
|
async def MemoryProfiling(self, request, context):
|
|
pid = request.pid
|
|
format = request.format
|
|
leaks = request.leaks
|
|
duration = request.duration
|
|
native = request.native
|
|
trace_python_allocators = request.trace_python_allocators
|
|
p = MemoryProfilingManager(self._log_dir)
|
|
success, profiler_filename, output = await p.attach_profiler(
|
|
pid, native=native, trace_python_allocators=trace_python_allocators
|
|
)
|
|
if not success:
|
|
return reporter_pb2.MemoryProfilingReply(output=output, success=success)
|
|
|
|
# add 1 second sleep for memray overhead
|
|
await asyncio.sleep(duration + 1)
|
|
success, output = await p.detach_profiler(pid)
|
|
warning = None if success else output
|
|
success, output = await p.get_profile_result(
|
|
pid, profiler_filename=profiler_filename, format=format, leaks=leaks
|
|
)
|
|
return reporter_pb2.MemoryProfilingReply(
|
|
output=output, success=success, warning=warning
|
|
)
|
|
|
|
async def HealthCheck(
|
|
self,
|
|
_request: reporter_pb2.HealthCheckRequest,
|
|
_context: ServicerContext,
|
|
) -> reporter_pb2.HealthCheckReply:
|
|
"""This is a health check endpoint for the reporter agent.
|
|
|
|
It is used to check if the reporter agent is ready to receive requests.
|
|
"""
|
|
return reporter_pb2.HealthCheckReply()
|
|
|
|
async def ReportOCMetrics(self, request, context):
|
|
# Do nothing if metrics collection is disabled.
|
|
if self._metrics_collection_disabled:
|
|
return reporter_pb2.ReportOCMetricsReply()
|
|
|
|
# This function receives a GRPC containing OpenCensus (OC) metrics
|
|
# from a Ray process, then exposes those metrics to Prometheus.
|
|
try:
|
|
worker_id = WorkerID(request.worker_id)
|
|
worker_id = None if worker_id.is_nil() else worker_id.hex()
|
|
self._metrics_agent.proxy_export_metrics(request.metrics, worker_id)
|
|
except Exception:
|
|
logger.error(traceback.format_exc())
|
|
return reporter_pb2.ReportOCMetricsReply()
|
|
|
|
def _export_histogram_data(
|
|
self,
|
|
metric: Metric,
|
|
) -> None:
|
|
"""
|
|
TODO(can-anyscale): once we launch the new open-telemetry stack, we need to
|
|
document and communicate that the histogram metric is an approximation to users.
|
|
The approximation is good enough for the dashboard to display the histogram
|
|
distribution. Only the sum of all data points will be the approximation. See
|
|
https://github.com/ray-project/ray/issues/54538 for the complete backlog of Ray
|
|
metric infra improvements.
|
|
|
|
Export histogram data points to OpenTelemetry Metric Recorder. A histogram
|
|
metric is aggregated into several internal representations in C++ side:
|
|
- sum of all buckets
|
|
- count of all buckets
|
|
- count per bucket
|
|
|
|
We reconstruct the histogram data points from these internal representations
|
|
and export them to OpenTelemetry Metric Recorder. The reconstruction is an
|
|
approximation, but it is good enough for the dashboard to display the histogram
|
|
data points.
|
|
"""
|
|
data_points = metric.histogram.data_points
|
|
if not data_points:
|
|
return
|
|
self._open_telemetry_metric_recorder.register_histogram_metric(
|
|
metric.name,
|
|
metric.description,
|
|
data_points[0].explicit_bounds,
|
|
)
|
|
# Collect all data points and record using a single call
|
|
batch_data_points = []
|
|
for data_point in data_points:
|
|
if data_point.count == 0:
|
|
continue
|
|
|
|
tags = {tag.key: tag.value.string_value for tag in data_point.attributes}
|
|
batch_data_points.append(
|
|
{
|
|
"tags": tags,
|
|
"bucket_counts": list(data_point.bucket_counts),
|
|
}
|
|
)
|
|
|
|
if batch_data_points:
|
|
# Keep a single label schema for each histogram batch before
|
|
# recording the reconstructed data points.
|
|
all_keys = sorted({k for dp in batch_data_points for k in dp["tags"]})
|
|
for dp in batch_data_points:
|
|
tags = dp["tags"]
|
|
dp["tags"] = {k: tags.get(k, "") for k in all_keys}
|
|
self._open_telemetry_metric_recorder.record_histogram_aggregated_batch(
|
|
metric.name,
|
|
batch_data_points,
|
|
)
|
|
|
|
def _export_number_data(
|
|
self,
|
|
metric: Metric,
|
|
) -> None:
|
|
data_points = []
|
|
if metric.WhichOneof("data") == "gauge":
|
|
self._open_telemetry_metric_recorder.register_gauge_metric(
|
|
metric.name,
|
|
metric.description,
|
|
)
|
|
data_points = metric.gauge.data_points
|
|
if metric.WhichOneof("data") == "sum":
|
|
if metric.sum.is_monotonic:
|
|
self._open_telemetry_metric_recorder.register_counter_metric(
|
|
metric.name,
|
|
metric.description,
|
|
)
|
|
else:
|
|
self._open_telemetry_metric_recorder.register_sum_metric(
|
|
metric.name,
|
|
metric.description,
|
|
)
|
|
data_points = metric.sum.data_points
|
|
for data_point in data_points:
|
|
self._open_telemetry_metric_recorder.set_metric_value(
|
|
metric.name,
|
|
{tag.key: tag.value.string_value for tag in data_point.attributes},
|
|
# Note that all data points received from other Ray components are
|
|
# always double values. This is because the c++ apis
|
|
# (open_telemetry_metric_recorder.cc) only create metrics with double
|
|
# values.
|
|
data_point.as_double,
|
|
)
|
|
|
|
async def Export(
|
|
self,
|
|
request: metrics_service_pb2.ExportMetricsServiceRequest,
|
|
context: ServicerContext,
|
|
) -> metrics_service_pb2.ExportMetricsServiceResponse:
|
|
"""
|
|
GRPC method that receives the open telemetry metrics exported from other Ray
|
|
components running in the same node (e.g., raylet, worker, etc.). This method
|
|
implements an interface of `metrics_service_pb2_grpc.MetricsServiceServicer` (https://github.com/open-telemetry/opentelemetry-proto/blob/main/opentelemetry/proto/collector/metrics/v1/metrics_service.proto#L30),
|
|
which is the default open-telemetry metrics service interface.
|
|
"""
|
|
for resource_metrics in request.resource_metrics:
|
|
for scope_metrics in resource_metrics.scope_metrics:
|
|
for metric in scope_metrics.metrics:
|
|
if metric.WhichOneof("data") == "histogram":
|
|
self._export_histogram_data(metric)
|
|
else:
|
|
self._export_number_data(metric)
|
|
|
|
return metrics_service_pb2.ExportMetricsServiceResponse()
|
|
|
|
@staticmethod
|
|
def _get_cpu_percent(in_k8s: bool):
|
|
if in_k8s:
|
|
return k8s_utils.cpu_percent()
|
|
else:
|
|
return psutil.cpu_percent()
|
|
|
|
def _get_gpu_usage(self):
|
|
"""Get GPU usage information using the GPU metric provider."""
|
|
return self._gpu_metric_provider.get_gpu_usage()
|
|
|
|
@staticmethod
|
|
def _get_tpu_usage() -> List[TpuUtilizationInfo]:
|
|
|
|
global enable_tpu_usage_check
|
|
if not enable_tpu_usage_check:
|
|
return []
|
|
|
|
if not TPU_DEVICE_PLUGIN_ADDR:
|
|
enable_tpu_usage_check = False
|
|
return []
|
|
|
|
endpoint = f"http://{TPU_DEVICE_PLUGIN_ADDR}/metrics"
|
|
try:
|
|
metrics = requests.get(endpoint).content
|
|
metrics = metrics.decode("utf-8")
|
|
except Exception as e:
|
|
logger.debug(
|
|
f"Failed to retrieve TPU information from device plugin: {endpoint} {e}"
|
|
)
|
|
enable_tpu_usage_check = False
|
|
return []
|
|
|
|
tpu_utilizations = []
|
|
# TPU metrics have a quirk where tensor core and memory bandwidth
|
|
# metrics are host metrics and indexed by chip 0 to N-1, while the
|
|
# other metrics are runtime metrics and are indexed globally in the
|
|
# slice.
|
|
# To make these useful in the dashboard, we want to re-canonicalize the
|
|
# global indices onto the local host indices.
|
|
# We partition the metrics into two groups to perform this re-indexing.
|
|
tpu_utilizations_host = []
|
|
tpu_utilizations_other = []
|
|
|
|
# See https://cloud.google.com/monitoring/api/metrics_gcp#gcp-tpu for
|
|
# schema.
|
|
# Sample should look like:
|
|
# Name: tensorcore_utilization_node Labels: {'accelerator_id': '4804690994094478883-0', 'make': 'cloud-tpu', 'model': 'tpu-v6e-slice', 'tpu_topology': '2x4'} Value: 0.0
|
|
try:
|
|
for family in text_string_to_metric_families(metrics):
|
|
for sample in family.samples:
|
|
# Skip irrelevant metrics
|
|
if not hasattr(sample, "labels"):
|
|
continue
|
|
if "accelerator_id" not in sample.labels:
|
|
continue
|
|
labels = sample.labels
|
|
accelerator_id = labels["accelerator_id"]
|
|
index = int(accelerator_id.split("-")[1])
|
|
|
|
info = TpuUtilizationInfo(
|
|
index=index,
|
|
name=accelerator_id,
|
|
tpu_type=labels["model"],
|
|
tpu_topology=labels["tpu_topology"],
|
|
tensorcore_utilization=0.0,
|
|
hbm_utilization=0.0,
|
|
duty_cycle=0.0,
|
|
memory_used=0,
|
|
memory_total=0,
|
|
)
|
|
|
|
known = True
|
|
is_host_metric = False
|
|
match sample.name:
|
|
case "memory_bandwidth_utilization":
|
|
info["hbm_utilization"] = sample.value
|
|
is_host_metric = True
|
|
case "tensorcore_utilization":
|
|
info["tensorcore_utilization"] = sample.value
|
|
is_host_metric = True
|
|
case "duty_cycle":
|
|
info["duty_cycle"] = sample.value
|
|
case "memory_used":
|
|
info["memory_used"] = sample.value
|
|
case "memory_total":
|
|
info["memory_total"] = sample.value
|
|
case _:
|
|
known = False
|
|
|
|
if known:
|
|
if is_host_metric:
|
|
tpu_utilizations_host.append(info)
|
|
else:
|
|
tpu_utilizations_other.append(info)
|
|
except Exception as e:
|
|
logger.debug(f"Failed to parse metrics from device plugin: {metrics} {e}")
|
|
return []
|
|
|
|
desired_indices = sorted({i["index"] for i in tpu_utilizations_host})
|
|
rewrite_indices = sorted({i["index"] for i in tpu_utilizations_other})
|
|
|
|
# Some TPU types do not have runtime metrics reported from the device
|
|
# plugin and the rewrite_indices list will be empty.
|
|
if len(rewrite_indices) > 0:
|
|
if len(rewrite_indices) != len(desired_indices):
|
|
logger.warning(
|
|
f"Failed to parse metrics from device plugin: two sets of metrics for different chip counts, {len(desired_indices)} vs {len(rewrite_indices)}"
|
|
)
|
|
return []
|
|
|
|
index_map = dict(zip(rewrite_indices, desired_indices))
|
|
for info in tpu_utilizations_other:
|
|
info["index"] = index_map[info["index"]]
|
|
|
|
tpu_utilizations = tpu_utilizations_host + tpu_utilizations_other
|
|
|
|
# Each collected sample records only one metric (e.g. duty cycle) during
|
|
# the metric interval for one TPU. So here we need to aggregate the
|
|
# sample records together. The aggregated list should be indexed by the
|
|
# TPU accelerator index.
|
|
merged_tpu_utilizations = {}
|
|
|
|
for info in tpu_utilizations:
|
|
index = int(info.get("index"))
|
|
if index in merged_tpu_utilizations:
|
|
merged_info = merged_tpu_utilizations[index]
|
|
merged_info["tensorcore_utilization"] += info.get(
|
|
"tensorcore_utilization"
|
|
)
|
|
merged_info["hbm_utilization"] += info.get("hbm_utilization")
|
|
merged_info["duty_cycle"] += info.get("duty_cycle")
|
|
merged_info["memory_used"] += info.get("memory_used")
|
|
merged_info["memory_total"] += info.get("memory_total")
|
|
else:
|
|
merged_info = TpuUtilizationInfo(
|
|
index=info.get("index"),
|
|
name=info.get("name"),
|
|
tpu_type=info.get("tpu_type"),
|
|
tpu_topology=info.get("tpu_topology"),
|
|
tensorcore_utilization=info.get("tensorcore_utilization"),
|
|
hbm_utilization=info.get("hbm_utilization"),
|
|
duty_cycle=info.get("duty_cycle"),
|
|
memory_used=info.get("memory_used"),
|
|
memory_total=info.get("memory_total"),
|
|
)
|
|
merged_tpu_utilizations[index] = merged_info
|
|
|
|
sorted_tpu_utilizations = [
|
|
value for _, value in sorted(merged_tpu_utilizations.items())
|
|
]
|
|
return sorted_tpu_utilizations
|
|
|
|
@staticmethod
|
|
def _get_boot_time():
|
|
if IN_KUBERNETES_POD:
|
|
# Return start time of container entrypoint
|
|
return psutil.Process(pid=1).create_time()
|
|
else:
|
|
return psutil.boot_time()
|
|
|
|
@staticmethod
|
|
def _get_network_stats():
|
|
ifaces = [
|
|
v for k, v in psutil.net_io_counters(pernic=True).items() if k[0] == "e"
|
|
]
|
|
|
|
sent = sum((iface.bytes_sent for iface in ifaces))
|
|
recv = sum((iface.bytes_recv for iface in ifaces))
|
|
return sent, recv
|
|
|
|
@staticmethod
|
|
def _get_mem_usage():
|
|
total = get_system_memory()
|
|
used = utils.get_used_memory()
|
|
available = total - used
|
|
percent = round(used / total, 3) * 100
|
|
return total, available, percent, used
|
|
|
|
@staticmethod
|
|
def _get_host_mem_usage():
|
|
vmem = psutil.virtual_memory()
|
|
return vmem.used, vmem.total
|
|
|
|
@staticmethod
|
|
def _get_disk_usage(temp_dir: str):
|
|
if IN_KUBERNETES_POD and not ENABLE_K8S_DISK_USAGE:
|
|
# If in a K8s pod, disable disk display by passing in dummy values.
|
|
sdiskusage = namedtuple("sdiskusage", ["total", "used", "free", "percent"])
|
|
return {"/": sdiskusage(total=1, used=0, free=1, percent=0.0)}
|
|
|
|
if sys.platform == "win32":
|
|
root = psutil.disk_partitions()[0].mountpoint
|
|
else:
|
|
root = os.sep
|
|
return {
|
|
"/": psutil.disk_usage(root),
|
|
temp_dir: psutil.disk_usage(temp_dir),
|
|
}
|
|
|
|
@staticmethod
|
|
def _get_disk_io_stats():
|
|
stats = psutil.disk_io_counters()
|
|
# stats can be None or {} if the machine is diskless.
|
|
# https://psutil.readthedocs.io/en/latest/#psutil.disk_io_counters
|
|
if not stats:
|
|
return (0, 0, 0, 0)
|
|
else:
|
|
return (
|
|
stats.read_bytes,
|
|
stats.write_bytes,
|
|
stats.read_count,
|
|
stats.write_count,
|
|
)
|
|
|
|
async def _async_get_worker_pids_from_raylet(self) -> List[int]:
|
|
try:
|
|
# Get worker pids from raylet via gRPC.
|
|
return await self._raylet_client.async_get_worker_pids()
|
|
except (GetTimeoutError, RpcError):
|
|
logger.exception("Failed to get worker pids from raylet")
|
|
return []
|
|
|
|
async def _async_get_agent_pids_from_raylet(self) -> List[int]:
|
|
try:
|
|
# Get agents pids from raylet via gRPC.
|
|
return await self._raylet_client.async_get_agent_pids()
|
|
except (GetTimeoutError, RpcError):
|
|
logger.exception("Failed to get agents pids from raylet")
|
|
return []
|
|
|
|
def _get_agent_proc(self) -> psutil.Process:
|
|
# Agent is the current process.
|
|
# This method is not necessary, but we have it for mock testing.
|
|
return psutil.Process()
|
|
|
|
def _generate_proc_key(self, proc: psutil.Process) -> Tuple[int, float]:
|
|
return (proc.pid, proc.create_time())
|
|
|
|
async def _async_get_worker_processes(self):
|
|
pids = await self._async_get_worker_pids_from_raylet()
|
|
logger.debug(f"Worker PIDs from raylet: {pids}")
|
|
if not pids:
|
|
return {}
|
|
workers = {}
|
|
for pid in pids:
|
|
try:
|
|
proc = psutil.Process(pid)
|
|
workers[self._generate_proc_key(proc)] = proc
|
|
except (psutil.NoSuchProcess, psutil.AccessDenied):
|
|
logger.error(f"Failed to access worker process {pid}")
|
|
continue
|
|
return workers
|
|
|
|
async def _async_get_agent_processes(self):
|
|
pids = await self._async_get_agent_pids_from_raylet()
|
|
logger.debug(f"Agent PIDs from raylet: {pids}")
|
|
if not pids:
|
|
return {}
|
|
agents = {}
|
|
for pid in pids:
|
|
try:
|
|
proc = psutil.Process(pid)
|
|
agents[self._generate_proc_key(proc)] = proc
|
|
except (psutil.NoSuchProcess, psutil.AccessDenied):
|
|
logger.error(f"Failed to access agent process {pid}")
|
|
continue
|
|
return agents
|
|
|
|
async def _async_get_workers_and_agents(
|
|
self, gpus: Optional[List[GpuUtilizationInfo]] = None
|
|
):
|
|
workers, agents = await asyncio.gather(
|
|
self._async_get_worker_processes(), self._async_get_agent_processes()
|
|
)
|
|
workers.update(agents)
|
|
if not workers:
|
|
return []
|
|
else:
|
|
# We should keep `raylet_proc.children()` in `self` because
|
|
# when `cpu_percent` is first called, it returns the meaningless 0.
|
|
# See more: https://github.com/ray-project/ray/issues/29848
|
|
keys_to_pop = []
|
|
# Add all new workers.
|
|
for key, worker in workers.items():
|
|
if key not in self._workers:
|
|
self._workers[key] = worker
|
|
|
|
# Pop out stale workers.
|
|
for key in self._workers:
|
|
if key not in workers:
|
|
keys_to_pop.append(key)
|
|
for k in keys_to_pop:
|
|
self._workers.pop(k)
|
|
|
|
# Build process ID -> GPU info mapping for faster lookups
|
|
gpu_pid_mapping = defaultdict(list)
|
|
if gpus is not None:
|
|
for gpu in gpus:
|
|
processes = gpu.get("processes_pids")
|
|
if processes:
|
|
for proc in processes.values():
|
|
gpu_pid_mapping[proc["pid"]].append(proc)
|
|
|
|
result = []
|
|
for w in self._workers.values():
|
|
try:
|
|
if w.status() == psutil.STATUS_ZOMBIE:
|
|
continue
|
|
|
|
# Get basic process info
|
|
worker_info = w.as_dict(attrs=PSUTIL_PROCESS_ATTRS)
|
|
|
|
# Add GPU information if available
|
|
worker_pid = worker_info["pid"]
|
|
gpu_memory_usage = 0
|
|
gpu_utilization = 0
|
|
|
|
if worker_pid in gpu_pid_mapping:
|
|
# Aggregate GPU memory and utilization across all GPUs for this process
|
|
for gpu_proc in gpu_pid_mapping[worker_pid]:
|
|
gpu_memory_usage += gpu_proc["gpu_memory_usage"]
|
|
utilization = gpu_proc["gpu_utilization"] or 0
|
|
gpu_utilization += utilization
|
|
|
|
# Add GPU information to worker info
|
|
worker_info["gpu_memory_usage"] = gpu_memory_usage # in MB
|
|
worker_info["gpu_utilization"] = gpu_utilization # percentage
|
|
|
|
result.append(worker_info)
|
|
except psutil.NoSuchProcess:
|
|
# the process may have terminated due to race condition.
|
|
continue
|
|
|
|
return result
|
|
|
|
def _get_raylet_proc(self):
|
|
try:
|
|
if not self._raylet_proc:
|
|
curr_proc = psutil.Process()
|
|
# The dashboard agent is a child of the raylet process.
|
|
# It is not necessarily the direct child (python-windows
|
|
# typically uses a py.exe runner to run python), so search
|
|
# up for a process named 'raylet'
|
|
candidate = curr_proc.parent()
|
|
while candidate:
|
|
if "raylet" in candidate.name():
|
|
break
|
|
candidate = candidate.parent()
|
|
self._raylet_proc = candidate
|
|
|
|
if self._raylet_proc is not None:
|
|
if self._raylet_proc.pid == 1:
|
|
return None
|
|
if self._raylet_proc.status() == psutil.STATUS_ZOMBIE:
|
|
return None
|
|
return self._raylet_proc
|
|
except (psutil.AccessDenied, ProcessLookupError):
|
|
pass
|
|
return None
|
|
|
|
def _get_gcs(self):
|
|
if self._gcs_pid:
|
|
if not self._gcs_proc or self._gcs_pid != self._gcs_proc.pid:
|
|
self._gcs_proc = psutil.Process(self._gcs_pid)
|
|
if self._gcs_proc:
|
|
dictionary = self._gcs_proc.as_dict(attrs=PSUTIL_PROCESS_ATTRS)
|
|
return dictionary
|
|
return {}
|
|
|
|
def _get_raylet(self):
|
|
raylet_proc = self._get_raylet_proc()
|
|
if raylet_proc is None:
|
|
return None
|
|
else:
|
|
return raylet_proc.as_dict(attrs=PSUTIL_PROCESS_ATTRS)
|
|
|
|
def _get_agent(self):
|
|
# Current proc == agent proc
|
|
if not self._agent_proc:
|
|
self._agent_proc = psutil.Process()
|
|
return self._agent_proc.as_dict(attrs=PSUTIL_PROCESS_ATTRS)
|
|
|
|
def _get_load_avg(self):
|
|
if sys.platform == "win32":
|
|
cpu_percent = psutil.cpu_percent()
|
|
load = (cpu_percent, cpu_percent, cpu_percent)
|
|
else:
|
|
load = os.getloadavg()
|
|
if self._cpu_counts[0] > 0:
|
|
per_cpu_load = tuple((round(x / self._cpu_counts[0], 2) for x in load))
|
|
else:
|
|
per_cpu_load = None
|
|
return load, per_cpu_load
|
|
|
|
@staticmethod
|
|
def _compute_speed_from_hist(hist):
|
|
while len(hist) > 7:
|
|
hist.pop(0)
|
|
then, prev_stats = hist[0]
|
|
now, now_stats = hist[-1]
|
|
time_delta = now - then
|
|
return tuple((y - x) / time_delta for x, y in zip(prev_stats, now_stats))
|
|
|
|
def _get_shm_usage(self):
|
|
"""Return the shm usage.
|
|
|
|
If shm doesn't exist (e.g., MacOS), it returns None.
|
|
"""
|
|
mem = psutil.virtual_memory()
|
|
if not hasattr(mem, "shared"):
|
|
return None
|
|
return mem.shared
|
|
|
|
async def _async_collect_stats(self):
|
|
now = dashboard_utils.to_posix_time(datetime.datetime.utcnow())
|
|
network_stats = self._get_network_stats()
|
|
self._network_stats_hist.append((now, network_stats))
|
|
network_speed_stats = self._compute_speed_from_hist(self._network_stats_hist)
|
|
|
|
disk_stats = self._get_disk_io_stats()
|
|
self._disk_io_stats_hist.append((now, disk_stats))
|
|
disk_speed_stats = self._compute_speed_from_hist(self._disk_io_stats_hist)
|
|
|
|
gpus = self._get_gpu_usage()
|
|
raylet = self._get_raylet()
|
|
stats = {
|
|
"now": now,
|
|
"hostname": self._hostname,
|
|
"ip": self._ip,
|
|
"cpu": self._get_cpu_percent(IN_KUBERNETES_POD),
|
|
"cpus": self._cpu_counts,
|
|
"mem": self._get_mem_usage(),
|
|
# Unit is in bytes. None if
|
|
"shm": self._get_shm_usage(),
|
|
"host_mem": self._get_host_mem_usage(),
|
|
"cgroup_mem": utils.get_cgroup_mem_stats(),
|
|
"workers": await self._async_get_workers_and_agents(gpus),
|
|
"raylet": raylet,
|
|
"agent": self._get_agent(),
|
|
"bootTime": self._get_boot_time(),
|
|
"loadAvg": self._get_load_avg(),
|
|
"disk": self._get_disk_usage(self._dashboard_agent.temp_dir),
|
|
"disk_io": disk_stats,
|
|
"disk_io_speed": disk_speed_stats,
|
|
"gpus": gpus,
|
|
"tpus": self._get_tpu_usage(),
|
|
"network": network_stats,
|
|
"network_speed": network_speed_stats,
|
|
# Deprecated field, should be removed with frontend.
|
|
"cmdline": raylet.get("cmdline", []) if raylet else [],
|
|
}
|
|
if self._is_head_node:
|
|
stats["gcs"] = self._get_gcs()
|
|
return stats
|
|
|
|
def _generate_reseted_stats_record(self, component_name: str) -> List[Record]:
|
|
"""Return a list of Record that will reset
|
|
the system metrics of a given component name.
|
|
|
|
Args:
|
|
component_name: a component name for a given stats.
|
|
|
|
Returns:
|
|
a list of Record instances of all values 0.
|
|
"""
|
|
tags = {"ip": self._ip, "Component": component_name}
|
|
|
|
records = []
|
|
records.append(
|
|
Record(
|
|
gauge=METRICS_GAUGES["component_cpu_percentage"],
|
|
value=0.0,
|
|
tags=tags,
|
|
)
|
|
)
|
|
records.append(
|
|
Record(
|
|
gauge=METRICS_GAUGES["component_shared_bytes"],
|
|
value=0.0,
|
|
tags=tags,
|
|
)
|
|
)
|
|
records.append(
|
|
Record(
|
|
gauge=METRICS_GAUGES["component_rss_mb"],
|
|
value=0.0,
|
|
tags=tags,
|
|
)
|
|
)
|
|
records.append(
|
|
Record(
|
|
gauge=METRICS_GAUGES["component_rss_bytes"],
|
|
value=0.0,
|
|
tags=tags,
|
|
)
|
|
)
|
|
records.append(
|
|
Record(
|
|
gauge=METRICS_GAUGES["component_uss_mb"],
|
|
value=0.0,
|
|
tags=tags,
|
|
)
|
|
)
|
|
records.append(
|
|
Record(
|
|
gauge=METRICS_GAUGES["component_uss_bytes"],
|
|
value=0.0,
|
|
tags=tags,
|
|
)
|
|
)
|
|
records.append(
|
|
Record(
|
|
gauge=METRICS_GAUGES["component_num_fds"],
|
|
value=0,
|
|
tags=tags,
|
|
)
|
|
)
|
|
|
|
return records
|
|
|
|
def _generate_system_stats_record(
|
|
self, stats: List[dict], component_name: str, pid: Optional[str] = None
|
|
) -> List[Record]:
|
|
"""Generate a list of Record class from a given component names.
|
|
|
|
Args:
|
|
stats: a list of stats dict generated by `psutil.as_dict`.
|
|
If empty, it will create the metrics of a given "component_name"
|
|
which has all 0 values.
|
|
component_name: a component name for a given stats.
|
|
pid: optionally provided pids.
|
|
|
|
Returns:
|
|
a list of Record class that will be exposed to Prometheus.
|
|
"""
|
|
total_cpu_percentage = 0.0
|
|
total_gpu_percentage = 0.0
|
|
total_gpu_memory = 0.0
|
|
total_rss_bytes = 0.0
|
|
total_uss_bytes = 0.0
|
|
total_shm_bytes = 0.0
|
|
total_num_fds = 0
|
|
for stat in stats:
|
|
total_cpu_percentage += float(stat.get("cpu_percent", 0.0)) # noqa
|
|
|
|
# Aggregate GPU stats if available
|
|
total_gpu_percentage += float(stat.get("gpu_utilization", 0.0))
|
|
total_gpu_memory += float(stat.get("gpu_memory_usage", 0.0))
|
|
|
|
memory_info = stat.get("memory_info")
|
|
if memory_info:
|
|
total_rss_bytes += float(memory_info.rss)
|
|
if hasattr(memory_info, "shared"):
|
|
total_shm_bytes += float(memory_info.shared)
|
|
mem_full_info = stat.get("memory_full_info")
|
|
if mem_full_info is not None:
|
|
# For Mac OS X, directly get USS metric from memory_full_info
|
|
total_uss_bytes += float(mem_full_info.uss)
|
|
elif memory_info is not None:
|
|
# For linux or windows, memory_full_info is not collected. Approximated USS from memory_info
|
|
if hasattr(memory_info, "shared"):
|
|
# Linux: USS ≈ RSS - shared
|
|
total_uss_bytes += float(memory_info.rss - memory_info.shared)
|
|
elif hasattr(memory_info, "private"):
|
|
# Windows: private IS USS
|
|
total_uss_bytes += float(memory_info.private)
|
|
total_num_fds += int(stat.get("num_fds", 0))
|
|
|
|
tags = {"ip": self._ip, "Component": component_name}
|
|
if pid:
|
|
tags["pid"] = pid
|
|
|
|
records = []
|
|
records.append(
|
|
Record(
|
|
gauge=METRICS_GAUGES["component_cpu_percentage"],
|
|
value=total_cpu_percentage,
|
|
tags=tags,
|
|
)
|
|
)
|
|
records.append(
|
|
Record(
|
|
gauge=METRICS_GAUGES["component_shared_bytes"],
|
|
value=total_shm_bytes,
|
|
tags=tags,
|
|
)
|
|
)
|
|
records.append(
|
|
Record(
|
|
gauge=METRICS_GAUGES["component_rss_mb"],
|
|
value=total_rss_bytes / 1.0e6,
|
|
tags=tags,
|
|
)
|
|
)
|
|
records.append(
|
|
Record(
|
|
gauge=METRICS_GAUGES["component_rss_bytes"],
|
|
value=total_rss_bytes,
|
|
tags=tags,
|
|
)
|
|
)
|
|
if total_uss_bytes > 0.0:
|
|
records.append(
|
|
Record(
|
|
gauge=METRICS_GAUGES["component_uss_mb"],
|
|
value=total_uss_bytes / 1.0e6,
|
|
tags=tags,
|
|
)
|
|
)
|
|
records.append(
|
|
Record(
|
|
gauge=METRICS_GAUGES["component_uss_bytes"],
|
|
value=total_uss_bytes,
|
|
tags=tags,
|
|
)
|
|
)
|
|
records.append(
|
|
Record(
|
|
gauge=METRICS_GAUGES["component_num_fds"],
|
|
value=total_num_fds,
|
|
tags=tags,
|
|
)
|
|
)
|
|
|
|
# Add GPU records if there's GPU usage
|
|
if total_gpu_memory > 0.0:
|
|
records.append(
|
|
Record(
|
|
gauge=METRICS_GAUGES["component_gpu_memory_mb"],
|
|
value=total_gpu_memory,
|
|
tags=tags,
|
|
)
|
|
)
|
|
|
|
if total_gpu_percentage > 0.0:
|
|
records.append(
|
|
Record(
|
|
gauge=METRICS_GAUGES["component_gpu_percentage"],
|
|
value=total_gpu_percentage,
|
|
tags=tags,
|
|
)
|
|
)
|
|
|
|
return records
|
|
|
|
def _generate_reseted_gpu_stats_record(self, component_name: str) -> List[Record]:
|
|
"""Return a list of Record that will reset
|
|
the GPU metrics of a given component name.
|
|
|
|
Args:
|
|
component_name: a component name for a given stats.
|
|
|
|
Returns:
|
|
a list of Record instances of GPU metrics with all values 0.
|
|
"""
|
|
tags = {"ip": self._ip, "Component": component_name}
|
|
|
|
records = []
|
|
records.append(
|
|
Record(
|
|
gauge=METRICS_GAUGES["component_gpu_memory_mb"],
|
|
value=0.0,
|
|
tags=tags,
|
|
)
|
|
)
|
|
records.append(
|
|
Record(
|
|
gauge=METRICS_GAUGES["component_gpu_percentage"],
|
|
value=0.0,
|
|
tags=tags,
|
|
)
|
|
)
|
|
|
|
return records
|
|
|
|
def generate_worker_stats_record(self, worker_stats: List[dict]) -> List[Record]:
|
|
"""Generate a list of Record class for worker processes.
|
|
|
|
This API automatically sets the component_name of record as
|
|
the name of worker processes. I.e., ray::* so that we can report
|
|
per task/actor (grouped by a func/class name) resource usages.
|
|
|
|
Args:
|
|
worker_stats: a list of stats dict generated by `psutil.as_dict`
|
|
for worker processes. Now with gpu usage information.
|
|
|
|
Returns:
|
|
A list of Record entries with per-process system and GPU stats,
|
|
including reset records for processes that no longer exist or no
|
|
longer report GPU usage.
|
|
"""
|
|
# worker cmd name (ray::*) -> stats dict.
|
|
proc_name_to_stats = defaultdict(list)
|
|
gpu_worker_proc_names = set() # Track processes with GPU usage
|
|
|
|
for stat in worker_stats:
|
|
cmdline = stat.get("cmdline")
|
|
# collect both worker and driver stats
|
|
if cmdline:
|
|
proc_name = cmdline[0]
|
|
proc_name_to_stats[proc_name].append(stat)
|
|
|
|
# Track if this process has GPU usage
|
|
if (
|
|
stat.get("gpu_memory_usage", 0) > 0
|
|
or stat.get("gpu_utilization", 0) > 0
|
|
):
|
|
gpu_worker_proc_names.add(proc_name)
|
|
|
|
records = []
|
|
|
|
# Generate system stats records (now includes GPU stats)
|
|
for proc_name, stats in proc_name_to_stats.items():
|
|
records.extend(self._generate_system_stats_record(stats, proc_name))
|
|
|
|
# Reset worker metrics that are from finished processes.
|
|
new_proc_names = set(proc_name_to_stats.keys())
|
|
stale_procs = self._latest_worker_proc_names - new_proc_names
|
|
self._latest_worker_proc_names = new_proc_names
|
|
|
|
for stale_proc_name in stale_procs:
|
|
records.extend(self._generate_reseted_stats_record(stale_proc_name))
|
|
|
|
# Reset GPU metrics for processes that no longer use GPU
|
|
stale_gpu_worker_proc_names = (
|
|
self._latest_gpu_worker_proc_names - gpu_worker_proc_names
|
|
)
|
|
self._latest_gpu_worker_proc_names = gpu_worker_proc_names
|
|
|
|
for stale_gpu_proc in stale_gpu_worker_proc_names:
|
|
records.extend(self._generate_reseted_gpu_stats_record(stale_gpu_proc))
|
|
|
|
return records
|
|
|
|
def _to_records(self, stats, cluster_stats) -> List[Record]:
|
|
records_reported = []
|
|
ip = stats["ip"]
|
|
ray_node_type = "head" if self._is_head_node else "worker"
|
|
is_head_node = "true" if self._is_head_node else "false"
|
|
|
|
# Common tags for node-level metrics
|
|
# We use RayNodeType to mark head/worker node, IsHeadNode is retained for backward compatibility
|
|
node_tags = {"ip": ip, "RayNodeType": ray_node_type, "IsHeadNode": is_head_node}
|
|
|
|
# -- Instance count of cluster --
|
|
# Only report cluster stats on head node
|
|
if "autoscaler_report" in cluster_stats and self._is_head_node:
|
|
active_nodes = cluster_stats["autoscaler_report"]["active_nodes"]
|
|
for node_type, active_node_count in active_nodes.items():
|
|
records_reported.append(
|
|
Record(
|
|
gauge=METRICS_GAUGES["cluster_active_nodes"],
|
|
value=active_node_count,
|
|
tags={"node_type": node_type},
|
|
)
|
|
)
|
|
|
|
# Emit cluster_idle_nodes only for autoscaler v2 (v1 has no "idle" state).
|
|
idle_nodes = cluster_stats.get("autoscaler_report", {}).get("idle_nodes")
|
|
if idle_nodes is not None:
|
|
for node_type, idle_node_count in idle_nodes.items():
|
|
records_reported.append(
|
|
Record(
|
|
gauge=METRICS_GAUGES["cluster_idle_nodes"],
|
|
value=idle_node_count,
|
|
tags={"node_type": node_type},
|
|
)
|
|
)
|
|
|
|
failed_nodes = cluster_stats["autoscaler_report"]["failed_nodes"]
|
|
failed_nodes_dict = {}
|
|
for node_ip, node_type in failed_nodes:
|
|
if node_type in failed_nodes_dict:
|
|
failed_nodes_dict[node_type] += 1
|
|
else:
|
|
failed_nodes_dict[node_type] = 1
|
|
|
|
for node_type, failed_node_count in failed_nodes_dict.items():
|
|
records_reported.append(
|
|
Record(
|
|
gauge=METRICS_GAUGES["cluster_failed_nodes"],
|
|
value=failed_node_count,
|
|
tags={"node_type": node_type},
|
|
)
|
|
)
|
|
|
|
pending_nodes = cluster_stats["autoscaler_report"]["pending_nodes"]
|
|
pending_nodes_dict = {}
|
|
for node_ip, node_type, status_message in pending_nodes:
|
|
if node_type in pending_nodes_dict:
|
|
pending_nodes_dict[node_type] += 1
|
|
else:
|
|
pending_nodes_dict[node_type] = 1
|
|
|
|
for node_type, pending_node_count in pending_nodes_dict.items():
|
|
records_reported.append(
|
|
Record(
|
|
gauge=METRICS_GAUGES["cluster_pending_nodes"],
|
|
value=pending_node_count,
|
|
tags={"node_type": node_type},
|
|
)
|
|
)
|
|
|
|
# -- CPU per node --
|
|
cpu_usage = float(stats["cpu"])
|
|
cpu_record = Record(
|
|
gauge=METRICS_GAUGES["node_cpu_utilization"],
|
|
value=cpu_usage,
|
|
tags=node_tags,
|
|
)
|
|
|
|
cpu_count, _ = stats["cpus"]
|
|
cpu_count_record = Record(
|
|
gauge=METRICS_GAUGES["node_cpu_count"], value=cpu_count, tags=node_tags
|
|
)
|
|
|
|
# -- Mem per node --
|
|
mem_total, mem_available, _, mem_used = stats["mem"]
|
|
mem_used_record = Record(
|
|
gauge=METRICS_GAUGES["node_mem_used"], value=mem_used, tags=node_tags
|
|
)
|
|
mem_available_record = Record(
|
|
gauge=METRICS_GAUGES["node_mem_available"],
|
|
value=mem_available,
|
|
tags=node_tags,
|
|
)
|
|
mem_total_record = Record(
|
|
gauge=METRICS_GAUGES["node_mem_total"], value=mem_total, tags=node_tags
|
|
)
|
|
|
|
shm_used = stats["shm"]
|
|
if shm_used:
|
|
node_mem_shared = Record(
|
|
gauge=METRICS_GAUGES["node_mem_shared_bytes"],
|
|
value=shm_used,
|
|
tags=node_tags,
|
|
)
|
|
records_reported.append(node_mem_shared)
|
|
|
|
host_mem_used, host_mem_total = stats["host_mem"]
|
|
records_reported.extend(
|
|
[
|
|
Record(
|
|
gauge=METRICS_GAUGES["node_mem_used_host"],
|
|
value=host_mem_used,
|
|
tags=node_tags,
|
|
),
|
|
Record(
|
|
gauge=METRICS_GAUGES["node_mem_total_host"],
|
|
value=host_mem_total,
|
|
tags=node_tags,
|
|
),
|
|
]
|
|
)
|
|
|
|
cgroup_stats = stats["cgroup_mem"]
|
|
if cgroup_stats is not None:
|
|
cgroup_used, cgroup_total = cgroup_stats
|
|
records_reported.extend(
|
|
[
|
|
Record(
|
|
gauge=METRICS_GAUGES["node_cgroup_mem_used"],
|
|
value=cgroup_used,
|
|
tags=node_tags,
|
|
),
|
|
Record(
|
|
gauge=METRICS_GAUGES["node_cgroup_mem_total"],
|
|
value=cgroup_total,
|
|
tags=node_tags,
|
|
),
|
|
]
|
|
)
|
|
|
|
# The output example of GpuUtilizationInfo.
|
|
"""
|
|
{'index': 0,
|
|
'uuid': 'GPU-36e1567d-37ed-051e-f8ff-df807517b396',
|
|
'name': 'NVIDIA A10G',
|
|
'utilization_gpu': 1,
|
|
'memory_used': 0,
|
|
'memory_total': 22731}
|
|
"""
|
|
# -- GPU per node --
|
|
gpus = stats["gpus"]
|
|
gpus_available = len(gpus)
|
|
|
|
if gpus_available:
|
|
for gpu in gpus:
|
|
gpus_utilization, gram_used, gram_total = 0, 0, 0
|
|
# Consume GPU may not report its utilization.
|
|
if gpu["utilization_gpu"] is not None:
|
|
gpus_utilization += gpu["utilization_gpu"]
|
|
gram_used += gpu["memory_used"]
|
|
gram_total += gpu["memory_total"]
|
|
gpu_index = gpu.get("index")
|
|
gpu_name = gpu.get("name")
|
|
gpu_power_mw = gpu.get("power_mw")
|
|
gpu_temperature_c = gpu.get("temperature_c")
|
|
|
|
gram_available = gram_total - gram_used
|
|
|
|
if gpu_index is not None:
|
|
gpu_tags = {**node_tags, "GpuIndex": str(gpu_index)}
|
|
if gpu_name:
|
|
gpu_tags["GpuDeviceName"] = gpu_name
|
|
|
|
# There's only 1 GPU per each index, so we record 1 here.
|
|
gpus_available_record = Record(
|
|
gauge=METRICS_GAUGES["node_gpus_available"],
|
|
value=1,
|
|
tags=gpu_tags,
|
|
)
|
|
gpus_utilization_record = Record(
|
|
gauge=METRICS_GAUGES["node_gpus_utilization"],
|
|
value=gpus_utilization,
|
|
tags=gpu_tags,
|
|
)
|
|
gram_used_record = Record(
|
|
gauge=METRICS_GAUGES["node_gram_used"],
|
|
value=gram_used,
|
|
tags=gpu_tags,
|
|
)
|
|
gram_available_record = Record(
|
|
gauge=METRICS_GAUGES["node_gram_available"],
|
|
value=gram_available,
|
|
tags=gpu_tags,
|
|
)
|
|
gpu_records_to_add = [
|
|
gpus_available_record,
|
|
gpus_utilization_record,
|
|
gram_used_record,
|
|
gram_available_record,
|
|
]
|
|
# Optional GPU power and temperature (e.g. NVIDIA, AMD)
|
|
if gpu_power_mw is not None:
|
|
gpu_records_to_add.append(
|
|
Record(
|
|
gauge=METRICS_GAUGES["node_gpu_power_milliwatts"],
|
|
value=gpu_power_mw,
|
|
tags=gpu_tags,
|
|
)
|
|
)
|
|
if gpu_temperature_c is not None:
|
|
gpu_records_to_add.append(
|
|
Record(
|
|
gauge=METRICS_GAUGES["node_gpu_temperature_celsius"],
|
|
value=gpu_temperature_c,
|
|
tags=gpu_tags,
|
|
)
|
|
)
|
|
records_reported.extend(gpu_records_to_add)
|
|
|
|
# -- TPU per node --
|
|
tpus = stats["tpus"]
|
|
|
|
for tpu in tpus:
|
|
tpu_index = tpu.get("index")
|
|
tpu_name = tpu.get("name")
|
|
tpu_type = tpu.get("tpu_type")
|
|
tpu_topology = tpu.get("tpu_topology")
|
|
tensorcore_utilization = tpu.get("tensorcore_utilization")
|
|
hbm_utilization = tpu.get("hbm_utilization")
|
|
duty_cycle = tpu.get("duty_cycle")
|
|
memory_used = tpu.get("memory_used")
|
|
memory_total = tpu.get("memory_total")
|
|
|
|
tpu_tags = {
|
|
**node_tags,
|
|
"TpuIndex": str(tpu_index),
|
|
"TpuDeviceName": tpu_name,
|
|
"TpuType": tpu_type,
|
|
"TpuTopology": tpu_topology,
|
|
}
|
|
tensorcore_utilization_record = Record(
|
|
gauge=METRICS_GAUGES["tpu_tensorcore_utilization"],
|
|
value=tensorcore_utilization,
|
|
tags=tpu_tags,
|
|
)
|
|
hbm_utilization_record = Record(
|
|
gauge=METRICS_GAUGES["tpu_memory_bandwidth_utilization"],
|
|
value=hbm_utilization,
|
|
tags=tpu_tags,
|
|
)
|
|
duty_cycle_record = Record(
|
|
gauge=METRICS_GAUGES["tpu_duty_cycle"],
|
|
value=duty_cycle,
|
|
tags=tpu_tags,
|
|
)
|
|
memory_used_record = Record(
|
|
gauge=METRICS_GAUGES["tpu_memory_used"],
|
|
value=memory_used,
|
|
tags=tpu_tags,
|
|
)
|
|
memory_total_record = Record(
|
|
gauge=METRICS_GAUGES["tpu_memory_total"],
|
|
value=memory_total,
|
|
tags=tpu_tags,
|
|
)
|
|
records_reported.extend(
|
|
[
|
|
tensorcore_utilization_record,
|
|
hbm_utilization_record,
|
|
duty_cycle_record,
|
|
memory_used_record,
|
|
memory_total_record,
|
|
]
|
|
)
|
|
|
|
# -- Disk per node --
|
|
disk_io_stats = stats["disk_io"]
|
|
disk_read_record = Record(
|
|
gauge=METRICS_GAUGES["node_disk_io_read"],
|
|
value=disk_io_stats[0],
|
|
tags=node_tags,
|
|
)
|
|
disk_write_record = Record(
|
|
gauge=METRICS_GAUGES["node_disk_io_write"],
|
|
value=disk_io_stats[1],
|
|
tags=node_tags,
|
|
)
|
|
disk_read_count_record = Record(
|
|
gauge=METRICS_GAUGES["node_disk_io_read_count"],
|
|
value=disk_io_stats[2],
|
|
tags=node_tags,
|
|
)
|
|
disk_write_count_record = Record(
|
|
gauge=METRICS_GAUGES["node_disk_io_write_count"],
|
|
value=disk_io_stats[3],
|
|
tags=node_tags,
|
|
)
|
|
disk_io_speed_stats = stats["disk_io_speed"]
|
|
disk_read_speed_record = Record(
|
|
gauge=METRICS_GAUGES["node_disk_io_read_speed"],
|
|
value=disk_io_speed_stats[0],
|
|
tags=node_tags,
|
|
)
|
|
disk_write_speed_record = Record(
|
|
gauge=METRICS_GAUGES["node_disk_io_write_speed"],
|
|
value=disk_io_speed_stats[1],
|
|
tags=node_tags,
|
|
)
|
|
disk_read_iops_record = Record(
|
|
gauge=METRICS_GAUGES["node_disk_read_iops"],
|
|
value=disk_io_speed_stats[2],
|
|
tags=node_tags,
|
|
)
|
|
disk_write_iops_record = Record(
|
|
gauge=METRICS_GAUGES["node_disk_write_iops"],
|
|
value=disk_io_speed_stats[3],
|
|
tags=node_tags,
|
|
)
|
|
used = stats["disk"]["/"].used
|
|
free = stats["disk"]["/"].free
|
|
disk_utilization = float(used / (used + free)) * 100
|
|
disk_usage_record = Record(
|
|
gauge=METRICS_GAUGES["node_disk_usage"], value=used, tags=node_tags
|
|
)
|
|
disk_free_record = Record(
|
|
gauge=METRICS_GAUGES["node_disk_free"], value=free, tags=node_tags
|
|
)
|
|
disk_utilization_percentage_record = Record(
|
|
gauge=METRICS_GAUGES["node_disk_utilization_percentage"],
|
|
value=disk_utilization,
|
|
tags=node_tags,
|
|
)
|
|
|
|
# -- Network speed (send/receive) stats per node --
|
|
network_stats = stats["network"]
|
|
network_sent_record = Record(
|
|
gauge=METRICS_GAUGES["node_network_sent"],
|
|
value=network_stats[0],
|
|
tags=node_tags,
|
|
)
|
|
network_received_record = Record(
|
|
gauge=METRICS_GAUGES["node_network_received"],
|
|
value=network_stats[1],
|
|
tags=node_tags,
|
|
)
|
|
|
|
# -- Network speed (send/receive) per node --
|
|
network_speed_stats = stats["network_speed"]
|
|
network_send_speed_record = Record(
|
|
gauge=METRICS_GAUGES["node_network_send_speed"],
|
|
value=network_speed_stats[0],
|
|
tags=node_tags,
|
|
)
|
|
network_receive_speed_record = Record(
|
|
gauge=METRICS_GAUGES["node_network_receive_speed"],
|
|
value=network_speed_stats[1],
|
|
tags=node_tags,
|
|
)
|
|
|
|
"""
|
|
Record system stats.
|
|
"""
|
|
|
|
if self._is_head_node:
|
|
gcs_stats = stats["gcs"]
|
|
if gcs_stats:
|
|
records_reported.extend(
|
|
self._generate_system_stats_record(
|
|
[gcs_stats], "gcs", pid=str(gcs_stats["pid"])
|
|
)
|
|
)
|
|
|
|
# Record component metrics.
|
|
raylet_stats = stats["raylet"]
|
|
if raylet_stats:
|
|
raylet_pid = str(raylet_stats["pid"])
|
|
records_reported.extend(
|
|
self._generate_system_stats_record(
|
|
[raylet_stats], "raylet", pid=raylet_pid
|
|
)
|
|
)
|
|
workers_stats = stats["workers"]
|
|
records_reported.extend(self.generate_worker_stats_record(workers_stats))
|
|
agent_stats = stats["agent"]
|
|
if agent_stats:
|
|
agent_pid = str(agent_stats["pid"])
|
|
records_reported.extend(
|
|
self._generate_system_stats_record(
|
|
[agent_stats], "agent", pid=agent_pid
|
|
)
|
|
)
|
|
|
|
# NOTE: Dashboard metrics is recorded within the dashboard because
|
|
# it can be deployed as a standalone instance. It shouldn't
|
|
# depend on the agent.
|
|
|
|
records_reported.extend(
|
|
[
|
|
cpu_record,
|
|
cpu_count_record,
|
|
mem_used_record,
|
|
mem_available_record,
|
|
mem_total_record,
|
|
disk_read_record,
|
|
disk_write_record,
|
|
disk_read_count_record,
|
|
disk_write_count_record,
|
|
disk_read_speed_record,
|
|
disk_write_speed_record,
|
|
disk_read_iops_record,
|
|
disk_write_iops_record,
|
|
disk_usage_record,
|
|
disk_free_record,
|
|
disk_utilization_percentage_record,
|
|
network_sent_record,
|
|
network_received_record,
|
|
network_send_speed_record,
|
|
network_receive_speed_record,
|
|
]
|
|
)
|
|
|
|
return records_reported
|
|
|
|
async def _run_loop(self):
|
|
"""Get any changes to the log files and push updates to kv."""
|
|
loop = get_or_create_event_loop()
|
|
|
|
while True:
|
|
try:
|
|
# Fetch autoscaler debug status
|
|
autoscaler_status_json_bytes: Optional[bytes] = None
|
|
autoscaler_v2_enabled = False
|
|
if self._is_head_node:
|
|
# Check autoscaler version once
|
|
autoscaler_v2_enabled = is_autoscaler_v2(
|
|
gcs_client=self._gcs_client
|
|
)
|
|
|
|
# Autoscaler v1 writes DEBUG_AUTOSCALING_STATUS to the internal KV; v2 does not.
|
|
if not autoscaler_v2_enabled:
|
|
autoscaler_status_json_bytes = (
|
|
await self._gcs_client.async_internal_kv_get(
|
|
DEBUG_AUTOSCALING_STATUS.encode(),
|
|
None,
|
|
timeout=GCS_RPC_TIMEOUT_SECONDS,
|
|
)
|
|
)
|
|
|
|
self._gcs_pid = await self._gcs_client.async_internal_kv_get(
|
|
GCS_PID_KEY.encode(),
|
|
None,
|
|
timeout=GCS_RPC_TIMEOUT_SECONDS,
|
|
)
|
|
self._gcs_pid = (
|
|
int(self._gcs_pid.decode()) if self._gcs_pid else None
|
|
)
|
|
|
|
# NOTE: Stats collection is executed inside the thread-pool
|
|
# executor (TPE) to avoid blocking the Agent's event-loop
|
|
json_payload = await loop.run_in_executor(
|
|
self._executor,
|
|
self._run_in_executor,
|
|
autoscaler_status_json_bytes,
|
|
autoscaler_v2_enabled,
|
|
)
|
|
|
|
await self._gcs_client.async_publish_node_resource_usage(
|
|
self._key, json_payload
|
|
)
|
|
|
|
except Exception:
|
|
logger.exception("Error publishing node physical stats.")
|
|
|
|
await asyncio.sleep(reporter_consts.REPORTER_UPDATE_INTERVAL_MS / 1000)
|
|
|
|
def _run_in_executor(
|
|
self,
|
|
cluster_autoscaling_stats_json: Optional[bytes],
|
|
autoscaler_v2_enabled: bool,
|
|
) -> str:
|
|
return asyncio.run(
|
|
self._async_compose_stats_payload(
|
|
cluster_autoscaling_stats_json,
|
|
autoscaler_v2_enabled,
|
|
)
|
|
)
|
|
|
|
async def _async_compose_stats_payload(
|
|
self,
|
|
cluster_autoscaling_stats_json: Optional[bytes],
|
|
autoscaler_v2_enabled: bool,
|
|
) -> str:
|
|
stats = await self._async_collect_stats()
|
|
|
|
# Report stats only when metrics collection is enabled.
|
|
if not self._metrics_collection_disabled:
|
|
cluster_stats = (
|
|
json.loads(cluster_autoscaling_stats_json.decode())
|
|
if cluster_autoscaling_stats_json
|
|
else {}
|
|
)
|
|
|
|
# Autoscaler v2 only - get cluster_status from gcs via RPC(get_cluster_status())
|
|
if self._is_head_node and autoscaler_v2_enabled:
|
|
cluster_stats = self._get_cluster_stats_v2()
|
|
|
|
records = self._to_records(stats, cluster_stats)
|
|
|
|
if RAY_ENABLE_OPEN_TELEMETRY:
|
|
self._open_telemetry_metric_recorder.record_and_export(
|
|
records,
|
|
global_tags={
|
|
"Version": ray.__version__,
|
|
"SessionName": self._session_name,
|
|
},
|
|
)
|
|
else:
|
|
self._metrics_agent.record_and_export(
|
|
records,
|
|
global_tags={
|
|
"Version": ray.__version__,
|
|
"SessionName": self._session_name,
|
|
},
|
|
)
|
|
|
|
self._metrics_agent.clean_all_dead_worker_metrics()
|
|
|
|
return self._generate_stats_payload(stats)
|
|
|
|
def _get_cluster_stats_v2(self) -> dict:
|
|
# Get cluster_status from gcs via RPC(get_cluster_status())
|
|
cluster_status = get_cluster_status(self.gcs_address)
|
|
|
|
# Aggregate node counts by ray_node_type_name
|
|
active_nodes = _count_by(cluster_status.active_nodes, "ray_node_type_name")
|
|
idle_nodes = _count_by(cluster_status.idle_nodes, "ray_node_type_name")
|
|
|
|
# Keep tuple schemas expected by _to_records()
|
|
pending_nodes = [
|
|
(n.ip_address, n.ray_node_type_name, str(n.details or "PENDING"))
|
|
for n in cluster_status.pending_nodes
|
|
]
|
|
failed_nodes = [
|
|
(n.ip_address, n.ray_node_type_name) for n in cluster_status.failed_nodes
|
|
]
|
|
|
|
return {
|
|
"autoscaler_report": {
|
|
"active_nodes": dict(active_nodes),
|
|
"idle_nodes": dict(idle_nodes),
|
|
"pending_nodes": pending_nodes,
|
|
"failed_nodes": failed_nodes,
|
|
},
|
|
}
|
|
|
|
def _generate_stats_payload(self, stats: dict) -> str:
|
|
# Convert processes_pids back to a list of dictionaries to maintain backwards-compatibility
|
|
for gpu in stats["gpus"]:
|
|
if isinstance(gpu.get("processes_pids"), dict):
|
|
gpu["processes_pids"] = list(gpu["processes_pids"].values())
|
|
|
|
if StatsPayload is not None:
|
|
stats_dict = dashboard_utils.to_google_style(recursive_asdict(stats))
|
|
|
|
parsed_stats = StatsPayload.parse_obj(stats_dict)
|
|
out = json.dumps(parsed_stats.dict())
|
|
return out
|
|
else:
|
|
# NOTE: This converts keys to "Google style", (e.g: "processes_pids" -> "processesPids")
|
|
return jsonify_asdict(stats)
|
|
|
|
async def run(self, server):
|
|
if server:
|
|
reporter_pb2_grpc.add_ReporterServiceServicer_to_server(self, server)
|
|
if RAY_ENABLE_OPEN_TELEMETRY:
|
|
metrics_service_pb2_grpc.add_MetricsServiceServicer_to_server(
|
|
self, server
|
|
)
|
|
|
|
# Initialize GPU metric provider when the agent starts
|
|
self._gpu_metric_provider.initialize()
|
|
|
|
await self._run_loop()
|
|
|
|
@staticmethod
|
|
def is_minimal_module():
|
|
return False
|