Repository: elastic/rally

Issue:
operation-type:composite fails when track_total_hits is false
We are using a composite operation to benchmark searchable snapshots using async search like below:

```json
{
  "operation": {
    "operation-type": "composite",
    "name": "date_histogram_by_day-90d-hotfrozen",
    "iterations": 10,
    "requests": [
      {
        "stream": [
          {
            "name": "async-search",
            "operation-type": "submit-async-search",
            "index": "hotfrozenlogs",
            "request-params": {
              "track_total_hits": "false"
            },
            "body": {
              "size": 0,
              "query": {
                "bool": {
                  "filter": [
                    {
                      "range": {
                        "@timestamp": {
                          "gte": "2024-01-01T00:00:00",
                          "lt": "2024-04-01T00:00:00"
                        }
                      }
                    },
                    {
                      "term": {
                        "log.file.path": {
                          "value": "/var/log/messages/salttouch"
                        }
                      }
                    }
                  ]
                }
              },
              "aggs": {
                "time": {
                  "date_histogram": {
                    "field": "@timestamp",
                    "calendar_interval": "1d"
                  }
                }
              }
            }
          }
        ]
      },
      {
        "operation-type": "get-async-search",
        "retrieve-results-for": [
          "async-search"
        ]
      },
      {
        "operation-type": "delete-async-search",
        "delete-results-for": [
          "async-search"
        ]
      }
    ]
  }
}
```
And its failing with:

`esrally.exceptions.RallyError: Cannot run task [date_histogram_by_day-90d-hotfrozen]: Cannot execute [user-defined context-manager enabled runner for [composite]]. Provided parameters are: ['operation-type', 'name', 'iterations', 'requests', 'include-in-reporting']. Error: ['total'].`

The log shows GetAsyncSearch is trying to read the `total` in: https://github.com/elastic/rally/blob/543b4dcf6d602db2e9ef37ddf134a7833b9f530f/esrally/driver/runner.py#L2530

```
File "/usr/local/lib/python3.8/site-packages/esrally/driver/runner.py", line 2530, in __call__
  "hits": response["response"]["hits"]["total"]["value"],
KeyError: 'total'
```

Hints from issue comments:
@ugosan: We would need a flag to check wether `track_total_hits` is present in request-params or in the body (where it can be also defined) and read the total accordingly.

```js
{
    "name": "async-search",
    "operation-type": "submit-async-search",
    "index": "hotfrozenlogs",
    "request-params": {
        "track_total_hits": false //here
    },
    "body": {
        "track_total_hits": false //or here
...
```

@ugosan: Here is an agnostic (to track_total_hits) approach that works just by checking if `total` is in `response["response"]["hits"]`, by splitting the stats[search] in two steps.

```py
class GetAsyncSearch(Runner):
    async def __call__(self, es, params):
        success = True
        searches = mandatory(params, "retrieve-results-for", self)
        request_params = params.get("request-params", {})
        stats = {}
        for search_id, search in async_search_ids(searches):
            response = await es.async_search.get(id=search_id, params=request_params)
            is_running = response["is_running"]
            success = success and not is_running
            if not is_running:
                
                stats[search] = {
                    "timed_out": response["response"]["timed_out"],
                    "took": response["response"]["took"],
                }
                
                if "total" in response["response"]["hits"].keys():
                    stats[search]["hits"] = response["response"]["hits"]["total"]["value"]
                    stats[search]["hits_relation"] = response["response"]["hits"]["total"]["relation"]
```

Another option could be to check in the `body` and `request_params` for `track_total_hits` inside `SubmitAsyncSearch` and put `track_total_hits` inside `CompositeContext` for checking in `GetAsyncSearch` - more verbose but also much more clear. 
```py

class SubmitAsyncSearch(Runner):
    async def __call__(self, es, params):
        ...
        #track_total_hits is true by default
        CompositeContext.put("track_total_hits", True)
        
        if "track_total_hits" in params.get("body").keys():
            CompositeContext.put("track_total_hits", bool(eval(params.get("body").get("track_total_hits").capitalize())))
            
        if "track_total_hits" in request_params.keys():
            CompositeContext.put("track_total_hits", bool(eval(request_params.get("track_total_hits").capitalize())))

```
 
```py
class GetAsyncSearch(Runner):
...
                if CompositeContext.get("track_total_hits"):
                    stats[search]["hits"] = response["response"]["hits"]["total"]["value"]
                    stats[search]["hits_relation"] = response["response"]["hits"]["total"]["relation"]
```


@gareth-ellis what do you think?@gareth-ellis: I like the agnostic change - it protects us from crashing out if some other parameter were to change the output behaviour too - presuming everything keeps working as expected, that is@ugosan: It looks like the tests are also checking for total, I will fix those too.

