MLOps Model Promotion and Drift Gate Blueprint for Kestra

id: mlops-model-promotion-drift-watchdog
namespace: company.team
description: |
  Promote only models that deserve it: the latest candidate's evaluation and drift
  score gate the promotion, a canary deploy plus smoke check verify it serves
  traffic, and a failed smoke rolls the canary back automatically with the
  registry updated either way.

inputs:
  - id: min_accuracy
    type: FLOAT
    displayName: Minimum evaluation accuracy
    description: Candidate accuracy below this is rejected before anything is deployed.
    defaults: 0.92

  - id: max_p95_latency_ms
    type: INT
    displayName: Maximum p95 latency (ms)
    description: Candidate p95 inference latency above this is rejected.
    defaults: 500

  - id: max_drift_score
    type: FLOAT
    displayName: Maximum drift score
    description: Latest monitored drift score above this blocks promotion (null
      drift counts as 0).
    defaults: 0.15

  - id: canary_image
    type: STRING
    displayName: Canary container image
    description: Image the canary deployment runs, e.g. your-registry/model-server:latest.
    defaults: your-registry/model-server:latest

  - id: canary_url
    type: STRING
    displayName: Canary health endpoint
    description: HTTP endpoint smoke-tested before promotion - 200 means the canary
      serves traffic.
    defaults: https://canary.example.com/health

  - id: canary_namespace
    type: STRING
    displayName: Kubernetes namespace for the canary
    defaults: ml-canary

  - id: repository
    type: STRING
    displayName: GitHub repository for promotion records
    description: owner/repo where rollbacks and rejections are recorded as issues.
    defaults: your-org/your-repo

