Feat/elasticsearch flatten docs

hpcbench/campaign.py

@@ -328,6 +328,7 @@ def _expand_tags(self):
         self.network.tags = expanded
 
 
+@listify(wrapper=set)
 def get_benchmark_types(campaign):
     """Get of benchmarks referenced in the configuration
 

hpcbench/export/es.py

@@ -10,7 +10,6 @@
 
 from hpcbench.campaign import (
     from_file,
-    get_benchmark_types,
     get_metrics,
     ReportNode,
 )
@@ -26,28 +25,11 @@ class ESExporter(object):
         bool: 'boolean',
         float: 'float',
         int: 'long',
-        six.text_type: 'text',
+        six.text_type: 'keyword',
         str: 'keyword',
     }
     PROPERTIES_FIELD_TYPE = dict(date='date')
-
-    COMMON_INDEX_MAPPING = dict(
-        benchmark=dict(
-            type='text',
-        ),
-        category=dict(
-            type='text',
-        ),
-        date=dict(
-            type='date',
-        ),
-        elapsed=dict(
-            type='float',
-        ),
-        exit_status=dict(
-            type='short'
-        ),
-    )
+    ES_DOC_TYPE = 'hpcbench_metric'
 
     def __init__(self, path, hosts):
         """
@@ -126,45 +108,42 @@ def _push_data(self):
     def index_mapping(self):
         """Get Elasticsearch index mapping
         """
-        fields = dict(
-            (doc_type, self._get_document_mapping(doc_type)) for
-            doc_type in self._document_types
-        )
-        return fields
+        return self._get_document_mapping
 
     @property
     def _documents(self):
-        for run in self._get_runs():
+        for run in self._runs:
             yield dict(
                 index=dict(
-                    _type=run['benchmark'],
-                    _id=run['id']
+                    _type=self.ES_DOC_TYPE,
+                    _id=self.document_id(run)
                 )
             )
             run['campaign_id'] = self.campaign.campaign_id
             yield run
 
-    @cached_property
-    def _document_types(self):
-        return [
-            benchmark
-            for benchmark in get_benchmark_types(self.campaign)
-        ]
+    def document_id(self, doc):
+        return doc['id'] + '/' + str(hash(frozenset(doc['context'].items())))
 
-    def _get_document_mapping(self, benchmark):
+    @property
+    def _get_document_mapping(self):
         fields = {}
-        for run in self._get_benchmark_runs(benchmark):
-            dict_merge(fields, ESExporter._get_dict_mapping(benchmark, run))
+        for run in self._runs:
+            dict_merge(
+                fields,
+                ESExporter._get_dict_mapping(self.ES_DOC_TYPE, run)
+            )
         return fields
 
     @classmethod
-    def _get_dict_mapping(cls, prop, data):
+    def _get_dict_mapping(cls, prop, data, root=None):
         mapping = {}
         for name, value in data.items():
             if isinstance(value, (dict, collections.Mapping)):
-                dict_merge(mapping, cls._get_dict_mapping(name, value))
+                ctx = (root or tuple()) + (name,)
+                dict_merge(mapping, cls._get_dict_mapping(name, value, ctx))
             else:
-                dict_merge(mapping, cls._get_field_mapping(name, value))
+                dict_merge(mapping, cls._get_field_mapping(name, value, root))
         return {
             prop: {
                 'properties': mapping
@@ -181,7 +160,7 @@ def _get_field_type(cls, value):
         return type(value)
 
     @classmethod
-    def _get_field_mapping(cls, name, value):
+    def _get_field_mapping(cls, name, value, root):
         extra_params = {}
         field_type = cls.PROPERTIES_FIELD_TYPE.get(name)
         if field_type is None:
@@ -192,6 +171,8 @@ def _get_field_mapping(cls, name, value):
                 extra_params = cls._get_dict_mapping(None, value)[None]
                 extra_params['dynamic'] = False
                 field_type = 'nested'
+            elif root == ('metas',) and isinstance(value, six.string_types):
+                field_type = 'text'
             else:
                 field_type = cls.PY_TYPE_TO_ES_FIELD_TYPE[obj_type]
         mapping = {
@@ -202,19 +183,13 @@ def _get_field_mapping(cls, name, value):
         mapping[name].update(extra_params)
         return mapping
 
-    def _get_runs(self):
-        for attrs, metrics in get_metrics(self.campaign, self.report):
-            for run in metrics:
-                eax = dict()
-                eax.update(attrs)
-                eax.update(run)
-                yield eax
-
-    def _get_benchmark_runs(self, benchmark):
-        for attrs, metrics in get_metrics(self.campaign, self.report):
-            for run in metrics:
-                if run['benchmark'] == benchmark:
-                    eax = dict()
-                    eax.update(attrs)
+    @property
+    def _runs(self):
+        for attrs, runs in get_metrics(self.campaign, self.report):
+            for run in runs:
+                metrics = run.pop('metrics')
+                for metric in metrics:
+                    eax = dict(metric)
                     eax.update(run)
+                    eax.update(attrs)
                     yield eax