add processing tasks in workflows

hasadna · Aug 5, 2024 · 3236fd1 · 3236fd1
1 parent 7c0dffd
commit 3236fd1
Show file tree

Hide file tree

Showing 4 changed files with 90 additions and 0 deletions.
diff --git a/workflows/templates/argo-workflows-configmap.yaml b/workflows/templates/argo-workflows-configmap.yaml
@@ -0,0 +1,6 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: argo-workflows
+data:
+  datacity-concurrency: {{ .Values.concurrency | quote }}
diff --git a/workflows/templates/continuous-processing-tasks-workflows.yaml b/workflows/templates/continuous-processing-tasks-workflows.yaml
@@ -0,0 +1,56 @@
+{{- range $instance, $config := .Values.continuousProcessingTasksInstanceWorkflows }}
+ {{- range $task := $.Values.continuousProcessingTasks }}
+apiVersion: argoproj.io/v1alpha1
+kind: WorkflowTemplate
+metadata:
+  name: cntprct-{{ $instance }}-{{ $task }}
+  labels:
+    datacity-workflow: continuous-processing-task
+    datacity-instance: {{ $instance | quote }}
+    datacity-cntprct-task: {{ $task | quote }}
+spec:
+  entrypoint: continuous-processing-task
+  synchronization:
+    semaphore:
+      configMapKeyRef:
+        name: argo-workflows
+        key: datacity-concurrency
+  templates:
+    - name: continuous-processing-task
+      synchronization:
+        mutex:
+          name: instance-{{ $instance }}
+      steps:
+        - - name: call-template
+            templateRef:
+              name: dgp-operator-runner
+              template: dgp-operator-runner
+            arguments:
+              parameters:
+                - name: operator
+                  value: packages_processing
+                - name: config_json
+                  value: |
+                    {
+                      "instance": {{ $instance | quote }},
+                      "task": {{ $task | quote }}
+                    }
+---
+apiVersion: argoproj.io/v1alpha1
+kind: CronWorkflow
+metadata:
+  name: cntprct-{{ $instance }}-{{ $task }}
+  labels:
+    datacity-workflow: continuous-processing-task
+    datacity-instance: {{ $instance | quote }}
+    datacity-cntprct-task: {{ $task | quote }}
+spec:
+  schedule: "@hourly"
+  concurrencyPolicy: Forbid
+  suspend: true
+  workflowSpec:
+    workflowTemplateRef:
+      name: cntprct-{{ $instance }}-{{ $task }}
+---
+ {{- end }}
+{{- end }}
diff --git a/workflows/templates/generic-fetcher-instance-organization-workflows.yaml b/workflows/templates/generic-fetcher-instance-organization-workflows.yaml
@@ -6,12 +6,21 @@ kind: WorkflowTemplate
 metadata:
   name: generic-fetcher-{{ $workflow.name_suffix }}
   labels:
+    datacity-workflow: generic-fetcher
     datacity-instance: {{ $instance | quote }}
     datacity-organization: {{ $organization | quote }}
 spec:
   entrypoint: generic-fetcher
+  synchronization:
+    semaphore:
+      configMapKeyRef:
+        name: argo-workflows
+        key: datacity-concurrency
   templates:
     - name: generic-fetcher
+      synchronization:
+        mutex:
+          name: instance-{{ $instance }}
       steps:
         - - name: call-template
             templateRef:
@@ -36,6 +45,7 @@ kind: CronWorkflow
 metadata:
   name: generic-fetcher-{{ $workflow.name_suffix }}
   labels:
+    datacity-workflow: generic-fetcher
     datacity-instance: {{ $instance | quote }}
     datacity-organization: {{ $organization | quote }}
 spec:

diff --git a/workflows/values.yaml b/workflows/values.yaml
@@ -1,3 +1,7 @@
+# The maximum concurrency for all datacity workflows
+# this is important because all workflows run on the dgp server
+concurrency: 4
+
 genericFetcherInstanceOrganizationWorkflows:
   haifa:
     israel-gov:
@@ -91,3 +95,17 @@ genericFetcherInstanceOrganizationWorkflows:
         target_package_id: bus-stops
         source_filter:
           MetropolinName: חיפה
+
+continuousProcessingTasks:
+  - xlsx
+  - geojson
+
+continuousProcessingTasksInstanceWorkflows:
+  haifa: {}
+  jaljulia: {}
+  jerusalem: {}
+  maale-adummim: {}
+  netanya: {}
+  rishonlezion: {}
+  kfar-saba: {}
+  test: {}