FP 0.27 Stream cans: parallel stream + process

Add flags to prevent writing to db and to disk
Generate ETA metric
Fix generator type hints
Add unit test

Author: Federico Ceratto

af/fastpath/debian/changelog

@@ -1,3 +1,9 @@
+fastpath (0.27) unstable; urgency=medium
+
+  * Stream cans
+
+ -- Federico Ceratto <[email protected]>  Fri, 01 May 2020 13:36:08 +0100
+
 fastpath (0.26) unstable; urgency=medium
 
   * Improve file trimming

af/fastpath/fastpath/core.py

@@ -106,6 +106,10 @@ def setup():
     ap.add_argument(
         "--stop-after", type=int, help="Stop after feeding N measurements", default=None
     )
+    ap.add_argument("--no-write-msmt", action="store_true",
+                    help="Do not write measurement on disk")
+    ap.add_argument("--no-write-to-db", action="store_true",
+                    help="Do not insert measurement in database")
     conf = ap.parse_args()
 
     if conf.devel or conf.stdout or no_journal_handler:
@@ -241,39 +245,6 @@ def clean_caches():
 }
 
 
-@metrics.timer("load_s3_measurements")
-def load_s3_measurements(day) -> Iterator[MsmtTup]:
-    # TODO: move this into s3feeder
-    t0 = time.time()
-    path = conf.s3cachedir / str(day)
-    log.info("Scanning %s", path.absolute())
-    files = []
-    with os.scandir(path) as d:
-        for e in d:
-            if not e.is_file():
-                continue
-            if e.name == "index.json.gz":
-                continue
-            files.append(e)
-
-    fcnt = 0
-    for e in sorted(files, key=lambda f: f.name):
-        fcnt += 1
-        log.info(f"Ingesting [{fcnt}/{len(files)}] {e.name}")
-        fn = os.path.join(path, e.name)
-        try:
-            for measurement_tup in s3feeder.load_multiple(fn):
-                yield measurement_tup
-        except:
-            log.error(f"Ingesting [{fcnt}/{len(files)}] {e.name}", exc_info=True)
-
-        remaining = (time.time() - t0) * (len(files) - fcnt) / fcnt
-        metrics.gauge("load_s3_measurements_eta", remaining)
-        metrics.gauge("load_s3_measurements_remaining_files", len(files) - fcnt)
-        remaining_td = timedelta(seconds=remaining)
-        log.info("load_s3_measurements remaining time: %s", remaining_td)
-
-
 def prepare_for_json_normalize(report):
     try:
         d = report["test_keys"]["control"]["tcp_connect"]
@@ -296,22 +267,9 @@ def fetch_measurements(start_day, end_day) -> Iterator[MsmtTup]:
     # --start-day                   -> Run on old cans, then over SSH
     # --start-day and --end-day     -> Run on old cans
     # --start-day and --end-day  with the same date     -> NOP
-    today = date.today()
-    if start_day and start_day < today:
-        # TODO: fetch day N+1 while processing day N
-        log.info("Fetching older cans from S3")
-        day = start_day
-        while day < (end_day or today):
-            log.info("Processing %s", day)
-            s3feeder.fetch_cans_for_a_day_with_cache(conf, day)
-            for measurement_tup in load_s3_measurements(day):
-                yield measurement_tup
-
-            day += timedelta(days=1)
-
-        if end_day:
-            log.info("Reached {end_day}, exiting")
-            return
+
+    for measurement_tup in s3feeder.stream_cans(conf, start_day, end_day):
+        yield measurement_tup
 
     if conf.nossh:
         log.info("Not fetching over SSH")
@@ -1295,7 +1253,11 @@ def file_trimmer(conf):
 def msm_processor(queue):
     """Measurement processor worker
     """
-    db.setup(conf)
+    if conf.no_write_to_db:
+        log.warning("not writing to database")
+    else:
+        db.setup(conf)
+
     while True:
         msm_tup = queue.get()
         if msm_tup is None:
@@ -1314,7 +1276,8 @@ def msm_processor(queue):
                 sshfeeder.log_ingestion_delay(measurement)
                 log.debug(f"Processing {tid} {rid} {inp}")
                 fn = generate_filename(tid)
-                writeout_measurement(msm_jstr, fn, conf.update, tid)
+                if not conf.no_write_msmt:
+                    writeout_measurement(msm_jstr, fn, conf.update, tid)
                 if measurement.get("test_name", None) == "web_connectivity":
                     matches = match_fingerprints(measurement)
                 else:
