Making snapshot deletes distributed across data nodes.

[backslasht]

Unlike snapshot creation operation, snapshot deletion operation is executed
in master. This is acceptable for small to medium sized clusters where the
delete operation completes in less than a minute. But, for bigger clusters with
few thousand primary shards, the deletion operation takes considerable amount of
time as there is only one thread working on deleting the snapshot data index by
index, shard by shard.

To reduce the time spent on deletion, parallelizing (distributing) the snapshot deletes
across all data nodes similar to snapshot creation but with no tight coupling between
data node and snapshot shards that are getting deleted.

With this changes, for a cluster with 3 dedicated masters and 20 data nodes and 20,000
primary shards, the deletion took around 1 minute where as without this change it took
around 70 minutes.

[ywelsch]

The problem is that deletes are processed sequentially on the master. I don't think we should add a whole new distributed layer for this, but just look into doing some of these calls concurrently on the master.

@backslasht before working on such a large item, it's best to open an issue first to discuss the solution.

[elasticmachine]

Pinging @elastic/es-distributed

[backslasht]

@ywelsch - Like you mentioned, I initially started with making calls concurrent on the master. It does help a little but still the time taken in quite high.

Let us a take following setup as an example.
Dedicated master available: Yes
Number of Data Nodes: 100
Number of Primary Shards: 10,000
Time taken to compute list of shards = y seconds (say 10 for example)
Average time to delete a snapshot of a shard = x seconds (say 5 for example)

1. Deletion time with single thread (as it exists today)
   -> y + 10000 * x

2. Deletion time with parallelization in master (5 threads) - ** 5 times improvement over 1**
   -> y + (10000/5) * x
   -> y + 2000 * x

3. Deletion time with multiple threads across data nodes (proposed change) - ** 500 times improvement over 1 and 100 times improvement over 2**
   -> y + ((10000/100)/5) * x
   -> y +( 100/5) * x
   -> y + 20 * x

So, the proposed change makes snapshot delete operation time as a function of number of data nodes and number of primary shards similar to snapshot creation. The deletion time will remain the same with increase in data nodes (and corresponding increase in shards).

[ywelsch]

I initially started with making calls concurrent on the master. It does help a little but still the time taken in quite high ...

That's a bit of an odd comparison. It all depends on how many concurrent calls you globally do. Whether the calls are done from a single node or multiple nodes does not matter given that the requests are very small and latency seems to be the determining factor.

We had a team discussion on this implementation and think it's too complex, which is why I'm closing this PR. We still think this is a problem worth tackling, though, but are looking at other ways of achieving this without building a full distributed task queue. Two options we're looking at is 1) parallelizing deletes on the master, possibly using async APIs, and 2) using repository-specific options (e.g. bulk-deletions on S3).