1342 lines
59 KiB
Python
1342 lines
59 KiB
Python
"""Periodic OKE remediator that runs three independent kagent checks.
|
|
|
|
The job performs:
|
|
1. Business application health validation for Orders.
|
|
2. HPA health validation for order-service.
|
|
3. ImagePull health validation for application deployments.
|
|
|
|
Each check is sent to kagent as an independent MCP task. The returned JSON is
|
|
normalized into a stable machine-readable shape so the job can:
|
|
- decide whether the issue is active,
|
|
- avoid duplicate notifications for the same active incident,
|
|
- trigger the matching OCI DevOps remediation deployment,
|
|
- send one email report with the approval link for that remediation.
|
|
"""
|
|
|
|
import asyncio
|
|
import ast
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
import traceback
|
|
import urllib.request
|
|
import uuid
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Any, Dict, List
|
|
from urllib.error import HTTPError
|
|
|
|
import oci
|
|
from mcp import ClientSession
|
|
|
|
try:
|
|
from mcp.client.streamable_http import streamable_http_client as streamablehttp_client
|
|
except ImportError:
|
|
from mcp.client.streamable_http import streamablehttp_client
|
|
|
|
|
|
STATE_CONFIGMAP_NAME = "kagent-oke-remediator-multi-state"
|
|
WORKLOAD_NAME = "kagent-oke-remediator-multi"
|
|
HEALTHY_STATUSES = {
|
|
"HEALTHY",
|
|
"OK",
|
|
"COMPLETED",
|
|
"ALL SYSTEMS OPERATIONAL",
|
|
"NO ACTIVE AUTOSCALING",
|
|
"NO ACTIVE IMAGE_PULL_FAILURE DETECTED",
|
|
"NO FAILURE",
|
|
"NO ISSUE DETECTED",
|
|
"NO HAY ERRORES DE OBTENCIÓN DE IMÁGENES.",
|
|
"NO HAY ERRORES DE OBTENCION DE IMAGENES.",
|
|
}
|
|
IMAGE_PULL_WAITING_REASONS = {
|
|
"ERRIMAGEPULL",
|
|
"IMAGEPULLBACKOFF",
|
|
"INVALIDIMAGENAME",
|
|
"CREATECONTAINERCONFIGERROR",
|
|
}
|
|
IMAGE_PULL_EVENT_REASONS = {
|
|
"FAILED",
|
|
"BACKOFF",
|
|
"INSPECTFAILED",
|
|
}
|
|
INTERNAL_FAILURE_TYPES = {
|
|
"AGENT_RESPONSE_PARSE_FAILURE",
|
|
"AGENT_INPUT_REQUIRED",
|
|
"JOB_RUNTIME_FAILURE",
|
|
}
|
|
|
|
|
|
# Structured logging is used heavily because this job is executed as a CronJob.
|
|
# The easiest way to understand behavior later is by reading pod logs, so every
|
|
# major phase emits a compact JSON event.
|
|
def _log_event(event: str, fields: Dict[str, Any]) -> None:
|
|
"""Emit a single structured log event for easy scraping from pod logs."""
|
|
print(json.dumps({"event": event, **fields}, ensure_ascii=False), flush=True)
|
|
|
|
|
|
def main() -> int:
|
|
"""Entrypoint used by the container process."""
|
|
app_namespace = os.environ.get("APP_NAMESPACE", "kagent-demo")
|
|
region = os.environ.get("OCI_REGION", "mx-monterrey-1")
|
|
kagent_url = os.environ.get("KAGENT_MCP_URL", "")
|
|
_log_event(
|
|
"KAGENT_MULTI_START",
|
|
{"namespace": app_namespace, "region": region, "kagentMcpUrl": kagent_url},
|
|
)
|
|
try:
|
|
result = asyncio.run(run_multi_checks())
|
|
_log_event(
|
|
"KAGENT_MULTI_RESULT",
|
|
{
|
|
"namespace": app_namespace,
|
|
"resultsCount": len(result.get("results", [])),
|
|
"notificationsSent": len(result.get("notificationsSent", [])),
|
|
"remediationsTriggered": len(result.get("remediationsTriggered", [])),
|
|
},
|
|
)
|
|
print(json.dumps(result, indent=2, ensure_ascii=False), flush=True)
|
|
return 0
|
|
except Exception as exc:
|
|
error = {
|
|
"status": "FAILED",
|
|
"failureType": "JOB_RUNTIME_FAILURE",
|
|
"rootCause": _exception_summary(exc),
|
|
"traceback": "".join(traceback.format_exception(exc)),
|
|
}
|
|
print(json.dumps(error, indent=2, ensure_ascii=False), flush=True)
|
|
return 1
|
|
|
|
|
|
def _fetch_json(url: str) -> Dict[str, Any]:
|
|
"""Fetch the Orders observability payload used by the business check."""
|
|
if not url:
|
|
return {"fetchStatus": "SKIPPED", "reason": "observabilityUrl is empty"}
|
|
|
|
try:
|
|
with urllib.request.urlopen(url, timeout=10) as response:
|
|
body = response.read(20000).decode("utf-8")
|
|
return json.loads(body)
|
|
except Exception as exc:
|
|
return {
|
|
"fetchStatus": "FAILED",
|
|
"error": _exception_summary(exc),
|
|
}
|
|
|
|
|
|
async def run_multi_checks() -> Dict[str, Any]:
|
|
"""Run the three checks, deduplicate incidents, trigger remediation, send mail."""
|
|
payload = _build_payload_from_env()
|
|
results: List[Dict[str, Any]] = []
|
|
notifications_sent: List[str] = []
|
|
remediations_triggered: List[Dict[str, Any]] = []
|
|
|
|
# Each prompt is executed as an independent MCP interaction so kagent treats
|
|
# them like separate conversations/checks instead of sharing one context.
|
|
for check in _build_checks(payload):
|
|
result = await _run_single_check(payload, check)
|
|
results.append(result)
|
|
|
|
# Persist one state entry per error_code so a still-open incident does not
|
|
# create another remediation deployment or another email every 3 minutes.
|
|
state = _load_notification_state(payload["appNamespace"])
|
|
for result in results:
|
|
result = _normalize_result(result, payload["appNamespace"])
|
|
error_code = result.get("error_code", "")
|
|
|
|
# When a check returns healthy again we clear only that incident family.
|
|
# This allows the same issue to be detected as "new" in the future.
|
|
if not _is_active_failure(result):
|
|
_clear_state_for_code(payload["appNamespace"], state, error_code)
|
|
continue
|
|
|
|
if _is_internal_failure(result):
|
|
_log_event(
|
|
"KAGENT_MULTI_INTERNAL_FAILURE",
|
|
{
|
|
"check": result.get("checkName", ""),
|
|
"errorCode": error_code,
|
|
"failureType": result.get("failureType", ""),
|
|
"rootCause": result.get("rootCause", ""),
|
|
},
|
|
)
|
|
_clear_state_for_code(payload["appNamespace"], state, error_code)
|
|
result["emailSent"] = False
|
|
result["notificationReason"] = "Internal agent/runtime failure; notification suppressed."
|
|
_replace_result(results, result)
|
|
continue
|
|
|
|
signature = _build_signature(result)
|
|
state_entry = state.get(error_code, {})
|
|
active_url = state_entry.get("deploymentUrl", "")
|
|
active_signature = state_entry.get("signature", "")
|
|
active_email = state_entry.get("emailSent", "")
|
|
|
|
# If the same incident is still active and we already have a remediation
|
|
# deployment link for it, suppress new emails and new deployments.
|
|
if signature == active_signature and active_url and active_email == "true":
|
|
result["emailSent"] = False
|
|
result["notificationReason"] = "Duplicate active issue; notification suppressed."
|
|
result["remediationDeployment"] = {
|
|
"triggered": True,
|
|
"deploymentUrl": active_url,
|
|
"deploymentId": state_entry.get("deploymentId", ""),
|
|
"reused": True,
|
|
}
|
|
continue
|
|
|
|
remediation = _safe_trigger_remediation(result, payload)
|
|
if remediation:
|
|
result["remediationDeployment"] = remediation
|
|
if remediation.get("triggered"):
|
|
remediations_triggered.append(remediation)
|
|
|
|
email_sent, notification_reason = _safe_send_email(result, payload)
|
|
result["emailSent"] = email_sent
|
|
result["notificationReason"] = notification_reason
|
|
if email_sent:
|
|
notifications_sent.append(error_code)
|
|
|
|
# Persist state only after this check has finished its notification/
|
|
# remediation cycle so future runs know whether the incident is new.
|
|
if email_sent and (result.get("remediationDeployment") or {}).get("deploymentUrl"):
|
|
state[error_code] = {
|
|
"signature": signature,
|
|
"deploymentUrl": result["remediationDeployment"].get("deploymentUrl", ""),
|
|
"deploymentId": result["remediationDeployment"].get("deploymentId", ""),
|
|
"emailSent": "true",
|
|
}
|
|
else:
|
|
state[error_code] = {
|
|
"signature": "",
|
|
"deploymentUrl": "",
|
|
"deploymentId": "",
|
|
"emailSent": "false",
|
|
}
|
|
|
|
_save_notification_state(payload["appNamespace"], state)
|
|
return {
|
|
"status": "COMPLETED",
|
|
"results": results,
|
|
"notificationsSent": notifications_sent,
|
|
"remediationsTriggered": remediations_triggered,
|
|
}
|
|
|
|
|
|
def _replace_result(results: List[Dict[str, Any]], updated: Dict[str, Any]) -> None:
|
|
"""Replace the stored result for one check without changing list order."""
|
|
for idx, item in enumerate(results):
|
|
if item.get("checkName") == updated.get("checkName"):
|
|
results[idx] = updated
|
|
return
|
|
|
|
|
|
def _build_payload_from_env() -> Dict[str, Any]:
|
|
"""Collect all runtime settings from environment variables."""
|
|
return {
|
|
"region": os.environ.get("OCI_REGION", "mx-monterrey-1"),
|
|
"appNamespace": os.environ.get("APP_NAMESPACE", "kagent-demo"),
|
|
"kagentMcpUrl": os.environ.get("KAGENT_MCP_URL", ""),
|
|
"agentName": os.environ.get("KAGENT_AGENT_NAME", "k8s-agent"),
|
|
"agentNamespace": os.environ.get("KAGENT_AGENT_NAMESPACE", "kagent"),
|
|
"observabilityUrl": os.environ.get("ORDER_OBSERVABILITY_URL", ""),
|
|
"notificationTopicOcid": os.environ.get("NOTIFICATION_TOPIC_OCID", ""),
|
|
"remediationPipelines": {
|
|
"business": {
|
|
"projectOcid": os.environ.get("BUSINESS_REMEDIATION_PROJECT_OCID", ""),
|
|
"pipelineOcid": os.environ.get("BUSINESS_REMEDIATION_PIPELINE_OCID", ""),
|
|
},
|
|
"imagePull": {
|
|
"projectOcid": os.environ.get("IMAGEPULL_REMEDIATION_PROJECT_OCID", ""),
|
|
"pipelineOcid": os.environ.get("IMAGEPULL_REMEDIATION_PIPELINE_OCID", ""),
|
|
},
|
|
"hpa": {
|
|
"projectOcid": os.environ.get("HPA_REMEDIATION_PROJECT_OCID", ""),
|
|
"pipelineOcid": os.environ.get("HPA_REMEDIATION_PIPELINE_OCID", ""),
|
|
},
|
|
},
|
|
}
|
|
|
|
|
|
async def _run_single_check(payload: Dict[str, Any], check: Dict[str, str]) -> Dict[str, Any]:
|
|
"""Open a fresh MCP interaction for one check and return the parsed payload."""
|
|
context_id = f"{check['name']}-{uuid.uuid4()}"
|
|
_log_event("KAGENT_MULTI_CHECK_START", {"check": check["name"], "contextId": context_id})
|
|
async with streamablehttp_client(payload["kagentMcpUrl"]) as streams:
|
|
read_stream, write_stream = streams[:2]
|
|
async with ClientSession(read_stream, write_stream) as session:
|
|
await session.initialize()
|
|
response = await session.call_tool(
|
|
"invoke_agent",
|
|
{
|
|
"agent": f"{payload['agentNamespace']}/{payload['agentName']}",
|
|
"task": check["prompt"],
|
|
"context_id": context_id,
|
|
},
|
|
)
|
|
|
|
text = _extract_text(response)
|
|
result = _parse_json_response(text)
|
|
result["checkName"] = check["name"]
|
|
result.setdefault("rawResponse", text)
|
|
_log_event(
|
|
"KAGENT_MULTI_CHECK_RESULT",
|
|
{
|
|
"check": check["name"],
|
|
"status": result.get("status", ""),
|
|
"failureType": result.get("failureType", ""),
|
|
"errorCode": result.get("error_code", ""),
|
|
},
|
|
)
|
|
return result
|
|
|
|
|
|
def _extract_text(response: Any) -> str:
|
|
"""Flatten MCP tool response content into one text block."""
|
|
parts = []
|
|
for item in getattr(response, "content", []) or []:
|
|
text = getattr(item, "text", None)
|
|
if text:
|
|
parts.append(text)
|
|
return "\n".join(parts) if parts else str(response)
|
|
|
|
|
|
def _parse_json_response(text: str) -> Dict[str, Any]:
|
|
"""Extract and parse the first JSON object returned by kagent."""
|
|
candidate = _extract_json_candidate(text)
|
|
if not candidate:
|
|
return {
|
|
"status": "FAILED",
|
|
"failureType": "AGENT_RESPONSE_PARSE_FAILURE",
|
|
"rootCause": "kagent did not return JSON",
|
|
"rawResponse": text,
|
|
}
|
|
try:
|
|
return json.loads(candidate)
|
|
except json.JSONDecodeError:
|
|
parsed = ast.literal_eval(candidate)
|
|
if not isinstance(parsed, dict):
|
|
raise ValueError("Parsed agent payload is not a JSON object")
|
|
return parsed
|
|
|
|
|
|
def _extract_json_candidate(text: str) -> str:
|
|
"""Prefer fenced ```json blocks; otherwise take the outermost object."""
|
|
fenced_match = re.search(r"```json\s*(\{.*?\})\s*```", text, re.DOTALL | re.IGNORECASE)
|
|
if fenced_match:
|
|
return fenced_match.group(1)
|
|
start = text.find("{")
|
|
end = text.rfind("}")
|
|
if start == -1 or end == -1 or end <= start:
|
|
return ""
|
|
return text[start : end + 1]
|
|
|
|
|
|
def _normalize_result(result: Dict[str, Any], app_namespace: str) -> Dict[str, Any]:
|
|
"""Normalize the heterogeneous kagent responses into a stable contract."""
|
|
normalized = dict(result)
|
|
normalized.setdefault("namespace", app_namespace)
|
|
|
|
# First normalize the "incident family" so downstream logic knows which
|
|
# remediation pipeline belongs to this result.
|
|
normalized["error_code"] = _normalize_error_code(normalized)
|
|
|
|
# Then normalize health/failure semantics so prose like "No Failure" or
|
|
# "No issue detected" is converted to a machine-usable state.
|
|
normalized["status"] = _normalize_status(normalized)
|
|
normalized["failureType"] = _normalize_failure_type(normalized)
|
|
|
|
# The report is what goes to the email body. We flatten dict-shaped evidence
|
|
# into a list so the email renderer can print one bullet per line.
|
|
if isinstance(normalized.get("report"), dict):
|
|
report = dict(normalized["report"])
|
|
if isinstance(report.get("evidence"), dict):
|
|
report["evidence"] = [f"{key}: {value}" for key, value in report["evidence"].items()]
|
|
normalized["report"] = report
|
|
if normalized["error_code"] == "BUSINESS_APPLICATION" and "pending_transactions" not in normalized:
|
|
normalized["pending_transactions"] = _extract_pending_transactions(normalized)
|
|
if normalized["error_code"] == "BUSINESS_APPLICATION":
|
|
normalized = _reconcile_business_with_observability(normalized)
|
|
if normalized["error_code"] == "IMAGE_PULL":
|
|
normalized = _reconcile_imagepull_with_live_state(normalized, app_namespace)
|
|
return normalized
|
|
|
|
|
|
def _reconcile_business_with_observability(result: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Suppress business hallucinations when observability clearly says the business flow is healthy."""
|
|
observability_url = os.environ.get("ORDER_OBSERVABILITY_URL", "")
|
|
observability_json = _compact_observability_json(_fetch_json(observability_url))
|
|
business = observability_json.get("businessHealth", {})
|
|
services = observability_json.get("services", [])
|
|
|
|
healthy = bool(business.get("healthy", True))
|
|
total_orders = _to_number(business.get("totalOrders", 0))
|
|
pending_orders = _to_number(business.get("pendingReviewOrders", 0))
|
|
failed_orders = _to_number(business.get("failedOrders", 0))
|
|
manual_review_rate = _to_number(business.get("manualReviewRate", 0))
|
|
revenue_at_risk = _to_number(business.get("revenueAtRisk", 0))
|
|
degraded_services = [
|
|
service.get("name", "")
|
|
for service in services
|
|
if isinstance(service, dict)
|
|
and str(service.get("status", "")).strip().upper() == "DEGRADED"
|
|
and bool(service.get("kagentSignal", False))
|
|
]
|
|
|
|
if healthy and total_orders == 0 and pending_orders == 0 and failed_orders == 0 and manual_review_rate == 0 and revenue_at_risk == 0 and not degraded_services:
|
|
normalized = dict(result)
|
|
normalized["status"] = "HEALTHY"
|
|
normalized["failureType"] = "NO_ACTIVE_FAILURE"
|
|
normalized["pending_transactions"] = 0
|
|
normalized["summary"] = "No hay degradación activa del proceso de negocio según la observabilidad actual."
|
|
normalized["rootCause"] = "La observabilidad actual muestra flujo de negocio saludable y sin órdenes pendientes, fallidas o revenue en riesgo."
|
|
normalized["businessImpact"] = ""
|
|
normalized["recommendedAction"] = "No se requiere remediación de negocio. Continúe monitoreando señales de runtime y negocio."
|
|
normalized["evidence"] = [
|
|
f"businessHealth.healthy={healthy}",
|
|
f"totalOrders={total_orders}",
|
|
f"pendingReviewOrders={pending_orders}",
|
|
f"failedOrders={failed_orders}",
|
|
f"manualReviewRate={manual_review_rate}",
|
|
f"revenueAtRisk={revenue_at_risk}",
|
|
"degradedBusinessServices=none",
|
|
]
|
|
normalized["report"] = {
|
|
"executiveSummary": "La observabilidad actual confirma que no existe degradación activa del proceso de negocio.",
|
|
"rootCause": normalized["rootCause"],
|
|
"businessImpact": "",
|
|
"recommendedAction": normalized["recommendedAction"],
|
|
"evidence": list(normalized["evidence"]),
|
|
"recommendedConfigMapData": {},
|
|
}
|
|
return normalized
|
|
|
|
return result
|
|
|
|
|
|
def _to_number(value: Any) -> float:
|
|
try:
|
|
return float(value)
|
|
except (TypeError, ValueError):
|
|
return 0.0
|
|
|
|
|
|
def _reconcile_imagepull_with_live_state(result: Dict[str, Any], namespace: str) -> Dict[str, Any]:
|
|
"""Ignore historical image-pull incidents when no current pod has an active pull error."""
|
|
snapshot = _get_live_imagepull_snapshot(namespace)
|
|
if snapshot["active"]:
|
|
if _is_model_runtime_failure(result):
|
|
return _build_live_imagepull_result(result, snapshot, namespace)
|
|
evidence = result.get("evidence", [])
|
|
if not isinstance(evidence, list):
|
|
evidence = [str(evidence)] if evidence else []
|
|
for line in snapshot.get("evidence", []):
|
|
if line not in evidence:
|
|
evidence.append(line)
|
|
result["evidence"] = evidence
|
|
return result
|
|
|
|
normalized = dict(result)
|
|
normalized["status"] = "HEALTHY"
|
|
normalized["failureType"] = "NO_ACTIVE_FAILURE"
|
|
normalized["rootCause"] = snapshot["reason"]
|
|
normalized["businessImpact"] = ""
|
|
evidence = normalized.get("evidence", [])
|
|
if not isinstance(evidence, list):
|
|
evidence = [str(evidence)] if evidence else []
|
|
for line in snapshot.get("evidence", []):
|
|
if line not in evidence:
|
|
evidence.append(line)
|
|
normalized["evidence"] = evidence
|
|
report = normalized.get("report", {})
|
|
if isinstance(report, dict):
|
|
report = dict(report)
|
|
report["rootCause"] = normalized["rootCause"]
|
|
report["businessImpact"] = ""
|
|
report["recommendedAction"] = "No remediation required for image pull. Current pods are not failing due to image extraction."
|
|
report_evidence = report.get("evidence", [])
|
|
if not isinstance(report_evidence, list):
|
|
report_evidence = [str(report_evidence)] if report_evidence else []
|
|
for line in snapshot.get("evidence", []):
|
|
if line not in report_evidence:
|
|
report_evidence.append(line)
|
|
report["evidence"] = report_evidence
|
|
normalized["report"] = report
|
|
normalized["recommendedAction"] = "No remediation required for image pull. Current pods are not failing due to image extraction."
|
|
return normalized
|
|
|
|
|
|
def _build_live_imagepull_result(result: Dict[str, Any], snapshot: Dict[str, Any], namespace: str) -> Dict[str, Any]:
|
|
"""Build a real IMAGE_PULL incident from live cluster evidence when the model response failed."""
|
|
active_pod = snapshot.get("activePod", "")
|
|
affected_deployment = _deployment_name_from_pod(active_pod)
|
|
affected_image = snapshot.get("affectedImage", "")
|
|
evidence = list(snapshot.get("evidence", []))
|
|
|
|
normalized = dict(result)
|
|
normalized["error_code"] = "IMAGE_PULL"
|
|
normalized["status"] = "FAILED"
|
|
normalized["failureType"] = "IMAGE_PULL_FAILURE"
|
|
normalized["affectedDeployment"] = affected_deployment
|
|
normalized["affectedPod"] = active_pod
|
|
normalized["affectedImage"] = affected_image
|
|
normalized["imagePullSecret"] = normalized.get("imagePullSecret", "")
|
|
normalized["oldReplicasAvailable"] = normalized.get("oldReplicasAvailable", 0)
|
|
normalized["newReplicaSetReadyReplicas"] = normalized.get("newReplicaSetReadyReplicas", 0)
|
|
normalized["rootCause"] = "Current pod state confirms an active image pull failure; model analysis was unavailable due to context length limits."
|
|
normalized["businessImpact"] = "The affected workload cannot start correctly, which can degrade application availability and delay business processing."
|
|
normalized["evidence"] = evidence
|
|
normalized["recommendedAction"] = "Review the image reference and registry credentials, then rerun the remediation pipeline for image pull recovery."
|
|
normalized["recommendedImagePullFix"] = {
|
|
"deploymentName": affected_deployment,
|
|
"secretName": normalized.get("imagePullSecret", ""),
|
|
"image": affected_image,
|
|
}
|
|
normalized["report"] = {
|
|
"executiveSummary": "Se confirmó una falla activa de image pull a partir del estado actual del pod en Kubernetes, por lo que el incidente debe tratarse como real aunque el análisis del modelo no haya completado.",
|
|
"rootCause": normalized["rootCause"],
|
|
"businessImpact": normalized["businessImpact"],
|
|
"recommendedAction": normalized["recommendedAction"],
|
|
"evidence": evidence,
|
|
}
|
|
_log_event(
|
|
"IMAGE_PULL_LIVE_FALLBACK",
|
|
{
|
|
"namespace": namespace,
|
|
"affectedPod": active_pod,
|
|
"affectedDeployment": affected_deployment,
|
|
"affectedImage": affected_image,
|
|
},
|
|
)
|
|
return normalized
|
|
|
|
|
|
def _get_live_imagepull_snapshot(namespace: str) -> Dict[str, Any]:
|
|
"""Return the current namespace image-pull state, prioritizing live pod state over history."""
|
|
try:
|
|
pods_response = _kube_api_request("GET", f"/api/v1/namespaces/{namespace}/pods")
|
|
except Exception as exc:
|
|
_log_event(
|
|
"IMAGE_PULL_LIVE_CHECK_FAILED",
|
|
{"namespace": namespace, "stage": "pods", "error": _exception_summary(exc)},
|
|
)
|
|
return {
|
|
"active": False,
|
|
"reason": "Live pod verification failed, so historical image pull events were ignored to avoid false positives.",
|
|
"evidence": [f"livePodImagePullCheckError={_exception_summary(exc)}"],
|
|
}
|
|
|
|
active_pods = []
|
|
pod_evidence: List[str] = []
|
|
affected_image = ""
|
|
for item in pods_response.get("items", []):
|
|
pod_name = item.get("metadata", {}).get("name", "")
|
|
status = item.get("status", {})
|
|
for container_status in status.get("containerStatuses", []) or []:
|
|
reason = _container_waiting_reason(container_status)
|
|
if reason in IMAGE_PULL_WAITING_REASONS:
|
|
active_pods.append(pod_name)
|
|
pod_evidence.append(f"activeImagePullPod={pod_name} reason={reason}")
|
|
if not affected_image:
|
|
affected_image = str(container_status.get("image", "")).strip()
|
|
for container_status in status.get("initContainerStatuses", []) or []:
|
|
reason = _container_waiting_reason(container_status)
|
|
if reason in IMAGE_PULL_WAITING_REASONS:
|
|
active_pods.append(pod_name)
|
|
pod_evidence.append(f"activeImagePullInitPod={pod_name} reason={reason}")
|
|
if not affected_image:
|
|
affected_image = str(container_status.get("image", "")).strip()
|
|
|
|
if active_pods:
|
|
recent_event_evidence = _recent_imagepull_event_evidence(namespace, active_pods)
|
|
return {
|
|
"active": True,
|
|
"reason": "Current pod state shows an active image pull failure.",
|
|
"activePod": active_pods[0],
|
|
"affectedImage": affected_image,
|
|
"evidence": pod_evidence + recent_event_evidence,
|
|
}
|
|
|
|
return {
|
|
"active": False,
|
|
"reason": "No current pod in the namespace is showing an active image pull error.",
|
|
"evidence": ["livePodImagePullState=clean", "livePodPhaseSummary=all-current-pods-running-or-completed"],
|
|
}
|
|
|
|
|
|
def _container_has_imagepull_reason(container_status: Dict[str, Any]) -> bool:
|
|
"""Check current waiting state for image-pull-specific failure reasons."""
|
|
reason = _container_waiting_reason(container_status)
|
|
return reason in IMAGE_PULL_WAITING_REASONS
|
|
|
|
|
|
def _container_waiting_reason(container_status: Dict[str, Any]) -> str:
|
|
waiting = ((container_status or {}).get("state") or {}).get("waiting") or {}
|
|
return str(waiting.get("reason", "")).strip().upper()
|
|
|
|
|
|
def _recent_imagepull_event_evidence(namespace: str, pod_names: List[str], lookback_minutes: int = 15) -> List[str]:
|
|
"""Capture only recent image-pull-related events for currently failing pods."""
|
|
try:
|
|
events_response = _kube_api_request("GET", f"/api/v1/namespaces/{namespace}/events")
|
|
except Exception as exc:
|
|
_log_event(
|
|
"IMAGE_PULL_LIVE_CHECK_FAILED",
|
|
{"namespace": namespace, "stage": "events", "error": _exception_summary(exc)},
|
|
)
|
|
return [f"recentImagePullEventCheckError={_exception_summary(exc)}"]
|
|
|
|
threshold = datetime.now(timezone.utc) - timedelta(minutes=lookback_minutes)
|
|
pod_name_set = {name for name in pod_names if name}
|
|
evidence: List[str] = []
|
|
for item in events_response.get("items", []):
|
|
involved_name = item.get("involvedObject", {}).get("name", "")
|
|
if involved_name not in pod_name_set:
|
|
continue
|
|
reason = str(item.get("reason", "")).strip().upper()
|
|
message = str(item.get("message", "")).strip()
|
|
if reason not in IMAGE_PULL_EVENT_REASONS and not any(
|
|
token in message.upper() for token in IMAGE_PULL_WAITING_REASONS
|
|
):
|
|
continue
|
|
event_time = _parse_k8s_event_time(item)
|
|
if event_time and event_time < threshold:
|
|
continue
|
|
evidence.append(f"recentEvent pod={involved_name} reason={reason} message={message[:180]}")
|
|
return evidence
|
|
|
|
|
|
def _parse_k8s_event_time(item: Dict[str, Any]) -> datetime | None:
|
|
"""Parse the most useful event timestamp field when present."""
|
|
for key in ("eventTime", "lastTimestamp", "firstTimestamp", "metadata.creationTimestamp"):
|
|
value = item.get(key)
|
|
if value is None and "." in key:
|
|
first, second = key.split(".", 1)
|
|
value = ((item.get(first) or {}).get(second))
|
|
if not value:
|
|
continue
|
|
try:
|
|
return datetime.fromisoformat(str(value).replace("Z", "+00:00"))
|
|
except ValueError:
|
|
continue
|
|
return None
|
|
|
|
|
|
def _deployment_name_from_pod(pod_name: str) -> str:
|
|
"""Best-effort extraction of deployment name from a ReplicaSet-style pod name."""
|
|
if not pod_name:
|
|
return ""
|
|
parts = pod_name.split("-")
|
|
if len(parts) >= 3:
|
|
return "-".join(parts[:-2])
|
|
return pod_name
|
|
|
|
|
|
def _normalize_error_code(result: Dict[str, Any]) -> str:
|
|
"""Map free-form responses back to one of the three supported error codes."""
|
|
raw = str(result.get("error_code", "")).strip().upper()
|
|
check_name = str(result.get("checkName", "")).strip().lower()
|
|
failure_type = str(result.get("failureType", "")).strip().upper()
|
|
status = str(result.get("status", "")).strip().upper()
|
|
problems = result.get("problems", [])
|
|
markers = set()
|
|
if failure_type:
|
|
markers.add(failure_type)
|
|
if isinstance(problems, list):
|
|
for problem in problems:
|
|
if isinstance(problem, str):
|
|
markers.add(problem.upper())
|
|
elif isinstance(problem, dict):
|
|
if problem.get("failureType"):
|
|
markers.add(str(problem.get("failureType")).upper())
|
|
if problem.get("type"):
|
|
markers.add(str(problem.get("type")).upper())
|
|
|
|
if raw in {"BUSINESS_APPLICATION", "HPA", "IMAGE_PULL"}:
|
|
return raw
|
|
if "IMAGE_PULL_FAILURE" in markers:
|
|
return "IMAGE_PULL"
|
|
if "HPA_AUTOSCALING_FAILURE" in markers:
|
|
return "HPA"
|
|
if raw in {"NO_ERROR", "NONE"} or status in {
|
|
"NO FAILURE",
|
|
"NO ISSUE DETECTED",
|
|
"NO HAY ERRORES DE OBTENCIÓN DE IMÁGENES.",
|
|
"NO HAY ERRORES DE OBTENCION DE IMAGENES.",
|
|
}:
|
|
if check_name == "hpa-failures":
|
|
return "HPA"
|
|
if check_name == "imagepull-error":
|
|
return "IMAGE_PULL"
|
|
if check_name == "hpa-failures":
|
|
return "HPA"
|
|
if check_name == "imagepull-error":
|
|
return "IMAGE_PULL"
|
|
return "BUSINESS_APPLICATION"
|
|
|
|
|
|
def _normalize_status(result: Dict[str, Any]) -> str:
|
|
"""Collapse non-standard healthy phrases to HEALTHY and active issues to DEGRADED."""
|
|
raw = str(result.get("status", "")).strip().upper()
|
|
failure_type = str(result.get("failureType", "") or "").strip().upper()
|
|
if raw in HEALTHY_STATUSES:
|
|
return "HEALTHY"
|
|
if raw in {"HEALTHY", "DEGRADED", "FAILED"}:
|
|
return raw
|
|
if failure_type in {"NONE", "NO_ACTIVE_FAILURE", "NO_HPA_DEFINED"}:
|
|
return "HEALTHY"
|
|
if failure_type in {"HPA_AUTOSCALING_FAILURE", "IMAGE_PULL_FAILURE", "BUSINESS_PROCESS_DEGRADATION"}:
|
|
return "DEGRADED"
|
|
return raw or "FAILED"
|
|
|
|
|
|
def _normalize_failure_type(result: Dict[str, Any]) -> str:
|
|
"""Infer a canonical failure type when kagent omits or varies the field."""
|
|
failure_type = str(result.get("failureType", "") or "").strip().upper()
|
|
status = str(result.get("status", "")).strip().upper()
|
|
if _is_model_runtime_failure(result):
|
|
return "JOB_RUNTIME_FAILURE"
|
|
if failure_type in {"NONE", "NO_HPA_DEFINED"}:
|
|
return "NO_ACTIVE_FAILURE"
|
|
if result.get("error_code") in {"HPA", "IMAGE_PULL"} and status in HEALTHY_STATUSES:
|
|
return "NO_ACTIVE_FAILURE"
|
|
if failure_type:
|
|
return failure_type
|
|
|
|
problems = result.get("problems", [])
|
|
if isinstance(problems, list):
|
|
for problem in problems:
|
|
if isinstance(problem, str) and problem.upper() in {"IMAGE_PULL_FAILURE", "HPA_AUTOSCALING_FAILURE", "BUSINESS_PROCESS_DEGRADATION"}:
|
|
return problem.upper()
|
|
if isinstance(problem, dict):
|
|
if problem.get("failureType"):
|
|
return str(problem.get("failureType")).upper()
|
|
if problem.get("type"):
|
|
return str(problem.get("type")).upper()
|
|
|
|
if result.get("error_code") in {"HPA", "IMAGE_PULL"} and _normalize_status(result) == "HEALTHY":
|
|
return "NO_ACTIVE_FAILURE"
|
|
return ""
|
|
|
|
|
|
def _is_model_runtime_failure(result: Dict[str, Any]) -> bool:
|
|
"""Detect provider/model failures so they are suppressed instead of remediated."""
|
|
code = str(result.get("code", "")).strip()
|
|
message = str(result.get("message", "")).strip().lower()
|
|
raw_response = str(result.get("rawResponse", "")).strip().lower()
|
|
root_cause = str(result.get("rootCause", "")).strip().lower()
|
|
payload = " ".join(part for part in [message, raw_response, root_cause] if part)
|
|
|
|
if code == "400" and "context_length_exceeded" in payload:
|
|
return True
|
|
if "context_length_exceeded" in payload:
|
|
return True
|
|
if "invalid_request_error" in payload and "maximum context length" in payload:
|
|
return True
|
|
return False
|
|
|
|
|
|
def _extract_pending_transactions(result: Dict[str, Any]) -> int:
|
|
"""Best-effort extraction of pending transactions from evidence lines."""
|
|
evidence = result.get("evidence", [])
|
|
if isinstance(evidence, list):
|
|
for item in evidence:
|
|
text = str(item)
|
|
if text.startswith("pendingReviewOrders="):
|
|
try:
|
|
return int(float(text.split("=", 1)[1]))
|
|
except ValueError:
|
|
return 0
|
|
return 0
|
|
|
|
|
|
def _compact_observability_json(data: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Keep the business prompt stateless but bounded so it does not grow with runtime history."""
|
|
if not isinstance(data, dict):
|
|
return {"fetchStatus": "FAILED", "error": "Observability payload is not a JSON object"}
|
|
|
|
business = data.get("businessHealth", {}) if isinstance(data.get("businessHealth"), dict) else {}
|
|
services = data.get("services", []) if isinstance(data.get("services"), list) else []
|
|
orders = data.get("recentOrders", []) if isinstance(data.get("recentOrders"), list) else []
|
|
recommendations = data.get("recommendedKagentActions", [])
|
|
if not isinstance(recommendations, list):
|
|
recommendations = [str(recommendations)] if recommendations else []
|
|
|
|
compact_orders = []
|
|
for order in orders[:5]:
|
|
if not isinstance(order, dict):
|
|
continue
|
|
compact_orders.append(
|
|
{
|
|
"orderId": order.get("orderId", ""),
|
|
"status": order.get("status", ""),
|
|
"amount": order.get("amount", 0),
|
|
"reason": order.get("reason", ""),
|
|
"createdAt": order.get("createdAt", ""),
|
|
}
|
|
)
|
|
|
|
compact_services = []
|
|
for service in services[:6]:
|
|
if not isinstance(service, dict):
|
|
continue
|
|
compact_services.append(
|
|
{
|
|
"name": service.get("name", ""),
|
|
"layer": service.get("layer", ""),
|
|
"status": service.get("status", ""),
|
|
"detail": service.get("detail", ""),
|
|
"kagentSignal": service.get("kagentSignal", False),
|
|
}
|
|
)
|
|
|
|
return {
|
|
"timestamp": data.get("timestamp", ""),
|
|
"businessHealth": {
|
|
"totalOrders": business.get("totalOrders", 0),
|
|
"confirmedOrders": business.get("confirmedOrders", 0),
|
|
"pendingReviewOrders": business.get("pendingReviewOrders", 0),
|
|
"failedOrders": business.get("failedOrders", 0),
|
|
"manualReviewRate": business.get("manualReviewRate", 0),
|
|
"totalRevenue": business.get("totalRevenue", 0),
|
|
"confirmedRevenue": business.get("confirmedRevenue", 0),
|
|
"revenueAtRisk": business.get("revenueAtRisk", 0),
|
|
"healthy": business.get("healthy", True),
|
|
"summary": business.get("summary", ""),
|
|
"businessImpact": business.get("businessImpact", ""),
|
|
},
|
|
"services": compact_services,
|
|
"recentOrders": compact_orders,
|
|
"recommendedKagentActions": recommendations[:5],
|
|
}
|
|
|
|
|
|
def _trigger_remediation(result: Dict[str, Any], payload: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Create the remediation deployment in OCI DevOps for one incident type."""
|
|
pipeline_key = {
|
|
"BUSINESS_APPLICATION": "business",
|
|
"IMAGE_PULL": "imagePull",
|
|
"HPA": "hpa",
|
|
}.get(result.get("error_code", ""), "business")
|
|
pipeline = payload["remediationPipelines"].get(pipeline_key, {})
|
|
pipeline_ocid = pipeline.get("pipelineOcid", "")
|
|
project_ocid = pipeline.get("projectOcid", "")
|
|
if not pipeline_ocid:
|
|
return {"triggered": False, "reason": "No remediation pipeline OCID was provided."}
|
|
|
|
client = oci.devops.DevopsClient(
|
|
config={"region": payload["region"]},
|
|
signer=oci.auth.signers.get_oke_workload_identity_resource_principal_signer(),
|
|
)
|
|
response = client.create_deployment(
|
|
create_deployment_details=oci.devops.models.CreateDeployPipelineDeploymentDetails(
|
|
deploy_pipeline_id=pipeline_ocid,
|
|
display_name=f"oke-remediator-{result.get('error_code', 'UNKNOWN').lower()}-{uuid.uuid4().hex[:8]}",
|
|
),
|
|
opc_retry_token=str(uuid.uuid4()),
|
|
)
|
|
deployment_id = response.data.id
|
|
return {
|
|
"triggered": True,
|
|
"deploymentId": deployment_id,
|
|
"deploymentUrl": (
|
|
f"https://cloud.oracle.com/devops-deployment/projects/{project_ocid}/pipelines/"
|
|
f"{pipeline_ocid}/deployments/{deployment_id}?region={payload['region']}"
|
|
),
|
|
}
|
|
|
|
|
|
def _safe_trigger_remediation(result: Dict[str, Any], payload: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Return a non-throwing remediation result so one failing check does not abort the whole job."""
|
|
try:
|
|
return _trigger_remediation(result, payload)
|
|
except Exception as exc:
|
|
return {"triggered": False, "reason": f"Failed to trigger remediation deployment: {_exception_summary(exc)}"}
|
|
|
|
|
|
def _maybe_send_email(result: Dict[str, Any], payload: Dict[str, Any]) -> tuple[bool, str]:
|
|
"""Publish one OCI Notifications email for an actionable incident."""
|
|
topic_id = payload.get("notificationTopicOcid", "")
|
|
if not topic_id:
|
|
return False, "NOTIFICATION_TOPIC_OCID is not configured."
|
|
|
|
client = oci.ons.NotificationDataPlaneClient(
|
|
config={"region": payload["region"]},
|
|
signer=oci.auth.signers.get_oke_workload_identity_resource_principal_signer(),
|
|
)
|
|
client.publish_message(
|
|
topic_id=topic_id,
|
|
message_details=oci.ons.models.MessageDetails(
|
|
title=_subject_for_result(result),
|
|
body=_body_for_result(result, payload["region"]),
|
|
),
|
|
)
|
|
return True, "Failure report published to OCI Notifications."
|
|
|
|
|
|
def _safe_send_email(result: Dict[str, Any], payload: Dict[str, Any]) -> tuple[bool, str]:
|
|
"""Return a non-throwing notification result for the current incident."""
|
|
try:
|
|
return _maybe_send_email(result, payload)
|
|
except Exception as exc:
|
|
return False, f"Failed to publish notification: {_exception_summary(exc)}"
|
|
|
|
|
|
def _subject_for_result(result: Dict[str, Any]) -> str:
|
|
"""Build a stable subject based on namespace, status and error code."""
|
|
return f"[OKE Job][kagent][{result.get('namespace', '')}] {result.get('status', '')} - {result.get('error_code', '')}"[:255]
|
|
|
|
|
|
def _body_for_result(result: Dict[str, Any], region: str) -> str:
|
|
"""Render the email body as a compact operational report."""
|
|
report = result.get("report", {}) if isinstance(result.get("report"), dict) else {}
|
|
remediation = result.get("remediationDeployment", {}) if isinstance(result.get("remediationDeployment"), dict) else {}
|
|
deployment_url = remediation.get("deploymentUrl", "")
|
|
evidence = report.get("evidence", result.get("evidence", []))
|
|
if not isinstance(evidence, list):
|
|
evidence = [str(evidence)]
|
|
|
|
lines = [
|
|
"kagent diagnostic report",
|
|
"",
|
|
f"Region: {region}",
|
|
f"Namespace: {result.get('namespace', '')}",
|
|
f"Status: {result.get('status', '')}",
|
|
f"Error code: {result.get('error_code', '')}",
|
|
f"Failure type: {result.get('failureType', '')}",
|
|
"",
|
|
"Executive summary:",
|
|
str(report.get("executiveSummary", result.get("summary", ""))),
|
|
"",
|
|
"Root cause:",
|
|
str(report.get("rootCause", result.get("rootCause", ""))),
|
|
"",
|
|
"Business impact:",
|
|
str(report.get("businessImpact", result.get("businessImpact", ""))),
|
|
"",
|
|
"Evidence:",
|
|
]
|
|
# Keep the body compact enough for email while still including the most
|
|
# useful evidence lines from the check.
|
|
lines.extend(f"- {item}" for item in evidence[:12])
|
|
lines.extend(
|
|
[
|
|
"",
|
|
"Recommended action:",
|
|
str(report.get("recommendedAction", result.get("recommendedAction", ""))),
|
|
"",
|
|
"Remediation deployment:",
|
|
f"- Triggered: {remediation.get('triggered', False)}",
|
|
f"- Deployment ID: {remediation.get('deploymentId', '')}",
|
|
]
|
|
)
|
|
if deployment_url:
|
|
lines.extend(["", "Approve or reject remediation deployment in OCI DevOps:", deployment_url])
|
|
elif remediation.get("reason"):
|
|
lines.extend(["", "Remediation deployment was not created:", str(remediation.get("reason", ""))])
|
|
return "\n".join(lines)
|
|
|
|
|
|
def _build_signature(result: Dict[str, Any]) -> str:
|
|
"""Build a stable signature used to suppress duplicate active incidents."""
|
|
payload = {
|
|
"namespace": result.get("namespace", ""),
|
|
"error_code": result.get("error_code", ""),
|
|
"status": result.get("status", ""),
|
|
"failureType": result.get("failureType", ""),
|
|
}
|
|
|
|
if result.get("error_code") == "BUSINESS_APPLICATION":
|
|
# For business incidents, use stable business metrics rather than the
|
|
# AI narrative text. This prevents duplicate mails when wording changes.
|
|
payload["pending_transactions"] = result.get("pending_transactions", 0)
|
|
payload["businessImpact"] = result.get("businessImpact", "")
|
|
payload["report"] = {
|
|
"status": result.get("status", ""),
|
|
"summary": result.get("summary", ""),
|
|
}
|
|
elif result.get("error_code") == "HPA":
|
|
# For HPA, rely on actual autoscaling measurements.
|
|
payload["currentReplicas"] = result.get("currentReplicas", 0)
|
|
payload["desiredReplicas"] = result.get("desiredReplicas", 0)
|
|
payload["maxReplicas"] = result.get("maxReplicas", 0)
|
|
payload["currentCpuUtilization"] = result.get("currentCpuUtilization", 0)
|
|
payload["targetCpuUtilization"] = result.get("targetCpuUtilization", 0)
|
|
elif result.get("error_code") == "IMAGE_PULL":
|
|
# For ImagePull, the deployment/image tuple is the key incident identity.
|
|
payload["affectedDeployment"] = result.get("affectedDeployment", "")
|
|
payload["affectedImage"] = result.get("affectedImage", "")
|
|
payload["imagePullSecret"] = result.get("imagePullSecret", "")
|
|
|
|
return hashlib.sha256(json.dumps(payload, sort_keys=True, ensure_ascii=False).encode("utf-8")).hexdigest()
|
|
|
|
|
|
def _is_active_failure(result: Dict[str, Any]) -> bool:
|
|
return str(result.get("status", "")).strip().upper() in {"DEGRADED", "FAILED"}
|
|
|
|
|
|
def _is_internal_failure(result: Dict[str, Any]) -> bool:
|
|
return str(result.get("failureType", "")).strip().upper() in INTERNAL_FAILURE_TYPES
|
|
|
|
|
|
def _load_notification_state(namespace: str) -> Dict[str, Dict[str, str]]:
|
|
"""Read the incident state ConfigMap from the current namespace."""
|
|
try:
|
|
result = _kube_api_request(
|
|
"GET",
|
|
f"/api/v1/namespaces/{namespace}/configmaps/{STATE_CONFIGMAP_NAME}",
|
|
)
|
|
raw = result.get("data", {}).get("state", "{}")
|
|
return json.loads(raw)
|
|
except Exception:
|
|
return {}
|
|
|
|
|
|
def _save_notification_state(namespace: str, state: Dict[str, Dict[str, str]]) -> None:
|
|
"""Create or update the ConfigMap that tracks active incidents by error_code."""
|
|
body = {
|
|
"apiVersion": "v1",
|
|
"kind": "ConfigMap",
|
|
"metadata": {"name": STATE_CONFIGMAP_NAME, "namespace": namespace},
|
|
"data": {"state": json.dumps(state, ensure_ascii=False)},
|
|
}
|
|
try:
|
|
_kube_api_request("PUT", f"/api/v1/namespaces/{namespace}/configmaps/{STATE_CONFIGMAP_NAME}", body=body)
|
|
except KeyError:
|
|
_kube_api_request("POST", f"/api/v1/namespaces/{namespace}/configmaps", body=body)
|
|
|
|
|
|
def _clear_state_for_code(namespace: str, state: Dict[str, Dict[str, str]], error_code: str) -> None:
|
|
"""Reset the saved incident state when a check becomes healthy again."""
|
|
state[error_code] = {
|
|
"signature": "",
|
|
"deploymentUrl": "",
|
|
"deploymentId": "",
|
|
"emailSent": "false",
|
|
}
|
|
_save_notification_state(namespace, state)
|
|
|
|
|
|
def _kube_api_request(method: str, path: str, body: Dict[str, Any] | None = None) -> Dict[str, Any]:
|
|
"""Small helper for reading/writing ConfigMaps with the pod service account token."""
|
|
token_path = "/var/run/secrets/kubernetes.io/serviceaccount/token"
|
|
ca_path = "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt"
|
|
with open(token_path, "r", encoding="utf-8") as token_file:
|
|
token = token_file.read().strip()
|
|
|
|
url = f"https://kubernetes.default.svc{path}"
|
|
data = None if body is None else json.dumps(body).encode("utf-8")
|
|
request = urllib.request.Request(url, data=data, method=method)
|
|
request.add_header("Authorization", f"Bearer {token}")
|
|
request.add_header("Content-Type", "application/json")
|
|
|
|
import ssl
|
|
|
|
context = ssl.create_default_context(cafile=ca_path)
|
|
try:
|
|
with urllib.request.urlopen(request, context=context, timeout=10) as response:
|
|
raw = response.read().decode("utf-8")
|
|
return json.loads(raw) if raw else {}
|
|
except HTTPError as exc:
|
|
if exc.code == 404:
|
|
raise KeyError("ConfigMap not found") from exc
|
|
raise
|
|
|
|
|
|
def _exception_summary(exc: Exception) -> str:
|
|
"""Flatten nested exceptions into one log-friendly string."""
|
|
nested = getattr(exc, "exceptions", None)
|
|
if nested:
|
|
return f"{exc.__class__.__name__}: " + "; ".join(
|
|
f"{item.__class__.__name__}: {item}" for item in nested
|
|
)
|
|
return str(exc)
|
|
|
|
|
|
# Final overrides: keep these definitions last so the runtime uses the hardened
|
|
# versions even if draft implementations remain above in the file.
|
|
async def run_multi_checks() -> Dict[str, Any]:
|
|
"""Run the three checks, prioritize IMAGE_PULL over BUSINESS_APPLICATION, then act."""
|
|
payload = _build_payload_from_env()
|
|
results: List[Dict[str, Any]] = []
|
|
notifications_sent: List[str] = []
|
|
remediations_triggered: List[Dict[str, Any]] = []
|
|
|
|
for check in _build_checks(payload):
|
|
result = await _run_single_check(payload, check)
|
|
results.append(result)
|
|
|
|
results = [_normalize_result(result, payload["appNamespace"]) for result in results]
|
|
imagepull_active = any(
|
|
result.get("error_code") == "IMAGE_PULL"
|
|
and _is_active_failure(result)
|
|
and not _is_internal_failure(result)
|
|
for result in results
|
|
)
|
|
|
|
state = _load_notification_state(payload["appNamespace"])
|
|
for result in results:
|
|
error_code = result.get("error_code", "")
|
|
|
|
if not _is_active_failure(result):
|
|
_clear_state_for_code(payload["appNamespace"], state, error_code)
|
|
result["emailSent"] = False
|
|
result["notificationReason"] = "No active failure detected."
|
|
_replace_result(results, result)
|
|
continue
|
|
|
|
if _is_internal_failure(result):
|
|
_log_event(
|
|
"KAGENT_MULTI_INTERNAL_FAILURE",
|
|
{
|
|
"check": result.get("checkName", ""),
|
|
"errorCode": error_code,
|
|
"failureType": result.get("failureType", ""),
|
|
"rootCause": result.get("rootCause", ""),
|
|
},
|
|
)
|
|
_clear_state_for_code(payload["appNamespace"], state, error_code)
|
|
result["emailSent"] = False
|
|
result["notificationReason"] = "Internal agent/runtime failure; notification suppressed."
|
|
_replace_result(results, result)
|
|
continue
|
|
|
|
if error_code == "BUSINESS_APPLICATION" and imagepull_active:
|
|
_clear_state_for_code(payload["appNamespace"], state, error_code)
|
|
result["emailSent"] = False
|
|
result["notificationReason"] = "Suppressed because IMAGE_PULL has priority over business_application."
|
|
_replace_result(results, result)
|
|
continue
|
|
|
|
signature = _build_signature(result)
|
|
state_entry = state.get(error_code, {})
|
|
active_url = state_entry.get("deploymentUrl", "")
|
|
active_signature = state_entry.get("signature", "")
|
|
active_email = state_entry.get("emailSent", "")
|
|
|
|
if signature == active_signature and active_url and active_email == "true":
|
|
result["emailSent"] = False
|
|
result["notificationReason"] = "Duplicate active issue; notification suppressed."
|
|
result["remediationDeployment"] = {
|
|
"triggered": True,
|
|
"deploymentUrl": active_url,
|
|
"deploymentId": state_entry.get("deploymentId", ""),
|
|
"reused": True,
|
|
}
|
|
_replace_result(results, result)
|
|
continue
|
|
|
|
remediation = _safe_trigger_remediation(result, payload)
|
|
if remediation:
|
|
result["remediationDeployment"] = remediation
|
|
if remediation.get("triggered"):
|
|
remediations_triggered.append(remediation)
|
|
|
|
email_sent, notification_reason = _safe_send_email(result, payload)
|
|
result["emailSent"] = email_sent
|
|
result["notificationReason"] = notification_reason
|
|
if email_sent:
|
|
notifications_sent.append(error_code)
|
|
|
|
if email_sent and (result.get("remediationDeployment") or {}).get("deploymentUrl"):
|
|
state[error_code] = {
|
|
"signature": signature,
|
|
"deploymentUrl": result["remediationDeployment"].get("deploymentUrl", ""),
|
|
"deploymentId": result["remediationDeployment"].get("deploymentId", ""),
|
|
"emailSent": "true",
|
|
}
|
|
else:
|
|
state[error_code] = {
|
|
"signature": "",
|
|
"deploymentUrl": "",
|
|
"deploymentId": "",
|
|
"emailSent": "false",
|
|
}
|
|
_replace_result(results, result)
|
|
|
|
_save_notification_state(payload["appNamespace"], state)
|
|
return {
|
|
"status": "COMPLETED",
|
|
"results": results,
|
|
"notificationsSent": notifications_sent,
|
|
"remediationsTriggered": remediations_triggered,
|
|
}
|
|
|
|
|
|
def _build_checks(payload: Dict[str, Any]) -> List[Dict[str, Any]]:
|
|
"""Create the three independent prompts with stricter contracts."""
|
|
observability_json = _fetch_json(payload["observabilityUrl"])
|
|
compact_observability_json = _compact_observability_json(observability_json)
|
|
_log_event(
|
|
"KAGENT_MULTI_OBSERVABILITY_SUMMARY",
|
|
{
|
|
"fetchStatus": observability_json.get("fetchStatus", "OK") if isinstance(observability_json, dict) else "UNKNOWN",
|
|
"keys": list(observability_json.keys())[:8] if isinstance(observability_json, dict) else [],
|
|
},
|
|
)
|
|
|
|
business_prompt = (
|
|
f"Responder en español. "
|
|
f"Analiza la salud de negocio de la aplicacion Orders desplegada en el namespace {payload['appNamespace']}. "
|
|
f"La fuente principal de verdad para esta evaluacion es el siguiente JSON de observabilidad: "
|
|
f"{json.dumps(compact_observability_json, ensure_ascii=False)}. "
|
|
"Debes evaluar especificamente si existe degradacion real del proceso de negocio. "
|
|
"Considera como senales de degradacion los siguientes casos: businessHealth.healthy=false, "
|
|
"manualReviewRate alto, pendingReviewOrders mayor que 0, revenueAtRisk alto, confirmedOrders igual a 0 "
|
|
"cuando existen ordenes, o cualquier businessImpact que indique riesgo operativo o financiero. "
|
|
"No bases tu respuesta solo en que los pods esten Running; esta evaluacion es de salud de negocio, no solo "
|
|
"de salud de Kubernetes. "
|
|
"Quiero una respuesta completa, concreta y util para correo operativo. "
|
|
"Responde exclusivamente con JSON valido, sin markdown, sin explicacion adicional, usando exactamente esta "
|
|
"estructura: "
|
|
"{"
|
|
"\"error_code\":\"BUSINESS_APPLICATION\","
|
|
"\"status\":\"HEALTHY|DEGRADED\","
|
|
"\"failureType\":\"BUSINESS_PROCESS_DEGRADATION|NO_ACTIVE_FAILURE\","
|
|
"\"pending_transactions\":0,"
|
|
"\"summary\":\"\","
|
|
"\"rootCause\":\"\","
|
|
"\"businessImpact\":\"\","
|
|
"\"recommendedAction\":\"\","
|
|
"\"evidence\":[\"\"],"
|
|
"\"report\":{"
|
|
"\"executiveSummary\":\"\","
|
|
"\"rootCause\":\"\","
|
|
"\"businessImpact\":\"\","
|
|
"\"recommendedAction\":\"\","
|
|
"\"evidence\":[\"\"],"
|
|
"\"recommendedConfigMapData\":{}"
|
|
"}"
|
|
"}. "
|
|
"Reglas obligatorias: "
|
|
"1. Si el JSON de observabilidad muestra degradacion de negocio, responde status=DEGRADED y "
|
|
"failureType=BUSINESS_PROCESS_DEGRADATION. "
|
|
"2. Si no hay degradacion real de negocio, responde status=HEALTHY y failureType=NO_ACTIVE_FAILURE. "
|
|
"3. pending_transactions debe contener el numero real de ordenes pendientes de revision. "
|
|
"4. evidence debe incluir evidencia real observada, por ejemplo manualReviewRate, pendingReviewOrders, "
|
|
"revenueAtRisk, confirmedOrders, totalOrders, services degradados y cualquier recommendedKagentActions "
|
|
"relevante. "
|
|
"5. executiveSummary, rootCause, businessImpact y recommendedAction deben venir completos y redactados "
|
|
"como un reporte operativo claro."
|
|
)
|
|
_log_event("KAGENT_MULTI_PROMPT_READY", {"check": "business-application"})
|
|
|
|
hpa_prompt = (
|
|
f"Responder en español. "
|
|
"Analiza el namespace kagent-demo y determina si existe un caso activo de HPA_AUTOSCALING_FAILURE "
|
|
"en order-service. Revisa HPA, en order-service. Revisa deployment y pods de order-service."
|
|
"Revisa métricas actuales del HPA."
|
|
"Clasifica HPA_AUTOSCALING_FAILURE solamente si se cumplen ambas condiciones al mismo "
|
|
"tiempo: desiredReplicas == maxReplicas y currentCpuUtilization > targetCpuUtilization. Solo si ambas "
|
|
"condiciones anteriores se cumplen, responde status = DEGRADED y failureType = HPA_AUTOSCALING_FAILURE. "
|
|
"Si alguna de esas dos condiciones no se cumple, entonces no debes clasificarlo como DEGRADED ni FAILED"
|
|
"Responde únicamente con datos reales observados a partir de HPA, deployment, pods, observabilidad y diagnóstico interno."
|
|
"Por favor responder en formato JSON valido con los valores reales solicitados en los puntos anteriores de la siguiente forma:"
|
|
"{"
|
|
"error_code: HPA,"
|
|
"status: HEALTHY o DEGRADED o FAILED de acuerdo al estado real obtenido anteriormente al revisar numero de replicas actuales con porcentaje de consumo si es mayor al target fijado en HPA,"
|
|
"failureType: HPA_AUTOSCALING_FAILURE si verificamos que esta el numero de replicas al maximo y el consumo es mayor al target configurado en HPA; DEPENDENCY_FAILURE si el problema principal es resolucion DNS, endpoints o conectividad interna; OKE_PLATFORM_FAILURE si existe un problema del cluster o plataforma; o NO_ACTIVE_FAILURE si no encontramos que se superan los limites anteriormente mencionados,"
|
|
"rootCause: colocar conclusion de modelo de Inteligencia artificial de Causa Raiz, solo si el servicio es detectado que esta fallando,"
|
|
"businessImpact: Colocar el impacto hacia el negocio definido por la IA, solo si esta fallando,"
|
|
"evidence: ["
|
|
"Colocar evidencia real observada"
|
|
"],"
|
|
"recommendedAction: colocar remediacion que puede ser activar mayor numero de replicas si es HPA, o revisar DNS/Services/Endpoints si es dependencia, solo si esta fallando con informacion anteriormente mencionada,"
|
|
"recommendedHpaPatch: {"
|
|
"minReplicas: 3,"
|
|
"maxReplicas: El modelo de IA debe colocar una recomendacion de numero de replicas de acuerdo a porcentaje de consumo superior al target solo si esta fallando por HPA,"
|
|
"averageUtilization: Colocar el verdadero porcentaje de utilizacion anteriormente encontrado"
|
|
"}"
|
|
"}"
|
|
|
|
|
|
)
|
|
_log_event("KAGENT_MULTI_PROMPT_READY", {"check": "hpa-failures"})
|
|
|
|
image_prompt = (
|
|
f"Responder en español. "
|
|
"""
|
|
Analiza el namespace kagent-demo y determina si existe un caso activo de IMAGE_PULL_FAILURE en payment-service o en otros deployments de la aplicacion.
|
|
|
|
Instrucciones:
|
|
|
|
1. Revisa deployments y pods del namespace kagent-demo.
|
|
2. Revisa los ReplicaSets relacionados con los deployments de la aplicacion.
|
|
3. Revisa eventos recientes del namespace kagent-demo.
|
|
4. Revisa si existen pods en estados como ErrImagePull, ImagePullBackOff, InvalidImageName o CreateContainerConfigError relacionados con imagen o imagePullSecrets.
|
|
5. Revisa si el deployment tiene una imagen invalida, un tag inexistente o un imagePullSecret incorrecto o faltante.
|
|
6. Revisa la URL de observabilidad y correlaciona si hay impacto en la aplicacion.
|
|
7. Clasifica IMAGE_PULL_FAILURE solo si se cumple al menos una de estas condiciones:
|
|
- existe un pod con estado ErrImagePull
|
|
- existe un pod con estado ImagePullBackOff
|
|
- existe un pod con estado InvalidImageName
|
|
- el deployment actualizado no puede crear pods listos por falla de extraccion de imagen
|
|
- el contenedor usa una imagen inexistente o un secret de registro invalido
|
|
8. No clasifiques IMAGE_PULL_FAILURE si todos los pods nuevos estan corriendo y listos, o si el rollout completo correctamente.
|
|
9. Si observas que los pods viejos siguen disponibles pero el nuevo ReplicaSet no puede arrancar por error de imagen, eso si debe clasificarse como IMAGE_PULL_FAILURE.
|
|
10. No confundas un error de imagen con fallas de readiness/liveness o con problemas de DNS interno.
|
|
11. Si el problema principal es DNS, endpoints o conectividad interna, clasificalo como DEPENDENCY_FAILURE o OKE_PLATFORM_FAILURE segun corresponda.
|
|
12. Revisa unicamente valores reales observados en deployments, ReplicaSets, pods, eventos y observabilidad.
|
|
|
|
Por favor responde exclusivamente en formato JSON valido con los valores reales observados y con la siguiente estructura:
|
|
|
|
{
|
|
"error_code": "IMAGE_PULL",
|
|
"status": "HEALTHY|DEGRADED|FAILED",
|
|
"failureType": "IMAGE_PULL_FAILURE|DEPENDENCY_FAILURE|OKE_PLATFORM_FAILURE|NO_ACTIVE_FAILURE",
|
|
"affectedDeployment": "",
|
|
"affectedPod": "",
|
|
"affectedImage": "",
|
|
"imagePullSecret": "",
|
|
"oldReplicasAvailable": 0,
|
|
"newReplicaSetReadyReplicas": 0,
|
|
"rootCause": "",
|
|
"businessImpact": "",
|
|
"evidence": [
|
|
"Colocar evidencia real observada"
|
|
],
|
|
"recommendedAction": "",
|
|
"recommendedImagePullFix": {
|
|
"deploymentName": "",
|
|
"secretName": "",
|
|
"image": ""
|
|
}
|
|
}
|
|
"""
|
|
)
|
|
_log_event("KAGENT_MULTI_PROMPT_READY", {"check": "imagepull-error"})
|
|
|
|
return [
|
|
{"name": "business-application", "prompt": business_prompt},
|
|
{"name": "hpa-failures", "prompt": hpa_prompt},
|
|
{"name": "imagepull-error", "prompt": image_prompt},
|
|
]
|
|
|
|
|
|
def _build_signature(result: Dict[str, Any]) -> str:
|
|
"""Build a stable signature used to suppress duplicate active incidents."""
|
|
payload = {
|
|
"namespace": result.get("namespace", ""),
|
|
"error_code": result.get("error_code", ""),
|
|
"status": result.get("status", ""),
|
|
"failureType": result.get("failureType", ""),
|
|
}
|
|
if result.get("error_code") == "BUSINESS_APPLICATION":
|
|
payload["pending_transactions"] = result.get("pending_transactions", 0)
|
|
payload["businessImpact"] = result.get("businessImpact", "")
|
|
payload["summary"] = result.get("summary", "")
|
|
elif result.get("error_code") == "HPA":
|
|
payload["currentReplicas"] = result.get("currentReplicas", 0)
|
|
payload["desiredReplicas"] = result.get("desiredReplicas", 0)
|
|
payload["maxReplicas"] = result.get("maxReplicas", 0)
|
|
payload["currentCpuUtilization"] = result.get("currentCpuUtilization", 0)
|
|
payload["targetCpuUtilization"] = result.get("targetCpuUtilization", 0)
|
|
elif result.get("error_code") == "IMAGE_PULL":
|
|
payload["affectedDeployment"] = result.get("affectedDeployment", "")
|
|
payload["affectedImage"] = result.get("affectedImage", "")
|
|
payload["imagePullSecret"] = result.get("imagePullSecret", "")
|
|
return hashlib.sha256(json.dumps(payload, sort_keys=True, ensure_ascii=False).encode("utf-8")).hexdigest()
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|