Ginnipahwa05/Meta-Pytorch
0
1#!/usr/bin/env python32"""3FastAPI server for a distributed incident war-room OpenEnv.4Deployable to Hugging Face Spaces.5"""6 7import os8import json9import time10import base6411import fnmatch12from datetime import datetime13from typing import Optional, Dict, Any, List14from fastapi import FastAPI, HTTPException15from fastapi.responses import JSONResponse, HTMLResponse16from pydantic import BaseModel17import requests18from dotenv import load_dotenv19 20from environment import make_env, DevOpsWarRoomEnv21from graders import safe_display_score, safe_task_score22from models import Action, ActionType, ServiceName, MetricType, Observation, Reward23from scripts.seed_project_errors_to_elastic import docs_for_scenario24from tasks import TASK_DEFINITIONS25 26load_dotenv()27 28 29app = FastAPI(30 title="Distributed Incident War Room OpenEnv",31 description="SRE debugging simulation environment for live distributed production incidents",32 version="1.0.0",33)34 35# Global environment instance (for HF Space stateful deployment)36current_env: Optional[DevOpsWarRoomEnv] = None37current_task_id: str = "easy_0"38LOCAL_DEMO_LOG_FILE = os.getenv(39 "LOCAL_DEMO_LOG_FILE",40 os.path.join(os.path.dirname(__file__), ".run", "local-demo-logs.jsonl"),41)42 43 44def _get_source_env() -> Optional[DevOpsWarRoomEnv]:45 source_env = current_env46 if source_env is None:47 try:48 source_env = make_env(task_id=current_task_id, seed=0)49 source_env.reset()50 except Exception:51 source_env = None52 return source_env53 54 55def _site_base_url(site: str) -> str:56 site = (site or "datadoghq.com").strip()57 return f"https://api.{site}" if not site.startswith("api.") else f"https://{site}"58 59 60def _get_datadog_settings() -> Dict[str, Any]:61 indexes = [62 item.strip()63 for item in os.getenv("DD_LOG_INDEXES", "").split(",")64 if item.strip()65 ]66 return {67 "api_key": os.getenv("DD_API_KEY", "").strip(),68 "app_key": os.getenv("DD_APP_KEY", "").strip(),69 "site": os.getenv("DD_SITE", "datadoghq.com").strip(),70 "indexes": indexes,71 }72 73 74def _datadog_enabled(settings: Optional[Dict[str, Any]] = None) -> bool:75 settings = settings or _get_datadog_settings()76 return bool(settings.get("api_key") and settings.get("app_key"))77 78 79def _get_elasticsearch_settings() -> Dict[str, Any]:80 return {81 "url": os.getenv("ELASTICSEARCH_URL", "").strip().rstrip("/"),82 "api_key": os.getenv("ELASTICSEARCH_API_KEY", "").strip(),83 "username": os.getenv("ELASTICSEARCH_USERNAME", "").strip(),84 "password": os.getenv("ELASTICSEARCH_PASSWORD", "").strip(),85 "index": os.getenv("ELASTICSEARCH_LOG_INDEX", "logs-*").strip() or "logs-*",86 "timestamp_field": os.getenv("ELASTICSEARCH_TIMESTAMP_FIELD", "@timestamp").strip() or "@timestamp",87 "service_field": os.getenv("ELASTICSEARCH_SERVICE_FIELD", "service").strip() or "service",88 "message_field": os.getenv("ELASTICSEARCH_MESSAGE_FIELD", "message").strip() or "message",89 "level_field": os.getenv("ELASTICSEARCH_LEVEL_FIELD", "log.level").strip() or "log.level",90 "verify_tls": os.getenv("ELASTICSEARCH_VERIFY_TLS", "true").strip().lower() not in {"0", "false", "no"},91 }92 93 94def _elasticsearch_enabled(settings: Optional[Dict[str, Any]] = None) -> bool:95 settings = settings or _get_elasticsearch_settings()96 return bool(settings.get("url"))97 98 99def _mask_secret(value: str) -> str:100 if not value:101 return ""102 if len(value) <= 8:103 return "*" * len(value)104 return f"{value[:4]}...{value[-4:]}"105 106 107def _elastic_auth_headers(settings: Dict[str, Any]) -> Dict[str, str]:108 headers: Dict[str, str] = {109 "Accept": "application/json",110 "Content-Type": "application/json",111 }112 if settings.get("api_key"):113 headers["Authorization"] = f"ApiKey {settings['api_key']}"114 elif settings.get("username") and settings.get("password"):115 token = base64.b64encode(f"{settings['username']}:{settings['password']}".encode("utf-8")).decode("utf-8")116 headers["Authorization"] = f"Basic {token}"117 return headers118 119 120def _build_log_search_payload(search_query: str, limit: int, minutes: int, indexes: Optional[List[str]] = None) -> Dict[str, Any]:121 payload = {122 "filter": {123 "query": search_query,124 "from": f"now-{max(1, minutes)}m",125 "to": "now",126 },127 "sort": "timestamp",128 "page": {"limit": max(1, min(limit, 100))},129 }130 if indexes:131 payload["filter"]["indexes"] = indexes132 return payload133 134 135def _build_datadog_search_query(query: str, service: Optional[str], default: str) -> str:136 normalized_query = (query or "").strip()137 ignore_queries = {"*", "*:*", "service:*", "service.name:*"}138 filters: List[str] = []139 if normalized_query and normalized_query not in ignore_queries:140 filters.append(f"({normalized_query})")141 if service:142 filters.append(f"service:{service}")143 return " AND ".join(filters) if filters else default144 145 146def _parse_timestamp(value: Any) -> float:147 if value in (None, ""):148 return 0.0149 if isinstance(value, (int, float)):150 return float(value)151 if isinstance(value, str):152 normalized = value.strip()153 if not normalized:154 return 0.0155 if normalized.endswith("Z"):156 normalized = normalized[:-1] + "+00:00"157 try:158 return datetime.fromisoformat(normalized).timestamp()159 except ValueError:160 try:161 return time.mktime(time.strptime(normalized[:19], "%Y-%m-%dT%H:%M:%S"))162 except ValueError:163 return 0.0164 return 0.0165 166 167def _field_candidates(payload: Dict[str, Any], field: str) -> List[Any]:168 mapping = {169 "service": ["service", "service.name"],170 "service.name": ["service.name", "service"],171 "message": ["message", "event.original"],172 "log.level": ["log.level", "level"],173 "level": ["level", "log.level"],174 "trace.id": ["trace.id", "trace_id"],175 "trace_id": ["trace_id", "trace.id"],176 "@timestamp": ["@timestamp", "timestamp"],177 "timestamp": ["timestamp", "@timestamp"],178 }179 values: List[Any] = []180 for key in mapping.get(field, [field]):181 if key in payload:182 values.append(payload[key])183 return values184 185 186def _matches_local_query_term(payload: Dict[str, Any], term: str) -> bool:187 normalized = term.strip().strip("()")188 if not normalized or normalized in {"*", "*:*", "service:*", "service.name:*"}:189 return True190 191 if ":" in normalized:192 field, pattern = normalized.split(":", 1)193 field = field.strip()194 pattern = pattern.strip().strip('"').strip("'")195 values = _field_candidates(payload, field)196 if pattern == "*":197 return any(value not in (None, "") for value in values)198 lowered_pattern = pattern.lower()199 for value in values:200 if value in (None, ""):201 continue202 candidate = str(value)203 candidate_lower = candidate.lower()204 if "*" in lowered_pattern or "?" in lowered_pattern:205 if fnmatch.fnmatchcase(candidate_lower, lowered_pattern):206 return True207 elif lowered_pattern == candidate_lower or lowered_pattern in candidate_lower:208 return True209 return False210 211 haystack = json.dumps(payload, default=str).lower()212 return normalized.lower() in haystack213 214 215def _matches_local_query(payload: Dict[str, Any], query: str) -> bool:216 normalized = (query or "*").strip()217 if not normalized or normalized in {"*", "*:*", "service:*", "service.name:*"}:218 return True219 if " OR " in normalized:220 return any(_matches_local_query(payload, part) for part in normalized.split(" OR "))221 if " AND " in normalized:222 return all(_matches_local_query(payload, part) for part in normalized.split(" AND "))223 return _matches_local_query_term(payload, normalized)224 225 226def _load_local_demo_logs(query: str, service: Optional[str], limit: int, minutes: int) -> List[Dict[str, Any]]:227 if not os.path.exists(LOCAL_DEMO_LOG_FILE):228 return []229 230 min_timestamp = time.time() - max(1, minutes) * 60231 logs: List[Dict[str, Any]] = []232 233 try:234 with open(LOCAL_DEMO_LOG_FILE, "r", encoding="utf-8") as handle:235 for line in handle:236 raw_line = line.strip()237 if not raw_line:238 continue239 try:240 payload = json.loads(raw_line)241 except json.JSONDecodeError:242 continue243 244 if service:245 service_value = str(246 _field_candidates(payload, "service")[0]247 ) if _field_candidates(payload, "service") else ""248 if service_value != service:249 continue250 if _parse_timestamp(payload.get("@timestamp")) < min_timestamp:251 continue252 if not _matches_local_query(payload, query):253 continue254 255 logs.append(256 {257 "timestamp": payload.get("@timestamp") or payload.get("timestamp"),258 "service": payload.get("service") or payload.get("service.name"),259 "level": payload.get("log.level") or payload.get("level") or "info",260 "message": payload.get("message") or json.dumps(payload)[:240],261 "trace_id": payload.get("trace.id") or payload.get("trace_id"),262 "is_relevant": True,263 "raw": payload,264 }265 )266 except OSError:267 return []268 269 logs.sort(key=lambda item: _parse_timestamp(item.get("timestamp")), reverse=True)270 return logs[: max(1, min(limit, 100))]271 272 273def _has_local_demo_logs() -> bool:274 if not os.path.exists(LOCAL_DEMO_LOG_FILE):275 return False276 try:277 return os.path.getsize(LOCAL_DEMO_LOG_FILE) > 0278 except OSError:279 return False280 281 282def _pick_first(payload: Dict[str, Any], paths: List[str], default: Any = None) -> Any:283 for path in paths:284 if isinstance(payload, dict) and path in payload and payload[path] not in (None, ""):285 return payload[path]286 cursor: Any = payload287 found = True288 for key in path.split("."):289 if isinstance(cursor, dict) and key in cursor:290 cursor = cursor[key]291 else:292 found = False293 break294 if found and cursor not in (None, ""):295 return cursor296 return default297 298 299def _build_elasticsearch_log_payload(300 query: str,301 service: Optional[str],302 limit: int,303 minutes: int,304 settings: Dict[str, Any],305) -> Dict[str, Any]:306 timestamp_field = settings["timestamp_field"]307 service_field = settings["service_field"]308 filters: List[Dict[str, Any]] = [309 {310 "range": {311 timestamp_field: {312 "gte": f"now-{max(1, minutes)}m",313 "lte": "now",314 }315 }316 }317 ]318 if service:319 filters.append(320 {321 "bool": {322 "should": [323 {"term": {service_field: service}},324 {"term": {f"{service_field}.keyword": service}},325 {"term": {"service.name": service}},326 {"term": {"service.name.keyword": service}},327 ],328 "minimum_should_match": 1,329 }330 }331 )332 333 must: List[Dict[str, Any]] = []334 normalized_query = (query or "").strip()335 if normalized_query and normalized_query not in {"*", "*:*", "service:*", "service.name:*"}:336 must.append(337 {338 "query_string": {339 "query": normalized_query,340 "default_operator": "AND",341 }342 }343 )344 345 return {346 "size": max(1, min(limit, 100)),347 "sort": [{timestamp_field: {"order": "desc", "unmapped_type": "date"}}],348 "query": {349 "bool": {350 "must": must,351 "filter": filters,352 }353 },354 }355 356 357def _serialize_log_entry(log: Any) -> Dict[str, Any]:358 if hasattr(log, "model_dump"):359 payload = log.model_dump()360 elif isinstance(log, dict):361 payload = dict(log)362 else:363 payload = {364 "timestamp": getattr(log, "timestamp", None),365 "service": getattr(getattr(log, "service", None), "value", getattr(log, "service", None)),366 "level": getattr(log, "level", None),367 "message": getattr(log, "message", None),368 "trace_id": getattr(log, "trace_id", None),369 "is_relevant": getattr(log, "is_relevant", None),370 }371 372 service_value = payload.get("service")373 if hasattr(service_value, "value"):374 payload["service"] = service_value.value375 return payload376 377 378def _fetch_datadog_logs(query: str = "*", service: Optional[str] = None, limit: int = 25, minutes: int = 15) -> Dict[str, Any]:379 settings = _get_datadog_settings()380 api_key = settings["api_key"]381 app_key = settings["app_key"]382 site = settings["site"]383 384 filters: List[str] = []385 search_query = _build_datadog_search_query(query, service, "*")386 387 if not api_key or not app_key:388 fallback_logs = _load_local_demo_logs(query=query, service=service, limit=limit, minutes=minutes)389 source_env = _get_source_env()390 391 if source_env is not None:392 matching_logs = [393 _serialize_log_entry(log)394 for log in source_env.all_logs395 if not service or str(getattr(log.service, "value", log.service)) == service396 ]397 fallback_logs.extend(matching_logs)398 399 fallback_logs.sort(key=lambda item: _parse_timestamp(item.get("timestamp")), reverse=True)400 fallback_logs = fallback_logs[: max(1, min(limit, 100))]401 402 return {403 "source": "local-fallback",404 "query": search_query,405 "logs": fallback_logs,406 "note": (407 "DD_API_KEY and DD_APP_KEY are not configured. "408 "Showing locally replayed demo logs and simulator logs instead."409 if fallback_logs410 else "DD_API_KEY or DD_APP_KEY not configured. Showing environment logs instead."411 ),412 }413 414 payload = _build_log_search_payload(415 search_query=search_query,416 limit=limit,417 minutes=minutes,418 indexes=settings.get("indexes"),419 )420 421 response = requests.post(422 f"{_site_base_url(site)}/api/v2/logs/events/search",423 headers={424 "DD-API-KEY": api_key,425 "DD-APPLICATION-KEY": app_key,426 "Accept": "application/json",427 "Content-Type": "application/json",428 },429 json=payload,430 timeout=12,431 )432 433 if response.status_code >= 400:434 raise HTTPException(status_code=502, detail=f"Datadog logs API error: {response.text}")435 436 body = response.json()437 raw_logs = body.get("data") or body.get("logs") or []438 parsed_logs: List[Dict[str, Any]] = []439 440 for item in raw_logs:441 attributes = item.get("attributes", {}) if isinstance(item, dict) else {}442 parsed_logs.append(443 {444 "timestamp": item.get("id") or attributes.get("timestamp") or attributes.get("ingested_at"),445 "service": attributes.get("service") or attributes.get("service_name") or attributes.get("host"),446 "level": attributes.get("status") or attributes.get("level") or attributes.get("alert_type") or "info",447 "message": attributes.get("message") or attributes.get("title") or attributes.get("text") or json.dumps(attributes)[:240],448 "trace_id": attributes.get("trace_id") or attributes.get("dd.trace_id"),449 "is_relevant": True,450 "raw": item,451 }452 )453 454 return {455 "source": "datadog",456 "query": search_query,457 "logs": parsed_logs,458 }459 460 461def _fetch_elasticsearch_logs(query: str = "*", service: Optional[str] = None, limit: int = 25, minutes: int = 15) -> Dict[str, Any]:462 settings = _get_elasticsearch_settings()463 if not _elasticsearch_enabled(settings):464 raise HTTPException(status_code=500, detail="Elasticsearch is not configured.")465 466 payload = _build_elasticsearch_log_payload(467 query=query,468 service=service,469 limit=limit,470 minutes=minutes,471 settings=settings,472 )473 response = requests.post(474 f"{settings['url']}/{settings['index']}/_search",475 headers=_elastic_auth_headers(settings),476 json=payload,477 timeout=12,478 verify=settings["verify_tls"],479 )480 481 if response.status_code >= 400:482 raise HTTPException(status_code=502, detail=f"Elasticsearch logs API error: {response.text}")483 484 body = response.json()485 hits = (((body or {}).get("hits") or {}).get("hits")) or []486 logs: List[Dict[str, Any]] = []487 message_field = settings["message_field"]488 level_field = settings["level_field"]489 service_field = settings["service_field"]490 timestamp_field = settings["timestamp_field"]491 492 for hit in hits:493 source = hit.get("_source", {}) if isinstance(hit, dict) else {}494 logs.append(495 {496 "timestamp": _pick_first(source, [timestamp_field], hit.get("sort", [None])[0] if isinstance(hit, dict) else None),497 "service": _pick_first(source, [service_field, f"{service_field}.name", "service.name", "kubernetes.container.name", "host.name"], hit.get("_index") if isinstance(hit, dict) else "unknown"),498 "level": _pick_first(source, [level_field, "level", "severity", "log.level"], "info"),499 "message": _pick_first(source, [message_field, "event.original", "message"], json.dumps(source)[:240] if source else ""),500 "trace_id": _pick_first(source, ["trace.id", "trace_id", "dd.trace_id"], None),501 "is_relevant": True,502 "raw": hit,503 }504 )505 506 return {507 "source": "elasticsearch",508 "query": query or "*",509 "logs": logs,510 "note": f"Showing Elasticsearch logs from index pattern `{settings['index']}`.",511 }512 513 514def _concrete_index_name(index_pattern: str) -> str:515 today = datetime.utcnow().strftime("%Y.%m.%d")516 if "*" in index_pattern:517 return index_pattern.replace("*", today)518 return index_pattern519 520 521def _bulk_lines(index_name: str, docs: List[Dict[str, Any]]) -> str:522 parts: List[str] = []523 for doc in docs:524 parts.append(json.dumps({"index": {"_index": index_name}}))525 parts.append(json.dumps(doc))526 return "\n".join(parts) + "\n"527 528 529def _seed_demo_logs_into_elasticsearch(scenario: str = "all") -> Dict[str, Any]:530 settings = _get_elasticsearch_settings()531 if not _elasticsearch_enabled(settings):532 raise HTTPException(status_code=400, detail="Elasticsearch is not configured for this dashboard.")533 534 docs = docs_for_scenario(scenario)535 target_index = _concrete_index_name(settings["index"])536 response = requests.post(537 f"{settings['url']}/_bulk",538 headers={539 **_elastic_auth_headers(settings),540 "Content-Type": "application/x-ndjson",541 },542 data=_bulk_lines(target_index, docs).encode("utf-8"),543 timeout=15,544 verify=settings["verify_tls"],545 )546 547 if response.status_code >= 400:548 raise HTTPException(status_code=502, detail=f"Elasticsearch bulk seed failed: {response.text}")549 550 body = response.json()551 if body.get("errors"):552 raise HTTPException(status_code=502, detail="Elasticsearch reported indexing errors while seeding demo logs.")553 554 incident_ids = sorted({doc["incident_id"] for doc in docs})555 return {556 "status": "seeded",557 "source": "elasticsearch",558 "scenario": scenario,559 "index": target_index,560 "count": len(docs),561 "service": "meta-pytorch-demo",562 "query": "*",563 "incident_ids": incident_ids,564 "note": "Fresh demo incidents were written into Elasticsearch and are ready in the dashboard log stream.",565 }566 567 568def _fetch_elasticsearch_traces(query: str = "service:*", service: Optional[str] = None, limit: int = 20, minutes: int = 15) -> Dict[str, Any]:569 settings = _get_elasticsearch_settings()570 if not _elasticsearch_enabled(settings):571 raise HTTPException(status_code=500, detail="Elasticsearch is not configured.")572 573 payload = _build_elasticsearch_log_payload(574 query=query,575 service=service,576 limit=limit,577 minutes=minutes,578 settings=settings,579 )580 response = requests.post(581 f"{settings['url']}/{settings['index']}/_search",582 headers=_elastic_auth_headers(settings),583 json=payload,584 timeout=12,585 verify=settings["verify_tls"],586 )587 588 if response.status_code >= 400:589 raise HTTPException(status_code=502, detail=f"Elasticsearch trace search error: {response.text}")590 591 body = response.json()592 hits = (((body or {}).get("hits") or {}).get("hits")) or []593 traces: List[Dict[str, Any]] = []594 595 for hit in hits:596 source = hit.get("_source", {}) if isinstance(hit, dict) else {}597 duration_nanos = _pick_first(598 source,599 [600 "event.duration",601 "transaction.duration.us",602 "span.duration.us",603 ],604 None,605 )606 if duration_nanos is not None:607 try:608 duration_value = float(duration_nanos)609 if duration_value > 100000:610 duration_ms: Optional[float] = round(duration_value / 1000000.0, 2)611 else:612 duration_ms = round(duration_value / 1000.0, 2)613 except (TypeError, ValueError):614 duration_ms = None615 else:616 duration_ms = None617 618 trace_id = _pick_first(source, ["trace.id", "trace_id", "transaction.id", "span.id"], None)619 message = _pick_first(620 source,621 [settings["message_field"], "event.original", "message"],622 json.dumps(source)[:160] if source else "",623 )624 traces.append(625 {626 "timestamp": _pick_first(source, [settings["timestamp_field"]], hit.get("sort", [None])[0] if isinstance(hit, dict) else None),627 "service": _pick_first(source, [settings["service_field"], "service.name", "service", "host.name"], "unknown"),628 "operation": _pick_first(source, ["event.action", "event.dataset", "transaction.name", "span.name"], "log_event"),629 "resource": message[:120],630 "duration_ms": duration_ms,631 "trace_id": trace_id,632 "derived": True,633 }634 )635 636 return {637 "source": "elasticsearch",638 "query": query or "*",639 "traces": traces,640 "note": "Showing Elastic log-derived trace events. Configure Datadog APM later if you want full distributed spans.",641 }642 643 644def _fetch_datadog_status() -> Dict[str, Any]:645 settings = _get_datadog_settings()646 site = settings["site"]647 configured = _datadog_enabled(settings)648 649 base_status = {650 "site": site,651 "configured": configured,652 "api_key_hint": _mask_secret(settings["api_key"]),653 "app_key_hint": _mask_secret(settings["app_key"]),654 "log_indexes": settings.get("indexes", []),655 }656 657 if not configured:658 return {659 **base_status,660 "connected": False,661 "source": "local-fallback",662 "message": "Set DD_API_KEY and DD_APP_KEY to stream Datadog logs into the dashboard.",663 }664 665 payload = _build_log_search_payload(search_query="*", limit=1, minutes=15, indexes=settings.get("indexes"))666 667 try:668 response = requests.post(669 f"{_site_base_url(site)}/api/v2/logs/events/search",670 headers={671 "DD-API-KEY": settings["api_key"],672 "DD-APPLICATION-KEY": settings["app_key"],673 "Accept": "application/json",674 "Content-Type": "application/json",675 },676 json=payload,677 timeout=10,678 )679 except requests.RequestException as exc:680 return {681 **base_status,682 "connected": False,683 "source": "datadog",684 "message": f"Datadog request failed: {exc}",685 }686 687 if response.status_code >= 400:688 try:689 details = response.json()690 except ValueError:691 details = {"errors": [response.text]}692 693 return {694 **base_status,695 "connected": False,696 "source": "datadog",697 "message": "Datadog credentials were found, but the log search request failed.",698 "details": details,699 "status_code": response.status_code,700 }701 702 body = response.json()703 event_count = len(body.get("data") or [])704 705 return {706 **base_status,707 "connected": True,708 "source": "datadog",709 "message": "Datadog log search is active for this dashboard.",710 "sample_count": event_count,711 }712 713 714def _fetch_elasticsearch_status() -> Dict[str, Any]:715 settings = _get_elasticsearch_settings()716 configured = _elasticsearch_enabled(settings)717 718 base_status = {719 "backend": "elasticsearch",720 "configured": configured,721 "url": settings["url"],722 "index": settings["index"],723 "timestamp_field": settings["timestamp_field"],724 "service_field": settings["service_field"],725 "message_field": settings["message_field"],726 "auth_mode": "api_key" if settings["api_key"] else ("basic" if settings["username"] and settings["password"] else "none"),727 "api_key_hint": _mask_secret(settings["api_key"]),728 "username_hint": settings["username"],729 }730 731 if not configured:732 return {733 **base_status,734 "connected": False,735 "source": "local-fallback",736 "message": "Set ELASTICSEARCH_URL to use Elasticsearch for dashboard logs.",737 }738 739 payload = _build_elasticsearch_log_payload(740 query="*",741 service=None,742 limit=1,743 minutes=15,744 settings=settings,745 )746 747 try:748 response = requests.post(749 f"{settings['url']}/{settings['index']}/_search",750 headers=_elastic_auth_headers(settings),751 json=payload,752 timeout=10,753 verify=settings["verify_tls"],754 )755 except requests.RequestException as exc:756 return {757 **base_status,758 "connected": False,759 "source": "elasticsearch",760 "message": f"Elasticsearch request failed: {exc}",761 }762 763 if response.status_code >= 400:764 return {765 **base_status,766 "connected": False,767 "source": "elasticsearch",768 "status_code": response.status_code,769 "message": "Elasticsearch credentials or index settings were found, but log search failed.",770 "details": response.text,771 }772 773 body = response.json()774 sample_count = len((((body or {}).get("hits") or {}).get("hits")) or [])775 return {776 **base_status,777 "connected": True,778 "source": "elasticsearch",779 "message": "Elasticsearch log search is active for this dashboard.",780 "sample_count": sample_count,781 }782 783 784def _fetch_observability_status() -> Dict[str, Any]:785 elastic_settings = _get_elasticsearch_settings()786 if _elasticsearch_enabled(elastic_settings):787 return _fetch_elasticsearch_status()788 789 datadog_status = _fetch_datadog_status()790 if datadog_status.get("configured"):791 return {792 **datadog_status,793 "backend": "datadog",794 }795 796 return {797 "backend": "local-fallback",798 "configured": False,799 "connected": False,800 "source": "local-fallback",801 "message": (802 "No external log provider configured. The dashboard will use locally replayed demo logs and simulator logs."803 if _has_local_demo_logs()804 else "No external log provider configured. The dashboard will use simulator logs."805 ),806 }807 808 809def _fetch_datadog_metrics(metric_query: str = "avg:system.cpu.user{*}", service: Optional[str] = None, points: int = 24, minutes: int = 15) -> Dict[str, Any]:810 api_key = os.getenv("DD_API_KEY", "").strip()811 app_key = os.getenv("DD_APP_KEY", "").strip()812 site = os.getenv("DD_SITE", "datadoghq.com").strip()813 814 if not api_key or not app_key:815 source_env = _get_source_env()816 fallback_series: List[Dict[str, Any]] = []817 if source_env is not None:818 summary = source_env._compute_metrics_summary()819 for svc, values in summary.items():820 if service and svc != service:821 continue822 fallback_series.append(823 {824 "service": svc,825 "latency_ms": values.get("latency_ms", 0.0),826 "error_rate": values.get("error_rate", 0.0),827 "cpu_percent": values.get("cpu_percent", 0.0),828 "memory_percent": values.get("memory_percent", 0.0),829 }830 )831 832 return {833 "source": "local-fallback",834 "query": metric_query,835 "series": fallback_series,836 "note": "DD_API_KEY or DD_APP_KEY not configured. Showing simulator metrics instead.",837 }838 839 effective_query = metric_query or "avg:system.cpu.user{*}"840 if service and "{*}" in effective_query:841 effective_query = effective_query.replace("{*}", f"{{service:{service}}}")842 843 to_ts = int(time.time())844 from_ts = to_ts - max(60, minutes * 60)845 response = requests.get(846 f"{_site_base_url(site)}/api/v1/query",847 headers={848 "DD-API-KEY": api_key,849 "DD-APPLICATION-KEY": app_key,850 },851 params={"from": from_ts, "to": to_ts, "query": effective_query},852 timeout=12,853 )854 855 if response.status_code >= 400:856 raise HTTPException(status_code=502, detail=f"Datadog metrics API error: {response.text}")857 858 body = response.json()859 series_items = body.get("series", [])860 parsed_series: List[Dict[str, Any]] = []861 for item in series_items:862 pointlist = item.get("pointlist", [])863 last_value = None864 for point in reversed(pointlist):865 if isinstance(point, list) and len(point) > 1 and point[1] is not None:866 last_value = point[1]867 break868 parsed_series.append(869 {870 "metric": item.get("metric"),871 "scope": item.get("scope"),872 "point_count": len(pointlist),873 "last_value": last_value,874 "display_name": item.get("display_name"),875 }876 )877 878 return {879 "source": "datadog",880 "query": effective_query,881 "series": parsed_series,882 }883 884 885def _fetch_datadog_apm(query: str = "service:*", service: Optional[str] = None, limit: int = 20, minutes: int = 15) -> Dict[str, Any]:886 api_key = os.getenv("DD_API_KEY", "").strip()887 app_key = os.getenv("DD_APP_KEY", "").strip()888 site = os.getenv("DD_SITE", "datadoghq.com").strip()889 890 search_query = _build_datadog_search_query(query, service, "service:*")891 892 if not api_key or not app_key:893 source_env = _get_source_env()894 fallback_traces: List[Dict[str, Any]] = []895 if source_env is not None:896 for action in source_env.actions_log[-max(1, min(limit, 100)):]:897 action_service = action.get("service") or "system"898 if service and action_service != service:899 continue900 fallback_traces.append(901 {902 "timestamp": action.get("step"),903 "service": action_service,904 "operation": action.get("action_type"),905 "resource": action.get("root_cause") or "incident-op",906 "duration_ms": 40 + (action.get("step", 0) % 5) * 10,907 }908 )909 910 if not fallback_traces:911 for log in source_env.all_logs[-max(1, min(limit, 100)):]:912 log_service = str(getattr(log.service, "value", log.service))913 if service and log_service != service:914 continue915 fallback_traces.append(916 {917 "timestamp": getattr(log, "timestamp", None),918 "service": log_service,919 "operation": "log_event",920 "resource": getattr(log, "message", "event")[:80],921 "duration_ms": 25,922 "trace_id": getattr(log, "trace_id", None),923 }924 )925 926 if not fallback_traces and service:927 for log in source_env.all_logs[-max(1, min(limit, 100)):]:928 fallback_traces.append(929 {930 "timestamp": getattr(log, "timestamp", None),931 "service": str(getattr(log.service, "value", log.service)),932 "operation": "log_event",933 "resource": getattr(log, "message", "event")[:80],934 "duration_ms": 25,935 "trace_id": getattr(log, "trace_id", None),936 }937 )938 939 return {940 "source": "local-fallback",941 "query": search_query,942 "traces": fallback_traces,943 "note": "DD_API_KEY or DD_APP_KEY not configured. Showing simulator operations as pseudo traces.",944 }945 946 payload = {947 "filter": {948 "query": search_query,949 "from": f"now-{max(1, minutes)}m",950 "to": "now",951 },952 "sort": "timestamp",953 "page": {954 "limit": max(1, min(limit, 100)),955 },956 }957 response = requests.post(958 f"{_site_base_url(site)}/api/v2/apm/events/search",959 headers={960 "DD-API-KEY": api_key,961 "DD-APPLICATION-KEY": app_key,962 "Content-Type": "application/json",963 },964 json=payload,965 timeout=12,966 )967 968 if response.status_code >= 400:969 raise HTTPException(status_code=502, detail=f"Datadog APM API error: {response.text}")970 971 body = response.json()972 rows = body.get("data", [])973 traces: List[Dict[str, Any]] = []974 for row in rows:975 attrs = row.get("attributes", {}) if isinstance(row, dict) else {}976 traces.append(977 {978 "timestamp": attrs.get("timestamp") or attrs.get("start_timestamp"),979 "service": attrs.get("service") or attrs.get("service_name"),980 "operation": attrs.get("operation_name") or attrs.get("name"),981 "resource": attrs.get("resource_name") or attrs.get("resource"),982 "duration_ms": attrs.get("duration") or attrs.get("duration_ms"),983 "trace_id": attrs.get("trace_id") or attrs.get("trace.id"),984 }985 )986 987 return {988 "source": "datadog",989 "query": search_query,990 "traces": traces,991 }992 993 994def _dashboard_html() -> str:995 return """<!DOCTYPE html>996<html lang="en">997<head>998 <meta charset="UTF-8" />999 <meta name="viewport" content="width=device-width, initial-scale=1.0" />1000 <title>Distributed Incident War Room</title>1001 <style>1002 @import url("https://fonts.googleapis.com/css2?family=JetBrains+Mono:wght@500;700&family=Manrope:wght@400;500;600;700;800&family=Sora:wght@500;600;700;800&display=swap");1003 1004 :root {1005 --bg: #061016;1006 --bg-soft: #0d1a21;1007 --panel: rgba(7, 14, 20, 0.82);1008 --panel-strong: rgba(5, 10, 16, 0.94);1009 --panel-tint: rgba(18, 32, 40, 0.84);1010 --line: rgba(133, 167, 182, 0.16);1011 --line-strong: rgba(133, 167, 182, 0.3);1012 --text: #f1f6f7;1013 --muted: #93a8b1;1014 --muted-strong: #c2d0d7;1015 --teal: #62f1d6;1016 --amber: #ffbf66;1017 --red: #ff6c72;1018 --green: #71f0a1;1019 --cyan: #6cbfff;1020 --violet: #7d8cff;1021 --shadow: 0 30px 80px rgba(0, 0, 0, 0.42);1022 --radius-xl: 28px;1023 --radius-lg: 22px;1024 --radius-md: 16px;1025 }1026 1027 * {1028 box-sizing: border-box;1029 }1030 1031 html {1032 scroll-behavior: smooth;1033 }1034 1035 body {1036 margin: 0;1037 min-height: 100vh;1038 font-family: "Manrope", "Segoe UI", sans-serif;1039 color: var(--text);1040 background:1041 radial-gradient(circle at top left, rgba(98, 241, 214, 0.2), transparent 30%),1042 radial-gradient(circle at 84% 12%, rgba(255, 191, 102, 0.15), transparent 24%),1043 radial-gradient(circle at 70% 88%, rgba(108, 191, 255, 0.12), transparent 28%),1044 radial-gradient(circle at 50% 50%, rgba(125, 140, 255, 0.06), transparent 42%),1045 linear-gradient(160deg, #040b10 0%, #09131a 46%, #050c12 100%);1046 overflow-x: hidden;1047 }1048 1049 body::before,1050 body::after {1051 content: "";1052 position: fixed;1053 inset: 0;1054 pointer-events: none;1055 }1056 1057 body::before {1058 background:1059 linear-gradient(rgba(255, 255, 255, 0.025) 1px, transparent 1px),1060 linear-gradient(90deg, rgba(255, 255, 255, 0.025) 1px, transparent 1px);1061 background-size: 56px 56px;1062 mask-image: linear-gradient(180deg, rgba(0, 0, 0, 0.8), transparent 100%);1063 opacity: 0.75;1064 }1065 1066 body::after {1067 background:1068 radial-gradient(circle at center, transparent 54%, rgba(0, 0, 0, 0.28) 100%),1069 linear-gradient(180deg, rgba(4, 8, 12, 0), rgba(4, 8, 12, 0.36));1070 }1071 1072 .shell {1073 position: relative;1074 z-index: 1;1075 width: min(1520px, calc(100vw - 28px));1076 margin: 0 auto;1077 padding: 24px 0 40px;1078 }1079 1080 .hero {1081 display: grid;1082 gap: 18px;1083 grid-template-columns: minmax(0, 1.45fr) minmax(340px, 0.85fr);1084 margin-bottom: 18px;1085 }1086 1087 .surface {1088 position: relative;1089 overflow: hidden;1090 border: 1px solid var(--line);1091 border-radius: var(--radius-xl);1092 background: linear-gradient(180deg, rgba(13, 24, 31, 0.92), rgba(7, 14, 20, 0.88));1093 box-shadow: var(--shadow);1094 backdrop-filter: blur(20px);1095 animation: rise 420ms ease both;1096 }1097 1098 .surface::before {1099 content: "";1100 position: absolute;1101 inset: 0;1102 background: linear-gradient(135deg, rgba(98, 241, 214, 0.08), transparent 42%, rgba(255, 191, 102, 0.06));1103 pointer-events: none;1104 }1105 1106 .surface::after {1107 content: "";1108 position: absolute;1109 inset: -40% auto -40% -30%;1110 width: 44%;1111 background: linear-gradient(90deg, transparent, rgba(255, 255, 255, 0.06), transparent);1112 transform: rotate(16deg);1113 animation: sheen 14s linear infinite;1114 pointer-events: none;1115 }1116 1117 .hero-card {1118 padding: 32px;1119 min-height: 240px;1120 display: flex;1121 flex-direction: column;1122 justify-content: space-between;1123 }1124 1125 .eyebrow {1126 display: inline-flex;1127 align-items: center;1128 gap: 10px;1129 color: var(--teal);1130 text-transform: uppercase;1131 letter-spacing: 0.22em;1132 font-size: 11px;1133 font-weight: 700;1134 font-family: "JetBrains Mono", monospace;1135 }1136 1137 .eyebrow::before {1138 content: "";1139 width: 26px;1140 height: 1px;1141 background: currentColor;1142 }1143 1144 h1,1145 h2,1146 h3 {1147 margin: 0;1148 font-family: "Sora", "Manrope", sans-serif;1149 letter-spacing: -0.03em;1150 }1151 1152 h1 {1153 margin-top: 18px;1154 max-width: 10ch;1155 font-size: clamp(38px, 6vw, 72px);1156 line-height: 0.92;1157 background: linear-gradient(135deg, #f5fbfc, #9beeff 44%, #ffd496 100%);1158 -webkit-background-clip: text;1159 color: transparent;1160 }1161 1162 .subtitle {1163 max-width: 60ch;1164 margin: 18px 0 0;1165 color: var(--muted-strong);1166 font-size: 15px;1167 line-height: 1.75;1168 }1169 1170 .hero-footer {1171 margin-top: 24px;1172 display: flex;1173 flex-wrap: wrap;1174 gap: 10px;1175 }1176 1177 .chip,1178 .badge {1179 display: inline-flex;1180 align-items: center;1181 gap: 8px;1182 min-height: 36px;1183 padding: 8px 12px;1184 border-radius: 999px;1185 border: 1px solid rgba(143, 174, 188, 0.18);1186 background: rgba(255, 255, 255, 0.04);1187 color: var(--muted);1188 font-size: 12px;1189 font-weight: 700;1190 backdrop-filter: blur(10px);1191 }1192 1193 .badge[data-tone="good"],1194 .source-card[data-tone="good"] strong,1195 .metric-card[data-tone="good"] .metric-value {1196 color: var(--green);1197 }1198 1199 .badge[data-tone="warn"],1200 .source-card[data-tone="warn"] strong,