tasks:
  - id: load_candidate
    type: io.kestra.plugin.jdbc.postgresql.Query
    description: Newest candidate's evaluation metrics with its latest drift score
      attached in one round trip.
    url: "{{ secret('POSTGRES_JDBC_URL') }}"
    username: "{{ secret('POSTGRES_USERNAME') }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    fetchType: FETCH_ONE
    sql: |
      SELECT e.model_id,
             e.version,
             e.accuracy,
             e.f1,
             e.p95_latency_ms,
             e.trained_at,
             d.drift_score
      FROM model_evals e
      LEFT JOIN LATERAL (
        SELECT md.drift_score
        FROM model_drift md
        WHERE md.model_id = e.model_id
          AND md.version = e.version
        ORDER BY md.checked_at DESC
        LIMIT 1
      ) d ON true
      WHERE e.status = 'candidate'
      ORDER BY e.trained_at DESC
      LIMIT 1

  - id: eval_gate
    type: io.kestra.plugin.core.flow.If
    description: Ship only when accuracy, latency, and drift all clear their
      thresholds; reject loudly otherwise.
    condition: |
      {{ outputs.load_candidate.row.model_id is not null
         and outputs.load_candidate.row.accuracy >= inputs.min_accuracy
         and outputs.load_candidate.row.p95_latency_ms <= inputs.max_p95_latency_ms
         and (outputs.load_candidate.row.drift_score ?? 0) <= inputs.max_drift_score }}
    then:
      - id: deploy_canary
        type: io.kestra.plugin.kubernetes.kubectl.Apply
        description: Roll out the candidate as a one-replica canary labeled with its
          model version.
        connection:
          masterUrl: "{{ secret('K8S_MASTER_URL') }}"
          oauthToken: "{{ secret('K8S_TOKEN') }}"
        namespace: "{{ inputs.canary_namespace }}"
        spec: |
          apiVersion: apps/v1
          kind: Deployment
          metadata:
            name: model-canary
            namespace: {{ inputs.canary_namespace }}
            labels:
              app: model-canary
              model-version: "{{ outputs.load_candidate.row.version }}"
          spec:
            replicas: 1
            selector:
              matchLabels:
                app: model-canary
            template:
              metadata:
                labels:
                  app: model-canary
                  model-version: "{{ outputs.load_candidate.row.version }}"
              spec:
                containers:
                  - name: model-server
                    image: {{ inputs.canary_image }}
                    env:
                      - name: MODEL_ID
                        value: "{{ outputs.load_candidate.row.model_id }}"
                      - name: MODEL_VERSION
                        value: "{{ outputs.load_candidate.row.version }}"
                    ports:
                      - containerPort: 8080

      - id: smoke_canary
        type: io.kestra.plugin.core.http.Request
        description: Probe the canary health endpoint - allowFailed so a non-200 reaches
          the gate instead of aborting the task.
        uri: "{{ inputs.canary_url }}"
        method: GET
        options:
          allowFailed: true

      - id: smoke_gate
        type: io.kestra.plugin.core.flow.If
        description: Registry promotion only after the canary actually serves a 200;
          otherwise tear it down and stop the pipeline.
        condition: "{{ outputs.smoke_canary.code == 200 }}"
        then:
          - id: demote_previous
            type: io.kestra.plugin.jdbc.postgresql.Queries
            description: Archive whichever version currently holds production status.
            url: "{{ secret('POSTGRES_JDBC_URL') }}"
            username: "{{ secret('POSTGRES_USERNAME') }}"
            password: "{{ secret('POSTGRES_PASSWORD') }}"
            fetchType: NONE
            sql: |
              UPDATE model_registry
              SET status = 'archived'
              WHERE model_id = '{{ outputs.load_candidate.row.model_id }}'
                AND status = 'production'
                AND version <> '{{ outputs.load_candidate.row.version }}';

          - id: promote_candidate
            type: io.kestra.plugin.jdbc.postgresql.Queries
            description: Mark the candidate as production in the registry, keeping promotion
              history.
            url: "{{ secret('POSTGRES_JDBC_URL') }}"
            username: "{{ secret('POSTGRES_USERNAME') }}"
            password: "{{ secret('POSTGRES_PASSWORD') }}"
            fetchType: NONE
            sql: |
              INSERT INTO model_registry (model_id, version, status, promoted_at)
              VALUES ('{{ outputs.load_candidate.row.model_id }}', '{{ outputs.load_candidate.row.version }}', 'production', now())
              ON CONFLICT (model_id, version)
              DO UPDATE SET status = 'production', promoted_at = now();

          - id: announce_promotion
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            messageText: |
              :rocket: Model promoted to production: *{{ outputs.load_candidate.row.model_id }}* v{{ outputs.load_candidate.row.version }} - accuracy {{ outputs.load_candidate.row.accuracy }}, p95 {{ outputs.load_candidate.row.p95_latency_ms }}ms, drift {{ outputs.load_candidate.row.drift_score ?? 'n/a' }}. Canary passed smoke at {{ inputs.canary_url }}.

        else:
          - id: rollback_canary
            type: io.kestra.plugin.kubernetes.kubectl.Delete
            description: Remove the canary deployment - a model that cannot answer /health
              never reaches production.
            connection:
              masterUrl: "{{ secret('K8S_MASTER_URL') }}"
              oauthToken: "{{ secret('K8S_TOKEN') }}"
            namespace: "{{ inputs.canary_namespace }}"
            resourceType: deployments
            resourcesNames:
              - model-canary

          - id: open_rollback_issue
            type: io.kestra.plugin.github.issues.Create
            description: Record the failed smoke and rollback so the model owner has an
              assignable ticket.
            jwtToken: "{{ secret('GITHUB_TOKEN') }}"
            repository: "{{ inputs.repository }}"
            title: "Canary rollback: {{ outputs.load_candidate.row.model_id }} v{{
              outputs.load_candidate.row.version }} failed smoke ({{
              outputs.smoke_canary.code }})"
            labels:
              - mlops
              - rollback
            body: |
              ## Canary smoke failure

              | Field | Value |
              |---|---|
              | Model | {{ outputs.load_candidate.row.model_id }} |
              | Version | {{ outputs.load_candidate.row.version }} |
              | Canary URL | {{ inputs.canary_url }} |
              | HTTP code | {{ outputs.smoke_canary.code }} |
              | Accuracy | {{ outputs.load_candidate.row.accuracy }} |
              | Drift | {{ outputs.load_candidate.row.drift_score ?? 'n/a' }} |

              The canary deployment was deleted by execution `{{ execution.id }}` and the registry was left untouched.

          - id: fail_smoke
            type: io.kestra.plugin.core.execution.Fail
            description: Fail the execution so any release pipeline waiting on this flow
              stops too.
            errorMessage: "Canary for {{ outputs.load_candidate.row.model_id }} v{{
              outputs.load_candidate.row.version }} returned {{
              outputs.smoke_canary.code }} at {{ inputs.canary_url }} - rolled
              back, registry unchanged."

    else:
      - id: fail_rejection
        type: io.kestra.plugin.core.execution.Fail
        description: Reject the candidate before any deploy - the errors handler carries
          the numbers to Slack.
        errorMessage: "Candidate {{ outputs.load_candidate.row.model_id ?? 'none' }} v{{
          outputs.load_candidate.row.version ?? '-' }} rejected: accuracy {{
          outputs.load_candidate.row.accuracy ?? 'n/a' }} (min {{
          inputs.min_accuracy }}), p95 {{
          outputs.load_candidate.row.p95_latency_ms ?? 'n/a' }}ms (max {{
          inputs.max_p95_latency_ms }}), drift {{
          outputs.load_candidate.row.drift_score ?? 'n/a' }} (max {{
          inputs.max_drift_score }})."

