How to

Task Management with the NEST Elasticsearch .NET Client

By
Stuart Cam

Schrödinger's ~~Cat~~ Task

"What is the cluster doing?" "Why is my search request taking so long?" "Can I cancel a search request?"

The answers to the above questions can all be found in the Task Management API – an Elasticsearch feature that moved from experimental to beta in version 6.0.

The Task Management API exposes a number of endpoints for retrieving the currently running tasks, how long they have been running, the nature of the tasks and the node(s) they are currently executing on.

Sometimes, as is the case with inefficiently designed search queries or indices, tasks can take quite a long time to execute; and in the pathological case may never complete within a reasonable time frame. Of course, there is no substitute for a good design in the first place, but being able to view and manage cluster tasks should be an important part of your ongoing cluster maintenance.

The interesting part, and the focus of this blog article, is how you can use the OpaqueId in the NEST .NET client (versions 6.2.0 and above) to help you trace requests originating from the client and to cancel them at a later date should you need to.

Under the Covers

The implementation uses a request header named X-OpaqueId to send the user specified value to the server. This header can be set on any REST action and is returned in the response header as well.

The Elasticsearch server does not enforce uniqueness on this value, so it is up to you as the user, to ensure that this value is sufficiently unique for your use case. In a multi-tenant cluster, or a cluster that is utilised by various different parts of your organisation, it might make sense to use a value that contains a description of the source application, department or even the user. This will help you in understanding the origination of tasks by simply looking at the value.

For example: SALES-DAVE-000001 or WEB-SEARCH-ABCDEF12345678

Of course, you are free to choose a strategy that is most appropriate for your use-case.

The Client

The .NET client exposes a string property named OpaqueId via the RequestConfiguration object. This property is available on all requests and as it implies, is available on all REST calls to the Elasticsearch server.

An example of how to use this property is given below:

var client = new ElasticClient();
var results = client.Search<Project>(new SearchRequest<Project>()
{
    RequestConfiguration = new RequestConfiguration
    {
        OpaqueId = "MyOpaqueId"
    },
    Scroll = "10m" // Create a scroll time of 10 minutes to keep this task around
});

Or by using the fluent syntax:

var results = client.Search<Project>(s => s
    .Scroll("10m")
    .RequestConfiguration(r => r
        .OpaqueId("MyOpaqueId")
    )
);

Either of these two requests will be dispatched to the Elasticsearch server with the following HTTP POST request:

POST http://127.0.0.1:9200/project/doc/_search?typed_keys=true&scroll=10m HTTP/1.1
Accept: application/json
Content-Type: application/json
X-Opaque-Id: MyOpaqueId
Host: 127.0.0.1:9200
Content-Length: 2

{}

Once set, this OpaqueId can used at a later date to locate the corresponding task(s) using the task management methods:

var tasks = client.ListTasks(o => o
    .RequestConfiguration(r => r
        .OpaqueId("MyOpaqueId")
    )
);

Giving us the following HTTP GET request:

GET http://127.0.0.1:9200/_tasks HTTP/1.1
Accept: application/json
Content-Type: application/json
X-Opaque-Id: MyOpaqueId
Host: 127.0.0.1:9200

and the following response:

HTTP/1.1 200 OK
X-Opaque-Id: MyOpaqueId
content-type: application/json; charset=UTF-8
content-length: 902

{
  "nodes": {
    "Bm_64yc0TrqBG9b0YzOK8Q": {
      "name": "xunit-node-b392a09200",
      "transport_address": "127.0.0.1:9300",
      "host": "127.0.0.1",
      "ip": "127.0.0.1:9300",
      "roles": [
        "master",
        "data",
        "ingest"
      ],
      "attributes": {
        "ml.machine_memory": "17010651136",
        "xpack.installed": "true",
        "ml.max_open_jobs": "20",
        "ml.enabled": "true",
        "gateway": "true"
      },
      "tasks": {
        "Bm_64yc0TrqBG9b0YzOK8Q:101": {
          "node": "Bm_64yc0TrqBG9b0YzOK8Q",
          "id": 101,
          "type": "transport",
          "action": "cluster:monitor\/tasks\/lists",
          "start_time_in_millis": 1535345746455,
          "running_time_in_nanos": 260584,
          "cancellable": false,
          "headers": {
            "X-Opaque-Id": "MyOpaqueId"
          }
        },
        "Bm_64yc0TrqBG9b0YzOK8Q:102": {
          "node": "Bm_64yc0TrqBG9b0YzOK8Q",
          "id": 102,
          "type": "direct",
          "action": "cluster:monitor\/tasks\/lists[n]",
          "start_time_in_millis": 1535345746455,
          "running_time_in_nanos": 50072,
          "cancellable": false,
          "parent_task_id": "Bm_64yc0TrqBG9b0YzOK8Q:101",
          "headers": {
            "X-Opaque-Id": "MyOpaqueId"
          }
        }
      }
    }
  }
}

Enumerating the tasks result will reveal the underlying Elasticsearch task id(s). These task id(s) can then be used with the other task management API methods; for example, to cancel the task or to retrieve its current status.

foreach (var node in tasks.Nodes)
{
    foreach (var task in node.Value.Tasks)
    {
        if (!task.Value.Cancellable) continue;
        var taskId = task.Key;
        var status = client.CancelTasks(o => o.TaskId(taskId)); // Cancel the task
    }
}

Keep It Green

In this short article hopefully you have learned how you can track and manage requests originating from your client application, and cancel them should you need to.

It's a fairly simple feature, but one that can become quite important for the ongoing management of your cluster.