Automated incident triage on GKE with Gemini 4 Argon and Model Context Protocol: Correlating cluster events and VPA logs

export PROJECT_ID="production-ai-platform"
export PROJECT_NUMBER=$(gcloud projects describe "${PROJECT_ID}" --format="value(projectNumber)")
export LOCATION="us-central1"
export CLUSTER_NAME="commerce-production-cluster"

# Check currently enabled logging components first
gcloud container clusters describe "${CLUSTER_NAME}" \
    --location="${LOCATION}" \
    --project="${PROJECT_ID}" \
    --format="value(loggingConfig.componentConfig.enableComponents)"

# Enable SYSTEM, WORKLOAD, and KCP_VPA logs
gcloud container clusters update "${CLUSTER_NAME}" \
    --location="${LOCATION}" \
    --project="${PROJECT_ID}" \
    --logging=SYSTEM,WORKLOAD,KCP_VPA

apiVersion: v1
kind: Namespace
metadata:
  name: sre-triage
---
apiVersion: v1
kind: ServiceAccount
metadata:
  name: gemini-triage-sa
  namespace: sre-triage
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: gemini-readonly-triage-role
rules:
- apiGroups: [""]
  resources: ["pods", "pods/log", "events", "services", "nodes", "namespaces", "configmaps"]
  verbs: ["get", "list", "watch"]
- apiGroups: ["events.k8s.io"]
  resources: ["events"]
  verbs: ["get", "list", "watch"]
- apiGroups: ["apps"]
  resources: ["deployments", "statefulsets", "daemonsets", "replicasets"]
  verbs: ["get", "list", "watch"]
- apiGroups: ["autoscaling"]
  resources: ["horizontalpodautoscalers"]
  verbs: ["get", "list", "watch"]
- apiGroups: ["autoscaling.k8s.io"]
  resources: ["verticalpodautoscalers"]
  verbs: ["get", "list", "watch"]
- apiGroups: ["cloud.google.com"]
  resources: ["computeclasses"]
  verbs: ["get", "list", "watch"]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
  name: gemini-readonly-triage-binding
subjects:
- kind: ServiceAccount
  name: gemini-triage-sa
  namespace: sre-triage
roleRef:
  kind: ClusterRole
  name: gemini-readonly-triage-role
  apiGroup: rbac.authorization.k8s.io

kubectl apply -f triage-agent-rbac.yaml

for ROLE in roles/aiplatform.user roles/container.viewer roles/logging.viewer; do
  gcloud projects add-iam-policy-binding "${PROJECT_ID}" \
      --member="principal://iam.googleapis.com/projects/${PROJECT_NUMBER}/locations/global/workloadIdentityPools/${PROJECT_ID}.svc.id.goog/subject/ns/sre-triage/sa/gemini-triage-sa" \
      --role="${ROLE}" \
      --condition=None
done

import asyncio
import os
from google import genai
from google.genai import types
from google.cloud import logging as cloud_logging
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

PROJECT_ID = os.environ["GOOGLE_CLOUD_PROJECT"]
LOCATION = os.environ.get("VERTEX_LOCATION", "us-central1")
MODEL_ID = os.environ.get("GEMINI_MODEL_ID", "gemini-4-argon")

genai_client = genai.Client(vertexai=True, project=PROJECT_ID, location=LOCATION)
logging_client = cloud_logging.Client(project=PROJECT_ID)