outputs:
  - id: promoted_model
    type: STRING
    value: "{{ outputs.load_candidate.row.model_id }}:{{
      outputs.load_candidate.row.version }}"

  - id: accuracy
    type: FLOAT
    value: "{{ outputs.load_candidate.row.accuracy }}"

  - id: drift_score
    type: FLOAT
    value: "{{ outputs.load_candidate.row.drift_score ?? 0 }}"

errors:
  - id: alert_on_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    messageText: ":rotating_light: Model promotion {{ flow.id }} FAILED (execution
      {{ execution.id }}): {{ error.message }}"

triggers:
  - id: promotion_window
    type: io.kestra.plugin.core.trigger.Schedule
    description: Nightly promotion window after training and evaluation jobs finish.
      Enable when ready.
    cron: "0 6 * * *"
    disabled: true

  - id: on_demand
    type: io.kestra.plugin.core.trigger.Webhook
    description: Promote right after an evaluation run completes - post from CI/CD
      or trigger manually. Replace the key with your own value.
    key: replace-with-a-random-key

Most model regressions reach production through a gap between "the evaluation looked fine" and "the deployment actually works." A notebook says the metrics passed, someone copies a version number into a manifest, and nothing checks that the new model can even answer a health request before traffic arrives. This blueprint closes that gap as one workflow: the candidate's evaluation and drift score decide whether a canary deploys at all, the canary has to prove it serves traffic before the registry flips to production, and every rejection or rollback leaves a trail in Slack and GitHub.

How it works

load_candidate (io.kestra.plugin.jdbc.postgresql.Query, FETCH_ONE) pulls the newest row from model_evals with its latest drift score from model_drift via a LEFT JOIN LATERAL - one round trip gives the gate everything it needs.
eval_gate (io.kestra.plugin.core.flow.If) compares accuracy, p95 latency, and drift against the threshold inputs. A rejection calls core.execution.Fail with the actual numbers in the message, so the flow-level errors handler delivers the reason to Slack and any upstream release pipeline stops.
deploy_canary (io.kestra.plugin.kubernetes.kubectl.Apply) rolls the candidate out as a one-replica canary labeled with its model version; smoke_canary (io.kestra.plugin.core.http.Request with allowFailed) probes the health endpoint without aborting on a non-200.
smoke_gate (io.kestra.plugin.core.flow.If) promotes only on a 200: demote_previous archives the old production version, promote_candidate upserts the new one (ON CONFLICT keeps the registry idempotent), and announce_promotion posts the numbers to Slack.
A failed smoke deletes the canary (io.kestra.plugin.kubernetes.kubectl.Delete), opens a labeled rollback issue (io.kestra.plugin.github.issues.Create), and fails the execution - the registry is never touched.

What you get