Solution patch:
diff --git a/catalog-info.yaml b/catalog-info.yaml
index d5c7c9e1b..8094a47bc 100644
--- a/catalog-info.yaml
+++ b/catalog-info.yaml
@@ -31,8 +31,8 @@ spec:
 
 
 # Declare Rally's Buildkite IT pipeline.
-# yaml-language-server: $schema=https://gist.githubusercontent.com/elasticmachine/988b80dae436cafea07d9a4a460a011d/raw/e57ee3bed7a6f73077a3f55a38e76e40ec87a7cf/rre.schema.json
 ---
+# yaml-language-server: $schema=https://gist.githubusercontent.com/elasticmachine/988b80dae436cafea07d9a4a460a011d/raw/rre.schema.json
 apiVersion: backstage.io/v1alpha1
 kind: Resource
 metadata:
@@ -54,6 +54,9 @@ spec:
       name: Rally - IT
       description:  Macrobenchmarking framework for Elasticsearch
     spec:
+      env:
+        ELASTIC_SLACK_NOTIFICATIONS_ENABLED: 'true'
+        SLACK_NOTIFICATIONS_CHANNEL: '#es-perf-build'
       pipeline_file: .buildkite/it/pipeline.yml
       repository: elastic/rally
       schedules:
@@ -66,9 +69,10 @@ spec:
         everyone:
           access_level: READ_ONLY
 
-# Declare Rally's Buildkite IT pipeline.
-# yaml-language-server: $schema=https://gist.githubusercontent.com/elasticmachine/988b80dae436cafea07d9a4a460a011d/raw/e57ee3bed7a6f73077a3f55a38e76e40ec87a7cf/rre.schema.json
+
+# Declare Rally's Buildkite IT pipeline for serverless.
 ---
+# yaml-language-server: $schema=https://gist.githubusercontent.com/elasticmachine/988b80dae436cafea07d9a4a460a011d/raw/rre.schema.json
 apiVersion: backstage.io/v1alpha1
 kind: Resource
 metadata:
@@ -90,6 +94,9 @@ spec:
       name: Rally - IT Serverless
       description:  Macrobenchmarking framework for Elasticsearch
     spec:
+      env:
+        ELASTIC_SLACK_NOTIFICATIONS_ENABLED: 'true'
+        SLACK_NOTIFICATIONS_CHANNEL: '#es-perf-build'
       pipeline_file: .buildkite/it/serverless-pipeline.yml
       repository: elastic/rally
       schedules:
@@ -102,9 +109,10 @@ spec:
         everyone:
           access_level: READ_ONLY
 
+
 # Declare Rally Release Docker pipeline.
-# yaml-language-server: $schema=https://gist.githubusercontent.com/elasticmachine/988b80dae436cafea07d9a4a460a011d/raw/e57ee3bed7a6f73077a3f55a38e76e40ec87a7cf/rre.schema.json
 ---
+# yaml-language-server: $schema=https://gist.githubusercontent.com/elasticmachine/988b80dae436cafea07d9a4a460a011d/raw/rre.schema.json
 apiVersion: backstage.io/v1alpha1
 kind: Resource
 metadata:
@@ -126,6 +134,9 @@ spec:
       name: Rally - Release Docker
       description: Release Docker Artifacts for Rally
     spec:
+      env:
+        ELASTIC_SLACK_NOTIFICATIONS_ENABLED: 'true'
+        SLACK_NOTIFICATIONS_CHANNEL: '#es-perf-build'
       pipeline_file: .buildkite/release-docker/pipeline.yml
       provider_settings:
         trigger_mode: none
@@ -133,9 +144,10 @@ spec:
       teams:
         es-perf: {}
 
+
 # Declare Rally Dev Docker build pipeline.
-# yaml-language-server: $schema=https://gist.githubusercontent.com/elasticmachine/988b80dae436cafea07d9a4a460a011d/raw/e57ee3bed7a6f73077a3f55a38e76e40ec87a7cf/rre.schema.json
 ---
+# yaml-language-server: $schema=https://gist.githubusercontent.com/elasticmachine/988b80dae436cafea07d9a4a460a011d/raw/rre.schema.json
 apiVersion: backstage.io/v1alpha1
 kind: Resource
 metadata:
@@ -156,6 +168,9 @@ spec:
       name: Rally - Dev Docker Build
       description: Build Dev Docker Artifacts for Rally
     spec:
+      env:
+        ELASTIC_SLACK_NOTIFICATIONS_ENABLED: 'true'
+        SLACK_NOTIFICATIONS_CHANNEL: '#es-perf-build'
       pipeline_file: .buildkite/dev-docker/pipeline.yml
       provider_settings:
         trigger_mode: none
