Add capability of copying files from lustre to s3 bucket on AWS

ci/cases/pr/C48_ATM.yaml

@@ -8,6 +8,7 @@ arguments:
   resdetatmos: 48
   comroot: {{ 'RUNTESTS' | getenv }}/COMROOT
   expdir: {{ 'RUNTESTS' | getenv }}/EXPDIR
+  bucketroot: /bucket/pr-gen-results
   idate: 2021032312
   edate: 2021032312
   yaml: {{ HOMEgfs }}/ci/cases/yamls/gfs_defaults_ci.yaml

ci/cases/pr/C48_S2SW.yaml

@@ -9,6 +9,7 @@ arguments:
   resdetocean: 5.0
   comroot: {{ 'RUNTESTS' | getenv }}/COMROOT
   expdir: {{ 'RUNTESTS' | getenv }}/EXPDIR
+  bucketroot: /bucket/pr-gen-results
   idate: 2021032312
   edate: 2021032312
   yaml: {{ HOMEgfs }}/ci/cases/yamls/gfs_defaults_ci.yaml

ci/cases/pr/C48_S2SWA_gefs.yaml

@@ -13,6 +13,7 @@ arguments:
   start: cold
   comroot: {{ 'RUNTESTS' | getenv }}/COMROOT
   expdir: {{ 'RUNTESTS' | getenv }}/EXPDIR
+  bucketroot: /bucket/Wei.Huang/gefsc48
   idate: 2021032312
   edate: 2021032312
   yaml: {{ HOMEgfs }}/ci/cases/yamls/gefs_ci_defaults.yaml

jobs/JGLOBAL_ATMOS_PRODUCTS

@@ -15,6 +15,12 @@ for grid in '0p25' '0p50' '1p00'; do
   prod_dir="COM_ATMOS_GRIB_${grid}"
   GRID=${grid} YMD=${PDY} HH=${cyc} declare_from_tmpl -rx "${prod_dir}:COM_ATMOS_GRIB_GRID_TMPL"
   if [[ ! -d "${!prod_dir}" ]]; then mkdir -m 775 -p "${!prod_dir}"; fi
+
+  bueckt_dir="BUCKET_ATMOS_GRIB_${grid}"
+  GRID=${grid} YMD=${PDY} HH=${cyc} declare_from_tmpl -rx "${bueckt_dir}:BUCKET_ATMOS_GRIB_GRID_TMPL"
+  if [[ ! -d "${!bueckt_dir}" ]]; then mkdir -m 775 -p "${!bueckt_dir}"; fi
+
+  echo "!bueckt_dir is set: ${!bueckt_dir}"
 done
 
 # Variables used in this job

jobs/JGLOBAL_OCEANICE_PRODUCTS

@@ -13,6 +13,9 @@ YMD="${PDY}" HH="${cyc}" declare_from_tmpl -rx "COM_${COMPONENT^^}_HISTORY"
 YMD="${PDY}" HH="${cyc}" declare_from_tmpl -rx "COM_${COMPONENT^^}_GRIB"
 YMD="${PDY}" HH="${cyc}" declare_from_tmpl -rx "COM_${COMPONENT^^}_NETCDF"
 
+YMD="${PDY}" HH="${cyc}" declare_from_tmpl -rx "BUCKET_${COMPONENT^^}_GRIB"
+YMD="${PDY}" HH="${cyc}" declare_from_tmpl -rx "BUCKET_${COMPONENT^^}_NETCDF"
+
 ###############################################################
 # Run exglobal script
 "${SCRgfs}/exglobal_oceanice_products.py"

parm/config/gefs/config.base

@@ -37,6 +37,7 @@ export FIXugwd=${FIXgfs}/ugwd
 # GLOBAL static environment parameters
 export PACKAGEROOT="@PACKAGEROOT@"    # TODO: set via prod_envir in Ops
 export COMROOT="@COMROOT@"    # TODO: set via prod_envir in Ops
+export BUCKETROOT="@BUCKETROOT@"    # TODO: set via prod_envir in Ops
 export COMINsyn="@COMINsyn@"
 
 # USER specific paths