A single gate for accuracy, latency, and drift thresholds - three JSON-free inputs, no SQL edits.
Canary-first promotion: nothing enters the registry without answering a live health probe.
Automatic rollback with an assignable GitHub issue carrying every metric.
promoted_model, accuracy, and drift_score outputs for dashboards and downstream release flows.
A nightly schedule or an event-driven webhook, so promotion fits either ops culture.

Who it's for

ML engineers who need promotion to be a pipeline, not a runbook performed at 11 p.m.
Platform teams operating model registries who want drift to veto stale candidates.
Organizations where release governance requires evidence (metrics, rollback, issue trail) for every production change.

Why orchestrate this with Kestra

Promotion scripts hide state: the notebook ran, the deploy happened, nobody knows which version gate failed. Kestra gives you explicit branches for each gate, execution history that records the metrics as outputs, a failure handler that notifies instead of exits silently, and native Kubernetes and GitHub plugins - so the gate, the deploy, and the evidence live in one reviewable workflow.

Prerequisites

PostgreSQL with three tables:

CREATE TABLE model_evals (
  id BIGSERIAL PRIMARY KEY,
  model_id TEXT NOT NULL,
  version TEXT NOT NULL,
  status TEXT NOT NULL DEFAULT 'candidate',
  accuracy NUMERIC(6,4) NOT NULL,
  f1 NUMERIC(6,4) NOT NULL DEFAULT 0,
  p95_latency_ms INT NOT NULL DEFAULT 0,
  trained_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE model_drift (
  id BIGSERIAL PRIMARY KEY,
  model_id TEXT NOT NULL,
  version TEXT NOT NULL,
  drift_score NUMERIC(6,4) NOT NULL,
  checked_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE model_registry (
  model_id TEXT NOT NULL,
  version TEXT NOT NULL,
  status TEXT NOT NULL,
  promoted_at TIMESTAMPTZ,
  PRIMARY KEY (model_id, version)
);

A Kubernetes cluster with your canary image, and a health endpoint reachable from Kestra.
A Kestra instance.

Secrets

POSTGRES_JDBC_URL: JDBC URL for the registry database.
POSTGRES_USERNAME: database username.
POSTGRES_PASSWORD: database password.
K8S_MASTER_URL: Kubernetes API server URL.
K8S_TOKEN: service-account token with apply/delete on the canary namespace.
SLACK_WEBHOOK_URL: Slack incoming webhook for promotion and failure messages.
GITHUB_TOKEN: token with issue-write access to the repository input.

Quick start

Add the seven secrets above and create the three tables.
Insert a candidate evaluation row and, optionally, a drift row for the same model and version.
Set min_accuracy, max_p95_latency_ms, max_drift_score, canary_image, canary_url, and repository.
Run via on_demand - watch load_candidate, the gate, the canary apply, and the smoke result.
Break the smoke on purpose (point canary_url at a dead port) and confirm the delete, the GitHub issue, and the failed execution.
Enable promotion_window when you trust the thresholds.

How to extend

Wait for readiness: add a LoopUntil around smoke_canary that retries for a minute before gating - cold models often need warm-up time.
Wire drift monitoring: a separate schedule that writes model_drift rows (from Prometheus or evidently) can chain into this flow with a Flow trigger, so fresh drift data immediately re-runs the gate.
Batch promotion: wrap the whole flow in a Subflow loop to promote several candidate models in one release train.
Progressive traffic: after smoke_gate, shift service traffic in steps (10% - 50% - 100%) with a Loop over kubectl apply weight patches before promote_candidate.

MLOps Canary Promotion Gate with Drift Watchdog

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

MLOps Canary Promotion Gate with Drift Watchdog

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More AI Blueprints

Build a point-in-time correct training set and block feature leakage

Pre-Publication Content Moderation Gate with Mistral

Measure a Presidio PII scrub, then block the load when residual PII survives

Run Mistral Agent Workflows from Kestra with Completion Polling

Summarize Incident Reports with Mistral and Post to Slack

ML Training Dataset Security Guard and Poisoning Prevention Gate

New to Kestra?