Add GCS telemtry with ThreadLocal #125452

mhl-b · 2025-03-22T07:11:02Z

A simplified, yet larger code size, implementation of GCP telemetry. All previous efforts to intercept HTTP calls and do meaningful correlation between operations and requests were fruitless. This approach stores stats in the ThreadLocal so HTTP response interceptor can populate request metrics for parent operation.

It's achieved with new abstraction on top of GCS interface - MeteredStorage. This class provides a family of "MeteredXXX" wrappers for GCS objects, such as Storage API itself, MeteredObjectsGetRequest, MeteredWriteChannel, MeteredReadChannel, MeteredBlobPage.

Interesting bits for review RepositoryStatsCollector, OperationStats, and MeteredStorage.

MeteredStorage will execute GCS implementation and attach metrics to the ThreadLocal. RepositoryStatsCollector will get stats from the ThreadLocal and populate metrics and then aggregate collected metrics and expose them to REST stats and APM.

Every SDK call is wrapped into one of those helpers to populate ThreadLocal, collect results, and cleanup. See RepositoryStatsCollector.

    public <T> T collectIOSupplier(OperationPurpose purpose, StorageOperation operation, IOSupplier<T> blobFn) throws IOException {
        var t = timer.absoluteTimeInMillis();
        var stats = initAndGetThreadLocal(purpose, operation);
        try {
            return blobFn.get();
        } finally {
            stats.totalDuration += timer.absoluteTimeInMillis() - t;
            collect(stats);
            clearThreadLocal();
        }
    }

Response interceptor simplified to

    public static final HttpResponseInterceptor METERING_INTERCEPTOR = response -> {
        var stats = getThreadLocal();
        var code = response.getStatusCode();
        stats.requestAttempts += 1;
        stats.isLastRequestSucceed = true;
        if (((code >= 200 && code < 300) || code == 308 || code == 404) == false) {
            stats.requestError += 1;
            stats.isLastRequestSucceed = false;
            switch (code) {
                case 416 -> stats.requestRangeError += 1;
                case 429 -> stats.requestThrottle += 1;
            }
        }
    };

PS, this approach does not require creation of clients for every OpearationPurpose, already removed.

elasticsearchmachine · 2025-03-22T07:16:08Z

Hi @mhl-b, I've created a changelog YAML for you.

elasticsearchmachine · 2025-03-22T07:31:20Z

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

elasticsearchmachine · 2025-03-22T07:31:21Z

Hi @mhl-b, I've created a changelog YAML for you.

…rch into gcp-metrics-threadlocal

docs/changelog/125452.yaml

...sitory-gcs/src/main/java/org/elasticsearch/repositories/gcs/GoogleCloudStorageBlobStore.java

...pository-gcs/src/main/java/org/elasticsearch/repositories/gcs/GoogleCloudStorageService.java

nicktindall · 2025-03-23T23:00:30Z

modules/repository-gcs/src/main/java/org/elasticsearch/repositories/gcs/OperationStats.java

