Merge pull request #947 from cmu-delphi/ds/jit-update-meta

[Draft] just-in-time (JIT) meta computations

Author: dshemetov

.github/workflows/ci.yaml

@@ -62,7 +62,7 @@ jobs:
       - name: Start services
         run: |
           docker network create --driver bridge delphi-net
-          docker run --rm -d -p 13306:3306 --network delphi-net --name delphi_database_epidata delphi_database_epidata
+          docker run --rm -d -p 13306:3306 --network delphi-net --name delphi_database_epidata --cap-add=sys_nice delphi_database_epidata
           docker run --rm -d -p 10080:80 --env "SQLALCHEMY_DATABASE_URI=mysql+mysqldb://user:pass@delphi_database_epidata:3306/epidata" --env "FLASK_SECRET=abc" --env "FLASK_PREFIX=/epidata" --network delphi-net --name delphi_web_epidata delphi_web_epidata
           docker ps
 

dev/docker/database/epidata/Dockerfile

@@ -1,5 +1,14 @@
-# start with the `delphi_database` image
-FROM delphi_database
+# start with a standard percona mysql image
+FROM percona:ps-8
+
+# percona exits with the mysql user but we need root for additional setup
+USER root
+
+# use delphi's timezome
+RUN ln -s -f /usr/share/zoneinfo/America/New_York /etc/localtime
+
+# specify a development-only password for the database user "root"
+ENV MYSQL_ROOT_PASSWORD pass
 
 # create the `epidata` database
 ENV MYSQL_DATABASE epidata
@@ -8,8 +17,17 @@ ENV MYSQL_DATABASE epidata
 ENV MYSQL_USER user
 ENV MYSQL_PASSWORD pass
 
