Dokumentacja API: Pipeline Engine

Pipeline Engine (openapi-pipeline-engine.yaml, port 8082) jest właścicielem definicji pipeline'ów, wykonywania, planowania, orkiestracji, wersjonowania opartego na Git oraz zadań strumieniowych Flink. Wszystkie punkty końcowe są serwowane pod /api/v1 i wymagają JWT bearerAuth — punkty końcowe webhooków uwierzytelniają się zamiast tego nagłówkami sygnatury HMAC.

Pipeline'y

Definicje pipeline'ów są zarządzane przez metadata-service za prefiksem bramy /api/v1/pipelines/**. Pipeline niesie definicję YAML oraz automatycznie inkrementowaną wersję. Nowe pipeline'y zaczynają w stanie DRAFT.

Parametry

Model Pipeline

Kluczowe pola Pipeline: id, workspaceId, name, description, yamlDefinition, version (int ≥ 1), status, scheduleCron, tags[], sourceConnectionId, targetConnectionId, maxRetries (domyślnie 3), retryDelaySeconds (domyślnie 60), timeoutSeconds (domyślnie 3600), notifications, isActive, createdAt, updatedAt, createdBy.

Enum status: DRAFT | SCHEDULED | RUNNING | SUCCESS | FAILED | WARNING | CANCELLED.

PipelineCreate wymaga name oraz yamlDefinition.

// POST /api/v1/pipelines  →  201 Created
// Request: PipelineCreate
{
  "name": "billing-daily-load",
  "description": "Daily Teradata → Snowflake billing extract",
  "yamlDefinition": "version: 1\nsource:\n  connection: teradata-prod\ntarget:\n  connection: snowflake-dwh\n",
  "scheduleCron": "0 2 * * *",
  "tags": ["billing", "daily"],
  "sourceConnectionId": "a1b2c3d4-0000-0000-0000-000000000001",
  "targetConnectionId": "a1b2c3d4-0000-0000-0000-000000000002"
}

// Response: Pipeline
{
  "id": "f47ac10b-58cc-4372-a567-0e02b2c3d479",
  "workspaceId": "9c1f0e2a-1111-2222-3333-444455556666",
  "name": "billing-daily-load",
  "description": "Daily Teradata → Snowflake billing extract",
  "yamlDefinition": "version: 1\nsource:\n  connection: teradata-prod\n...",
  "version": 1,
  "status": "DRAFT",
  "scheduleCron": "0 2 * * *",
  "tags": ["billing", "daily"],
  "maxRetries": 3,
  "retryDelaySeconds": 60,
  "timeoutSeconds": 3600,
  "isActive": true,
  "createdAt": "2025-12-15T14:30:00Z",
  "updatedAt": "2025-12-15T14:30:00Z",
  "createdBy": "u-1001"
}

Wykonywanie i uruchomienia

Uruchomienie to skompilowane, zaplanowane wykonanie pipeline'u. runPipeline zwraca natychmiast z 202 Accepted — odpytuj getRunStatus, aby śledzić postęp.

Parametry i ciała

RunResponse: runId, pipelineId, status (RUNNING | PENDING), taskCount, engine (SPARK | FLINK | JDBC). RunStatusResponse.overallStatus: PENDING | RUNNING | COMPLETED | FAILED | CANCELLED; taskStatuses oraz errors to mapy kluczowane identyfikatorem zadania.

// POST /api/v1/pipelines/{id}/run  →  202 Accepted
// Request: RunRequest (all fields optional)
{
  "triggeredBy": "u-1001"
}

// Response: RunResponse
{
  "runId": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
  "pipelineId": "f47ac10b-58cc-4372-a567-0e02b2c3d479",
  "status": "RUNNING",
  "taskCount": 4,
  "engine": "SPARK"
}

// GET /api/v1/runs/{runId}  →  200 OK  (RunStatusResponse)
{
  "runId": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
  "pipelineId": "f47ac10b-58cc-4372-a567-0e02b2c3d479",
  "overallStatus": "RUNNING",
  "taskStatuses": {
    "extract": "COMPLETED",
    "transform": "RUNNING",
    "load": "PENDING"
  },
  "errors": {}
}