@@ -95,6 +96,7 @@ export assim_freq=6
 export PSLOT="@PSLOT@"
 export EXPDIR="@EXPDIR@/${PSLOT}"
 export ROTDIR="@COMROOT@/${PSLOT}"
+export BUCKETDIR="@BUCKETROOT@/${PSLOT}"
 
 export ARCDIR="${NOSCRUB}/archive/${PSLOT}"
 export ATARDIR="@ATARDIR@"
@@ -345,9 +347,12 @@ export DELETE_COM_IN_ARCHIVE_JOB="YES"   # NO=retain ROTDIR.  YES default in arc
 # Number of regional collectives to create soundings for
 export NUM_SND_COLLECTIVES=${NUM_SND_COLLECTIVES:-9}
 
-# The tracker, genesis, and METplus jobs are not supported on AWS yet
-# TODO: we should place these in workflow/hosts/awspw.yaml as part of AWS setup, not for general.
-if [[ "${machine}" == "AWSPW" ]]; then
+# The tracker, genesis, and METplus jobs are not supported on CSPs yet
+# TODO: we should place these in workflow/hosts/[csp]pw.yaml as part of AWS/AZURE/GOOGLE setup, not for general.
+if [[ "${machine}" =~ "PW" ]]; then
+   export DO_TRACKER="NO"
+   export DO_GENESIS="NO"
+   export DO_METP="NO"
    export DO_WAVE="NO"
 fi
 

parm/config/gfs/config.base

@@ -45,6 +45,7 @@ export FIXgdas="${FIXgfs}/gdas"
 # GLOBAL static environment parameters
 export PACKAGEROOT="@PACKAGEROOT@"    # TODO: set via prod_envir in Ops
 export COMROOT="@COMROOT@"    # TODO: set via prod_envir in Ops
+export BUCKETROOT="@BUCKETROOT@"    # TODO: set via prod_envir in Ops
 export COMINsyn="@COMINsyn@"
 export DMPDIR="@DMPDIR@"
 
@@ -121,6 +122,7 @@ export assim_freq=6
 export PSLOT="@PSLOT@"
 export EXPDIR="@EXPDIR@/${PSLOT}"
 export ROTDIR="@COMROOT@/${PSLOT}"
+export BUCKETDIR="@BUCKETROOT@/${PSLOT}"
 export ROTDIR_DUMP="YES"                #Note: A value of "NO" does not currently work
 export DUMP_SUFFIX=""
 if [[ "${PDY}${cyc}" -ge "2019092100" && "${PDY}${cyc}" -le "2019110700" ]]; then

parm/config/gfs/config.com

@@ -100,3 +100,26 @@ declare -rx COM_CHEM_HISTORY_TMPL=${COM_BASE}'/model/chem/history'
 declare -rx COM_CHEM_ANALYSIS_TMPL=${COM_BASE}'/analysis/chem'
 
 declare -rx COM_MED_RESTART_TMPL=${COM_BASE}'/model/med/restart'
+
+BUCKET_BASE='${BUCKETDIR}/${RUN}.${YMD}/${HH}/${MEMDIR}'
+
+declare -rx BUCKET_TOP_TMPL='${BUCKETDIR}/${RUN}.${YMD}/${HH}'
+
+declare -rx BUCKET_CONF_TMPL=${BUCKET_BASE}'/conf'
+declare -rx BUCKET_OBS_JEDI=${BUCKET_BASE}'/obs_jedi'
+
+declare -rx BUCKET_ATMOS_INPUT_TMPL=${BUCKET_BASE}'/model/atmos/input'
+declare -rx BUCKET_ATMOS_RESTART_TMPL=${BUCKET_BASE}'/model/atmos/restart'
+declare -rx BUCKET_ATMOS_HISTORY_TMPL=${BUCKET_BASE}'/model/atmos/history'
+declare -rx BUCKET_ATMOS_MASTER_TMPL=${BUCKET_BASE}'/model/atmos/master'
+declare -rx BUCKET_ATMOS_GRIB_TMPL=${BUCKET_BASE}'/products/atmos/grib2'
+declare -rx BUCKET_ATMOS_GRIB_GRID_TMPL=${BUCKET_ATMOS_GRIB_TMPL}'/${GRID}'
+
+declare -rx BUCKET_OCEAN_NETCDF_TMPL=${BUCKET_BASE}'/products/ocean/netcdf'
+declare -rx BUCKET_OCEAN_GRIB_TMPL=${BUCKET_BASE}'/products/ocean/grib2'
+declare -rx BUCKET_OCEAN_GRIB_GRID_TMPL=${BUCKET_OCEAN_GRIB_TMPL}'/${GRID}'
+
+declare -rx BUCKET_ICE_NETCDF_TMPL=${BUCKET_BASE}'/products/ice/netcdf'
+declare -rx BUCKET_ICE_GRIB_TMPL=${BUCKET_BASE}'/products/ice/grib2'
+declare -rx BUCKET_ICE_GRIB_GRID_TMPL=${BUCKET_ICE_GRIB_TMPL}'/${GRID}'
+

