Control Plane API Architecture¶

Version: 1.3.0 (April 2026) Status: Current Implementation

Revision History¶

1. Overview¶

The Control Plane API is the central gateway for the Lablet Cloud Manager system. It serves as the Backend-for-Frontend (BFF) providing:

• REST API for LabletDefinition, LabletInstance, and Worker CRUD operations
• Bootstrap 5 SPA with Server-Side Events (SSE) for real-time updates
• OAuth2/OIDC Authentication via Keycloak
• CQRS Pattern with Commands (writes) and Queries (reads) via Mediator
• CloudEvents Publishing for external system integration (fire-and-forget)
• etcd State Publishing for controller watches

2. Entity Hierarchy¶

erDiagram
    CMLWorker ||--o{ LabRecord : hosts
    CMLWorker ||--o{ NodeDefinition : contains
    CMLWorker ||--o{ ImageDefinition : contains
    LabletDefinition ||--o{ LabletInstance : instantiates
    LabletInstance }o--|| CMLWorker : "assigned_to"
    LabletInstance ||--o| LabRecord : "maps_to"

    CMLWorker {
        string id PK
        string name
        string region
        string instance_type
        string instance_id
        WorkerStatus status
        string ip_address
        int idle_timeout
        datetime created_at
    }

    LabRecord {
        string id PK
        string lab_id
        string title
        string owner
        LabState state
        int node_count
        datetime created
    }

    LabletDefinition {
        string id PK
        string name
        string description
        string topology_yaml
        string[] required_nodes
        int estimated_duration
        boolean is_public
    }

    LabletInstance {
        string id PK
        string definition_id FK
        string worker_id FK
        string user_id
        InstanceState state
        datetime scheduled_start
        datetime scheduled_end
        dict port_mappings
    }

    NodeDefinition {
        string id PK
        string node_definition_id
        string label
        string description
        boolean is_custom
    }

    ImageDefinition {
        string id PK
        string image_id
        string node_definition_id FK
        string label
    }

3. Storage Strategy¶

4. Layer Architecture¶

control-plane-api/
├── api/                          # HTTP Layer
│   ├── controllers/              # REST endpoints (auto-prefixed)
│   │   ├── workers_controller.py
│   │   ├── lablet_definitions_controller.py
│   │   ├── lablet_instances_controller.py
│   │   └── auth_controller.py
│   ├── dependencies.py           # DI for auth, sessions
│   └── services/
│       └── auth.py               # DualAuthService (Cookie + JWT)
│
├── application/                  # Business Logic Layer
│   ├── commands/                 # Write operations (self-contained)
│   │   ├── create_worker_command.py
│   │   ├── start_worker_command.py
│   │   └── create_lablet_instance_command.py
│   ├── queries/                  # Read operations (self-contained)
│   │   ├── get_workers_query.py
│   │   └── get_lablet_instances_query.py
│   ├── dtos/                     # Data Transfer Objects
│   ├── services/                 # Application services
│   │   ├── worker_monitoring_scheduler.py
│   │   └── sse_event_relay.py
│   ├── jobs/                     # Background tasks
│   │   ├── worker_metrics_collection_job.py
│   │   └── labs_refresh_job.py
│   └── settings.py               # Configuration (Pydantic Settings)
│
├── domain/                       # Core Domain Layer
│   ├── entities/                 # Aggregates
│   │   ├── cml_worker.py
│   │   ├── lab_record.py
│   │   ├── lablet_definition.py
│   │   └── lablet_instance.py
│   ├── events/                   # Domain events (@cloudevent)
│   ├── enums/                    # Value objects
│   │   ├── worker_status.py
│   │   └── instance_state.py
│   └── repositories/             # Abstract interfaces
│
├── integration/                  # External Service Adapters
│   ├── repositories/             # MongoDB implementations
│   │   ├── mongo_worker_repository.py
│   │   └── mongo_lablet_instance_repository.py
│   └── services/
│       ├── aws_ec2_api_client.py
│       ├── cml_api_client.py
│       └── control_plane_api_client.py
│
├── infrastructure/               # Technical Adapters
│   └── session_store.py          # Redis/InMemory session storage
│
├── ui/                           # Frontend Assets
│   ├── src/                      # Parcel source (JS, SCSS)
│   └── package.json
│
└── observability/                # Instrumentation
    └── otel_config.py

5. Authentication Architecture¶

The Control Plane API implements Dual Authentication supporting both browser-based and API-based access:

sequenceDiagram
    participant Browser
    participant API as Control Plane API
    participant Redis
    participant Keycloak

    Note over Browser,Keycloak: Cookie-Based Flow (Browser)

    Browser->>API: GET /api/auth/login
    API->>Keycloak: Redirect to OIDC authorize
    Keycloak-->>Browser: Login form
    Browser->>Keycloak: Credentials
    Keycloak->>API: Callback with code
    API->>Keycloak: Exchange code for tokens
    API->>Redis: Store tokens (session_id → tokens)
    API-->>Browser: Set-Cookie: session_id (httpOnly)

    Note over Browser,API: Subsequent requests
    Browser->>API: GET /api/workers (Cookie: session_id)
    API->>Redis: Lookup tokens by session_id
    API->>API: Validate JWT
    API-->>Browser: 200 OK

    Note over Browser,Keycloak: Bearer Token Flow (API Clients)

    Browser->>Keycloak: POST /token (client_credentials)
    Keycloak-->>Browser: access_token
    Browser->>API: GET /api/workers (Authorization: Bearer token)
    API->>Keycloak: Validate JWT (JWKS)
    API-->>Browser: 200 OK

Security Model¶

6. Background Services¶

The Control Plane API runs several background services:

Worker Monitoring Scheduler¶

Coordinates per-worker metrics collection:

class WorkerMonitoringScheduler:
    """
    Lifecycle:
    1. On startup: Query active workers
    2. For each worker: Schedule WorkerMetricsCollectionJob
    3. On worker state change: Add/remove jobs dynamically
    """

    async def start_async(self):
        workers = await self._get_active_workers()
        for worker in workers:
            self._scheduler.add_job(
                WorkerMetricsCollectionJob(worker.id),
                trigger='interval',
                seconds=self._settings.metrics_poll_interval
            )

SSE Event Relay¶

Broadcasts real-time updates to connected clients:

class SSEEventRelay:
    """
    - Clients subscribe with optional filters (worker_ids, event_types)
    - Events broadcast to all matching subscribers
    - Common events: worker_status, worker_metrics, lab_status
    """

7. API Design¶

Controller Routing¶

Neuroglia auto-generates route prefixes from controller class names:

# ✅ Correct
@route(HttpMethod.GET, "/{worker_id}")  # → GET /workers/{worker_id}

# ❌ Wrong (double prefix)
@route(HttpMethod.GET, "/workers/{worker_id}")  # → GET /workers/workers/{worker_id}

CQRS Pattern¶

All operations use the Mediator pattern:

# In controller
result = await self.mediator.execute_async(
    GetWorkersQuery(status_filter=WorkerStatus.RUNNING)
)
return self.process(result)

# In handler
class GetWorkersQueryHandler(QueryHandler[GetWorkersQuery, OperationResult[list[WorkerDto]]]):
    async def handle_async(self, request, cancellation_token=None):
        workers = await self._repository.get_all_async(cancellation_token)
        return self.ok([WorkerDto.from_entity(w) for w in workers])

8. External Integrations¶

AWS EC2 Client¶

class AwsEc2Client:
    """
    Singleton service with boto3 client pooling.

    Operations:
    - describe_instances() - Query instance state
    - start_instances() - Start stopped instances
    - stop_instances() - Stop running instances
    - terminate_instances() - Terminate instances
    - get_cloudwatch_metrics() - CPU, memory, network, disk
    """

CML API Client¶

class CMLApiClient:
    """
    Per-worker client for CML REST API.

    Endpoints:
    - /api/v0/system_information - No auth, worker info
    - /api/v0/system_stats - Auth required, resource usage
    - /api/v0/labs - Lab lifecycle operations
    - /api/v0/node_definitions - Available node types
    """

CloudEvent Ingestion (LDS Integration — ADR-040)¶

The Control Plane API receives CloudEvents from external systems via its CloudEventMiddleware (mounted on the main app). Per ADR-040, CPA handles simple state-transition CloudEvents directly, while complex orchestration events are routed to lablet-controller per ADR-022.

sequenceDiagram
    participant LDS as Lab Delivery System
    participant CPA as Control Plane API
    participant Handler as LdsSessionRunningHandler
    participant MongoDB
    participant SSE as SSE Event Relay
    participant Browser

    LDS->>CPA: POST / (CloudEvent: io.lablet.lds.session.running.v1)
    CPA->>Handler: CloudEventIngestor dispatches
    Handler->>Handler: Deduplication check
    Handler->>MongoDB: Load LabletSession aggregate
    Handler->>Handler: session.mark_running() (READY → RUNNING)
    Handler->>MongoDB: Save aggregate (emits domain events)
    MongoDB-->>SSE: Domain event → SSE handler
    SSE-->>Browser: lablet.session.status.changed
    CPA-->>LDS: 202 Accepted

Supported LDS CloudEvent Types¶

LDS Integration Settings¶

9. CQRS Commands Reference¶

Key commands implemented in Control Plane API:

CollectAndGrade Command (FR-2.2.7)¶

Triggers the grading workflow for a running lablet instance:

@dataclass
class CollectAndGradeCommand(Command[OperationResult[GradingResultDto]]):
    """Collect lab state and submit for grading."""
    instance_id: str
    grading_criteria: dict | None = None  # Optional custom criteria


class CollectAndGradeCommandHandler(CommandHandler):
    async def handle_async(self, request, cancellation_token=None):
        # 1. Get instance from repository
        instance = await self._repository.get_by_id_async(request.instance_id)

        if not instance:
            return self.not_found("LabletInstance", f"Instance {request.instance_id} not found")

        if instance.state.status != LabletInstanceStatus.RUNNING:
            return self.bad_request("Instance must be RUNNING to collect and grade")

        # 2. Trigger grading (publishes domain event)
        instance.collect_and_grade(request.grading_criteria)

        # 3. Persist
        await self._repository.update_async(instance, cancellation_token)

        return self.accepted(GradingResultDto(instance_id=request.instance_id, status="GRADING"))

10. Configuration¶

Key environment variables:

11. Related Documentation¶

• Authentication Flows
• Worker Monitoring
• Real-Time Updates
• CML Worker Domain Model
• ADR-018: Lab Delivery System Integration