497 lines
26 KiB
Python
497 lines
26 KiB
Python
import contextvars
|
|
import json
|
|
import logging
|
|
import math
|
|
import os
|
|
import re
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import uuid
|
|
from collections import Counter
|
|
from contextlib import contextmanager
|
|
from datetime import datetime, timezone
|
|
from logging.handlers import RotatingFileHandler
|
|
from pathlib import Path
|
|
|
|
from .config import TelemetryConfig, TelemetryError
|
|
from .inspection import assess, scan_workspace, scheduling, count_cases
|
|
from .safety import EVENTS, OPERATIONS, STATUSES, diagnostic, safe_code, detail, hashed, error_code
|
|
|
|
_context = contextvars.ContextVar("vba_telemetry_context", default=None)
|
|
|
|
|
|
class _GatewayFilter(logging.Filter):
|
|
def __init__(self, service_id):
|
|
super().__init__()
|
|
self.service_id = service_id
|
|
|
|
def filter(self, record):
|
|
# Replace legacy provider-name messages only inside this observed task.
|
|
return (_context.get() or {}).get("service_id") != self.service_id
|
|
|
|
|
|
def identity(value):
|
|
import hashlib
|
|
return hashlib.sha256(value.encode(errors="surrogatepass")).hexdigest() if isinstance(value, str) else None
|
|
|
|
|
|
class _Handler(RotatingFileHandler):
|
|
def handleError(self, record):
|
|
raise TelemetryError("TELEMETRY_STORAGE")
|
|
|
|
|
|
class _Scope:
|
|
def __init__(self, telemetry):
|
|
self.telemetry = telemetry
|
|
self.status = "SUCCESS"
|
|
self.code = None
|
|
self.reused = False
|
|
self.failed = False
|
|
self.detail = {}
|
|
|
|
def result(self, result):
|
|
status = getattr(result, "status", "SUCCESS")
|
|
self.status = status if status in STATUSES else "SUCCESS"
|
|
code = getattr(result, "code", None)
|
|
self.code = safe_code(code) if code else None
|
|
self.reused = getattr(result, "reused", False) is True
|
|
if hasattr(result, "source"):
|
|
self.detail["source_key"] = identity(Path(result.source).name)
|
|
if self.status == "QUARANTINED" and getattr(result, "package", None) is not None:
|
|
self.detail["quarantine_key"] = Path(result.package).parent.name
|
|
self.telemetry.guard(self.telemetry.result_metrics, result)
|
|
if isinstance(result, list) and any(getattr(item, "status", None) == "ERROR" for item in result):
|
|
self.status = "ERROR"
|
|
|
|
|
|
class _PhaseLogger:
|
|
def __init__(self, telemetry):
|
|
self.telemetry = telemetry
|
|
|
|
def _log(self, level, message, args):
|
|
if not self.telemetry.started:
|
|
from viano_bug_analysis.llm.logging import case_logger
|
|
getattr(case_logger(self.telemetry.workspace), level.lower())(message, *args)
|
|
return
|
|
index = message.split("code=%s")[0].count("%s") if isinstance(message, str) and "code=%s" in message else None
|
|
code = safe_code(args[index]) if index is not None and index < len(args) else None
|
|
if code is None and isinstance(message, str) and "no reachable LLM server" in message:
|
|
code = "LLM_UNAVAILABLE"
|
|
self.telemetry.emit("phase_diagnostic", level=level, code=code)
|
|
|
|
def info(self, message, *args):
|
|
self._log("INFO", message, args)
|
|
|
|
def warning(self, message, *args):
|
|
self._log("WARNING", message, args)
|
|
|
|
def error(self, message, *args):
|
|
self._log("ERROR", message, args)
|
|
|
|
|
|
class Telemetry:
|
|
"""Single-writer observation session. Never owns case state or recovery policy."""
|
|
def __init__(self, workspace, config=None, *, package_limits=None, clock=None, monotonic=None):
|
|
self.workspace = workspace
|
|
self.config = config or TelemetryConfig()
|
|
self.package_limits = package_limits
|
|
self.clock = clock or (lambda: datetime.now(timezone.utc))
|
|
self.monotonic = monotonic or time.monotonic
|
|
self.directory = workspace.runtime / "telemetry"
|
|
self.started = False
|
|
self.service_id = uuid.uuid4().hex
|
|
self._lock = threading.RLock()
|
|
self._stop = threading.Event()
|
|
self._thread = self._inventory_thread = self._handler = self._lease = None
|
|
self._active, self._metrics = {}, Counter()
|
|
self._cases = {}
|
|
self._case_progress = {}
|
|
self._budgets = {}
|
|
self._inventory = None
|
|
self._started_at = self._heartbeat = self._stopped_at = self._progress = None
|
|
self._last_error = None
|
|
self._failures = 0
|
|
self._healthy = True
|
|
self._channels = {channel: True for channel in ("log", "snapshot", "scan", "observation")}
|
|
self._last_fallback = float("-inf")
|
|
self._last_scan = float("-inf")
|
|
self._last_flush = float("-inf")
|
|
self.logger = _PhaseLogger(self)
|
|
self._gateway_filter = _GatewayFilter(self.service_id)
|
|
|
|
def _now(self):
|
|
now = self.clock()
|
|
if now.tzinfo is None or now.utcoffset() is None:
|
|
raise TelemetryError("TELEMETRY_CLOCK")
|
|
return now.isoformat()
|
|
|
|
def __enter__(self):
|
|
return self.start()
|
|
|
|
def __exit__(self, *args):
|
|
self.stop()
|
|
|
|
def start(self):
|
|
from viano_bug_analysis.lifecycle.locking import FileLock
|
|
if self.started or self._stopped_at:
|
|
raise TelemetryError("TELEMETRY_SESSION")
|
|
self._lease = FileLock(self.workspace.runtime / "locks" / "telemetry.lock")
|
|
self._lease.__enter__()
|
|
try:
|
|
if self.directory.is_symlink():
|
|
raise TelemetryError("TELEMETRY_PATH")
|
|
self.directory.mkdir(exist_ok=True)
|
|
self._check_paths()
|
|
log = self.directory / "service.jsonl"
|
|
if log.exists() and log.stat().st_size:
|
|
with log.open("rb") as stream:
|
|
stream.seek(-1, 2)
|
|
truncated = stream.read(1) != b"\n"
|
|
if truncated:
|
|
with log.open("ab") as stream:
|
|
stream.write(b"\n")
|
|
self._handler = _Handler(self.directory / "service.jsonl", maxBytes=self.config.log_max_bytes,
|
|
backupCount=self.config.log_backups, encoding="utf-8", delay=True)
|
|
self._handler.setFormatter(logging.Formatter("%(message)s"))
|
|
self._started_at = self._heartbeat = self._now()
|
|
self.started = True
|
|
logging.getLogger("viano_bug_analysis.llm").addFilter(self._gateway_filter)
|
|
self.emit("observer_started")
|
|
self._flush(force=True)
|
|
self._thread = threading.Thread(target=self._pulse, name="vba-telemetry", daemon=True)
|
|
self._thread.start()
|
|
self.refresh_inventory(force=True)
|
|
self._inventory_thread = threading.Thread(target=self._inventory_pulse, name="vba-inventory", daemon=True)
|
|
self._inventory_thread.start()
|
|
return self
|
|
except BaseException:
|
|
self._stop.set()
|
|
self.started = False
|
|
logging.getLogger("viano_bug_analysis.llm").removeFilter(self._gateway_filter)
|
|
if self._handler:
|
|
self._handler.close()
|
|
self._lease.__exit__()
|
|
raise
|
|
|
|
def stop(self):
|
|
if not self.started:
|
|
return
|
|
self._stop.set()
|
|
if self._thread and threading.current_thread() is not self._thread:
|
|
self._thread.join(timeout=2)
|
|
if self._inventory_thread:
|
|
self._inventory_thread.join(timeout=2)
|
|
with self._lock:
|
|
self._stopped_at = self._heartbeat = self._now()
|
|
self.emit("observer_stopped")
|
|
self._flush(force=True)
|
|
self.started = False
|
|
logging.getLogger("viano_bug_analysis.llm").removeFilter(self._gateway_filter)
|
|
self._handler.close()
|
|
self._lease.__exit__()
|
|
|
|
def _check_paths(self):
|
|
if self.directory.is_symlink():
|
|
raise TelemetryError("TELEMETRY_PATH")
|
|
paths = [self.directory / name for name in ("service.jsonl", "status.json", "metrics.prom")]
|
|
paths += [self.directory / f"service.jsonl.{i}" for i in range(1, self.config.log_backups + 1)]
|
|
if any(path.is_symlink() for path in paths):
|
|
raise TelemetryError("TELEMETRY_PATH")
|
|
|
|
def _failed(self, channel="observation"):
|
|
self._failures += 1
|
|
self._channels[channel] = False
|
|
self._healthy = False
|
|
if self.monotonic() - self._last_fallback >= 30:
|
|
self._last_fallback = self.monotonic()
|
|
try:
|
|
sys.stderr.write('{"event":"telemetry_write_failed","code":"TELEMETRY_STORAGE"}\n')
|
|
except OSError:
|
|
pass
|
|
|
|
def _metric(self, name, amount=1, **labels):
|
|
if not math.isfinite(amount) or amount < 0:
|
|
return
|
|
key = (name, tuple(sorted(labels.items())))
|
|
if key not in self._metrics and len(self._metrics) >= 2000:
|
|
self._failed()
|
|
return
|
|
self._metrics[key] += amount
|
|
|
|
def emit(self, event, *, level="INFO", code=None, status=None, context=None, extra=None):
|
|
if not self.started:
|
|
return
|
|
if event not in EVENTS or level not in {"INFO", "WARNING", "ERROR"}:
|
|
raise TelemetryError("TELEMETRY_EVENT")
|
|
with self._lock:
|
|
ctx = context or _context.get() or {}
|
|
record = {"format": "viano-service-event", "format_version": 1, "timestamp": self._now(),
|
|
"service_id": self.service_id, "level": level, "event": event,
|
|
"span_id": hashed(ctx.get("span_id"), 32), "parent_span_id": hashed(ctx.get("parent_span_id"), 32),
|
|
"operation": ctx.get("operation") if ctx.get("operation") in OPERATIONS else None,
|
|
"case_key": hashed(ctx.get("case_key")), "package_key": hashed(ctx.get("package_key")),
|
|
"status": status if status in STATUSES else None, "code": safe_code(code) if code else None}
|
|
if extra:
|
|
record["detail"] = detail(event, extra)
|
|
if level == "ERROR":
|
|
self._last_error = {key: record[key] for key in ("timestamp", "event", "span_id", "case_key", "code")}
|
|
try:
|
|
self._check_paths()
|
|
text = json.dumps(record, ensure_ascii=True, allow_nan=False)
|
|
if len(text.encode()) > min(self.config.max_snapshot_bytes, self.config.log_max_bytes):
|
|
raise TelemetryError("TELEMETRY_LIMIT")
|
|
self._handler.emit(logging.LogRecord("vba.telemetry", getattr(logging, level), "", 0, text, (), None))
|
|
self._channels["log"] = True
|
|
self._healthy = all(self._channels.values())
|
|
except Exception:
|
|
self._failed("log")
|
|
|
|
@contextmanager
|
|
def operation(self, operation, case_id=None, package_id=None):
|
|
if operation not in OPERATIONS:
|
|
raise TelemetryError("TELEMETRY_OPERATION")
|
|
parent = _context.get() or {}
|
|
ctx = {"service_id": self.service_id, "span_id": uuid.uuid4().hex, "parent_span_id": parent.get("span_id"), "operation": operation,
|
|
"case_key": identity(case_id) or parent.get("case_key"), "package_key": identity(package_id) or parent.get("package_key")}
|
|
token = _context.set(ctx)
|
|
scope = _Scope(self)
|
|
begun = self.monotonic()
|
|
with self._lock:
|
|
self._active[ctx["span_id"]] = {**ctx, "started_at": self._now()}
|
|
self.emit("operation_started", context=ctx)
|
|
self._flush(force=ctx["parent_span_id"] is None)
|
|
try:
|
|
yield scope
|
|
except BaseException as error:
|
|
scope.failed = True
|
|
scope.status = "CANCELLED" if type(error).__name__ in {"CancelledError", "KeyboardInterrupt", "SystemExit"} else "BUSY" if type(error).__name__ == "LockBusy" else "ERROR"
|
|
scope.code = error_code(error)
|
|
self.emit("operation_failed", level="WARNING" if scope.status in {"BUSY", "CANCELLED"} else "ERROR", status=scope.status, code=scope.code, context=ctx,
|
|
extra={"exceptions": diagnostic(error, self.config.max_frames)})
|
|
raise
|
|
finally:
|
|
elapsed = max(0, self.monotonic() - begun)
|
|
with self._lock:
|
|
self._active.pop(ctx["span_id"], None)
|
|
self._metric("vba_operations_total", operation=operation, status=scope.status)
|
|
self._metric("vba_operation_duration_seconds_count", operation=operation)
|
|
self._metric("vba_operation_duration_seconds_sum", elapsed, operation=operation)
|
|
if not scope.failed:
|
|
self.emit("operation_finished", level="ERROR" if scope.status == "ERROR" else "WARNING" if scope.status in {"BLOCKED", "REJECTED", "QUARANTINED"} else "INFO",
|
|
status=scope.status, code=scope.code, context=ctx, extra={"duration_seconds": elapsed, "reused": scope.reused, **scope.detail})
|
|
self._flush(force=ctx["parent_span_id"] is None)
|
|
_context.reset(token)
|
|
|
|
def exception(self, error):
|
|
self.emit("phase_diagnostic", level="ERROR", code=error_code(error), extra={"exceptions": diagnostic(error, self.config.max_frames)})
|
|
|
|
def guard(self, callback, *args):
|
|
try:
|
|
return callback(*args)
|
|
except Exception as error:
|
|
with self._lock:
|
|
self._failed()
|
|
try:
|
|
self.emit("phase_diagnostic", level="ERROR", code="TELEMETRY_CALLBACK", extra={"exceptions": diagnostic(error, self.config.max_frames)})
|
|
except Exception:
|
|
pass
|
|
|
|
def result_metrics(self, result):
|
|
with self._lock:
|
|
ctx = _context.get() or {}
|
|
if ctx.get("operation") == "inbox_poll" and isinstance(result, list):
|
|
for item in result:
|
|
status = getattr(item, "status", None)
|
|
if status in STATUSES:
|
|
self._metric("vba_inbox_results_total", status=status)
|
|
if ctx.get("operation") == "evidence_execute" and not getattr(result, "reused", True):
|
|
self._metric("vba_evidence_records_published_total", len(result.record_refs))
|
|
if ctx.get("operation") == "budget_execute" and hasattr(result, "state"):
|
|
ident = (ctx.get("case_key"), ctx.get("package_key"))
|
|
if ident not in self._budgets and len(self._budgets) >= self.config.max_cases:
|
|
self._failed()
|
|
return
|
|
accounts = result.state["accounts"].values()
|
|
self._budgets[ident] = {"case_key": ident[0], "package_key": ident[1], "analysis_revision": result.analysis_revision,
|
|
**{field: sum(a[field] for a in accounts) for field in ("remaining", "spent", "rewarded")},
|
|
"pending_actions": sum(action["status"] != "settled" for a in accounts for action in a["actions"].values())}
|
|
|
|
def case_checkpoint(self, manifest, details=None):
|
|
if not self.started:
|
|
return
|
|
with self._lock:
|
|
case_key, package_key = identity(manifest["case_id"]), identity(manifest["package_id"])
|
|
ident = (case_key, package_key)
|
|
if ident not in self._cases and len(self._cases) >= self.config.max_cases:
|
|
self._failed()
|
|
return
|
|
item = {"case_key": case_key, "package_key": package_key, "state": manifest["state"],
|
|
"package_generation": manifest["package_generation"], "analysis_revision": manifest["analysis_revision"],
|
|
"reason": safe_code(manifest["routing"]["reason"]) if manifest["routing"]["reason"] else None,
|
|
"retry_not_before": manifest["routing"]["retry_not_before"]}
|
|
item["scheduling"] = scheduling(manifest, details or {}, details if details and details.get("operation") == "llm-round" else None, self.clock())
|
|
if item["scheduling"] == "delayed" and details and details.get("status") == "RETRY_PENDING":
|
|
item["retry_not_before"] = details["retry_not_before"]
|
|
previous = self._cases.get(ident)
|
|
if previous and previous["package_generation"] >= item["package_generation"]:
|
|
return
|
|
self._cases[ident] = item
|
|
self._progress = self._now()
|
|
self._case_progress[ident] = self._progress
|
|
self._metric("vba_case_checkpoints_total", state=item["state"])
|
|
self.emit("case_checkpoint", context={**(_context.get() or {}), "case_key": case_key, "package_key": package_key}, extra=item)
|
|
|
|
def llm_round(self, record):
|
|
if not self.started:
|
|
return
|
|
with self._lock:
|
|
phase, status = record["phase"], record["status"]
|
|
self._metric("vba_llm_rounds_total", phase=phase, status=status)
|
|
metadata = record["outcome"].get("metadata", {})
|
|
for field in ("input_tokens", "output_tokens"):
|
|
value = metadata.get(field)
|
|
if type(value) is int and value >= 0:
|
|
self._metric("vba_llm_tokens_total", value, direction=field)
|
|
else:
|
|
self._metric("vba_llm_usage_unknown_total", field=field)
|
|
cost, currency = metadata.get("estimated_cost"), metadata.get("cost_currency")
|
|
currency_valid = isinstance(currency, str) and re.fullmatch("[A-Z]{3}", currency) is not None
|
|
if type(cost) in {int, float} and math.isfinite(cost) and cost >= 0 and currency_valid:
|
|
self._metric("vba_llm_estimated_cost_total", cost, currency=currency)
|
|
else:
|
|
self._metric("vba_llm_usage_unknown_total", field="estimated_cost")
|
|
attempts = metadata.get("attempts", record["outcome"].get("attempts", []))
|
|
self._metric("vba_llm_provider_failures_total", len(attempts), phase=phase)
|
|
self._metric("vba_llm_failed_attempt_usage_unknown_total", len(attempts))
|
|
self.emit("llm_round", level="INFO" if status == "SUCCESS" else "WARNING", status=status, code=record.get("code"),
|
|
extra={"phase": phase, "provider_key": identity(metadata.get("provider")), "model_key": identity(metadata.get("model")),
|
|
"http_status": record["outcome"].get("http_status"),
|
|
"input_tokens": metadata.get("input_tokens"), "output_tokens": metadata.get("output_tokens"),
|
|
"estimated_cost": cost if currency_valid else None,
|
|
"currency": currency if currency_valid else None,
|
|
"failed_providers": [{"provider_key": identity(a.get("provider")), "code": safe_code(a.get("code")), "http_status": a.get("http_status")} for a in attempts]})
|
|
|
|
def refresh_inventory(self, *, force=False):
|
|
if not self.started or not force and self.monotonic() - self._last_scan < self.config.inventory_interval_seconds:
|
|
return
|
|
try:
|
|
with self.operation("inventory_scan"):
|
|
scanned = scan_workspace(self.workspace.root, limits=self.package_limits, max_cases=self.config.max_cases, now=self.clock(), cancelled=self._stop.is_set)
|
|
with self._lock:
|
|
self._inventory = scanned
|
|
for item in scanned["cases"]:
|
|
ident = (item["case_key"], item["package_key"])
|
|
previous = self._cases.get(ident)
|
|
if previous and item["package_generation"] > previous["package_generation"]:
|
|
self._progress = self._now()
|
|
self._case_progress[ident] = self._progress
|
|
if ident in self._cases or len(self._cases) < self.config.max_cases:
|
|
if previous is None or previous["package_generation"] <= item["package_generation"]:
|
|
self._cases[ident] = item
|
|
self._channels["scan"] = scanned["complete"]
|
|
self._healthy = all(self._channels.values())
|
|
self._last_scan = self.monotonic()
|
|
self.emit("inventory_updated", level="INFO" if scanned["complete"] else "WARNING", extra={"complete": scanned["complete"], "counts": scanned["counts"], "inbox_files": scanned["inbox_files"]})
|
|
except Exception:
|
|
# Observation errors never change the analysis result.
|
|
with self._lock:
|
|
self._failed("scan")
|
|
|
|
def snapshot(self):
|
|
from dataclasses import asdict
|
|
with self._lock:
|
|
now = self.clock()
|
|
active = []
|
|
for item in self._active.values():
|
|
elapsed = max(0, (now - datetime.fromisoformat(item["started_at"])).total_seconds())
|
|
progressed = self._case_progress.get((item["case_key"], item["package_key"]), item["started_at"])
|
|
progress_age = max(0, (now - datetime.fromisoformat(progressed)).total_seconds())
|
|
active.append({**item, "elapsed_seconds": elapsed, "progress_age_seconds": progress_age,
|
|
"stalled": elapsed > self.config.progress_warning_seconds and progress_age > self.config.progress_warning_seconds})
|
|
inventory = json.loads(json.dumps(self._inventory)) if self._inventory else None
|
|
if inventory:
|
|
rows = {(item["case_key"], item["package_key"]): item for item in inventory["cases"]}
|
|
for ident, item in self._cases.items():
|
|
if ident not in rows or rows[ident]["package_generation"] <= item["package_generation"]:
|
|
rows[ident] = item
|
|
inventory["cases"] = list(rows.values())
|
|
for row in inventory["cases"]:
|
|
if row.get("scheduling") == "delayed" and row.get("retry_not_before") and datetime.fromisoformat(row["retry_not_before"].replace("Z", "+00:00")) <= now:
|
|
row["scheduling"] = "ready"
|
|
inventory["counts"] = count_cases(inventory["cases"])
|
|
data = {"format": "viano-service-status", "format_version": 1, "service_id": self.service_id, "pid": os.getpid(),
|
|
"started_at": self._started_at, "heartbeat_at": self._heartbeat, "stopped_at": self._stopped_at,
|
|
"last_progress_at": self._progress, "last_error": self._last_error, "telemetry_failures": self._failures,
|
|
"telemetry_healthy": self._healthy, "budgets": list(self._budgets.values()),
|
|
"active_operations": active, "cases": list(self._cases.values()), "inventory": inventory,
|
|
"config": asdict(self.config), "counters": [{"name": name, "labels": dict(labels), "value": value} for (name, labels), value in sorted(self._metrics.items())]}
|
|
return assess(data, self.clock())
|
|
|
|
def prometheus(self):
|
|
data = self.snapshot()
|
|
lines = [f"vba_observer_alive {int(data['observer'] == 'LIVE')}", f"vba_active_operations {len(data['active_operations'])}",
|
|
f"vba_telemetry_failures_total {self._failures}", f"vba_inventory_fresh {int(data['inventory_fresh'])}",
|
|
f"vba_observer_heartbeat_time_seconds {datetime.fromisoformat(self._heartbeat).timestamp()}",
|
|
f"vba_last_progress_time_seconds {datetime.fromisoformat(self._progress).timestamp() if self._progress else 0}",
|
|
f"vba_observer_start_time_seconds {datetime.fromisoformat(self._started_at).timestamp()}"]
|
|
for kind in ("remaining", "spent", "rewarded", "pending_actions"):
|
|
lines.append('vba_observed_budget{kind="' + kind + '"} ' + str(sum(row[kind] for row in data["budgets"])))
|
|
for metric in data["counters"]:
|
|
labels = ",".join(name + "=" + json.dumps(value) for name, value in sorted(metric["labels"].items()))
|
|
lines.append(metric["name"] + ("{" + labels + "}" if labels else "") + " " + str(metric["value"]))
|
|
if data["inventory_fresh"]:
|
|
lines.append("vba_inbox_files " + str(data["inventory"]["inbox_files"]))
|
|
for category, value in data["inventory"]["counts"].items():
|
|
lines.append('vba_cases_observed{category="' + category + '"} ' + str(value))
|
|
return "\n".join(lines) + "\n"
|
|
|
|
def _write(self, name, data):
|
|
temporary = None
|
|
try:
|
|
if len(data) > self.config.max_snapshot_bytes:
|
|
raise TelemetryError("TELEMETRY_LIMIT")
|
|
with tempfile.NamedTemporaryFile(dir=self.directory, prefix=".telemetry-", delete=False) as stream:
|
|
temporary = Path(stream.name)
|
|
stream.write(data)
|
|
stream.flush()
|
|
os.fsync(stream.fileno())
|
|
os.replace(temporary, self.directory / name)
|
|
finally:
|
|
if temporary:
|
|
temporary.unlink(missing_ok=True)
|
|
|
|
def flush(self):
|
|
self._flush(force=True)
|
|
|
|
def _flush(self, *, force=False):
|
|
if not self.started:
|
|
return
|
|
with self._lock:
|
|
if not force and self.monotonic() - self._last_flush < self.config.snapshot_interval_seconds:
|
|
return
|
|
try:
|
|
self._check_paths()
|
|
self._write("metrics.prom", self.prometheus().encode())
|
|
self._channels["snapshot"] = True
|
|
self._healthy = all(self._channels.values())
|
|
self._write("status.json", json.dumps(self.snapshot(), allow_nan=False).encode())
|
|
self._last_flush = self.monotonic()
|
|
except Exception:
|
|
self._failed("snapshot")
|
|
|
|
def _pulse(self):
|
|
while not self._stop.wait(self.config.heartbeat_seconds):
|
|
try:
|
|
with self._lock:
|
|
self._heartbeat = self._now()
|
|
self._flush(force=True)
|
|
except Exception:
|
|
with self._lock:
|
|
self._failed("snapshot")
|
|
|
|
def _inventory_pulse(self):
|
|
while not self._stop.wait(self.config.inventory_interval_seconds):
|
|
self.refresh_inventory(force=True)
|