From 726df66d628ea3499e22bcf61c50bd2507b6f8fa Mon Sep 17 00:00:00 2001 From: Gerd Zellweger Date: Mon, 3 Aug 2026 21:42:51 -0700 Subject: [PATCH 1/3] ci: authenticate runtime tests with OIDC instead of a stored API key The runtime integration tests drop FELDERA_API_KEY_CI_GKE_AMD64 and FELDERA_API_KEY_CI_GKE_ARM64. The job presents its GitHub OIDC token as the bearer credential, which each CI instance matches against a registered trust relationship, and the auth step confirms the instance accepts it before the suites start. testutils hands the SDK a callable rather than a token string when it runs under Actions. An ID token expires well inside a 50-minute test run, and the SDK re-resolves a callable per request and retries once on 401, so a token that lapses mid-run is replaced instead of failing every request after it. The suites also drop FELDERA_TLS_INSECURE, since the CI instances present publicly trusted certificates. test-integration-platform keeps it, talking to a docker service container over a self-signed certificate, and keeps creating an API key: that path is what covers key creation, against a local instance, so no cluster secret is involved. Signed-off-by: Gerd Zellweger --- .github/workflows/ci.yml | 5 +++ .../workflows/test-integration-runtime.yml | 21 ++++++++---- python/feldera/testutils.py | 33 ++++++++++++++++++- 3 files changed, 51 insertions(+), 8 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 647686de674..ada56567528 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -87,6 +87,11 @@ jobs: name: Integration Tests needs: [invoke-build-java] uses: ./.github/workflows/test-integration-runtime.yml + # A called workflow gets no more than the calling job grants, and the + # runtime tests authenticate to the CI instances with an OIDC token. + permissions: + contents: read + id-token: write secrets: inherit invoke-tests-java: diff --git a/.github/workflows/test-integration-runtime.yml b/.github/workflows/test-integration-runtime.yml index 351f3abaf40..a90e68fa9cb 100644 --- a/.github/workflows/test-integration-runtime.yml +++ b/.github/workflows/test-integration-runtime.yml @@ -9,6 +9,12 @@ on: description: "ID of the workflow run that uploaded the artifact" required: true +# Reaching the CI instances needs an OIDC token. A caller must grant the same +# permission, which ci.yml does on the job that invokes this workflow. +permissions: + contents: read + id-token: write + jobs: runtime-tests: if: ${{ !contains(vars.CI_SKIP_JOBS, 'runtime-tests') }} @@ -18,14 +24,10 @@ jobs: include: - runner: [gke-runners-amd64] feldera_host: ${{ vars.FELDERA_HOST_CI_GKE_AMD64 }} - feldera_api_key: FELDERA_API_KEY_CI_GKE_AMD64 - runner: [gke-runners-arm64] feldera_host: ${{ vars.FELDERA_HOST_CI_GKE_ARM64 }} - feldera_api_key: FELDERA_API_KEY_CI_GKE_ARM64 runs-on: ${{ matrix.runner }} env: - FELDERA_HOST: ${{ matrix.feldera_host }} - FELDERA_API_KEY: ${{ secrets[matrix.feldera_api_key] }} FELDERA_RUNTIME_VERSION: ${{ github.sha }} # GCS test bucket over the S3-interop endpoint; the env names are the # standard ones the tests read (python/tests/utils.py). @@ -45,13 +47,20 @@ jobs: - name: Checkout repository uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6 + # The instances identify this workflow by the audience it requests, so + # the value has to match the one their trust pins. + - name: Authenticate to Feldera + uses: feldera/oidc-auth-action@d1fce411d12fcda95a1f1a0bc441d2dc191e3f2c # v1.0.0 + with: + host: ${{ matrix.feldera_host }} + audience: feldera-ci-runtime-tests + - name: Python runtime tests if: ${{ vars.CI_DRY_RUN != 'true' && !contains(vars.CI_SKIP_JOBS, 'runtime-pytest') }} run: uv run --locked pytest -n ${{ vars.PYTEST_WORKERS }} tests/runtime --timeout=3600 --maxfail=1 -vv ${{ vars.PYTEST_EXTRA }} working-directory: python env: PYTHONPATH: ${{ github.workspace }}/python - FELDERA_TLS_INSECURE: true - name: Python runtime_aggtest test if: ${{ vars.CI_DRY_RUN != 'true' && !contains(vars.CI_SKIP_JOBS, 'runtime-aggtest') }} @@ -60,7 +69,6 @@ jobs: env: RUNTIME_AGGTEST_JOBS: ${{ vars.RUNTIME_AGGTEST_JOBS }} PYTHONPATH: ${{ github.workspace }}/python - FELDERA_TLS_INSECURE: true - name: Python workload tests if: ${{ vars.CI_DRY_RUN != 'true' && !contains(vars.CI_SKIP_JOBS, 'runtime-workload') }} @@ -68,7 +76,6 @@ jobs: working-directory: python env: PYTHONPATH: ${{ github.workspace }}/python - FELDERA_TLS_INSECURE: true KAFKA_BOOTSTRAP_SERVERS: ${{ vars.CI_KAFKA_BOOTSTRAP }} SCHEMA_REGISTRY_URL: ${{ vars.CI_SCHEMA_REGISTRY }} RUN_ID: 1 # we use this to run a single variant of the Kafka tests in test_kafka_avro.py diff --git a/python/feldera/testutils.py b/python/feldera/testutils.py index d3091a46ecf..400285e00a8 100644 --- a/python/feldera/testutils.py +++ b/python/feldera/testutils.py @@ -3,6 +3,8 @@ import logging import math import os +import urllib.parse +import urllib.request import platform import re import time @@ -44,6 +46,35 @@ def _get_effective_api_key(): return oidc_token if oidc_token else API_KEY +def _github_oidc_token() -> str: + """Mint a fresh GitHub Actions ID token for this job.""" + request_url = os.environ["ACTIONS_ID_TOKEN_REQUEST_URL"] + audience = os.environ.get("FELDERA_OIDC_AUDIENCE") + if audience: + request_url += "&audience=" + urllib.parse.quote(audience, safe="") + request = urllib.request.Request(request_url) + request.add_header( + "Authorization", f"bearer {os.environ['ACTIONS_ID_TOKEN_REQUEST_TOKEN']}" + ) + with urllib.request.urlopen(request, timeout=30) as response: + return json.load(response)["value"] + + +def _feldera_credential(): + """Bearer credential for the test client. + + Under GitHub Actions the credential is an ID token that expires well inside + a long test run, so hand the SDK the callable rather than a token: it + re-resolves per request and retries once on 401, which a fixed string + cannot do. The OIDC login flow, where configured, still wins. + """ + if os.environ.get("OIDC_TEST_ISSUER"): + return _get_effective_api_key() + if os.environ.get("ACTIONS_ID_TOKEN_REQUEST_URL"): + return _github_oidc_token + return API_KEY + + BASE_URL = os.environ.get("FELDERA_HOST") or "http://localhost:8080" FELDERA_REQUESTS_VERIFY = requests_verify_from_env() FELDERA_TEST_NUM_WORKERS = int(os.environ.get("FELDERA_TEST_NUM_WORKERS", "8")) @@ -87,7 +118,7 @@ def _ensure(self): if self._client is None: self._client = FelderaClient( connection_timeout=10, - api_key=_get_effective_api_key(), + api_key=_feldera_credential(), ) return self._client From c3906154725eeb15683e01aa6f27cf4dfccac0da Mon Sep 17 00:00:00 2001 From: Gerd Zellweger Date: Mon, 3 Aug 2026 21:57:35 -0700 Subject: [PATCH 2/3] ci: cache the OIDC token instead of minting one per request The SDK resolves a callable credential before every request, so the provider as written asked GitHub for a token on every API call. A suite that polls in loops across parallel workers sends enough of those to be throttled, and it arrives as a connection timeout inside the client rather than as anything naming the token endpoint. The token is now held until two minutes before its own expiry, which takes a 500-request run from 500 mints to one. An unreadable payload falls back to minting each time, so a token is never served past its expiry. Signed-off-by: Gerd Zellweger --- python/feldera/testutils.py | 42 ++++++++++++++++++++++++++++++++++--- 1 file changed, 39 insertions(+), 3 deletions(-) diff --git a/python/feldera/testutils.py b/python/feldera/testutils.py index 400285e00a8..a6f89b1864a 100644 --- a/python/feldera/testutils.py +++ b/python/feldera/testutils.py @@ -1,5 +1,6 @@ "Utility functions for writing tests against a Feldera instance." +import base64 import logging import math import os @@ -46,10 +47,28 @@ def _get_effective_api_key(): return oidc_token if oidc_token else API_KEY +# Token and the moment it stops being handed out, keyed by audience. +_oidc_token_cache: dict[str, tuple[str, float]] = {} + +# Re-mint this long before the token's own expiry, so a request issued just +# before the check cannot arrive after it. +_OIDC_REFRESH_MARGIN_S = 120.0 + + def _github_oidc_token() -> str: - """Mint a fresh GitHub Actions ID token for this job.""" + """A GitHub Actions ID token, re-minted shortly before it expires. + + The SDK resolves this before every request, so it has to be cheap. Minting + per request adds a round trip to GitHub each time, and a suite that polls in + loops across parallel workers sends enough of them to be throttled, which + arrives as a connection timeout rather than an error. + """ + audience = os.environ.get("FELDERA_OIDC_AUDIENCE", "") + cached = _oidc_token_cache.get(audience) + if cached is not None and time.time() < cached[1]: + return cached[0] + request_url = os.environ["ACTIONS_ID_TOKEN_REQUEST_URL"] - audience = os.environ.get("FELDERA_OIDC_AUDIENCE") if audience: request_url += "&audience=" + urllib.parse.quote(audience, safe="") request = urllib.request.Request(request_url) @@ -57,7 +76,24 @@ def _github_oidc_token() -> str: "Authorization", f"bearer {os.environ['ACTIONS_ID_TOKEN_REQUEST_TOKEN']}" ) with urllib.request.urlopen(request, timeout=30) as response: - return json.load(response)["value"] + token = json.load(response)["value"] + + _oidc_token_cache[audience] = (token, _token_expiry(token) - _OIDC_REFRESH_MARGIN_S) + return token + + +def _token_expiry(token: str) -> float: + """`exp` out of a JWT payload, or now if it cannot be read. + + An unreadable payload means every call re-mints, which is the old + behaviour: slow, but never serving a token past its expiry. + """ + try: + payload = token.split(".")[1] + payload += "=" * (-len(payload) % 4) + return float(json.loads(base64.urlsafe_b64decode(payload))["exp"]) + except Exception: + return time.time() def _feldera_credential(): From 7e9ba1a06877be8a82fe3dc8d3b8471aac21e5f4 Mon Sep 17 00:00:00 2001 From: Gerd Zellweger Date: Mon, 3 Aug 2026 23:26:09 -0700 Subject: [PATCH 3/3] ci: clarify the token cache comments Review feedback: name the units the cache and the refresh margin are measured in, say plainly what an unreadable payload does rather than referring to previous behaviour, and drop the asides about what a fixed credential cannot do and which credential wins. Signed-off-by: Gerd Zellweger --- python/feldera/testutils.py | 31 +++++++++++++++++-------------- 1 file changed, 17 insertions(+), 14 deletions(-) diff --git a/python/feldera/testutils.py b/python/feldera/testutils.py index a6f89b1864a..010ffb373d1 100644 --- a/python/feldera/testutils.py +++ b/python/feldera/testutils.py @@ -47,21 +47,21 @@ def _get_effective_api_key(): return oidc_token if oidc_token else API_KEY -# Token and the moment it stops being handed out, keyed by audience. +# Audience -> (token, seconds since the epoch after which it is re-minted). _oidc_token_cache: dict[str, tuple[str, float]] = {} -# Re-mint this long before the token's own expiry, so a request issued just -# before the check cannot arrive after it. -_OIDC_REFRESH_MARGIN_S = 120.0 +# Re-mint a token this many seconds before its own expiry, so a request issued +# just before the check cannot arrive after it. +_OIDC_REFRESH_MARGIN_SECONDS = 120.0 def _github_oidc_token() -> str: """A GitHub Actions ID token, re-minted shortly before it expires. - The SDK resolves this before every request, so it has to be cheap. Minting - per request adds a round trip to GitHub each time, and a suite that polls in - loops across parallel workers sends enough of them to be throttled, which - arrives as a connection timeout rather than an error. + The SDK resolves this before every request, so the implementation has to be + cheap. Minting per request adds a round trip to GitHub each time, and a + suite that polls in loops across parallel workers sends enough of them to be + throttled, which results in a connection timeout rather than an error. """ audience = os.environ.get("FELDERA_OIDC_AUDIENCE", "") cached = _oidc_token_cache.get(audience) @@ -78,15 +78,18 @@ def _github_oidc_token() -> str: with urllib.request.urlopen(request, timeout=30) as response: token = json.load(response)["value"] - _oidc_token_cache[audience] = (token, _token_expiry(token) - _OIDC_REFRESH_MARGIN_S) + _oidc_token_cache[audience] = ( + token, + _token_expiry(token) - _OIDC_REFRESH_MARGIN_SECONDS, + ) return token def _token_expiry(token: str) -> float: - """`exp` out of a JWT payload, or now if it cannot be read. + """`exp` from a JWT payload, in seconds since the epoch. - An unreadable payload means every call re-mints, which is the old - behaviour: slow, but never serving a token past its expiry. + Returns the current time if the payload cannot be read, which re-mints on + every call rather than serving a token past its expiry. """ try: payload = token.split(".")[1] @@ -101,8 +104,8 @@ def _feldera_credential(): Under GitHub Actions the credential is an ID token that expires well inside a long test run, so hand the SDK the callable rather than a token: it - re-resolves per request and retries once on 401, which a fixed string - cannot do. The OIDC login flow, where configured, still wins. + re-resolves per request and retries once on 401. A configured OIDC login + flow takes precedence over both. """ if os.environ.get("OIDC_TEST_ISSUER"): return _get_effective_api_key()