wehub-resource-sync
293 lines
12 KiB
Python
293 lines
12 KiB
Python
from typing import Optional
|
|
|
|
|
|
class NullMetric:
|
|
"""Mock metric class to be used in case of prometheus_client import error."""
|
|
|
|
def set(self, *args, **kwargs):
|
|
pass
|
|
|
|
def observe(self, *args, **kwargs):
|
|
pass
|
|
|
|
def inc(self, *args, **kwargs):
|
|
pass
|
|
|
|
def labels(self, *args, **kwargs):
|
|
return self
|
|
|
|
def clear(self):
|
|
pass
|
|
|
|
|
|
try:
|
|
|
|
from prometheus_client import CollectorRegistry, Counter, Gauge, Histogram
|
|
|
|
# The metrics in this class should be kept in sync with
|
|
# python/ray/tests/test_metrics_agent.py
|
|
class AutoscalerPrometheusMetrics:
|
|
def __init__(
|
|
self, session_name: str = None, registry: Optional[CollectorRegistry] = None
|
|
):
|
|
self.registry: CollectorRegistry = registry or CollectorRegistry(
|
|
auto_describe=True
|
|
)
|
|
self._session_name = session_name
|
|
# Buckets: 5 seconds, 10 seconds, 20 seconds, 30 seconds,
|
|
# 45 seconds, 1 minute, 1.5 minutes, 2 minutes,
|
|
# 3 minutes, 4 minutes, 5 minutes, 6 minutes,
|
|
# 8 minutes, 10 minutes, 12 minutes, 15 minutes
|
|
# 20 minutes, 25 minutes, 30 minutes
|
|
# used for both worker launch time and worker update time
|
|
histogram_buckets = [
|
|
5,
|
|
10,
|
|
20,
|
|
30,
|
|
45,
|
|
60,
|
|
90,
|
|
120,
|
|
180,
|
|
240,
|
|
300,
|
|
360,
|
|
480,
|
|
600,
|
|
720,
|
|
900,
|
|
1200,
|
|
1500,
|
|
1800,
|
|
]
|
|
# Buckets: .01 seconds to 1000 seconds.
|
|
# Used for autoscaler update time.
|
|
update_time_buckets = [0.01, 0.1, 1, 10, 100, 1000]
|
|
self.worker_create_node_time: Histogram = Histogram(
|
|
"worker_create_node_time_seconds",
|
|
"Worker launch time. This is the time it takes for a call to "
|
|
"a node provider's create_node method to return. Note that "
|
|
"when nodes are launched in batches, the launch time for that "
|
|
"batch will be observed once for *each* node in that batch. "
|
|
"For example, if 8 nodes are launched in 3 minutes, a launch "
|
|
"time of 3 minutes will be observed 8 times.",
|
|
labelnames=("SessionName",),
|
|
unit="seconds",
|
|
namespace="autoscaler",
|
|
registry=self.registry,
|
|
buckets=histogram_buckets,
|
|
).labels(SessionName=session_name)
|
|
self.worker_update_time: Histogram = Histogram(
|
|
"worker_update_time_seconds",
|
|
"Worker update time. This is the time between when an updater "
|
|
"thread begins executing and when it exits successfully. This "
|
|
"metric only observes times for successful updates.",
|
|
labelnames=("SessionName",),
|
|
unit="seconds",
|
|
namespace="autoscaler",
|
|
registry=self.registry,
|
|
buckets=histogram_buckets,
|
|
).labels(SessionName=session_name)
|
|
self.update_time: Histogram = Histogram(
|
|
"update_time",
|
|
"Autoscaler update time. This is the time for an autoscaler "
|
|
"update iteration to complete.",
|
|
labelnames=("SessionName",),
|
|
unit="seconds",
|
|
namespace="autoscaler",
|
|
registry=self.registry,
|
|
buckets=update_time_buckets,
|
|
).labels(SessionName=session_name)
|
|
self.pending_nodes: Gauge = Gauge(
|
|
"pending_nodes",
|
|
"Number of nodes pending to be started.",
|
|
labelnames=(
|
|
"NodeType",
|
|
"SessionName",
|
|
),
|
|
unit="nodes",
|
|
namespace="autoscaler",
|
|
registry=self.registry,
|
|
)
|
|
self.active_nodes: Gauge = Gauge(
|
|
"active_nodes",
|
|
"Number of nodes in the cluster.",
|
|
labelnames=(
|
|
"NodeType",
|
|
"SessionName",
|
|
),
|
|
unit="nodes",
|
|
namespace="autoscaler",
|
|
registry=self.registry,
|
|
)
|
|
self.recently_failed_nodes = Gauge(
|
|
"recently_failed_nodes",
|
|
"The number of recently failed nodes. This count could reset "
|
|
"at undefined times.",
|
|
labelnames=(
|
|
"NodeType",
|
|
"SessionName",
|
|
),
|
|
unit="nodes",
|
|
namespace="autoscaler",
|
|
registry=self.registry,
|
|
)
|
|
self.started_nodes: Counter = Counter(
|
|
"started_nodes",
|
|
"Number of nodes started.",
|
|
labelnames=("SessionName",),
|
|
unit="nodes",
|
|
namespace="autoscaler",
|
|
registry=self.registry,
|
|
).labels(SessionName=session_name)
|
|
self.stopped_nodes: Counter = Counter(
|
|
"stopped_nodes",
|
|
"Number of nodes stopped.",
|
|
labelnames=("SessionName",),
|
|
unit="nodes",
|
|
namespace="autoscaler",
|
|
registry=self.registry,
|
|
).labels(SessionName=session_name)
|
|
self.updating_nodes: Gauge = Gauge(
|
|
"updating_nodes",
|
|
"Number of nodes in the process of updating.",
|
|
labelnames=("SessionName",),
|
|
unit="nodes",
|
|
namespace="autoscaler",
|
|
registry=self.registry,
|
|
).labels(SessionName=session_name)
|
|
self.recovering_nodes: Gauge = Gauge(
|
|
"recovering_nodes",
|
|
"Number of nodes in the process of recovering.",
|
|
labelnames=("SessionName",),
|
|
unit="nodes",
|
|
namespace="autoscaler",
|
|
registry=self.registry,
|
|
).labels(SessionName=session_name)
|
|
self.running_workers: Gauge = Gauge(
|
|
"running_workers",
|
|
"Number of worker nodes running.",
|
|
labelnames=("SessionName",),
|
|
unit="nodes",
|
|
namespace="autoscaler",
|
|
registry=self.registry,
|
|
).labels(SessionName=session_name)
|
|
self.failed_create_nodes: Counter = Counter(
|
|
"failed_create_nodes",
|
|
"Number of nodes that failed to be created due to an "
|
|
"exception in the node provider's create_node method.",
|
|
labelnames=("SessionName",),
|
|
unit="nodes",
|
|
namespace="autoscaler",
|
|
registry=self.registry,
|
|
).labels(SessionName=session_name)
|
|
self.failed_updates: Counter = Counter(
|
|
"failed_updates",
|
|
"Number of failed worker node updates.",
|
|
labelnames=("SessionName",),
|
|
unit="updates",
|
|
namespace="autoscaler",
|
|
registry=self.registry,
|
|
).labels(SessionName=session_name)
|
|
self.successful_updates: Counter = Counter(
|
|
"successful_updates",
|
|
"Number of succesfful worker node updates.",
|
|
labelnames=("SessionName",),
|
|
unit="updates",
|
|
namespace="autoscaler",
|
|
registry=self.registry,
|
|
).labels(SessionName=session_name)
|
|
self.failed_recoveries: Counter = Counter(
|
|
"failed_recoveries",
|
|
"Number of failed node recoveries.",
|
|
labelnames=("SessionName",),
|
|
unit="recoveries",
|
|
namespace="autoscaler",
|
|
registry=self.registry,
|
|
).labels(SessionName=session_name)
|
|
self.successful_recoveries: Counter = Counter(
|
|
"successful_recoveries",
|
|
"Number of successful node recoveries.",
|
|
labelnames=("SessionName",),
|
|
unit="recoveries",
|
|
namespace="autoscaler",
|
|
registry=self.registry,
|
|
).labels(SessionName=session_name)
|
|
self.update_loop_exceptions: Counter = Counter(
|
|
"update_loop_exceptions",
|
|
"Number of exceptions raised in the update loop of the autoscaler.",
|
|
labelnames=("SessionName",),
|
|
unit="exceptions",
|
|
namespace="autoscaler",
|
|
registry=self.registry,
|
|
).labels(SessionName=session_name)
|
|
self.node_launch_exceptions: Counter = Counter(
|
|
"node_launch_exceptions",
|
|
"Number of exceptions raised while launching nodes.",
|
|
labelnames=("SessionName",),
|
|
unit="exceptions",
|
|
namespace="autoscaler",
|
|
registry=self.registry,
|
|
).labels(SessionName=session_name)
|
|
self.reset_exceptions: Counter = Counter(
|
|
"reset_exceptions",
|
|
"Number of exceptions raised while resetting the autoscaler.",
|
|
labelnames=("SessionName",),
|
|
unit="exceptions",
|
|
namespace="autoscaler",
|
|
registry=self.registry,
|
|
).labels(SessionName=session_name)
|
|
self.config_validation_exceptions: Counter = Counter(
|
|
"config_validation_exceptions",
|
|
"Number of exceptions raised while validating the config "
|
|
"during a reset.",
|
|
labelnames=("SessionName",),
|
|
unit="exceptions",
|
|
namespace="autoscaler",
|
|
registry=self.registry,
|
|
).labels(SessionName=session_name)
|
|
self.drain_node_exceptions: Counter = Counter(
|
|
"drain_node_exceptions",
|
|
"Number of exceptions raised when making a DrainNode rpc"
|
|
"prior to node termination.",
|
|
labelnames=("SessionName",),
|
|
unit="exceptions",
|
|
namespace="autoscaler",
|
|
registry=self.registry,
|
|
).labels(SessionName=session_name)
|
|
# This represents the autoscaler's view of essentially
|
|
# `ray.cluster_resources()`, it may be slightly different from the
|
|
# core metric from an eventual consistency perspective.
|
|
self.cluster_resources: Gauge = Gauge(
|
|
"cluster_resources",
|
|
"Total logical resources in the cluster.",
|
|
labelnames=("resource", "SessionName"),
|
|
unit="resources",
|
|
namespace="autoscaler",
|
|
registry=self.registry,
|
|
)
|
|
# This represents the pending launches + nodes being set up for the
|
|
# autoscaler.
|
|
self.pending_resources: Gauge = Gauge(
|
|
"pending_resources",
|
|
"Pending logical resources in the cluster.",
|
|
labelnames=("resource", "SessionName"),
|
|
unit="resources",
|
|
namespace="autoscaler",
|
|
registry=self.registry,
|
|
)
|
|
|
|
@property
|
|
def session_name(self):
|
|
return self._session_name
|
|
|
|
except ImportError:
|
|
|
|
class AutoscalerPrometheusMetrics(object):
|
|
def __init__(self, session_name: str = None):
|
|
pass
|
|
|
|
def __getattr__(self, attr):
|
|
return NullMetric()
|