baldur.services — Control API & Metrics

The control-API request/response surface for runtime operations, and the metrics helpers. The metrics symbols resolve lazily and require the [prometheus] extra at runtime.

Control API

ControlAPIService

ControlAPIService()

Baldur Control API Service.

Provides a unified, auditable, reversible, and governed control surface to manage reliability behaviors across testing, chaos experimentation, and real production operations.

Usage

service = ControlAPIService()

Execute control action

response = service.execute(ControlRequest( service_name="payment", action="allow", reason="PG recovered", environment="ops" ))

Get current status

status = service.get_status(environment="ops")

Get audit logs

logs = service.get_audit_logs(service_name="payment")

Initialize the Control API Service.

execute

execute(request: ControlRequest) -> ControlResponse

Execute a control API action.

Parameters:

Name	Type	Description	Default
request	ControlRequest	Control request	required

Returns:

Type	Description
ControlResponse	ControlResponse with outcome

get_status

get_status(environment: str = 'ops') -> dict

Get the current status of all services.

Parameters:

Name	Type	Description	Default
environment	str	Current environment context	'ops'

Returns:

Type	Description
dict	Status dictionary with all service states

get_service_status

get_service_status(service_name: str) -> dict

Get the status of a specific service.

Parameters:

Name	Type	Description	Default
service_name	str	Service to check	required

Returns:

Type	Description
dict	Service state dictionary

is_failure_injection_active

is_failure_injection_active(service_name: str) -> bool

Check if failure injection is active for a service.

Parameters:

Name	Type	Description	Default
service_name	str	Service to check	required

Returns:

Type	Description
bool	True if failures should be injected

get_failure_injection_config

get_failure_injection_config(
    service_name: str,
) -> dict | None

Get failure injection configuration for a service.

Parameters:

Name	Type	Description	Default
service_name	str	Service to check	required

Returns:

Type	Description
dict | None	Configuration dict or None

get_metrics

get_metrics() -> dict

Collect comprehensive baldur metrics for trend analysis.

Returns operational metrics for dashboards, AI agents, and monitoring. Unlike status (point-in-time snapshot), metrics provide trend data.

Consumers: - Admin UI: Dashboard visualization - AI Agent: Automated decision making - Prometheus/Grafana: Metrics scraping - External Monitoring: Alerting integration

Returns:

Type	Description
dict	Dictionary with comprehensive metrics data

ControlRequest dataclass

ControlRequest(
    service_name: str,
    action: str,
    reason: str,
    environment: str,
    ttl_minutes: int | None = None,
    request_id: str = (lambda: str(uuid.uuid4()))(),
    metadata: dict = dict(),
    actor: str = "system",
    actor_role: str = "automation",
)

Internal representation of a control API request.

ControlResponse dataclass

ControlResponse(
    status: str,
    action_applied: str,
    system_state: str = "",
    effective_until: str | None = None,
    reason_classification: str = "",
    evidence: dict = dict(),
    correlation_id: str = (lambda: str(uuid.uuid4()))(),
    error_code: str = "",
    error_message: str = "",
    risk_level: str = "",
)

Bases: SerializableMixin

Internal representation of a control API response.

Metrics

record_sla_breach

record_sla_breach(domain: str) -> None

Record an SLA breach event.

Parameters:

Name	Type	Description	Default
domain	str	Business domain where breach occurred	required

collect_all_metrics

collect_all_metrics() -> dict

Collect all baldur metrics.

This should be called by a periodic Celery task.

Returns:

Type	Description
dict	Dictionary with all current metric values