"""Jira issue source preparation pipeline. Supported issue/alert types: - ALT-*: Alertstore operational alerts (fetched from the DevOps gateway, not Jira) - O11Y-*: Observability alerts (use alert extraction method with raw JSON in custom fields) - TCOC-*: TiDBCloud Oncall tickets (use alert extraction method with raw JSON in custom fields) - TCID-*: TiDBCloud Ticket (use ticket extraction method with direct custom fields) - ST-*: TiDBCloud Ticket for premium plan (use ticket extraction method with direct custom fields) Note: O11Y are alerts, while TCOC, TCID, and ST are tickets. TCOC uses the same extraction method as O11Y but represents a ticket type. Pipeline flow: 1. Fetch issue details from Jira to get cluster_id, tenant_id, org_id 2. Fetch cluster metadata from Clinic HTTP API 3. Optionally fetch analysis insights from Guardian API + S3 4. Fetch Jira comments, write comments.json; then call OpenAI to summarize status into comments_summary_ai.md """ import json import importlib.util import logging import os import subprocess import threading import base64 from datetime import datetime from pathlib import Path from typing import Optional import aioboto3 import httpx from investibook import config as app_config from investibook.core.source_types import normalize_issue_id from .config import settings logger = logging.getLogger(__name__) logger.setLevel(logging.INFO) RUNBOOKS_REPO_URL = "https://github.com/tidbcloud/runbooks" RUNBOOKS_BRANCH = "master" _RUNBOOKS_REPO_LOCK = threading.Lock() def _source_prepare_fixtures_root() -> Path | None: raw = os.environ.get("INVESTIBOOK_SOURCE_PREPARE_FIXTURES_DIR", "").strip() if raw: path = Path(raw) return path if path.exists() else None default = Path(__file__).resolve().parents[1] / "tests" / "fixtures" / "source_prepare" return default if default.exists() else None def _load_local_source_prepare_fixture(issue_id: str) -> tuple[dict | None, dict | None]: root = _source_prepare_fixtures_root() if root is None: return None, None fixture_dir = root / issue_id issue_meta_path = fixture_dir / "issue_meta.json" cluster_meta_path = fixture_dir / "cluster_meta.json" if not issue_meta_path.exists() or not cluster_meta_path.exists(): return None, None try: issue_meta = json.loads(issue_meta_path.read_text(encoding="utf-8")) cluster_meta = json.loads(cluster_meta_path.read_text(encoding="utf-8")) except (OSError, ValueError, json.JSONDecodeError): logger.warning("Invalid source-prepare fixture for %s under %s", issue_id, fixture_dir) return None, None if not isinstance(issue_meta, dict) or not isinstance(cluster_meta, dict): return None, None logger.info("Using local source-prepare fixture for %s from %s", issue_id, fixture_dir) return issue_meta, cluster_meta def _get_file_type(filename: str) -> str: """Map filename to frontend file type.""" if "insight" in filename: return "insight_report" elif "diagnostic" in filename: return "diagnostic" elif "cluster_meta" in filename: return "cluster_meta" elif "issue_meta" in filename: return "issue_meta" else: return "unknown" async def _fetch_alert_from_jira( alert_id: str, ) -> tuple[Optional[str], Optional[str], Optional[dict]]: """Fetch observability alert details from Jira and extract cluster_id, tenant_id, and full issue data. This function handles observability alerts (O11Y-*, TCOC-*) which use raw alert JSON in custom fields. - O11Y: Observability alerts - TCOC: TiDBCloud Oncall alerts Uses a two-tier extraction strategy: 1. Fast Path (Direct Custom Fields): - customfield_10149 → cluster_id - customfield_10157 → tenant_id If both are present, use them immediately and skip JSON parsing. 2. Fallback Path (Parse Raw Alert JSON from customfield_10160): - cluster_id: labels.tidb_cluster_id → labels.cluster_id → labels.name - tenant_id: labels.o11y_tenant_id → labels.tenant_id Cluster ID Processing: - If value starts with "tidb" prefix, it will be trimmed (e.g., "tidb123456" → "123456") Args: alert_id: Jira issue ID (e.g., "O11Y-1956041", "TCOC-12345") Returns: Tuple of (cluster_id, tenant_id, issue_dict) or (None, None, None) if failed """ if ( not settings.jira_base_url or not settings.jira_username or not settings.jira_api_token ): logger.warning("Jira not configured, skipping Jira API call") return None, None, None try: logger.info(f"Fetching alert details from Jira for {alert_id}") async with httpx.AsyncClient() as client: resp = await client.get( f"{settings.jira_base_url}/rest/api/2/issue/{alert_id}", auth=(settings.jira_username, settings.jira_api_token), timeout=30.0, ) resp.raise_for_status() issue = resp.json() # Extract cluster_id and tenant_id using two-tier strategy: # 1. Fast path: Try direct custom fields (customfield_10149, customfield_10157) # 2. Fallback: Parse raw alert JSON (customfield_10160) fields = issue.get("fields", {}) raw_alert_str = fields.get("customfield_10160") cluster_id_customfield = fields.get("customfield_10149") tenant_id_customfield = fields.get("customfield_10157") if cluster_id_customfield and tenant_id_customfield: return cluster_id_customfield, tenant_id_customfield, issue cluster_id = None tenant_id = None if raw_alert_str: try: raw_alert = json.loads(raw_alert_str) labels = raw_alert.get("labels", {}) # Try multiple fallbacks: tidb_cluster_id → cluster_id → name cluster_id = labels.get("tidb_cluster_id") or labels.get("cluster_id") or labels.get("name") # Trim "tidb" prefix if present (e.g., "tidb123456" -> "123456") if ( cluster_id and isinstance(cluster_id, str) and cluster_id.startswith("tidb") ): cluster_id = cluster_id[4:] # Remove "tidb" prefix # Try multiple fallbacks: o11y_tenant_id → tenant_id tenant_id = labels.get("o11y_tenant_id") or labels.get("tenant_id") except (json.JSONDecodeError, AttributeError, TypeError) as e: logger.warning( f"Failed to parse raw alert JSON from Jira issue {alert_id}: {e}" ) logger.info( f"Extracted from Jira alert {alert_id}: cluster_id={cluster_id}, tenant_id={tenant_id}" ) return cluster_id, tenant_id, issue except Exception as e: logger.warning(f"Failed to fetch alert from Jira: {e}") return None, None, None def _first_label(labels: dict, *keys: str) -> Optional[str]: """Return the first non-empty label value among ``keys``.""" for key in keys: value = labels.get(key) if value: return str(value) return None async def _fetch_alert_from_alertstore( alert_id: str, ) -> tuple[Optional[str], Optional[str], Optional[dict]]: """Fetch an alertstore alert (ALT-*) and normalize it into a Jira-issue-like dict. Alertstore is the SSOT for TiDB Cloud operational alerts. It shares the DevOps gateway, so we authenticate with ``devops_api_token`` (X-API-KEY header). The returned dict is shaped like a Jira issue so ``_build_issue_meta`` can consume it unchanged: the alert's label set is packed into ``customfield_10160`` (the same field O11Y alerts use) and top-level fields map to summary/status/created/updated. Returns: Tuple of (cluster_id, tenant_id, issue_like_dict) or (None, None, None) on failure. """ if not settings.devops_api_token: raise RuntimeError( "Alertstore credentials are missing: configure DEVOPS_API_KEY, " "DEVOPS_API_TOKEN, or INVESTIBOOK_DEVOPS_API_TOKEN" ) base_url = (settings.devops_base_url or "https://prod.devops.pingcap.com").rstrip("/") url = f"{base_url}/api/v1/alertstore/alerts/{alert_id}" try: logger.info(f"Fetching alert detail from alertstore for {alert_id}") async with httpx.AsyncClient() as client: resp = await client.get( url, headers={ "Accept": "application/json", "X-API-KEY": settings.devops_api_token, }, timeout=30.0, ) if resp.is_error: raise RuntimeError( f"Alertstore HTTP {resp.status_code} for {alert_id}: {resp.text[:500]}" ) detail = resp.json() or {} if not isinstance(detail, dict) or not detail: raise RuntimeError( f"Alertstore returned an empty or invalid alert for {alert_id}" ) attrs = detail.get("attrs") if isinstance(detail.get("attrs"), dict) else {} common_labels = ( detail.get("common_labels") if isinstance(detail.get("common_labels"), dict) else {} ) labels = {**attrs, **common_labels} cluster_id = ( detail.get("cluster_id") or _first_label(labels, "tidb_cluster_id", "cluster_id", "name") ) if isinstance(cluster_id, str) and cluster_id.startswith("tidb"): cluster_id = cluster_id[4:] # normalize "tidb123456" -> "123456" tenant_id = ( detail.get("tenant_id") or _first_label(labels, "o11y_tenant_id", "tenant_id") ) alert_name = detail.get("alertname") or _first_label(labels, "alertname") # Best-effort human message(s) for downstream alert-rule scoring. messages: list[str] = [] common_facts = detail.get("common_facts") message_val = _first_label(labels, "message", "summary", "description") if message_val: messages = [message_val] elif isinstance(common_facts, dict): fact_msg = common_facts.get("message") or common_facts.get("summary") if fact_msg: messages = [str(fact_msg)] # Repack into the O11Y-style raw alert JSON that _build_issue_meta parses. raw_alert = {"labels": labels, "messages": messages} summary = alert_name or detail.get("id") or alert_id issue_like = { "fields": { "summary": summary, "description": detail.get("external_issue_url", ""), "customfield_10160": json.dumps(raw_alert, ensure_ascii=False), "status": {"name": detail.get("state")} if detail.get("state") else None, "priority": {"name": detail.get("classification")} if detail.get("classification") else None, "created": detail.get("created_at"), "updated": detail.get("updated_at"), }, # Preserve the raw alertstore payload for callers that want it. "alertstore": detail, } logger.info( f"Extracted from alertstore {alert_id}: cluster_id={cluster_id}, tenant_id={tenant_id}" ) return cluster_id, tenant_id, issue_like except Exception: logger.exception("Failed to fetch alert from alertstore alert_id=%s", alert_id) raise async def _fetch_ticket_from_jira( ticket_id: str, ) -> tuple[Optional[str], Optional[str], Optional[dict]]: """Fetch TiDBCloud Ticket details from Jira and extract cluster_id, tenant_id, and full issue data. This function handles TiDBCloud Tickets (TCID-* / ST-*) which use direct custom fields for cluster and org metadata. Extraction Strategy: - cluster_id: from customfield_10088 (FIELD_TIDBCLOUD_CLUSTER_ID) - tenant_id: from customfield_10086 (FIELD_TIDBCLOUD_ORG_ID) Note: Asset fields (customfield_10110, customfield_10111) are not supported yet as they require JSM Asset API access which is not available in investibook. Args: ticket_id: Jira issue ID (e.g., "TCID-12345", "ST-12345") Returns: Tuple of (cluster_id, tenant_id, issue_dict) or (None, None, None) if failed """ if ( not settings.jira_base_url or not settings.jira_username or not settings.jira_api_token ): logger.warning("Jira not configured, skipping Jira API call") return None, None, None try: logger.info(f"Fetching ticket details from Jira for {ticket_id}") async with httpx.AsyncClient() as client: resp = await client.get( f"{settings.jira_base_url}/rest/api/2/issue/{ticket_id}", auth=(settings.jira_username, settings.jira_api_token), timeout=30.0, ) resp.raise_for_status() issue = resp.json() # Extract cluster_id and tenant_id from JSM ticket custom fields fields = issue.get("fields", {}) # JSM tickets use customfield_10088 for cluster_id and customfield_10086 for tenant_id cluster_id = fields.get("customfield_10088") tenant_id = fields.get("customfield_10086") # Convert to string if present if cluster_id: cluster_id = str(cluster_id) if tenant_id: tenant_id = str(tenant_id) logger.info( f"Extracted from Jira ticket {ticket_id}: cluster_id={cluster_id}, tenant_id={tenant_id}" ) return cluster_id, tenant_id, issue except Exception as e: logger.warning(f"Failed to fetch ticket from Jira: {e}") return None, None, None async def _fetch_comments_from_jira(issue_id: str) -> Optional[dict]: """Fetch all comments for a Jira issue from Jira API. Args: issue_id: Jira issue ID (e.g., "O11Y-1956041", "TCOC-12345", "TCID-67890", "ST-13579") Returns: Full Jira comments payload dict, or None if Jira is not configured or the call fails. """ if ( not settings.jira_base_url or not settings.jira_username or not settings.jira_api_token ): logger.warning("Jira not configured, skipping Jira comments API call") return None try: logger.info(f"Fetching comments from Jira for {issue_id}") async with httpx.AsyncClient() as client: resp = await client.get( f"{settings.jira_base_url}/rest/api/2/issue/{issue_id}/comment", auth=(settings.jira_username, settings.jira_api_token), timeout=30.0, ) resp.raise_for_status() comments = resp.json() logger.info( "Fetched %s comments for Jira issue %s", len(comments.get("comments", [])) if isinstance(comments, dict) else "unknown", issue_id, ) return comments except Exception as e: logger.warning(f"Failed to fetch comments from Jira for {issue_id}: {e}") return None async def _summarize_comments_with_openai(comments_payload: dict) -> Optional[str]: """Call OpenAI directly to summarize Jira comments into a status summary. Uses OpenAI's HTTP API via httpx (no extra SDK dependency). """ comments = comments_payload.get("comments", []) if isinstance(comments_payload, dict) else [] if not comments: return None # Build a compact plain-text representation of comments lines: list[str] = [] for c in comments: if not isinstance(c, dict): continue author = (c.get("author") or {}).get("displayName") or "unknown" created = c.get("created") or c.get("updated") or "" body = (c.get("body") or "").replace("\r", " ").replace("\n", " ") # Truncate each comment to avoid overly long prompts if len(body) > 600: body = body[:600] + "..." lines.append(f"[{created}] {author}: {body}") if not lines: return None comments_text = "\n".join(lines) openai_api_key = settings.openai_api_key if not openai_api_key: logger.warning("OpenAI API key (settings.openai_api_key) is empty, skipping LLM summary") return None api_url = "https://api.openai.com/v1/chat/completions" headers = { "Authorization": f"Bearer {openai_api_key}", "Content-Type": "application/json", } system_prompt = """You are an AIOps expert. Summarize Jira comments into a compact status summary for an on-call alert card. Requirements: 1. Output one or two concise sentences, not a timeline or per-comment list. 2. Do not include action items, recommendations, or risk assessment. 3. Only mention notification/escalation status, whether the alert recovered or is still active, whether comments include a human diagnosis/root cause, and the latest known status. 4. If comments only contain notifications or requests to check, say that no human diagnosis or clear root cause is present. 5. Do not invent facts beyond the comments. """ user_prompt = f"Here are the comments for a Jira alert: {comments_text}" payload = { "model": settings.openai_model or "gpt-4o", "messages": [ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}, ], "temperature": 0.2, "max_tokens": 800, } try: async with httpx.AsyncClient(timeout=60.0) as client: resp = await client.post(api_url, headers=headers, json=payload) resp.raise_for_status() data = resp.json() choice = (data.get("choices") or [{}])[0] message = (choice.get("message") or {}).get("content") if not isinstance(message, str): logger.warning("Unexpected OpenAI response format for comments summary") return None return message.strip() except Exception as e: logger.warning(f"Failed to call OpenAI for comments summary: {e}") return None def _build_comments_summary(comments_payload: dict) -> dict: """Build a lightweight summary from Jira comments payload. This is intentionally simple and deterministic so it can be used as a fallback or pre-processing step before AI summarization. """ comments = comments_payload.get("comments", []) if isinstance(comments_payload, dict) else [] total_comments = len(comments) authors = sorted( { (c.get("author") or {}).get("displayName") for c in comments if isinstance(c, dict) and c.get("author") and (c.get("author") or {}).get("displayName") } ) def _comment_ts(c: dict) -> str: return ( c.get("updated") or c.get("created") or "" ) last_comment = max(comments, key=_comment_ts, default=None) if comments else None last_comment_excerpt = None last_comment_author = None last_comment_timestamp = None if isinstance(last_comment, dict): body = last_comment.get("body") or "" # Simple truncation to keep summary compact last_comment_excerpt = body[:200] if len(body) > 200: last_comment_excerpt += "..." author = last_comment.get("author") or {} last_comment_author = author.get("displayName") last_comment_timestamp = last_comment.get("updated") or last_comment.get("created") return { "total_comments": total_comments, "authors": authors, "last_comment_author": last_comment_author, "last_comment_timestamp": last_comment_timestamp, "last_comment_excerpt": last_comment_excerpt, } def _format_fallback_comments_summary(comments_payload: dict) -> str: """Format a deterministic fallback when AI status summarization is unavailable.""" basic = _build_comments_summary(comments_payload) last_author = basic.get("last_comment_author") or "unknown" last_ts = basic.get("last_comment_timestamp") or "" last_excerpt = basic.get("last_comment_excerpt") or "" if last_ts: header = f"Latest Jira comment from {last_author} at {last_ts}:" else: header = f"Latest Jira comment from {last_author}:" body = last_excerpt or "(no comment body available)" return f"{header}\n\n{body}" def _build_issue_meta( issue_id: str, issue: dict, cluster_id: Optional[str], tenant_id: Optional[str] ) -> dict: """Build issue metadata from Jira issue data. Args: issue_id: Jira issue ID (e.g., "O11Y-1956041", "TCOC-12345", "TCID-67890", "ST-13579") issue: Full Jira issue dict cluster_id: Extracted cluster ID tenant_id: Extracted tenant ID Returns: Issue metadata dict for issue_meta.json """ fields = issue.get("fields", {}) # Parse raw alert data for additional details (for O11Y/TCOC types) raw_alert_str = fields.get("customfield_10160") alert_name = None component = None messages: list[str] = [] if raw_alert_str: try: raw_alert = json.loads(raw_alert_str) labels = raw_alert.get("labels", {}) alert_name = labels.get("alertname") component = labels.get("component") raw_messages = raw_alert.get("messages") if isinstance(raw_messages, str): text = raw_messages.strip() if text: messages = [text] elif isinstance(raw_messages, list): messages = [str(item).strip() for item in raw_messages if str(item).strip()] except (json.JSONDecodeError, AttributeError, TypeError): pass # Build issue metadata issue_meta = { "jira_issue_id": issue_id, "summary": fields.get("summary", ""), "description": fields.get("description", ""), "cluster_id": cluster_id, "tenant_id": tenant_id, "alert_name": alert_name, "component": component, "messages": messages, "rule": None, "rule_expr": None, "rule_for": None, "status": fields.get("status", {}).get("name") if isinstance(fields.get("status"), dict) else None, "priority": fields.get("priority", {}).get("name") if isinstance(fields.get("priority"), dict) else None, "created": fields.get("created"), "updated": fields.get("updated"), } return issue_meta def _load_alert_rule_lookup_module(): """Load the alert-rule lookup script as a module for backend-side reuse.""" candidates = [ app_config.AGENTKIT_SKILLS_DIR / "builtin" / "o11y-alert-rule-lookup" / "scripts" / "find_alert_rule.py", Path(__file__).resolve().parents[2] / "agentkit" / "skills" / "builtin" / "o11y-alert-rule-lookup" / "scripts" / "find_alert_rule.py", ] script_path = next((path for path in candidates if path.exists()), None) if script_path is None: searched = ", ".join(str(path) for path in candidates) raise FileNotFoundError( f"Alert rule lookup script not found in any known path: {searched}" ) spec = importlib.util.spec_from_file_location("o11y_alert_rule_lookup", script_path) if not spec or not spec.loader: raise ImportError(f"Unable to load module spec from {script_path}") module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module def _run_git_command(args: list[str], *, cwd: Path | None = None) -> None: env = os.environ.copy() env["GIT_TERMINAL_PROMPT"] = "0" token = (os.environ.get("SKILL_GITHUB_TOKEN") or "").strip() if token: basic = base64.b64encode(f"x-access-token:{token}".encode("utf-8")).decode("ascii") env["GIT_CONFIG_COUNT"] = "1" env["GIT_CONFIG_KEY_0"] = "http.https://github.com/.extraheader" env["GIT_CONFIG_VALUE_0"] = f"AUTHORIZATION: basic {basic}" completed = subprocess.run( args, cwd=str(cwd) if cwd else None, check=False, capture_output=True, env=env, text=True, timeout=120, ) if completed.returncode != 0: stderr = (completed.stderr or "").strip() stdout = (completed.stdout or "").strip() detail = stderr or stdout or f"exit code {completed.returncode}" raise RuntimeError(f"Git command failed: {' '.join(args)} :: {detail}") def _managed_runbooks_dir() -> Path: raw = os.environ.get("TIDBCLOUD_RUNBOOKS_DIR", "").strip() if raw: return Path(raw).expanduser().resolve() return (app_config.DATA_DIR / "runbooks").resolve() def _ensure_runbooks_checkout() -> Path: """Ensure a local runbooks checkout exists and is updated to origin/master.""" with _RUNBOOKS_REPO_LOCK: runbooks_dir = _managed_runbooks_dir() git_dir = runbooks_dir / ".git" if not git_dir.exists(): runbooks_dir.parent.mkdir(parents=True, exist_ok=True) logger.info("Cloning shared runbooks repository into %s", runbooks_dir) _run_git_command( [ "git", "clone", "--branch", RUNBOOKS_BRANCH, "--single-branch", RUNBOOKS_REPO_URL, str(runbooks_dir), ] ) logger.info("Refreshing shared runbooks repository at %s", runbooks_dir) _run_git_command(["git", "-C", str(runbooks_dir), "fetch", "origin", RUNBOOKS_BRANCH]) _run_git_command(["git", "-C", str(runbooks_dir), "checkout", RUNBOOKS_BRANCH]) _run_git_command(["git", "-C", str(runbooks_dir), "pull", "origin", RUNBOOKS_BRANCH]) rules_dir = runbooks_dir / "rules" if not rules_dir.is_dir(): raise FileNotFoundError(f"Runbooks checkout missing rules/ directory: {runbooks_dir}") return runbooks_dir def _lookup_alert_rule(issue_meta: dict) -> dict | None: """Best-effort resolve of the matching runbooks rule for an alert.""" alert_name = str(issue_meta.get("alert_name") or "").strip() if not alert_name: return None try: module = _load_alert_rule_lookup_module() result = module.search_rules( _ensure_runbooks_checkout(), alert_name, issue_meta.get("component"), issue_meta.get("messages") or [], 10, ) except FileNotFoundError as exc: logger.info("Alert rule lookup skipped: %s", exc) return None except RuntimeError as exc: logger.warning("Alert rule repository refresh failed for %s: %s", alert_name, exc) return None except Exception as exc: logger.warning("Alert rule lookup failed for %s: %s", alert_name, exc) return None top_match = result.get("top_match") if isinstance(top_match, dict): issue_meta["rule"] = top_match.get("file") issue_meta["rule_expr"] = top_match.get("expr") issue_meta["rule_for"] = top_match.get("for") return result async def _fetch_cluster_meta_from_clinic( cluster_id: str, tenant_id: Optional[str] ) -> Optional[dict]: """Fetch cluster metadata from Clinic HTTP API. Args: cluster_id: Cluster ID to fetch tenant_id: Tenant ID (optional, used to fill cluster_meta) Returns: Cluster metadata dict or None if failed """ # Resolve Clinic API key with same fallback chain as acp.py: # SKILL_CLINIC_API_KEY → CLINIC_API_KEY → INVESTIBOOK_CLINIC_API_KEY (settings) clinic_api_key = ( os.environ.get("SKILL_CLINIC_API_KEY") or os.environ.get("CLINIC_API_KEY") or settings.clinic_api_key or "" ).strip() has_api_key = bool(clinic_api_key) has_cookie_auth = bool(settings.clinic_cookie and settings.clinic_csrf_token) if not settings.clinic_endpoint or not (has_api_key or has_cookie_auth): logger.warning("Clinic API not configured, skipping cluster metadata fetch") return None try: logger.info(f"Fetching cluster metadata from Clinic API for {cluster_id}") async with httpx.AsyncClient() as client: # Use API key if available, otherwise use cookie + CSRF token headers = {} if has_api_key: headers["Authorization"] = f"Bearer {clinic_api_key}" logger.info("Using Clinic API key authentication") else: headers["Cookie"] = settings.clinic_cookie headers["X-CSRF-Token"] = settings.clinic_csrf_token logger.info("Using Clinic cookie + CSRF token authentication") resp = await client.get( f"{settings.clinic_endpoint}/clinic/api/v1/dashboard/clusters", headers=headers, params={ "cluster_id": cluster_id, "show_deleted": False, "limit": 1, "page": 1, }, timeout=30.0, ) resp.raise_for_status() result = resp.json() items = result.get("items", []) if not items: logger.warning(f"Cluster {cluster_id} not found in Clinic API") return None cluster = items[0] org_id = cluster.get("orgID", "") shoot_name = cluster.get("clusterShootName", "") # For Starter/Essential clusters, clusterShootName may be empty # in the List API — fall back to Get Cluster details API deploy_type_v2 = cluster.get("clusterDeployTypeV2", "") if not shoot_name and deploy_type_v2 in ("starter", "essential") and org_id: try: detail_resp = await client.get( f"{settings.clinic_endpoint}/clinic/api/v1/orgs/{org_id}/clusters/{cluster_id}", headers=headers, timeout=15.0, ) detail_resp.raise_for_status() detail = detail_resp.json() shoot_name = detail.get("shootName", "") logger.info(f"Got shootName from Get Cluster API: {shoot_name}") except Exception as e: logger.warning(f"Failed to fetch cluster detail for shootName: {e}") # For Starter/Essential clusters, discover keyspace_id from metrics keyspace_id = "" if deploy_type_v2 in ("starter", "essential"): try: series_resp = await client.get( f"{settings.clinic_endpoint}/data-proxy/metrics/{cluster_id}/api/v1/series", headers=headers, params={ "match[]": f'tidb_server_connections{{cluster_id="{cluster_id}"}}', }, timeout=15.0, ) series_resp.raise_for_status() series_data = series_resp.json().get("data", []) ks_ids = {s.get("keyspace_id", "") for s in series_data} - {""} if ks_ids: keyspace_id = sorted(ks_ids)[0] # pick first if multiple logger.info(f"Discovered keyspace_id={keyspace_id} for cluster {cluster_id}") except Exception as e: logger.warning(f"Failed to discover keyspace_id: {e}") # Build cluster metadata with comprehensive fields cluster_meta = { "tenant_id": tenant_id or cluster.get("tenantID", ""), "clinic_org_id": org_id, "cluster_id": cluster_id, "cluster_name": cluster.get("clusterName", ""), "tidb_version": cluster.get("clusterVersion", "unknown"), "cluster_type": cluster.get("clusterDeployType", ""), "cluster_type_v2": deploy_type_v2, "parent_id": cluster.get("parentID", ""), "shoot_name": shoot_name, "keyspace_id": keyspace_id, "project_id": cluster.get("projectID", ""), "cloud_provider": cluster.get("clusterProviderName", ""), "region": cluster.get("clusterRegionName", ""), "status": cluster.get("clusterStatus", ""), "create_timestamp": cluster.get("clusterCreatedAt", ""), } logger.info(f"Fetched cluster_meta: {cluster_meta}") return cluster_meta except Exception as e: logger.warning(f"Failed to fetch cluster metadata from Clinic: {e}") return None async def _fetch_analysis_from_guardian( issue_id: str, tenant_id: str, cluster_id: str, dest_dir: Path ) -> bool: """Optionally fetch analysis insights from Guardian API and S3. Args: issue_id: Jira issue ID (e.g., "O11Y-1956041", "TCOC-12345", "TCID-67890", "ST-13579") tenant_id: Tenant ID for S3 path cluster_id: Cluster ID for S3 path dest_dir: Destination directory to write files Returns: True if analysis was fetched successfully, False otherwise """ if not settings.aiops_api_url: logger.info("Guardian API not configured, skipping analysis fetch") return False try: logger.info(f"Fetching analysis for issue {issue_id} from Guardian API") async with httpx.AsyncClient() as client: resp = await client.get( f"{settings.aiops_api_url}/api/v1/aiops/guardian/analyses", params={"jira_issue_id": issue_id, "page_size": 50}, timeout=30.0, ) resp.raise_for_status() data = resp.json() items = data.get("items", []) if not items: logger.info(f"No analyses found for issue {issue_id}") return False # Filter for COMPLETED, sort by created_at desc completed = [a for a in items if a.get("status") == "COMPLETED"] candidates = completed if completed else items analysis = sorted( candidates, key=lambda x: x.get("created_at", ""), reverse=True )[0] logger.info( f"Selected analysis {analysis.get('analysis_uuid')} (status={analysis.get('status')})" ) # Extract S3 path components analysis_uuid = analysis["analysis_uuid"] created_at = datetime.fromisoformat( analysis["created_at"].replace("Z", "+00:00") ) # Construct S3 keys (UNDERSCORE separator!) date_str = created_at.strftime("%Y_%m") insight_key = f"guardian_service/analysis/{tenant_id}/{cluster_id}/{date_str}/{analysis_uuid}/insight.en.json" diagnostic_key = f"guardian_service/analysis/{tenant_id}/{cluster_id}/{date_str}/{analysis_uuid}/diagnostic_package.json" logger.info(f"S3 keys: insight={insight_key}, diagnostic={diagnostic_key}") # Download from S3 using aioboto3 logger.info(f"Downloading files from S3 bucket {settings.s3_bucket}") session = aioboto3.Session() async with session.client("s3", region_name=settings.aws_region) as s3: # Download insight.en.json → save as insight.json insight_obj = await s3.get_object( Bucket=settings.s3_bucket, Key=insight_key ) async with insight_obj["Body"] as stream: insight_data = await stream.read() (dest_dir / "insight.json").write_bytes(insight_data) logger.info(f"Downloaded insight.json ({len(insight_data)} bytes)") # Download diagnostic_package.json diag_obj = await s3.get_object( Bucket=settings.s3_bucket, Key=diagnostic_key ) async with diag_obj["Body"] as stream: diag_data = await stream.read() (dest_dir / "diagnostic_package.json").write_bytes(diag_data) logger.info(f"Downloaded diagnostic_package.json ({len(diag_data)} bytes)") return True except Exception as e: logger.warning(f"Failed to fetch analysis from Guardian/S3: {e}") return False async def prepare_issue_source( task_id: str, issue_id: str, workspace_id: str, prepare_tasks: dict, workspaces_dir: Path, templates_dir: Path, skip_jira_comments: bool = False, ) -> None: """Prepare Jira issue source data via new pipeline. Supports O11Y alerts, TCOC tickets, TCID tickets, and ST tickets. Note: O11Y are alerts; TCOC, TCID, and ST are tickets. New flow: 1. Fetch issue details from Jira (get cluster_id, tenant_id, org_id) - For O11Y/TCOC: use _fetch_alert_from_jira (raw JSON extraction) - For TCID: use _fetch_ticket_from_jira (direct custom fields) 2. Fetch cluster metadata from Clinic HTTP API 3. Optionally fetch analysis insights from Guardian API + S3 4. Fetch Jira comments and generate AI summary 5. Write cluster_meta.json and issue_meta.json 6. Build file list and update task dict (Note: .opencode/opencode.json is written at session-creation time by main.py) """ task = prepare_tasks[task_id] issue_id = normalize_issue_id(issue_id) try: workspace_dir = workspaces_dir / workspace_id workspace_dir.mkdir(parents=True, exist_ok=True) # Each issue gets its own subdirectory under issues/ dest_dir = workspace_dir / "issues" / issue_id dest_dir.mkdir(parents=True, exist_ok=True) # Step 1: Fetch issue details from Jira # Determine issue type based on prefix: # - O11Y-*: Observability alert (use _fetch_alert_from_jira) # - TCOC-*: TiDBCloud Oncall ticket (use _fetch_alert_from_jira, same extraction method) # - TCID-*: TiDBCloud Ticket (use _fetch_ticket_from_jira) # - ST-*: TiDBCloud Ticket for premium plan (use _fetch_ticket_from_jira) # - Other prefixes: Unsupported (fail immediately) is_alertstore_alert = issue_id.startswith("ALT-") if is_alertstore_alert: logger.info(f"Detected alertstore alert: {issue_id}") cluster_id, tenant_id, issue = await _fetch_alert_from_alertstore(issue_id) elif issue_id.startswith("O11Y-"): logger.info(f"Detected O11Y alert: {issue_id}") cluster_id, tenant_id, issue = await _fetch_alert_from_jira(issue_id) elif issue_id.startswith("TCOC-"): logger.info(f"Detected TCOC ticket: {issue_id}") cluster_id, tenant_id, issue = await _fetch_alert_from_jira(issue_id) elif issue_id.startswith("TCID-"): logger.info(f"Detected TCID ticket: {issue_id}") cluster_id, tenant_id, issue = await _fetch_ticket_from_jira(issue_id) elif issue_id.startswith("ST-"): logger.info(f"Detected ST ticket: {issue_id}") cluster_id, tenant_id, issue = await _fetch_ticket_from_jira(issue_id) else: logger.error( f"Unsupported issue prefix for {issue_id}. " f"Supported prefixes: ALT- (alertstore alert), O11Y- (alert), " f"TCOC- (ticket), TCID- (ticket), ST- (ticket)" ) raise ValueError( f"Unsupported issue type for {issue_id}. " f"Only ALT-*, O11Y-*, TCOC-*, TCID-*, and ST-* issues are supported." ) if is_alertstore_alert and not issue: raise RuntimeError(f"No source data returned for {issue_id}") # Build normalized issue metadata for the shared downstream pipeline. issue_meta = _build_issue_meta(issue_id, issue or {}, cluster_id, tenant_id) alertstore_detail = issue.get("alertstore") if is_alertstore_alert else None if is_alertstore_alert: attrs = ( alertstore_detail.get("attrs") if isinstance(alertstore_detail, dict) and isinstance(alertstore_detail.get("attrs"), dict) else {} ) issue_meta.update( { "source_type": "alertstore_alert", "owned": attrs.get("owned"), "external_issue_url": ( alertstore_detail.get("external_issue_url") if isinstance(alertstore_detail, dict) else None ), } ) # Step 2: Fetch cluster metadata from Clinic cluster_meta = None if cluster_id and tenant_id: cluster_meta = await _fetch_cluster_meta_from_clinic(cluster_id, tenant_id) if not cluster_meta: # Fallback to minimal cluster_meta if Clinic fetch failed logger.warning("Using minimal cluster_meta due to Clinic API failure") cluster_meta = { "tenant_id": tenant_id or "", "clinic_org_id": "", "cluster_id": cluster_id, "cluster_name": "", "tidb_version": "unknown", "cluster_type": "", "cluster_type_v2": "", "parent_id": "", "shoot_name": "", "keyspace_id": "", "project_id": "", "cloud_provider": "", "region": "", "status": "", "create_timestamp": "", } if not issue_meta.get("cluster_id") or not issue_meta.get("created"): fixture_issue_meta, fixture_cluster_meta = _load_local_source_prepare_fixture(issue_id) if fixture_issue_meta and fixture_cluster_meta: issue_meta = { **fixture_issue_meta, "jira_issue_id": issue_id, "summary": issue_meta.get("summary") or fixture_issue_meta.get("summary", ""), "description": issue_meta.get("description") or fixture_issue_meta.get("description", ""), "status": issue_meta.get("status") or fixture_issue_meta.get("status"), "priority": issue_meta.get("priority") or fixture_issue_meta.get("priority"), "cluster_id": issue_meta.get("cluster_id") or fixture_issue_meta.get("cluster_id"), "tenant_id": issue_meta.get("tenant_id") or fixture_issue_meta.get("tenant_id"), "created": issue_meta.get("created") or fixture_issue_meta.get("created"), "updated": issue_meta.get("updated") or fixture_issue_meta.get("updated"), } cluster_meta = { **fixture_cluster_meta, "cluster_id": cluster_meta.get("cluster_id") or fixture_cluster_meta.get("cluster_id"), "tenant_id": cluster_meta.get("tenant_id") or fixture_cluster_meta.get("tenant_id"), "clinic_org_id": cluster_meta.get("clinic_org_id") or fixture_cluster_meta.get("clinic_org_id"), "cluster_name": cluster_meta.get("cluster_name") or fixture_cluster_meta.get("cluster_name"), "cluster_type": cluster_meta.get("cluster_type") or fixture_cluster_meta.get("cluster_type"), "cluster_type_v2": cluster_meta.get("cluster_type_v2") or fixture_cluster_meta.get("cluster_type_v2"), "parent_id": cluster_meta.get("parent_id") or fixture_cluster_meta.get("parent_id"), "shoot_name": cluster_meta.get("shoot_name") or fixture_cluster_meta.get("shoot_name"), "keyspace_id": cluster_meta.get("keyspace_id") or fixture_cluster_meta.get("keyspace_id"), "project_id": cluster_meta.get("project_id") or fixture_cluster_meta.get("project_id"), "cloud_provider": cluster_meta.get("cloud_provider") or fixture_cluster_meta.get("cloud_provider"), "region": cluster_meta.get("region") or fixture_cluster_meta.get("region"), "status": cluster_meta.get("status") or fixture_cluster_meta.get("status"), "create_timestamp": cluster_meta.get("create_timestamp") or fixture_cluster_meta.get("create_timestamp"), } # Step 3: Optionally fetch analysis insights from Guardian + S3 if tenant_id and cluster_id: await _fetch_analysis_from_guardian( issue_id, tenant_id, cluster_id, dest_dir ) else: logger.info( "Skipping analysis fetch: missing tenant_id or cluster_id from Jira" ) # Best-effort alert rule lookup for O11Y/TCOC alerts. alert_rule_match = _lookup_alert_rule(issue_meta) if alert_rule_match and isinstance(alert_rule_match.get("top_match"), dict): top_match = alert_rule_match["top_match"] issue_meta["rule"] = top_match.get("file") issue_meta["rule_expr"] = top_match.get("expr") issue_meta["rule_for"] = top_match.get("for") # Step 4: Write cluster_meta.json and issue_meta.json (dest_dir / "cluster_meta.json").write_text( json.dumps(cluster_meta, indent=2, ensure_ascii=False), encoding="utf-8", ) logger.info("Wrote cluster_meta.json") (dest_dir / "issue_meta.json").write_text( json.dumps(issue_meta, indent=2, ensure_ascii=False), encoding="utf-8", ) logger.info("Wrote issue_meta.json") if isinstance(alertstore_detail, dict): (dest_dir / "alertstore_alert.json").write_text( json.dumps(alertstore_detail, indent=2, ensure_ascii=False), encoding="utf-8", ) logger.info("Wrote alertstore_alert.json") if alert_rule_match: (dest_dir / "alert_rule_match.json").write_text( json.dumps(alert_rule_match, indent=2, ensure_ascii=False), encoding="utf-8", ) logger.info("Wrote alert_rule_match.json") # Step 5: Fetch comments from Jira and write comments.json + comments_summary_ai.md if is_alertstore_alert or skip_jira_comments or settings.skip_jira_comments: # Re-preparing an existing workspace must not leave previously fetched # comments behind when the request now excludes them. for filename in ("comments.json", "comments_summary_ai.md"): (dest_dir / filename).unlink(missing_ok=True) logger.info( "Skipping Jira comments fetch (Alertstore source, request, or server policy)" ) comments_payload = None else: comments_payload = await _fetch_comments_from_jira(issue_id) if comments_payload: comments_json_path = dest_dir / "comments.json" comments_json_path.write_text(json.dumps(comments_payload, indent=2)) logger.info("Wrote comments.json") # Try direct LLM status summary via OpenAI summary_md_path = dest_dir / "comments_summary_ai.md" summary_text = await _summarize_comments_with_openai(comments_payload) if summary_text: summary_md_path.write_text(summary_text) logger.info("Wrote comments_summary_ai.md via OpenAI status summary") else: # Fallback to a deterministic latest-comment excerpt if LLM fails. summary_md_path.write_text(_format_fallback_comments_summary(comments_payload)) logger.info("Wrote comments_summary_ai.md via fallback summary builder (last comment only)") # Note: .opencode/opencode.json is written by _write_workspace_opencode_json() # at session-creation time (main.py) using AGENTS_DIR absolute paths. # We no longer copy .opencode/ or toolset/ from the template here. # Step 6: Build file list and update task dict copied_files = [] for fpath in dest_dir.rglob("*"): if fpath.is_file(): # Paths are relative to the workspace root so frontend can fetch them rel = str(fpath.relative_to(workspace_dir)) copied_files.append( { "path": rel, "type": _get_file_type(rel), "size": fpath.stat().st_size, } ) logger.info( f"Prepared {len(copied_files)} files for workspace {workspace_id} (issue dir: issues/{issue_id})" ) task["status"] = "completed" task["files"] = copied_files task["cluster_meta"] = cluster_meta task["issue_dir"] = f"issues/{issue_id}" task["error"] = None except Exception as e: logger.exception(f"Error preparing issue source {issue_id}: {e}") task["status"] = "error" task["error"] = str(e) # Backward compatibility alias prepare_alert_source = prepare_issue_source