+# provide DDL which will configure dev environment at container startup
+COPY repos/delphi/delphi-epidata/dev/docker/database/epidata/_init.sql /docker-entrypoint-initdb.d/
+
 # provide DDL which will create empty tables at container startup
 COPY repos/delphi/delphi-epidata/src/ddl/*.sql /docker-entrypoint-initdb.d/
 
+# provide additional configuration needed for percona
+COPY repos/delphi/delphi-epidata/dev/docker/database/mysql.d/*.cnf /etc/my.cnf.d/
+
 # grant access to SQL scripts
 RUN chmod o+r /docker-entrypoint-initdb.d/*.sql
+
+# restore mysql user for percona
+USER mysql

dev/docker/database/epidata/_init.sql

@@ -0,0 +1,2 @@
+CREATE DATABASE covid;
+GRANT ALL ON covid.* TO 'user';

dev/docker/database/mysql.d/my.cnf

@@ -0,0 +1,2 @@
+[mysqld]
+default_authentication_plugin=mysql_native_password

dev/local/Makefile

@@ -10,19 +10,19 @@
 # Creates all prereq images (delphi_database, delphi_python) only if they don't
 # exist. If you need to rebuild a prereq, you're probably doing something
 # complicated, and can figure out the rebuild command on your own.
-# 
-# 
+#
+#
 # Commands:
-#            
+#
 #   web:     Stops currently-running delphi_web_epidata instances, if any.
 #            Rebuilds delphi_web_epidata image.
 #            Runs image in the background and pipes stdout to a log file.
-# 
+#
 #   db:      Stops currently-running delphi_database_epidata instances, if any.
 #            Rebuilds delphi_database_epidata image.
 #            Runs image in the background and pipes stdout to a log file.
 #            Blocks until database is ready to receive connections.
-# 
+#
 #   python:  Rebuilds delphi_web_python image. You shouldn't need to do this
 #            often; only if you are installing a new environment, or have
 #            made changes to delphi-epidata/dev/docker/python/Dockerfile.
@@ -35,7 +35,7 @@
 #
 #   clean:   Cleans up dangling Docker images.
 #
-# 
+#
 # Optional arguments:
 #   pdb=1         Drops you into debug mode upon test failure, if running tests.
 #   test=         Only runs tests in the directories provided here, e.g.
@@ -105,11 +105,12 @@ db:
 	@# Run the database
 	@docker run --rm -p 127.0.0.1:13306:3306 \
 		--network delphi-net --name delphi_database_epidata \
+                --cap-add=sys_nice \
 		delphi_database_epidata >$(LOG_DB) 2>&1 &
 
 	@# Block until DB is ready
 	@while true; do \
-		sed -n '/Temporary server stopped/,/mysqld: ready for connections/p' $(LOG_DB) | grep "ready for connections" && break; \
+		sed -n '/mysqld: ready for connections/p' $(LOG_DB) | grep "ready for connections" && break; \
 		tail -1 $(LOG_DB); \
 		sleep 1; \
 	done
@@ -127,7 +128,7 @@ py:
 all: web db py
 
 .PHONY=test
-test: 
+test:
 	@docker run -i --rm --network delphi-net \
 		--mount type=bind,source=$(CWD)repos/delphi/delphi-epidata,target=/usr/src/app/repos/delphi/delphi-epidata,readonly \
 		--mount type=bind,source=$(CWD)repos/delphi/delphi-epidata/src,target=/usr/src/app/delphi/epidata,readonly \

docs/Gemfile.lock

@@ -246,7 +246,7 @@ GEM
     thread_safe (0.3.6)
     typhoeus (1.4.0)
       ethon (>= 0.9.0)
-    tzinfo (1.2.9)
+    tzinfo (1.2.10)
       thread_safe (~> 0.1)
     tzinfo-data (1.2021.1)
       tzinfo (>= 1.0.0)

integrations/acquisition/covidcast/test_covidcast_meta_caching.py

@@ -10,9 +10,9 @@
 
 # first party
 from delphi_utils import Nans
-from delphi.epidata.client.delphi_epidata import Epidata
 import delphi.operations.secrets as secrets
-import delphi.epidata.acquisition.covidcast.database as live
+from delphi.epidata.client.delphi_epidata import Epidata
+from delphi.epidata.acquisition.covidcast.database_meta import DatabaseMeta
 from delphi.epidata.acquisition.covidcast.covidcast_meta_cache_updater import main
 
 # py3tester coverage target (equivalent to `import *`)
@@ -40,9 +40,9 @@ def setUp(self):
     cur = cnx.cursor()
 
     # clear all tables
-    cur.execute("truncate table signal_load")
-    cur.execute("truncate table signal_history")
-    cur.execute("truncate table signal_latest")
+    cur.execute("truncate table epimetric_load")
+    cur.execute("truncate table epimetric_full")
+    cur.execute("truncate table epimetric_latest")
     cur.execute("truncate table geo_dim")
     cur.execute("truncate table signal_dim")
     # reset the `covidcast_meta_cache` table (it should always have one row)
@@ -71,14 +71,19 @@ def test_caching(self):
 
     # insert dummy data
     self.cur.execute(f'''
-      INSERT INTO `signal_dim` (`signal_key_id`, `source`, `signal`) VALUES (42, 'src', 'sig');
+      INSERT INTO `signal_dim` (`signal_key_id`, `source`, `signal`)
+      VALUES
+        (42, 'src', 'sig');
     ''')
     self.cur.execute(f'''
-      INSERT INTO `geo_dim` (`geo_key_id`, `geo_type`, `geo_value`) VALUES (96, 'state', 'pa'), (97, 'state', 'wa');
+      INSERT INTO `geo_dim` (`geo_key_id`, `geo_type`, `geo_value`)
+      VALUES
+        (96, 'state', 'pa'), 
+        (97, 'state', 'wa');
     ''')
     self.cur.execute(f'''
       INSERT INTO
-        `signal_latest` (`signal_data_id`, `signal_key_id`, `geo_key_id`, `time_type`,
+        `epimetric_latest` (`epimetric_id`, `signal_key_id`, `geo_key_id`, `time_type`,
 	      `time_value`, `value_updated_timestamp`,
         `value`, `stderr`, `sample_size`,
         `issue`, `lag`, `missing_value`,
@@ -92,7 +97,7 @@ def test_caching(self):
     self.cnx.commit()
 
     # make sure the live utility is serving something sensible
-    cvc_database = live.Database()
+    cvc_database = DatabaseMeta()
     cvc_database.connect()
     epidata1 = cvc_database.compute_covidcast_meta()
     cvc_database.disconnect(False)

integrations/acquisition/covidcast/test_csv_uploading.py

@@ -15,7 +15,6 @@
 from delphi_utils import Nans
 from delphi.epidata.client.delphi_epidata import Epidata
 from delphi.epidata.acquisition.covidcast.csv_to_database import main
-from delphi.epidata.acquisition.covidcast.dbjobs_runner import main as dbjobs_main
 import delphi.operations.secrets as secrets
 
 # py3tester coverage target (equivalent to `import *`)
@@ -37,9 +36,9 @@ def setUp(self):
     cur = cnx.cursor()
 
     # clear all tables
-    cur.execute("truncate table signal_load")
-    cur.execute("truncate table signal_history")
-    cur.execute("truncate table signal_latest")
+    cur.execute("truncate table epimetric_load")
+    cur.execute("truncate table epimetric_full")
+    cur.execute("truncate table epimetric_latest")
     cur.execute("truncate table geo_dim")
     cur.execute("truncate table signal_dim")
     # reset the `covidcast_meta_cache` table (it should always have one row)
@@ -79,9 +78,9 @@ def apply_lag(expected_epidata):
 
   def verify_timestamps_and_defaults(self):
     self.cur.execute('''
-select value_updated_timestamp from signal_history
+select value_updated_timestamp from epimetric_full
 UNION ALL
-select value_updated_timestamp from signal_latest''')
+select value_updated_timestamp from epimetric_latest''')
     for (value_updated_timestamp,) in self.cur:
       self.assertGreater(value_updated_timestamp, 0)
 
@@ -102,8 +101,6 @@ def test_uploading(self):
           log_file=log_file_directory +
           "output.log",
           data_dir=data_dir,
-          is_wip_override=False,
-          not_wip_override=False,
           specific_issue_date=False)
     uploader_column_rename = {"geo_id": "geo_value", "val": "value", "se": "stderr", "missing_val": "missing_value", "missing_se": "missing_stderr"}
 
@@ -123,7 +120,6 @@ def test_uploading(self):
 
       # upload CSVs
       main(args)
-      dbjobs_main()
       response = Epidata.covidcast('src-name', signal_name, 'day', 'state', 20200419, '*')
 
       expected_values = pd.concat([values, pd.DataFrame({ "time_value": [20200419] * 3, "signal": [signal_name] * 3, "direction": [None] * 3})], axis=1).rename(columns=uploader_column_rename).to_dict(orient="records")
@@ -152,7 +148,6 @@ def test_uploading(self):
 
       # upload CSVs
       main(args)
-      dbjobs_main()
       response = Epidata.covidcast('src-name', signal_name, 'day', 'state', 20200419, '*')
 
       expected_values = pd.concat([values, pd.DataFrame({
@@ -187,7 +182,6 @@ def test_uploading(self):
 
       # upload CSVs
       main(args)
-      dbjobs_main()
       response = Epidata.covidcast('src-name', signal_name, 'day', 'state', 20200419, '*')
 
       expected_response = {'result': -2, 'message': 'no results'}
@@ -213,7 +207,6 @@ def test_uploading(self):
 
       # upload CSVs
       main(args)
-      dbjobs_main()
       response = Epidata.covidcast('src-name', signal_name, 'day', 'state', 20200419, '*')
 
       expected_values_df = pd.concat([values, pd.DataFrame({
@@ -232,42 +225,6 @@ def test_uploading(self):
       self.setUp()
 
 
-    with self.subTest("Valid wip"):
-      values = pd.DataFrame({
-        "geo_id": ["me", "nd", "wa"],
-        "val": [10.0, 20.0, 30.0],
-        "se": [0.01, 0.02, 0.03],
-        "sample_size": [100.0, 200.0, 300.0],
-        "missing_val": [Nans.NOT_MISSING] * 3,
-        "missing_se": [Nans.NOT_MISSING] * 3,
-        "missing_sample_size": [Nans.NOT_MISSING] * 3
-      })
-      signal_name = "wip_prototype"
-      values.to_csv(source_receiving_dir + f'/20200419_state_{signal_name}.csv', index=False)
-
-      # upload CSVs
-      main(args)
-      dbjobs_main()
-      response = Epidata.covidcast('src-name', signal_name, 'day', 'state', 20200419, '*')
-
-      expected_values = pd.concat([values, pd.DataFrame({
-        "time_value": [20200419] * 3,
-        "signal": [signal_name] * 3,
-        "direction": [None] * 3
-      })], axis=1).rename(columns=uploader_column_rename).to_dict(orient="records")
-      expected_response = {'result': 1, 'epidata': self.apply_lag(expected_values), 'message': 'success'}
-
-      self.assertEqual(response, expected_response)
-      self.verify_timestamps_and_defaults()
-
-      # Verify that files were archived
-      path = data_dir + f'/archive/successful/src-name/20200419_state_wip_prototype.csv.gz'
-      self.assertIsNotNone(os.stat(path))
-
-      self.tearDown()
-      self.setUp()
-
-
     with self.subTest("Valid signal with name length 32<x<64"):
       values = pd.DataFrame({
         "geo_id": ["pa"],
@@ -278,12 +235,11 @@ def test_uploading(self):
         "missing_se": [Nans.NOT_MISSING],
         "missing_sample_size": [Nans.NOT_MISSING]
       })
-      signal_name = "wip_really_long_name_that_will_be_accepted"
+      signal_name = "really_long_name_that_will_be_accepted"
       values.to_csv(source_receiving_dir + f'/20200419_state_{signal_name}.csv', index=False)
 
       # upload CSVs
       main(args)
-      dbjobs_main()
       response = Epidata.covidcast('src-name', signal_name, 'day', 'state', 20200419, '*')
 
       expected_values = pd.concat([values, pd.DataFrame({
@@ -310,12 +266,11 @@ def test_uploading(self):
         "missing_se": [Nans.NOT_MISSING],
         "missing_sample_size": [Nans.NOT_MISSING]
       })
-      signal_name = "wip_really_long_name_that_will_get_truncated_lorem_ipsum_dolor_sit_amet"
+      signal_name = "really_long_name_that_will_get_truncated_lorem_ipsum_dolor_sit_amet"
       values.to_csv(source_receiving_dir + f'/20200419_state_{signal_name}.csv', index=False)
 
       # upload CSVs
       main(args)
-      dbjobs_main()
       response = Epidata.covidcast('src-name', signal_name, 'day', 'state', 20200419, '*')
 
       expected_response = {'result': -2, 'message': 'no results'}
@@ -332,7 +287,6 @@ def test_uploading(self):
         f.write('this,header,is,wrong\n')
 
       main(args)
-      dbjobs_main()
 
       path = data_dir + '/archive/failed/src-name/20200420_state_test.csv'
       self.assertIsNotNone(os.stat(path))
@@ -346,7 +300,6 @@ def test_uploading(self):
         f.write('file name is wrong\n')
 
       main(args)
-      dbjobs_main()
 
       path = data_dir + '/archive/failed/unknown/hello.csv'
       self.assertIsNotNone(os.stat(path))