generated from nathanwoodburn/python-webserver-template
147 lines
5.0 KiB
Python
147 lines
5.0 KiB
Python
from typing import Dict, List
|
|
|
|
import requests
|
|
|
|
from config import AppConfig
|
|
from .base import BaseCollector, CollectionResult
|
|
|
|
|
|
class CoolifyCollector(BaseCollector):
|
|
source_name = "coolify"
|
|
|
|
def __init__(self, config: AppConfig):
|
|
self.config = config
|
|
|
|
def collect(self) -> CollectionResult:
|
|
if not self.config.coolify_enabled:
|
|
return CollectionResult(source=self.source_name, assets=[], status="disabled")
|
|
if not self.config.coolify_endpoints:
|
|
return CollectionResult(source=self.source_name, assets=[], status="skipped", error="No COOLIFY_ENDPOINTS configured")
|
|
if not self.config.coolify_api_token:
|
|
return CollectionResult(source=self.source_name, assets=[], status="skipped", error="No COOLIFY_API_TOKEN configured")
|
|
|
|
headers = {
|
|
"Accept": "application/json",
|
|
"Authorization": f"Bearer {self.config.coolify_api_token}",
|
|
}
|
|
|
|
assets: List[Dict] = []
|
|
errors: List[str] = []
|
|
|
|
for endpoint in self.config.coolify_endpoints:
|
|
base = endpoint.rstrip("/")
|
|
try:
|
|
resp = requests.get(
|
|
f"{base}/api/v1/applications",
|
|
headers=headers,
|
|
timeout=self.config.request_timeout_seconds,
|
|
)
|
|
resp.raise_for_status()
|
|
for app in self._extract_app_list(resp.json()):
|
|
app_status = self._derive_status(app)
|
|
assets.append(
|
|
{
|
|
"asset_type": "service",
|
|
"external_id": str(app.get("id", app.get("uuid", "unknown-app"))),
|
|
"name": app.get("name", "unknown-service"),
|
|
"hostname": app.get("fqdn") or app.get("name"),
|
|
"status": app_status,
|
|
"ip_addresses": [],
|
|
"node": endpoint,
|
|
"metadata": {
|
|
"coolify_uuid": app.get("uuid"),
|
|
"environment": app.get("environment_name"),
|
|
"repository": app.get("git_repository"),
|
|
"raw_status": app.get("status"),
|
|
"health": app.get("health"),
|
|
"deployment_status": app.get("deployment_status"),
|
|
},
|
|
}
|
|
)
|
|
except Exception as exc:
|
|
errors.append(f"{endpoint}: {exc}")
|
|
|
|
if errors and not assets:
|
|
return CollectionResult(source=self.source_name, assets=[], status="error", error=" | ".join(errors))
|
|
if errors:
|
|
return CollectionResult(source=self.source_name, assets=assets, status="degraded", error=" | ".join(errors))
|
|
return CollectionResult(source=self.source_name, assets=assets, status="ok")
|
|
|
|
@staticmethod
|
|
def _extract_app_list(payload: object) -> List[Dict]:
|
|
if isinstance(payload, list):
|
|
return [item for item in payload if isinstance(item, dict)]
|
|
|
|
if isinstance(payload, dict):
|
|
for key in ("data", "applications", "items", "result"):
|
|
value = payload.get(key)
|
|
if isinstance(value, list):
|
|
return [item for item in value if isinstance(item, dict)]
|
|
|
|
return []
|
|
|
|
@staticmethod
|
|
def _derive_status(app: Dict) -> str:
|
|
candidate_fields = [
|
|
app.get("status"),
|
|
app.get("health"),
|
|
app.get("deployment_status"),
|
|
app.get("current_status"),
|
|
app.get("state"),
|
|
]
|
|
|
|
for value in candidate_fields:
|
|
normalized = CoolifyCollector._normalize_status(value)
|
|
if normalized != "unknown":
|
|
return normalized
|
|
|
|
if app.get("is_running") is True or app.get("running") is True:
|
|
return "running"
|
|
if app.get("is_running") is False or app.get("running") is False:
|
|
return "stopped"
|
|
|
|
return "unknown"
|
|
|
|
@staticmethod
|
|
def _normalize_status(value: object) -> str:
|
|
if value is None:
|
|
return "unknown"
|
|
|
|
text = str(value).strip().lower()
|
|
if not text:
|
|
return "unknown"
|
|
|
|
online = {
|
|
"running",
|
|
"online",
|
|
"healthy",
|
|
"up",
|
|
"active",
|
|
"ready",
|
|
"started",
|
|
"success",
|
|
"completed",
|
|
}
|
|
offline = {
|
|
"stopped",
|
|
"offline",
|
|
"down",
|
|
"unhealthy",
|
|
"error",
|
|
"failed",
|
|
"crashed",
|
|
"dead",
|
|
"exited",
|
|
}
|
|
|
|
if text in online:
|
|
return "running"
|
|
if text in offline:
|
|
return "stopped"
|
|
if "running" in text or "healthy" in text:
|
|
return "running"
|
|
if "stop" in text or "fail" in text or "unhealthy" in text:
|
|
return "stopped"
|
|
|
|
return text
|