626 lines
26 KiB
Python
626 lines
26 KiB
Python
#!/usr/bin/env python3
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
|
|
import requests
|
|
|
|
HAPROXY_API_BASE = "http://:admin@127.0.0.1:5555/v3"
|
|
CERT_DIR = "/home/fourlights/.acme.sh"
|
|
ACME_SCRIPT = "/usr/local/bin/acme.sh"
|
|
|
|
|
|
class PodmanHAProxyACMESync:
|
|
def __init__(self):
|
|
self.ssl_services = set()
|
|
self.session = requests.Session()
|
|
self.session.headers.update({"Content-Type": "application/json"})
|
|
|
|
def get_next_index(self, path):
|
|
response = self.session.get(
|
|
f"{HAPROXY_API_BASE}/services/haproxy/configuration/{path}"
|
|
)
|
|
return len(response.json()) if response.status_code == 200 else None
|
|
|
|
def get_dataplaneapi_version(self):
|
|
response = self.session.get(
|
|
f"{HAPROXY_API_BASE}/services/haproxy/configuration/version"
|
|
)
|
|
return response.json() if response.status_code == 200 else None
|
|
|
|
def get_container_labels(self, container_id):
|
|
try:
|
|
result = subprocess.run(
|
|
["podman", "inspect", container_id], capture_output=True, text=True
|
|
)
|
|
if result.returncode == 0:
|
|
data = json.loads(result.stdout)
|
|
return data[0]["Config"]["Labels"] or {}
|
|
except Exception as e:
|
|
print(f"Error getting labels for {container_id}: {e}")
|
|
return {}
|
|
|
|
# ===== PHASE 1: Helper Methods (Check/Query Operations) =====
|
|
|
|
def check_acl_exists(self, frontend, acl_name, criterion, value):
|
|
"""
|
|
Check if an ACL exists in the specified frontend by matching name, criterion, and value.
|
|
Returns: (exists: bool, index: int | None, current_config: dict | None)
|
|
"""
|
|
try:
|
|
response = self.session.get(
|
|
f"{HAPROXY_API_BASE}/services/haproxy/configuration/frontends/{frontend}/acls"
|
|
)
|
|
if response.status_code == 200:
|
|
acls = response.json()
|
|
for idx, acl in enumerate(acls):
|
|
if (acl.get("acl_name") == acl_name and
|
|
acl.get("criterion") == criterion and
|
|
acl.get("value") == value):
|
|
return (True, idx, acl)
|
|
return (False, None, None)
|
|
except Exception as e:
|
|
print(f"[UPSERT-ERROR] Error checking ACL existence in frontend {frontend}: {e}")
|
|
return (False, None, None)
|
|
|
|
def check_backend_rule_exists(self, frontend, rule_name, cond_test):
|
|
"""
|
|
Check if a backend switching rule exists in the specified frontend by matching name and condition test.
|
|
Returns: (exists: bool, index: int | None, current_config: dict | None)
|
|
"""
|
|
try:
|
|
response = self.session.get(
|
|
f"{HAPROXY_API_BASE}/services/haproxy/configuration/frontends/{frontend}/backend_switching_rules"
|
|
)
|
|
if response.status_code == 200:
|
|
rules = response.json()
|
|
for idx, rule in enumerate(rules):
|
|
if (rule.get("name") == rule_name and
|
|
rule.get("cond_test") == cond_test):
|
|
return (True, idx, rule)
|
|
return (False, None, None)
|
|
except Exception as e:
|
|
print(f"[UPSERT-ERROR] Error checking backend rule existence in frontend {frontend}: {e}")
|
|
return (False, None, None)
|
|
|
|
def check_http_request_rule_exists(self, frontend, cond_test, redirect_type, redirect_value):
|
|
"""
|
|
Check if an http_request_rule (redirect rule) exists by matching condition test and redirect parameters.
|
|
Returns: (exists: bool, index: int | None, current_config: dict | None)
|
|
"""
|
|
try:
|
|
response = self.session.get(
|
|
f"{HAPROXY_API_BASE}/services/haproxy/configuration/frontends/{frontend}/http_request_rules"
|
|
)
|
|
if response.status_code == 200:
|
|
rules = response.json()
|
|
for idx, rule in enumerate(rules):
|
|
if (rule.get("cond_test") == cond_test and
|
|
rule.get("type") == "redirect"):
|
|
redirect_cfg = rule.get("redirect_rule", {})
|
|
if (redirect_cfg.get("type") == redirect_type and
|
|
redirect_cfg.get("value") == redirect_value):
|
|
return (True, idx, rule)
|
|
return (False, None, None)
|
|
except Exception as e:
|
|
print(f"[UPSERT-ERROR] Error checking http_request_rule existence in frontend {frontend}: {e}")
|
|
return (False, None, None)
|
|
|
|
# ===== PHASE 2: Upsert Operations =====
|
|
|
|
def upsert_acl(self, frontend, acl_name, criterion, value):
|
|
"""
|
|
Create or update an ACL in the specified frontend.
|
|
Returns: (success: bool, operation_type: str)
|
|
- operation_type: "CREATE", "UPDATE", "SKIP", or "ERROR"
|
|
"""
|
|
try:
|
|
exists, idx, current_config = self.check_acl_exists(frontend, acl_name, criterion, value)
|
|
|
|
if exists and current_config:
|
|
# ACL exists with same config - skip
|
|
print(f"[UPSERT-SKIP] ACL {acl_name} already exists in frontend {frontend}, skipping")
|
|
return (True, "SKIP")
|
|
elif exists and idx is not None:
|
|
# ACL exists but with different config - delete and recreate
|
|
try:
|
|
self.session.delete(
|
|
f"{HAPROXY_API_BASE}/services/haproxy/configuration/frontends/{frontend}/acls/{idx}?version={self.get_dataplaneapi_version()}"
|
|
)
|
|
except Exception as e:
|
|
print(f"[UPSERT-ERROR] Failed to delete old ACL {acl_name}: {e}")
|
|
return (False, "ERROR")
|
|
|
|
# Create new ACL
|
|
acl_data = {
|
|
"acl_name": acl_name,
|
|
"criterion": criterion,
|
|
"value": value,
|
|
}
|
|
response = self.session.post(
|
|
f"{HAPROXY_API_BASE}/services/haproxy/configuration/frontends/{frontend}/acls/{self.get_next_index(f'frontends/{frontend}/acls')}?version={self.get_dataplaneapi_version()}",
|
|
json=acl_data,
|
|
)
|
|
|
|
if response.status_code in [200, 201]:
|
|
op_type = "UPDATE" if (exists and idx is not None) else "CREATE"
|
|
print(f"[UPSERT-{op_type}] ACL {acl_name} {op_type.lower()}d in frontend {frontend}")
|
|
return (True, op_type)
|
|
else:
|
|
print(f"[UPSERT-ERROR] Failed to create ACL {acl_name}: {response.status_code} - {response.text}")
|
|
return (False, "ERROR")
|
|
except Exception as e:
|
|
print(f"[UPSERT-ERROR] Exception upserting ACL {acl_name}: {e}")
|
|
return (False, "ERROR")
|
|
|
|
def upsert_backend_rule(self, frontend, rule_name, cond_test):
|
|
"""
|
|
Create or update a backend switching rule in the specified frontend.
|
|
Returns: (success: bool, operation_type: str)
|
|
- operation_type: "CREATE", "UPDATE", "SKIP", or "ERROR"
|
|
"""
|
|
try:
|
|
exists, idx, current_config = self.check_backend_rule_exists(frontend, rule_name, cond_test)
|
|
|
|
if exists and current_config:
|
|
# Rule exists with same config - skip
|
|
print(f"[UPSERT-SKIP] Backend rule {rule_name} already exists in frontend {frontend}, skipping")
|
|
return (True, "SKIP")
|
|
elif exists and idx is not None:
|
|
# Rule exists but with different config - delete and recreate
|
|
try:
|
|
self.session.delete(
|
|
f"{HAPROXY_API_BASE}/services/haproxy/configuration/frontends/{frontend}/backend_switching_rules/{idx}?version={self.get_dataplaneapi_version()}"
|
|
)
|
|
except Exception as e:
|
|
print(f"[UPSERT-ERROR] Failed to delete old backend rule {rule_name}: {e}")
|
|
return (False, "ERROR")
|
|
|
|
# Create new backend rule
|
|
rule_data = {
|
|
"name": rule_name,
|
|
"cond": "if",
|
|
"cond_test": cond_test,
|
|
}
|
|
response = self.session.post(
|
|
f"{HAPROXY_API_BASE}/services/haproxy/configuration/frontends/{frontend}/backend_switching_rules/{self.get_next_index(f'frontends/{frontend}/backend_switching_rules')}?version={self.get_dataplaneapi_version()}",
|
|
json=rule_data,
|
|
)
|
|
|
|
if response.status_code in [200, 201]:
|
|
op_type = "UPDATE" if (exists and idx is not None) else "CREATE"
|
|
print(f"[UPSERT-{op_type}] Backend rule {rule_name} {op_type.lower()}d in frontend {frontend}")
|
|
return (True, op_type)
|
|
else:
|
|
print(f"[UPSERT-ERROR] Failed to create backend rule {rule_name}: {response.status_code} - {response.text}")
|
|
return (False, "ERROR")
|
|
except Exception as e:
|
|
print(f"[UPSERT-ERROR] Exception upserting backend rule {rule_name}: {e}")
|
|
return (False, "ERROR")
|
|
|
|
def upsert_http_request_rule(self, frontend, cond_test, redirect_rule_data):
|
|
"""
|
|
Create or update an http_request_rule (redirect rule) in the specified frontend.
|
|
redirect_rule_data should contain: {"type": "redirect", "redirect_rule": {...}, "cond": "if", "cond_test": ...}
|
|
Returns: (success: bool, operation_type: str)
|
|
- operation_type: "CREATE", "UPDATE", "SKIP", or "ERROR"
|
|
"""
|
|
try:
|
|
redirect_type = redirect_rule_data.get("redirect_rule", {}).get("type")
|
|
redirect_value = redirect_rule_data.get("redirect_rule", {}).get("value")
|
|
|
|
exists, idx, current_config = self.check_http_request_rule_exists(
|
|
frontend, cond_test, redirect_type, redirect_value
|
|
)
|
|
|
|
if exists and current_config:
|
|
# Rule exists with same config - skip
|
|
print(f"[UPSERT-SKIP] Redirect rule for {cond_test} already exists in frontend {frontend}, skipping")
|
|
return (True, "SKIP")
|
|
elif exists and idx is not None:
|
|
# Rule exists but with different config - delete and recreate
|
|
try:
|
|
self.session.delete(
|
|
f"{HAPROXY_API_BASE}/services/haproxy/configuration/frontends/{frontend}/http_request_rules/{idx}?version={self.get_dataplaneapi_version()}"
|
|
)
|
|
except Exception as e:
|
|
print(f"[UPSERT-ERROR] Failed to delete old redirect rule for {cond_test}: {e}")
|
|
return (False, "ERROR")
|
|
|
|
# Create new http_request_rule
|
|
response = self.session.post(
|
|
f"{HAPROXY_API_BASE}/services/haproxy/configuration/frontends/{frontend}/http_request_rules/{self.get_next_index(f'frontends/{frontend}/http_request_rules')}?version={self.get_dataplaneapi_version()}",
|
|
json=redirect_rule_data,
|
|
)
|
|
|
|
if response.status_code in [200, 201]:
|
|
op_type = "UPDATE" if (exists and idx is not None) else "CREATE"
|
|
print(f"[UPSERT-{op_type}] Redirect rule for {cond_test} {op_type.lower()}d in frontend {frontend}")
|
|
return (True, op_type)
|
|
else:
|
|
print(f"[UPSERT-ERROR] Failed to create redirect rule for {cond_test}: {response.status_code} - {response.text}")
|
|
return (False, "ERROR")
|
|
except Exception as e:
|
|
print(f"[UPSERT-ERROR] Exception upserting redirect rule for {cond_test}: {e}")
|
|
return (False, "ERROR")
|
|
|
|
def request_certificate(self, domain):
|
|
print(f"[CERT-REQUEST] About to request certificate for {domain}")
|
|
sys.stdout.flush()
|
|
|
|
try:
|
|
cmd = [
|
|
ACME_SCRIPT,
|
|
"--issue",
|
|
"-d",
|
|
domain,
|
|
"--standalone",
|
|
"--httpport",
|
|
"8888",
|
|
"--server",
|
|
"letsencrypt",
|
|
"--listen-v4",
|
|
"--debug",
|
|
"2",
|
|
]
|
|
|
|
# Log the command being executed
|
|
print(f"[CERT-REQUEST] Executing: {' '.join(cmd)}")
|
|
sys.stdout.flush()
|
|
|
|
result = subprocess.run(cmd, capture_output=True, text=True)
|
|
|
|
# Log both stdout and stderr for complete debugging
|
|
if result.stdout:
|
|
print(f"[CERT-STDOUT] {result.stdout}")
|
|
sys.stdout.flush()
|
|
if result.stderr:
|
|
print(f"[CERT-STDERR] {result.stderr}")
|
|
sys.stderr.flush()
|
|
|
|
if result.returncode == 0:
|
|
print(f"[CERT-SUCCESS] Certificate obtained for {domain}")
|
|
sys.stdout.flush()
|
|
self.install_certificate(domain)
|
|
return True
|
|
else:
|
|
print(f"[CERT-FAILED] Failed to obtain certificate for {domain}")
|
|
print(f"[CERT-FAILED] Return code: {result.returncode}")
|
|
sys.stdout.flush()
|
|
return False
|
|
|
|
except Exception as e:
|
|
print(f"[CERT-ERROR] Error requesting certificate: {e}")
|
|
sys.stdout.flush()
|
|
return False
|
|
|
|
def install_certificate(self, domain):
|
|
cert_file = f"{CERT_DIR}/{domain}.pem"
|
|
|
|
try:
|
|
acme_cert_dir = f"/home/fourlights/.acme.sh/{domain}_ecc"
|
|
|
|
with open(cert_file, "w") as outfile:
|
|
with open(f"{acme_cert_dir}/fullchain.cer") as cert:
|
|
outfile.write(cert.read())
|
|
with open(f"{acme_cert_dir}/{domain}.key") as key:
|
|
outfile.write(key.read())
|
|
try:
|
|
with open(f"{acme_cert_dir}/ca.cer") as ca:
|
|
outfile.write(ca.read())
|
|
except FileNotFoundError:
|
|
pass
|
|
|
|
os.chmod(cert_file, 0o600)
|
|
print(f"Certificate installed at {cert_file}")
|
|
|
|
self.update_haproxy_ssl_bind(domain)
|
|
|
|
except Exception as e:
|
|
print(f"Error installing certificate for {domain}: {e}")
|
|
|
|
def update_haproxy_ssl_bind(self, domain):
|
|
print(f"Updating ssl bind for {domain}")
|
|
try:
|
|
ssl_bind_data = {
|
|
"address": "*",
|
|
"port": 443,
|
|
"ssl": True,
|
|
"ssl_certificate": f"{CERT_DIR}/{domain}.pem",
|
|
}
|
|
|
|
response = self.session.post(
|
|
f"{HAPROXY_API_BASE}/services/haproxy/configuration/frontends/https_main/binds?version={self.get_dataplaneapi_version()}",
|
|
json=ssl_bind_data,
|
|
)
|
|
print(response.json())
|
|
|
|
if response.status_code in [200, 201]:
|
|
print(f"Updated HAProxy SSL bind for {domain}")
|
|
|
|
except Exception as e:
|
|
print(f"Error updating HAProxy SSL bind: {e}")
|
|
|
|
def setup_certificate_renewal(self, domain):
|
|
renewal_script = f"/etc/cron.d/acme-{domain.replace('.', '-')}"
|
|
|
|
cron_content = f"""0 0 * * * root {ACME_SCRIPT} --renew -d {domain} --post-hook "systemctl reload haproxy" >/dev/null 2>&1
|
|
"""
|
|
|
|
with open(renewal_script, "w") as f:
|
|
f.write(cron_content)
|
|
|
|
print(f"Setup automatic renewal for {domain}")
|
|
|
|
def update_haproxy_backend(self, service_name, host, port, action="add"):
|
|
backend_name = f"backend_{service_name}"
|
|
server_name = f"{service_name}_server"
|
|
|
|
if action == "add":
|
|
backend_data = {
|
|
"name": backend_name,
|
|
"mode": "http",
|
|
"balance": {"algorithm": "roundrobin"},
|
|
}
|
|
backends = self.session.post(
|
|
f"{HAPROXY_API_BASE}/services/haproxy/configuration/backends?version={self.get_dataplaneapi_version()}",
|
|
json=backend_data,
|
|
)
|
|
print(backends.json())
|
|
|
|
server_data = {
|
|
"name": server_name,
|
|
"address": host,
|
|
"port": int(port),
|
|
"check": "enabled",
|
|
}
|
|
tweak = self.session.post(
|
|
f"{HAPROXY_API_BASE}/services/haproxy/configuration/backends/{backend_name}/servers?version={self.get_dataplaneapi_version()}",
|
|
json=server_data,
|
|
)
|
|
print(tweak.json())
|
|
|
|
elif action == "remove":
|
|
self.session.delete(
|
|
f"{HAPROXY_API_BASE}/services/haproxy/configuration/backends/{backend_name}/servers/{server_name}?version={self.get_dataplaneapi_version()}"
|
|
)
|
|
|
|
def update_haproxy_frontend_rule(
|
|
self, service_name, domain, ssl_enabled=False, action="add"
|
|
):
|
|
"""
|
|
Update HAProxy frontend rules (ACLs, backend switching rules, and redirect rules).
|
|
Uses upsert behavior to prevent duplicate rules on multiple syncs.
|
|
"""
|
|
if action == "add":
|
|
if ssl_enabled and domain and domain not in self.ssl_services:
|
|
print(f"Setting up SSL for {domain}")
|
|
if self.request_certificate(domain):
|
|
self.setup_certificate_renewal(domain)
|
|
self.ssl_services.add(domain)
|
|
|
|
# Upsert ACL in HTTP frontend (main)
|
|
self.upsert_acl(
|
|
"main",
|
|
f"is_{service_name}",
|
|
"hdr(host)",
|
|
domain
|
|
)
|
|
|
|
# Upsert ACL in HTTPS frontend if SSL enabled
|
|
if ssl_enabled:
|
|
self.upsert_acl(
|
|
"https_main",
|
|
f"is_{service_name}",
|
|
"hdr(host)",
|
|
domain
|
|
)
|
|
|
|
# Upsert backend switching rule in HTTP frontend
|
|
self.upsert_backend_rule(
|
|
"main",
|
|
f"backend_{service_name}",
|
|
f"is_{service_name}"
|
|
)
|
|
|
|
# Upsert backend switching rule in HTTPS frontend if SSL enabled
|
|
if ssl_enabled:
|
|
self.upsert_backend_rule(
|
|
"https_main",
|
|
f"backend_{service_name}",
|
|
f"is_{service_name}"
|
|
)
|
|
|
|
# Upsert redirect rule (HTTP to HTTPS)
|
|
redirect_rule = {
|
|
"type": "redirect",
|
|
"redirect_rule": {"type": "scheme", "value": "https", "code": 301},
|
|
"cond": "if",
|
|
"cond_test": f"is_{service_name}",
|
|
}
|
|
self.upsert_http_request_rule(
|
|
"main",
|
|
f"is_{service_name}",
|
|
redirect_rule
|
|
)
|
|
|
|
def process_container_event(self, event):
|
|
# DIAGNOSTIC: Log raw event structure
|
|
print(
|
|
f"[EVENT-DEBUG] Received event - Type: {event.get('Type', 'MISSING')}, Action: {event.get('Action', 'MISSING')}"
|
|
)
|
|
sys.stdout.flush()
|
|
|
|
# DIAGNOSTIC: Check for Actor key
|
|
if "Actor" not in event:
|
|
print(
|
|
f"[EVENT-SKIP] Skipping event without 'Actor' key - Full event: {json.dumps(event)}"
|
|
)
|
|
sys.stdout.flush()
|
|
return
|
|
|
|
# DIAGNOSTIC: Check for ID in Actor
|
|
if "ID" not in event["Actor"]:
|
|
print(
|
|
f"[EVENT-SKIP] Skipping event without 'Actor.ID' - Actor content: {json.dumps(event['Actor'])}"
|
|
)
|
|
sys.stdout.flush()
|
|
return
|
|
|
|
container_id = event["Actor"]["ID"][:12]
|
|
action = event["Action"]
|
|
|
|
print(
|
|
f"[EVENT-PROCESS] Processing '{action}' event for container {container_id}"
|
|
)
|
|
sys.stdout.flush()
|
|
|
|
labels = self.get_container_labels(container_id)
|
|
|
|
# Dictionary to store discovered services
|
|
services = {}
|
|
|
|
# First, check for namespaced labels (haproxy.{service_name}.enable)
|
|
for label_key, label_value in labels.items():
|
|
if (
|
|
label_key.startswith("haproxy.")
|
|
and label_key.endswith(".enable")
|
|
and label_value.lower() == "true"
|
|
):
|
|
# Extract service name from label key
|
|
parts = label_key.split(".")
|
|
if len(parts) == 3: # haproxy.{service_name}.enable
|
|
service_name = parts[1]
|
|
|
|
# Extract properties for this service namespace
|
|
service_config = {
|
|
"service_name": service_name,
|
|
"host": labels.get(f"haproxy.{service_name}.host", "127.0.0.1"),
|
|
"port": labels.get(f"haproxy.{service_name}.port", "8080"),
|
|
"domain": labels.get(f"haproxy.{service_name}.domain", None),
|
|
"ssl_enabled": labels.get(
|
|
f"haproxy.{service_name}.tls", "false"
|
|
).lower()
|
|
== "true",
|
|
}
|
|
services[service_name] = service_config
|
|
|
|
# Backward compatibility: If no namespaced labels found, check for flat labels
|
|
if (
|
|
not services
|
|
and "haproxy.enable" in labels
|
|
and labels["haproxy.enable"].lower() == "true"
|
|
):
|
|
service_name = labels.get("haproxy.service", container_id)
|
|
services[service_name] = {
|
|
"service_name": service_name,
|
|
"host": labels.get("haproxy.host", "127.0.0.1"),
|
|
"port": labels.get("haproxy.port", "8080"),
|
|
"domain": labels.get("haproxy.domain", None),
|
|
"ssl_enabled": labels.get("haproxy.tls", "false").lower() == "true",
|
|
}
|
|
|
|
# Process each discovered service
|
|
for service_name, config in services.items():
|
|
if action in ["start", "restart"]:
|
|
print(
|
|
f"Adding service {config['service_name']} to HAProxy (SSL: {config['ssl_enabled']}, Domain: {config['domain']})"
|
|
)
|
|
sys.stdout.flush()
|
|
self.update_haproxy_backend(
|
|
config["service_name"], config["host"], config["port"], "add"
|
|
)
|
|
if config["domain"]:
|
|
self.update_haproxy_frontend_rule(
|
|
config["service_name"],
|
|
config["domain"],
|
|
config["ssl_enabled"],
|
|
"add",
|
|
)
|
|
|
|
elif action in ["stop", "remove", "died"]:
|
|
print(f"Removing service {config['service_name']} from HAProxy")
|
|
sys.stdout.flush()
|
|
self.update_haproxy_backend(
|
|
config["service_name"], config["host"], config["port"], "remove"
|
|
)
|
|
|
|
def watch_events(self):
|
|
print("Starting Podman-HAProxy-ACME sync...")
|
|
|
|
# Track last sync time
|
|
last_full_sync = 0
|
|
SYNC_INTERVAL = 60 # Re-scan all containers every 60 seconds
|
|
|
|
def do_full_sync():
|
|
"""Perform a full sync of all running containers"""
|
|
print("Performing full container sync...")
|
|
try:
|
|
result = subprocess.run(
|
|
["podman", "ps", "--format", "json"], capture_output=True, text=True
|
|
)
|
|
if result.returncode == 0:
|
|
containers = json.loads(result.stdout)
|
|
for container in containers:
|
|
event = {
|
|
"Type": "container",
|
|
"Action": "start",
|
|
"Actor": {"ID": container.get("Id", "")},
|
|
}
|
|
self.process_container_event(event)
|
|
print(f"Synced {len(containers)} containers")
|
|
except Exception as e:
|
|
print(f"Error during full sync: {e}")
|
|
|
|
# Initial sync
|
|
do_full_sync()
|
|
last_full_sync = time.time()
|
|
|
|
print("Watching for container events...")
|
|
|
|
cmd = ["podman", "events", "--format", "json"]
|
|
process = subprocess.Popen(cmd, stdout=subprocess.PIPE, text=True)
|
|
|
|
# Use select/poll for non-blocking read so we can do periodic syncs
|
|
import select
|
|
|
|
while True:
|
|
# Check if it's time for periodic sync
|
|
if time.time() - last_full_sync >= SYNC_INTERVAL:
|
|
do_full_sync()
|
|
last_full_sync = time.time()
|
|
|
|
# Check for events with timeout
|
|
ready, _, _ = select.select([process.stdout], [], [], 5)
|
|
|
|
if ready:
|
|
line = process.stdout.readline()
|
|
if line:
|
|
try:
|
|
event = json.loads(line.strip())
|
|
if event["Type"] == "container":
|
|
self.process_container_event(event)
|
|
except json.JSONDecodeError as e:
|
|
print(
|
|
f"[EVENT-ERROR] JSON decode error: {e} - Line: {line[:100]}"
|
|
)
|
|
sys.stdout.flush()
|
|
except KeyError as e:
|
|
print(
|
|
f"[EVENT-ERROR] Missing key {e} in event: {json.dumps(event)}"
|
|
)
|
|
sys.stdout.flush()
|
|
except Exception as e:
|
|
print(f"[EVENT-ERROR] Error processing event: {e}")
|
|
print(f"[EVENT-ERROR] Event structure: {json.dumps(event)}")
|
|
sys.stdout.flush()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
os.makedirs(CERT_DIR, exist_ok=True)
|
|
sync = PodmanHAProxyACMESync()
|
|
sync.watch_events()
|