Merge pull request #982 from cmu-delphi/main

Deploy runner-based logging fixes to production

Author: krivard

_delphi_utils_python/delphi_utils/logger.py

@@ -40,16 +40,12 @@ def get_structured_logger(name=__name__,
     is a good choice.
     filename: An (optional) file to write log output.
     """
-    # Configure the underlying logging configuration
-    handlers = [logging.StreamHandler()]
-    if filename:
-        handlers.append(logging.FileHandler(filename))
-
+    # Configure the basic underlying logging configuration
     logging.basicConfig(
         format="%(message)s",
         level=logging.INFO,
-        handlers=handlers
-        )
+        handlers=[logging.StreamHandler()]
+    )
 
     # Configure structlog. This uses many of the standard suggestions from
     # the structlog documentation.
@@ -84,7 +80,12 @@ def get_structured_logger(name=__name__,
         cache_logger_on_first_use=True,
     )
 
-    logger = structlog.get_logger(name)
+    # Create the underlying python logger and wrap it with structlog
+    system_logger = logging.getLogger(name)
+    if filename:
+        system_logger.addHandler(logging.FileHandler(filename))
+    system_logger.setLevel(logging.INFO)
+    logger = structlog.wrap_logger(system_logger)
 
     if log_exceptions:
         handle_exceptions(logger)

_delphi_utils_python/delphi_utils/runner.py

@@ -3,6 +3,7 @@
 import importlib
 from typing import Any, Callable, Dict, Optional
 from .archive import ArchiveDiffer, archiver_from_params
+from .logger import get_structured_logger
 from .utils import read_params
 from .validator.validate import Validator
 from .validator.run import validator_from_params
@@ -40,6 +41,7 @@ def run_indicator_pipeline(indicator_fn:  Callable[[Params], None],
     archiver = archiver_fn(params)
     if validator:
         validation_report = validator.validate()
+        validation_report.log(get_structured_logger(params["common"].get("log_filename", None)))
     if archiver and (not validator or validation_report.success()):
         archiver.archive()
 

_delphi_utils_python/delphi_utils/validator/report.py

@@ -4,8 +4,6 @@
 from ..logger import get_structured_logger
 from .errors import ValidationFailure
 
-logger = get_structured_logger(__name__)
-
 class ValidationReport:
     """Class for reporting the results of validation."""
 
@@ -83,14 +81,17 @@ def summary(self):
         out_str += f"{len(self.raised_warnings)} warnings\n"
         return out_str
 
-    def log(self):
+    def log(self, logger=None):
         """Log errors and warnings."""
+        if logger is None:
+            logger = get_structured_logger(__name__)
+
         for error in self.unsuppressed_errors:
             logger.critical(str(error))
         for warning in self.raised_warnings:
             logger.warning(str(warning))
 
-    def print_and_exit(self, die_on_failures=True):
+    def print_and_exit(self, logger=None, die_on_failures=True):
         """Print results and exit.
 
         Arguments
@@ -99,7 +100,7 @@ def print_and_exit(self, die_on_failures=True):
             Whether to return non-zero status if any failures were encountered.
         """
         print(self.summary())
-        self.log()
+        self.log(logger)
         if self.success():
             sys.exit(0)
         elif die_on_failures:

_delphi_utils_python/delphi_utils/validator/run.py

@@ -5,7 +5,7 @@
 when the module is run with `python -m delphi_utils.validator`.
 """
 import argparse as ap
-from .. import read_params
+from .. import read_params, get_structured_logger
 from .validate import Validator
 
 
@@ -15,8 +15,12 @@ def run_module():
     parser.add_argument("--dry_run", action="store_true", help="When provided, return zero exit"
                         " status irrespective of the number of failures")
     args = parser.parse_args()
-    validator = Validator(read_params())
-    validator.validate().print_and_exit(not args.dry_run)
+    params = read_params()
+    validator = Validator(params)
+    validator.validate().print_and_exit(
+        get_structured_logger(__name__,
+                              params["common"].get("log_filename", None),
+                              not args.dry_run))
 
 
 def validator_from_params(params):

_delphi_utils_python/tests/test_runner.py

@@ -35,6 +35,7 @@ class TestRunIndicator:
     """Fixture for running indicators."""
     # arbitrary params to pass to function generators
     PARAMS = {
+        "common": {},
         "indicator": {"a": 1},
         "validation": {"b": 2},
         "archive": {"c": 3}

_delphi_utils_python/tests/validator/test_report.py

@@ -44,9 +44,9 @@ def test_str(self):
         assert report.summary() ==\
             "3 checks run\n1 checks failed\n1 checks suppressed\n2 warnings\n"
 
-    @mock.patch("delphi_utils.validator.report.logger")
-    def test_log(self, mock_logger):
+    def test_log(self):
         """Test that the logs contain all failures and warnings."""
+        mock_logger = mock.Mock()
         report = ValidationReport([self.ERROR_1])
         report.increment_total_checks()
         report.increment_total_checks()
@@ -56,7 +56,7 @@ def test_log(self, mock_logger):
         report.add_raised_error(self.ERROR_1)
         report.add_raised_error(self.ERROR_2)
 
-        report.log()
+        report.log(mock_logger)
         mock_logger.critical.assert_called_once_with(
             "bad failed for sig2 at resolution county on 2020-11-07: msg 2")
         mock_logger.warning.assert_has_calls([mock.call("wrong import"), mock.call("right import")])

ansible/templates/google_symptoms-params-prod.json.j2

@@ -5,7 +5,7 @@
   },
   "indicator": {
     "export_start_date": "2020-02-20",
-    "num_export_days": 14,
+    "num_export_days": null,
     "bigquery_credentials": {
       "type": "{{ google_symptoms_account_type }}",
       "project_id": "{{ google_symptoms_project_id }}",

google_symptoms/delphi_google_symptoms/run.py

@@ -7,6 +7,7 @@
 import time
 from datetime import datetime, date
 from itertools import product
+import covidcast
 
 import numpy as np
 from delphi_utils import (
@@ -52,7 +53,13 @@ def run_module(params):
         ), "%Y-%m-%d")
 
     export_dir = params["common"]["export_dir"]
-    num_export_days = params["indicator"].get("num_export_days", "all")
+    num_export_days = params["indicator"]["num_export_days"]
+
+    if num_export_days is None:
+        # Get number of days based on what's missing from the API.
+        metadata = covidcast.metadata()
+        gs_metadata = metadata[(metadata.data_source == "google-symptoms")]
+        num_export_days = max(gs_metadata.min_lag)
 
     logger = get_structured_logger(
         __name__, filename=params["common"].get("log_filename"),

google_symptoms/params.json.template

@@ -5,7 +5,7 @@
   },
   "indicator": {
     "export_start_date": "2020-02-20",
-    "num_export_days": 14,
+    "num_export_days": null,
     "bigquery_credentials": {}
   },
   "validation": {