mirror of
https://github.com/kerberos-io/helm-charts.git
synced 2026-08-23 15:18:33 +00:00
Merge pull request #109 from kerberos-io/Workflows->-standalone-setup
Workflows > standalone setup
This commit is contained in:
Kilian
@@ -0,0 +1,42 @@
|
||||
{{/*
|
||||
Assemble the workflows engine stage registry (PIPELINE_STAGE_REGISTRY) as a
|
||||
JSON array from every *enabled* stage under kerberoshub.workflows.stages, so the
|
||||
engine's routing stays in lockstep with the deployed stage workers and the
|
||||
per-stage queue cannot drift. The `stages:` block is the routing source of truth;
|
||||
each stage's deployment (and its queue) lives under the matching
|
||||
kerberoshub.services.<name> entry.
|
||||
|
||||
Each enabled stage contributes one descriptor:
|
||||
operation defaults to the stage's map key.
|
||||
dispatch defaults to "always".
|
||||
queue taken from the matching services.<key>.queue (authoritative; the
|
||||
same value the worker consumes via <NAME>_QUEUE). Omitted when the
|
||||
service or its queue is unset, so the engine derives
|
||||
"kcloud-<operation>-queue.fifo".
|
||||
needs conditional stages only: the upstream dependencies (each
|
||||
{operation, condition?}) carried through verbatim.
|
||||
needsMode conditional stages with more than one need: how they combine —
|
||||
"any" (default; fire on the first matching upstream) or "all"
|
||||
(a join; fire only once every need has resolved and matched).
|
||||
Carried through verbatim; omitted when unset (engine defaults any).
|
||||
kind delegated-ingest stages only: the ingest handler the engine routes
|
||||
the stage's typed payload through (e.g. "detection"). Carried
|
||||
through verbatim; omitted when unset (engine treats the stage as
|
||||
self-persisting).
|
||||
*/}}
|
||||
{{- define "kerberoshub.workflows.stageRegistry" -}}
|
||||
{{- $entries := list -}}
|
||||
{{- $services := .Values.kerberoshub.services | default dict -}}
|
||||
{{- range $name, $stage := .Values.kerberoshub.workflows.stages -}}
|
||||
{{- if $stage.enabled -}}
|
||||
{{- $entry := dict "operation" (default $name $stage.operation) "dispatch" (default "always" $stage.dispatch) -}}
|
||||
{{- $service := index $services $name -}}
|
||||
{{- if $service }}{{- with $service.queue }}{{- $_ := set $entry "queue" . -}}{{- end }}{{- end }}
|
||||
{{- with $stage.needs }}{{- $_ := set $entry "needs" . -}}{{- end }}
|
||||
{{- with $stage.needsMode }}{{- $_ := set $entry "needsMode" . -}}{{- end }}
|
||||
{{- with $stage.kind }}{{- $_ := set $entry "kind" . -}}{{- end }}
|
||||
{{- $entries = append $entries $entry -}}
|
||||
{{- end -}}
|
||||
{{- end -}}
|
||||
{{- $entries | toJson -}}
|
||||
{{- end -}}
|
||||
124
charts/hub/templates/kerberos-pipeline/hub-stage.yaml
Normal file
124
charts/hub/templates/kerberos-pipeline/hub-stage.yaml
Normal file
@@ -0,0 +1,124 @@
|
||||
{{- /*
|
||||
Generic workflow-stage worker.
|
||||
|
||||
Renders a Deployment + Service for every stage under
|
||||
kerberoshub.workflows.stages that has a matching, enabled worker under
|
||||
kerberoshub.services.<name>. A custom stage joins the pipeline by values
|
||||
alone — no per-stage template needed.
|
||||
|
||||
Every stage worker receives the same connection contract; the only value that
|
||||
varies by stage is the consume-queue variable name, <NAME>_QUEUE (a stage
|
||||
keyed "anpr" gets ANPR_QUEUE, "my-stage" gets MY_STAGE_QUEUE). To run a worker
|
||||
outside the chart instead, leave services.<name>.enabled unset (or false)
|
||||
while keeping the stage under workflows.stages so the engine still routes to
|
||||
the queue you publish.
|
||||
*/ -}}
|
||||
{{- if and (or (eq .Values.mode "all") (eq .Values.mode "pipeline")) .Values.kerberoshub.workflows.enabled -}}
|
||||
{{- $root := . -}}
|
||||
{{- $services := .Values.kerberoshub.services | default dict -}}
|
||||
{{- range $name, $stage := .Values.kerberoshub.workflows.stages -}}
|
||||
{{- $svc := index $services $name -}}
|
||||
{{- if and $svc $svc.enabled -}}
|
||||
{{- $queueEnv := printf "%s_QUEUE" ($name | upper | replace "-" "_") -}}
|
||||
---
|
||||
apiVersion: apps/v1
|
||||
kind: Deployment
|
||||
metadata:
|
||||
name: hub-{{ $name }}
|
||||
namespace: {{ $root.Release.Namespace }}
|
||||
spec:
|
||||
replicas: {{ $svc.replicas | default 1 }}
|
||||
selector:
|
||||
matchLabels:
|
||||
app: hub-{{ $name }}
|
||||
minReadySeconds: 10
|
||||
strategy:
|
||||
type: RollingUpdate
|
||||
rollingUpdate:
|
||||
maxUnavailable: 1
|
||||
maxSurge: 1
|
||||
template:
|
||||
metadata:
|
||||
labels:
|
||||
app: hub-{{ $name }}
|
||||
spec:
|
||||
{{- if $root.Values.kerberoshub.serviceAccount.create }}
|
||||
serviceAccountName: {{ default (printf "%s-%s-sa" $root.Release.Name $root.Chart.Name | trunc 63 | trimSuffix "-") $root.Values.kerberoshub.serviceAccount.name }}
|
||||
{{- end }}
|
||||
{{- with $root.Values.imagePullSecrets }}
|
||||
imagePullSecrets:
|
||||
{{- toYaml . | nindent 8 }}
|
||||
{{- end }}
|
||||
{{- with $svc.volumes }}
|
||||
volumes:
|
||||
{{- toYaml . | nindent 8 }}
|
||||
{{- end }}
|
||||
{{- with $svc.topologySpreadConstraints }}
|
||||
topologySpreadConstraints:
|
||||
{{- toYaml . | nindent 8 }}
|
||||
{{- end }}
|
||||
containers:
|
||||
- name: hub-{{ $name }}
|
||||
image: "{{ $root.Values.global.imageRegistry }}{{ $svc.repository }}:{{ $svc.tag }}"
|
||||
imagePullPolicy: {{ $svc.pullPolicy | default "IfNotPresent" }}
|
||||
{{- with $svc.resources }}
|
||||
resources:
|
||||
{{- toYaml . | nindent 12 }}
|
||||
{{- end }}
|
||||
{{- with $svc.volumeMounts }}
|
||||
volumeMounts:
|
||||
{{- toYaml . | nindent 12 }}
|
||||
{{- end }}
|
||||
env:
|
||||
- name: LOG_LEVEL
|
||||
value: "{{ $svc.logLevel | default "info" }}"
|
||||
- name: QUEUE_SYSTEM
|
||||
value: "{{ $root.Values.queueProvider }}"
|
||||
|
||||
# The queue this stage worker consumes dispatched "{{ $name }}" stages
|
||||
# from ({{ $queueEnv }}) and the workflows engine queue it routes its
|
||||
# result back to (WORKFLOWS_QUEUE, so the run records the resolution and
|
||||
# any stage that needs "{{ $name }}" can fire).
|
||||
- name: {{ $queueEnv }}
|
||||
value: "{{ $svc.queue }}"
|
||||
- name: WORKFLOWS_QUEUE
|
||||
value: "{{ $root.Values.kerberoshub.workflows.queue }}"
|
||||
|
||||
# RabbitMQ settings
|
||||
- name: RABBITMQ_HOST
|
||||
value: "{{ $root.Values.rabbitmq.host }}"
|
||||
- name: RABBITMQ_EXCHANGE
|
||||
value: "{{ $root.Values.rabbitmq.exchange }}"
|
||||
- name: RABBITMQ_USERNAME
|
||||
value: "{{ $root.Values.rabbitmq.username }}"
|
||||
- name: RABBITMQ_PASSWORD
|
||||
value: "{{ $root.Values.rabbitmq.password }}"
|
||||
|
||||
# Kerberos Vault — global storage credentials this stage uses to fetch
|
||||
# the media it operates on.
|
||||
- name: KERBEROS_STORAGE_URI
|
||||
value: "{{ $root.Values.kerberosvault.uri }}"
|
||||
- name: KERBEROS_STORAGE_ACCESS_KEY
|
||||
value: "{{ $root.Values.kerberosvault.accesskey }}"
|
||||
- name: KERBEROS_STORAGE_SECRET
|
||||
value: "{{ $root.Values.kerberosvault.secretkey }}"
|
||||
---
|
||||
apiVersion: v1
|
||||
kind: Service
|
||||
metadata:
|
||||
name: hub-{{ $name }}
|
||||
namespace: {{ $root.Release.Namespace }}
|
||||
labels:
|
||||
app: hub-{{ $name }}
|
||||
service: pipe
|
||||
spec:
|
||||
ports:
|
||||
- name: hub-metrics
|
||||
port: 8080
|
||||
targetPort: 8080
|
||||
protocol: TCP
|
||||
selector:
|
||||
app: hub-{{ $name }}
|
||||
{{- end }}
|
||||
{{- end }}
|
||||
{{- end }}
|
||||
114
charts/hub/templates/kerberos-pipeline/hub-workflows.yaml
Normal file
114
charts/hub/templates/kerberos-pipeline/hub-workflows.yaml
Normal file
@@ -0,0 +1,114 @@
|
||||
{{- if and (or (eq .Values.mode "all") (eq .Values.mode "pipeline")) .Values.kerberoshub.workflows.enabled -}}
|
||||
apiVersion: apps/v1
|
||||
kind: Deployment
|
||||
metadata:
|
||||
name: hub-workflows
|
||||
namespace: {{ .Release.Namespace }}
|
||||
spec:
|
||||
replicas: {{ .Values.kerberoshub.workflows.replicas }}
|
||||
selector:
|
||||
matchLabels:
|
||||
app: hub-workflows
|
||||
minReadySeconds: 10
|
||||
strategy:
|
||||
type: RollingUpdate
|
||||
rollingUpdate:
|
||||
maxUnavailable: 1
|
||||
maxSurge: 1
|
||||
template:
|
||||
metadata:
|
||||
annotations:
|
||||
checksum/config: {{ include (print $.Template.BasePath "/configmap-mongodb.yaml") . | sha256sum }}
|
||||
labels:
|
||||
app: hub-workflows
|
||||
spec:
|
||||
{{- if .Values.kerberoshub.serviceAccount.create }}
|
||||
serviceAccountName: {{ default (printf "%s-%s-sa" .Release.Name .Chart.Name | trunc 63 | trimSuffix "-") .Values.kerberoshub.serviceAccount.name }}
|
||||
{{- end }}
|
||||
{{- with .Values.imagePullSecrets }}
|
||||
imagePullSecrets:
|
||||
{{- toYaml . | nindent 8 }}
|
||||
{{- end }}
|
||||
{{- with .Values.kerberoshub.workflows.volumes }}
|
||||
volumes:
|
||||
{{- toYaml . | nindent 8 }}
|
||||
{{- end }}
|
||||
{{- with .Values.kerberoshub.workflows.topologySpreadConstraints }}
|
||||
topologySpreadConstraints:
|
||||
{{- toYaml . | nindent 8 }}
|
||||
{{- end }}
|
||||
containers:
|
||||
- name: hub-workflows
|
||||
image: "{{ .Values.global.imageRegistry }}{{ .Values.kerberoshub.workflows.repository }}:{{ .Values.kerberoshub.workflows.tag }}"
|
||||
imagePullPolicy: {{ .Values.kerberoshub.workflows.pullPolicy }}
|
||||
{{- with .Values.kerberoshub.workflows.resources }}
|
||||
resources:
|
||||
{{- toYaml . | nindent 12 }}
|
||||
{{- end }}
|
||||
{{- with .Values.kerberoshub.workflows.volumeMounts }}
|
||||
volumeMounts:
|
||||
{{- toYaml . | nindent 12 }}
|
||||
{{- end }}
|
||||
envFrom:
|
||||
- configMapRef:
|
||||
name: mongodb-config
|
||||
env:
|
||||
- name: LOG_LEVEL
|
||||
value: "{{ .Values.kerberoshub.workflows.logLevel }}"
|
||||
- name: QUEUE_SYSTEM
|
||||
value: "{{ .Values.queueProvider }}"
|
||||
|
||||
# Queue this service consumes from (WORKFLOWS_QUEUE) and the custom
|
||||
# pipeline stage registry it may dispatch (PIPELINE_STAGE_REGISTRY).
|
||||
# The registry is assembled from the enabled stages under
|
||||
# kerberoshub.workflows.stages (see _workflows-helpers.tpl).
|
||||
- name: WORKFLOWS_QUEUE
|
||||
value: "{{ .Values.kerberoshub.workflows.queue }}"
|
||||
- name: PIPELINE_STAGE_REGISTRY
|
||||
value: {{ include "kerberoshub.workflows.stageRegistry" . | quote }}
|
||||
|
||||
# RabbitMQ settings
|
||||
- name: RABBITMQ_HOST
|
||||
value: "{{ .Values.rabbitmq.host }}"
|
||||
- name: RABBITMQ_EXCHANGE
|
||||
value: "{{ .Values.rabbitmq.exchange }}"
|
||||
- name: RABBITMQ_USERNAME
|
||||
value: "{{ .Values.rabbitmq.username }}"
|
||||
- name: RABBITMQ_PASSWORD
|
||||
value: "{{ .Values.rabbitmq.password }}"
|
||||
|
||||
# Kerberos Vault — global storage credentials a dispatched stage worker
|
||||
# uses to fetch the media. Per-recording vault overrides (site/account)
|
||||
# are resolved at dispatch time from the database.
|
||||
- name: KERBEROS_STORAGE_URI
|
||||
value: "{{ .Values.kerberosvault.uri }}"
|
||||
- name: KERBEROS_STORAGE_ACCESS_KEY
|
||||
value: "{{ .Values.kerberosvault.accesskey }}"
|
||||
- name: KERBEROS_STORAGE_SECRET
|
||||
value: "{{ .Values.kerberosvault.secretkey }}"
|
||||
|
||||
# Open Telemetry tracing
|
||||
- name: OTEL_EXPORTED_OTLP_ENABLED
|
||||
value: "{{ .Values.opentelemetry.enabled }}"
|
||||
- name: OTEL_EXPORTED_OTLP_ROUTING_ENABLED
|
||||
value: "{{ .Values.opentelemetry.routingEnabled }}"
|
||||
- name: OTEL_EXPORTER_OTLP_ENDPOINT
|
||||
value: "{{ .Values.opentelemetry.collector.endpoint }}"
|
||||
---
|
||||
apiVersion: v1
|
||||
kind: Service
|
||||
metadata:
|
||||
name: hub-workflows
|
||||
namespace: {{ .Release.Namespace }}
|
||||
labels:
|
||||
app: hub-workflows
|
||||
service: pipe
|
||||
spec:
|
||||
ports:
|
||||
- name: hub-metrics
|
||||
port: 8080
|
||||
targetPort: 8080
|
||||
protocol: TCP
|
||||
selector:
|
||||
app: hub-workflows
|
||||
{{- end }}
|
||||
@@ -81,6 +81,12 @@ spec:
|
||||
value: "{{ .Values.rabbitmq.username }}"
|
||||
- name: RABBITMQ_PASSWORD
|
||||
value: "{{ .Values.rabbitmq.password }}"
|
||||
|
||||
# When true, analysis tees the classify result to the hub-workflows
|
||||
# service in parallel with the throttler/notification tail (which still
|
||||
# runs unchanged). Kept in sync with whether the workflows service runs.
|
||||
- name: WORKFLOWS_ENABLED
|
||||
value: "{{ .Values.kerberoshub.workflows.enabled }}"
|
||||
|
||||
# Kerberos Vault
|
||||
- name: KERBEROS_STORAGE_URI
|
||||
|
||||
@@ -592,6 +592,139 @@ kerberoshub:
|
||||
requests:
|
||||
memory: 10Mi
|
||||
cpu: 10m
|
||||
# hub-workflows is the standalone, queue-driven workflow engine. It consumes
|
||||
# pipeline events and dispatches custom stages declared in the stage registry,
|
||||
# tracking each run in its own `workflow_runs` collection. It shares events
|
||||
# (not a document) with the analysis pipeline and is meant to grow into the
|
||||
# primary orchestrator. See https://github.com/uug-ai/hub-workflows.
|
||||
workflows:
|
||||
# Disabled by default. When enabled, the analysis service tees each
|
||||
# classify result to this service (via WORKFLOWS_ENABLED on pipe-analysis)
|
||||
# in parallel with the normal throttler/notification tail, which still runs
|
||||
# unchanged. Flip to true to run the workflows engine.
|
||||
enabled: false
|
||||
repository: ghcr.io/uug-ai/hub-workflows
|
||||
pullPolicy: IfNotPresent
|
||||
tag: "v1.0.0"
|
||||
replicas: 1 # Number of pods for the service.
|
||||
topologySpreadConstraints: [] # Optional pod topology spread constraints (empty = none).
|
||||
# Optional extra volumes / volumeMounts for this deployment (empty = none).
|
||||
#volumes:
|
||||
# - name: extra
|
||||
# emptyDir: {}
|
||||
#volumeMounts:
|
||||
# - name: extra
|
||||
# mountPath: /data
|
||||
logLevel: "info" # possible values: trace, debug, info, warn, error
|
||||
# Queue this service consumes ingest events and upstream results from
|
||||
# (WORKFLOWS_QUEUE). Must be fed the same messages the analysis service sees.
|
||||
queue: "kcloud-workflows-queue"
|
||||
resources:
|
||||
requests:
|
||||
memory: 10Mi
|
||||
cpu: 10m
|
||||
# Stage routing — how each custom stage fits into the workflow. The engine's
|
||||
# PIPELINE_STAGE_REGISTRY is assembled automatically from the *enabled*
|
||||
# stages below: each contributes its operation and dispatch/needs routing,
|
||||
# plus the queue from its matching `services.<name>` entry — so the engine's
|
||||
# routing stays in lockstep with the deployed worker and the queue cannot
|
||||
# drift. There is nothing to set for the registry directly.
|
||||
#
|
||||
# Each stage is keyed to a worker of the same name under kerberoshub.services.
|
||||
# Routing (here) and deployment (services) are toggled independently:
|
||||
# workflows.stages.<name>.enabled -> include the stage in the registry (route to it)
|
||||
# services.<name>.enabled -> deploy the worker (pipe-<name>.yaml)
|
||||
stages:
|
||||
# hub-anpr — automatic number-plate recognition stage. Routing only; its
|
||||
# worker deployment lives under kerberoshub.services.anpr.
|
||||
anpr:
|
||||
# Include this stage in the engine's registry (route to it).
|
||||
enabled: false
|
||||
# operation unique stage id (defaults to the key "anpr" if omitted);
|
||||
# binds the queue (taken from services.anpr.queue) and how
|
||||
# the result is recorded.
|
||||
# dispatch "always" (run on every workflow) | "conditional".
|
||||
# needs conditional stages only: upstream dependencies, each
|
||||
# {operation?, condition}. operation is the readiness GATE —
|
||||
# the upstream op whose data must be present before the
|
||||
# condition is read; leave it empty for a check on the run
|
||||
# root itself (device/user/identity), read as soon as the run
|
||||
# opens. condition shape:
|
||||
# {path: <abs-path>, op: eq|ne|contains|in|exists|gt|gte|lt|lte, value: <operand>}.
|
||||
# path is ABSOLUTE from the run root and resolves through
|
||||
# string-keyed maps only (it CANNOT index into arrays):
|
||||
# inputs.<op>.<field> (a trigger result, e.g. classify),
|
||||
# results.<op>.<field> (a finished stage), device.<field>,
|
||||
# user.<field>, or a top-level scalar (operation/runId/key).
|
||||
# classify is the only operation teed to workflows, so it is
|
||||
# the reliable upstream: match inputs.classify.properties (the
|
||||
# detected class strings) with `contains` — there is no
|
||||
# top-level `label`, and inputs.classify.details is an array a
|
||||
# path cannot index into — or gate on inputs.classify.objectCount
|
||||
# (top-level int) numerically. The engine rejects an unknown
|
||||
# path at boot.
|
||||
# needsMode conditional stages with more than one need: how they
|
||||
# combine. "any" (default) fires on the first matching
|
||||
# need; "all" is a join — the stage fires only once every
|
||||
# need has resolved and each condition matches, and only once.
|
||||
operation: anpr
|
||||
dispatch: conditional
|
||||
# kind routes this stage's typed result (the PostANPRRequest its worker
|
||||
# returns in the run Payload) through the engine's shared ingest core,
|
||||
# which persists it into the dedicated "anpr" collection and creates one
|
||||
# marker per recognised plate. Without it the result is only recorded for
|
||||
# routing, not persisted.
|
||||
kind: anpr
|
||||
# Default routing — run anpr whenever classify reports a car (any camera).
|
||||
needsMode: any
|
||||
needs:
|
||||
- operation: classify
|
||||
condition: {path: "inputs.classify.properties", op: contains, value: car}
|
||||
# Restrict plate recognition to a SINGLE camera: add an ungated need on
|
||||
# the recording's device (empty operation = checked as soon as the run
|
||||
# opens) and switch needsMode to "all", so both must hold — "that camera
|
||||
# AND a car was detected". For several cameras use op: in with a list.
|
||||
#needsMode: all
|
||||
#needs:
|
||||
# - operation: classify
|
||||
# condition: {path: "inputs.classify.properties", op: contains, value: car}
|
||||
# - operation:
|
||||
# condition: {path: "device.deviceKey", op: eq, value: device02}
|
||||
|
||||
# Workflow stage worker deployments (siblings of the workflows engine). Each
|
||||
# entry is a standalone service the engine dispatches to; it consumes its own
|
||||
# queue and routes a result back to the workflows queue. Keyed to match a
|
||||
# workflows.stages entry of the same name. Deploying a worker here is
|
||||
# independent from routing to it (workflows.stages.<name>.enabled) — a worker
|
||||
# only runs when its `enabled` is true AND the workflows engine is enabled.
|
||||
services:
|
||||
# hub-anpr — automatic number-plate recognition worker. Lives in the
|
||||
# hub-workflows repository as its own module. See
|
||||
# https://github.com/uug-ai/hub-workflows/tree/main/hub-anpr.
|
||||
anpr:
|
||||
# Deploy the hub-anpr worker.
|
||||
enabled: false
|
||||
repository: ghcr.io/uug-ai/hub-anpr
|
||||
pullPolicy: IfNotPresent
|
||||
tag: "v1.0.0"
|
||||
replicas: 1 # Number of pods for the worker.
|
||||
topologySpreadConstraints: [] # Optional pod topology spread constraints (empty = none).
|
||||
#volumes:
|
||||
# - name: extra
|
||||
# emptyDir: {}
|
||||
#volumeMounts:
|
||||
# - name: extra
|
||||
# mountPath: /data
|
||||
logLevel: "info" # possible values: trace, debug, info, warn, error
|
||||
# Queue this worker consumes dispatched messages from (ANPR_QUEUE). This
|
||||
# same value is read into the engine's generated registry entry for the
|
||||
# matching stage, so the engine dispatches and the worker consumes the same
|
||||
# queue with no drift. Convention: "kcloud-<operation>-queue.fifo".
|
||||
queue: "kcloud-anpr-queue.fifo"
|
||||
resources:
|
||||
requests:
|
||||
memory: 10Mi
|
||||
cpu: 10m
|
||||
monitordevice:
|
||||
repository: ghcr.io/uug-ai/hub-monitor-device
|
||||
pullPolicy: IfNotPresent
|
||||
|
||||
Reference in New Issue
Block a user