utils/test/dashboard/dashboard/mongo2elastic/main.py

   1 #! /usr/bin/env python
   2
   3 import datetime
   4 import json
   5 import os
   6 import subprocess
   7 import traceback
   8 import urlparse
   9 import uuid
  10
  11 import argparse
  12
  13 from common import logger_utils, elastic_access
  14 from conf import testcases
  15 from conf.config import APIConfig
  16 from mongo2elastic import format
  17
  18 logger = logger_utils.DashboardLogger('mongo2elastic').get
  19
  20 parser = argparse.ArgumentParser()
  21 parser.add_argument("-c", "--config-file",
  22                     dest='config_file',
  23                     help="Config file location")
  24 parser.add_argument('-ld', '--latest-days',
  25                     default=0,
  26                     type=int,
  27                     metavar='N',
  28                     help='get entries old at most N days from mongodb and'
  29                          ' parse those that are not already in elasticsearch.'
  30                          ' If not present, will get everything from mongodb, which is the default')
  31
  32 args = parser.parse_args()
  33 CONF = APIConfig().parse(args.config_file)
  34
  35
  36 tmp_docs_file = './mongo-{}.json'.format(uuid.uuid4())
  37
  38
  39 class DocumentPublisher:
  40
  41     def __init__(self, doc, fmt, exist_docs, creds, elastic_url):
  42         self.doc = doc
  43         self.fmt = fmt
  44         self.creds = creds
  45         self.exist_docs = exist_docs
  46         self.elastic_url = elastic_url
  47         self.is_formatted = True
  48
  49     def format(self):
  50         try:
  51             if self._verify_document() and self.fmt:
  52                 self.is_formatted = vars(format)[self.fmt](self.doc)
  53             else:
  54                 self.is_formatted = False
  55         except Exception:
  56             logger.error("Fail in format testcase[%s]\nerror message: %s" %
  57                          (self.doc, traceback.format_exc()))
  58             self.is_formatted = False
  59         finally:
  60             return self
  61
  62     def publish(self):
  63         if self.is_formatted and self.doc not in self.exist_docs:
  64             self._publish()
  65
  66     def _publish(self):
  67         status, data = elastic_access.publish_docs(self.elastic_url, self.creds, self.doc)
  68         if status > 300:
  69             logger.error('Publish record[{}] failed, due to [{}]'
  70                          .format(self.doc, json.loads(data)['error']['reason']))
  71
  72     def _fix_date(self, date_string):
  73         if isinstance(date_string, dict):
  74             return date_string['$date']
  75         else:
  76             return date_string[:-3].replace(' ', 'T') + 'Z'
  77
  78     def _verify_document(self):
  79         """
  80         Mandatory fields:
  81             installer
  82             pod_name
  83             version
  84             case_name
  85             date
  86             project
  87             details
  88
  89             these fields must be present and must NOT be None
  90
  91         Optional fields:
  92             description
  93
  94             these fields will be preserved if the are NOT None
  95         """
  96         mandatory_fields = ['installer',
  97                             'pod_name',
  98                             'version',
  99                             'case_name',
 100                             'project_name',
 101                             'details']
 102         mandatory_fields_to_modify = {'start_date': self._fix_date}
 103         fields_to_swap_or_add = {'scenario': 'version'}
 104         if '_id' in self.doc:
 105             mongo_id = self.doc['_id']
 106         else:
 107             mongo_id = None
 108         optional_fields = ['description']
 109         for key, value in self.doc.items():
 110             if key in mandatory_fields:
 111                 if value is None:
 112                     # empty mandatory field, invalid input
 113                     logger.info("Skipping testcase with mongo _id '{}' because the testcase was missing value"
 114                                 " for mandatory field '{}'".format(mongo_id, key))
 115                     return False
 116                 else:
 117                     mandatory_fields.remove(key)
 118             elif key in mandatory_fields_to_modify:
 119                 if value is None:
 120                     # empty mandatory field, invalid input
 121                     logger.info("Skipping testcase with mongo _id '{}' because the testcase was missing value"
 122                                 " for mandatory field '{}'".format(mongo_id, key))
 123                     return False
 124                 else:
 125                     self.doc[key] = mandatory_fields_to_modify[key](value)
 126                     del mandatory_fields_to_modify[key]
 127             elif key in fields_to_swap_or_add:
 128                 if value is None:
 129                     swapped_key = fields_to_swap_or_add[key]
 130                     swapped_value = self.doc[swapped_key]
 131                     logger.info("Swapping field '{}' with value None for '{}' with value '{}'.".format(key, swapped_key,
 132                                                                                                        swapped_value))
 133                     self.doc[key] = swapped_value
 134                     del fields_to_swap_or_add[key]
 135                 else:
 136                     del fields_to_swap_or_add[key]
 137             elif key in optional_fields:
 138                 if value is None:
 139                     # empty optional field, remove
 140                     del self.doc[key]
 141                 optional_fields.remove(key)
 142             else:
 143                 # unknown field
 144                 del self.doc[key]
 145
 146         if len(mandatory_fields) > 0:
 147             # some mandatory fields are missing
 148             logger.info("Skipping testcase with mongo _id '{}' because the testcase was missing"
 149                         " mandatory field(s) '{}'".format(mongo_id, mandatory_fields))
 150             return False
 151         elif len(mandatory_fields_to_modify) > 0:
 152             # some mandatory fields are missing
 153             logger.info("Skipping testcase with mongo _id '{}' because the testcase was missing"
 154                         " mandatory field(s) '{}'".format(mongo_id, mandatory_fields_to_modify.keys()))
 155             return False
 156         else:
 157             if len(fields_to_swap_or_add) > 0:
 158                 for key, swap_key in fields_to_swap_or_add.iteritems():
 159                     self.doc[key] = self.doc[swap_key]
 160
 161             return True
 162
 163
 164 class DocumentsPublisher:
 165
 166     def __init__(self, project, case, fmt, days, elastic_url, creds):
 167         self.project = project
 168         self.case = case
 169         self.fmt = fmt
 170         self.days = days
 171         self.elastic_url = elastic_url
 172         self.creds = creds
 173         self.existed_docs = []
 174
 175     def export(self):
 176         if self.days > 0:
 177             past_time = datetime.datetime.today() - datetime.timedelta(days=self.days)
 178             query = '''{{
 179                           "project_name": "{}",
 180                           "case_name": "{}",
 181                           "start_date": {{"$gt" : "{}"}}
 182                         }}'''.format(self.project, self.case, past_time)
 183         else:
 184             query = '''{{
 185                            "project_name": "{}",
 186                            "case_name": "{}"
 187                         }}'''.format(self.project, self.case)
 188         cmd = ['mongoexport',
 189                '--db', 'test_results_collection',
 190                '--collection', 'results',
 191                '--query', '{}'.format(query),
 192                '--out', '{}'.format(tmp_docs_file)]
 193         try:
 194             subprocess.check_call(cmd)
 195             return self
 196         except Exception, err:
 197             logger.error("export mongodb failed: %s" % err)
 198             self._remove()
 199             exit(-1)
 200
 201     def get_existed_docs(self):
 202         if self.days == 0:
 203             body = '''{{
 204                         "query": {{
 205                             "bool": {{
 206                                 "must": [
 207                                     {{ "match": {{ "project_name": "{}" }} }},
 208                                     {{ "match": {{ "case_name": "{}" }} }}
 209                                 ]
 210                             }}
 211                         }}
 212                     }}'''.format(self.project, self.case)
 213         elif self.days > 0:
 214             body = '''{{
 215                        "query": {{
 216                            "bool": {{
 217                                "must": [
 218                                    {{ "match": {{ "project_name": "{}" }} }},
 219                                    {{ "match": {{ "case_name": "{}" }} }}
 220                                ],
 221                                "filter": {{
 222                                    "range": {{
 223                                        "start_date": {{ "gte": "now-{}d" }}
 224                                    }}
 225                                }}
 226                            }}
 227                        }}
 228                    }}'''.format(self.project, self.case, self.days)
 229         else:
 230             raise Exception('Update days must be non-negative')
 231         self.existed_docs = elastic_access.get_docs(self.elastic_url, self.creds, body)
 232         return self
 233
 234     def publish(self):
 235         try:
 236             with open(tmp_docs_file) as fdocs:
 237                 for doc_line in fdocs:
 238                     DocumentPublisher(json.loads(doc_line),
 239                                       self.fmt,
 240                                       self.existed_docs,
 241                                       self.creds,
 242                                       self.elastic_url).format().publish()
 243         finally:
 244             fdocs.close()
 245             self._remove()
 246
 247     def _remove(self):
 248         if os.path.exists(tmp_docs_file):
 249             os.remove(tmp_docs_file)
 250
 251
 252 def main():
 253     base_elastic_url = urlparse.urljoin(CONF.es_url, '/test_results/mongo2elastic')
 254     days = args.latest_days
 255     es_creds = CONF.es_creds
 256
 257     for project, case_dicts in testcases.testcases_yaml.items():
 258         for case_dict in case_dicts:
 259             case = case_dict.get('name')
 260             fmt = testcases.compose_format(case_dict.get('format'))
 261             DocumentsPublisher(project,
 262                                case,
 263                                fmt,
 264                                days,
 265                                base_elastic_url,
 266                                es_creds).export().get_existed_docs().publish()