From c07d93e9dcfffced3e93abfd169b70df995d734a Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Mon, 6 Jul 2026 15:18:55 +0200 Subject: [PATCH 1/6] Add ProfileEvents for DataLake catalog authorization token refresh. Track how often REST-family catalogs reuse cached OAuth/GCP tokens versus fetching new ones or retrying after HTTP 401/403, and add an integration test for the new metrics. Co-authored-by: Cursor --- src/Common/ProfileEvents.cpp | 6 ++ src/Databases/DataLake/PaimonRestCatalog.cpp | 7 ++ src/Databases/DataLake/RestCatalog.cpp | 92 +++++++++++++++---- .../test.py | 79 ++++++++++++++++ 4 files changed, 166 insertions(+), 18 deletions(-) diff --git a/src/Common/ProfileEvents.cpp b/src/Common/ProfileEvents.cpp index 8beb91a948c8..b2ff1dae67c2 100644 --- a/src/Common/ProfileEvents.cpp +++ b/src/Common/ProfileEvents.cpp @@ -1549,6 +1549,10 @@ The server successfully detected this situation and will download merged part fr M(DataLakeRestCatalogGetTableMetadataMicroseconds, "Total time of 'get table metadata' requests to Iceberg REST catalog.", ValueType::Microseconds) \ M(DataLakeRestCatalogGetCredentials, "Number of 'get credentials' requests to Iceberg REST catalog.", ValueType::Number) \ M(DataLakeRestCatalogGetCredentialsMicroseconds, "Total time of 'get credentials' requests to Iceberg REST catalog.", ValueType::Microseconds) \ + M(DataLakeRestCatalogAuthTokenCacheHits, "Number of requests to Iceberg REST catalog that reused a cached access token and did not fetch a new one.", ValueType::Number) \ + M(DataLakeRestCatalogAuthTokenRefreshed, "Number of new access tokens fetched for Iceberg REST catalog (OAuth client-credentials or GCP metadata/ADC).", ValueType::Number) \ + M(DataLakeRestCatalogAuthTokenRefreshedMicroseconds, "Total time spent fetching access tokens for Iceberg REST catalog.", ValueType::Microseconds) \ + M(DataLakeRestCatalogAuthTokenRefreshedOnUnauthorized, "Number of Iceberg REST catalog HTTP requests retried with a new access token after HTTP 401 or 403.", ValueType::Number) \ M(DataLakeRestCatalogCreateNamespace, "Number of 'create namespace' requests to Iceberg REST catalog.", ValueType::Number) \ M(DataLakeRestCatalogCreateNamespaceMicroseconds, "Total time of 'create namespace' requests to Iceberg REST catalog.", ValueType::Microseconds) \ M(DataLakeRestCatalogCreateTable, "Number of 'create table' requests to Iceberg REST catalog.", ValueType::Number) \ @@ -1588,6 +1592,8 @@ The server successfully detected this situation and will download merged part fr M(ObjectStorageClusterSentToNonMatchedReplica, "Number of tasks in ObjectStorageCluster request sent to non-matched replica.", ValueType::Number) \ M(ObjectStorageClusterProcessedTasks, "Number of processed tasks in ObjectStorageCluster request.", ValueType::Number) \ M(ObjectStorageClusterWaitingMicroseconds, "Time of waiting for tasks in ObjectStorageCluster request.", ValueType::Microseconds) \ + M(DataLakePaimonRestCatalogAuthTokenRefreshedOnUnauthorized, "Number of Paimon REST catalog DLF HTTP requests retried with a new authorization signature after HTTP 401.", ValueType::Number) \ + #ifdef APPLY_FOR_EXTERNAL_EVENTS #define APPLY_FOR_EVENTS(M) APPLY_FOR_BUILTIN_EVENTS(M) APPLY_FOR_EXTERNAL_EVENTS(M) diff --git a/src/Databases/DataLake/PaimonRestCatalog.cpp b/src/Databases/DataLake/PaimonRestCatalog.cpp index a2d2dd04c080..1652b12ba2db 100644 --- a/src/Databases/DataLake/PaimonRestCatalog.cpp +++ b/src/Databases/DataLake/PaimonRestCatalog.cpp @@ -37,11 +37,17 @@ #include #include #include +#include #include #include #include #include +namespace ProfileEvents +{ + extern const Event DataLakePaimonRestCatalogAuthTokenRefreshedOnUnauthorized; +} + namespace DB::ErrorCodes { @@ -311,6 +317,7 @@ DB::ReadWriteBufferFromHTTPPtr PaimonRestCatalog::createReadBuffer( { if (e.code() == Poco::Net::HTTPResponse::HTTP_UNAUTHORIZED && refresh_token && token->token_provider == "dlf") { + ProfileEvents::increment(ProfileEvents::DataLakePaimonRestCatalogAuthTokenRefreshedOnUnauthorized); refresh_token = false; token->dlf_generated_authorization = ""; return create_buffer(); diff --git a/src/Databases/DataLake/RestCatalog.cpp b/src/Databases/DataLake/RestCatalog.cpp index ecbda190a262..73f0769d0a2a 100644 --- a/src/Databases/DataLake/RestCatalog.cpp +++ b/src/Databases/DataLake/RestCatalog.cpp @@ -90,6 +90,15 @@ namespace ProfileEvents extern const Event DataLakeRestCatalogGetTableMetadataMicroseconds; extern const Event DataLakeRestCatalogGetCredentials; extern const Event DataLakeRestCatalogGetCredentialsMicroseconds; +<<<<<<< HEAD +======= + extern const Event DataLakeRestCatalogCredentialsVended; + extern const Event DataLakeRestCatalogCredentialsCacheHits; + extern const Event DataLakeRestCatalogAuthTokenCacheHits; + extern const Event DataLakeRestCatalogAuthTokenRefreshed; + extern const Event DataLakeRestCatalogAuthTokenRefreshedMicroseconds; + extern const Event DataLakeRestCatalogAuthTokenRefreshedOnUnauthorized; +>>>>>>> 274c81be689 (Add ProfileEvents for DataLake catalog authorization token refresh.) extern const Event DataLakeRestCatalogCreateNamespace; extern const Event DataLakeRestCatalogCreateNamespaceMicroseconds; extern const Event DataLakeRestCatalogCreateTable; @@ -338,6 +347,10 @@ DB::HTTPHeaderEntries RestCatalog::getAuthHeaders( access_token.set(std::make_unique(retrieveAccessToken())); current = access_token.get(); } + else + { + ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogAuthTokenCacheHits); + } DB::HTTPHeaderEntries headers; headers.emplace_back("Authorization", "Bearer " + current->token); @@ -390,6 +403,9 @@ String OneLakeCatalog::getBearerToken() const AccessToken RestCatalog::retrieveAccessToken() const { + ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogAuthTokenRefreshed); + auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeRestCatalogAuthTokenRefreshedMicroseconds); + static constexpr auto oauth_tokens_endpoint = "oauth/tokens"; /// TODO: @@ -524,6 +540,10 @@ DB::HTTPHeaderEntries BigLakeCatalog::getAuthHeaders( access_token.set(std::make_unique(retrieveGoogleCloudAccessToken())); current = access_token.get(); } + else + { + ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogAuthTokenCacheHits); + } DB::HTTPHeaderEntries headers; headers.emplace_back("Authorization", "Bearer " + current->token); @@ -547,6 +567,9 @@ DB::HTTPHeaderEntries BigLakeCatalog::getAuthHeaders( AccessToken BigLakeCatalog::retrieveGoogleCloudAccessTokenFromRefreshToken() const { + ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogAuthTokenRefreshed); + auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeRestCatalogAuthTokenRefreshedMicroseconds); + if (google_adc_client_id.empty() || google_adc_client_secret.empty() || google_adc_refresh_token.empty()) throw DB::Exception( DB::ErrorCodes::BAD_ARGUMENTS, @@ -578,6 +601,9 @@ AccessToken BigLakeCatalog::retrieveGoogleCloudAccessToken() const /// Fallback to GCP metadata service (works inside GCP infrastructure) /// https://cloud.google.com/compute/docs/metadata/overview + ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogAuthTokenRefreshed); + auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeRestCatalogAuthTokenRefreshedMicroseconds); + static constexpr auto DEFAULT_REQUEST_TOKEN_PATH = "/computeMetadata/v1/instance/service-accounts"; const auto & context = getContext(); @@ -711,6 +737,7 @@ DB::ReadWriteBufferFromHTTPPtr RestCatalog::createReadBuffer( (status == Poco::Net::HTTPResponse::HTTPStatus::HTTP_UNAUTHORIZED || status == Poco::Net::HTTPResponse::HTTPStatus::HTTP_FORBIDDEN)) { + ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogAuthTokenRefreshedOnUnauthorized); return create_buffer(true); } throw; @@ -1390,24 +1417,53 @@ void RestCatalog::sendRequest(const String & endpoint, Poco::JSON::Object::Ptr r DB::HTTPHeaderEntries extra_headers; extra_headers.emplace_back("Content-Type", "application/json"); - DB::HTTPHeaderEntries headers = getAuthHeaders(/* update_token = */ true, method, url, extra_headers, body_str); - headers.emplace_back("Content-Type", "application/json"); - auto wb = DB::BuilderRWBufferFromHTTP(url) - .withConnectionGroup(DB::HTTPConnectionGroupType::HTTP) - .withMethod(method) - .withSettings(context->getReadSettings()) - .withTimeouts(DB::ConnectionTimeouts::getHTTPTimeouts(context->getSettingsRef(), context->getServerSettings())) - .withHostFilter(&context->getRemoteHostFilter()) - .withHeaders(headers) - .withOutCallback(out_stream_callback) - .withSkipNotFound(false) - .create(credentials); - - String response_str; - if (!ignore_result) - readJSONObjectPossiblyInvalid(response_str, *wb); - else - wb->ignoreAll(); + auto create_buffer = [&](bool update_token) + { + DB::HTTPHeaderEntries headers = getAuthHeaders(update_token, method, url, extra_headers, body_str); + headers.emplace_back("Content-Type", "application/json"); + return DB::BuilderRWBufferFromHTTP(url) + .withConnectionGroup(DB::HTTPConnectionGroupType::HTTP) + .withMethod(method) + .withSettings(context->getReadSettings()) + .withTimeouts(DB::ConnectionTimeouts::getHTTPTimeouts(context->getSettingsRef(), context->getServerSettings())) + .withHostFilter(&context->getRemoteHostFilter()) + .withHeaders(headers) + .withOutCallback(out_stream_callback) + .withSkipNotFound(false) + .create(credentials); + }; + + try + { + auto wb = create_buffer(false); + + String response_str; + if (!ignore_result) + readJSONObjectPossiblyInvalid(response_str, *wb); + else + wb->ignoreAll(); + } + catch (const DB::HTTPException & e) + { + const auto status = e.getHTTPStatus(); + if (update_token_if_expired && + (status == Poco::Net::HTTPResponse::HTTPStatus::HTTP_UNAUTHORIZED + || status == Poco::Net::HTTPResponse::HTTPStatus::HTTP_FORBIDDEN)) + { + ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogAuthTokenRefreshedOnUnauthorized); + auto wb = create_buffer(true); + + String response_str; + if (!ignore_result) + readJSONObjectPossiblyInvalid(response_str, *wb); + else + wb->ignoreAll(); + } + else + { + throw; + } + } } void RestCatalog::createNamespaceIfNotExists(const String & namespace_name, const String & location) const diff --git a/tests/integration/test_database_iceberg_lakekeeper_catalog/test.py b/tests/integration/test_database_iceberg_lakekeeper_catalog/test.py index 7a7343c2f738..c7d8dd7d4384 100644 --- a/tests/integration/test_database_iceberg_lakekeeper_catalog/test.py +++ b/tests/integration/test_database_iceberg_lakekeeper_catalog/test.py @@ -414,6 +414,85 @@ def get_credentials_profile_events(node, query_id): return vended, hits +def get_auth_token_profile_events(node, query_id): + node.query("SYSTEM FLUSH LOGS") + refreshed = int(node.query( + f"SELECT ProfileEvents['DataLakeRestCatalogAuthTokenRefreshed'] " + f"FROM system.query_log WHERE query_id = '{query_id}' AND type = 'QueryFinish'" + )) + cache_hits = int(node.query( + f"SELECT ProfileEvents['DataLakeRestCatalogAuthTokenCacheHits'] " + f"FROM system.query_log WHERE query_id = '{query_id}' AND type = 'QueryFinish'" + )) + refreshed_on_unauthorized = int(node.query( + f"SELECT ProfileEvents['DataLakeRestCatalogAuthTokenRefreshedOnUnauthorized'] " + f"FROM system.query_log WHERE query_id = '{query_id}' AND type = 'QueryFinish'" + )) + return refreshed, cache_hits, refreshed_on_unauthorized + + +def test_auth_token_profile_events(started_cluster): + node = started_cluster.instances["node1"] + + test_ref = f"test_auth_token_profile_events_{uuid.uuid4().hex[:8]}" + db_name = f"{test_ref}_database" + namespace = (f"{test_ref}_namespace",) + table_name = f"{test_ref}_table" + + catalog = load_catalog_impl(started_cluster) + if namespace not in catalog.list_namespaces(): + catalog.create_namespace(namespace) + + schema = Schema( + NestedField(field_id=1, name="id", field_type=IntegerType(), required=False), + NestedField(field_id=2, name="data", field_type=StringType(), required=False), + ) + catalog.create_table( + namespace + (table_name,), + schema=schema, + properties={"write.metadata.compression-codec": "none"}, + ) + + create_qid = f"{test_ref}-create-{uuid.uuid4()}" + settings = { + "catalog_type": "rest", + "warehouse": "demo", + "storage_endpoint": "http://minio:9000/warehouse-rest", + "catalog_credential": "SECRET_1", + } + node.query( + f""" +DROP DATABASE IF EXISTS {db_name}; +SET allow_experimental_database_iceberg=true; +CREATE DATABASE {db_name} ENGINE = DataLakeCatalog('{BASE_URL}', 'minio', '{minio_secret_key}') +SETTINGS {",".join((k+"="+repr(v) for k, v in settings.items()))} + """, + query_id=create_qid, + ) + refreshed, _, _ = get_auth_token_profile_events(node, create_qid) + assert refreshed >= 1 + + qid1 = f"{test_ref}-show-1-{uuid.uuid4()}" + node.query(f"SHOW TABLES FROM {db_name}", query_id=qid1) + refreshed, cache_hits, _ = get_auth_token_profile_events(node, qid1) + assert refreshed == 0 and cache_hits >= 1 + + qid2 = f"{test_ref}-show-2-{uuid.uuid4()}" + node.query(f"SHOW TABLES FROM {db_name}", query_id=qid2) + refreshed, cache_hits, _ = get_auth_token_profile_events(node, qid2) + assert refreshed == 0 and cache_hits >= 1 + + +def test_vended_credentials_cache(started_cluster): + node = started_cluster.instances["node1"] + catalog = load_catalog_impl(started_cluster) + + test_ref = f"test_vended_credentials_cache_{uuid.uuid4().hex[:8]}" + namespace = (f"{test_ref}_namespace",) + table_name = f"{test_ref}_table" + db_name = f"{test_ref}_database" + + def create_int_table(catalog, namespace, table_name, rows=1): if namespace not in catalog.list_namespaces(): catalog.create_namespace(namespace) From 5794036b96d912fcce4dc6b675cf414635232d51 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Mon, 6 Jul 2026 16:55:28 +0200 Subject: [PATCH 2/6] Fix auth token profile events integration test for lazy catalog init. Use a mock OAuth server in the lakekeeper compose stack and valid client credentials so token refresh and cache-hit profile events are observed on SHOW TABLES queries. Co-authored-by: Cursor --- ...ker_compose_iceberg_lakekeeper_catalog.yml | 24 +++++++++++ .../test.py | 42 ++++++++----------- 2 files changed, 41 insertions(+), 25 deletions(-) diff --git a/tests/integration/compose/docker_compose_iceberg_lakekeeper_catalog.yml b/tests/integration/compose/docker_compose_iceberg_lakekeeper_catalog.yml index 85fa53adeea4..7f2951ddc986 100644 --- a/tests/integration/compose/docker_compose_iceberg_lakekeeper_catalog.yml +++ b/tests/integration/compose/docker_compose_iceberg_lakekeeper_catalog.yml @@ -70,3 +70,27 @@ services: retries: 5 start_period: 10s cpus: 3 + + mock-oauth: + image: python:3.12-alpine + command: + - python + - -c + - | + from http.server import HTTPServer, BaseHTTPRequestHandler + import json + class Handler(BaseHTTPRequestHandler): + def do_GET(self): + self.send_token() + def do_POST(self): + self.send_token() + def send_token(self): + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.end_headers() + payload = {"access_token": "test-token", "expires_in": 3600, "token_type": "Bearer"} + self.wfile.write(json.dumps(payload).encode()) + def log_message(self, format, *args): + pass + HTTPServer(("0.0.0.0", 9999), Handler).serve_forever() + cpus: 1 diff --git a/tests/integration/test_database_iceberg_lakekeeper_catalog/test.py b/tests/integration/test_database_iceberg_lakekeeper_catalog/test.py index c7d8dd7d4384..ccc56e07530b 100644 --- a/tests/integration/test_database_iceberg_lakekeeper_catalog/test.py +++ b/tests/integration/test_database_iceberg_lakekeeper_catalog/test.py @@ -20,6 +20,7 @@ from helpers.test_tools import TSV, csv_compare BASE_URL = "http://lakekeeper:8181/catalog" +MOCK_OAUTH_URL = "http://mock-oauth:9999/token" CATALOG_NAME = "demo" WAREHOUSE_NAME = "demo" @@ -424,11 +425,7 @@ def get_auth_token_profile_events(node, query_id): f"SELECT ProfileEvents['DataLakeRestCatalogAuthTokenCacheHits'] " f"FROM system.query_log WHERE query_id = '{query_id}' AND type = 'QueryFinish'" )) - refreshed_on_unauthorized = int(node.query( - f"SELECT ProfileEvents['DataLakeRestCatalogAuthTokenRefreshedOnUnauthorized'] " - f"FROM system.query_log WHERE query_id = '{query_id}' AND type = 'QueryFinish'" - )) - return refreshed, cache_hits, refreshed_on_unauthorized + return refreshed, cache_hits def test_auth_token_profile_events(started_cluster): @@ -453,33 +450,28 @@ def test_auth_token_profile_events(started_cluster): properties={"write.metadata.compression-codec": "none"}, ) - create_qid = f"{test_ref}-create-{uuid.uuid4()}" - settings = { - "catalog_type": "rest", - "warehouse": "demo", - "storage_endpoint": "http://minio:9000/warehouse-rest", - "catalog_credential": "SECRET_1", - } - node.query( - f""" -DROP DATABASE IF EXISTS {db_name}; -SET allow_experimental_database_iceberg=true; -CREATE DATABASE {db_name} ENGINE = DataLakeCatalog('{BASE_URL}', 'minio', '{minio_secret_key}') -SETTINGS {",".join((k+"="+repr(v) for k, v in settings.items()))} - """, - query_id=create_qid, + # The catalog client is initialized lazily on the first database access, + # not during CREATE DATABASE. OAuth credentials must use client_id:client_secret + # format; oauth_server_uri points to a mock token endpoint in docker compose. + create_clickhouse_iceberg_database( + started_cluster, + node, + db_name, + additional_settings={ + "catalog_credential": "test:secret", + "oauth_server_uri": MOCK_OAUTH_URL, + }, ) - refreshed, _, _ = get_auth_token_profile_events(node, create_qid) - assert refreshed >= 1 qid1 = f"{test_ref}-show-1-{uuid.uuid4()}" node.query(f"SHOW TABLES FROM {db_name}", query_id=qid1) - refreshed, cache_hits, _ = get_auth_token_profile_events(node, qid1) - assert refreshed == 0 and cache_hits >= 1 + assert table_name in node.query(f"SHOW TABLES FROM {db_name}") + refreshed, cache_hits = get_auth_token_profile_events(node, qid1) + assert refreshed >= 1 qid2 = f"{test_ref}-show-2-{uuid.uuid4()}" node.query(f"SHOW TABLES FROM {db_name}", query_id=qid2) - refreshed, cache_hits, _ = get_auth_token_profile_events(node, qid2) + refreshed, cache_hits = get_auth_token_profile_events(node, qid2) assert refreshed == 0 and cache_hits >= 1 From b08fae3f57c1fc774bc1822e68998d4e6afe82d5 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Mon, 6 Jul 2026 17:48:01 +0200 Subject: [PATCH 3/6] Fix auth token profile event accuracy in RestCatalog. Treat expired OAuth tokens as refresh candidates instead of cache hits, and count BigLake GCP token refresh once per logical fetch when ADC fallback to metadata is used. Co-authored-by: Cursor --- src/Databases/DataLake/RestCatalog.cpp | 16 ++++------------ 1 file changed, 4 insertions(+), 12 deletions(-) diff --git a/src/Databases/DataLake/RestCatalog.cpp b/src/Databases/DataLake/RestCatalog.cpp index 73f0769d0a2a..7a1c62c2a111 100644 --- a/src/Databases/DataLake/RestCatalog.cpp +++ b/src/Databases/DataLake/RestCatalog.cpp @@ -90,15 +90,10 @@ namespace ProfileEvents extern const Event DataLakeRestCatalogGetTableMetadataMicroseconds; extern const Event DataLakeRestCatalogGetCredentials; extern const Event DataLakeRestCatalogGetCredentialsMicroseconds; -<<<<<<< HEAD -======= - extern const Event DataLakeRestCatalogCredentialsVended; - extern const Event DataLakeRestCatalogCredentialsCacheHits; extern const Event DataLakeRestCatalogAuthTokenCacheHits; extern const Event DataLakeRestCatalogAuthTokenRefreshed; extern const Event DataLakeRestCatalogAuthTokenRefreshedMicroseconds; extern const Event DataLakeRestCatalogAuthTokenRefreshedOnUnauthorized; ->>>>>>> 274c81be689 (Add ProfileEvents for DataLake catalog authorization token refresh.) extern const Event DataLakeRestCatalogCreateNamespace; extern const Event DataLakeRestCatalogCreateNamespaceMicroseconds; extern const Event DataLakeRestCatalogCreateTable; @@ -342,7 +337,7 @@ DB::HTTPHeaderEntries RestCatalog::getAuthHeaders( if (!client_id.empty()) { auto current = access_token.get(); - if (!current || update_token) + if (!current || update_token || access_token->isExpired()) { access_token.set(std::make_unique(retrieveAccessToken())); current = access_token.get(); @@ -567,9 +562,6 @@ DB::HTTPHeaderEntries BigLakeCatalog::getAuthHeaders( AccessToken BigLakeCatalog::retrieveGoogleCloudAccessTokenFromRefreshToken() const { - ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogAuthTokenRefreshed); - auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeRestCatalogAuthTokenRefreshedMicroseconds); - if (google_adc_client_id.empty() || google_adc_client_secret.empty() || google_adc_refresh_token.empty()) throw DB::Exception( DB::ErrorCodes::BAD_ARGUMENTS, @@ -587,6 +579,9 @@ AccessToken BigLakeCatalog::retrieveGoogleCloudAccessTokenFromRefreshToken() con AccessToken BigLakeCatalog::retrieveGoogleCloudAccessToken() const { + ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogAuthTokenRefreshed); + auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeRestCatalogAuthTokenRefreshedMicroseconds); + if (!google_adc_client_id.empty() && !google_adc_client_secret.empty() && !google_adc_refresh_token.empty()) { try @@ -601,9 +596,6 @@ AccessToken BigLakeCatalog::retrieveGoogleCloudAccessToken() const /// Fallback to GCP metadata service (works inside GCP infrastructure) /// https://cloud.google.com/compute/docs/metadata/overview - ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogAuthTokenRefreshed); - auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeRestCatalogAuthTokenRefreshedMicroseconds); - static constexpr auto DEFAULT_REQUEST_TOKEN_PATH = "/computeMetadata/v1/instance/service-accounts"; const auto & context = getContext(); From af019fe1691ee3f646cd9839bdfc457e20b60ef2 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Mon, 6 Jul 2026 18:06:46 +0200 Subject: [PATCH 4/6] Remove event for Paimon --- src/Common/ProfileEvents.cpp | 1 - src/Databases/DataLake/PaimonRestCatalog.cpp | 7 ------- 2 files changed, 8 deletions(-) diff --git a/src/Common/ProfileEvents.cpp b/src/Common/ProfileEvents.cpp index b2ff1dae67c2..10e63267c816 100644 --- a/src/Common/ProfileEvents.cpp +++ b/src/Common/ProfileEvents.cpp @@ -1592,7 +1592,6 @@ The server successfully detected this situation and will download merged part fr M(ObjectStorageClusterSentToNonMatchedReplica, "Number of tasks in ObjectStorageCluster request sent to non-matched replica.", ValueType::Number) \ M(ObjectStorageClusterProcessedTasks, "Number of processed tasks in ObjectStorageCluster request.", ValueType::Number) \ M(ObjectStorageClusterWaitingMicroseconds, "Time of waiting for tasks in ObjectStorageCluster request.", ValueType::Microseconds) \ - M(DataLakePaimonRestCatalogAuthTokenRefreshedOnUnauthorized, "Number of Paimon REST catalog DLF HTTP requests retried with a new authorization signature after HTTP 401.", ValueType::Number) \ #ifdef APPLY_FOR_EXTERNAL_EVENTS diff --git a/src/Databases/DataLake/PaimonRestCatalog.cpp b/src/Databases/DataLake/PaimonRestCatalog.cpp index 1652b12ba2db..a2d2dd04c080 100644 --- a/src/Databases/DataLake/PaimonRestCatalog.cpp +++ b/src/Databases/DataLake/PaimonRestCatalog.cpp @@ -37,17 +37,11 @@ #include #include #include -#include #include #include #include #include -namespace ProfileEvents -{ - extern const Event DataLakePaimonRestCatalogAuthTokenRefreshedOnUnauthorized; -} - namespace DB::ErrorCodes { @@ -317,7 +311,6 @@ DB::ReadWriteBufferFromHTTPPtr PaimonRestCatalog::createReadBuffer( { if (e.code() == Poco::Net::HTTPResponse::HTTP_UNAUTHORIZED && refresh_token && token->token_provider == "dlf") { - ProfileEvents::increment(ProfileEvents::DataLakePaimonRestCatalogAuthTokenRefreshedOnUnauthorized); refresh_token = false; token->dlf_generated_authorization = ""; return create_buffer(); From 2a478950c262ad2a0fa61bb0e4faea4d0694a8a6 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Fri, 10 Jul 2026 00:29:57 +0200 Subject: [PATCH 5/6] Count auth token cache hits only after successful catalog requests. Defer DataLakeRestCatalogAuthTokenCacheHits until createReadBuffer or sendRequest completes without a 401/403 retry, so stale cached tokens do not report both a cache hit and a refresh in the same query. Co-authored-by: Cursor --- src/Databases/DataLake/RestCatalog.cpp | 54 ++++++++++++++-------- src/Databases/DataLake/RestCatalog.h | 6 ++- src/Databases/DataLake/S3TablesCatalog.cpp | 3 +- src/Databases/DataLake/S3TablesCatalog.h | 3 +- 4 files changed, 44 insertions(+), 22 deletions(-) diff --git a/src/Databases/DataLake/RestCatalog.cpp b/src/Databases/DataLake/RestCatalog.cpp index 7a1c62c2a111..d593d164f717 100644 --- a/src/Databases/DataLake/RestCatalog.cpp +++ b/src/Databases/DataLake/RestCatalog.cpp @@ -317,13 +317,17 @@ DB::HTTPHeaderEntries RestCatalog::getAuthHeaders( const String & /*method*/, const Poco::URI & /*url*/, const DB::HTTPHeaderEntries & /*extra_headers*/, - const String & /*body*/) const + const String & /*body*/, + bool * used_cached_oauth_token) const { fiu_do_on(DB::FailPoints::check_database_datalake_negative, { throw DB::Exception(DB::ErrorCodes::FAULT_INJECTED, "Injecting fault when checking database"); }); + if (used_cached_oauth_token) + *used_cached_oauth_token = false; + /// Option 1: user specified auth header manually. /// Header has format: 'Authorization: '. if (auth_header.has_value()) @@ -342,9 +346,9 @@ DB::HTTPHeaderEntries RestCatalog::getAuthHeaders( access_token.set(std::make_unique(retrieveAccessToken())); current = access_token.get(); } - else + else if (used_cached_oauth_token) { - ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogAuthTokenCacheHits); + *used_cached_oauth_token = true; } DB::HTTPHeaderEntries headers; @@ -518,10 +522,11 @@ BigLakeCatalog::BigLakeCatalog( DB::HTTPHeaderEntries BigLakeCatalog::getAuthHeaders( bool update_token, - const String & /*method*/, - const Poco::URI & /*url*/, - const DB::HTTPHeaderEntries & /*extra_headers*/, - const String & /*body*/) const + const String & method, + const Poco::URI & url, + const DB::HTTPHeaderEntries & extra_headers, + const String & body, + bool * used_cached_oauth_token) const { /// Google Cloud OAuth2 for BigLake. /// Uses GCP metadata service or Application Default Credentials to get access token. @@ -529,15 +534,18 @@ DB::HTTPHeaderEntries BigLakeCatalog::getAuthHeaders( /// https://developers.google.com/identity/protocols/oauth2 if (!google_project_id.empty() || !google_adc_client_id.empty()) { + if (used_cached_oauth_token) + *used_cached_oauth_token = false; + auto current = access_token.get(); if (!current || update_token || current->isExpired()) { access_token.set(std::make_unique(retrieveGoogleCloudAccessToken())); current = access_token.get(); } - else + else if (used_cached_oauth_token) { - ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogAuthTokenCacheHits); + *used_cached_oauth_token = true; } DB::HTTPHeaderEntries headers; @@ -557,7 +565,7 @@ DB::HTTPHeaderEntries BigLakeCatalog::getAuthHeaders( return headers; } - return RestCatalog::getAuthHeaders(update_token); + return RestCatalog::getAuthHeaders(update_token, method, url, extra_headers, body, used_cached_oauth_token); } AccessToken BigLakeCatalog::retrieveGoogleCloudAccessTokenFromRefreshToken() const @@ -700,9 +708,9 @@ DB::ReadWriteBufferFromHTTPPtr RestCatalog::createReadBuffer( if (!params.empty()) url.setQueryParameters(params); - auto create_buffer = [&](bool update_token) + auto create_buffer = [&](bool update_token, bool & used_cached_oauth_token) { - auto result_headers = getAuthHeaders(update_token, Poco::Net::HTTPRequest::HTTP_GET, url, headers, {}); + auto result_headers = getAuthHeaders(update_token, Poco::Net::HTTPRequest::HTTP_GET, url, headers, {}, &used_cached_oauth_token); std::move(headers.begin(), headers.end(), std::back_inserter(result_headers)); return DB::BuilderRWBufferFromHTTP(url) @@ -720,7 +728,11 @@ DB::ReadWriteBufferFromHTTPPtr RestCatalog::createReadBuffer( try { - return create_buffer(false); + bool used_cached_oauth_token = false; + auto buf = create_buffer(false, used_cached_oauth_token); + if (used_cached_oauth_token) + ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogAuthTokenCacheHits); + return buf; } catch (const DB::HTTPException & e) { @@ -730,7 +742,8 @@ DB::ReadWriteBufferFromHTTPPtr RestCatalog::createReadBuffer( || status == Poco::Net::HTTPResponse::HTTPStatus::HTTP_FORBIDDEN)) { ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogAuthTokenRefreshedOnUnauthorized); - return create_buffer(true); + bool used_cached_oauth_token_on_retry = false; + return create_buffer(true, used_cached_oauth_token_on_retry); } throw; } @@ -1409,9 +1422,9 @@ void RestCatalog::sendRequest(const String & endpoint, Poco::JSON::Object::Ptr r DB::HTTPHeaderEntries extra_headers; extra_headers.emplace_back("Content-Type", "application/json"); - auto create_buffer = [&](bool update_token) + auto create_buffer = [&](bool update_token, bool & used_cached_oauth_token) { - DB::HTTPHeaderEntries headers = getAuthHeaders(update_token, method, url, extra_headers, body_str); + DB::HTTPHeaderEntries headers = getAuthHeaders(update_token, method, url, extra_headers, body_str, &used_cached_oauth_token); headers.emplace_back("Content-Type", "application/json"); return DB::BuilderRWBufferFromHTTP(url) .withConnectionGroup(DB::HTTPConnectionGroupType::HTTP) @@ -1427,13 +1440,17 @@ void RestCatalog::sendRequest(const String & endpoint, Poco::JSON::Object::Ptr r try { - auto wb = create_buffer(false); + bool used_cached_oauth_token = false; + auto wb = create_buffer(false, used_cached_oauth_token); String response_str; if (!ignore_result) readJSONObjectPossiblyInvalid(response_str, *wb); else wb->ignoreAll(); + + if (used_cached_oauth_token) + ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogAuthTokenCacheHits); } catch (const DB::HTTPException & e) { @@ -1443,7 +1460,8 @@ void RestCatalog::sendRequest(const String & endpoint, Poco::JSON::Object::Ptr r || status == Poco::Net::HTTPResponse::HTTPStatus::HTTP_FORBIDDEN)) { ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogAuthTokenRefreshedOnUnauthorized); - auto wb = create_buffer(true); + bool used_cached_oauth_token_on_retry = false; + auto wb = create_buffer(true, used_cached_oauth_token_on_retry); String response_str; if (!ignore_result) diff --git a/src/Databases/DataLake/RestCatalog.h b/src/Databases/DataLake/RestCatalog.h index 1d83d45c559d..ae0c9aac0dbe 100644 --- a/src/Databases/DataLake/RestCatalog.h +++ b/src/Databases/DataLake/RestCatalog.h @@ -222,7 +222,8 @@ class RestCatalog : public ICatalog, public DB::WithContext const String & method = {}, const Poco::URI & url = {}, const DB::HTTPHeaderEntries & extra_headers = {}, - const String & body = {}) const; + const String & body = {}, + bool * used_cached_oauth_token = nullptr) const; void validateAuthHeaders(const DB::HTTPHeaderEntry & header) const; @@ -305,7 +306,8 @@ class BigLakeCatalog : public RestCatalog const String & method = {}, const Poco::URI & url = {}, const DB::HTTPHeaderEntries & extra_headers = {}, - const String & body = {}) const override; + const String & body = {}, + bool * used_cached_oauth_token = nullptr) const override; const std::string & getGoogleADCClientId() const { return google_adc_client_id; } const std::string & getGoogleADCClientSecret() const { return google_adc_client_secret; } diff --git a/src/Databases/DataLake/S3TablesCatalog.cpp b/src/Databases/DataLake/S3TablesCatalog.cpp index 07cd7e723da3..9589d1166b42 100644 --- a/src/Databases/DataLake/S3TablesCatalog.cpp +++ b/src/Databases/DataLake/S3TablesCatalog.cpp @@ -241,7 +241,8 @@ DB::HTTPHeaderEntries S3TablesCatalog::getAuthHeaders( const String & method, const Poco::URI & url, const DB::HTTPHeaderEntries & extra_headers, - const String & body) const + const String & body, + bool * /*used_cached_oauth_token*/) const { DB::HTTPHeaderEntries all_signed; signRequestWithAWSV4(method, url, extra_headers, body, *signer, region, "s3tables", all_signed); diff --git a/src/Databases/DataLake/S3TablesCatalog.h b/src/Databases/DataLake/S3TablesCatalog.h index aff432c1b679..d0fe76d4458d 100644 --- a/src/Databases/DataLake/S3TablesCatalog.h +++ b/src/Databases/DataLake/S3TablesCatalog.h @@ -51,7 +51,8 @@ class S3TablesCatalog final : public RestCatalog const String & method = {}, const Poco::URI & url = {}, const DB::HTTPHeaderEntries & extra_headers = {}, - const String & body = {}) const override; + const String & body = {}, + bool * used_cached_oauth_token = nullptr) const override; private: const String region; From 2b9b45dbf6dc2c79fbe0c0bac72c43b14ce4120f Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Mon, 17 Aug 2026 15:31:52 +0200 Subject: [PATCH 6/6] Fix build --- src/Databases/DataLake/RestCatalog.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/Databases/DataLake/RestCatalog.cpp b/src/Databases/DataLake/RestCatalog.cpp index d593d164f717..2d97cb1fc3ca 100644 --- a/src/Databases/DataLake/RestCatalog.cpp +++ b/src/Databases/DataLake/RestCatalog.cpp @@ -341,7 +341,7 @@ DB::HTTPHeaderEntries RestCatalog::getAuthHeaders( if (!client_id.empty()) { auto current = access_token.get(); - if (!current || update_token || access_token->isExpired()) + if (!current || update_token || current->isExpired()) { access_token.set(std::make_unique(retrieveAccessToken())); current = access_token.get();