scripts/deploy_observation_qa.py
Rama main · réplica de lectura del repositorio original.
"""Observability acceptance on the disposable Linux Deploy QA host only."""
import http.server
import json
import subprocess
import threading
import time
import uuid
class Collector:
"""Local OTLP/HTTP receiver with a controllable failure response."""
def __init__(self):
self.records = []
self.errors = []
self.fail = False
self.lock = threading.Lock()
owner = self
class Handler(http.server.BaseHTTPRequestHandler):
def do_POST(self):
try:
length = int(self.headers.get("Content-Length", "0"))
assert self.path == "/v1/logs" and self.headers.get_content_type() == "application/json"
assert 0 < length <= 1024 * 1024
body = json.loads(self.rfile.read(length))
with owner.lock:
owner.records.append(body)
failure = owner.fail
response = b"{}"
self.send_response(400 if failure else 200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(response)))
self.end_headers()
self.wfile.write(response)
except Exception as error:
with owner.lock:
owner.errors.append(str(error))
self.send_error(400)
def log_message(self, *args):
pass
self.server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
self.thread.start()
@property
def endpoint(self):
return f"http://127.0.0.1:{self.server.server_port}"
def set_failure(self, fail):
with self.lock:
self.fail = fail
def observations(self):
with self.lock:
requests = list(self.records)
output = []
for request in requests:
for resource in request["resourceLogs"]:
attrs = {item["key"]: item["value"]["stringValue"] for item in resource["resource"]["attributes"]}
for scope in resource["scopeLogs"]:
for record in scope["logRecords"]:
fields = {item["key"]: item["value"]["stringValue"] for item in record["attributes"]}
output.append((attrs, fields, record["body"]["stringValue"]))
return output
def close(self):
self.server.shutdown()
self.server.server_close()
self.thread.join(timeout=5)
assert not self.thread.is_alive(), "collector did not stop"
assert not self.errors, self.errors
def run_observation(api, wait_for, traffic, network, spec_a, port, evidence):
services = []
collector = Collector()
errors = []
metrics = {"stages": [], "cleanupErrors": errors}
evidence["observability"] = metrics
started = time.monotonic()
def path(service, suffix=""):
return "/api/micro-services/" + service["id"] + suffix
def current(service):
return api.call("GET", path(service, "/deployment"))
def snapshot(service):
return api.call("GET", path(service, "/observability"))
def log_exists(service, stream, message):
return any(row["stream"] == stream and message in row["message"] for row in snapshot(service)["logs"])
def create(label):
name = "observe-" + uuid.uuid4().hex[:8]
spec = {**spec_a, "name": name, "domain": name + ".example.test", "replicas": 1,
"env": {**spec_a.get("env", {}), "OBS_ID": label}}
service = {"id": api.call("POST", "/api/micro-services", spec)["id"], "spec": spec}
services.append(service)
api.call("POST", path(service, "/deployment"), {"microNetworkId": network})
service["deployment"] = wait_for(lambda: (record if (record := current(service))["phase"] == "running" else None), timeout=180)
return service
def stop_service(service):
phase = wait_for(lambda: (record if (record := current(service))["phase"] in ["running", "failed", "degraded", "stopped"] else None), timeout=180)["phase"]
if phase != "stopped":
api.call("POST", path(service, "/stop"))
wait_for(lambda: current(service)["phase"] == "stopped", timeout=180)
api.call("DELETE", path(service))
def request_log(service):
assert traffic("127.0.0.1", port, service["spec"]["domain"], "/cgi-bin/observation") == (200, "observed")
try:
alpha = create("alpha")
beta = create("beta")
for service, label in [(alpha, "alpha"), (beta, "beta")]:
for stream in ["stdout", "stderr"]:
wait_for(lambda: log_exists(service, stream, f"fixture-{stream}-{label}"), timeout=60,
message=f"{label} {stream} startup output was not collected")
assert "fixture-stdout-beta" not in json.dumps(snapshot(alpha))
assert "fixture-stderr-alpha" not in json.dumps(snapshot(beta))
assert any(event["kind"] == "deployment" for event in snapshot(alpha)["events"])
metrics["stages"].append("isolated stdout, stderr and deployment events")
result = api.call("PUT", path(alpha, "/observability/otlp"), {"endpoint": collector.endpoint})
assert result["endpoint"].rstrip("/") == collector.endpoint
request_log(alpha)
wait_for(lambda: any(attrs.get("firecrab.service.id") == alpha["id"] and
fields.get("log.iostream") == "stderr" and "fixture-cgi-alpha" in body
for attrs, fields, body in collector.observations()), timeout=60,
message="collector did not receive the service log")
assert not any("fixture-stdout-alpha" in body for _, _, body in collector.observations()), "pre-opt-in history exported"
assert not any(attrs.get("firecrab.service.id") == beta["id"] for attrs, _, _ in collector.observations())
metrics["stages"].append("platform logs exported as OTLP/HTTP JSON after opt-in with service and stream labels")
old = current(alpha)
updated_spec = {**alpha["spec"], "env": {**alpha["spec"]["env"], "OBS_REDEPLOY": "1"}}
update = api.call("POST", path(alpha, "/updates"), {"operationId": str(uuid.uuid4()),
"expectedReleaseId": old["id"], "spec": updated_spec, "strategy": "rolling"})
wait_for(lambda: (record if (record := current(alpha))["id"] == update["id"] and record["phase"] == "running" else None), timeout=180,
message="observability environment redeploy failed")
status, app_env = traffic("127.0.0.1", port, alpha["spec"]["domain"], "/cgi-bin/otel-environment")
assert status == 200
endpoint, protocol, service_name = app_env.split("|")
assert endpoint.rstrip("/") == collector.endpoint and protocol == "http/protobuf" and service_name == alpha["spec"]["name"]
metrics["stages"].append("app OTLP environment applied at new VM boot; app metrics/traces are not generated by this fixture")
collector.set_failure(True)
request_log(alpha)
wait_for(lambda: snapshot(alpha)["otlp"]["failedBatches"] > 0, timeout=60,
message="collector failure was not visible")
collector.set_failure(False)
metrics["stages"].append("collector failure visible in export status")
subprocess.run(["systemctl", "stop", "firecrab-api.service"], check=True)
try:
address = beta["deployment"]["instances"][0]["ipv4"]
assert traffic(address, 8080, "fixture", "/cgi-bin/observation") == (200, "observed")
finally:
subprocess.run(["systemctl", "start", "firecrab-api.service"], check=True)
api.ready()
wait_for(lambda: log_exists(beta, "stderr", "fixture-cgi-beta"), timeout=60,
message="output produced while the API was down was not recovered")
assert "fixture-cgi-beta" not in json.dumps(snapshot(alpha))
metrics["stages"].append("API restart buffered guest output without cross-service exposure")
api.call("POST", path(alpha, "/stop"))
wait_for(lambda: current(alpha)["phase"] == "stopped", timeout=180)
wait_for(lambda: any(attrs.get("firecrab.service.id") == alpha["id"] and
fields.get("firecrab.observation.kind") == "event" and "stopped" in body
for attrs, fields, body in collector.observations()), timeout=60,
message="collector did not receive the lifecycle event")
metrics["stages"].append("lifecycle event exported after API restart")
print("PASS isolated service logs, API restart buffer, events and OTLP export", flush=True)
except BaseException as error:
metrics["failure"] = str(error)
raise
finally:
try:
api.ready()
except Exception as error:
errors.append("API readiness: " + str(error))
for service in reversed(services):
try:
stop_service(service)
except Exception as error:
errors.append("service cleanup: " + str(error))
try:
collector.close()
except Exception as error:
errors.append("collector cleanup: " + str(error))
metrics["elapsedSeconds"] = round(time.monotonic() - started, 2)
assert not errors, errors