137 lines
4.2 KiB
Python
137 lines
4.2 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from collections.abc import Mapping
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from urllib.parse import urlparse
|
|
|
|
|
|
class HostAgentConfigurationError(ValueError):
|
|
"""Raised when Host Agent process configuration is invalid."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class HostAgentConfig:
|
|
control_plane_url: str
|
|
host_id: str = ""
|
|
token: str = field(default="", repr=False)
|
|
enrollment_token: str = field(default="", repr=False)
|
|
identity_path: Path = Path("tasks/host_identity.json")
|
|
enrollment_managed: bool = False
|
|
display_name: str | None = None
|
|
heartbeat_interval_seconds: float = 30.0
|
|
poll_timeout_seconds: float = 20.0
|
|
retry_backoff_seconds: float = 1.0
|
|
max_retry_backoff_seconds: float = 30.0
|
|
max_retry_attempts: int = 5
|
|
|
|
|
|
def load_host_agent_config(
|
|
env: Mapping[str, str] | None = None,
|
|
) -> HostAgentConfig:
|
|
values = os.environ if env is None else env
|
|
control_plane_url = (
|
|
values.get(
|
|
"HOST_AGENT_CONTROL_PLANE_URL",
|
|
"http://127.0.0.1:8001",
|
|
)
|
|
.strip()
|
|
.rstrip("/")
|
|
)
|
|
parsed_url = urlparse(control_plane_url)
|
|
if parsed_url.scheme not in {"http", "https"} or not parsed_url.netloc:
|
|
raise HostAgentConfigurationError(
|
|
"HOST_AGENT_CONTROL_PLANE_URL must be an HTTP(S) URL"
|
|
)
|
|
|
|
host_id = values.get("HOST_AGENT_HOST_ID", "").strip()
|
|
token = values.get("HOST_AGENT_TOKEN", "").strip()
|
|
if bool(host_id) != bool(token):
|
|
raise HostAgentConfigurationError(
|
|
"HOST_AGENT_HOST_ID and HOST_AGENT_TOKEN must be configured together"
|
|
)
|
|
enrollment_token = values.get("HOST_AGENT_ENROLLMENT_TOKEN", "").strip()
|
|
identity_path = Path(
|
|
values.get("HOST_AGENT_IDENTITY_PATH", "tasks/host_identity.json").strip()
|
|
)
|
|
if not host_id and not enrollment_token and not identity_path.is_file():
|
|
raise HostAgentConfigurationError(
|
|
"explicit Host credentials, an enrollment token, or existing identity state "
|
|
"is required"
|
|
)
|
|
|
|
config = HostAgentConfig(
|
|
control_plane_url=control_plane_url,
|
|
host_id=host_id,
|
|
token=token,
|
|
enrollment_token=enrollment_token,
|
|
identity_path=identity_path,
|
|
enrollment_managed=not bool(host_id),
|
|
display_name=values.get("HOST_AGENT_DISPLAY_NAME") or None,
|
|
heartbeat_interval_seconds=_positive_float(
|
|
values,
|
|
"HOST_AGENT_HEARTBEAT_INTERVAL_SECONDS",
|
|
30.0,
|
|
),
|
|
poll_timeout_seconds=_positive_float(
|
|
values,
|
|
"HOST_AGENT_POLL_TIMEOUT_SECONDS",
|
|
20.0,
|
|
),
|
|
retry_backoff_seconds=_positive_float(
|
|
values,
|
|
"HOST_AGENT_RETRY_BACKOFF_SECONDS",
|
|
1.0,
|
|
),
|
|
max_retry_backoff_seconds=_positive_float(
|
|
values,
|
|
"HOST_AGENT_MAX_RETRY_BACKOFF_SECONDS",
|
|
30.0,
|
|
),
|
|
max_retry_attempts=_positive_int(
|
|
values,
|
|
"HOST_AGENT_MAX_RETRY_ATTEMPTS",
|
|
5,
|
|
),
|
|
)
|
|
if config.max_retry_backoff_seconds < config.retry_backoff_seconds:
|
|
raise HostAgentConfigurationError(
|
|
"maximum retry backoff must not be less than initial backoff"
|
|
)
|
|
return config
|
|
|
|
|
|
def _positive_float(
|
|
values: Mapping[str, str],
|
|
name: str,
|
|
default: float,
|
|
) -> float:
|
|
raw_value = values.get(name)
|
|
if raw_value is None:
|
|
return default
|
|
try:
|
|
value = float(raw_value)
|
|
except ValueError as exc:
|
|
raise HostAgentConfigurationError(f"{name} must be a number") from exc
|
|
if value <= 0:
|
|
raise HostAgentConfigurationError(f"{name} must be greater than zero")
|
|
return value
|
|
|
|
|
|
def _positive_int(
|
|
values: Mapping[str, str],
|
|
name: str,
|
|
default: int,
|
|
) -> int:
|
|
raw_value = values.get(name)
|
|
if raw_value is None:
|
|
return default
|
|
try:
|
|
value = int(raw_value)
|
|
except ValueError as exc:
|
|
raise HostAgentConfigurationError(f"{name} must be an integer") from exc
|
|
if value <= 0:
|
|
raise HostAgentConfigurationError(f"{name} must be greater than zero")
|
|
return value
|