parm/post/oceanice_products.yaml

@@ -40,8 +40,10 @@ ocean:
   data_out:
     mkdir:
       - "{{ COM_OCEAN_NETCDF }}"
+      - "{{ BUCKET_OCEAN_NETCDF }}"
       {% for grid in product_grids %}
       - "{{ COM_OCEAN_GRIB }}/{{ grid }}"
+      - "{{ BUCKET_OCEAN_GRIB }}/{{ grid }}"
       {% endfor %}
     copy:
       - ["{{ DATA }}/ocean_subset.nc", "{{ COM_OCEAN_NETCDF }}/{{ RUN }}.ocean.t{{ current_cycle | strftime('%H') }}z.native.f{{ '%03d' % forecast_hour }}.nc"]
@@ -50,6 +52,12 @@ ocean:
       - ["{{ DATA }}/ocean.{{ grid }}.grib2.idx", "{{ COM_OCEAN_GRIB }}/{{ grid }}/{{ RUN }}.ocean.t{{ current_cycle | strftime('%H') }}z.{{ grid }}.f{{ '%03d' % forecast_hour }}.grib2.idx"]
       {% endfor %}
 
+      - ["{{ DATA }}/ocean_subset.nc", "{{ BUCKET_OCEAN_NETCDF }}/{{ RUN }}.ocean.t{{ current_cycle | strftime('%H') }}z.native.f{{ '%03d' % forecast_hour }}.nc"]
+      {% for grid in product_grids %}
+      - ["{{ DATA }}/ocean.{{ grid }}.grib2", "{{ BUCKET_OCEAN_GRIB }}/{{ grid }}/{{ RUN }}.ocean.t{{ current_cycle | strftime('%H') }}z.{{ grid }}.f{{ '%03d' % forecast_hour }}.grib2"]
+      - ["{{ DATA }}/ocean.{{ grid }}.grib2.idx", "{{ BUCKET_OCEAN_GRIB }}/{{ grid }}/{{ RUN }}.ocean.t{{ current_cycle | strftime('%H') }}z.{{ grid }}.f{{ '%03d' % forecast_hour }}.grib2.idx"]
+      {% endfor %}
+
 ice:
   namelist:
     ftype: "ice"
@@ -64,12 +72,20 @@ ice:
   data_out:
     mkdir:
       - "{{ COM_ICE_NETCDF }}"
+      - "{{ BUCKET_ICE_NETCDF }}"
       {% for grid in product_grids %}
       - "{{ COM_ICE_GRIB }}/{{ grid }}"
+      - "{{ BUCKET_ICE_GRIB }}/{{ grid }}"
       {% endfor %}
     copy:
       - ["{{ DATA }}/ice_subset.nc", "{{ COM_ICE_NETCDF }}/{{ RUN }}.ice.t{{ current_cycle | strftime('%H') }}z.native.f{{ '%03d' % forecast_hour }}.nc"]
       {% for grid in product_grids %}
       - ["{{ DATA }}/ice.{{ grid }}.grib2", "{{ COM_ICE_GRIB }}/{{ grid }}/{{ RUN }}.ice.t{{ current_cycle | strftime('%H') }}z.{{ grid }}.f{{ '%03d' % forecast_hour }}.grib2"]
       - ["{{ DATA }}/ice.{{ grid }}.grib2.idx", "{{ COM_ICE_GRIB }}/{{ grid }}/{{ RUN }}.ice.t{{ current_cycle | strftime('%H') }}z.{{ grid }}.f{{ '%03d' % forecast_hour }}.grib2.idx"]
       {% endfor %}
