Azure ML Training Job Governance Gate with Kestra

id: azure-ml-training-job-governance-gate
namespace: company.team

description: |
  Audit a named Azure Machine Learning job, classify it CRITICAL, WARNING, IN_PROGRESS,
  COMPLETED or PROBE_FAILED from its real JobState, alert Slack with a direct Studio link,
  optionally request cancellation after operator review, and log a final audit summary
  either way.

inputs:
  - id: resource_group_name
    type: STRING
    displayName: Resource group name
    description: Resource group containing the Machine Learning workspace. Maps to
      the GetJob and CancelJob tasks' resourceGroupName field.
    defaults: "ml-rg"

  - id: workspace_name
    type: STRING
    displayName: Workspace name
    description: Name of the Azure Machine Learning workspace. Maps to the GetJob
      and CancelJob tasks' workspaceName field.
    defaults: "ml-workspace"

  - id: job_name
    type: STRING
    displayName: Job name
    description: Name of the Azure Machine Learning job to audit. Maps to the GetJob
      and CancelJob tasks' jobName field.
    defaults: "training-run-001"

  - id: job_type_label
    type: SELECT
    displayName: Job type label
    description: Label only, for the log and alert text, Azure ML distinguishes
      command, pipeline, and sweep jobs but GetJob reports the same JobState for
      all three.
    values:
      - "command"
      - "pipeline"
      - "sweep"
    defaults: "command"

  - id: failure_is_critical
    type: BOOL
    displayName: Treat failure as critical
    description: When true, FAILED, CANCELED, or NOT_RESPONDING classifies CRITICAL.
      When false, classifies WARNING instead, useful for low-priority or
      experimental jobs that should not page anyone.
    defaults: true

  - id: request_cancel
    type: BOOL
    displayName: Request cancellation
    description: When true, together with dry_run false, requests cancellation of
      this job regardless of its current classification. Use only after
      reviewing the Slack alert and the job's Studio URL, this is not triggered
      automatically by a bad classification.
    defaults: false

  - id: dry_run
    type: BOOL
    displayName: Dry run
    description: Must be false, together with request_cancel true, for the
      cancellation to actually run.
    defaults: true

tasks:
  - id: probe_job_status
    type: io.kestra.plugin.azure.ml.GetJob
    description: Read-only. Reads the job's current JobState, Studio URL, and any
      MLflow metrics or named outputs it has logged.
    allowFailure: true
    tenantId: "{{ secret('AZURE_TENANT_ID') }}"
    clientId: "{{ secret('AZURE_CLIENT_ID') }}"
    clientSecret: "{{ secret('AZURE_CLIENT_SECRET') }}"
    subscriptionId: "{{ secret('AZURE_SUBSCRIPTION_ID') }}"
    resourceGroupName: "{{ inputs.resource_group_name }}"
    workspaceName: "{{ inputs.workspace_name }}"
    jobName: "{{ inputs.job_name }}"

  - id: classify_job_status
    type: io.kestra.plugin.core.debug.Return
    description: PROBE_FAILED when the probe produced no output at all, CRITICAL or
      WARNING on FAILED, CANCELED, or NOT_RESPONDING depending on
      failure_is_critical, COMPLETED on a clean finish, IN_PROGRESS for every
      non-terminal state.
    format: "{{ outputs.probe_job_status is not defined ? 'PROBE_FAILED' :
      (((outputs.probe_job_status.status == 'FAILED') or
      (outputs.probe_job_status.status == 'CANCELED') or
      (outputs.probe_job_status.status == 'NOT_RESPONDING')) ?
      (inputs.failure_is_critical ? 'CRITICAL' : 'WARNING') :
      (outputs.probe_job_status.status == 'COMPLETED' ? 'COMPLETED' :
      'IN_PROGRESS')) }}"

  - id: route_by_job_status
    type: io.kestra.plugin.core.flow.Switch
    description: Branch A logs a completed or still-running job. Branch B alerts
      Slack with a direct Studio link when the job failed, could not be reached
      for status, or failure_is_critical is false.
    value: "{{ outputs.classify_job_status.value }}"
    cases:
      "COMPLETED":
        - id: log_completed
          type: io.kestra.plugin.core.log.Log
          message: "Azure ML {{ inputs.job_type_label }} job {{ inputs.job_name }}
            completed. {{ outputs.probe_job_status.studioUrl }}"
      "IN_PROGRESS":
        - id: log_in_progress
          type: io.kestra.plugin.core.log.Log
          message: "Azure ML {{ inputs.job_type_label }} job {{ inputs.job_name }} in
            progress, status={{ outputs.probe_job_status.status }}."
      "WARNING":
        - id: alert_warning
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {{ {"text": "Azure ML " ~ inputs.job_type_label ~ " job " ~ inputs.job_name ~ " ended as " ~ outputs.probe_job_status.status ~ " (treated as WARNING). " ~ outputs.probe_job_status.studioUrl} | toJson }}
      "CRITICAL":
        - id: alert_critical
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {{ {"text": "Azure ML " ~ inputs.job_type_label ~ " job " ~ inputs.job_name ~ " CRITICAL, ended as " ~ outputs.probe_job_status.status ~ ". " ~ outputs.probe_job_status.studioUrl} | toJson }}
      "PROBE_FAILED":
        - id: alert_probe_failed
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {{ {"text": "Azure ML job status probe for " ~ inputs.job_name ~ " in " ~ inputs.workspace_name ~ " did not return a result. Check resource_group_name, workspace_name, job_name and Azure credentials."} | toJson }}
    defaults:
      - id: log_unhandled_job_status
        type: io.kestra.plugin.core.log.Log
        message: "Unhandled Azure ML job status classification: {{
          outputs.classify_job_status.value }}"

  - id: remediate_gate
    type: io.kestra.plugin.core.flow.If
    description: Runs only when request_cancel is true and dry_run is false,
      independent of the classification above, an operator may want to cancel a
      job for reasons this gate cannot see, such as a known bad input dataset.
    condition: "{{ inputs.request_cancel and not inputs.dry_run }}"
    then:
      - id: cancel_job
        type: io.kestra.plugin.azure.ml.CancelJob
        description: Cancellation is asynchronous in Azure ML, this task polls until the
          job reaches a terminal state before returning, per the plugin's own
          documented behavior. Cancelling an already-terminal job is a no-op,
          not a failure.
        allowFailure: true
        tenantId: "{{ secret('AZURE_TENANT_ID') }}"
        clientId: "{{ secret('AZURE_CLIENT_ID') }}"
        clientSecret: "{{ secret('AZURE_CLIENT_SECRET') }}"
        subscriptionId: "{{ secret('AZURE_SUBSCRIPTION_ID') }}"
        resourceGroupName: "{{ inputs.resource_group_name }}"
        workspaceName: "{{ inputs.workspace_name }}"
        jobName: "{{ inputs.job_name }}"

      - id: report_cancel_result
        type: io.kestra.plugin.core.flow.If
        description: Fails the flow when the cancel call did not return at all, so a
          failed cancel request is never silently reported as handled.
        condition: "{{ outputs.cancel_job is not defined }}"
        then:
          - id: cancel_request_failed
            type: io.kestra.plugin.core.execution.Fail
            errorMessage: "Cancel request for job {{ inputs.job_name }} did not return a
              result."
        else:
          - id: cancel_result_logged
            type: io.kestra.plugin.core.log.Log
            message: "Cancel of job {{ inputs.job_name }} returned status {{
              outputs.cancel_job.status }}."

  - id: log_audit_summary
    type: io.kestra.plugin.core.log.Log
    description: Final audit log confirming the gate ran, regardless of the branch
      taken above.
    message: "Azure ML training job governance gate completed. job={{
      inputs.job_name }} status={{ outputs.classify_job_status.value }}
      request_cancel={{ inputs.request_cancel }} dry_run={{ inputs.dry_run }}"

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK') }}"
    payload: |
      {{ {"text": "Azure ML training job governance gate flow failed (execution " ~ execution.id ~ "). Check the execution logs."} | toJson }}

triggers:
  - id: every_fifteen_minutes
    type: io.kestra.plugin.core.trigger.Schedule
    description: Shipped disabled, and always in safe mode by default.
    cron: "*/15 * * * *"
    disabled: true
    inputs:
      resource_group_name: "ml-rg"
      workspace_name: "ml-workspace"
      job_name: "training-run-001"
      job_type_label: "command"
      failure_is_critical: true
      request_cancel: false
      dry_run: true

A training or pipeline job that fails on Azure Machine Learning does not page anyone, it just sits there in a workspace nobody is watching until a data scientist happens to check Studio. A job that should have finished in an hour and is still RUNNING a day later is burning compute the whole time, with the same silence. GetJob already reports the exact JobState and a direct Studio link, this blueprint is what reads that report on a schedule and turns it into an alert, with a narrow, operator-reviewed path to cancel the job if that is the right call.

How it works

probe_job_status (io.kestra.plugin.azure.ml.GetJob) reads jobName's current JobState, studioUrl, best-effort MLflow metrics, and named outputs, with allowFailure: true so a bad credential or a renamed job does not stop the flow before it can be classified as PROBE_FAILED below.
classify_job_status (io.kestra.plugin.core.debug.Return) returns PROBE_FAILED when probe_job_status produced no output at all (guarded with is not defined), CRITICAL or WARNING on FAILED, CANCELED, or NOT_RESPONDING depending on failure_is_critical, COMPLETED on a clean finish, and IN_PROGRESS for every other, non-terminal JobState (QUEUED, RUNNING, PROVISIONING, and so on).
route_by_job_status (io.kestra.plugin.core.flow.Switch) branches on that status: COMPLETED and IN_PROGRESS log it, WARNING, CRITICAL, and PROBE_FAILED each alert Slack, and the two real failure alerts include probe_job_status.studioUrl, a direct link into Azure ML Studio for the exact run, not a generic workspace link.
remediate_gate (io.kestra.plugin.core.flow.If) runs only when request_cancel is true and dry_run is false, independent of the classification above, since an operator may want to cancel a job for a reason this gate cannot see on its own, such as a known bad input dataset:
- cancel_job (io.kestra.plugin.azure.ml.CancelJob) requests cancellation and, per the task's own documented behavior, polls until the job actually reaches a terminal state before returning, since Azure ML cancellation is asynchronous.
- report_cancel_result (io.kestra.plugin.core.flow.If) fails the flow with io.kestra.plugin.core.execution.Fail if the cancel call produced no output at all.
log_audit_summary always runs last and logs the final classification.
The errors block posts to Slack if the flow itself fails outside the gate logic above.

What you get

A real answer to whether a specific training or pipeline job actually finished, built from Azure ML's own JobState, not a timeout guess.
A Slack alert that lands with a direct Studio link to the exact failing run, not just a job name someone has to go search for.
A deliberately decoupled cancel path: request_cancel is never set by the classification itself, only by an operator who has already looked at the alert and the Studio run.

Who it's for

ML platform and MLOps teams running training or pipeline jobs on Azure Machine Learning who need to know a job failed, was cancelled, or stopped responding, without opening Studio to check.
Teams evaluating the Azure Machine Learning plugin who want a working example of ml.GetJob and ml.CancelJob together, including the real JobState values and the Studio deep link helper both tasks share.

Why orchestrate this with Kestra

GetJob and CancelJob are single point-in-time calls, Azure ML itself has no concept of a schedule, a policy on what counts as a failure worth paging for, or an alert built from the job's own Studio link. Kestra adds the recurring audit, the failure_is_critical policy choice, the Slack alert carrying a working deep link, and the reviewed cancel path, all as one declarative YAML file you can version and review like code.

Prerequisites

An Azure Machine Learning workspace with a job to audit, and a service principal with Microsoft.MachineLearningServices/workspaces/jobs/read permission (roles/AzureML Data Scientist or a narrower custom role is enough to read, cancelling additionally needs write on the job).
A Slack incoming webhook for alerts.
Azure Machine Learning is a managed service with no self-hostable open-source emulator, there is no local container to start for it. This plugin calls the Azure Resource Manager API directly from the Kestra worker process, there is no sub-container taskRunner and no Docker network to join.

Secrets

AZURE_TENANT_ID, AZURE_CLIENT_ID, AZURE_CLIENT_SECRET, AZURE_SUBSCRIPTION_ID: service principal credentials used by probe_job_status and cancel_job.
SLACK_WEBHOOK: Slack incoming webhook URL used by the alert tasks and the errors block.

In Kestra OSS, set all five as base64-encoded environment variables, for example SECRET_AZURE_TENANT_ID, SECRET_AZURE_SUBSCRIPTION_ID, SECRET_SLACK_WEBHOOK.

Inputs

Input	Type	Default	Allowed values
`resource_group_name`	STRING	`ml-rg`	any resource group name
`workspace_name`	STRING	`ml-workspace`	any Azure ML workspace name
`job_name`	STRING	`training-run-001`	any Azure ML job name
`job_type_label`	SELECT	`command`	`command`, `pipeline`, `sweep`
`failure_is_critical`	BOOL	`true`	`true`, `false`
`request_cancel`	BOOL	`false`	`true`, `false`
`dry_run`	BOOL	`true`	`true`, `false`

Outputs

{{ outputs.probe_job_status.status }}: current JobState string of the Azure ML job.
{{ outputs.probe_job_status.studioUrl }}: deep link URL directly to the job run in Azure Machine Learning Studio.
{{ outputs.probe_job_status.metrics }}: map of MLflow logged metrics (best-effort).
{{ outputs.probe_job_status.outputs }}: map of named job output artifacts.
{{ outputs.classify_job_status.value }}: governance classification (CRITICAL, WARNING, IN_PROGRESS, COMPLETED, or PROBE_FAILED).
{{ outputs.cancel_job.status }}: verified terminal status of the job after cancellation.

Quick start

Add the Azure and SLACK_WEBHOOK secrets.
Submit a job on Azure ML (SubmitCommandJob or your own pipeline) and note its name.
Set job_name to that job and run this flow manually, read log_audit_summary for the classification.
If it reports CRITICAL or WARNING and cancellation is the right call, set request_cancel: true and dry_run: false and run it again.
Enable the every_fifteen_minutes trigger, which always ships in safe mode.

How to extend

Loop this flow over a list of job_name values with io.kestra.plugin.core.flow.Loop to audit every job in a pipeline run from one flow instead of one job at a time.
Read a specific metric from outputs.probe_job_status.metrics (best-effort, from MLflow) and add a metric-based classification, for example a validation loss that never improved, alongside the state-based one.
Chain SubmitCommandJob before this gate in the same flow to submit, then immediately audit, a retrain job end to end.

Pitfalls

GetJob's output has no start time or duration, only the current JobState. This gate cannot tell you a job has been RUNNING too long on its own, only that it is not yet terminal, track duration separately if runaway jobs are the specific risk you care about.
metrics is best-effort: if the service principal lacks MLflow tracking scope, or the tracking server is unreachable, it comes back as an empty map and a warning is logged, not a task failure, do not treat an empty metrics map as proof a job logged nothing.
CancelJob treats cancelling an already-terminal job as a no-op, not an error, running remediate_gate against a job that finished between your last check and now is safe.
NOT_RESPONDING is itself a real JobState, not only FAILED and CANCELED count as failures per the plugin's own JobState.isFailure(), this blueprint checks all three.

Azure ML Training Job Governance Gate

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Inputs

Outputs

Quick start

How to extend

Pitfalls

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Azure ML Training Job Governance Gate

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Inputs

Outputs

Quick start

How to extend

Pitfalls

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Infrastructure Blueprints

Kueue Workload Admission Check and Quota Reservation Sentinel

Run a Python script in a Kubernetes pod

Kubernetes Sealed Secrets Unparseable Decryption Sentinel

Kubernetes Secret Exposure Sentinel

Kubernetes Vertical Pod Autoscaler (VPA) Recommendation Drift Sentinel

Rotate Vault Secrets into Kubernetes with a Rolling Restart

More Cloud Blueprints

Kubernetes StatefulSet Replica Health and Quorum Degradation Sentinel

Find the Kubernetes manifests that will break on your next cluster upgrade

Kueue Workload Admission Check and Quota Reservation Sentinel

Kubernetes cert-manager ACME Order Failure and Ingress TLS Sentinel

Kubernetes Vertical Pod Autoscaler (VPA) Recommendation Drift Sentinel

KubeVirt Virtual Machine Live Migration Stall Sentinel

New to Kestra?