def query_vpa_decision_logs(
    cluster_name: str, namespace: str, workload_name: str, limit: int = 10
) -> list[dict]:
    """Queries GKE KCP_VPA control plane logs for a specific workload to inspect resize or eviction decisions.

    Args:
        cluster_name: Name of the GKE cluster.
        namespace: Kubernetes namespace of the target workload.
        workload_name: Name of the target Deployment or workload.
        limit: Maximum number of log entries to return.
    """
    log_filter = (
        'resource.type="k8s_control_plane_component" '
        f'AND resource.labels.cluster_name="{cluster_name}" '
        f'AND logName="projects/{PROJECT_ID}/logs/container.googleapis.com%2Fvpa-controller" '
        f'AND jsonPayload.target.namespace="{namespace}" '
        f'AND jsonPayload.target.name="{workload_name}"'
    )
    entries = logging_client.list_entries(
        filter_=log_filter,
        order_by=cloud_logging.DESCENDING,
        max_results=limit,
    )
    results = []
    for entry in entries:
        payload = entry.payload if isinstance(entry.payload, dict) else {}
        results.append({
            "timestamp": str(entry.timestamp),
            "operation": payload.get("operation"),
            "state": payload.get("state"),
            "reason": payload.get("reason"),
            "confidence": payload.get("confidence"),
        })
    return results


async def investigate_workload_incident(cluster_name: str, namespace: str, workload_name: str) -> str:
    """Connects Gemini to the GKE MCP server and Cloud Logging to diagnose a failing workload."""
    server_params = StdioServerParameters(
        command="gke-mcp",
        args=[],
        env=os.environ.copy(),
    )

    async with stdio_client(server_params) as (read_stream, write_stream):
        async with ClientSession(read_stream, write_stream) as mcp_session:
            await mcp_session.initialize()

            prompt = (
                f"Investigate the degraded Deployment '{workload_name}' in namespace '{namespace}' "
                f"on GKE cluster '{cluster_name}'. "
                "1. List recent Kubernetes events and inspect the Deployment and Pod statuses. "
                "2. Check the container termination exit codes and recent Pod logs. "
                "3. Query VPA decision logs using query_vpa_decision_logs to see if an in-place resize "
                "or eviction occurred or was skipped. "
                "4. Provide a concise root-cause timeline and a git-ready YAML patch."
            )

            config = types.GenerateContentConfig(
                system_instruction=(
                    "You are a senior GKE Site Reliability Engineer. Always verify Pod status, exit codes, "
                    "Kubernetes events, and VPA decision logs before proposing a fix. Never guess when "
                    "telemetry can be queried via your tools."
                ),
                tools=[mcp_session, query_vpa_decision_logs],
                automatic_function_calling=types.AutomaticFunctionCallingConfig(
                    maximum_remote_calls=8
                ),
                temperature=0.1,
            )

            response = await genai_client.aio.models.generate_content(
                model=MODEL_ID,
                contents=prompt,
                config=config,
            )
            return response.text

Root-cause summary for Deployment payments/checkout-service:
1. Pod status & exit code: Pod checkout-service-7c9d8b6f4-x82lp terminated with reason OOMKilled (exit code 137) at 06:14:22 UTC after memory usage reached its 1Gi limit.
2. Kubernetes events: BackOff restarting failed container; InPlaceResizeFailed event reported on node gke-commerce-pool-4a91f.
3. VPA control plane logs (KCP_VPA):
   - Operation: APPLY_RECOMMENDATION_IN_PLACE | State: FAILED | Confidence: HIGH
   - Reason: Recommended memory target (2Gi) exceeded maxAllowed policy bound (1Gi) defined in VerticalPodAutoscaler checkout-service-vpa.
4. Recommended remediation: Raise resourcePolicy.containerPolicies[0].maxAllowed.memory in checkout-service-vpa from 1Gi to 3Gi so VPA can apply high-confidence in-place memory resizes without triggering OOM terminations.

# Verify the VPA resource policy cap on the cluster
kubectl get vpa checkout-service-vpa -n payments -o yaml

# Inspect the corresponding KCP_VPA log entry in Cloud Logging
gcloud logging read "resource.type=\"k8s_control_plane_component\" AND resource.labels.cluster_name=\"${CLUSTER_NAME}\" AND logName=\"projects/${PROJECT_ID}/logs/container.googleapis.com%2Fvpa-controller\" AND jsonPayload.target.name=\"checkout-service\"" \
    --project="${PROJECT_ID}" \
    --limit=3 \
    --format=json