Purge + Restore user timeseries data with long-term storage

bin/debug/load_multi_timeline_for_range.py

@@ -14,7 +14,7 @@
 import emission.storage.json_wrappers as esj
 import argparse
 
-import common
+import bin.debug.common as common
 import os
 
 import gzip
@@ -26,19 +26,19 @@
 
 args = None
 
-def register_fake_users(prefix, unique_user_list):
+def register_fake_users(prefix, unique_user_list, verbose):
     logging.info("Creating user entries for %d users" % len(unique_user_list))
 
     format_string = "{0}-%0{1}d".format(prefix, len(str(len(unique_user_list))))
     logging.info("pattern = %s" % format_string)
 
     for i, uuid in enumerate(unique_user_list):
         username = (format_string % i)
-        if args.verbose is not None and i % args.verbose == 0:
+        if verbose is not None and i % verbose == 0:
             logging.info("About to insert mapping %s -> %s" % (username, uuid))
         user = ecwu.User.registerWithUUID(username, uuid)
 
-def register_mapped_users(mapfile, unique_user_list):
+def register_mapped_users(mapfile, unique_user_list, verbose):
     uuid_entries = json.load(open(mapfile), object_hook=esj.wrapped_object_hook)
     logging.info("Creating user entries for %d users from map of length %d" % (len(unique_user_list), len(mapfile)))
 
@@ -50,17 +50,17 @@ def register_mapped_users(mapfile, unique_user_list):
         # register this way
         # Pro: will do everything that register does, including creating the profile
         # Con: will insert only username and uuid - id and update_ts will be different
-        if args.verbose is not None and i % args.verbose == 0:
+        if verbose is not None and i % verbose == 0:
             logging.info("About to insert mapping %s -> %s" % (username, uuid))
         user = ecwu.User.registerWithUUID(username, uuid)
 
-def get_load_ranges(entries):
-    start_indices = list(range(0, len(entries), args.batch_size))
+def get_load_ranges(entries, batch_size):
+    start_indices = list(range(0, len(entries), batch_size))
     ranges = list(zip(start_indices, start_indices[1:]))
     ranges.append((start_indices[-1], len(entries)))
     return ranges
 
-def load_pipeline_states(file_prefix, all_uuid_list, continue_on_error):
+def load_pipeline_states(file_prefix, all_uuid_list, continue_on_error, verbose):
     import emission.core.get_database as edb
     import pymongo
 
@@ -70,7 +70,7 @@ def load_pipeline_states(file_prefix, all_uuid_list, continue_on_error):
             (curr_uuid, pipeline_filename))
         with gzip.open(pipeline_filename) as gfd:
             states = json.load(gfd, object_hook = esj.wrapped_object_hook)
-            if args.verbose:
+            if verbose:
                 logging.debug("Loading states of length %s" % len(states))
             if len(states) > 0:
                 try:
@@ -109,6 +109,55 @@ def post_check(unique_user_list, all_rerun_list):
     else:
         logging.info("timeline contains a mixture of analysis results and raw data - complain to shankari!")
 
