Merge pull request #415 from benjaminysmith/ingestion_logging

Add logging to CSV ingestion

Author: krivard

integrations/acquisition/covidcast/test_csv_uploading.py

@@ -60,7 +60,9 @@ def test_uploading(self):
     # make some fake data files
     data_dir = 'covid/data'
     source_receiving_dir = data_dir + '/receiving/src-name'
+    log_file_directory = "/var/log/"
     os.makedirs(source_receiving_dir, exist_ok=True)
+    os.makedirs(log_file_directory, exist_ok=True)
 
     # valid
     with open(source_receiving_dir + '/20200419_state_test.csv', 'w') as f:
@@ -96,7 +98,13 @@ def test_uploading(self):
 
     # upload CSVs
     # TODO: use an actual argparse object for the args instead of a MagicMock
-    args = MagicMock(data_dir=data_dir, is_wip_override=False, not_wip_override=False, specific_issue_date=False)
+    args = MagicMock(
+        log_file=log_file_directory +
+        "output.log",
+        data_dir=data_dir,
+        is_wip_override=False,
+        not_wip_override=False,
+        specific_issue_date=False)
     main(args)
 
     # request CSV data from the API

src/acquisition/covidcast/csv_to_database.py

@@ -3,11 +3,13 @@
 # standard library
 import argparse
 import os
+import time
 
 # first party
 from delphi.epidata.acquisition.covidcast.csv_importer import CsvImporter
 from delphi.epidata.acquisition.covidcast.database import Database, CovidcastRow
 from delphi.epidata.acquisition.covidcast.file_archiver import FileArchiver
+from delphi.epidata.acquisition.covidcast.logger import get_structured_logger
 
 
 def get_argument_parser():
@@ -30,6 +32,9 @@ def get_argument_parser():
     '--not_wip_override',
     action='store_true',
     help='overrides all signals to mark them as *not* WIP.  NOTE: specify neither or only one of --is_wip_override and --not_wip_override.')
+  parser.add_argument(
+    '--log_file',
+    help="filename for log output (defaults to stdout)")
   return parser
 
 def collect_files(data_dir, specific_issue_date, csv_importer_impl=CsvImporter):
@@ -77,6 +82,7 @@ def upload_archive(
     path_details,
     database,
     handlers,
+    logger,
     is_wip_override=None,
     csv_importer_impl=CsvImporter):
   """Upload CSVs to the database and archive them using the specified handlers.
@@ -123,6 +129,15 @@ def upload_archive(
       try:
         result = database.insert_or_update_bulk(rows_list)
         print(f"insert_or_update_bulk {filename} returned {result}")
+        logger.info(
+          "Inserted database rows",
+          row_count = result,
+          source = source,
+          signal = signal,
+          geo_type = geo_type,
+          time_value = time_value,
+          issue = issue,
+          lag = lag)
         if result is None or result: # else would indicate zero rows inserted
           database.commit()
       except Exception as e:
@@ -143,6 +158,9 @@ def main(
     upload_archive_impl=upload_archive):
   """Find, parse, and upload covidcast signals."""
 
+  logger = get_structured_logger("csv_ingestion", filename=args.log_file)
+  start_time = time.time()
+
   if args.is_wip_override and args.not_wip_override:
     print('conflicting overrides for forcing WIP option!  exiting...')
     return
@@ -157,7 +175,9 @@ def main(
   if not path_details:
     print('nothing to do; exiting...')
     return
-    
+  
+  logger.info("Ingesting CSVs", csv_count = len(path_details))
+
   database = database_impl()
   database.connect()
   num_starting_rows = database.count_all_rows()
@@ -167,16 +187,22 @@ def main(
       path_details,
       database,
       make_handlers(args.data_dir, args.specific_issue_date),
+      logger,
       is_wip_override=wip_override)
   finally:
     # no catch block so that an exception above will cause the program to fail
     # after the following cleanup
     try:
       num_inserted_rows = database.count_all_rows() - num_starting_rows
+      logger.info("Finished inserting database rows", row_count = num_inserted_rows)
       print('inserted/updated %d rows' % num_inserted_rows)
     finally:
       # unconditionally commit database changes since CSVs have been archived
       database.disconnect(True)
+  
+  logger.info(
+      "Ingested CSVs into database",
+      total_runtime_in_seconds=round(time.time() - start_time, 2))
 
 if __name__ == '__main__':
   main(get_argument_parser().parse_args())

tests/acquisition/covidcast/test_csv_to_database.py

@@ -76,12 +76,14 @@ def load_csv_impl(path, *args):
     mock_csv_importer = MagicMock()
     mock_csv_importer.load_csv = load_csv_impl
     mock_file_archiver = MagicMock()
+    mock_logger = MagicMock()
 
     upload_archive(
       self._path_details(),
       mock_database,
       make_handlers(data_dir, False,
                     file_archiver_impl=mock_file_archiver),
+      mock_logger,
       csv_importer_impl=mock_csv_importer)
 
     # verify that appropriate rows were added to the database
@@ -114,7 +116,7 @@ def test_main_successful(self):
     """Run the main program successfully, then commit changes."""
 
     # TODO: use an actual argparse object for the args instead of a MagicMock
-    args = MagicMock(data_dir='data', is_wip_override=False, not_wip_override=False, specific_issue_date=False)
+    args = MagicMock(log_file=None, data_dir='data', is_wip_override=False, not_wip_override=False, specific_issue_date=False)
     mock_database = MagicMock()
     mock_database.count_all_rows.return_value = 0
     fake_database_impl = lambda: mock_database
@@ -142,7 +144,7 @@ def test_main_unsuccessful(self):
     """Run the main program with failure, then commit changes."""
 
     # TODO: use an actual argparse object for the args instead of a MagicMock
-    args = MagicMock(data_dir='data', is_wip_override=False, not_wip_override=False, specific_issue_date=False)
+    args = MagicMock(log_file=None, data_dir='data', is_wip_override=False, not_wip_override=False, specific_issue_date=False)
     mock_database = MagicMock()
     mock_database.count_all_rows.return_value = 0
     fake_database_impl = lambda: mock_database
@@ -168,7 +170,7 @@ def test_main_early_exit(self):
     """Run the main program with an empty receiving directory."""
 
     # TODO: use an actual argparse object for the args instead of a MagicMock
-    args = MagicMock(data_dir='data', is_wip_override=False, not_wip_override=False, specific_issue_date=False)
+    args = MagicMock(log_file=None, data_dir='data', is_wip_override=False, not_wip_override=False, specific_issue_date=False)
     mock_database = MagicMock()
     mock_database.count_all_rows.return_value = 0
     fake_database_impl = lambda: mock_database
@@ -204,11 +206,13 @@ def test_database_exception_is_handled(self):
       MagicMock(geo_value='geo', value=1, stderr=1, sample_size=1),
     ]
     mock_file_archiver = MagicMock()
+    mock_logger = MagicMock()
 
     upload_archive(
         collect_files(data_dir, False, csv_importer_impl=mock_csv_importer),
         mock_database,
         make_handlers(data_dir, False, file_archiver_impl=mock_file_archiver),
+        mock_logger,
         csv_importer_impl=mock_csv_importer,
         )