Multistage Restores

schema/versions/5661f79ddf29_multistage_restores.py

+"""multistage restores
+
+Revision ID: 5661f79ddf29
+Revises: 09dec567edcd
+Create Date: 2021-10-11 15:55:57.500671
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = "5661f79ddf29"
+down_revision = "09dec567edcd"
+branch_labels = None
+depends_on = None
+
+
+dag_content = """JOB FETCH restore_cms_fetch.condor
+VARS FETCH jobname="$(JOB)"
+
+JOB ENVOY restore_cms_envoy.condor
+VARS ENVOY jobname="$(JOB)"
+
+JOB DELIVER restore_cms_deliver.condor
+VARS DELIVER jobname="$(JOB)"
+
+PARENT FETCH CHILD ENVOY
+PARENT ENVOY CHILD DELIVER
+
+"""
+
+fetch_content = """executable = restore_cms_fetch.sh
+arguments = {{product_locator}} {{cal_locator}}
+
+output = restore_fetch.out
+error = restore_fetch.err
+log = condor.log
+
+SBIN_PATH = /lustre/aoc/cluster/pipeline/$ENV(CAPO_PROFILE)/workspaces/sbin
+should_transfer_files = yes
+transfer_input_files = $(SBIN_PATH)/pycapo, $(SBIN_PATH)/framework.sh, $(SBIN_PATH)/productfetcher
+transfer_output_files = working, rawdata, products
+
+getenv = True
+environment = "CAPO_PATH=/home/casa/capo"
+
+requirements = HasLustre == True
+
+queue
+
+"""
+
+envoy_content = """executable = restore_cms_envoy.sh
+arguments = metadata.json PPR.xml
+
+output = restore_envoy.out
+error = restore_envoy.err
+log = condor.log
+
+SBIN_PATH = /lustre/aoc/cluster/pipeline/$ENV(CAPO_PROFILE)/workspaces/sbin
+should_transfer_files = yes
+transfer_input_files = $(SBIN_PATH)/.matplotlib, $(SBIN_PATH)/pycapo, $(SBIN_PATH)/casa_envoy, $(SBIN_PATH)/vela, ./working, ./rawdata, ./products{{#remote}}, {{profile}}.properties{{/remote}}, ./PPR.xml, ./metadata.json{{files_to_transfer}}
+transfer_output_files = working, rawdata, products
+
+request_memory = {{ramInGb}}
+getenv = True
+
+{{^remote}}
+environment = "CAPO_PATH=/home/casa/capo"
+requirements = HasLustre == True
+{{/remote}}
+{{#remote}}
+requirements = (VLASS == True)
++partition = "VLASS"
+Rank = (TARGET.VLASS == True) + (TARGET.VLASSTEST =!= True)
+{{/remote}}
+
+queue
+
+"""
+
+deliver_content = """executable = restore_cms_deliver.sh
+arguments = {{request_id}}
+
+output = restore_deliver.out
+error = restore_deliver.err
+log = condor.log
+
+SBIN_PATH = /lustre/aoc/cluster/pipeline/$ENV(CAPO_PROFILE)/workspaces/sbin
+should_transfer_files = yes
+transfer_input_files = $(SBIN_PATH)/pycapo, $(SBIN_PATH)/deliver, ./products, ./working
+
+getenv = True
+environment = "CAPO_PATH=/home/casa/capo"
+requirements = HasLustre == True
+
+queue
+
+"""
+
+old_content = """executable = restore_cms.sh
+arguments = {{product_locator}} {{cal_locator}} {{request_id}} metadata.json PPR.xml
+
+output = restore.out
+error = restore.err
+log = condor.log
+
+SBIN_PATH = /lustre/aoc/cluster/pipeline/$ENV(CAPO_PROFILE)/workspaces/sbin
+should_transfer_files = yes
+transfer_input_files = $(SBIN_PATH)/.matplotlib, $(SBIN_PATH)/pycapo, $(SBIN_PATH)/framework.sh, $(SBIN_PATH)/productfetcher, $(SBIN_PATH)/casa_envoy, $(SBIN_PATH)/vela, $(SBIN_PATH)/deliver, ./PPR.xml, ./metadata.json{{files_to_transfer}}
+transfer_output_files = working, rawdata, products, delivery.json
+
+request_memory = {{ramInGb}}
+getenv = True
+environment = "CAPO_PATH=/home/casa/capo"
+
+requirements = HasLustre == True
+
+queue
+
+"""
+
+
+def upgrade():
+    op.execute(
+        """
+        DELETE FROM workflow_templates WHERE filename='restore_cms.condor'
+        """
+    )
+
+    op.execute(
+        f"""
+        INSERT INTO workflow_templates (filename, content, workflow_name) 
+        VALUES ('restore_cms.dag', E'{dag_content}', 'restore_cms') 
+        """
+    )
+    op.execute(
+        f"""
+        INSERT INTO workflow_templates (filename, content, workflow_name) 
+        VALUES ('restore_cms_fetch.condor', E'{fetch_content}', 'restore_cms') 
+        """
+    )
+    op.execute(
+        f"""
+        INSERT INTO workflow_templates (filename, content, workflow_name) 
+        VALUES ('restore_cms_envoy.condor', E'{envoy_content}', 'restore_cms') 
+        """
+    )
+    op.execute(
+        f"""
+        INSERT INTO workflow_templates (filename, content, workflow_name) 
+        VALUES ('restore_cms_deliver.condor', E'{deliver_content}', 'restore_cms') 
+        """
+    )
+
+
+def downgrade():
+    op.execute(
+        """
+        DELETE FROM workflow_templates 
+        WHERE filename like ('restore_cms_%.condor') 
+        AND workflow_name = 'restore_cms'
+        """
+    )
+
+    op.execute(
+        f"""
+        INSERT INTO workflow_templates (filename, content, workflow_name) 
+        VALUES ('restore_cms.condor', E'{old_content}', 'restore_cms')
+        """
+    )

schema/versions/8e38c2b4bbdd_multistage_restores_scripts.py

+"""multistage restores scripts
+
+Revision ID: 8e38c2b4bbdd
+Revises: 5661f79ddf29
+Create Date: 2021-10-11 15:56:02.052972
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = "8e38c2b4bbdd"
+down_revision = "5661f79ddf29"
+branch_labels = None
+depends_on = None
+
+fetch_content = """#!/bin/sh
+
+export HOME=$TMPDIR
+
+set -o errexit
+./framework.sh -d .
+chmod 770 .
+cd rawdata/
+../productfetcher --product-locator $1 --product-locator $2
+cd ../
+
+"""
+
+envoy_content = """#!/bin/sh
+
+export HOME=$TMPDIR
+
+set -o errexit
+
+{{#remote}}
+export CAPO_PATH=.
+export CAPO_PROFILE={{profile}}
+{{/remote}}
+./casa_envoy --restore -c $1 $2
+
+"""
+
+deliver_content = """#!/bin/sh
+
+export HOME=$TMPDIR
+
+set -o errexit
+
+cp -r ./working/{{sdmId}}.ms ./products/
+./deliver -p ./products --prefix $1
+
+"""
+
+old_content = """#!/bin/sh
+
+export HOME=$TMPDIR
+
+set -o errexit
+./framework.sh -d .
+chmod 770 .
+cd rawdata/
+../productfetcher --product-locator $1 --product-locator $2
+cd ../
+./casa_envoy --restore -c $4 $5
+cp -r ./working/{{sdmId}}.ms ./products/
+./deliver -p ./products --prefix $3
+"""
+
+
+def upgrade():
+    op.execute(
+        """
+        DELETE FROM workflow_templates WHERE filename='restore_cms.sh'
+        """
+    )
+    op.execute(
+        f"""
+        INSERT INTO workflow_templates (filename, content, workflow_name) 
+        VALUES ('restore_cms_fetch.sh', E'{fetch_content}', 'restore_cms') 
+        """
+    )
+    op.execute(
+        f"""
+        INSERT INTO workflow_templates (filename, content, workflow_name) 
+        VALUES ('restore_cms_envoy.sh', E'{envoy_content}', 'restore_cms') 
+        """
+    )
+    op.execute(
+        f"""
+        INSERT INTO workflow_templates (filename, content, workflow_name) 
+        VALUES ('restore_cms_deliver.sh', E'{deliver_content}', 'restore_cms') 
+        """
+    )
+
+
+def downgrade():
+    op.execute(
+        """
+        DELETE FROM workflow_templates 
+        WHERE filename like ('restore_cms_%.sh') 
+        AND workflow_name = 'restore_cms'
+        """
+    )
+
+    op.execute(
+        f"""
+        INSERT INTO workflow_templates (filename, content, workflow_name) 
+        VALUES ('restore_cms.sh', E'{old_content}', 'restore_cms')
+        """
+    )