Control Plane Services

The Control Plane is the stable orchestration layer that coordinates capabilities without coupling to any specific provider.

Service Overview

graph TB
    subgraph "Control Plane"
        IN[Intake Service]
        PW[Pathway Service]
        PE[Policy Engine]
        EC[Echo Service]
        PR[Provider Registry]
        MCP[MCP Layer]
        DE[Documentation Engine]
    end

    IN -->|"metrics, lineage"| EC
    PW -->|"metrics, lineage"| EC
    PE -->|"reads state"| EC
    MCP -->|"invokes"| IN & PW & PE & PR
    DE -->|"reads"| EC

Intake Service

Orchestrates data acquisition through the Source Acquisition capability.

public interface IntakeService {
    AcquisitionResult acquire(Pipeline pipeline, Stage stage);
    void checkpoint(String executionId);
    AcquisitionResult resume(String executionId, CheckpointState checkpoint);
}

Key behaviours:

• Resolves the Source Provider from the Provider Registry
• Supports both replayable and non-replayable sources
• Persists acquisition state (offsets, watermarks) for resumption
• Buffers non-replayable sources into replayable intermediate stores

Pathway Service

Orchestrates transformation execution with dual-mode support.

public interface PathwayService {
    TransformResult transform(Pipeline pipeline, Stage stage, DataReference input);
    ExecutionStrategy resolveExecutionStrategy(Stage stage, ComputePlatformProfile profile);
}

Execution strategies:

Strategy	When Used
SPARK_API	Databricks, EMR, Synapse Spark Pool
PURE_SQL	Snowflake, Athena, Synapse SQL Pool
PLATFORM_NATIVE	Platform-specific optimised path

Echo Service

The unified telemetry hub — receives and exposes state, metrics, lineage, events, quality results, and recovery metadata.

public interface EchoService {
    // State management
    StateTransitionResult transitionState(StateKey key, String targetState, String event);
    PipelineState getState(StateKey key);

    // Multi-channel ingestion
    StateTransitionResult handleRestUpdate(StateUpdate update);
    StateTransitionResult handleWebhookCallback(WebhookPayload payload);
    StateTransitionResult handleEventBusMessage(PatternOpsEvent event);

    // Metrics, lineage, events, quality, recovery, webhooks, contracts...
}

Ingestion channels:

1. REST API — Direct POST/PUT calls
2. Inbound Webhooks — External systems push updates
3. Event Bus — Messaging Provider subscription

Policy Engine

The authoritative decision-maker for operational automation.

public interface PolicyEngine {
    List<PolicyDecision> evaluate(PipelineState state, List<MetricsEnvelope> metrics);
    void registerPolicy(PolicyDefinition policy);
    RetryDecision handleFailure(StageFailure failure);
    ScalingAction handleScalingTrigger(ScalingTrigger trigger);
    QualityAction handleQualityBreach(QualityBreach breach);
    CatchUpPlan scheduleCatchUp(String pipeline, String fromDate, String toDate);
}

Policy actions:

Action	Description
RETRY	Retry with configured backoff
SCALE	Scale up or down
ALERT	Send notifications
QUARANTINE	Route failing data to quarantine
HALT	Stop pipeline execution
RESTART_FROM_CHECKPOINT	Resume from last good state

Provider Registry

Central catalogue for provider lifecycle management.

public interface ProviderRegistry {
    RegistrationResult register(ProviderDescriptor descriptor);
    DeactivationResult deactivate(String providerId);
    BaseProviderContract resolve(String capability, TenancyContext context);
    BaseProviderContract resolveSubCapability(String capability, String subCapability, TenancyContext context);
    List<ProviderDescriptor> list(TenancyContext context);
    HealthStatus healthCheck(String providerId);
}

MCP Layer

Exposes all PatternOps capabilities as Model Context Protocol tools for AI integration.

public interface MCPLayer {
    MCPToolCatalog discover();
    MCPResult invoke(String toolName, Map<String, Object> params, Identity identity);
    SafetyClass getSafetyClass(String toolName);
}

Safety classifications:

Class	Description	Example
READ_ONLY	No side effects	Query pipeline state
VALIDATION	Check but don't modify	Validate a pipeline definition
CONTROLLED_WRITE	Write with controlled scope	Submit a pipeline
RESTRICTED_ADMIN	Elevated privileges required	Deactivate a provider

Documentation Engine

Generates documentation automatically for every deployed pipeline.

public interface DocumentationEngine {
    DocumentationBundle generate(Pipeline pipeline, DocumentationContext context);
    DocumentationBundle regenerate(Pipeline pipeline, List<String> changedSources);
    DashboardBundle generateDashboards(ComputePlatformProfile profile);
}

Output formats: MkDocs, OpenAPI, JSON Schema docs, MCP tool catalog

Dashboard targets: Grafana, Kibana, Splunk, Datadog, CloudWatch