references/acceptance-criteria.md
# Acceptance Criteria
## Current Events
- `huawei_get_cce_events` returns `success: true`, the requested `cluster_id`, a non-empty `access_method`, and an `events` array when Events exist.
- `huawei_analyze_cce_events` with `region` and `cluster_id` returns `source: current`, a `query` object, and aggregate counters.
## Historical LTS Events
- `huawei_query_k8s_events_from_lts` reads the cluster `default-event` LogConfig through `kubectl cce` and returns its configured LTS group and stream IDs.
- A bounded query that has collected Events returns normalized records with `type`, `reason`, timestamp, and affected-object fields when available.
- `huawei_analyze_cce_events event_source=lts` returns `source: lts`, query time-range metadata, and aggregate counters.
## Analysis Quality
- Analysis returns event-record and occurrence totals, type breakdown, top reasons, namespaces, affected objects, and repeated patterns.
- Supplying an `events` array continues to perform offline analysis without a cloud request.
- Missing credentials, missing LTS collection, inaccessible clusters, and invalid time windows return actionable errors without changing cloud or Kubernetes resources.
references/kubectl-cce.md
# kubectl-cce Usage
`huawei_get_cce_events` uses `kubectl` to read Events. It first uses a temporary kubeconfig through the cluster external endpoint. When no external endpoint is available, it falls back to the `kubectl cce` plugin.
## Install
Install `kubectl` with the system package manager and verify it:
```bash
kubectl version --client
```
Install `kubectl-cce` v0.1.0 from the GitHub release that matches the local OS and architecture:
```bash
curl -LO https://github.com/pancake0001/kubectl-cce-plugin/releases/download/v0.1.0/kubectl-cce_0.1.0_linux_amd64.tar.gz
tar -xzf kubectl-cce_0.1.0_linux_amd64.tar.gz
chmod +x kubectl-cce && mv kubectl-cce /usr/local/bin/
kubectl plugin list
```
The executable must be named `kubectl-cce` so that kubectl discovers it as `kubectl cce`.
## Plugin Credentials
The plugin requires AK, SK, and the target cluster's project ID. The skill supplies compatible credential environment variables internally and passes the project ID explicitly with `--project-id`. Temporary credentials also require a security token.
Set these values through an approved local credential provider before invoking the plugin. Never place credential values in this document, shell history, source control, or command output.
## Example
```bash
kubectl cce --cluster-id <cluster-id> --region <region> --project-id <project-id> get events -A
```
references/output-schema.md
# Output Schema
This is the single source of truth for the public response fields emitted by this skill.
## Current Event Response (`huawei_get_cce_events`)
| Field | Description |
| --- | --- |
| `success` | Whether the query completed successfully |
| `region`, `cluster_id` | Requested Huawei Cloud region and CCE cluster ID |
| `namespace` | Requested namespace or `all` |
| `event_type` | Applied Event type filter |
| `access_method` | `kubectl_kubeconfig_external` or `kubectl_cce_plugin` |
| `count`, `limit` | Returned Event record count and requested maximum |
| `events` | Normalized Event records |
## Historical Event Response (`huawei_query_k8s_events_from_lts`)
| Field | Description |
| --- | --- |
| `success` | Whether the LTS query completed successfully |
| `region`, `cluster_id` | Requested Huawei Cloud region and CCE cluster ID |
| `log_group_id`, `log_stream_id` | LTS source identifiers read from `default-event` |
| `event_type`, `keywords` | Applied LTS type and keyword filter |
| `event_count`, `events` | Returned historical Event record count and parsed records |
| `time_range` | Requested UTC start and end time |
| `log_config` | `default-event` LogConfig metadata and discovery method |
| `pagination` | Page count, limit, and whether more results are available |
## Event Analysis Response (`huawei_analyze_cce_events`)
| Field | Description |
| --- | --- |
| `source` | `current`, `lts`, or caller-provided source label |
| `event_records`, `total_occurrences` | Input record count and sum of Event `count` values |
| `event_type_breakdown` | Occurrence totals by Event type |
| `warning_count`, `normal_count` | Warning and Normal occurrence totals |
| `time_range` | First and last observed Event timestamps |
| `top_reasons` | Most frequent reasons with occurrence count, Warning count, and time range |
| `namespace_breakdown`, `affected_objects` | Most affected namespaces and resources |
| `repeated_patterns` | Event records whose `count` is greater than one |
| `resource_status` | Current-state checks for Event resources when `region` and `cluster_id` are available |
| `query` | Source-query metadata when the tool fetched Events itself |
### Resource Status
| Field | Description |
| --- | --- |
| `checked` | Number of distinct resources checked, capped by `max_groups` |
| `summary` | Resource counts grouped by current state |
| `resources` | Per-resource kind, name, namespace, state, and message |
| `state` | `normal`, `abnormal`, `unknown`, `not_found`, `unsupported`, or `query_failed` |
## Normalized Event Record
| Field | Description |
| --- | --- |
| `name`, `namespace` | Event name and namespace when available |
| `type`, `reason`, `message` | Event type, reason, and message |
| `involved_object` | Referenced resource `kind`, `name`, and `namespace` |
| `count` | Number of occurrences represented by the record |
| `first_timestamp`, `last_timestamp` | First and latest observed timestamps |
references/risk-rules.md
# Risk Rules & Guardrails
## Hard Constraints (NEVER violate)
### H1: Read-Only Operations
This skill only queries Kubernetes events and lists related resources. No modifications are made to any cluster resource.
**Rationale**: Event analysis should never alter cluster state. Remediation must be handled by dedicated diagnosis/remediation skills with appropriate confirmation mechanisms.
### H2: Data Redaction
Do not expose sensitive data such as node names, pod names, or workload names that could identify production systems in public outputs. Use redacted or fictional examples in summaries when possible.
**Rationale**: Production system identifiers in event summaries can leak infrastructure details to unauthorized parties.
### H3: Hand Off Remediation
If event analysis reveals a clear remediation path, provide evidence and hand off to the appropriate diagnosis or remediation skill instead of executing recovery actions here.
**Rationale**: This skill lacks confirmation mechanisms for write operations. Diagnosis skills have proper guardrails for remediation actions.
### H4: Time-Bounded Queries
Keep event queries time-bounded. Prefer recent windows (1-24 hours) to avoid overwhelming results.
**Rationale**: Unbounded queries can return thousands of events, making analysis impractical and consuming excessive API resources.
## Soft Constraints (SHOULD follow, exceptions documented)
### S1: Start with K8s API
Use `huawei_get_cce_events` as the primary query method. Fall back to `huawei_query_k8s_events_from_lts` only when precise time-range filtering or keyword search is needed.
**Rationale**: K8s API is simpler. LTS provides server-side filtering and requires the log-agent default Event-to-LTS collection to be enabled.
### S2: Filter Warning First
When analyzing events, filter `type == "Warning"` first. Warning events are the primary diagnostic signal.
**Rationale**: Normal events are informational noise. Warning events indicate actual problems requiring attention.
### S3: Group by Reason Before Detail
Always group events by `reason` before examining individual events. This reveals systemic patterns faster.
**Rationale**: Individual event inspection without grouping misses recurring patterns that indicate root causes.
## Guardrails
1. **Read-only**: This skill never modifies, deletes, or creates Kubernetes resources
2. **No auto-remediation**: If the user asks to take action based on event findings, redirect to `huawei-cloud-cce-auto-remediation-runner` with the evidence summarized
3. **Data redaction**: Never expose production pod/node/workload names in summaries
4. **Handoff required**: Event findings that indicate specific failures must be handed off to diagnosis skills with evidence
5. **Time-bounded**: Default to recent 1-24 hour windows; never query without time bounds unless user explicitly requests
references/workflow.md
# Event Query Workflow
## Event Query Sequence
1. Identify `region`, `cluster_id`, and optional `namespace` from the user query.
2. Use `huawei_get_cce_events` for current Events. Use `huawei_query_k8s_events_from_lts` for historical windows longer than one hour.
3. Apply follow-up filters based on user needs:
- `reason` patterns (FailedScheduling, ImagePullBackOff, FailedMount, etc.)
- `involved_object.kind` / `involved_object.name` for specific resources
- `first_timestamp` / `last_timestamp` for time-window analysis
4. Group events by `reason`, `type`, or `namespace`.
5. Summarize top reasons, repeated patterns, and affected resources.
## Event Pattern Recognition
| Pattern | Likely Cause | Handoff Target |
|---------|-------------|---------------|
| `ImagePullBackOff` repeated | Wrong image or pull secret missing | `huawei-cloud-cce-pod-failure-diagnoser` |
| `FailedScheduling` + `insufficient` | Resource pressure or node not ready | `huawei-cloud-cce-workload-failure-diagnoser` |
| `FailedMount` | Volume attach or PVC issue | `huawei-cloud-cce-storage-failure-diagnoser` |
| `Evicted` pods | Budget disruption or node pressure | `huawei-cloud-cce-pod-failure-diagnoser` |
| `NodeNotReady` | Node agent or network issue | `huawei-cloud-cce-node-failure-diagnoser` |
| `Unhealthy` + Readiness probe | Application issue or startup failure | `huawei-cloud-cce-pod-failure-diagnoser` |
| `FailedCreatePodSandBox` | CNI or network issue | `huawei-cloud-cce-network-failure-diagnoser` |
| `OOMKilled` | Memory limit exceeded | `huawei-cloud-cce-pod-failure-diagnoser` |
## Time-Window Analysis
1. Events include `first_timestamp` and `last_timestamp` fields.
2. If the user provides an incident window, filter events by these timestamps.
3. Compare event frequency before, during, and after the incident window.
4. Flag events that started or peaked during the incident window.
5. Report `warning_count` vs `normal_count` ratio within each window segment.
## Event Aggregation
1. For large event volumes, aggregate by `reason` and show top N patterns with total counts.
2. For repeated events (count > 1), show the first and last timestamp and the object involved.
3. If a namespace has > 50% of events, flag it as high-noise and suggest namespace-level investigation.
4. Report `warning_count` vs `normal_count` to give a quick health signal.
## LTS vs K8s API Selection Guide
| Criteria | Use K8s API (`huawei_get_cce_events`) | Use LTS (`huawei_query_k8s_events_from_lts`) |
|----------|---------------------------------------|---------------------------------------------|
| Current Events or a recent check within one hour | Yes | No (needs time range) |
| Precise time range | No (client-side only) | Yes (server-side filter) |
| Keyword search | No (client-side only) | Yes (keywords parameter) |
| Historical Events over one hour | No | Yes |
| Requires Kubernetes LogConfig | No | Yes (`default-event` with LTS output) |
| Default route | Primary for current Events | Primary for historical windows over one hour |
scripts/huawei_cloud/__init__.py
"""Huawei Cloud service modules."""
scripts/huawei_cloud/cce_app_logs.py
"""Read CCE Event-to-LTS LogConfig custom resources through kubectl-cce."""
from __future__ import annotations
from typing import Any, Dict, List
from . import kubectl_client
def get_cce_logconfigs_action(params: Dict[str, str]) -> Dict[str, Any]:
"""Return LogConfig resources needed to locate CCE Event LTS streams."""
region = params.get("region")
cluster_id = params.get("cluster_id")
if not region:
return {"success": False, "error": "region is required"}
if not cluster_id:
return {"success": False, "error": "cluster_id is required"}
result = kubectl_client.get_cce_logconfigs_with_cce_plugin(
region=region,
cluster_id=cluster_id,
ak=params.get("ak"),
sk=params.get("sk"),
project_id=params.get("project_id"),
security_token=params.get("security_token"),
)
if not result.get("success"):
return result
logconfigs: List[Dict[str, Any]] = []
for item in result.get("items") or []:
metadata = item.get("metadata") or {}
spec = item.get("spec") or {}
input_detail = spec.get("inputDetail") or {}
output_detail = spec.get("outputDetail") or {}
logconfigs.append(
{
"name": metadata.get("name"),
"namespace": metadata.get("namespace"),
"input_type": input_detail.get("type"),
"output_type": output_detail.get("type"),
"spec": spec,
"api_version": item.get("apiVersion"),
}
)
return {
"success": True,
"cluster_id": cluster_id,
"access_method": result.get("access_method"),
"count": len(logconfigs),
"logconfigs": logconfigs,
}
scripts/huawei_cloud/cce_events_lts.py
"""
Query Kubernetes events from LTS log streams.
This module implements huawei_query_k8s_events_from_lts tool which:
1. Reads Event-to-LTS LogConfig resources through kubectl-cce
2. Queries LTS for K8s events in the specified time range
3. Parses and returns structured event data
"""
import json
import re
import time
from datetime import datetime, timezone
from typing import Dict, Any, Optional, List
try:
from . import cce_app_logs, lts as lts_mod
_lts_available = True
except ImportError:
_lts_available = False
cce_app_logs = None
lts_mod = None
def _convert_timestamp_to_ms(time_str: str) -> int:
"""Convert a UTC 'YYYY-MM-DD HH:MM:SS' timestamp to milliseconds."""
dt = datetime.strptime(time_str, "%Y-%m-%d %H:%M:%S").replace(tzinfo=timezone.utc)
return int(dt.timestamp() * 1000)
def _parse_event_content(log_content: str) -> Optional[Dict[str, Any]]:
"""
Parse K8s event from LTS log content.
Supports two formats:
- Format A: lowercase keys (standard K8s event format)
- Format B: uppercase keys (Huawei CCE event format)
Returns normalized dict with lowercase keys, or None if parsing fails.
"""
try:
data = json.loads(log_content)
except (json.JSONDecodeError, TypeError):
return None
if not isinstance(data, dict):
return None
normalized = {}
# Handle both lowercase and uppercase key formats
key_mapping = {
'reason': 'reason',
'message': 'message',
'type': 'type',
'count': 'count',
'firsttimestamp': 'first_timestamp',
'lasttimestamp': 'last_timestamp',
'involvedobject': 'involved_object',
}
# Uppercase format mapping
uppercase_mapping = {
'Reason': 'reason',
'Message': 'message',
'Type': 'type',
'Count': 'count',
'FirstTimestamp': 'first_timestamp',
'LastTimestamp': 'last_timestamp',
'InvolvedObject': 'involved_object',
}
# Determine which format we're dealing with.
if 'reason' in data:
# Format A (lowercase)
for k, v in data.items():
if k in key_mapping:
normalized[key_mapping[k]] = v
else:
normalized[k] = v
elif 'Reason' in data:
# Format B (uppercase) - convert to lowercase
for k, v in data.items():
mapped_key = uppercase_mapping.get(k, k.lower())
normalized[mapped_key] = v
else:
# Unknown format, just lowercase everything
for k, v in data.items():
normalized[k.lower()] = v
# Cloud Native Log Collection writes Event records with this compact schema:
# `name` is the Kubernetes Event reason and `reason` contains the message.
if data.get("resource_kind") and data.get("name"):
if not normalized.get("message") and data.get("reason"):
normalized["message"] = data["reason"]
normalized["reason"] = data["name"]
elif not normalized.get("reason") and data.get("name"):
normalized["reason"] = data["name"]
if not normalized.get("first_timestamp") and data.get("start_time"):
normalized["first_timestamp"] = data["start_time"]
normalized["last_timestamp"] = data["start_time"]
if not normalized.get("involved_object") and (data.get("resource_kind") or data.get("resource_name")):
normalized["involved_object"] = {
"kind": data.get("resource_kind"),
"name": data.get("resource_name"),
}
try:
if int(normalized.get("count", 1)) <= 0:
normalized["count"] = 1
except (TypeError, ValueError):
normalized["count"] = 1
for field in ("type", "reason", "message"):
if isinstance(normalized.get(field), str):
normalized[field] = re.sub(r"<[^>]+>", "", normalized[field])
return normalized
def _normalize_involved_object(obj: Any) -> Optional[Dict[str, Any]]:
"""Normalize involvedObject field to consistent format."""
if not obj:
return None
if isinstance(obj, dict):
result = {}
# Handle both formats
for k, v in obj.items():
key_lower = k.lower()
if key_lower in ('kind', 'name', 'namespace'):
result[key_lower] = v
return result if result else None
return None
def _query_k8s_events_from_lts(
region: str,
cluster_id: str,
start_time: str,
end_time: str,
keywords: Optional[str] = None,
event_type: Optional[str] = None,
limit: int = 500,
ak: Optional[str] = None,
sk: Optional[str] = None,
project_id: Optional[str] = None
) -> Dict[str, Any]:
"""
Query K8s events from LTS based on LogConfig settings.
Args:
region: Huawei Cloud region
cluster_id: CCE cluster ID
start_time: Start time 'YYYY-MM-DD HH:MM:SS'
end_time: End time 'YYYY-MM-DD HH:MM:SS'
keywords: Optional keywords to filter events
event_type: Optional Event type (`Warning`, `Normal`, or `all`) to filter server-side
limit: Maximum number of events to return (default 500)
ak: Access key (optional, uses env if not provided)
sk: Secret key (optional, uses env if not provided)
project_id: Project ID (optional)
Returns:
Dict with success status, events, and metadata
"""
if not _lts_available:
return {
"success": False,
"error": "LTS query module is not available."
}
effective_event_type = event_type or "Warning"
if effective_event_type not in {"Warning", "Normal", "all"}:
return {"success": False, "error": "event_type must be Warning, Normal, or all"}
if effective_event_type != "all":
if keywords and keywords != effective_event_type:
return {
"success": False,
"error": "LTS supports one server-side keyword filter; use event_type=all before providing keywords",
}
keywords = effective_event_type
# Step 1: Read LogConfig CRs through kubectl-cce and find the Event-to-LTS rule.
logconfigs_result = cce_app_logs.get_cce_logconfigs_action({
"region": region,
"cluster_id": cluster_id,
"ak": ak,
"sk": sk,
"project_id": project_id,
})
if not logconfigs_result.get("success"):
return {
"success": False,
"error": f"Failed to get LogConfigs: {logconfigs_result.get('error', 'Unknown error')}",
}
event_config = next(
(
config for config in logconfigs_result.get("logconfigs") or []
if config.get("name") == "default-event"
and config.get("input_type") == "event"
and config.get("output_type") == "LTS"
),
None,
)
if not event_config:
return {
"success": False,
"error": "No default-event LogConfig with LTS output was found in the cluster.",
"checked_logconfigs": logconfigs_result.get("count", 0),
}
lts_config = (event_config.get("spec") or {}).get("outputDetail", {}).get("LTS", {})
log_group_id = lts_config.get("ltsGroupID")
log_stream_id = lts_config.get("ltsStreamID")
if not log_group_id or not log_stream_id:
return {
"success": False,
"error": "The default-event LogConfig does not contain LTS group and stream IDs.",
}
# Step 4: Convert time to milliseconds
try:
start_ms = _convert_timestamp_to_ms(start_time)
end_ms = _convert_timestamp_to_ms(end_time)
except ValueError as e:
return {
"success": False,
"error": f"Invalid UTC time format; expected 'YYYY-MM-DD HH:MM:SS': {e}",
}
# Step 5: Query LTS with pagination
all_events = []
scroll_id = None
total_fetched = 0
page_count = 0
page_limit = 1000 # LTS API page size
page_request_delay_seconds = 0.1
while total_fetched < limit:
page_count += 1
page_remaining = limit - total_fetched
current_page_limit = min(page_limit, page_remaining)
lts_result = lts_mod.query_logs(
region=region,
log_group_id=log_group_id,
log_stream_id=log_stream_id,
start_time=str(start_ms),
end_time=str(end_ms),
keywords=keywords,
limit=current_page_limit,
scroll_id=scroll_id,
ak=ak,
sk=sk,
project_id=project_id
)
if not lts_result.get("success"):
return {
"success": False,
"error": f"LTS query failed: {lts_result.get('error', 'Unknown error')}",
"log_group_id": log_group_id,
"log_stream_id": log_stream_id,
"events_fetched": total_fetched,
"pages_fetched": page_count - 1
}
# Step 6: Parse events from logs
raw_logs = lts_result.get("logs", [])
for log in raw_logs:
content = log.get("content", "")
if not content:
continue
parsed = _parse_event_content(content)
if not parsed:
continue
# Normalize involved object
involved_obj = parsed.get("involved_object")
if involved_obj:
parsed["involved_object"] = _normalize_involved_object(involved_obj)
all_events.append(parsed)
total_fetched += 1
if total_fetched >= limit:
break
# Check for next page
scroll_id = lts_result.get("scroll_id")
if not scroll_id:
break
if total_fetched < limit:
time.sleep(page_request_delay_seconds)
# Step 7: Build response
return {
"success": True,
"region": region,
"cluster_id": cluster_id,
"log_group_id": log_group_id,
"log_stream_id": log_stream_id,
"keywords": keywords,
"event_type": effective_event_type,
"event_count": len(all_events),
"events": all_events,
"time_range": {
"start": start_time,
"end": end_time
},
"log_config": {
"name": event_config.get("name"),
"namespace": event_config.get("namespace"),
"input_type": "event",
"discovery_method": "kubectl_cce_logconfig",
"access_method": logconfigs_result.get("access_method"),
},
"pagination": {
"pages_fetched": page_count,
"limit": limit,
"has_more": scroll_id is not None and total_fetched >= limit
}
}
def query_k8s_events_from_lts_action(params: Dict[str, str]) -> Dict[str, Any]:
"""
Action handler for huawei_query_k8s_events_from_lts tool.
Expected parameters:
- region: Huawei Cloud region (required)
- cluster_id: CCE cluster ID (required)
- start_time: Start time 'YYYY-MM-DD HH:MM:SS' (required)
- end_time: End time 'YYYY-MM-DD HH:MM:SS' (required)
- keywords: Optional keywords to filter events
- event_type: Optional Event type (`Warning`, `Normal`, or `all`) for server-side filtering
- limit: Maximum number of events to return (default 500)
"""
region = params.get("region")
cluster_id = params.get("cluster_id")
start_time = params.get("start_time")
end_time = params.get("end_time")
keywords = params.get("keywords")
event_type = params.get("event_type")
# Validate required parameters
if not region:
return {"success": False, "error": "region is required"}
if not cluster_id:
return {"success": False, "error": "cluster_id is required"}
if not start_time:
return {"success": False, "error": "start_time is required"}
if not end_time:
return {"success": False, "error": "end_time is required"}
# Parse limit parameter
try:
limit = int(params.get("limit", 500))
except (ValueError, TypeError):
limit = 500
return _query_k8s_events_from_lts(
region=region,
cluster_id=cluster_id,
start_time=start_time,
end_time=end_time,
keywords=keywords,
event_type=event_type,
limit=limit,
ak=params.get("ak"),
sk=params.get("sk"),
project_id=params.get("project_id")
)
scripts/huawei_cloud/cce.py
"""CCE helpers used by the current Kubernetes Event query."""
from __future__ import annotations
from typing import Any, Dict, Optional
from . import kubectl_client
def get_kubernetes_events(
region: str,
cluster_id: str,
ak: Optional[str] = None,
sk: Optional[str] = None,
project_id: Optional[str] = None,
security_token: Optional[str] = None,
namespace: Optional[str] = None,
event_type: Optional[str] = None,
limit: int = 500,
) -> Dict[str, Any]:
"""Read and normalize CCE Events through the kubectl access strategy."""
effective_event_type = event_type or "Warning"
result = kubectl_client.get_cce_events_with_kubectl(
region=region,
cluster_id=cluster_id,
namespace=namespace,
event_type=effective_event_type,
limit=limit,
ak=ak,
sk=sk,
project_id=project_id,
security_token=security_token,
)
if not result.get("success"):
return result
events = []
for item in result.get("items") or []:
metadata = item.get("metadata") or {}
involved_object = item.get("involvedObject") or {}
series = item.get("series") or {}
events.append(
{
"name": metadata.get("name"),
"namespace": metadata.get("namespace"),
"type": item.get("type"),
"reason": item.get("reason"),
"message": item.get("message"),
"first_timestamp": item.get("firstTimestamp") or item.get("eventTime") or metadata.get("creationTimestamp"),
"last_timestamp": item.get("lastTimestamp") or series.get("lastObservedTime") or item.get("eventTime"),
"count": item.get("count") or series.get("count") or 1,
"involved_object": {
"kind": involved_object.get("kind"),
"name": involved_object.get("name"),
"namespace": involved_object.get("namespace"),
}
if involved_object
else None,
}
)
return {
"success": True,
"region": region,
"cluster_id": cluster_id,
"action": "get_cce_events",
"namespace": namespace or "all",
"event_type": effective_event_type,
"access_method": result.get("access_method"),
"count": len(events),
"limit": limit,
"events": events,
}
scripts/huawei_cloud/common.py
"""Credential helpers for CCE Event queries."""
from __future__ import annotations
import os
import re
from typing import Optional
def get_credentials(
ak: Optional[str] = None, sk: Optional[str] = None, project_id: Optional[str] = None
) -> tuple[Optional[str], Optional[str], Optional[str]]:
"""Resolve explicit credentials before environment-variable fallback."""
return (
ak or os.environ.get("HUAWEI_AK") or os.environ.get("HUAWEICLOUD_SDK_AK") or os.environ.get("HW_ACCESS_KEY"),
sk or os.environ.get("HUAWEI_SK") or os.environ.get("HUAWEICLOUD_SDK_SK") or os.environ.get("HW_SECRET_KEY"),
project_id or os.environ.get("HUAWEI_PROJECT_ID") or os.environ.get("HUAWEICLOUD_SDK_PROJECT_ID") or os.environ.get("HW_PROJECT_ID"),
)
def has_hcloud_profile() -> bool:
"""Return whether a usable local hcloud profile is present."""
config_dir = os.environ.get("HCLOUD_CONFIG_DIR")
candidates = [os.path.join(config_dir, "config.json")] if config_dir else []
candidates.extend(
[
os.path.expanduser("~/.hcloud/config.json"),
os.path.expanduser("~/.hcloud/config.yaml"),
os.path.expanduser("~/.hcloud/config.yml"),
]
)
return any(os.path.isfile(path) and os.path.getsize(path) > 0 for path in candidates)
def resolve_hcloud_credentials(
ak: Optional[str] = None,
sk: Optional[str] = None,
project_id: Optional[str] = None,
) -> tuple[Optional[str], Optional[str], Optional[str]]:
"""Resolve hcloud auth in priority order: arguments, profile, environment."""
if ak or sk or project_id:
return ak, sk, project_id
if has_hcloud_profile():
return None, None, None
return get_credentials()
def redact_command(command: list[str]) -> list[str]:
"""Redact credential values before a command is returned to callers."""
return [
re.sub(r"(--cli-(?:access-key|secret-key|security-token)=).*", r"\1***", part)
for part in command
]
scripts/huawei_cloud/dispatcher.py
"""Dispatcher for the public CCE Kubernetes Event Analyzer tools."""
from __future__ import annotations
from typing import Any, Callable, Dict
from . import cce, cce_events_lts, event_analysis
Handler = Callable[[Dict[str, str]], Dict[str, Any]]
def _require(params: Dict[str, str], *keys: str) -> str | None:
missing = [key for key in keys if not params.get(key)]
if not missing:
return None
return f"{', '.join(missing)} are required" if len(missing) > 1 else f"{missing[0]} is required"
def _to_int(value: str | None, default: int) -> int:
try:
return int(value) if value is not None else default
except (TypeError, ValueError):
return default
def _get_cce_events(params: Dict[str, str]) -> Dict[str, Any]:
return cce.get_kubernetes_events(
region=params["region"],
cluster_id=params["cluster_id"],
namespace=params.get("namespace"),
event_type=params.get("event_type"),
limit=_to_int(params.get("limit"), 500),
ak=params.get("ak"),
sk=params.get("sk"),
project_id=params.get("project_id"),
security_token=params.get("security_token"),
)
ACTION_SPECS: Dict[str, tuple[tuple[str, ...], Handler]] = {
"huawei_get_cce_events": (("region", "cluster_id"), _get_cce_events),
"huawei_query_k8s_events_from_lts": (
("region", "cluster_id", "start_time", "end_time"),
cce_events_lts.query_k8s_events_from_lts_action,
),
"huawei_analyze_cce_events": ((), event_analysis.analyze_cce_events_action),
}
def is_registered_action(action: str) -> bool:
return action in ACTION_SPECS
def dispatch_action(action: str, params: Dict[str, str]) -> Dict[str, Any]:
required, handler = ACTION_SPECS[action]
error = _require(params, *required)
return {"success": False, "error": error} if error else handler(params)
scripts/huawei_cloud/event_analysis.py
"""Local aggregation for Kubernetes Event query results."""
from __future__ import annotations
import json
from collections import Counter, defaultdict
from typing import Any, Dict, Iterable, List, Optional
def _as_int(value: Any, default: int = 1) -> int:
try:
return max(int(value), 0)
except (TypeError, ValueError):
return default
def _event_times(event: Dict[str, Any]) -> List[Optional[str]]:
first = event.get("first_timestamp") or event.get("eventTime")
last = event.get("last_timestamp") or event.get("eventTime")
return [first, last] if first != last else [first]
def _parse_events(value: Optional[str]) -> List[Dict[str, Any]]:
if not value:
raise ValueError("events is required and must be a JSON array or an object containing an events array")
parsed = json.loads(value)
if isinstance(parsed, dict):
parsed = parsed.get("events")
if not isinstance(parsed, list):
raise ValueError("events must be a JSON array or an object containing an events array")
return [event for event in parsed if isinstance(event, dict)]
def _range(values: Iterable[Optional[str]]) -> Dict[str, Optional[str]]:
timestamps = sorted(value for value in values if value)
return {"first": timestamps[0] if timestamps else None, "last": timestamps[-1] if timestamps else None}
def _as_bool(value: Optional[str], default: bool) -> bool:
if value is None:
return default
return value.strip().lower() in {"1", "true", "yes", "on"}
def _query_events(params: Dict[str, str]) -> tuple[List[Dict[str, Any]], str, Dict[str, Any]]:
"""Query a supported Event source before local aggregation."""
source = (
params.get("event_source")
or ("lts" if params.get("start_time") or params.get("end_time") else "current")
).lower()
region = params.get("region")
cluster_id = params.get("cluster_id")
if source not in {"current", "lts"}:
raise ValueError("event_source must be current or lts")
if not region or not cluster_id:
raise ValueError("region and cluster_id are required when events is not provided")
if source == "current":
from . import cce
try:
limit = int(params.get("limit", "500"))
except ValueError:
raise ValueError("limit must be an integer")
result = cce.get_kubernetes_events(
region=region,
cluster_id=cluster_id,
namespace=params.get("namespace"),
event_type=params.get("event_type"),
limit=max(1, min(limit, 1000)),
ak=params.get("ak"),
sk=params.get("sk"),
project_id=params.get("project_id"),
security_token=params.get("security_token"),
)
else:
from . import cce_events_lts
if not params.get("start_time") or not params.get("end_time"):
raise ValueError("start_time and end_time are required when event_source=lts")
result = cce_events_lts.query_k8s_events_from_lts_action(params)
if not result.get("success"):
raise RuntimeError(result.get("error") or f"failed to query {source} events")
return result.get("events") or [], source, result
def analyze_cce_events_action(params: Dict[str, str]) -> Dict[str, Any]:
"""Query an Event source when needed, then aggregate its Event records locally."""
query_result: Optional[Dict[str, Any]] = None
try:
if params.get("events"):
events = _parse_events(params.get("events"))
source = params.get("event_source") or "provided_events"
else:
events, source, query_result = _query_events(params)
except (ValueError, json.JSONDecodeError, RuntimeError) as exc:
return {"success": False, "error": str(exc)}
try:
max_groups = max(1, min(int(params.get("max_groups", "10")), 100))
except ValueError:
return {"success": False, "error": "max_groups must be an integer between 1 and 100"}
reason_counts: Counter[str] = Counter()
reason_warnings: Counter[str] = Counter()
reason_times: Dict[str, List[Optional[str]]] = defaultdict(list)
namespace_counts: Counter[str] = Counter()
object_counts: Counter[str] = Counter()
type_counts: Counter[str] = Counter()
repeated_patterns: List[Dict[str, Any]] = []
for event in events:
event_type = str(event.get("type") or "Unknown")
reason = str(event.get("reason") or "Unknown")
namespace = str(event.get("namespace") or (event.get("involved_object") or {}).get("namespace") or "unknown")
occurrences = _as_int(event.get("count"))
involved_object = event.get("involved_object") or event.get("involvedObject") or {}
if not isinstance(involved_object, dict):
involved_object = {}
object_key = "/".join(
filter(
None,
(
namespace,
str(involved_object.get("kind") or "Unknown"),
str(involved_object.get("name") or "Unknown"),
),
)
)
reason_counts[reason] += occurrences
namespace_counts[namespace] += occurrences
object_counts[object_key] += occurrences
type_counts[event_type] += occurrences
reason_times[reason].extend(_event_times(event))
if event_type.lower() == "warning":
reason_warnings[reason] += occurrences
if occurrences > 1:
repeated_patterns.append(
{
"reason": reason,
"type": event_type,
"namespace": namespace,
"involved_object": involved_object or None,
"count": occurrences,
"first_timestamp": event.get("first_timestamp"),
"last_timestamp": event.get("last_timestamp"),
}
)
top_reasons = [
{
"reason": reason,
"count": count,
"warning_count": reason_warnings[reason],
"time_range": _range(reason_times[reason]),
}
for reason, count in reason_counts.most_common(max_groups)
]
repeated_patterns.sort(key=lambda item: item["count"], reverse=True)
response = {
"success": True,
"source": source,
"event_records": len(events),
"total_occurrences": sum(type_counts.values()),
"event_type_breakdown": dict(type_counts.most_common()),
"warning_count": type_counts.get("Warning", 0),
"normal_count": type_counts.get("Normal", 0),
"time_range": _range(timestamp for event in events for timestamp in _event_times(event)),
"top_reasons": top_reasons,
"namespace_breakdown": [
{"namespace": namespace, "count": count}
for namespace, count in namespace_counts.most_common(max_groups)
],
"affected_objects": [
{"object": object_key, "count": count}
for object_key, count in object_counts.most_common(max_groups)
],
"repeated_patterns": repeated_patterns[:max_groups],
}
region = params.get("region") or (query_result or {}).get("region")
cluster_id = params.get("cluster_id") or (query_result or {}).get("cluster_id")
should_check_resources = _as_bool(params.get("check_resource_status"), bool(region and cluster_id))
if should_check_resources and region and cluster_id:
from . import resource_status
try:
response["resource_status"] = resource_status.check_event_resource_statuses(
events=events,
region=region,
cluster_id=cluster_id,
max_resources=max_groups,
ak=params.get("ak"),
sk=params.get("sk"),
project_id=params.get("project_id"),
security_token=params.get("security_token"),
)
except Exception as exc:
response["resource_status"] = {
"checked": 0,
"summary": {"query_failed": 1},
"resources": [],
"message": f"Resource status checks failed: {exc}",
}
else:
response["resource_status"] = {
"checked": 0,
"summary": {},
"resources": [],
"message": "Resource status checks require region and cluster_id, or can be disabled with check_resource_status=false",
}
if query_result is not None:
response["query"] = {
"source": source,
"region": query_result.get("region"),
"cluster_id": query_result.get("cluster_id"),
"event_count": len(events),
"access_method": query_result.get("access_method"),
"time_range": query_result.get("time_range"),
"log_config": query_result.get("log_config"),
}
return response
scripts/huawei_cloud/kubectl_client.py
"""kubectl-based read-only Kubernetes access for CCE event queries."""
from __future__ import annotations
import json
import os
import re
import subprocess
import tempfile
from typing import Any, Dict, List, Optional
from . import common
_NAMESPACE_PATTERN = re.compile(r"^[a-z0-9](?:[-a-z0-9]*[a-z0-9])?$")
def _parse_json_output(output: str, source: str) -> Dict[str, Any]:
text = (output or "").strip()
try:
return {"success": True, "data": json.loads(text or "{}")}
except json.JSONDecodeError as exc:
return {"success": False, "error": f"{source} returned non-JSON output: {exc}"}
def _run_command(cmd: List[str], env: Optional[Dict[str, str]] = None, timeout: int = 60) -> Dict[str, Any]:
safe_cmd = common.redact_command(cmd)
try:
proc = subprocess.run(cmd, text=True, capture_output=True, timeout=timeout, env=env)
except FileNotFoundError:
return {"success": False, "error": f"{cmd[0]} not found in PATH", "command": safe_cmd}
except subprocess.TimeoutExpired:
return {"success": False, "error": f"command timed out after {timeout}s", "command": safe_cmd}
if proc.returncode:
return {
"success": False,
"error": (proc.stderr or proc.stdout or f"command exited with code {proc.returncode}")[:2000],
"command": safe_cmd,
"returncode": proc.returncode,
}
result = _parse_json_output(proc.stdout, cmd[0])
result["command"] = safe_cmd
return result
def _run_hcloud(
service: str,
operation: str,
region: str,
params: Dict[str, Any],
ak: Optional[str],
sk: Optional[str],
project_id: Optional[str],
) -> Dict[str, Any]:
access_key, secret_key, resolved_project_id = common.resolve_hcloud_credentials(ak, sk, project_id)
cmd = [
"hcloud", service, operation, f"--cli-region={region}", "--cli-output=json",
"--cli-connect-timeout=10", "--cli-read-timeout=60",
]
if access_key:
cmd.append(f"--cli-access-key={access_key}")
if secret_key:
cmd.append(f"--cli-secret-key={secret_key}")
if resolved_project_id:
cmd.append(f"--cli-project-id={resolved_project_id}")
for key, value in params.items():
if value is not None:
cmd.append(f"--{key}={value}")
return _run_command(cmd)
def _cluster_has_external_access(cluster: Dict[str, Any]) -> bool:
status = cluster.get("status") or {}
for condition in status.get("conditions", []) or []:
if condition.get("type") == "ElasticPublicIP":
return condition.get("status") not in {"UNBOUND", "False", "", None}
return any(
endpoint.get("type") == "External" and endpoint.get("url")
for endpoint in status.get("endpoints", []) or []
)
def _prefer_external_context(kubeconfig: Dict[str, Any]) -> None:
external_name = next(
(
item.get("name") for item in kubeconfig.get("clusters", []) or []
if "external" in item.get("name", "") and "TLS" not in item.get("name", "")
),
None,
)
if not external_name:
return
for context in kubeconfig.get("contexts", []) or []:
if (context.get("context") or {}).get("cluster") == external_name:
kubeconfig["current-context"] = context.get("name")
return
def _get_events_with_external_kubeconfig(
region: str, cluster_id: str, args: List[str], ak: Optional[str], sk: Optional[str], project_id: Optional[str]
) -> Dict[str, Any]:
cluster_result = _run_hcloud("CCE", "ShowCluster", region, {"cluster_id": cluster_id}, ak, sk, project_id)
if not cluster_result.get("success"):
return cluster_result
if not _cluster_has_external_access(cluster_result.get("data") or {}):
return {"success": False, "error": "cluster has no bound EIP/external endpoint"}
cert_result = _run_hcloud(
"CCE", "CreateKubernetesClusterCert", region, {"cluster_id": cluster_id, "duration": 1}, ak, sk, project_id
)
if not cert_result.get("success"):
return cert_result
kubeconfig = cert_result.get("data") or {}
if not kubeconfig.get("clusters"):
return {"success": False, "error": "CreateKubernetesClusterCert returned no kubeconfig clusters"}
_prefer_external_context(kubeconfig)
kubeconfig_file = None
try:
with tempfile.NamedTemporaryFile("w", delete=False, suffix=".json") as handle:
json.dump(kubeconfig, handle)
kubeconfig_file = handle.name
result = _run_command(["kubectl", "--kubeconfig", kubeconfig_file, "get", *args, "-o", "json"])
if result.get("success"):
result["access_method"] = "kubectl_kubeconfig_external"
return result
finally:
if kubeconfig_file and os.path.exists(kubeconfig_file):
os.remove(kubeconfig_file)
def _get_events_with_cce_plugin(
region: str, cluster_id: str, args: List[str], ak: Optional[str], sk: Optional[str], project_id: Optional[str], security_token: Optional[str]
) -> Dict[str, Any]:
env_ak, env_sk, env_project_id = common.get_credentials()
access_key = ak or env_ak
secret_key = sk or env_sk
resolved_project_id = project_id or env_project_id
env = os.environ.copy()
env.update({"CCE_CLUSTER_ID": cluster_id, "CCE_REGION": region, "HW_REGION": region})
if resolved_project_id:
env.update({"CCE_PROJECT_ID": resolved_project_id, "HW_PROJECT_ID": resolved_project_id})
if access_key:
env.update({"HW_ACCESS_KEY": access_key, "HUAWEICLOUD_SDK_AK": access_key})
if secret_key:
env.update({"HW_SECRET_KEY": secret_key, "HUAWEICLOUD_SDK_SK": secret_key})
if security_token:
env.update({"HW_SECURITY_TOKEN": security_token, "HUAWEICLOUD_SECURITY_TOKEN": security_token})
command = ["kubectl", "cce", "--cluster-id", cluster_id, "--region", region]
if resolved_project_id:
command.extend(["--project-id", resolved_project_id])
command.extend(["get", *args, "-o", "json"])
result = _run_command(command, env=env)
if result.get("success"):
result["access_method"] = "kubectl_cce_plugin"
return result
def get_cce_events_with_kubectl(
region: str,
cluster_id: str,
namespace: Optional[str] = None,
event_type: Optional[str] = None,
limit: int = 500,
ak: Optional[str] = None,
sk: Optional[str] = None,
project_id: Optional[str] = None,
security_token: Optional[str] = None,
) -> Dict[str, Any]:
"""Read Kubernetes Events through external kubeconfig, then kubectl-cce."""
if not cluster_id:
return {"success": False, "error": "cluster_id is required"}
if namespace and (len(namespace) > 63 or not _NAMESPACE_PATTERN.fullmatch(namespace)):
return {
"success": False,
"error": "namespace must be a Kubernetes DNS label (lowercase letters, digits, and hyphens; max 63 characters)",
}
effective_event_type = event_type or "Warning"
if effective_event_type not in {"Warning", "Normal", "all"}:
return {"success": False, "error": "event_type must be Warning, Normal, or all"}
args = ["events", "-n", namespace] if namespace else ["events", "-A"]
if effective_event_type != "all":
args.extend(["--field-selector", f"type={effective_event_type}"])
external_result = _get_events_with_external_kubeconfig(region, cluster_id, args, ak, sk, project_id)
if external_result.get("success"):
result = external_result
else:
token = security_token or os.environ.get("HUAWEI_SECURITY_TOKEN") or os.environ.get("HW_SECURITY_TOKEN")
plugin_result = _get_events_with_cce_plugin(region, cluster_id, args, ak, sk, project_id, token)
if not plugin_result.get("success"):
return {
"success": False,
"error": "failed to get Kubernetes events via external kubeconfig or kubectl cce plugin",
"kubeconfig_error": external_result.get("error"),
"plugin_error": plugin_result.get("error"),
}
result = plugin_result
return {
"success": True,
"region": region,
"cluster_id": cluster_id,
"namespace": namespace or "all",
"event_type": effective_event_type,
"access_method": result.get("access_method"),
"items": ((result.get("data") or {}).get("items") or [])[:limit],
}
def get_cce_resource_with_kubectl(
region: str,
cluster_id: str,
resource: str,
name: str,
namespace: Optional[str] = None,
ak: Optional[str] = None,
sk: Optional[str] = None,
project_id: Optional[str] = None,
security_token: Optional[str] = None,
) -> Dict[str, Any]:
"""Read one Kubernetes resource through external kubeconfig or kubectl-cce."""
if not cluster_id:
return {"success": False, "error": "cluster_id is required"}
if not resource or not name:
return {"success": False, "error": "resource and name are required"}
if namespace and (len(namespace) > 63 or not _NAMESPACE_PATTERN.fullmatch(namespace)):
return {
"success": False,
"error": "namespace must be a Kubernetes DNS label (lowercase letters, digits, and hyphens; max 63 characters)",
}
args = [resource, name]
if namespace:
args.extend(["-n", namespace])
external_result = _get_events_with_external_kubeconfig(region, cluster_id, args, ak, sk, project_id)
if external_result.get("success"):
result = external_result
else:
token = security_token or os.environ.get("HUAWEI_SECURITY_TOKEN") or os.environ.get("HW_SECURITY_TOKEN")
plugin_result = _get_events_with_cce_plugin(region, cluster_id, args, ak, sk, project_id, token)
if not plugin_result.get("success"):
return {
"success": False,
"error": f"failed to get Kubernetes resource {resource}/{name}",
"kubeconfig_error": external_result.get("error"),
"plugin_error": plugin_result.get("error"),
}
result = plugin_result
return {
"success": True,
"access_method": result.get("access_method"),
"item": result.get("data") or {},
}
def get_cce_logconfigs_with_cce_plugin(
region: str,
cluster_id: str,
ak: Optional[str] = None,
sk: Optional[str] = None,
project_id: Optional[str] = None,
security_token: Optional[str] = None,
) -> Dict[str, Any]:
"""Read LogConfig CRs through the kubectl-cce plugin."""
if not cluster_id:
return {"success": False, "error": "cluster_id is required"}
token = security_token or os.environ.get("HUAWEI_SECURITY_TOKEN") or os.environ.get("HW_SECURITY_TOKEN")
result = _get_events_with_cce_plugin(
region,
cluster_id,
["logconfigs.logging.openvessel.io", "-A"],
ak,
sk,
project_id,
token,
)
if not result.get("success"):
return {
"success": False,
"error": "failed to get LogConfigs through kubectl cce plugin",
"plugin_error": result.get("error"),
}
return {
"success": True,
"region": region,
"cluster_id": cluster_id,
"access_method": result.get("access_method"),
"items": (result.get("data") or {}).get("items") or [],
}
scripts/huawei_cloud/lts.py
"""LTS log queries through hcloud for historical Kubernetes Event queries."""
from __future__ import annotations
import json
import subprocess
from datetime import datetime, timedelta
from typing import Any, Dict, Optional
from . import common
def _timestamp(value: Optional[str], default: datetime) -> int:
if not value:
return int(default.timestamp() * 1000)
if "-" in value:
return int(datetime.strptime(value, "%Y-%m-%d %H:%M:%S").timestamp() * 1000)
return int(value)
def _parse_hcloud_json(output: str) -> Dict[str, Any]:
text = (output or "").strip()
return json.loads(text or "{}")
def _run_hcloud(cmd: list[str]) -> Dict[str, Any]:
safe_cmd = common.redact_command(cmd)
try:
completed = subprocess.run(cmd, text=True, capture_output=True, timeout=75, check=False)
except FileNotFoundError:
return {"success": False, "error": "hcloud not found in PATH", "command": safe_cmd}
except subprocess.TimeoutExpired:
return {"success": False, "error": "hcloud command timed out after 75 seconds", "command": safe_cmd}
if completed.returncode:
return {
"success": False,
"error": (completed.stderr or completed.stdout or f"hcloud exited with code {completed.returncode}")[:2000],
"command": safe_cmd,
}
try:
return {"success": True, "data": _parse_hcloud_json(completed.stdout)}
except (ValueError, json.JSONDecodeError) as exc:
return {
"success": False,
"error": f"hcloud response parsing failed: {exc}",
"command": safe_cmd,
}
def _hcloud_base_command(
service: str,
operation: str,
region: str,
ak: Optional[str],
sk: Optional[str],
project_id: Optional[str],
) -> list[str]:
access_key, secret_key, resolved_project_id = common.resolve_hcloud_credentials(ak, sk, project_id)
cmd = [
"hcloud", service, operation, f"--cli-region={region}", "--cli-output=json",
"--cli-connect-timeout=10", "--cli-read-timeout=60",
]
if resolved_project_id:
cmd.append(f"--cli-project-id={resolved_project_id}")
if access_key:
cmd.append(f"--cli-access-key={access_key}")
if secret_key:
cmd.append(f"--cli-secret-key={secret_key}")
return cmd
def query_logs(
region: str,
log_group_id: str,
log_stream_id: str,
start_time: Optional[str] = None,
end_time: Optional[str] = None,
keywords: Optional[str] = None,
limit: int = 1000,
scroll_id: Optional[str] = None,
ak: Optional[str] = None,
sk: Optional[str] = None,
project_id: Optional[str] = None,
**_: Any,
) -> Dict[str, Any]:
"""Query one LTS stream through ``hcloud LTS ListLogs``."""
now = datetime.now()
cmd = _hcloud_base_command("LTS", "ListLogs", region, ak, sk, project_id)
cmd.extend(
[
f"--log_group_id={log_group_id}",
f"--log_stream_id={log_stream_id}",
f"--start_time={_timestamp(start_time, now - timedelta(hours=1))}",
f"--end_time={_timestamp(end_time, now)}",
f"--limit={max(1, min(limit, 1000))}",
"--is_desc=true",
]
)
if keywords:
cmd.append(f"--keywords={keywords}")
if scroll_id:
cmd.append(f"--scroll_id={scroll_id}")
query_result = _run_hcloud(cmd)
if not query_result.get("success"):
return query_result
response = query_result["data"]
logs = [
{
"content": item.get("content", ""),
"timestamp": item.get("timestamp"),
"log_group_id": log_group_id,
"log_stream_id": log_stream_id,
}
for item in (response.get("logs") or [])
if isinstance(item, dict)
]
next_scroll_id = response.get("scroll_id")
return {
"success": True,
"log_group_id": log_group_id,
"log_stream_id": log_stream_id,
"total": len(logs),
"scroll_id": next_scroll_id,
"has_more": bool(next_scroll_id),
"logs": logs,
}
scripts/huawei_cloud/resource_status.py
"""Current-state checks for Kubernetes resources referenced by Events."""
from __future__ import annotations
from collections import Counter
from typing import Any, Dict, Iterable, Optional
from . import kubectl_client
_RESOURCE_MAP = {
"pod": ("pods", True),
"node": ("nodes", False),
"deployment": ("deployments.apps", True),
"statefulset": ("statefulsets.apps", True),
"daemonset": ("daemonsets.apps", True),
"replicaset": ("replicasets.apps", True),
"job": ("jobs.batch", True),
"cronjob": ("cronjobs.batch", True),
"persistentvolumeclaim": ("persistentvolumeclaims", True),
"persistentvolume": ("persistentvolumes", False),
"service": ("services", True),
}
def _conditions(item: Dict[str, Any]) -> Dict[str, str]:
return {
str(condition.get("type")): str(condition.get("status"))
for condition in ((item.get("status") or {}).get("conditions") or [])
if isinstance(condition, dict) and condition.get("type")
}
def _count(value: Any) -> int:
try:
return int(value or 0)
except (TypeError, ValueError):
return 0
def _status(item: Dict[str, Any], kind: str) -> tuple[str, str]:
spec = item.get("spec") or {}
status = item.get("status") or {}
conditions = _conditions(item)
if kind == "pod":
phase = status.get("phase")
if phase == "Succeeded":
return "normal", "Pod completed successfully"
if phase == "Running" and conditions.get("Ready") == "True":
return "normal", "Pod is Running and Ready"
if phase in {"Pending", "Failed", "Unknown"} or conditions.get("Ready") == "False":
return "abnormal", f"Pod phase is {phase or 'Unknown'}"
return "unknown", f"Pod phase is {phase or 'Unknown'}"
if kind == "node":
pressure = ("MemoryPressure", "DiskPressure", "PIDPressure")
if conditions.get("Ready") == "True" and all(conditions.get(name) != "True" for name in pressure):
return "normal", "Node is Ready without resource pressure"
if conditions.get("Ready") == "False":
return "abnormal", "Node is not Ready"
return "unknown", "Node readiness is not reported"
if kind in {"deployment", "replicaset"}:
desired = _count(spec.get("replicas", 1))
available = _count(status.get("availableReplicas", status.get("readyReplicas", 0)))
return ("normal", "All desired replicas are available") if available >= desired else (
"abnormal", f"Available replicas {available}/{desired}"
)
if kind == "statefulset":
desired = _count(spec.get("replicas", 1))
ready = _count(status.get("readyReplicas"))
return ("normal", "All desired replicas are Ready") if ready >= desired else (
"abnormal", f"Ready replicas {ready}/{desired}"
)
if kind == "daemonset":
desired = _count(status.get("desiredNumberScheduled"))
ready = _count(status.get("numberReady"))
if desired == 0:
return "normal", "No Pods are currently scheduled"
return ("normal", "All scheduled Pods are Ready") if ready >= desired else (
"abnormal", f"Ready Pods {ready}/{desired}"
)
if kind == "job":
if conditions.get("Complete") == "True":
return "normal", "Job completed successfully"
if conditions.get("Failed") == "True":
return "abnormal", "Job has failed"
return "unknown", "Job is still active or has no terminal condition"
if kind == "persistentvolumeclaim":
phase = status.get("phase")
return ("normal", "PVC is Bound") if phase == "Bound" else ("abnormal", f"PVC phase is {phase or 'Unknown'}")
return "unknown", "No health rule is defined for this resource kind"
def check_event_resource_statuses(
events: Iterable[Dict[str, Any]],
region: str,
cluster_id: str,
max_resources: int,
ak: Optional[str] = None,
sk: Optional[str] = None,
project_id: Optional[str] = None,
security_token: Optional[str] = None,
) -> Dict[str, Any]:
"""Check the current state of distinct resources referenced by Event records."""
seen = set()
results = []
for event in events:
involved = event.get("involved_object") or event.get("involvedObject") or {}
if not isinstance(involved, dict):
continue
kind = str(involved.get("kind") or "")
name = str(involved.get("name") or "")
namespace = involved.get("namespace") or event.get("namespace")
key = (kind.lower(), str(namespace or ""), name)
if not kind or not name or key in seen:
continue
seen.add(key)
if len(results) >= max_resources:
break
mapping = _RESOURCE_MAP.get(kind.lower())
base = {"kind": kind, "name": name, "namespace": namespace}
if not mapping:
results.append({**base, "state": "unsupported", "message": "Resource kind is not supported for status checks"})
continue
resource, namespaced = mapping
if namespaced and not namespace:
results.append({**base, "state": "query_failed", "message": "Event does not include the resource namespace"})
continue
lookup = kubectl_client.get_cce_resource_with_kubectl(
region=region,
cluster_id=cluster_id,
resource=resource,
name=name,
namespace=str(namespace) if namespaced and namespace else None,
ak=ak,
sk=sk,
project_id=project_id,
security_token=security_token,
)
if not lookup.get("success"):
error = lookup.get("plugin_error") or lookup.get("error") or "resource query failed"
state = "not_found" if "NotFound" in error or "not found" in error.lower() else "query_failed"
results.append({**base, "state": state, "message": error[:500]})
continue
state, message = _status(lookup.get("item") or {}, kind.lower())
results.append({**base, "state": state, "message": message, "access_method": lookup.get("access_method")})
counts = Counter(item["state"] for item in results)
return {"checked": len(results), "summary": dict(counts), "resources": results}
scripts/huawei-cloud.py
#!/usr/bin/env python3
"""Command-line entry point for CCE Kubernetes Event queries."""
from __future__ import annotations
import json
import sys
from pathlib import Path
from typing import Dict, List
def _parse_cli_params(args: List[str]) -> Dict[str, str]:
"""Parse key=value and --key=value/--key value arguments."""
params: Dict[str, str] = {}
index = 0
while index < len(args):
argument = args[index]
if argument.startswith("--"):
normalized = argument[2:]
if "=" in normalized:
key, value = normalized.split("=", 1)
params[key.replace("-", "_")] = value
elif index + 1 < len(args) and not args[index + 1].startswith("--"):
params[normalized.replace("-", "_")] = args[index + 1]
index += 1
else:
params[normalized.replace("-", "_")] = "true"
elif "=" in argument:
key, value = argument.split("=", 1)
params[key.lstrip("-").replace("-", "_")] = value
index += 1
return params
def main() -> int:
if len(sys.argv) < 2:
print(json.dumps({"success": False, "error": "action is required"}))
return 1
script_dir = str(Path(__file__).resolve().parent)
if script_dir not in sys.path:
sys.path.insert(0, script_dir)
from huawei_cloud.dispatcher import dispatch_action, is_registered_action
action = sys.argv[1]
if not is_registered_action(action):
print(json.dumps({"success": False, "error": f"unknown action: {action}"}))
return 1
print(json.dumps(dispatch_action(action, _parse_cli_params(sys.argv[2:])), ensure_ascii=True, indent=2))
return 0
if __name__ == "__main__":
raise SystemExit(main())
skill-profile.yaml
name: huawei-cloud-cce-kubernetes-event-analyzer
level: L2
domain: observability
description: Query and analyze Kubernetes Events in Huawei Cloud CCE clusters. Trigger when users ask about CCE events, Kubernetes warning events, FailedScheduling, FailedMount, ImagePullBackOff, event patterns, historical events in LTS, or event-based diagnosis for a CCE cluster or namespace.
tools:
- huawei_get_cce_events
- huawei_query_k8s_events_from_lts
- huawei_analyze_cce_events
references:
- references/workflow.md
- references/risk-rules.md
- references/output-schema.md
- references/acceptance-criteria.md
- references/kubectl-cce.md
guardrails:
max_auto_risk: R3
SKILL.md
---
name: huawei-cloud-cce-kubernetes-event-analyzer
description: Query and analyze Kubernetes Events in Huawei Cloud CCE clusters. Trigger when users ask about CCE events, Kubernetes warning events, FailedScheduling, FailedMount, ImagePullBackOff, event patterns, historical events in LTS, or event-based diagnosis for a CCE cluster or namespace.
tags: [CCE, Kubernetes, events, observability]
---
# Huawei Cloud CCE Kubernetes Event Analyzer
## Overview
Query and analyze Kubernetes Events in Huawei Cloud CCE clusters to identify warnings, repeated failure patterns, affected resources, and useful diagnosis handoffs. The skill supports a current Event view through `kubectl` and a historical Event view through LTS.
**Architecture**: `python3 scripts/huawei-cloud.py` dispatcher -> `kubectl` through external kubeconfig or `kubectl cce` for current Events / `kubectl cce` LogConfig discovery plus `hcloud LTS ListLogs` for historical Events -> filtering and grouping -> diagnosis handoff.
**Execution Method**: Invoke only the bundled dispatcher. Do not query Kubernetes Events with raw Python Kubernetes SDK calls, direct Kubernetes API calls, or ad hoc cloud commands. The `huawei_get_cce_events` implementation invokes `kubectl` internally: external kubeconfig access first, then the `kubectl cce` plugin fallback.
**Related Skills**:
- `huawei-cloud-kubectl-cce-installer` - Install `kubectl` and the `kubectl-cce` plugin required for cluster access
- `huawei-cloud-cce-metric-analyzer` - CCE and cloud-resource metrics
**Capabilities**:
- Query current Kubernetes Events across a cluster or in a namespace
- Read Events through external `kubectl` kubeconfig access or `kubectl cce`
- Query historical Event records from LTS within an explicit time window
- Filter and group Events by type, reason, namespace, resource, and timestamps
- Check the current status of supported resources referenced by Events
- Analyze a supplied current or historical Event result locally without another cloud request
- Identify repeated warning patterns and hand off evidence to diagnosis skills
**Typical Use Cases**:
- "List Warning events for this CCE cluster"
- "Find repeated FailedScheduling events in namespace default"
- "Query historical ImagePullBackOff events from LTS"
- "Analyze the top Kubernetes event reasons during an incident"
## Prerequisites
### 1. Runtime Dependencies
- Python 3.8+ for the dispatcher and result processing
- `hcloud` (KooCLI) for cluster lookup and temporary external kubeconfig generation
- `kubectl` for current Event reads
- `kubectl-cce` when the cluster has no usable external endpoint; see [kubectl-cce.md](references/kubectl-cce.md)
- `hcloud` LTS command support and the Cloud Native Log Collection add-on (`log-agent`) with a `default-event` Event-to-LTS `LogConfig`. `huawei_query_k8s_events_from_lts` reads `logconfigs.logging.openvessel.io` through `kubectl cce`, then invokes `hcloud LTS ListLogs` using the configured LTS IDs.
### 2. Credential Configuration
- External kubeconfig access uses hcloud credential priority: explicit tool parameters > local hcloud profile > environment variables.
- The `kubectl cce` fallback requires AK/SK and the target cluster's `project_id` from explicit tool parameters or environment variables; encrypted hcloud profile credentials cannot be reused by the plugin. When `project_id` is available, the implementation passes it explicitly as `kubectl cce --project-id <project-id>`.
- LTS queries require valid Huawei Cloud credentials and an authorized project.
**Security Rules**:
- Never print, persist, or hardcode AK/SK, security tokens, kubeconfig content, or temporary client credentials.
- Never use `echo $HUAWEI_AK` or `echo $HUAWEI_SK` to inspect credentials.
- Prefer a local hcloud profile for external kubeconfig access.
- Use least-privilege IAM identities and read-only Kubernetes RBAC permissions.
**Optional Environment Fallback**:
```bash
export HUAWEI_AK=<your-ak>
export HUAWEI_SK=<your-sk>
export HUAWEI_REGION=cn-north-4
export HUAWEI_PROJECT_ID=<project-id>
export HUAWEI_SECURITY_TOKEN=<security-token>
```
### 3. IAM Permission Requirements
| Permission | Purpose |
| ---------- | ------- |
| `cce:cluster:get` | Inspect cluster external endpoint availability |
| `cce:cluster:createCert` | Generate temporary kubeconfig for external `kubectl` access |
| `lts:logs:search` | Query historical Event records in LTS |
The effective Kubernetes identity also needs read-only `get` and `list` permission for Events in the target namespace or cluster.
**Permission Failure Handling**:
1. Report the failed operation and required permission.
2. Ask the user to grant the missing IAM or Kubernetes RBAC permission.
3. Do not retry until the user confirms the permission is ready.
## Core Commands
All commands use the bundled dispatcher:
```bash
python3 scripts/huawei-cloud.py <tool-name> key=value key=value
```
## KooCLI Command Format Standard
Users invoke the dispatcher rather than raw `hcloud` commands. For current Event queries, the dispatcher internally uses hcloud only to inspect the CCE cluster and generate a temporary external kubeconfig when appropriate.
```bash
python3 scripts/huawei-cloud.py huawei_get_cce_events \
region=cn-north-4 cluster_id=<cluster-id>
```
Follow these rules:
- Use `key=value` parameters and quote values containing spaces or special shell characters.
- Do not print or persist credentials, security tokens, or temporary kubeconfig files.
- Use exact `cluster_id` values for cluster-scoped queries.
- Keep LTS queries time-bounded with both `start_time` and `end_time`.
### 1. Current Kubernetes Events
```bash
# Query Warning Events (default)
python3 scripts/huawei-cloud.py huawei_get_cce_events \
region=cn-north-4 cluster_id=<cluster-id>
# Query Events in a namespace
python3 scripts/huawei-cloud.py huawei_get_cce_events \
region=cn-north-4 cluster_id=<cluster-id> namespace=default
# Limit returned Event records
python3 scripts/huawei-cloud.py huawei_get_cce_events \
region=cn-north-4 cluster_id=<cluster-id> limit=100
# Query all Event types only when explicitly needed
python3 scripts/huawei-cloud.py huawei_get_cce_events \
region=cn-north-4 cluster_id=<cluster-id> event_type=all limit=100
```
The tool returns only Warning Events by default, using the Kubernetes API server-side field selector. It first uses the external endpoint with a temporary kubeconfig; it then falls back to `kubectl cce`. For large clusters, full Event history can be substantial; query all types only after the user explicitly requests it with `event_type=all`.
### 2. Historical Events From LTS
```bash
# Query an explicit historical window
python3 scripts/huawei-cloud.py huawei_query_k8s_events_from_lts \
region=cn-north-4 cluster_id=<cluster-id> \
start_time="2026-05-30 06:00:00" \
end_time="2026-05-30 08:00:00"
# Query with an LTS keyword filter
python3 scripts/huawei-cloud.py huawei_query_k8s_events_from_lts \
region=cn-north-4 cluster_id=<cluster-id> \
start_time="2026-05-30 00:00:00" \
end_time="2026-05-30 23:59:59" \
keywords=FailedScheduling
```
LTS time format is UTC `YYYY-MM-DD HH:MM:SS`; the tool always interprets input values as UTC, not the local time zone of the host. The cluster must have the Cloud Native Log Collection add-on (`log-agent`) installed and healthy with the `default-event` Event-to-LTS `LogConfig`. The tool uses `kubectl cce --cluster-id <cluster-id> --region <region> get logconfigs.logging.openvessel.io -A -o json`, selects `default-event`, and reads `outputDetail.LTS.ltsGroupID` and `ltsStreamID`. LTS queries default to `event_type=Warning`, using `Warning` as a server-side keyword filter. For large clusters, request full Event history only after user confirmation with `event_type=all`; this removes the type keyword filter. LTS filtering is keyword matching, not a structured-field selector.
### 3. Query and Analyze Event Results
Without `events`, the tool queries and analyzes current cluster Events by default. For historical requests spanning more than one hour, use LTS with a bounded time window. Providing `start_time` or `end_time` automatically selects LTS; `event_source=lts` may also be set explicitly. Passing an `events` array (or a complete response object containing it) retains offline analysis behavior.
```bash
# Query and analyze current Events
python3 scripts/huawei-cloud.py huawei_analyze_cce_events \
region=cn-north-4 cluster_id=<cluster-id>
# Query and analyze historical LTS Events
python3 scripts/huawei-cloud.py huawei_analyze_cce_events \
region=cn-north-4 cluster_id=<cluster-id> event_source=lts \
start_time="2026-05-30 06:00:00" end_time="2026-05-30 08:00:00"
# Analyze supplied Events without a cloud query
python3 scripts/huawei-cloud.py huawei_analyze_cce_events \
events='[{"type":"Warning","reason":"FailedScheduling","namespace":"default","count":3}]' \
max_groups=10
```
## Risk Levels
This skill is read-only. It never changes cloud resources, Kubernetes resources, LTS configuration, or local cluster access configuration.
| Level | Meaning | Execution Guidance |
| ----- | ------- | ------------------ |
| R3 | Read-only Event query or local Event analysis | May run automatically |
| Tool | Operation Type | Risk Level | Description |
| ---- | -------------- | ---------- | ----------- |
| `huawei_get_cce_events` | Query | R3 | Query current cluster or namespace Events through `kubectl` |
| `huawei_query_k8s_events_from_lts` | Query | R3 | Query historical Event records from configured LTS collection |
| `huawei_analyze_cce_events` | Query and analyze | R3 | Query current or LTS Events when needed, then aggregate by type, reason, namespace, and resource |
## Parameter Reference
### Common Parameters
| Parameter | Required/Optional | Description | Default |
| --------- | ----------------- | ----------- | ------- |
| `region` | Required | Huawei Cloud region | `HUAWEI_REGION` |
| `cluster_id` | Required | Exact CCE cluster ID | N/A |
| `ak` | Optional | Explicit AK for access paths that support it | profile/environment fallback |
| `sk` | Optional | Explicit SK for access paths that support it | profile/environment fallback |
| `project_id` | Required for `kubectl cce`; optional otherwise | Target cluster's Huawei Cloud project ID | hcloud profile/IAM/environment fallback for external kubeconfig access |
### Current Event Query Parameters
| Tool | Required | Optional |
| ---- | -------- | -------- |
| `huawei_get_cce_events` | `region`, `cluster_id` | `namespace`, `event_type` (`Warning` default, `Normal`, or `all`), `limit`, `ak`, `sk`, `project_id` (required for `kubectl cce`), `security_token` |
### Historical Event Query Parameters
| Tool | Required | Optional |
| ---- | -------- | -------- |
| `huawei_query_k8s_events_from_lts` | `region`, `cluster_id`, `start_time`, `end_time`, `project_id` | `event_type` (`Warning` default, `Normal`, or `all`), `keywords` (requires `event_type=all`), `ak`, `sk` |
### Event Analysis Parameters
| Tool | Required | Optional |
| ---- | -------- | -------- |
| `huawei_analyze_cce_events` | Either `events`, or `region` + `cluster_id` | `event_source` (`current` default or `lts`), `start_time`/`end_time` (required for `lts`), `namespace`, `event_type`, `keywords`, `limit`, `max_groups` (1-100, default 10), `check_resource_status` (default true when `region` and `cluster_id` are present), `ak`, `sk`, `project_id`, `security_token` |
## Output Format
All public response fields, Event record fields, and resource-status states are defined in [output-schema.md](references/output-schema.md). That reference is the single source of truth for output contracts.
## Workflow
1. Identify `region`, exact `cluster_id`, optional namespace, and incident time window.
2. Use `huawei_get_cce_events` for current Event inspection.
3. Use `huawei_query_k8s_events_from_lts` for historical Event windows longer than one hour, or when a precise LTS time range or keyword filtering is required.
4. Pass the returned `events` to `huawei_analyze_cce_events` to aggregate reasons, namespaces, resources, and repeated patterns.
5. Hand off evidence to the relevant Pod, Workload, Node, Storage, or Network diagnosis skill.
See [workflow.md](references/workflow.md) for pattern recognition and time-window analysis guidance.
## Verification
Run a current Event query first:
```bash
python3 scripts/huawei-cloud.py huawei_get_cce_events \
region=cn-north-4 cluster_id=<cluster-id> limit=10
```
When default Event-to-LTS collection is enabled, verify a bounded historical query:
```bash
python3 scripts/huawei-cloud.py huawei_query_k8s_events_from_lts \
region=cn-north-4 cluster_id=<cluster-id> \
start_time="2026-05-30 06:00:00" \
end_time="2026-05-30 07:00:00"
```
Verify that the current Event response includes `access_method`, and that the LTS response identifies the default LTS group and stream. Do not create or change logging configuration as part of verification.
## Best Practices
1. **Start with warnings** - filter `type == "Warning"` before detailed inspection.
2. **Group by reason** - repeated reasons reveal systemic issues faster than individual records.
3. **Use exact cluster IDs** - do not infer a cluster from its name.
4. **Keep LTS windows bounded** - use the smallest incident window that answers the question.
5. **Use LTS for history** - current Kubernetes Events have limited retention.
6. **Hand off rather than remediate** - this skill provides evidence only.
## Notes
- No active warning does not prove a cluster is healthy; inspect historical LTS Events for recent or recovered incidents when available.
- The Event-to-LTS path depends on a healthy log-agent add-on with default Event collection enabled.
- Event summaries should redact sensitive production workload, Pod, and node identifiers where the audience does not need them.
- Do not modify Kubernetes, CCE logging, LTS, or cloud resources through this skill.
## Troubleshooting
| Symptom | Likely Cause | Action |
| ------- | ------------ | ------ |
| External kubeconfig access fails | No external endpoint, invalid profile, or missing CCE permission | Verify `cce:cluster:get` and `cce:cluster:createCert`; the tool then tries `kubectl cce` |
| `kubectl cce` fallback fails | Plugin missing or plugin credentials unavailable | Install/configure the plugin using [kubectl-cce.md](references/kubectl-cce.md) |
| LTS query finds no default Event stream | Default Event collection is not enabled or has not finished provisioning | Enable default Event collection through the log-agent add-on, then retry |
| LTS query returns no records | Time window, keywords, retention, or event collection does not match | Narrow or correct the window and verify the default LTS group and stream |
| Too many current Events | Broad cluster query | Warning is the default; provide `namespace` and a lower `limit` to further reduce data at the source |
| Permission denied | Missing IAM or Kubernetes RBAC permission | Grant the reported least-privilege permission, then retry |
## Limitations
- The skill provides only the two documented read-only Event tools.
- Current Event queries support only namespace and Event type (`Warning`, `Normal`, or `all`) server-side selection.
- Historical queries require default Event-to-LTS collection enabled before the incident; the skill cannot recover uncollected history.
- The skill cannot create, modify, or delete LTS streams, Kubernetes resources, or CCE resources.
- The skill does not automatically select a cluster, namespace, event filter, or diagnosis/remediation action for the user.
## References
| Document | Use |
| -------- | --- |
| [Workflow](references/workflow.md) | Event query sequence, grouping, patterns, and time-window analysis |
| [Risk Rules](references/risk-rules.md) | Read-only boundaries, redaction, and handoff constraints |
| [Output Schema](references/output-schema.md) | Query, analysis, and Event record fields |
| [kubectl-cce](references/kubectl-cce.md) | kubectl-cce installation, credentials, and access fallback |
| [Acceptance Criteria](references/acceptance-criteria.md) | Expected outcomes for current, historical, and combined query-and-analysis flows |