// POST /api/v1/runs/{runId}/cancel  →  200 OK
{
  "runId": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
  "status": "CANCELLED",
  "message": "Run cancelled by user request"
}

Scheduler

Scheduler zarządza harmonogramami cron, zadaniami backfill oraz grafami zależności pipeline'ów.

Harmonogramy

Pola CreateScheduleRequest: cronExpression (5-polowe), timezone, businessDaysOnly, skipHolidays, maintenanceWindows[], overlapStrategy (SKIP | QUEUE | CANCEL_RUNNING), catchupStrategy (SKIP_TO_CURRENT | RUN_ALL | RUN_LATEST_N). getNextRuns akceptuje limit (domyślnie 20).

// POST /api/v1/scheduler/schedules  →  201 Created
// Request: CreateScheduleRequest
{
  "pipelineId": "f47ac10b-58cc-4372-a567-0e02b2c3d479",
  "cronExpression": "0 2 * * *",
  "timezone": "Europe/Warsaw",
  "businessDaysOnly": false,
  "skipHolidays": true,
  "overlapStrategy": "SKIP",
  "catchupStrategy": "SKIP_TO_CURRENT"
}

// Response: ScheduleResponse
{
  "id": "7b2e1c44-aaaa-bbbb-cccc-ddddeeeeffff",
  "pipelineId": "f47ac10b-58cc-4372-a567-0e02b2c3d479",
  "cronExpression": "0 2 * * *",
  "timezone": "Europe/Warsaw",
  "status": "ACTIVE",
  "nextRunAt": "2025-12-16T01:00:00Z"
}

Backfille

Pola CreateBackfillRequest: granularity (HOURLY | DAILY | WEEKLY | MONTHLY), concurrency, priority (LOW | MEDIUM | HIGH), dryRun, parameterOverrides.

// POST /api/v1/scheduler/backfill  →  201 Created
// Request: CreateBackfillRequest
{
  "pipelineId": "f47ac10b-58cc-4372-a567-0e02b2c3d479",
  "from": "2025-11-01",
  "to": "2025-11-30",
  "granularity": "DAILY",
  "concurrency": 4,
  "priority": "MEDIUM",
  "dryRun": false
}

// Response: BackfillResponse
{
  "id": "c9d8e7f6-1234-5678-9abc-def012345678",
  "pipelineId": "f47ac10b-58cc-4372-a567-0e02b2c3d479",
  "status": "RUNNING",
  "totalIntervals": 30,
  "completedIntervals": 11,
  "failedIntervals": 0
}

Zależności

AddDependencyRequest.type: STRONG | WEAK. Dodanie międzyworkspace'owej zależności zwraca 403.

// POST /api/v1/scheduler/dependencies  →  201 Created
// Request: AddDependencyRequest
{
  "upstreamPipelineId": "f47ac10b-58cc-4372-a567-0e02b2c3d479",
  "downstreamPipelineId": "a1b2c3d4-9999-8888-7777-666655554444",
  "type": "STRONG"
}

Orkiestrator

Orkiestrator integruje zewnętrzne schedulery (Airflow, AutomateNow) poprzez webhooki, callbacki oraz synchronizację DAG-ów. Webhooki wyzwalające uwierzytelniają się nagłówkami HMAC, a nie JWT.

triggerPipeline wymaga nagłówków X-Webhook-Signature oraz X-Webhook-Id, a także ciała TriggerRequest wymagającego external_job_id. WebhookRegistrationRequest wymaga name, source, hmac_secret, callback_url. triggerPipeline oraz triggerAirflowDag zwracają 202 Accepted.

// POST /api/v1/orchestrator/trigger  →  202 Accepted
// Headers: X-Webhook-Signature: <hmac>, X-Webhook-Id: <uuid>
// Request: TriggerRequest
{
  "external_job_id": "airflow-billing-2025-12-15",
  "pipelineId": "f47ac10b-58cc-4372-a567-0e02b2c3d479",
  "parameters": { "loadDate": "2025-12-15" }
}

// Response: TriggerResponse
{
  "runId": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
  "external_job_id": "airflow-billing-2025-12-15",
  "status": "ACCEPTED"
}

// POST /api/v1/orchestrator/airflow/trigger/{dagId}  →  202 Accepted
{
  "dagId": "dataflow_billing_daily_load",
  "dagRunId": "manual__2025-12-15T14:30:00+00:00",
  "state": "queued"
}

Git

Definicje pipeline'ów są wersjonowane w Git. Te punkty końcowe zarządzają repozytoriami, historią, różnicami, wycofywaniem, gałęziami oraz scalaniami. Webhook Git uwierzytelnia się nagłówkami HMAC dostawcy.

Parametry

ConnectRepoRequest.credentialType: NONE | TOKEN | SSH_KEY. WebhookResult.action: SYNC | DEPLOY | VALIDATE | IGNORE | ERROR. mergeBranch zwraca 409 w przypadku konfliktu scalania.

// POST /api/v1/git/repos  →  201 Created
// Request: ConnectRepoRequest
{
  "workspaceSlug": "billing-team",
  "remoteUrl": "https://gitlab.polkomtel.pl/dataflow/billing.git",
  "credentialType": "TOKEN",
  "credentialRef": "vault://git/billing-token"
}

// GET /api/v1/git/diff/{pipelineId}?from=3&to=5  →  200 OK  (DiffResponse)
{
  "pipelineId": "f47ac10b-58cc-4372-a567-0e02b2c3d479",
  "fromVersion": 3,
  "toVersion": 5,
  "diff": "@@ -2,3 +2,3 @@\n-  connection: teradata-stage\n+  connection: teradata-prod"
}

// POST /api/v1/git/merge  →  200 OK  (VersionOperationResult)
{
  "success": true,
  "mergedVersion": 6,
  "message": "Branch feature/new-mapping merged into main"
}

Flink

Punkty końcowe Flink zarządzają zadaniami strumieniowymi oraz klastrami sesyjnymi na Kubernetes.

Zadania Flink

FlinkJobConfig.deploymentMode: APPLICATION | SESSION. FlinkJobRecord.status: CREATED | RUNNING | FAILING | FAILED | CANCELLING | CANCELED | FINISHED | RESTARTING | SUSPENDED. triggerFlinkSavepoint wymaga path w ciele. getFlinkJobLogs akceptuje lines (domyślnie 500).

// POST /api/v1/flink/jobs  →  201 Created
// Request: FlinkJobConfig
{
  "jobName": "cdr-stream-enrich",
  "deploymentMode": "APPLICATION",
  "jarUri": "gs://dataflow-artifacts/cdr-enrich-1.4.0.jar",
  "parallelism": 8
}

// Response: FlinkJobHandle
{
  "jobId": "00000000-cdr-enrich-0001",
  "status": "CREATED",
  "deploymentMode": "APPLICATION"
}

// POST /api/v1/flink/jobs/{jobId}/savepoint  →  200 OK
// Request: { "path": "gs://dataflow-savepoints/cdr-enrich/" }
{
  "savepointPath": "gs://dataflow-savepoints/cdr-enrich/savepoint-abc123"
}

Klastry Flink

FlinkClusterInfo.status: STARTING | RUNNING | SCALING | DEGRADED | STOPPING | STOPPED | ERROR. scaleFlinkCluster wymaga taskManagers (≥ 1) w ciele; createFlinkCluster zwraca 409, jeśli klaster o tej nazwie już istnieje.

// POST /api/v1/flink/clusters  →  201 Created
// Request: FlinkClusterCreateRequest
{
  "name": "streaming-prod",
  "taskManagers": 3,
  "taskSlotsPerManager": 4
}

// POST /api/v1/flink/clusters/{name}/scale  →  200 OK  (FlinkClusterInfo)
// Request: { "taskManagers": 6 }
{
  "name": "streaming-prod",
  "status": "SCALING",
  "taskManagers": 6,
  "taskSlotsPerManager": 4
}