Introduce field-caps operation-type

docs/track.rst

@@ -2533,12 +2533,16 @@ Properties
 
 The ``composite`` operation only supports the following operation-types:
 
+* ``open-point-in-time``
+* ``close-point-in-time``
+* ``search``
+* ``paginated-search``
 * ``raw-request``
 * ``sleep``
-* ``search``
 * ``submit-async-search``
 * ``get-async-search``
 * ``delete-async-search``
+* ``field-caps``
 
 **Examples**
 
@@ -2964,6 +2968,41 @@ The following meta data is always returned:
 * ``weight``: The number of fetched pages. Should always equal to the ``pages`` parameter.
 * ``unit``: The unit in which to interpret ``weight``. Always "ops".
 
+field-caps
+~~~~~~~~~~~~~~~~~~~
+
+Retrieve `the capabilities of fields among indices <https://www.elastic.co/guide/en/elasticsearch/reference/current/search-field-caps.html>`_.
+
+Properties
+""""""""""
+
+* ``index`` (optional, defaults to ``_all``): An `index pattern <https://www.elastic.co/guide/en/elasticsearch/reference/current/multi-index.html>`_ that defines which indices should be targeted by this operation.
+* ``fields`` (optional, default to ``*``): Comma-separated list of fields to retrieve capabilities for.
+* ``index_filter`` (optional): An index_filter to limit this operation to target only indices that match this index_filter
+
+**Example**
+
+Retrieve the capabilities of fields starting with ``message`` of the ``log-*`` indices::
+
+    {
+      "operation": {
+        "operation-type": "field-caps",
+        "name": "field-caps-logs-indices",
+        "index": "log-*",
+        "fields": "message*"
+      },
+      "target-throughput": 10,
+      "clients": 2,
+      "warmup-iterations": 50,
+      "iterations": 100
+    }
+
+
+Meta-data
+"""""""""
+
+The operation returns no meta-data.
+
 
 .. _track_dependencies:
 

esrally/driver/runner.py

@@ -57,6 +57,7 @@ def register_default_runners():
     register_runner(track.OperationType.OpenPointInTime, OpenPointInTime(), async_runner=True)
     register_runner(track.OperationType.ClosePointInTime, ClosePointInTime(), async_runner=True)
     register_runner(track.OperationType.Sql, Sql(), async_runner=True)
+    register_runner(track.OperationType.FieldCaps, FieldCaps(), async_runner=True)
 
     # This is an administrative operation but there is no need for a retry here as we don't issue a request
     register_runner(track.OperationType.Sleep, Sleep(), async_runner=True)
@@ -2343,6 +2344,7 @@ def __init__(self, *args, **kwargs):
             "submit-async-search",
             "get-async-search",
             "delete-async-search",
+            "field-caps",
         ]
 
     async def run_stream(self, es, stream, connection_limit):
@@ -2482,6 +2484,28 @@ def __repr__(self, *args, **kwargs):
         return "sql"
 
 
+class FieldCaps(Runner):
+    """
+    Retrieve `the capabilities of fields among indices.
+    <https://www.elastic.co/guide/en/elasticsearch/reference/current/search-field-caps.html>` _.
+    """
+
+    async def __call__(self, es, params):
+        index = params.get("index", "_all")
+        fields = params.get("fields", "*")
+        body = params.get("body", {})
+        index_filter = params.get("index_filter")
+        if index_filter:
+            body["index_filter"] = index_filter
+        request_params = params.get("request-params")
+        await es.field_caps(index=index, body=body, fields=fields, params=request_params)
+
+        return {"weight": 1, "unit": "ops", "success": True}
+
+    def __repr__(self, *args, **kwargs):
+        return "field-caps"
+
+
 class RequestTiming(Runner, Delegator):
     def __init__(self, delegate):
         super().__init__(delegate=delegate)

esrally/track/track.py

@@ -684,6 +684,7 @@ class OperationType(Enum):
     OpenPointInTime = 14
     ClosePointInTime = 15
     Sql = 16
+    FieldCaps = 17
 
     # administrative actions
     ForceMerge = 1001
@@ -843,6 +844,8 @@ def from_hyphenated_string(cls, v):
             return OperationType.DeleteIlmPolicy
         elif v == "sql":
             return OperationType.Sql
+        elif v == "field-caps":
+            return OperationType.FieldCaps
         else:
             raise KeyError(f"No enum value for [{v}]")
 

tests/driver/runner_test.py

@@ -5874,7 +5874,8 @@ async def test_rejects_unsupported_operations(self, es):
 
         assert exc.value.args[0] == (
             "Unsupported operation-type [bulk]. Use one of [open-point-in-time, close-point-in-time, "
-            "search, paginated-search, raw-request, sleep, submit-async-search, get-async-search, delete-async-search]."
+            "search, paginated-search, raw-request, sleep, submit-async-search, get-async-search, "
+            "delete-async-search, field-caps]."
         )
 
 
@@ -6227,3 +6228,27 @@ async def test_refresh_request_timeout(self, es):
         await refresh(es, params={"index": "_all", "request-timeout": 50000})
 
         es.indices.refresh.assert_awaited_once_with(index="_all", request_timeout=50000)
+
+
+class TestFieldCapsRunner:
+    @mock.patch("elasticsearch.Elasticsearch")
+    @run_async
+    async def test_field_caps_without_index_filter(self, es):
+        es.field_caps = mock.AsyncMock()
+        field_caps = runner.FieldCaps()
+        result = await field_caps(es, params={"index": "log-*"})
+        assert result == {"weight": 1, "unit": "ops", "success": True}
+
+        es.field_caps.assert_awaited_once_with(index="log-*", fields="*", body={}, params=None)
+
+    @mock.patch("elasticsearch.Elasticsearch")
+    @run_async
+    async def test_field_caps_with_index_filter(self, es):
+        es.field_caps = mock.AsyncMock()
+        field_caps = runner.FieldCaps()
+        index_filter = {"range": {"@timestamp": {"gte": "2022"}}}
+        result = await field_caps(es, params={"fields": "time-*", "index_filter": index_filter})
+        assert result == {"weight": 1, "unit": "ops", "success": True}
+
+        expected_body = {"index_filter": index_filter}
+        es.field_caps.assert_awaited_once_with(index="_all", fields="time-*", body=expected_body, params=None)