Tighter integration towards argo workflows

[olaals]

Describe the improvement you would like to see

We are kind of creating an workflow orchestrator (sara) on top of a workflow orchestrator (argo workflows) in a sense.

If we need more advanced workflow features that argo workflows already provides (such as skip logic in #389) i think we should lean more into the logic that is implemented in argo workflows and sara only acts as a translation layer of our config -> argo workflows config

I think the correct direction to sara is (in terms of the workflow aspect of sara) to act as a thin wrapper around argo workflows, then argo workflows can do the heavy lifting in terms of workflow logic and execution. Argo workflows is a well maintained codebase and they have done many smart choices when it comes to orchestration that we might want to hook into, this is easier if we do not have to create the logic ourself, but just change how we configure workflows against it.

Argo workflows has support for

• Linear and non-linear workflows (DAGs), right now we only do linear jobs but i think non-linear jobs might be wanted at some point)
• Mapping of inputs and outputs from different steps (now we map the output of one job directly as the input to the next)
• Skip logic (also flexible in terms of running jobs on failed / success)
  and more

so if we need this functionality its better to hand it off to argo workflows than implementing it in sara (if we are continuing to use argo workflows)

Going in this direction kind of rules out Radix more, because we are more radix compatible as of now than if we do a tighter integration against argo workflows. With Radix we will have to do more of this orchestration logic ourself since radix only has concepts of single jobs, not series of jobs etc.

With the large refactor recently, i think it is easier to do this tighter integration against argo workflows since the db models maps better against the concepts in argo.

Key differences:

• Workflows are not in analytics-infrastructure, only workflow templates are in the infrastructure code
• (argo) Workflows are generated dynamically in sara and submitted directly to kubernetes.
• 1 workflow in sara is 1 workflow in argo workflows (now 1 workflow in sara is 1 or more workflows in argo worklows)
• We keep our own definition of workflows (like we do in appsettings.json) but we map our own definition to argo workflows definition.
• We let argo workflows handle the progression of the whole workflow, so the whole workflow definition is generated upfront.
• Workflow notifier steps are not in the exit handler, but rather just steps in the pipeline. There can be an overall analysis notifier that notifies of the overall analysis status at the end. We can also utilize the watcher API from k8s to watch the workflow as it progresses if we need more information about the workflow.
• No need for Argo Sensor setup, which simplifies things and needs less resources running in the cluster

What should stay the same:

• MQTT messages are generated "mid-pipeline", e.g. for example that the anonymized image is available if that step is completed but the overall workflow is not
• Most db models should stay the same
• The workflow config in appsettings.json should stay the same unless there is a very good reason to change it
• Workflow configs such as memory usage and image version should stay in the infrastructure repo, through workflow templates

TODOS:

• Refactor to generate argo workflow definitions based on our config
• Refactor so that SARA can understand from the workflow generation step how to wire inputs and outputs in argo workflows
• Generate k8s roles for allowing SARA container to create workflows directly
• Investigate the need for watching the workflow via k8s api
• Move workflows out of analytics infrastructure, only keep templates where it makes sense
• For the initial iteration: try to keep db models the same as they are now, unless a big workaround is needed

AI generated proposal:

Current State Summary

Right now, SARA uses Argo Events (EventSource + Sensor) as an indirection layer between itself and Argo Workflows. The flow is:

SARA -> HTTP POST -> EventSource webhook -> Sensor -> submits Argo Workflow

And SARA handles multi-step chaining itself: when step N completes, the workflow-notifier calls back into SARA, and SARA triggers step N+1 via another webhook POST. Argo only ever sees individual single-analyzer workflows (notify-start -> run -> notify-result), with no awareness of the broader pipeline.

Proposed Architecture: SARA as Direct Argo Workflow Submitter

Current Flow

MQTT message arrives
-> SARA creates DB records (Analysis, AnalysisRun, Workflow rows)
-> SARA POSTs to Argo EventSource webhook
-> Argo Sensor picks up event, submits single-step Argo Workflow
-> Argo runs: notify-started -> run-analyzer -> notify-result
-> workflow-notifier calls back to SARA with result
-> SARA triggers next step via another webhook POST
-> repeat until chain complete
Each analyzer runs as its own isolated Argo Workflow. SARA orchestrates the chain. Three Kubernetes resources (EventSource, Service, Sensor) + an EventBus exist per workflow type.

Proposed Flow

MQTT message arrives
-> SARA creates DB records (Analysis, AnalysisRun)
-> SARA pre-computes all parameters (blob locations, extras, thermal reference metadata)
-> SARA submits a single Argo Workflow directly via K8s API
-> Argo executes the full pipeline as steps within one Workflow
-> SARA watches Workflow CR status or receives exit handler callback
-> SARA updates DB with final outcome
Example: Anonymizer + Thermal Reading Pipeline
SARA submits a single Argo Workflow that looks roughly like this:

apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  generateName: analysis-run-
  labels:
    sara.equinor.com/analysis-run-id: "<analysisRunId>"
spec:
  arguments:
    parameters:
      - name: analysis-run-id
        value: "<analysisRunId>"
      - name: sara-callback-url
        value: "https://sara.example.com/api"
      - name: input-blob-storage-locations
        value: '<json>'
      - name: anonymizer-output-location
        value: '<json>'
      - name: thermal-reading-output-location
        value: '<json>'
      - name: thermal-reference-image-location
        value: '<json>'
      - name: thermal-reference-polygon-location
        value: '<json>'

  entrypoint: pipeline
  onExit: notify-exited

  templates:
    - name: pipeline
      steps:
        # Step 1: Notify SARA that the pipeline has started
        - - name: notify-started
            templateRef:
              name: workflow-notifier
              template: notify
            arguments:
              parameters:
                - name: callback-url
                  value: "{{workflow.parameters.sara-callback-url}}"
                - name: analysis-run-id
                  value: "{{workflow.parameters.analysis-run-id}}"
                - name: status
                  value: "started"

        # Step 2: Run anonymizer
        - - name: run-anonymizer
            templateRef:
              name: anonymizer
              template: run
            arguments:
              parameters:
                - name: input-blob-storage-locations
                  value: "{{workflow.parameters.input-blob-storage-locations}}"
                - name: output-blob-storage-location
                  value: "{{workflow.parameters.anonymizer-output-location}}"

        # Step 3: Notify SARA that anonymizer completed, pass result
        - - name: notify-anonymizer-done
            templateRef:
              name: workflow-notifier
              template: notify
            arguments:
              parameters:
                - name: callback-url
                  value: "{{workflow.parameters.sara-callback-url}}"
                - name: analysis-run-id
                  value: "{{workflow.parameters.analysis-run-id}}"
                - name: step
                  value: "anonymizer"
                - name: status
                  value: "step-completed"
                - name: result
                  value: "{{steps.run-anonymizer.outputs.parameters.result}}"

        # Step 4: Run thermal reading (inputs reference anonymizer output)
        - - name: run-thermal-reading
            templateRef:
              name: thermal-reading
              template: run
            arguments:
              parameters:
                - name: input-blob-storage-locations
                  value: "{{steps.run-anonymizer.outputs.parameters.pre-processed-location}}"
                - name: output-blob-storage-location
                  value: "{{workflow.parameters.thermal-reading-output-location}}"
                - name: reference-image-location
                  value: "{{workflow.parameters.thermal-reference-image-location}}"
                - name: reference-polygon-location
                  value: "{{workflow.parameters.thermal-reference-polygon-location}}"

        # Step 5: Notify SARA that thermal reading completed
        - - name: notify-thermal-done
            templateRef:
              name: workflow-notifier
              template: notify
            arguments:
              parameters:
                - name: callback-url
                  value: "{{workflow.parameters.sara-callback-url}}"
                - name: analysis-run-id
                  value: "{{workflow.parameters.analysis-run-id}}"
                - name: step
                  value: "thermal-reading"
                - name: status
                  value: "step-completed"
                - name: result
                  value: "{{steps.run-thermal-reading.outputs.parameters.result}}"

    # Always runs regardless of success/failure
    - name: notify-exited
      templateRef:
        name: workflow-notifier
        template: notify
      arguments:
        parameters:
          - name: callback-url
            value: "{{workflow.parameters.sara-callback-url}}"
          - name: analysis-run-id
            value: "{{workflow.parameters.analysis-run-id}}"
          - name: status
            value: "{{workflow.status}}"
          - name: error
            value: "{{workflow.failures}}"

Analytics-infrastructure has one WorkflowTemplate for the notifier with a single generic notify template that accepts parameters like callback-url, workflow-id, status, step-name, result-json. SARA generates the full Workflow CR with inline step definitions that call this generic template.

What the generic notifier template would look like
apiVersion: argoproj.io/v1alpha1
kind: WorkflowTemplate
metadata:
  name: workflow-notifier
spec:
  templates:
    - name: notify
      serviceAccountName: workflow-notifier-sa
      container:
        image: workflow-notifier
        command: ["python", "main.py"]
        args:
          - "{{inputs.parameters.command}}"      # started | result | exited
          - "{{inputs.parameters.workflow-id}}"
          # result and exited pass additional args
        envFrom:
          - configMapRef:
              name: workflow-notifier-config
          - secretRef:
              name: workflow-notifier-secrets
      inputs:
        parameters:
          - name: command
          - name: workflow-id
          - name: result-json
            default: ""
          - name: exit-status
            default: ""
          - name: error-message
            default: ""

How will this change existing functionality?
Less code in sara and making it easier to implement advanced funcionality

How will this improvement affect the current Threat Model?
Letting Sara create workflows might be less secure as it has more access in k8s, would have to go deeper into this how it can be remediated.

[github-actions]

This issue has been automatically marked as stale due to inactivity.