@@ -1330,16 +1293,17 @@ def msm_processor(queue):
                     log.debug(
                         f"Storing {tid} {rid} {inp} A{int(anomaly)} F{int(failure)} C{int(confirmed)}"
                     )
-                db.upsert_summary(
-                    measurement,
-                    scores,
-                    anomaly,
-                    confirmed,
-                    failure,
-                    tid,
-                    fn,
-                    conf.update,
-                )
+                if not conf.no_write_to_db:
+                    db.upsert_summary(
+                        measurement,
+                        scores,
+                        anomaly,
+                        confirmed,
+                        failure,
+                        tid,
+                        fn,
+                        conf.update,
+                    )
             except Exception as e:
                 log.exception(e)
                 metrics.incr("unhandled_exception")
@@ -1366,7 +1330,10 @@ def core():
     scores = None
 
     # Spawn file trimmer process
-    mp.Process(target=file_trimmer, args=(conf,)).start()
+    if conf.no_write_msmt:
+        log.warning("Not trimming old measurement on disk")
+    else:
+        mp.Process(target=file_trimmer, args=(conf,)).start()
 
     # Spawn worker processes
     # 'queue' is a singleton from the portable_queue module

af/fastpath/fastpath/db.py

@@ -9,8 +9,6 @@
 
 from textwrap import dedent
 import logging
-import os
-import time
 
 import psycopg2  # debdeps: python3-psycopg2
 from psycopg2.extras import Json

af/fastpath/fastpath/s3feeder.py

@@ -15,7 +15,9 @@
 
 """
 
-from typing import Iterator
+from datetime import date, timedelta
+from typing import Generator
+from pathlib import Path
 import logging
 import os
 import time
@@ -42,14 +44,11 @@
     logging.getLogger(l).setLevel(logging.INFO)
 
 
-def load_multiple(fn, touch=True) -> Iterator[MsmtTup]:
+def load_multiple(fn: str) -> Generator[MsmtTup, None, None]:
     """Load contents of cans. Decompress tar archives if found.
     Yields measurements one by one as:
         (string of JSON, None) or (None, msmt dict)
     """
-    if touch:
-        os.utime(fn)  # update access time - used for cache cleanup
-
     # TODO: handle:
     # RuntimeError: LZ4F_decompress failed with code: ERROR_decompressionFailed
     if fn.endswith(".tar.lz4"):
@@ -81,10 +80,10 @@ def load_multiple(fn, touch=True) -> Iterator[MsmtTup]:
     elif fn.endswith(".yaml.lz4"):
         with lz4frame.open(fn) as f:
             raise Exception("Unsupported format: YAML")
-            bucket_tstamp = "FIXME"
-            for msm in iter_yaml_msmt_normalized(f, bucket_tstamp):
-                metrics.incr("yaml_normalization")
-                yield (None, msm)
+            # bucket_tstamp = "FIXME"
+            # for msm in iter_yaml_msmt_normalized(f, bucket_tstamp):
+            #     metrics.incr("yaml_normalization")
+            #     yield (None, msm)
 
     else:
         raise RuntimeError(fn)
@@ -94,16 +93,6 @@ def create_s3_client():
     return boto3.Session(profile_name=AWS_PROFILE).client("s3")
 
 
-def list_cans_on_s3(prefix):
-    # TODO list files based on date and return them
-    s3 = create_s3_client()
-    r = s3.list_objects_v2(Bucket=BUCKET_NAME, Prefix="canned/" + prefix)
-    for filedesc in r["Contents"]:
-        fn = filedesc["Key"]
-        size = filedesc["Size"]
-        print("%s size %d MB" % (fn, size / 1024 / 1024))
-
-
 def list_cans_on_s3_for_a_day(s3, day):
     prefix = f"{day}/"
     r = s3.list_objects_v2(Bucket=BUCKET_NAME, Prefix="canned/" + prefix)
@@ -116,10 +105,11 @@ def list_cans_on_s3_for_a_day(s3, day):
 
 
 @metrics.timer("fetch_cans")
-def fetch_cans(s3, conf, files):
+def fetch_cans(s3, conf, files) -> Generator[Path, None, None]:
     """
     Download cans to a local directory
     fnames = [("2013-09-12/20130912T150305Z-MD-AS1547-http_", size), ... ]
+    yield each can file Path
     """
     # fn: can filename without path
     # diskf: File in the s3cachedir directory
@@ -144,7 +134,7 @@ def _cb(bytes_count):
         _cb.count += bytes_count
         _cb.total_count += bytes_count
         metrics.gauge("s3_download_percentage", _cb.total_count / _cb.total_size * 100)
-        log.debug("s3_download_percentage %d", _cb.total_count / _cb.total_size * 100)
+        # log.debug("s3_download_percentage %d", _cb.total_count / _cb.total_size * 100)
         try:
             speed = _cb.count / 131_072 / (time.time() - _cb.start_time)
             metrics.gauge("s3_download_speed_avg_Mbps", speed)
@@ -169,16 +159,71 @@ def _cb(bytes_count):
         metrics.gauge("fetching", 0)
         tmpf.rename(diskf)
         assert size == diskf.stat().st_size
+        yield diskf
 
     metrics.gauge("s3_download_speed_avg_Mbps", 0)
 
 
+# TODO: merge with stream_daily_cans, add caching to the latter to be used
+# during functional tests
 @metrics.timer("fetch_cans_for_a_day_with_cache")
 def fetch_cans_for_a_day_with_cache(conf, day):
     s3 = create_s3_client()
     fns = list_cans_on_s3_for_a_day(s3, day)
-    #can_names = [
-    #    "2020-03-03/web_connectivity.14.tar.lz4",
-    #]
-    #fns = [(name, size) for name, size in fns if name in can_names]
-    fetch_cans(s3, conf, fns)
+    list(fetch_cans(s3, conf, fns))
+
+
+def _calculate_etr(t0, now, start_day, day, stop_day, can_num, can_tot_count) -> int:
+    """Estimate total runtime in seconds.
+    stop_day is not included, can_num starts from 0
+    """
+    tot_days_count = (stop_day - start_day).days
+    elapsed = now - t0
+    days_done = (day - start_day).days
+    fraction_of_day_done = (can_num + 1) / float(can_tot_count)
+    etr = elapsed * tot_days_count / (days_done + fraction_of_day_done)
+    return etr
+
+
+def _update_eta(t0, start_day, day, stop_day, can_num, can_tot_count):
+    """Generate metric process_s3_measurements_eta expressed as epoch
+    """
+    try:
+        now = time.time()
+        etr = _calculate_etr(t0, now, start_day, day, stop_day, can_num, can_tot_count)
+        eta = t0 + etr
+        metrics.gauge("process_s3_measurements_eta", eta)
+    except:
+        pass
+
+
+def stream_cans(conf, start_day: date, end_day: date) -> Generator[MsmtTup, None, None]:
+    """Stream cans from S3
+    """
+    today = date.today()
+    if not start_day or start_day >= today:
+        return
+
+    log.info("Fetching older cans from S3")
+    t0 = time.time()
+    day = start_day
+    s3 = create_s3_client()
+    # the last day is not included
+    stop_day = end_day if end_day < today else today
+    while day < stop_day:
+        log.info("Processing day %s", day)
+        cans_fns = list_cans_on_s3_for_a_day(s3, day)
+        for cn, can_f in enumerate(fetch_cans(s3, conf, cans_fns)):
+            try:
+                _update_eta(t0, start_day, day, stop_day, cn, len(cans_fns))
+                log.info("can %s ready", can_f.name)
+                for msmt_tup in load_multiple(can_f.as_posix()):
+                    yield msmt_tup
+            except Exception as e:
+                log.error(str(e), exc_info=True)
+
+        day += timedelta(days=1)
+
+    if end_day:
+        log.info("Reached {end_day}, stream_cans finished")
+        return

af/fastpath/fastpath/tests/test_unit.py

@@ -2,7 +2,10 @@
 # Fastpath - unit tests
 #
 
+from datetime import date
+
 import fastpath.core as fp
+import fastpath.s3feeder as s3feeder
 import ujson
 
 
@@ -165,3 +168,24 @@ def test_bug_backend352():
         "blocking_isp": 0.0,
         "blocking_local": 0.0,
     }
+
+
+def test_s3feeder_eta():
+    t0 = 1588200000
+    now = t0 + 3600
+    start_day = date(2020, 1, 1)
+    day = date(2020, 1, 1)
+    stop_day = date(2020, 1, 2)
+
+    etr = s3feeder._calculate_etr(t0, now, start_day, day, stop_day, 0, 4)
+    assert etr / 3600 == 4
+    etr = s3feeder._calculate_etr(t0, now, start_day, day, stop_day, 3, 4)
+    assert etr / 3600 == 1
+    etr = s3feeder._calculate_etr(
+        t0, now, start_day, date(2020, 1, 2), date(2020, 1, 5), -1, 9
+    )
+    assert etr / 3600 == 4.0
+    etr = s3feeder._calculate_etr(
+        t0, now, start_day, date(2020, 1, 4), date(2020, 1, 5), 9, 10
+    )
+    assert etr / 3600 == 1.0