fix: support clustered Redis workflow cleanup

Author: ecolternv

deployments/charts/osmo/templates/worker.yaml

@@ -128,6 +128,8 @@ spec:
         volumeMounts:
         - name: osmo-progress-files
           mountPath: /var/run/osmo
+        - name: osmo-runtime-tmp
+          mountPath: /tmp
         {{- if .Values.secrets.masterEncryptionKey.existingSecret.name }}
         - mountPath: {{ include "osmo.secrets.mekMountPath" . | quote }}
           name: mek-volume
@@ -160,6 +162,8 @@ spec:
       volumes:
         - name: osmo-progress-files
           emptyDir: {}
+        - name: osmo-runtime-tmp
+          emptyDir: {}
         {{- include "osmo.pod.extraVolumes" .Values.services.worker | nindent 8 }}
         {{- include "osmo.secrets.mekVolume" . | nindent 8 }}
         {{- include "osmo.configuration.volumes" . | nindent 8 }}

deployments/charts/osmo/tests/test_osmo_charts.sh

@@ -69,6 +69,18 @@ require_occurrences() {
         fail "expected '$expected' $count times in $file, found $actual"
 }
 
+require_empty_dir_volume() {
+    local file=$1
+    local volume_name=$2
+    awk -v volume_name="$volume_name" '
+        $0 == "        - name: " volume_name {
+            getline
+            if ($0 == "          emptyDir: {}") found = 1
+        }
+        END { exit !found }
+    ' "$file" || fail "expected emptyDir volume '$volume_name' in $file"
+}
+
 require_line_count() {
     local file=$1
     local expected=$2
@@ -1398,11 +1410,13 @@ EOF
         require_contains "$TEST_DIRECTORY/osmo-$hardened_component.yaml" \
             "readOnlyRootFilesystem: true"
     done
-    for hardened_component in api router logger agent; do
+    for hardened_component in api worker router logger agent; do
         require_contains "$TEST_DIRECTORY/osmo-$hardened_component.yaml" \
             "mountPath: /tmp"
         require_contains "$TEST_DIRECTORY/osmo-$hardened_component.yaml" \
             "name: osmo-runtime-tmp"
+        require_empty_dir_volume "$TEST_DIRECTORY/osmo-$hardened_component.yaml" \
+            "osmo-runtime-tmp"
     done
     for hardened_component in worker logger agent delayed-job-monitor; do
         require_contains "$TEST_DIRECTORY/osmo-$hardened_component.yaml" \

src/utils/job/jobs.py

@@ -1268,7 +1268,7 @@ def _notify_barrier(self, database, redis_client, total_timeout: int):
             retry_ids = task.Task.batch_fetch_latest_retry_ids(
                 database, self.workflow_id, task_names)
 
-            pipe = redis_client.pipeline()
+            pipe = redis_client.pipeline(transaction=False)
             pipe.set(action_key, json.dumps(attributes))
             pipe.expire(action_key, total_timeout, nx=True)
 
@@ -1441,7 +1441,7 @@ def execute(self, context: JobExecutionContext,
 
         redis_client = redis.from_url(workflow_obj.logs)
 
-        redis_batch_pipeline = redis_client.pipeline()
+        redis_batch_pipeline = redis_client.pipeline(transaction=False)
 
         if workflow_obj.status.failed():
             start_delimiter = '\n' + '-' * 100 + '\n'

src/utils/job/tests/test_jobs_pure.py

@@ -372,6 +372,7 @@ def test_when_enough_members_in_barrier_emits_notifications(self):
              mock.patch.object(task.Task, 'batch_fetch_latest_retry_ids',
                                return_value={'t1': 0, 't2': 0}):
             ug._notify_barrier(database, client, total_timeout=60)
+        client.pipeline.assert_called_once_with(transaction=False)
         pipe.execute.assert_called_once()
         # Two members should each lpush once.
         self.assertEqual(pipe.lpush.call_count, 2)
@@ -1923,6 +1924,7 @@ def test_full_execute_with_credential_calls_storage_upload(self):
                                new=mock.AsyncMock(return_value=mock.Mock())):
             result = cw.execute(ctx, progress_writer)
         self.assertIsInstance(result, jobs_base.JobResult)
+        redis_client.pipeline.assert_called_once_with(transaction=False)
         # Update logs called for both logs and events
         wf.update_log_to_db.assert_called_once()
         wf.update_events_to_db.assert_called_once()