+
+      - ["{{ DATA }}/ice_subset.nc", "{{ BUCKET_ICE_NETCDF }}/{{ RUN }}.ice.t{{ current_cycle | strftime('%H') }}z.native.f{{ '%03d' % forecast_hour }}.nc"]
+      {% for grid in product_grids %}
+      - ["{{ DATA }}/ice.{{ grid }}.grib2", "{{ BUCKET_ICE_GRIB }}/{{ grid }}/{{ RUN }}.ice.t{{ current_cycle | strftime('%H') }}z.{{ grid }}.f{{ '%03d' % forecast_hour }}.grib2"]
+      - ["{{ DATA }}/ice.{{ grid }}.grib2.idx", "{{ BUCKET_ICE_GRIB }}/{{ grid }}/{{ RUN }}.ice.t{{ current_cycle | strftime('%H') }}z.{{ grid }}.f{{ '%03d' % forecast_hour }}.grib2.idx"]
+      {% endfor %}

scripts/exglobal_atmos_products.sh

@@ -163,6 +163,18 @@ for (( nset=1 ; nset <= downset ; nset++ )); do
     prod_dir="COM_ATMOS_GRIB_${grid}"
     ${NCP} "pgb2${grp}file_${fhr3}_${grid}" "${!prod_dir}/${PREFIX}pgrb2${grp}.${grid}.${fhr3}"
     ${WGRIB2} -s "pgb2${grp}file_${fhr3}_${grid}" > "${!prod_dir}/${PREFIX}pgrb2${grp}.${grid}.${fhr3}.idx"
+
+    echo "!prod_dir: ${!prod_dir}"
+    bucket_dir="BUCKET_ATMOS_GRIB_${grid}"
+    if [[ -z ${!bucket_dir+x} ]]; then
+      echo "!bucket_dir is unset"
+    else
+      echo "!bucket_dir is set to: ${!bucket_dir}"
+      if [[ "${!prod_dir}" != "${!bucket_dir}" ]]; then
+        cp "pgb2${grp}file_${fhr3}_${grid}" "${!bucket_dir}/${PREFIX}pgrb2${grp}.${grid}.${fhr3}"
+        cp "${!prod_dir}/${PREFIX}pgrb2${grp}.${grid}.${fhr3}.idx" "${!bucket_dir}/${PREFIX}pgrb2${grp}.${grid}.${fhr3}.idx"
+      fi
+    fi
   done
 
   echo "Finished processing nset = ${nset}"
@@ -195,6 +207,18 @@ if [[ "${FLXGF:-}" == "YES" ]]; then
     prod_dir="COM_ATMOS_GRIB_${grid}"
     ${NCP} "sflux_${fhr3}_${grid}" "${!prod_dir}/${PREFIX}flux.${grid}.${fhr3}"
     ${WGRIB2} -s "sflux_${fhr3}_${grid}" > "${!prod_dir}/${PREFIX}flux.${grid}.${fhr3}.idx"
+
+    echo "!prod_dir: ${!prod_dir}"
+    bucket_dir="BUCKET_ATMOS_GRIB_${grid}"
+    if [[ -z ${!bucket_dir+x} ]]; then
+      echo "!bucket_dir is unset"
+    else
+      echo "!bucket_dir is set to: ${!bucket_dir}"
+      if [[ "${!prod_dir}" != "${!bucket_dir}" ]]; then
+        cp "sflux_${fhr3}_${grid}" "${!bucket_dir}/${PREFIX}flux.${grid}.${fhr3}"
+        cp "${!prod_dir}/${PREFIX}flux.${grid}.${fhr3}.idx" "${!bucket_dir}/${PREFIX}flux.${grid}.${fhr3}.idx"
+      fi
+    fi
   done
 fi
 

workflow/create_experiment.py

@@ -17,7 +17,7 @@
         Name of the experiment
 
     RUNTESTS
-        Root directory where the test EXPDIR and COMROOT will be placed
+        Root directory where the test EXPDIR and COMROOT/BUCKETROOT will be placed
 
 Output
 ------

workflow/rocoto/tasks.py

