Add ETA for Reindex, update_by_query, delete_by_query

[amooncake]

Closes #76904

[github-actions]

Documentation preview:

• ✨ Changed pages

[amooncake]

Add eta_millis in status, a human readable eta will also be returned when human=true in query params

{
  "node": "Xqj-2eUEQES4sqBzcX4FVA",
  "id": 172,
  "type": "transport",
  "action": "indices:data/write/delete/byquery",
  "status": {
    "slice_id": 1,
    "total": 113,
    "updated": 0,
    "created": 0,
    "deleted": 8,
    "batches": 4,
    "version_conflicts": 0,
    "noops": 0,
    "retries": {
      "bulk": 0,
      "search": 0
    },
    "throttled": "11.9s",
    "throttled_millis": 11999,
    "requests_per_second": 0.5,
    "throttled_until": "3.3s",
    "throttled_until_millis": 3325,
    "eta": "3.4m",
    "eta_millis": 209999
  },
  "description": "delete-by-query [sales]",
  "start_time": "2022-09-04T07:57:14.465Z",
  "start_time_in_millis": 1662278234465,
  "running_time": "13s",
  "running_time_in_nanos": 13015756458,
  "cancellable": true,
  "cancelled": false,
  "parent_task_id": "Xqj-2eUEQES4sqBzcX4FVA:168",
  "headers": {
    
  }
}

For reviewers: eta was calculated by rest/lastBatchRequestsPerMillis

elasticsearch/server/src/main/java/org/elasticsearch/index/reindex/WorkerBulkByScrollTaskState.java

Lines 210 to 218 in 072b9c2

	public TimeValue throttleWaitTime(long lastBatchStartTimeNS, long nowNS, int lastBatchSize) {
	long earliestNextBatchStartTime = nowNS + (long) perfectlyThrottledBatchTime(lastBatchSize);
	long waitTime = min(MAX_THROTTLE_WAIT_TIME.nanos(), max(0, earliestNextBatchStartTime - System.nanoTime()));
	this.lastBatchRequestsPerMillis = 1.0f * lastBatchSize / max(
	1,
	TimeUnit.NANOSECONDS.toMillis(earliestNextBatchStartTime - lastBatchStartTimeNS)
	);
	return timeValueNanos(waitTime);
	}

elasticsearch/server/src/main/java/org/elasticsearch/index/reindex/WorkerBulkByScrollTaskState.java

Lines 174 to 184 in 072b9c2

	TimeValue eta() {
	if (lastBatchRequestsPerMillis > 0) {
	long done = updated.get() + created.get() + deleted.get() + noops.get() + versionConflicts.get();
	long rest = total.get() - done;
	// `done` should be less than `total`, just in case
	rest = rest < 0 ? 0 : rest;
	return timeValueMillis((long) Math.floor(rest / lastBatchRequestsPerMillis));
	} else {
	return timeValueMillis(0);
	}
	}

DELETE sales

PUT sales

POST sales/_bulk?refresh
{"index":{}}
{"date": "2015/01/01 00:00:00", "price": 200, "promoted": true, "rating": 1, "type": "hat"}
{"index":{}}
{"date": "2015/01/01 00:00:00", "price": 200, "promoted": true, "rating": 1, "type": "t-shirt"}
{"index":{}}
{"date": "2015/01/01 00:00:00", "price": 150, "promoted": true, "rating": 5, "type": "bag"}
{"index":{}}
{"date": "2015/02/01 00:00:00", "price": 50, "promoted": false, "rating": 1, "type": "hat"}
{"index":{}}
{"date": "2015/02/01 00:00:00", "price": 10, "promoted": true, "rating": 4, "type": "t-shirt"}

POST sales/_delete_by_query?scroll_size=2&wait_for_completion=false&requests_per_second=1&slices=2&&conflicts=proceed
{
  "query": {
    "match_all": {}
  }
}

GET _tasks?detailed=true&actions=*/write/*&human=true&group_by=parents

[elasticsearchmachine]

Pinging @elastic/es-distributed (Team:Distributed)