+def load_multi_timeline_for_range(file_prefix, info_only=None, verbose=None, continue_on_error=None, mapfile=None, prefix=None, batch_size=10000, raw_timeseries_only=False):
+    fn = file_prefix
+    logging.info("Loading file or prefix %s" % fn)
+    sel_file_list = common.read_files_with_prefix(fn)
+
+    all_user_list = []
+    all_rerun_list = []
+    (tsdb_count, ucdb_count) = (0,0)
+
+    for i, filename in enumerate(sel_file_list):
+        if "pipelinestate" in filename:
+            continue
+        logging.info("=" * 50)
+        logging.info("Loading data from file %s" % filename)
+
+        entries = json.load(gzip.open(filename), object_hook = esj.wrapped_object_hook)
+
+        # Obtain uuid and rerun information from entries
+        curr_uuid_list, needs_rerun = common.analyse_timeline(entries)
+        if len(curr_uuid_list) > 1:
+            logging.warning("Found %d users, %s in filename, aborting! " % 
+                (len(curr_uuid_list), curr_uuid_list))
+            raise RuntimeException("Found %d users, %s in filename, expecting 1, %s" %
+                (len(curr_uuid_list), curr_uuid_list, common.split_user_id(filename)))
+        curr_uuid = curr_uuid_list[0]
+        all_user_list.append(curr_uuid)
+        all_rerun_list.append(needs_rerun)
+
+        load_ranges = get_load_ranges(entries, batch_size)
+        if not info_only:
+            for j, curr_range in enumerate(load_ranges):
+                if verbose is not None and j % verbose == 0:
+                    logging.info("About to load range %s -> %s" % (curr_range[0], curr_range[1]))
+                wrapped_entries = [ecwe.Entry(e) for e in entries[curr_range[0]:curr_range[1]]]
+                (tsdb_count, ucdb_count) = estcs.insert_entries(curr_uuid, wrapped_entries, continue_on_error)
+        logging.debug("For uuid %s, finished loading %d entries into the usercache and %d entries into the timeseries" % (curr_uuid, ucdb_count, tsdb_count))
+
+    unique_user_list = set(all_user_list)
+    if not info_only:
+        if not raw_timeseries_only:
+            load_pipeline_states(file_prefix, unique_user_list, continue_on_error, verbose)
+        if mapfile is not None:
+            register_mapped_users(mapfile, unique_user_list, verbose)
+        elif prefix is not None:
+            register_fake_users(prefix, unique_user_list, verbose)
+
+    post_check(unique_user_list, all_rerun_list) 
+    return (tsdb_count, ucdb_count)
+
 if __name__ == '__main__':
     parser = argparse.ArgumentParser()
     parser.add_argument("file_prefix",
@@ -128,6 +177,9 @@ def post_check(unique_user_list, all_rerun_list):
 
     parser.add_argument("-s", "--batch-size", default=10000, type=int,
         help="batch size to use for the entries")
+
+    parser.add_argument("-t", "--raw-timeseries-only", default=False, action='store_true',
+        help="load only raw timeseries data; if not set load both raw and analysis timeseries data")
 
     group = parser.add_mutually_exclusive_group(required=False)
     group.add_argument("-p", "--prefix", default="user",
@@ -141,47 +193,5 @@ def post_check(unique_user_list, all_rerun_list):
     else:
         logging.basicConfig(level=logging.INFO)
 
-    fn = args.file_prefix
-    logging.info("Loading file or prefix %s" % fn)
-    sel_file_list = common.read_files_with_prefix(fn)
-
-    all_user_list = []
-    all_rerun_list = []
-
-    for i, filename in enumerate(sel_file_list):
-        if "pipelinestate" in filename:
-            continue
-        logging.info("=" * 50)
-        logging.info("Loading data from file %s" % filename)
-
-        entries = json.load(gzip.open(filename), object_hook = esj.wrapped_object_hook)
-
-        # Obtain uuid and rerun information from entries
-        curr_uuid_list, needs_rerun = common.analyse_timeline(entries)
-        if len(curr_uuid_list) > 1:
-            logging.warning("Found %d users, %s in filename, aborting! " % 
-                (len(curr_uuid_list), curr_uuid_list))
-            raise RuntimeException("Found %d users, %s in filename, expecting 1, %s" %
-                (len(curr_uuid_list), curr_uuid_list, common.split_user_id(filename)))
-        curr_uuid = curr_uuid_list[0]
-        all_user_list.append(curr_uuid)
-        all_rerun_list.append(needs_rerun)
-
-        load_ranges = get_load_ranges(entries)
-        if not args.info_only:
-            for j, curr_range in enumerate(load_ranges):
-                if args.verbose is not None and j % args.verbose == 0:
-                    logging.info("About to load range %s -> %s" % (curr_range[0], curr_range[1]))
-                wrapped_entries = [ecwe.Entry(e) for e in entries[curr_range[0]:curr_range[1]]]
-                (tsdb_count, ucdb_count) = estcs.insert_entries(curr_uuid, wrapped_entries, args.continue_on_error)
-        print("For uuid %s, finished loading %d entries into the usercache and %d entries into the timeseries" % (curr_uuid, ucdb_count, tsdb_count))
-
-    unique_user_list = set(all_user_list)
-    if not args.info_only:
-        load_pipeline_states(args.file_prefix, unique_user_list, args.continue_on_error)
-        if args.mapfile is not None:
-            register_mapped_users(args.mapfile, unique_user_list)
-        elif args.prefix is not None:
-            register_fake_users(args.prefix, unique_user_list)
-
-    post_check(unique_user_list, all_rerun_list) 
+    # load_multi_timeline_for_range(args.file_prefix, args.info_only, args.verbose, args.continue_on_error, args.mapfile, args.prefix, args.batch_size)
+    load_multi_timeline_for_range(args.file_prefix, args.info_only, args.verbose, args.continue_on_error, args.mapfile, args.prefix, args.batch_size, args.raw_timeseries_only)

conf/log/purge.conf.sample

@@ -0,0 +1,43 @@
+{
+    "handlers": {
+        "errors": {
+            "backupCount": 2,
+            "mode": "a",
+            "level": "ERROR",
+            "formatter": "detailed",
+            "class": "logging.handlers.RotatingFileHandler",
+            "maxBytes": 1073741824,
+            "filename": "/var/tmp/purge-errors.log",
+            "encoding": "UTF-8"
+        },
+        "console": {
+            "class": "logging.StreamHandler",
+            "level": "WARNING"
+        },
+        "file": {
+            "backupCount": 8,
+            "filename": "/var/tmp/purge.log",
+            "maxBytes": 1073741824,
+            "mode": "a",
+            "formatter": "detailed",
+            "class": "logging.handlers.RotatingFileHandler",
+            "encoding": "UTF-8"
+        }
+    },
+    "version": 1,
+    "root": {
+        "handlers": [
+            "console",
+            "file",
+            "errors"
+        ],
+        "level": "DEBUG"
+    },
+    "formatters": {
+        "detailed": {
+            "class": "logging.Formatter",
+            "format": "%(asctime)s:%(levelname)s:%(thread)d:%(message)s"
+        }
+    }
+}
+

conf/log/restore.conf.sample

@@ -0,0 +1,43 @@
+{
+    "handlers": {
+        "errors": {
+            "backupCount": 2,
+            "mode": "a",
+            "level": "ERROR",
+            "formatter": "detailed",
+            "class": "logging.handlers.RotatingFileHandler",
+            "maxBytes": 1073741824,
+            "filename": "/var/tmp/restore-errors.log",
+            "encoding": "UTF-8"
+        },
+        "console": {
+            "class": "logging.StreamHandler",
+            "level": "WARNING"
+        },
+        "file": {
+            "backupCount": 8,
+            "filename": "/var/tmp/restore.log",
+            "maxBytes": 1073741824,
+            "mode": "a",
+            "formatter": "detailed",
+            "class": "logging.handlers.RotatingFileHandler",
+            "encoding": "UTF-8"
+        }
+    },
+    "version": 1,
+    "root": {
+        "handlers": [
+            "console",
+            "file",
+            "errors"
+        ],
+        "level": "DEBUG"
+    },
+    "formatters": {
+        "detailed": {
+            "class": "logging.Formatter",
+            "format": "%(asctime)s:%(levelname)s:%(thread)d:%(message)s"
+        }
+    }
+}
+

emission/core/wrapper/pipelinestate.py

@@ -28,7 +28,9 @@ class PipelineStages(enum.Enum):
     USER_MODEL = 7
     RECOMMENDATION = 8
     OUTPUT_GEN = 9
-    EXPORT_DATA = 15
+    EXPORT_DATA = 18
+    PURGE_TIMESERIES_DATA = 19
+    RESTORE_TIMESERIES_DATA = 20
 
 class PipelineState(ecwb.WrapperBase):
     props = {"pipeline_stage": ecwb.WrapperBase.Access.RW,  # the value of the stage from the enum above