@@ -52,6 +52,8 @@ def __init__(self, app_config: AppConfig, run: str) -> None:
 
         self.HOMEgfs = self._base['HOMEgfs']
         self.rotdir = self._base['ROTDIR']
+        self._base['BUCKETDIR'] = '/lustre/Wei.Huang/run/GEFSTESTS/COMROOT/c48gefs'
+        self.bucketdir = self._base['BUCKETDIR']
         self.pslot = self._base['PSLOT']
         if self.run == "enkfgfs":
             self.nmem = int(self._base['NMEM_ENS_GFS'])
@@ -66,6 +68,7 @@ def __init__(self, app_config: AppConfig, run: str) -> None:
         envar_dict = {'RUN_ENVIR': self._base.get('RUN_ENVIR', 'emc'),
                       'HOMEgfs': self.HOMEgfs,
                       'EXPDIR': self._base.get('EXPDIR'),
+                      'BUCKETDIR': self._base.get('BUCKETDIR'),
                       'NET': self._base.get('NET'),
                       'RUN': self.run,
                       'CDATE': '<cyclestr>@Y@m@d@H</cyclestr>',

workflow/rocoto_viewer.py

@@ -1328,6 +1328,10 @@ def main(screen):
         EXPDIR = entity_values['EXPDIR']
     else:
         EXPDIR = '.'
+    if 'BUCKETDIR' in entity_values:
+        BUCKETDIR = entity_values['BUCKETDIR']
+    else:
+        BUCKETDIR = 'no_bucketdir'
 
     if html_output:
         html_ptr = None

workflow/setup_expt.py

@@ -114,6 +114,7 @@ def edit_baseconfig(host, inputs, yaml_dict):
         "@OCNRES@": f"{int(100.*inputs.resdetocean):03d}",
         "@EXPDIR@": inputs.expdir,
         "@COMROOT@": inputs.comroot,
+        "@BUCKETROOT@": inputs.bucketroot,
         "@EXP_WARM_START@": is_warm_start,
         "@MODE@": inputs.mode,
         "@gfs_cyc@": inputs.gfs_cyc,
@@ -197,6 +198,8 @@ def _common_args(parser):
                             type=str, required=False, default=os.getenv('HOME'))
         parser.add_argument('--expdir', help='full path to EXPDIR',
                             type=str, required=False, default=os.getenv('HOME'))
+        parser.add_argument('--bucketroot', help='full path to BUCKETROOT',
+                            type=str, required=False, default=os.getenv('HOME'))
         parser.add_argument('--idate', help='starting date of experiment, initial conditions must exist!',
                             required=True, type=lambda dd: to_datetime(dd))
         parser.add_argument('--edate', help='end date experiment', required=True, type=lambda dd: to_datetime(dd))
@@ -347,6 +350,9 @@ def main(*argv):
     user_inputs = input_args(*argv)
     host = Host()
 
+    print('user_inputs = ', user_inputs)
+    print('host = ', host)
+
     validate_user_request(host, user_inputs)
 
     # Determine ocean resolution if not provided
@@ -355,13 +361,18 @@ def main(*argv):
 
     rotdir = os.path.join(user_inputs.comroot, user_inputs.pslot)
     expdir = os.path.join(user_inputs.expdir, user_inputs.pslot)
+    bucketdir = os.path.join(user_inputs.bucketroot, user_inputs.pslot)
 
     create_rotdir = query_and_clean(rotdir, force_clean=user_inputs.overwrite)
     create_expdir = query_and_clean(expdir, force_clean=user_inputs.overwrite)
+    create_bucketdir = query_and_clean(bucketdir, force_clean=user_inputs.overwrite)
 
     if create_rotdir:
         makedirs_if_missing(rotdir)
 
+    if create_bucketdir:
+        makedirs_if_missing(bucketdir)
+
     if create_expdir:
         makedirs_if_missing(expdir)
         fill_EXPDIR(user_inputs)
@@ -370,6 +381,7 @@ def main(*argv):
     print(f"*" * 100)
     print(f'EXPDIR: {expdir}')
     print(f'ROTDIR: {rotdir}')
+    print(f'BUCKETDIR: {bucketdir}')
     print(f"*" * 100)