diff --git a/docs/advanced.rst b/docs/advanced.rst
index 1960affae..b1d9fcd3f 100644
--- a/docs/advanced.rst
+++ b/docs/advanced.rst
@@ -6,7 +6,7 @@ Advanced Topics
 Templating Track Files with Jinja2
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 
-Rally uses `Jinja2 <http://jinja.pocoo.org/docs/dev/>`_ as a template language so you can use Jinja2 expressions in track files.
+Rally uses `Jinja2 <https://jinja.palletsprojects.com/en/latest/>`_ as a template language so you can use Jinja2 expressions in track files.
 
 Elasticsearch utilizes Mustache formatting in a few places, notably in `search templates <https://www.elastic.co/guide/en/elasticsearch/reference/7.4/search-template.html>`_ and `Watcher templates <https://www.elastic.co/guide/en/elasticsearch/reference/7.4/actions-email.html>`_. If you are using Mustache in your Rally tracks you must `escape them properly <https://jinja.palletsprojects.com/en/2.10.x/templates/#escaping>`_. See :ref:`put_pipeline` for an example.
 
@@ -17,8 +17,8 @@ Extensions
 
 Rally also provides a few extensions to Jinja2:
 
-* ``now``: a global variable that represents the current date and time when the template is evaluated by Rally.
-* ``days_ago()``: a `filter <http://jinja.pocoo.org/docs/dev/templates/#filters>`_ that you can use for date calculations.
+* ``now``: a global variable that represents the current date and time in seconds since the epoch (Jan 1, 1970 00:00:00 UTC) when the template is evaluated by Rally.
+* ``days_ago()``: a `filter <https://jinja.palletsprojects.com/en/latest/templates/#filters>`_ that you can use for date calculations.
 
 You can find an example in the ``http_logs`` track::
 
@@ -90,6 +90,45 @@ The parameter ``default_value`` controls the value to use for the setting if it
 * ``build_flavor``: a global variable that holds build flavor reported by Elasticsearch cluster targetted by Rally.
 * ``serverless_operator`` (reserved for Elastic internal use): a global variable that holds ``true`` when Rally targets serverless cluster with operator privileges, and ``false`` otherwise.
 
+Dynamic Date Ranges
+~~~~~~~~~~~~~~~~~~~
+
+* ``get_start_date('%Y-%m-%d', 'timezone.utc')``: pair with the ``now`` global variable to generate a start date in the specified date format and time zone. The default format is ``%Y-%m-%d`` and time zone is ``timezone.utc``.  The filter will use the default values when no parameters are specified.
+* ``get_end_date(1, '%Y-%m-%d', 'timezone.utc')``: pair with the ``now`` global variable to generate an end date with the specified number of days, date format, and time zone. The default number of days is 1, format is ``%Y-%m-%d``, and time zone is ``timezone.utc``. The filter will use the default values when no parameters are specified.
+
+Example 1::
+
+    {
+      ...,
+      "start-date": {{ now | get_start_date | tojson }},
+      "end-date": {{ now | get_end_date | tojson }}
+    }
+
+Results in::
+
+    {
+      ...,
+      "start-date": "2021-01-01",
+      "end-date": "2021-01-02"
+    }
+
+Example 2::
+
+    {
+      ...,
+      "start-date": {{ now | get_start_date('%Y-%m-%d') | tojson }},
+      "end-date": {{ now | get_end_date(3, '%Y-%m-%d') | tojson }}
+    }
+
+Results in::
+
+    {
+      ...,
+      "start-date": "2021-01-01",
+      "end-date": "2021-01-04"
+    }
+
+
 .. _adding_tracks_custom_param_sources:
 
 Controlling Operation Parameters Using Custom Parameter Sources
diff --git a/esrally/driver/runner.py b/esrally/driver/runner.py
index 137bf4a4d..1b5018f05 100644
--- a/esrally/driver/runner.py
+++ b/esrally/driver/runner.py
@@ -2495,10 +2495,14 @@ def __repr__(self, *args, **kwargs):
 class SubmitAsyncSearch(Runner):
     async def __call__(self, es, params):
         request_params = params.get("request-params", {})
+
+        # defaults wait_for_completion_timeout = 0 to avoid sync fallback for fast searches
+        if "wait_for_completion_timeout" not in request_params:
+            request_params["wait_for_completion_timeout"] = 0
+
         response = await es.async_search.submit(body=mandatory(params, "body", self), index=params.get("index"), params=request_params)
 
         op_name = mandatory(params, "name", self)
-        # id may be None if the operation has already returned
         search_id = response.get("id")
         CompositeContext.put(op_name, search_id)
 
@@ -2510,7 +2514,6 @@ def async_search_ids(op_names):
     subjects = [op_names] if isinstance(op_names, str) else op_names
     for subject in subjects:
         subject_id = CompositeContext.get(subject)
-        # skip empty ids, searches have already completed
         if subject_id:
             yield subject_id, subject
 