+    static OperationStats initAndGet(OperationPurpose purpose, StorageOperation operation) {
+        var stats = new OperationStats(purpose, operation);
+        OPERATION_STATS.set(stats);
+        return stats;


Should we assert here that we're not over-writing an existing OperationStats? it seems like if things are working properly we should never do that?

And below in set perhaps?

nicktindall · 2025-03-23T23:05:43Z

...epository-gcs/src/main/java/org/elasticsearch/repositories/gcs/RepositoryStatsCollector.java

+    public Map<String, BlobStoreActionStats> operationsStats() {
+        return restMetering.entrySet().stream().collect(Collectors.toMap(e -> e.getKey().key, e -> {
+            var ops = e.getValue().sum();
+            return new BlobStoreActionStats(ops, ops);


This should be ops and requests (ops/ops) was just a stop-gap until we had the ability to count requests

I'm thinking BlobStoreActionStats is not practical. REST metering tracks operation counts only and APM structure does not fit in it, the error metrics in particular. I think we can remove BlobStoreActionStats.

For example record BillableRequests(long count){} and Map<Operation,BillableRequests>. That would be more accurate representation for REST API metering endpoint.

nicktindall · 2025-03-23T23:11:07Z

...epository-gcs/src/main/java/org/elasticsearch/repositories/gcs/RepositoryStatsCollector.java

+     * Continue collecting metrics with given OperationStats. Useful for readers and writers.
+     */
+    public void continueAndCollect(OperationStats stats, CheckedRunnable<IOException> runnable) throws IOException {
+        OperationStats.set(stats);


I wonder if we should be more assertive about the current state of OperationStats. In this case it's probably safe to clear it after each action right? then we could assert that it was empty each time we went to start a new action, to make it fail noisily if they started being interleaved somehow?

...ory-gcs/src/test/java/org/elasticsearch/repositories/gcs/GoogleCloudStorageServiceTests.java

nicktindall · 2025-03-23T23:15:40Z

server/src/main/java/org/elasticsearch/repositories/RepositoriesMetrics.java

+     * Create the map of attributes we expect to see on repository metrics
+     */
+    public static Map<String, Object> createAttributesMap(String repoType, String repoName, OperationPurpose purpose, String operation) {
+        return Map.of("repo_type", repoType, "repo_name", repoName, "operation", operation, "purpose", purpose.getKey());


do we need the additional layer of indirection?

not really, will inline

nicktindall

I like this approach much better, a few comments

…ocal

mhl-b · 2025-04-03T07:14:09Z

@nicktindall , it's ready for another review. I havn't done APM testing yet, PR is already large. Will do a follow up PR.

...sitory-gcs/src/main/java/org/elasticsearch/repositories/gcs/GcsRepositoryStatsCollector.java

nicktindall

Looking good, I have some queries around the counting logic and other minor stuff

...sitory-gcs/src/main/java/org/elasticsearch/repositories/gcs/GcsRepositoryStatsCollector.java

nicktindall · 2025-04-04T00:52:48Z

...epository-gcs/src/main/java/org/elasticsearch/repositories/gcs/GoogleCloudStoragePlugin.java

@@ -34,14 +35,15 @@ public class GoogleCloudStoragePlugin extends Plugin implements RepositoryPlugin

    @SuppressWarnings("this-escape")
    public GoogleCloudStoragePlugin(final Settings settings) {
-        this.storageService = createStorageService(settings);
+        var isServerless = settings.getAsBoolean(DiscoveryNode.STATELESS_ENABLED_SETTING_NAME, false);


I think there's a helper for this org.elasticsearch.cluster.node.DiscoveryNode#isStateless, there is some sentiment that its use shouldn't be expanded but it just does what you're doing here.

I think we're supposed to prefer plugins/SPI to having explicit flag checks like this (maybe something we could consider later)

...pository-gcs/src/main/java/org/elasticsearch/repositories/gcs/GoogleCloudStorageService.java

...sitory-gcs/src/main/java/org/elasticsearch/repositories/gcs/GcsRepositoryStatsCollector.java

mhl-b · 2025-04-04T05:12:12Z

...Test/java/org/elasticsearch/repositories/gcs/GoogleCloudStorageBlobStoreRepositoryTests.java

+                trackRequest(StorageOperation.LIST.key());
+            } else if (Regex.simpleMatch("POST /upload/storage/v1/b/*uploadType=resumable*", request)) {
+                trackRequest(StorageOperation.INSERT.key());
+            } else if (Regex.simpleMatch("PUT /upload/storage/v1/b/*uploadType=resumable*", request)) {


count all requests now, not operations

nicktindall · 2025-04-04T05:22:58Z

...Test/java/org/elasticsearch/repositories/gcs/GoogleCloudStorageBlobStoreRepositoryTests.java

+    @Override
+    public void testRequestStats() throws Exception {
+        super.testRequestStats();
+    }


helper to run from ide

nicktindall · 2025-04-04T05:26:40Z

...sitory-gcs/src/main/java/org/elasticsearch/repositories/gcs/GcsRepositoryStatsCollector.java

+
+    /**
+     * Continue collecting metrics with given OperationStats. Useful for readers and writers.
+     */


copy/paste error on the comment here? or just out of date?

out of date

nicktindall · 2025-04-04T05:28:31Z

...sitory-gcs/src/main/java/org/elasticsearch/repositories/gcs/GcsRepositoryStatsCollector.java

+    }
+
+    private void collect(OperationStats stats) {
+        if (stats.reqAtt == 0) {


I kinda think we shouldn't use abbreviations here, requestsAttempted reads much nicer than Att especially because att can mean so many things

And we have attr floating around in the same block of code. It's a bit confusing.

nicktindall

LGTM, with some minor comments 🏆

mhl-b · 2025-04-04T06:06:01Z

A followup issue on removing serverless flag from BlobStore#stats() #126262

elasticsearchmachine added the v9.1.0 label Mar 22, 2025

mhl-b added >enhancement :Distributed Coordination/Snapshot/Restore Anything directly related to the `_snapshot/*` APIs Team:Distributed Coordination Meta label for Distributed Coordination team labels Mar 22, 2025

add GCS telemtry with threadlocals

b5da931

mhl-b force-pushed the gcp-metrics-threadlocal branch from 6983065 to b5da931 Compare March 22, 2025 07:30

mhl-b marked this pull request as ready for review March 22, 2025 07:30

Update docs/changelog/125452.yaml

4af0aed

mhl-b added 3 commits March 22, 2025 19:26

fix compilation

63166e4

Merge branch 'gcp-metrics-threadlocal' of github.com:mhl-b/elasticsea…

5b7fe05

…rch into gcp-metrics-threadlocal

Merge branch 'main' into gcp-metrics-threadlocal

0d4f8f8

mhl-b requested a review from nicktindall March 23, 2025 05:02