Add Email Notification System for scrontab-Launched Rocoto Workflows

dev/workflow/rocoto/rocoto_scron_template.sh

@@ -0,0 +1,79 @@
+#!/usr/bin/env bash
+source "{HOMEgfs}/dev/ush/gw_setup.sh"
+# Set platform variable if hostname contains 'gaea'
+if [[ "$(hostname -s)" == *gaea* ]]; then
+    platform="gaea"
+else
+    platform="unknown"
+fi
+# Run rocotorun
+# Template variable {{rocotorunstr}} will be substituted by the Python template system
+bash -c "{rocotorunstr}"
+
+# Monitor for failed jobs using rocotostat
+LOCKFILE="{expdir}/.failed_jobs.lock"
+ROCOTOSTAT=$(command -v rocotostat)
+
+if [[ -n "${{ROCOTOSTAT}}" ]]; then
+    FAILED_JOBS=$(${{ROCOTOSTAT}} -d "{expdir}/{pslot}.db" -w "{expdir}/{pslot}.xml" -c all 2> /dev/null | grep -E 'DEAD')
+
+    if [[ -n "${{FAILED_JOBS}}" ]]; then
+        # Read previously reported failures
+        PREV_FAILED=""
+        if [[ -f "${{LOCKFILE}}" ]]; then
+            PREV_FAILED=$(cat "${{LOCKFILE}}")
+        fi
+
+        # Check for NEW failures only (not just changes)
+        NEW_FAILURES=""
+        while IFS= read -r job; do
+            if [[ -n "${{job}}" ]] && ! echo "${{PREV_FAILED}}" | grep -qF "${{job}}"; then
+                NEW_FAILURES="${{NEW_FAILURES}}${{job}}"$'\n'
+            fi
+        done <<< "${{FAILED_JOBS}}"
+
+        # Send email only if there are NEW failures
+        if [[ -n "${{NEW_FAILURES}}" ]]; then
+            MSGFILE="/tmp/rocoto_fail_msg_$$.txt"
+            {{
+                echo "The following jobs have failed in experiment {pslot} on ${{platform}}:"
+                echo ""
+
+                # Format each failed job with detailed information
+                while IFS= read -r line; do
+                    if [[ -n "${{line}}" ]]; then
+                        # Parse rocotostat output: Cycle Task JobID State Try MaxTries Duration
+                        read -r cycle task jobid state try maxtries duration <<< "${{line}}"
+                        # Extract YYYYMMDDHH from cycle (first 10 characters)
+                        cycle_short=${{cycle:0:10}}
+                        # Get current timestamp
+                        timestamp=$(date -u '+%m/%d/%y %H:%M:%S UTC')
+
+                        # Format similar to user's example
+                        echo "${{timestamp}} :: {pslot}.xml :: Cycle ${{cycle}}, Task ${{task}}, jobid=${{jobid}}, in state ${{state}}, ran for ${{duration}} seconds, try=${{try}} (of ${{maxtries}})"
+                        echo "Check log: {comroot}/{pslot}/logs/${{cycle_short}}/${{task}}.log"
+                        echo ""
+                    fi
+                done <<< "${{NEW_FAILURES}}"
+            }} > "${{MSGFILE}}"
+
+            # Try to send email
+            EMAIL="{replyto}"
+            hostname_domain=$(hostname -d)
+            FROM_EMAIL="no-reply@${{hostname_domain}}"
+            if [[ -n "${{EMAIL}}" ]] && command -v mail &> /dev/null; then
+                # On Gaea, the mail utility requires the -v (verbose) flag to ensure delivery.
+                # To avoid receiving verbose output as an actual email, a spoofed 'from' address is used for notifications.
+                mail -r "${{FROM_EMAIL}}" -v -s "[{pslot}] Workflow Job Failures Detected" "${{EMAIL}}" < "${{MSGFILE}}" 2>&1
+            fi
+
+            rm -f "${{MSGFILE}}"
+        fi
+
+        # Always update lockfile to reflect current failures
+        echo "${{FAILED_JOBS}}" > "${{LOCKFILE}}"
+    else
+        # No failures, remove lockfile if it exists
+        [[ -f "${{LOCKFILE}}" ]] && rm -f "${{LOCKFILE}}"
+    fi
+fi