@@ -2527,12 +2530,14 @@ async def __call__(self, es, params):
             success = success and not is_running
             if not is_running:
                 stats[search] = {
-                    "hits": response["response"]["hits"]["total"]["value"],
-                    "hits_relation": response["response"]["hits"]["total"]["relation"],
                     "timed_out": response["response"]["timed_out"],
                     "took": response["response"]["took"],
                 }
 
+                if "total" in response["response"]["hits"].keys():
+                    stats[search]["hits"] = response["response"]["hits"]["total"]["value"]
+                    stats[search]["hits_relation"] = response["response"]["hits"]["total"]["relation"]
+
         return {
             # only count completed searches - there is one key per search id in `stats`
             "weight": len(stats),
diff --git a/esrally/time.py b/esrally/time.py
index ad6e58316..17cc9b778 100644
--- a/esrally/time.py
+++ b/esrally/time.py
@@ -16,7 +16,7 @@
 # under the License.
 
 import time
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 
 
 def to_epoch_millis(t):
@@ -84,6 +84,33 @@ def days_ago(start_date, end_date, date_format="%d-%m-%Y", default_tz=timezone.u
     return (end - start).days
 
 
+def get_start_date(the_date, date_format="%Y-%m-%d", default_tz=timezone.utc):
+    """
+    Return a formatted date string for a given date.
+
+    :param the_date: The date to format. May be a datetime instance, a unix timestamp or a string representation of a date.
+    :param date_format: The format to use for the date. Default: "%Y-%m-%d"
+    :param default_tz: Timezone to use for dates using unix timestamp. Default: timezone.UTC
+    :return: Returns a formatted date string.
+    """
+    start_date = _to_datetime(the_date, default_tz)
+    return start_date.strftime(date_format)
+
+
+def get_end_date(start_date, duration_days=1, date_format="%Y-%m-%d", default_tz=timezone.utc):
+    """
+    Return a formatted date string for a given date plus a duration.
+
+    :param start_date: The date to start from. May be a datetime instance, a unix timestamp or a string representation of a date.
+    :param duration_days: The duration in days to add to the start date. Default: 1
+    :param date_format: The format to use for the date. Default: "%Y-%m-%d"
+    :param default_tz: Timezone to use for dates using unix timestamp. Default: timezone.UTC
+    :return: Returns a formatted date string.
+    """
+    end_date = _to_datetime(start_date, default_tz) + timedelta(days=duration_days)
+    return end_date.strftime(date_format)
+
+
 class Clock:
     """
     A clock abstracts time measurements. Its main purpose is to ease testing
diff --git a/esrally/track/loader.py b/esrally/track/loader.py
index 6c6f947e1..f20709f1c 100644
--- a/esrally/track/loader.py
+++ b/esrally/track/loader.py
@@ -757,6 +757,8 @@ def default_internal_template_vars(glob_helper=lambda f: [], clock=time.Clock, b
         },
         "filters": {
             "days_ago": time.days_ago,
+            "get_start_date": time.get_start_date,
+            "get_end_date": time.get_end_date,
         },
     }
 


Test patch:
diff --git a/tests/driver/runner_test.py b/tests/driver/runner_test.py
index 561404752..22d60ddad 100644
--- a/tests/driver/runner_test.py
+++ b/tests/driver/runner_test.py
@@ -6119,7 +6119,9 @@ async def test_submit_async_search(self, es):
             # search id is registered in context
             assert runner.CompositeContext.get("search-1") == "12345"
 
-        es.async_search.submit.assert_awaited_once_with(body={"query": {"match_all": {}}}, index="_all", params={})
+        es.async_search.submit.assert_awaited_once_with(
+            body={"query": {"match_all": {}}}, index="_all", params={"wait_for_completion_timeout": 0}
+        )
 
 
 class TestGetAsyncSearch:
diff --git a/tests/track/loader_test.py b/tests/track/loader_test.py
index 56cc65911..e60271528 100644
--- a/tests/track/loader_test.py
+++ b/tests/track/loader_test.py
@@ -974,7 +974,9 @@ def test_render_simple_template(self):
             "key": {{'01-01-2000' | days_ago(now)}},
             "key2": "static value",
             "key3": "{{build_flavor}}",
-            "key4": {{serverless_operator}}
+            "key4": {{serverless_operator}},
+            "key5": {{now | get_start_date('%Y-%m-%d') | tojson}},
+            "key6": {{now | get_end_date(1, '%Y-%m-%d') | tojson}}
         }
         """
 
@@ -985,7 +987,9 @@ def test_render_simple_template(self):
             "key": 5864,
             "key2": "static value",
             "key3": "test",
-            "key4": False
+            "key4": False,
+            "key5": "2016-01-21",
+            "key6": "2016-01-22"
         }
         """
         assert rendered == expected