diff --git a/docs/monitor/quick_start.md b/docs/monitor/quick_start.md index d4359755..0dc28b54 100644 --- a/docs/monitor/quick_start.md +++ b/docs/monitor/quick_start.md @@ -131,6 +131,8 @@ At startup, RL-Insight copies them into the runtime dashboards directory and pro If you add or update a dashboard JSON file such as `quick_start_demo.json`, place it in the bundled dashboards directory before starting Grafana, or restart the stack so RL-Insight copies the latest file into the runtime directory and Grafana provisions it. Prometheus metrics and Tempo traces are persisted under `~/.rl-insight/data` by default. Stopping the server does not delete collected data. +Prometheus scrape targets registered by trainers or `rl-insight server targets add` are stored separately in `~/.rl-insight/data/targets/prometheus-targets.yml`. The generated `prometheus.yml` references this persistent file through Prometheus file-based service discovery, so restarting the server stack does not clear registered targets. Target updates are written atomically under a cross-process lock. Existing registration paths continue to reload Prometheus for API compatibility, while file-based service discovery also refreshes the target file every five seconds. + ## 6. Stop Services Foreground mode: @@ -182,5 +184,3 @@ rl-insight server install If metrics do not appear, check that the monitor hub process is reachable from Prometheus and that the Prometheus configuration points to the hub `/metrics` endpoint. - - diff --git a/rl_insight/server/runtime.py b/rl_insight/server/runtime.py index 89fe9a85..1cfab558 100644 --- a/rl_insight/server/runtime.py +++ b/rl_insight/server/runtime.py @@ -18,13 +18,16 @@ import configparser import datetime as _dt +import fcntl import json import os +import re import shutil import signal import subprocess import sys import time +from contextlib import contextmanager from dataclasses import dataclass from pathlib import Path from typing import Any, Sequence @@ -34,6 +37,10 @@ from .catalog import DEFAULT_STATE_ROOT, STATE_FILE from .network import format_host_port, local_addresses +from ..utils.constants import ( + PrometheusScrape, + prometheus_targets_file_from_config, +) from .dependencies import ( MissingDependencyError, DependencyManager, @@ -378,12 +385,251 @@ def _service_data_root(conf: DictConfig, _install_root: Path) -> Path: def _render_prometheus_config(conf: DictConfig, runtime_dir: Path) -> Path: target = runtime_dir / "prometheus.yml" + targets_file = prometheus_targets_file_from_config(conf) + legacy_targets_file = (runtime_dir / PrometheusScrape.TARGETS_FILE_NAME).resolve() source = Path(str(OmegaConf.select(conf, "prometheus.config_file"))) data = yaml.safe_load(source.read_text(encoding="utf-8")) or {} - target.write_text(yaml.safe_dump(data, sort_keys=False), encoding="utf-8") + scrape_configs = data.get("scrape_configs") or [] + if not isinstance(scrape_configs, list): + raise ValueError("Prometheus scrape_configs must be a list") + + with _prometheus_targets_lock(targets_file): + if not targets_file.exists(): + migrated_targets = ( + _read_file_sd_targets(legacy_targets_file) + if legacy_targets_file.exists() + else _migrate_prometheus_static_targets( + target, scrape_configs, targets_file + ) + ) + _write_yaml_atomically(targets_file, migrated_targets) + + scrape_configs = [ + item + for item in scrape_configs + if not _is_managed_prometheus_job(item, targets_file) + ] + source_job_names: list[str] = [] + managed_jobs: list[dict[str, Any]] = [] + for source_job in scrape_configs: + if not isinstance(source_job, dict): + continue + source_job_name = str(source_job.get("job_name") or "").strip() + if not source_job_name: + continue + source_job_names.append(source_job_name) + profile_name = _available_prometheus_job_name([*scrape_configs, *managed_jobs]) + managed_jobs.append( + _profiled_prometheus_job( + profile_name, source_job_name, source_job, targets_file + ) + ) + + dynamic_job_name = _available_prometheus_job_name([*scrape_configs, *managed_jobs]) + managed_jobs.append( + _managed_prometheus_job( + dynamic_job_name, targets_file, excluded_jobs=source_job_names + ) + ) + scrape_configs.extend(managed_jobs) + data["scrape_configs"] = scrape_configs + _write_yaml_atomically(target, data) return target +def _managed_prometheus_job( + job_name: str, + targets_file: Path, + *, + excluded_jobs: Sequence[str] = (), +) -> dict[str, Any]: + relabel_configs: list[dict[str, Any]] = [] + if excluded_jobs: + relabel_configs.append( + { + "source_labels": [PrometheusScrape.DYNAMIC_JOB_LABEL], + "regex": "|".join(re.escape(name) for name in excluded_jobs), + "action": "drop", + } + ) + relabel_configs.extend( + [ + { + "source_labels": [PrometheusScrape.DYNAMIC_JOB_LABEL], + "target_label": "job", + }, + { + "regex": PrometheusScrape.DYNAMIC_JOB_LABEL, + "action": "labeldrop", + }, + ] + ) + return { + "job_name": job_name, + "file_sd_configs": [ + { + "files": [str(targets_file)], + "refresh_interval": PrometheusScrape.TARGETS_REFRESH_INTERVAL, + } + ], + "relabel_configs": relabel_configs, + } + + +def _profiled_prometheus_job( + profile_name: str, + source_job_name: str, + source_job: dict[str, Any], + targets_file: Path, +) -> dict[str, Any]: + discovery_keys = { + key + for key in source_job + if key == "static_configs" or key.endswith("_sd_configs") + } + profile = { + "job_name": profile_name, + **{ + key: value + for key, value in source_job.items() + if key not in discovery_keys and key not in {"job_name", "relabel_configs"} + }, + "file_sd_configs": [ + { + "files": [str(targets_file)], + "refresh_interval": PrometheusScrape.TARGETS_REFRESH_INTERVAL, + } + ], + "relabel_configs": [ + { + "source_labels": [PrometheusScrape.DYNAMIC_JOB_LABEL], + "regex": re.escape(source_job_name), + "action": "keep", + }, + { + "source_labels": [PrometheusScrape.DYNAMIC_JOB_LABEL], + "target_label": "job", + }, + { + "regex": PrometheusScrape.DYNAMIC_JOB_LABEL, + "action": "labeldrop", + }, + *(source_job.get("relabel_configs") or []), + ], + } + return profile + + +def _is_managed_prometheus_job(item: Any, targets_file: Path) -> bool: + if not isinstance(item, dict): + return False + name = str(item.get("job_name") or "") + if not name.startswith(PrometheusScrape.DYNAMIC_CONFIG_JOB): + return False + file_sd_configs = item.get("file_sd_configs") or [] + if not any( + isinstance(config, dict) and config.get("files") == [str(targets_file)] + for config in file_sd_configs + ): + return False + return any( + isinstance(config, dict) + and config.get("regex") == PrometheusScrape.DYNAMIC_JOB_LABEL + and config.get("action") == "labeldrop" + for config in item.get("relabel_configs") or [] + ) + + +def _available_prometheus_job_name(scrape_configs: list[Any]) -> str: + used_names = { + str(item.get("job_name")) + for item in scrape_configs + if isinstance(item, dict) and item.get("job_name") is not None + } + base = PrometheusScrape.DYNAMIC_CONFIG_JOB + if base not in used_names: + return base + suffix = 1 + while f"{base}-{suffix}" in used_names: + suffix += 1 + return f"{base}-{suffix}" + + +def _write_yaml_atomically(path: Path, data: Any) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + tmp_path = path.with_name(f".{path.name}.{os.getpid()}.tmp") + try: + tmp_path.write_text(yaml.safe_dump(data, sort_keys=False), encoding="utf-8") + os.replace(tmp_path, path) + except BaseException: + try: + tmp_path.unlink() + except FileNotFoundError: + pass + raise + + +@contextmanager +def _prometheus_targets_lock(path: Path): + path.parent.mkdir(parents=True, exist_ok=True) + lock_file = path.with_name(f".{path.name}.lock") + with lock_file.open("a+", encoding="utf-8") as handle: + fcntl.flock(handle.fileno(), fcntl.LOCK_EX) + try: + yield + finally: + fcntl.flock(handle.fileno(), fcntl.LOCK_UN) + + +def _read_file_sd_targets(path: Path) -> list[dict[str, Any]]: + data = yaml.safe_load(path.read_text(encoding="utf-8")) or [] + if not isinstance(data, list): + raise ValueError("Prometheus file_sd targets must be a list") + return data + + +def _migrate_prometheus_static_targets( + config_file: Path, + source_scrape_configs: list[Any], + targets_file: Path, +) -> list[dict[str, Any]]: + if not config_file.exists(): + return [] + legacy = yaml.safe_load(config_file.read_text(encoding="utf-8")) or {} + source_targets = { + (str(job.get("job_name") or ""), str(target)) + for job in source_scrape_configs + if isinstance(job, dict) + for static_group in job.get("static_configs") or [] + if isinstance(static_group, dict) + for target in static_group.get("targets") or [] + } + groups: list[dict[str, Any]] = [] + for job in legacy.get("scrape_configs") or []: + if not isinstance(job, dict): + continue + job_name = str(job.get("job_name") or "").strip() + if not job_name or _is_managed_prometheus_job(job, targets_file): + continue + for static_group in job.get("static_configs") or []: + if not isinstance(static_group, dict): + continue + targets = [ + str(item) + for item in static_group.get("targets") or [] + if (job_name, str(item)) not in source_targets + ] + if not targets: + continue + labels = { + str(key): str(value) + for key, value in (static_group.get("labels") or {}).items() + } + labels[PrometheusScrape.DYNAMIC_JOB_LABEL] = job_name + groups.append({"targets": targets, "labels": labels}) + return groups + + def _render_tempo_config( conf: DictConfig, runtime_dir: Path, data_root: Path, tempo_version: str ) -> Path: diff --git a/rl_insight/utils/constants.py b/rl_insight/utils/constants.py index b699216d..016fcc17 100644 --- a/rl_insight/utils/constants.py +++ b/rl_insight/utils/constants.py @@ -18,6 +18,8 @@ from pathlib import Path +from omegaconf import DictConfig, OmegaConf + _MONITOR_DIR = Path(__file__).resolve().parents[1] @@ -83,3 +85,21 @@ class PrometheusScrape: """Prometheus scrape job names managed by the monitor hub.""" TRAINER_METRICS_JOB = "trainer_metrics" + DYNAMIC_CONFIG_JOB = "rl-insight-dynamic" + DYNAMIC_JOB_LABEL = "rl_insight_job" + TARGETS_FILE_NAME = "prometheus-targets.yml" + TARGETS_REFRESH_INTERVAL = "5s" + + +def prometheus_targets_file_from_config(conf: DictConfig) -> Path: + """Return the persistent file_sd target store configured for the server.""" + # TODO: Partition file_sd targets by experiment once stable experiment + # identity and lifecycle management are available, then have Prometheus + # watch all experiment target files collectively. + raw_data_dir = OmegaConf.select(conf, "server.data_dir") + data_dir = ( + Path(str(raw_data_dir)).expanduser().resolve() + if raw_data_dir + else (MonitorPaths.STATE_ROOT / "data").resolve() + ) + return (data_dir / "targets" / PrometheusScrape.TARGETS_FILE_NAME).resolve() diff --git a/rl_insight/utils/prometheus_utils.py b/rl_insight/utils/prometheus_utils.py index de28747f..9a21de79 100644 --- a/rl_insight/utils/prometheus_utils.py +++ b/rl_insight/utils/prometheus_utils.py @@ -20,6 +20,7 @@ import os import threading from collections.abc import Mapping, Sequence +from contextlib import contextmanager from dataclasses import dataclass, field from pathlib import Path from typing import Any @@ -30,7 +31,12 @@ from prometheus_client import Counter, Gauge, Histogram, start_http_server from ..server.network import format_host_port, local_addresses -from .constants import MonitorEnv, MonitorPaths, PrometheusScrape +from .constants import ( + MonitorEnv, + MonitorPaths, + PrometheusScrape, + prometheus_targets_file_from_config, +) logger = logging.getLogger(__file__) logger.setLevel(logging.WARNING) @@ -54,10 +60,20 @@ class PrometheusTarget: class PrometheusTargetStore: - """Maintain Prometheus scrape targets in the runtime config file.""" + """Maintain Prometheus scrape targets in a file_sd discovery file.""" - def __init__(self, config_file: str | Path, prometheus_port: int): + def __init__( + self, + config_file: str | Path, + prometheus_port: int, + targets_file: str | Path | None = None, + ): self.config_file = Path(config_file).expanduser().resolve() + self.targets_file = ( + Path(targets_file).expanduser().resolve() + if targets_file is not None + else self.config_file.with_name(PrometheusScrape.TARGETS_FILE_NAME) + ) self.prometheus_port = prometheus_port self._lock = threading.Lock() @@ -70,7 +86,11 @@ def from_config(cls, conf: DictConfig) -> "PrometheusTargetStore": else (MonitorPaths.STATE_ROOT / "runtime").resolve() ) prometheus_port = int(OmegaConf.select(conf, "prometheus.prometheus_port")) - return cls(base / "prometheus.yml", prometheus_port) + return cls( + base / "prometheus.yml", + prometheus_port, + prometheus_targets_file_from_config(conf), + ) def register( self, job_name: str, targets: Sequence[PrometheusTarget] @@ -80,52 +100,94 @@ def register( for item in targets } - with self._lock: - source = ( - self.config_file - if self.config_file.exists() - else MonitorPaths.PROMETHEUS_CONFIG_FILE - ) - data = yaml.safe_load(source.read_text(encoding="utf-8")) or {} - scrape_configs = data.setdefault("scrape_configs", []) - - job_config = next( - ( - config - for config in scrape_configs - if config.get("job_name") == job_name - ), - None, - ) - if job_config is None: - job_config = {"job_name": job_name} - scrape_configs.append(job_config) - - target_map = { - target: group.get("labels", {}) - for group in job_config.get("static_configs", []) - for target in group.get("targets", []) - } - target_map.update(incoming) - job_config["static_configs"] = [ - {"targets": [target], **({"labels": labels} if labels else {})} - for target, labels in sorted(target_map.items()) - ] - - self.config_file.parent.mkdir(parents=True, exist_ok=True) - payload = yaml.safe_dump(data, sort_keys=False) - tmp_path = self.config_file.with_name( - f".{self.config_file.name}.{os.getpid()}.tmp" + with self._lock, self._file_lock(): + target_map = self._read_targets() + target_map.update( + {(str(job_name), target): labels for target, labels in incoming.items()} ) - tmp_path.write_text(payload, encoding="utf-8") - os.replace(tmp_path, self.config_file) + self._write_targets(target_map) return { "job_name": job_name, - "target_count": len(target_map), + "target_count": sum( + 1 for stored_job, _target in target_map if stored_job == job_name + ), "config_file": str(self.config_file), + "targets_file": str(self.targets_file), } + @contextmanager + def _file_lock(self): + """Serialize read-modify-write updates made by different processes.""" + try: + import fcntl + except ImportError as exc: + raise RuntimeError( + "Prometheus target persistence requires a POSIX server" + ) from exc + + self.targets_file.parent.mkdir(parents=True, exist_ok=True) + lock_file = self.targets_file.with_name(f".{self.targets_file.name}.lock") + with lock_file.open("a+", encoding="utf-8") as handle: + fcntl.flock(handle.fileno(), fcntl.LOCK_EX) + try: + yield + finally: + fcntl.flock(handle.fileno(), fcntl.LOCK_UN) + + def _write_targets( + self, target_map: Mapping[tuple[str, str], Mapping[str, str]] + ) -> None: + groups = [ + { + "targets": [target], + "labels": { + **labels, + PrometheusScrape.DYNAMIC_JOB_LABEL: stored_job, + }, + } + for (stored_job, target), labels in sorted(target_map.items()) + ] + payload = yaml.safe_dump(groups, sort_keys=False) + tmp_path = self.targets_file.with_name( + f".{self.targets_file.name}.{os.getpid()}.{threading.get_ident()}.tmp" + ) + try: + tmp_path.write_text(payload, encoding="utf-8") + os.replace(tmp_path, self.targets_file) + except BaseException: + try: + tmp_path.unlink() + except FileNotFoundError: + pass + raise + + def _read_targets(self) -> dict[tuple[str, str], dict[str, str]]: + if not self.targets_file.exists(): + return {} + groups = yaml.safe_load(self.targets_file.read_text(encoding="utf-8")) or [] + if not isinstance(groups, list): + raise ValueError("Prometheus file_sd targets must be a list") + + target_map: dict[tuple[str, str], dict[str, str]] = {} + for group in groups: + if not isinstance(group, Mapping): + raise ValueError( + "Each Prometheus file_sd target group must be an object" + ) + raw_labels = group.get("labels") or {} + if not isinstance(raw_labels, Mapping): + raise ValueError("Prometheus file_sd target labels must be an object") + labels = {str(key): str(value) for key, value in raw_labels.items()} + stored_job = labels.pop(PrometheusScrape.DYNAMIC_JOB_LABEL, "").strip() + if not stored_job: + raise ValueError( + "Prometheus file_sd target group is missing managed job label" + ) + for target in group.get("targets") or []: + target_map[(stored_job, str(target))] = dict(labels) + return target_map + def reload(self) -> bool: url = ( "http://" @@ -298,9 +360,9 @@ def update_prometheus_config( ) -> None: """Register trainer metrics endpoints with the RL-Insight server. - The RL-Insight server writes these targets into the runtime Prometheus - config and reloads the managed Prometheus process. ``server_addresses`` - should contain scrape targets in ``host:port`` or ``[ipv6]:port`` form, not full URLs. + The RL-Insight server writes these targets into its file_sd discovery file; + Prometheus observes the update automatically. ``server_addresses`` should + contain scrape targets in ``host:port`` or ``[ipv6]:port`` form, not full URLs. Args: server_addresses: Prometheus scrape targets exposed by trainer-side diff --git a/tests/monitor/ut/test_server_http_api.py b/tests/monitor/ut/test_server_http_api.py new file mode 100644 index 00000000..794f4014 --- /dev/null +++ b/tests/monitor/ut/test_server_http_api.py @@ -0,0 +1,60 @@ +# Copyright (c) 2026 verl-project authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Unit tests for the RL-Insight server HTTP API.""" + +from __future__ import annotations + +from typing import Any, cast +from unittest.mock import MagicMock + +from omegaconf import OmegaConf + +from rl_insight.server.http_api import create_app +from rl_insight.utils.prometheus_utils import PrometheusTargetStore + + +def test_register_targets_should_persist_file_sd_targets_and_reload_prometheus( + monkeypatch, tmp_path +) -> None: + conf = OmegaConf.create( + { + "server": { + "runtime_dir": str(tmp_path / "runtime"), + "data_dir": str(tmp_path / "data"), + }, + "prometheus": {"prometheus_port": 9090}, + } + ) + + reload_prometheus = MagicMock(return_value=True) + monkeypatch.setattr(PrometheusTargetStore, "reload", reload_prometheus) + app = create_app(conf) + endpoint = next( + cast(Any, route).endpoint + for route in app.routes + if getattr(route, "path", "") == "/api/v1/prometheus/targets" + ) + + result = endpoint( + { + "job_name": "node-exporter", + "targets": ["node-a:9100"], + } + ) + + assert result["status"] == "ok" + assert result["prometheus_reloaded"] is True + reload_prometheus.assert_called_once_with() + assert (tmp_path / "data" / "targets" / "prometheus-targets.yml").exists() diff --git a/tests/monitor/ut/test_server_runtime.py b/tests/monitor/ut/test_server_runtime.py new file mode 100644 index 00000000..ad3e0096 --- /dev/null +++ b/tests/monitor/ut/test_server_runtime.py @@ -0,0 +1,313 @@ +# Copyright (c) 2026 verl-project authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Unit tests for rendering local server runtime files.""" + +from __future__ import annotations + +import os + +import pytest +import yaml +from omegaconf import OmegaConf + +from rl_insight.server import runtime as runtime_module + + +_render_prometheus_config = runtime_module._render_prometheus_config + + +def _targets_file(tmp_path): + return tmp_path / "data" / "targets" / "prometheus-targets.yml" + + +def _prometheus_conf(source, tmp_path): + return OmegaConf.create( + { + "server": {"data_dir": str(tmp_path / "data")}, + "prometheus": {"config_file": str(source)}, + } + ) + + +def test_render_prometheus_config_should_reference_and_preserve_file_sd_targets( + tmp_path, +) -> None: + source = tmp_path / "source-prometheus.yml" + source.write_text( + yaml.safe_dump( + { + "global": {"scrape_interval": "10s"}, + "scrape_configs": [], + }, + sort_keys=False, + ), + encoding="utf-8", + ) + runtime_dir = tmp_path / "runtime" + runtime_dir.mkdir() + targets_file = _targets_file(tmp_path) + targets_file.parent.mkdir(parents=True) + existing_targets = [ + { + "targets": ["node-a:9100"], + "labels": {"rl_insight_job": "node-exporter"}, + } + ] + targets_file.write_text(yaml.safe_dump(existing_targets), encoding="utf-8") + conf = _prometheus_conf(source, tmp_path) + + rendered_path = _render_prometheus_config(conf, runtime_dir) + rendered = yaml.safe_load(rendered_path.read_text(encoding="utf-8")) + + assert yaml.safe_load(targets_file.read_text(encoding="utf-8")) == existing_targets + assert rendered["scrape_configs"] == [ + { + "job_name": "rl-insight-dynamic", + "file_sd_configs": [ + { + "files": [str(targets_file.resolve())], + "refresh_interval": "5s", + } + ], + "relabel_configs": [ + { + "source_labels": ["rl_insight_job"], + "target_label": "job", + }, + {"regex": "rl_insight_job", "action": "labeldrop"}, + ], + } + ] + + +def test_render_prometheus_config_should_migrate_existing_static_targets( + tmp_path, +) -> None: + source = tmp_path / "source-prometheus.yml" + source.write_text( + yaml.safe_dump({"global": {"scrape_interval": "10s"}, "scrape_configs": []}), + encoding="utf-8", + ) + runtime_dir = tmp_path / "runtime" + runtime_dir.mkdir() + runtime_config = runtime_dir / "prometheus.yml" + runtime_config.write_text( + yaml.safe_dump( + { + "scrape_configs": [ + { + "job_name": "node-exporter", + "static_configs": [ + { + "targets": ["node-a:9100"], + "labels": {"node": "node-a"}, + } + ], + } + ] + } + ), + encoding="utf-8", + ) + conf = _prometheus_conf(source, tmp_path) + + _render_prometheus_config(conf, runtime_dir) + + targets_file = _targets_file(tmp_path) + assert yaml.safe_load(targets_file.read_text(encoding="utf-8")) == [ + { + "targets": ["node-a:9100"], + "labels": { + "rl_insight_job": "node-exporter", + "node": "node-a", + }, + } + ] + + +def test_render_prometheus_config_should_move_legacy_runtime_targets_to_data( + tmp_path, +) -> None: + source = tmp_path / "source-prometheus.yml" + source.write_text("scrape_configs: []\n", encoding="utf-8") + runtime_dir = tmp_path / "runtime" + runtime_dir.mkdir() + legacy_targets = [ + { + "targets": ["trainer:9092"], + "labels": {"rl_insight_job": "trainer_metrics"}, + } + ] + (runtime_dir / "prometheus-targets.yml").write_text( + yaml.safe_dump(legacy_targets), encoding="utf-8" + ) + + _render_prometheus_config(_prometheus_conf(source, tmp_path), runtime_dir) + + assert ( + yaml.safe_load(_targets_file(tmp_path).read_text(encoding="utf-8")) + == legacy_targets + ) + + +def test_render_prometheus_config_should_migrate_only_runtime_added_targets( + tmp_path, +) -> None: + source = tmp_path / "source-prometheus.yml" + source_job = { + "job_name": "node-exporter", + "metrics_path": "/custom-metrics", + "static_configs": [{"targets": ["node-a:9100"]}], + } + source.write_text( + yaml.safe_dump({"scrape_configs": [source_job]}, sort_keys=False), + encoding="utf-8", + ) + runtime_dir = tmp_path / "runtime" + runtime_dir.mkdir() + (runtime_dir / "prometheus.yml").write_text( + yaml.safe_dump( + { + "scrape_configs": [ + { + **source_job, + "static_configs": [{"targets": ["node-a:9100", "node-b:9100"]}], + } + ] + }, + sort_keys=False, + ), + encoding="utf-8", + ) + conf = _prometheus_conf(source, tmp_path) + + rendered_path = _render_prometheus_config(conf, runtime_dir) + + rendered = yaml.safe_load(rendered_path.read_text(encoding="utf-8")) + assert rendered["scrape_configs"][0] == source_job + profile_job = rendered["scrape_configs"][1] + assert profile_job["metrics_path"] == "/custom-metrics" + assert profile_job["file_sd_configs"] == [ + { + "files": [str(_targets_file(tmp_path).resolve())], + "refresh_interval": "5s", + } + ] + assert profile_job["relabel_configs"][:3] == [ + { + "source_labels": ["rl_insight_job"], + "regex": "node\\-exporter", + "action": "keep", + }, + {"source_labels": ["rl_insight_job"], "target_label": "job"}, + {"regex": "rl_insight_job", "action": "labeldrop"}, + ] + assert yaml.safe_load(_targets_file(tmp_path).read_text(encoding="utf-8")) == [ + { + "targets": ["node-b:9100"], + "labels": {"rl_insight_job": "node-exporter"}, + } + ] + + +def test_render_prometheus_config_should_preserve_colliding_custom_job( + tmp_path, +) -> None: + source = tmp_path / "source-prometheus.yml" + custom_job = { + "job_name": "rl-insight-dynamic", + "static_configs": [{"targets": ["custom:9090"]}], + } + source.write_text( + yaml.safe_dump({"scrape_configs": [custom_job]}, sort_keys=False), + encoding="utf-8", + ) + runtime_dir = tmp_path / "runtime" + runtime_dir.mkdir() + conf = _prometheus_conf(source, tmp_path) + + rendered_path = _render_prometheus_config(conf, runtime_dir) + + jobs = yaml.safe_load(rendered_path.read_text(encoding="utf-8"))["scrape_configs"] + assert jobs[0] == custom_job + assert [job["job_name"] for job in jobs] == [ + "rl-insight-dynamic", + "rl-insight-dynamic-1", + "rl-insight-dynamic-2", + ] + + +def test_render_prometheus_config_should_migrate_targets_from_colliding_job( + tmp_path, +) -> None: + source = tmp_path / "source-prometheus.yml" + source_job = { + "job_name": "rl-insight-dynamic", + "static_configs": [{"targets": ["node-a:9100"]}], + } + source.write_text( + yaml.safe_dump({"scrape_configs": [source_job]}, sort_keys=False), + encoding="utf-8", + ) + runtime_dir = tmp_path / "runtime" + runtime_dir.mkdir() + (runtime_dir / "prometheus.yml").write_text( + yaml.safe_dump( + { + "scrape_configs": [ + { + **source_job, + "static_configs": [{"targets": ["node-a:9100", "node-b:9100"]}], + } + ] + }, + sort_keys=False, + ), + encoding="utf-8", + ) + conf = _prometheus_conf(source, tmp_path) + + _render_prometheus_config(conf, runtime_dir) + + assert yaml.safe_load(_targets_file(tmp_path).read_text(encoding="utf-8")) == [ + { + "targets": ["node-b:9100"], + "labels": {"rl_insight_job": "rl-insight-dynamic"}, + } + ] + + +def test_render_prometheus_config_should_not_leave_partial_targets_on_write_failure( + monkeypatch, tmp_path +) -> None: + source = tmp_path / "source-prometheus.yml" + source.write_text("scrape_configs: []\n", encoding="utf-8") + runtime_dir = tmp_path / "runtime" + runtime_dir.mkdir() + targets_file = _targets_file(tmp_path) + real_replace = os.replace + + def fail_targets_replace(source_path, target_path) -> None: + if target_path == targets_file.resolve(): + raise OSError("simulated interrupted migration") + real_replace(source_path, target_path) + + monkeypatch.setattr(runtime_module.os, "replace", fail_targets_replace) + conf = _prometheus_conf(source, tmp_path) + + with pytest.raises(OSError, match="simulated interrupted migration"): + _render_prometheus_config(conf, runtime_dir) + + assert not targets_file.exists() diff --git a/tests/monitor/ut/utils/test_prometheus_utils.py b/tests/monitor/ut/utils/test_prometheus_utils.py index c8760ca3..6df87e7b 100644 --- a/tests/monitor/ut/utils/test_prometheus_utils.py +++ b/tests/monitor/ut/utils/test_prometheus_utils.py @@ -16,6 +16,7 @@ from __future__ import annotations +import multiprocessing from typing import Any from unittest.mock import MagicMock @@ -25,6 +26,19 @@ from rl_insight.utils import prometheus_utils as prometheus_module +def _register_target_in_process( + config_file: str, targets_file: str, start_event, index: int +) -> None: + store = prometheus_module.PrometheusTargetStore( + config_file, 9090, targets_file=targets_file + ) + start_event.wait() + store.register( + "trainers", + [prometheus_module.PrometheusTarget(f"host-{index}:9000")], + ) + + def _samples(collector: Any) -> dict[str, Any]: return { sample.name: sample @@ -55,41 +69,47 @@ def test_metric_registry_should_store_real_samples_when_all_metric_types_are_rec assert histogram["monitor_ut_latency_sum"].value == 12 -def test_register_should_merge_and_sort_targets_when_config_already_exists( +def test_register_should_write_file_sd_targets_without_changing_main_config( tmp_path, ) -> None: config_file = tmp_path / "prometheus.yml" - config_file.write_text( - yaml.safe_dump( - { - "scrape_configs": [ - { - "job_name": "trainers", - "static_configs": [{"targets": ["host-b:9000"]}], - } - ] - } - ), - encoding="utf-8", - ) + original_config = {"global": {"scrape_interval": "10s"}} + config_file.write_text(yaml.safe_dump(original_config), encoding="utf-8") store = prometheus_module.PrometheusTargetStore(config_file, 9090) result = store.register( "trainers", [ prometheus_module.PrometheusTarget("host-a:9000", {"rank": 0}), - prometheus_module.PrometheusTarget("host-b:9000", {"rank": 1}), + prometheus_module.PrometheusTarget("host-b:9000"), ], ) - saved = yaml.safe_load(config_file.read_text(encoding="utf-8")) - groups = saved["scrape_configs"][0]["static_configs"] + targets_file = tmp_path / "prometheus-targets.yml" + saved = yaml.safe_load(targets_file.read_text(encoding="utf-8")) assert result["target_count"] == 2 - assert groups == [ - {"targets": ["host-a:9000"], "labels": {"rank": "0"}}, - {"targets": ["host-b:9000"], "labels": {"rank": "1"}}, + assert yaml.safe_load(config_file.read_text(encoding="utf-8")) == original_config + assert saved == [ + { + "targets": ["host-a:9000"], + "labels": {"rl_insight_job": "trainers", "rank": "0"}, + }, + { + "targets": ["host-b:9000"], + "labels": {"rl_insight_job": "trainers"}, + }, ] + store.register( + "trainers", + [prometheus_module.PrometheusTarget("host-b:9000", {"rank": 1})], + ) + saved = yaml.safe_load(targets_file.read_text(encoding="utf-8")) + assert saved[1] == { + "targets": ["host-b:9000"], + "labels": {"rl_insight_job": "trainers", "rank": "1"}, + } + def test_reload_should_post_to_local_prometheus_when_store_is_configured( monkeypatch: pytest.MonkeyPatch, tmp_path: Any @@ -114,6 +134,32 @@ def test_reload_should_post_to_local_prometheus_when_store_is_configured( response.raise_for_status.assert_called_once_with() +def test_register_should_preserve_concurrent_cross_process_updates(tmp_path) -> None: + config_file = tmp_path / "prometheus.yml" + targets_file = tmp_path / "prometheus-targets.yml" + context = multiprocessing.get_context("fork") + start_event = context.Event() + processes = [ + context.Process( + target=_register_target_in_process, + args=(str(config_file), str(targets_file), start_event, index), + ) + for index in range(8) + ] + + for process in processes: + process.start() + start_event.set() + for process in processes: + process.join(timeout=10) + + assert [process.exitcode for process in processes] == [0] * len(processes) + saved = yaml.safe_load(targets_file.read_text(encoding="utf-8")) + assert {group["targets"][0] for group in saved} == { + f"host-{index}:9000" for index in range(8) + } + + def test_update_prometheus_config_should_send_normalized_targets_when_server_is_set( monkeypatch: pytest.MonkeyPatch, ) -> None: