josh/electric-cherry
1
0
Fork 0
Code Issues 18 Pull Requests Actions Packages Projects Releases Wiki Activity

Report safe diagnostics for unexpected agent failures

Browse Source
This commit is contained in:
josh
2026-10-08 14:00:13 -04:00
parent b3fa486886
commit 511f090552
4 changed files with 372 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
static/agent/ec-agent.py
+124 -3
View File
@@ -10,6 +10,7 @@ import base64
from concurrent.futures import ThreadPoolExecutor, as_completed
import contextlib
from datetime import datetime, timezone
import errno
import math
import hashlib
import http.client
@@ -28,12 +29,13 @@ import subprocess
import sys
import threading
import time
import types
import urllib.error
import urllib.parse
import urllib.request
import uuid
VERSION = "1.8.2"
VERSION = "1.8.3"
LABEL = "io.electric-cherry"
MAX_FILE = 32 * 1024 * 1024
MANIFEST_MAX_ENTRIES = 10000
@@ -79,6 +81,106 @@ class AutoscalingInvalidated(Rejected):
"""This removal needs a new policy decision, never a later local retry."""
OPERATION_ERROR_PHASES = {
"executing": "executing the operation",
"checking-instance": "checking the existing instance",
"validating-create-spec": "validating the container specification",
"preparing-create-directory": "preparing the instance directory",
"seeding-runtime": "seeding the runtime cache",
"writing-create-files": "writing initial files",
"persisting-create-metadata": "saving instance metadata",
"creating-container": "creating the container",
"starting-container": "starting the container",
"recording-created-instance": "recording the started container",
"inspecting-created-instance": "checking the created container",
"removing-old-instance": "removing the old instance directory",
}
def unexpected_operation_error(error, phase):
"""Return bounded diagnostics without exception text, paths, or runtime data."""
def attribute(value, name):
try:
return getattr(value, name, None)
except Exception:
return None
def category(value):
# Names come from this fixed list, never an arbitrary exception class.
for kind in (urllib.error.HTTPError, urllib.error.URLError, http.client.IncompleteRead,
http.client.RemoteDisconnected, http.client.HTTPException,
subprocess.TimeoutExpired, TimeoutError,
PermissionError, FileNotFoundError, FileExistsError, IsADirectoryError,
NotADirectoryError, ConnectionError, OSError, sqlite3.OperationalError, sqlite3.Error,
json.JSONDecodeError, UnicodeError, KeyError, TypeError, ValueError,
MemoryError, RuntimeError):
if isinstance(value, kind):
return kind.__name__
return "Exception"
phase = phase if isinstance(phase, str) and phase in OPERATION_ERROR_PHASES else "executing"
details = {"type": category(error), "phase": phase, "agentVersion": VERSION}
evidence = error
if isinstance(error, urllib.error.HTTPError):
status = attribute(error, "code")
if type(status) is int and 100 <= status <= 599:
details["httpStatus"] = status
elif isinstance(error, urllib.error.URLError):
reason = attribute(error, "reason")
if isinstance(reason, BaseException):
details["causeType"] = category(reason)
evidence = reason
number = attribute(evidence, "errno") if isinstance(evidence, OSError) else None
number = number if type(number) is int and number in errno.errorcode else None
if number is not None:
details.update(errno=number, errnoName=errno.errorcode[number])
frames = []
trace = attribute(error, "__traceback__")
while isinstance(trace, types.TracebackType):
# Only agent-owned code locations are safe to report. Exclude source
# lines, filenames, other modules, frame locals, and exception chains.
if trace.tb_frame.f_globals is globals():
frames.append({"function": trace.tb_frame.f_code.co_name, "line": trace.tb_lineno})
trace = trace.tb_next
details["frames"] = frames[-8:]
if details.get("httpStatus") in (401, 403):
guidance = "Check host enrollment and controller artifact access permissions."
elif details.get("httpStatus") == 404:
guidance = "Check that the requested artifact is still available on the controller."
elif isinstance(error, urllib.error.HTTPError):
guidance = "Check controller availability and artifact delivery."
elif number == errno.ENOSPC:
guidance = "Check free disk space and inodes on the host."
elif number == errno.EDQUOT:
guidance = "Check the host filesystem quota."
elif number in (errno.EACCES, errno.EPERM) or isinstance(evidence, PermissionError):
guidance = "Check permissions on the host's managed files and directories."
elif number == errno.EROFS:
guidance = "Check whether the host filesystem is mounted read-only."
elif isinstance(error, (urllib.error.URLError, http.client.HTTPException)) or isinstance(evidence, (TimeoutError, ConnectionError)):
guidance = "Check host network, DNS, TLS, and controller connectivity."
elif isinstance(error, subprocess.TimeoutExpired):
guidance = "Check whether the host service command is stalled or unavailable."
elif isinstance(evidence, FileNotFoundError):
guidance = "Check that the required host files and directories are present."
elif isinstance(evidence, OSError):
guidance = "Check host storage, filesystem access, and service availability."
else:
guidance = "Inspect the saved operation diagnostics and host setup before retrying."
labels = [details["type"]]
if "httpStatus" in details:
labels.append(f"HTTP {details['httpStatus']}")
if "errnoName" in details:
labels.append(f"{details['errnoName']} ({details['errno']})")
if "causeType" in details:
labels.append("cause " + details["causeType"])
if frames:
labels.append(f"at {frames[-1]['function']}:{frames[-1]['line']}")
return f"Operation error while {OPERATION_ERROR_PHASES[phase]} ({'; '.join(labels)}). {guidance}", details
def systemctl(*args):
try:
result = subprocess.run(["systemctl", *args], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=60, check=False)
@@ -962,6 +1064,7 @@ class Agent:
self.plutonium_lock = threading.RLock()
self.config_path = config.get("configPath", "/etc/electric-cherry/agent.json")
self.executing_id = None
self.operation_phase = None
self.lane_lock = threading.Lock()
self.lanes = {name: {"thread": None, "lastPollAt": None, "polled": time.monotonic(), "busy": False}
for name in ("health-1", "health-2", "read", "write")}
@@ -1767,6 +1870,7 @@ class Agent:
return True
def create(self, instance_id, generation, payload):
self.operation_phase = "checking-instance"
self.check_creation_cancellation()
directory = self.instance_dir(instance_id, generation)
current_marker = directory.parent / "current.json"
@@ -1779,12 +1883,17 @@ class Agent:
if self.spec(instance_id, generation) != payload:
raise AgentError("Instance already exists with a different specification")
return existing
self.operation_phase = "validating-create-spec"
game, library, image, port, bind, env = self.validate_spec(payload)
app = directory / "app"
self.operation_phase = "preparing-create-directory"
app.mkdir(parents=True, exist_ok=True)
self.make_writable(app, directory=True)
self.operation_phase = "seeding-runtime"
self.seed_plutonium(app, image)
self.operation_phase = "writing-create-files"
self.write_initial_files(app, payload.get("files", []))
self.operation_phase = "persisting-create-metadata"
atomic_json(directory / "spec.json", payload)
env_path = directory / "environment"
atomic_bytes(env_path, "".join(f"{key}={value}\n" for key, value in env.items()).encode())
@@ -1800,12 +1909,16 @@ class Agent:
args.extend(["--publish", f"{bind_arg}:{port}:{port}/{protocol}"])
args.append(image)
self.check_creation_cancellation()
self.operation_phase = "creating-container"
self.checkpoint({"phase": "creating", "generation": generation, "specSha256": spec_sha})
self.docker.call(*args, timeout=120)
self.check_creation_cancellation()
self.operation_phase = "starting-container"
self.docker.call("container", "start", name)
self.check_creation_cancellation()
self.operation_phase = "recording-created-instance"
atomic_json(current_marker, {"generation": generation})
self.operation_phase = "inspecting-created-instance"
return self.inspect(instance_id, generation)
def check_creation_cancellation(self):
@@ -1913,6 +2026,7 @@ class Agent:
self.docker.call("container", "rm", name)
if kind == "instance.rebuild":
result = self.create(instance_id, next_generation, spec)
self.operation_phase = "removing-old-instance"
old = self.instance_dir(instance_id, generation)
if old.exists():
shutil.rmtree(old, ignore_errors=False)
@@ -2507,6 +2621,7 @@ class Agent:
except AgentError:
return self.journal.update(operation["id"], "accepted", {"blocked": "controller-unavailable", "message": "Waiting for controller contact before executing the operation"})
self.executing_id = operation["id"]
self.operation_phase = "executing"
try:
result = self.dispatch(operation)
except Held as err:
@@ -2518,11 +2633,17 @@ class Agent:
except Exception as err:
# The command may already have changed Docker or files. Never infer
# failure means nothing happened, and never retry without inspection.
message = str(err) if isinstance(err, AgentError) else "Operation could not be completed; inspect its current state"
diagnostic = None
if isinstance(err, AgentError):
message = str(err)
else:
message, details = unexpected_operation_error(err, self.operation_phase)
diagnostic = {"unexpectedError": details}
checkpoint = self.journal.get(operation["id"]).get("checkpoint")
return self.journal.update(operation["id"], "failed" if isinstance(err, Rejected) or not checkpoint else "unknown", error=message)
return self.journal.update(operation["id"], "failed" if isinstance(err, Rejected) or not checkpoint else "unknown", diagnostic, error=message)
finally:
self.executing_id = None
self.operation_phase = None
return self.journal.update(operation["id"], "running" if operation["kind"] == "host.decommission" else "completed", result)
def recover_credential(self):