Repository: desihub/desitransfer

Issue:
Guard against truncated status file
At 2020-01-09 01:00 PST, the desi_transfer_status.json file was truncated, causing the transfer daemon to crash when it subsequently tried to read the empty file.

Workaround for quick restart: Add `[]` (an array) to the empty file.

dtn04.nersc.gov was rebooted sometime after the file was truncated (approximately 02:30 PST), suggesting transient problems with the filesystem.

Discussion:

Patch:
diff --git a/bin/desi_transfer_status_restore b/bin/desi_transfer_status_restore
new file mode 100755
index 0000000..4766fc8
--- /dev/null
+++ b/bin/desi_transfer_status_restore
@@ -0,0 +1,71 @@
+#!/usr/bin/env python
+"""
+Quick and dirty script to restore raw data transfer status.
+
+1. Obtain rsync time from modification time of exposure directory.
+2. Set checksum time to rsync time.
+3. Ignore pipeline time (as of early 2020).
+4. Obtain backup time from HPSS backup files.
+
+This script is deliberately kept separate from the package because it
+uses hpsspy.
+"""
+from sys import exit
+import json
+import os
+import hpsspy.os as hpos
+
+
+def backup_times(path='desi/spectro/data'):
+    """Obtain backup times from HPSS files.
+
+    Parameters
+    ----------
+    path : :class:`str`
+        The HPSS path to the raw data backup files.
+
+    Returns
+    -------
+    :class:`dict`
+        A mapping of night to backup time.  The backup time is in milliseconds
+        for compatibility with JavaScript.
+    """
+    ls = hpos.listdir(path)
+    return dict([(os.path.splitext(f.name)[0].split('_')[-1], f.st_mtime*1000)
+                 for f in ls if f.name.endswith('.tar')])
+
+
+def main():
+    """Entry point for :command:`desi_transfer_status_restore`.
+
+    Returns
+    -------
+    :class:`int`
+        An integer suitable for passing to :func:`sys.exit`.
+    """
+    bt = backup_times()
+    nights = os.listdir(os.environ['DESI_SPECTRO_DATA'])
+    status = list()
+    for night in nights:
+        if night != 'README.html':
+            exposures = os.listdir(os.path.join(os.environ['DESI_SPECTRO_DATA'], night))
+            for exp in exposures:
+                rt = int(os.stat(os.path.join(os.environ['DESI_SPECTRO_DATA'], night, exp)).st_mtime * 1000)
+                status.append([int(night), int(exp), 'rsync', True, '', rt])
+                status.append([int(night), int(exp), 'checksum', True, '', rt])
+                try:
+                    status.append([int(night), int(exp), 'backup', True, '', bt[night]])
+                except KeyError:
+                    pass
+    status = sorted(status, key=lambda x: x[0]*10000000 + x[1], reverse=True)
+    with open('desi_transfer_status_restore.json', 'w') as j:
+        json.dump(status, j, indent=None, separators=(',', ':'))
+    return 0
+
+
+if __name__ == '__main__':
+    try:
+        foo = os.environ['HPSS_DIR']
+    except KeyError:
+        os.environ['HPSS_DIR'] = '/usr/common/mss'
+    exit(main())
diff --git a/doc/api.rst b/doc/api.rst
index 418593d..9ee155f 100644
--- a/doc/api.rst
+++ b/doc/api.rst
@@ -5,6 +5,9 @@ desitransfer API
 .. automodule:: desitransfer
     :members:
 
+.. automodule:: desitransfer.common
+    :members:
+
 .. automodule:: desitransfer.daemon
     :members:
 
diff --git a/doc/changes.rst b/doc/changes.rst
index c55f0ac..44903e1 100644
--- a/doc/changes.rst
+++ b/doc/changes.rst
@@ -5,7 +5,10 @@ Change Log
 0.3.4 (unreleased)
 ------------------
 
-* No changes yet.
+* Guard against corrupted status JSON files; restore transfer status;
+  additional daily transfers (PR `#10`_).
+
+.. _`#10`: https://github.com/desihub/desitransfer/pull/10
 
 0.3.3 (2019-12-18)
 ------------------