dev/workflow/rocoto/rocoto_xml.py

@@ -123,6 +123,37 @@ def write(self, xml_file: str = None, crontab_file: str = None):
         if self._base["DO_ARCHCOM"] and self._base["ARCHCOM_TO"] == "globus_hpss":
             self._write_server_crontab()
 
+    def _get_scron_script_content(self, rocotorunstr: str, replyto: str) -> str:
+        """
+        Load and format the cron script template with experiment-specific values.
+
+        Parameters
+        ----------
+        rocotorunstr : str
+            The rocotorun command string
+        replyto : str
+            Email address for notifications
+
+        Returns
+        -------
+        str
+            Formatted bash script content
+        """
+        template_path = os.path.join(os.path.dirname(__file__), 'rocoto_scron_template.sh')
+
+        with open(template_path, 'r') as fh:
+            template_content = fh.read()
+
+        # Format the template with experiment-specific values
+        return template_content.format(
+            HOMEgfs=self.HOMEgfs,
+            rocotorunstr=rocotorunstr,
+            expdir=self.expdir,
+            pslot=self.pslot,
+            replyto=replyto,
+            comroot=self._base.get('COMROOT')
+        )
+
     def _write_xml(self, xml_file: str = None) -> None:
 
         if xml_file is None:
@@ -148,6 +179,10 @@ def _write_crontab(self, crontab_file: str = None, cronint: int = 5) -> None:
         cronintstr = f'*/{cronint} * * * *'
 
         replyto = os.environ.get('REPLYTO', "")
+        # Use fallback email if REPLYTO is not set
+        if not replyto:
+            user = os.environ.get('USER', 'user')
+            replyto = f"{user}@noaa.gov"
 
         crontab_strings = [
             '',
@@ -169,28 +204,25 @@ def _write_crontab(self, crontab_file: str = None, cronint: int = 5) -> None:
             crontab_strings.extend([
                 f'#SCRON --partition={partition}',
                 f'#SCRON --account={account}',
-                f'#SCRON --mail-user={replyto}',
                 f'#SCRON --job-name={self.pslot}_scron',
                 f'#SCRON --output={self.expdir}/logs/scron.log',
-                '#SCRON --time=00:10:00',
-                '#SCRON --dependency=singleton'
+                f'#SCRON --time=00:10:00',
+                f'#SCRON --dependency=singleton'
             ])
 
-            # Now write the script that actually runs rocotorun
+            # Now write the script that actually runs rocotorun and monitors for failures
             cron_cmd = f"{self.expdir}/{self.pslot}.scron.sh"
+            script_content = "#!/usr/bin/env bash\nset -x\n" + self._get_scron_script_content(rocotorunstr, replyto)
+
             with open(cron_cmd, "w") as script_fh:
-                script_fh.write(
-                    "#!/usr/bin/env bash\n" +
-                    "set -x\n" +
-                    f"source {self.HOMEgfs}/dev/ush/gw_setup.sh" + "\n" +
-                    rocotorunstr + "\n"
-                )
+                script_fh.write(script_content)
 
             # Make the script executable
             mode = os.stat(cron_cmd)
             os.chmod(cron_cmd, mode.st_mode | stat.S_IEXEC)
         else:
-            cron_cmd = rocotorunstr
+            # For regular crontab, create a wrapper script with monitoring
+            cron_cmd = f"{self.expdir}/{self.pslot}.cron.sh"
             crontab_strings.extend([
                 'SHELL="/bin/bash"',
                 f'MAILTO="{replyto}"'