[AdmissionControl] Using threadcontext to collect resource usage stats

[bharath-techie]

Overview

The parent RFC #8910 discusses admission control framework which uses resource usage stats of the peer data nodes in the coordinator to help in ranking / rejection in coordinator node.

We proposed using Threadcontext to carry resource usage / health stats of the downstream nodes to the coordinator. ( As part of response headers )
This document covers the low level design / implementation details of the usage of threadcontext for the same.

Low level design

Storing stats as part of ResponseHeaders in ThreadContext

1. We will have a cluster setting to control this stats collection , by default the value will be ‘enabled’
2. Transport actions which are registered with admission control will transfer node resource usage stats as part of the thread context in response flow.
   1. Should we have a setting to enable the resource usage collection as part of 'all' transport actions ? ( this will improve accuracy but will cause more chattiness ) - will update the doc with performance impact.
3. Storing resource usage stats of the node in 'ThreadContext' is done in core ‘TransportService’ as shown in the above diagram.

--- In TransportService.java ---

public final <T extends TransportResponse> void sendRequest(
        final Transport.Connection connection,
        final String action,
        final TransportRequest request,
        final TransportRequestOptions options,
        final TransportResponseHandler<T> handler
    ) {
        // same method where we start span and end span in tracing flow
        ...
        delegate = new TransportResponseHandler<T>() {
            @Override
            public void handleResponse(T response) {
                // We will add resource usage stats to thread context
               // based on if action is registered with admission control
                addResourceUsageStatsToThreadContext(action);
                ....
            }
        ...
    }

Retrieving stats from response headers

We will check if the incoming request has resource usage stats in thread context response headers and we’ll add to the collector in coordinator.

1. Time to live : Here, 'TTL' refers to whether to use stats for 'admission control' or 'ranking' based on the 'timestamp' when the stats are collected.
   1. For example, if TTL is 5 seconds for the stats , if the collected stats timestamp is more than 5 seconds old, we won’t use the data to take relevant actions. ( DataNodeTimestamp and CoordinatorTimestamp )
2. Delayed responses : We will use the 'DataNodeTimestamp' stored along with stats in coordinator to ignore older stats of delayed responses from data node.
3. Avoid constant updation : We’ll update stats in coordinator only if the data stored in coordinator is more than '1 second' old ( configurable ) or absent. We’ll make sure the 'node id' is parsed quickly and rest of the data is parsed only if needed avoiding unnecessary computation.

--- In TransportService.java ---

/**
 * called by the {@link Transport} implementation when an incoming request arrives but before
 * any parsing of it has happened (with the exception of the requestId and action)
 */
@Override
public void onRequestReceived(long requestId, String action) {
       // Parse resource usage stats from response headers of thread context
       // Add resource usage stats to 'ResourceUsageCollectorService'
}

Format of Data stored in Response headers of ThreadContext

Resource usage stats is stored as a 'string' in the response headers of the threadcontext. Format of the string should follow these key principles:

1. Format should be extensible to newer usage stats.
2. Format should be extensible to multiple nodes.
3. ‘Node ID’ should be easily parseable , to make a decision whether to discard the stats or if we need to parse the whole string.
4. Payload should be lightweight to minimize any additional overhead

Format :

• Each node stats separated by semicolon ( ; )
• Each stat separated by comma ( , ) - newer stats can easily get appended and we’ll parse each value using index number
• Node and its stats are separated by colon ( : ) - can optimally parse node id before parsing further stats ( helpful in discard cases )

Key : Resource_usage_stats
Value : NodeID:CpuUsageStress,HeapUsageStress,IoUsage,timestamp;NodeID:<Stats>

Example :

{
    "Resource_Usage_Stats" : 
    "node1id:92.6,84.6,82.5,1700031452;node2id:55.6,45.6,77.5,1700031489"
 }

[gashutos]

Transport actions which are registered with admission control will transfer node resource usage stats as part of the thread context in response flow.
Should we have a setting to enable the resource usage collection as part of 'all' transport actions ? ( this will improve accuracy but will cause more chattiness ) - will update the doc with performance impact.

I would say lets have all by default for more accuracy, in case we see congestion, we can have ac_only in settings.

Rest looks good ! Nice work. Will be great to see PR.

[bharath-techie]

When used with security plugin , the response headers gets reset as part of stash - we might have to add custom logic in stash to not reset 'Resource_usage_stats' field in response headers , instead initialize the newly created threadcontext with them.

[msfroh]

@ansjcy -- Do you think this would be superseded by the work you're doing on Query Insights, like #13172?

@bharath-techie -- how much overlap is there between this PR and your proposal here?

[ansjcy]

@msfroh Actually the proposed solution to store stats as part of ResponseHeaders in ThreadContext is exactly how we get query-level resource usage metrics on the coordinator node in #13172 :)

After the shard search responses are sent to the coordinator node, we gather the resource usage for all nodes into SearchRequestContext (https://github.com/opensearch-project/OpenSearch/pull/13172/files#diff-f779f0258605cbf3db7076d76314256f74ba476da7b868e4fb05c5841ef9c781R47). In this case I think the missing piece to finish this propsal is to implement the desired listener.

[bharath-techie]

Hi @msfroh @ansjcy ,

#13172 certainly seems like a working solution for threadcontext based resource usage transmission from data nodes to coordinator ndoes.

In this proposal , we explored sending node level resource usage stats as part of search and indexing phases.

1. Is there a possibility to extend the search solution later on to send node level stats , so that node level decisions can be taken in coordinator utilizing them ? ( ranking / admission control ) Doubt if we can derive node level stats based on node id of the current TaskResourceUsage stats. ( please confirm if otherwise )
2. For indexing flows, we can implement fresh based on the learnings from Query-level resource usages tracking #13172.

Tagging @Bukhtawar @ajaymovva as well to check if they have anything to add.