diff --git a/py/desitransfer/daily.py b/py/desitransfer/daily.py
index 94b63d5..d27b1f7 100644
--- a/py/desitransfer/daily.py
+++ b/py/desitransfer/daily.py
@@ -98,6 +98,8 @@ def _config():
             #                os.path.join(engineering, 'fxc')),
             DailyDirectory('/data/focalplane/logs/calib_logs',
                            os.path.join(engineering, 'focalplane', 'logs', 'calib_logs')),
+            DailyDirectory('/data/focalplane/logs/kpno',
+                           os.path.join(engineering, 'focalplane', 'logs', 'kpno')),
             DailyDirectory('/data/focalplane/logs/xytest_data',
                            os.path.join(engineering, 'focalplane', 'logs', 'xytest_data')),
             DailyDirectory('/data/fvc/data',
diff --git a/py/desitransfer/status.py b/py/desitransfer/status.py
index 21cb7a9..e92998f 100644
--- a/py/desitransfer/status.py
+++ b/py/desitransfer/status.py
@@ -14,10 +14,6 @@ import time
 from argparse import ArgumentParser
 from pkg_resources import resource_filename
 from . import __version__ as dtVersion
-# from desiutil.log import get_logger
-
-
-# log = None
 
 
 class TransferStatus(object):
@@ -47,11 +43,44 @@ class TransferStatus(object):
             return
         try:
             with open(self.json) as j:
-                self.status = json.load(j)
+                try:
+                    self.status = json.load(j)
+                except json.JSONDecodeError:
+                    self._handle_malformed()
         except FileNotFoundError:
             pass
         return
 
+    def _handle_malformed(self):
+        """Handle malformed JSON files.
+
+        This function will save the malformed file to a .bad file for
+        later analysis, and write an empty array to a new status file.
+        """
+        from .daemon import log
+        bad = self.json + '.bad'
+        m = "Malformed JSON file detected: %s; saving original file as %s."
+        try:
+            log.error(m, self.json, bad)
+        except AttributeError:
+            # If the status code is running stand-alone, the log object
+            # will be None.
+            print("ERROR: " + (m % (self.json, bad)))
+        m = "shutil.copy2('%s', '%s')"
+        try:
+            log.debug(m, self.json, bad)
+        except AttributeError:
+            print("DEBUG: " + (m % (self.json, bad)))
+        shutil.copy2(self.json, bad)
+        m = "Writing empty array to %s."
+        try:
+            log.info(m, self.json)
+        except AttributeError:
+            print("INFO: " + (m % (self.json,)))
+        with open(self.json, 'w') as j:
+            j.write('[]')
+        return
+
     def update(self, night, exposure, stage, failure=False, last=''):
         """Update the transfer status.
 
@@ -92,6 +121,14 @@ class TransferStatus(object):
             self.status.insert(0, row)
         self.status = sorted(self.status, key=lambda x: x[0]*10000000 + x[1],
                              reverse=True)
+        #
+        # Copy the original file before modifying.
+        # This will overwrite any existing .bak file
+        #
+        try:
+            shutil.copy2(self.json, self.json + '.bak')
+        except FileNotFoundError:
+            pass
         with open(self.json, 'w') as j:
             json.dump(self.status, j, indent=None, separators=(',', ':'))
 
diff --git a/requirements.txt b/requirements.txt
index 345f714..98fc19f 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,3 +1,3 @@
 setuptools
 pytz
-git+https://github.com/desihub/desiutil.git@1.9.15#egg=desiutil
+git+https://github.com/desihub/desiutil.git@2.0.1#egg=desiutil


Test patch:
diff --git a/py/desitransfer/test/t/bad.json b/py/desitransfer/test/t/bad.json
new file mode 100644
index 0000000..0f25ee4
--- /dev/null
+++ b/py/desitransfer/test/t/bad.json
@@ -0,0 +1,1 @@
+This is a bad JSON file!
diff --git a/py/desitransfer/test/test_daemon.py b/py/desitransfer/test/test_daemon.py
index f6247df..ab51215 100644
--- a/py/desitransfer/test/test_daemon.py
+++ b/py/desitransfer/test/test_daemon.py
@@ -135,10 +135,10 @@ class TestDaemon(unittest.TestCase):
     @patch('desitransfer.daemon.SMTPHandler')
     @patch('desitransfer.daemon.RotatingFileHandler')
     @patch('desitransfer.daemon.get_logger')
-    def test_TransferDaemon_configure_log(self, gl, rfh, smtp):
+    @patch('desitransfer.daemon.log')  # Needed to restore the module-level log object after test.
+    def test_TransferDaemon_configure_log(self, mock_log, gl, rfh, smtp):
         """Test logging configuration.
         """
-        ll = gl.return_value = MagicMock()
         with patch.dict('os.environ',
                         {'CSCRATCH': self.tmp.name,
                          'DESI_ROOT': '/desi/root',
@@ -149,7 +149,7 @@ class TestDaemon(unittest.TestCase):
         rfh.assert_called_once_with('/desi/root/spectro/staging/logs/desi_transfer_daemon.log',
                                     backupCount=100, maxBytes=100000000)
         gl.assert_called_once_with(timestamp=True)
-        ll.setLevel.assert_called_once_with(logging.DEBUG)
+        gl().setLevel.assert_called_once_with(logging.DEBUG)
 
     @patch.object(TransferDaemon, 'checksum_lock')
     @patch.object(TransferDaemon, 'directory')
diff --git a/py/desitransfer/test/test_daily.py b/py/desitransfer/test/test_daily.py
index 1d8c180..9089131 100644
--- a/py/desitransfer/test/test_daily.py
+++ b/py/desitransfer/test/test_daily.py
@@ -110,7 +110,7 @@ class TestDaily(unittest.TestCase):
                              call().__exit__(None, None, None)])
         mock_popen.assert_has_calls([call(),
                                      call(['fix_permissions.sh', '-a', '/dst/d0'],
-                                          stdout=mo(), stderr=-2), 
+                                          stdout=mo(), stderr=-2),
                                      call().wait()])
 
 
diff --git a/py/desitransfer/test/test_status.py b/py/desitransfer/test/test_status.py
index 0194031..23adbe1 100644
--- a/py/desitransfer/test/test_status.py
+++ b/py/desitransfer/test/test_status.py
@@ -4,9 +4,10 @@
 """
 import json
 import os
+import shutil
 import sys
 import unittest
-from unittest.mock import patch
+from unittest.mock import patch, call
 from tempfile import TemporaryDirectory
 from pkg_resources import resource_filename
 from ..status import TransferStatus, _options
@@ -85,6 +86,45 @@ class TestStatus(unittest.TestCase):
         cp.assert_called_once_with(j, d)
         cf.assert_called_once_with(h, os.path.join(d, 'index.html'))
 
+    @patch('desitransfer.daemon.log')
+    def test_TransferStatus_handle_malformed_with_log(self, mock_log):
+        """Test handling of malformed JSON files.
+        """
+        bad = resource_filename('desitransfer.test', 't/bad.json')
+        with TemporaryDirectory() as d:
+            shutil.copy(bad, os.path.join(d, 'desi_transfer_status.json'))
+            s = TransferStatus(d)
+            self.assertTrue(os.path.exists(os.path.join(d, 'desi_transfer_status.json.bad')))
+            self.assertListEqual(s.status, [])
+            self.assertListEqual(os.listdir(d), ['desi_transfer_status.json.bad',
+                                                 'desi_transfer_status.json'])
+        mock_log.error.assert_called_once_with('Malformed JSON file detected: %s; saving original file as %s.',
+                                               os.path.join(d, 'desi_transfer_status.json'),
+                                               os.path.join(d, 'desi_transfer_status.json.bad'))
+        mock_log.debug.assert_called_once_with("shutil.copy2('%s', '%s')",
+                                               os.path.join(d, 'desi_transfer_status.json'),
+                                               os.path.join(d, 'desi_transfer_status.json.bad'))
+        mock_log.info.assert_called_once_with('Writing empty array to %s.',
+                                              os.path.join(d, 'desi_transfer_status.json'))
+
+    @patch('builtins.print')
+    def test_TransferStatus_handle_malformed_without_log(self, mock_print):
+        """Test handling of malformed JSON files (no log object).
+        """
+        bad = resource_filename('desitransfer.test', 't/bad.json')
+        with TemporaryDirectory() as d:
+            shutil.copy(bad, os.path.join(d, 'desi_transfer_status.json'))
+            s = TransferStatus(d)
+            self.assertTrue(os.path.exists(os.path.join(d, 'desi_transfer_status.json.bad')))
+            self.assertListEqual(s.status, [])
+            self.assertListEqual(os.listdir(d), ['desi_transfer_status.json.bad',
+                                                 'desi_transfer_status.json'])
+        mock_print.assert_has_calls([call('ERROR: Malformed JSON file detected: %s; saving original file as %s.' % (os.path.join(d, 'desi_transfer_status.json'),
+                                                                                                                    os.path.join(d, 'desi_transfer_status.json.bad'))),
+                                     call("DEBUG: shutil.copy2('%s', '%s')" % (os.path.join(d, 'desi_transfer_status.json'),
+                                                                               os.path.join(d, 'desi_transfer_status.json.bad'))),
+                                     call("INFO: Writing empty array to %s." % (os.path.join(d, 'desi_transfer_status.json'),))])
+
     @patch('time.time')
     def test_TransferStatus_update(self, mock_time):
         """Test status reporting mechanism updates.
@@ -98,6 +138,7 @@ class TestStatus(unittest.TestCase):
                 json.dump(st, f, indent=None, separators=(',', ':'))
             s = TransferStatus(d)
             s.update('20200703', '12345678', 'checksum')
+            self.assertTrue(os.path.exists(js + '.bak'))
             self.assertEqual(s.status[0], [20200703, 12345678, 'checksum', True, '', 1565300090000])
             s.update('20200703', '12345680', 'rsync')
             self.assertEqual(s.status[0], [20200703, 12345680, 'rsync', True, '', 1565300090000])
@@ -112,6 +153,35 @@ class TestStatus(unittest.TestCase):
             self.assertTrue(all(b))
             self.assertEqual(len(b), 4)
 
+    @patch('time.time')
+    def test_TransferStatus_update_empty(self, mock_time):
+        """Test status reporting mechanism updates (with no initial JSON file).
+        """
+        mock_time.return_value = 1565300090
+        # st = [[20200703, 12345678, 'rsync', True, '', 1565300074664],
+        #       [20200703, 12345677, 'rsync', True, '', 1565300073000]]
+        with TemporaryDirectory() as d:
+            js = os.path.join(d, 'desi_transfer_status.json')
+            # with open(js, 'w') as f:
+            #     json.dump(st, f, indent=None, separators=(',', ':'))
+            s = TransferStatus(d)
+            s.update('20200703', '12345678', 'checksum')
+            self.assertFalse(os.path.exists(js + '.bak'))
+            self.assertEqual(s.status[0], [20200703, 12345678, 'checksum', True, '', 1565300090000])
+            s.update('20200703', '12345680', 'rsync')
+            self.assertTrue(os.path.exists(js + '.bak'))
+            self.assertEqual(s.status[0], [20200703, 12345680, 'rsync', True, '', 1565300090000])
+            s.update('20200703', '12345678', 'checksum', failure=True)
+            self.assertEqual(s.status[1], [20200703, 12345678, 'checksum', False, '', 1565300090000])
+            s.update('20200703', '12345681', 'pipeline')
+            self.assertEqual(s.status[0], [20200703, 12345681, 'pipeline', True, '', 1565300090000])
+            s.update('20200703', '12345681', 'pipeline', last='arcs')
+            self.assertEqual(s.status[0], [20200703, 12345681, 'pipeline', True, 'arcs', 1565300090000])
+            s.update('20200703', 'all', 'backup')
+            b = [i[3] for i in s.status if i[2] == 'backup']
+            self.assertTrue(all(b))
+            self.assertEqual(len(b), 3)
+
     def test_TransferStatus_find(self):
         """Test status search.
         """