Merge pull request #700 from benjaminysmith/even_more_logging

Add logging to google health and symptoms.

Author: krivard

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

@@ -13,5 +13,6 @@
   },
   "bucket_name": "delphi-covidcast-indicator-output",
   "test": false,
-  "test_data_dir": ""
+  "test_data_dir": "",
+  "log_filename": "/var/log/indicators/google_health.log"
 }

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

@@ -3,5 +3,6 @@
   "static_file_dir": "./static",
   "export_dir": "/common/covidcast/receiving/google-symptoms",
   "cache_dir": "./cache",
-  "base_url": "https://raw.githubusercontent.com/google-research/open-covid-19-data/master/data/exports/search_trends_symptoms_dataset/United%20States%20of%20America{sub_url}2020_US_{state}daily_symptoms_dataset.csv"
+  "base_url": "https://raw.githubusercontent.com/google-research/open-covid-19-data/master/data/exports/search_trends_symptoms_dataset/United%20States%20of%20America{sub_url}2020_US_{state}daily_symptoms_dataset.csv",
+  "log_filename": "/var/log/indicators/google_symptoms.log"
 }

google_health/delphi_google_health/run.py

@@ -7,14 +7,16 @@
 
 import datetime
 import logging
+import time
 
 import pandas as pd
 
 from delphi_utils import (
     read_params,
     S3ArchiveDiffer,
     add_prefix,
-    create_export_csv
+    create_export_csv,
+    get_structured_logger
 )
 
 from .data_tools import format_for_export
@@ -32,7 +34,10 @@ def run_module():
     the directory defined by the `export_dir` (should be "receiving" expect for
     testing purposes).
     """
-
+    start_time = time.time()
+    csv_export_count = 0
+    oldest_final_export_date = None
+    
     # read parameters
     params = read_params()
     ght_key = params["ght_key"]
@@ -44,6 +49,8 @@ def run_module():
     wip_signal = params["wip_signal"]
     cache_dir = params["cache_dir"]
 
+    logger = get_structured_logger(__name__, filename = params.get("log_filename"))
+
     arch_diff = S3ArchiveDiffer(
         cache_dir, export_dir,
         params["bucket_name"], "ght",
@@ -95,11 +102,19 @@ def run_module():
     for signal in signal_names:
         is_smoothed = signal.endswith(SMOOTHED)
         for geo_res, df in df_by_geo_res.items():
-            create_export_csv(format_for_export(df, is_smoothed),
-                              geo_res=geo_res,
-                              sensor=signal,
-                              start_date=start_date,
-                              export_dir=export_dir)
+            exported_csv_dates = create_export_csv(
+                                    format_for_export(df, is_smoothed),
+                                    geo_res=geo_res,
+                                    sensor=signal,
+                                    start_date=start_date,
+                                    export_dir=export_dir)
+
+            if not exported_csv_dates.empty:
+                csv_export_count += exported_csv_dates.size
+                if not oldest_final_export_date:
+                    oldest_final_export_date = max(exported_csv_dates)
+                oldest_final_export_date = min(
+                    oldest_final_export_date, max(exported_csv_dates))
 
     if not params["test"]:
         # Diff exports, and make incremental versions
@@ -117,3 +132,15 @@ def run_module():
         # Report failures: someone should probably look at them
         for exported_file in fails:
             print(f"Failed to archive '{exported_file}'")
+    
+    elapsed_time_in_seconds = round(time.time() - start_time, 2)
+    max_lag_in_days = None
+    formatted_oldest_final_export_date = None
+    if oldest_final_export_date:
+        max_lag_in_days = (datetime.datetime.now() - oldest_final_export_date).days
+        formatted_oldest_final_export_date = oldest_final_export_date.strftime("%Y-%m-%d")
+    logger.info("Completed indicator run",
+        elapsed_time_in_seconds = elapsed_time_in_seconds,
+        csv_export_count = csv_export_count,
+        max_lag_in_days = max_lag_in_days,
+        oldest_final_export_date = formatted_oldest_final_export_date)

google_symptoms/delphi_google_symptoms/run.py

@@ -8,7 +8,13 @@
 from itertools import product
 
 import numpy as np
-from delphi_utils import read_params, create_export_csv, geomap
+import time
+from delphi_utils import (
+    read_params, 
+    create_export_csv, 
+    geomap,
+    get_structured_logger
+)
 
 from .pull import pull_gs_data
 from .geo import geo_map
@@ -18,11 +24,17 @@
 
 def run_module():
     """Run Google Symptoms module."""
+    start_time = time.time()
+    csv_export_count = 0
+    oldest_final_export_date = None
+
     params = read_params()
     export_start_date = datetime.strptime(params["export_start_date"], "%Y-%m-%d")
     export_dir = params["export_dir"]
     base_url = params["base_url"]
 
+    logger = get_structured_logger(__name__, filename = params.get("log_filename"))
+
     # Pull GS data
     dfs = pull_gs_data(base_url)
     gmpr = geomap.GeoMapper()
@@ -47,10 +59,29 @@ def run_module():
             df = df.loc[~df["val"].isnull(), :]
             df = df.reset_index()
             sensor_name = "_".join([smoother, "search"])
-            create_export_csv(
+            exported_csv_dates = create_export_csv(
                 df,
                 export_dir=export_dir,
                 start_date=SMOOTHERS_MAP[smoother][1](export_start_date),
                 metric=metric.lower(),
                 geo_res=geo_res,
                 sensor=sensor_name)
+            
+            if not exported_csv_dates.empty:
+                csv_export_count += exported_csv_dates.size
+                if not oldest_final_export_date:
+                    oldest_final_export_date = max(exported_csv_dates)
+                oldest_final_export_date = min(
+                    oldest_final_export_date, max(exported_csv_dates))
+
+    elapsed_time_in_seconds = round(time.time() - start_time, 2)
+    max_lag_in_days = None
+    formatted_oldest_final_export_date = None
+    if oldest_final_export_date:
+        max_lag_in_days = (datetime.now() - oldest_final_export_date).days
+        formatted_oldest_final_export_date = oldest_final_export_date.strftime("%Y-%m-%d")
+    logger.info("Completed indicator run",
+        elapsed_time_in_seconds = elapsed_time_in_seconds,
+        csv_export_count = csv_export_count,
+        max_lag_in_days = max_lag_in_days,
+        oldest_final_export_date = formatted_oldest_final_export_date)