Oracle-ES consistency

Utils/Dataflow/069_upload2es/README

@@ -4,7 +4,7 @@
 
 Description
 -----------
-Uploads prepared data to ElasticSearch.
+load_data.sh uploads prepared data to ElasticSearch.
 
 Input
 -----
@@ -18,6 +18,24 @@ JSON documents, one per line:
 ...
 }}}
 
+Consistency
+-----------
+consistency.py checks that the data is present in ElasticSearch instead of
+uploading it. Input comes from Stage 009(in consistency mode) and only needs 2
+fields for now:
+{{{
+{taskid, task_timestamp}
+...
+}}}
+
+Consistency check can be run as following:
+
+  ./consistency.py --conf elasticsearch_config
+
+For more information about running the check and its arguments, use:
+
+  ./consistency.py -h
+
 TODO
 ----
 Make the stage aware of EOProcess/EOMessage markers

Utils/Dataflow/069_upload2es/consistency.py

@@ -1,6 +1,9 @@
 #!/bin/env python
 '''
-Script for consistency checking.
+Script for checking the supplied task's presence in elasticsearch.
+
+Currently it performs the check by comparing the supplied timestamp
+with the one in elasticsearch.
 
 Authors:
   Vasilii Aulov (vasilii.aulov@cern.ch)
@@ -15,7 +18,7 @@
 
 
 def log(msg, prefix='DEBUG'):
-    ''' ??? '''
+    ''' Add prefix and current time to message and write it to stderr. '''
     prefix = '(%s)' % (prefix)
     prefix = prefix.ljust(8)
     sys.stderr.write('%s%s %s\n' % (prefix, datetime.now().isoformat(), msg))
@@ -41,7 +44,13 @@ def log(msg, prefix='DEBUG'):
 
 
 def load_config(fname):
-    ''' ??? '''
+    ''' Open elasticsearch config and obtain parameters from it.
+
+    Setup INDEX as global variable.
+
+    :param fname: config file's name
+    :type fname: str
+    '''
     cfg = {
         'ES_HOST': 'localhost',
         'ES_PORT': '9200',
@@ -67,9 +76,10 @@ def load_config(fname):
 
 
 def es_connect(cfg):
-    ''' Establish a connection to elasticsearch.
+    ''' Establish a connection to elasticsearch, as a global variable.
 
-    TODO: take parameters from es config.
+    :param cfg: connection parameters
+    :type cfg: dict
     '''
     global es
     if cfg['ES_USER'] and cfg['ES_PASSWORD']:
@@ -110,6 +120,11 @@ def process(stage, message):
 
     Implementation of :py:meth:`.AbstractProcessorStage.process` for hooking
     the stage into DKB workflow.
+
+    :param stage: stage instance
+    :type stage: pyDKB.dataflow.stage.ProcessorStage
+    :param msg: input message with task info
+    :type msg: pyDKB.dataflow.Message
     '''
     data = message.content()
     if type(data) is not dict:
@@ -141,7 +156,11 @@ def process(stage, message):
 
 
 def main(args):
-    ''' ??? '''
+    ''' Parse command line arguments and run the stage.
+
+    :param argv: arguments
+    :type argv: list
+    '''
 
     stage = JSONProcessorStage()
     stage.add_argument('--conf', help='elasticsearch config', required=True)