Reuse local node in async shard fetch responses

[DaveCTurner]

Reuse local node in async shard fetch responses

We read various objects from the wire that already exist in the cluster
state. The most notable is DiscoveryNode which can consume ~2kB in
heap for each fresh object, but rarely changes, so it's pretty wasteful
to use fresh objects here. There could be thousands (millions?) of
DiscoveryNode objects in flight from various TransportNodesAction
responses.

This branch adds a DiscoveryNode parameter to the response
deserialisation method and makes sure that the worst offenders re-use
the local object rather than creating a fresh one:

• TransportNodesListShardStoreMetadata
• TransportNodesListGatewayStartedShards

Relates #77266

[elasticmachine]

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

[henningandersen]

LGTM.

[DaveCTurner]

Thanks Henning :)

[elasticsearchmachine]

💔 Backport failed

Status	Branch	Result
❌	7.x	Commit could not be cherrypicked due to conflicts

You can use sqren/backport to manually backport by running backport --upstream elastic/elasticsearch --pr 77991

[howardhuanghua]

@DaveCTurner In our production env, we double checked the optimization about fetcing respose memory consumption.
Before optimization, we could see DiscoveryNode costs 1.7k memory:

After this PR optimization, it only has 128 bytes that contains TransportAddress :

However, the master's heap is still crash, due to huge inflight fetch shard requests. In our case, we have 75 data nodes, and 3 dedicated master nodes, each master node has 4 GB heap, 1.5w shards. After full restarting cluster, master node memory will used up in several seconds. We dump the memory and found netty inflight sending request used lots of heap:

Each WriteOperation should be single shard request to specific node (16k buffer size per each):

From Netty4MessageChannelHandler class we could see a queuedWrites, messages are flushed asynchronously:

elasticsearch/modules/transport-netty4/src/main/java/org/elasticsearch/transport/netty4/Netty4MessageChannelHandler.java

Line 42 in 025dbdc

private final Queue<WriteOperation> queuedWrites = new ArrayDeque<>();

So besides cutting fetch shard response, we also need to handle massive shard sending requests.

[DaveCTurner]

That sounds like a separate problem @howardhuanghua, although it's related. Would you open a new issue about it?

1.5w shards

I think that's a typo, but this is important - how many shards are there in this cluster?

[howardhuanghua]

@DaveCTurner Sorry about the typo. 15000 shards total in cluster. I will open another issue.

[howardhuanghua]

Opened a new issue #80694.

[DaveCTurner]

I think that's a typo

TIL that you used "w" to abbreviate "wan", i.e. 万, meaning 10,000. I didn't know that was a thing, but I do now 😄

[howardhuanghua]

😄 Yes, you are right. It's a Chinese style.