Stop generate a new field every time when a new result is being put (#3708)

* stop generate a new field every time when a new result is being put

The current elasticsearch backend generate a new field every time
a new result is being put. This is really inefficient for elasticsearch.

Switch to use the same field every time since the key is already stored in
elasticsearch's _id field.

* fix test cases for elasticsearch backend

celery/backends/elasticsearch.py

@@ -64,7 +64,7 @@ class ElasticsearchBackend(KeyValueStoreBackend):
             )
             try:
                 if res['found']:
-                    return res['_source'][key]
+                    return res['_source']['result']
             except (TypeError, KeyError):
                 pass
         except elasticsearch.exceptions.NotFoundError:
@@ -75,7 +75,7 @@ class ElasticsearchBackend(KeyValueStoreBackend):
             self._index(
                 id=key,
                 body={
-                    key: value,
+                    'result': value,
                     '@timestamp': '{0}Z'.format(
                         datetime.utcnow().isoformat()[:-3]
                     ),
@@ -103,7 +103,7 @@ class ElasticsearchBackend(KeyValueStoreBackend):
 
     def _get_server(self):
         """Connect to the Elasticsearch server."""
-        return elasticsearch.Elasticsearch(self.host)
+        return elasticsearch.Elasticsearch('%s:%s' % (self.host, self.port))
 
     @property
     def server(self):

t/unit/backends/test_elasticsearch.py

@@ -26,7 +26,7 @@ class test_ElasticsearchBackend:
         x._server = Mock()
         x._server.get = Mock()
         # expected result
-        r = dict(found=True, _source={sentinel.task_id: sentinel.result})
+        r = dict(found=True, _source={'result': sentinel.result})
         x._server.get.return_value = r
         dict_result = x.get(sentinel.task_id)