Skip to content

Commit 7da5e2b

Browse files
committed
feat(spark): add namespace ServiceAccount OIDC auth
Allow each workflow namespace to select a Databricks service principal through an annotated ServiceAccount while retaining connector identity as the OIDC fallback. Signed-off-by: Rohit Sharma <rohitrsh@gmail.com>
1 parent e8ba14e commit 7da5e2b

4 files changed

Lines changed: 620 additions & 2 deletions

File tree

plugins/flytekit-spark/README.md

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -78,3 +78,41 @@ The connector deployment is responsible for projecting a JWT at one of these
7878
paths and configuring a matching federation policy for the Databricks service
7979
principal. PAT remains the default unless `oidc_federation` is selected
8080
explicitly.
81+
82+
#### Per-namespace ServiceAccount identity
83+
84+
When OIDC federation is selected, the connector first looks for one
85+
ServiceAccount in the workflow namespace with this configuration:
86+
87+
```yaml
88+
apiVersion: v1
89+
kind: ServiceAccount
90+
metadata:
91+
name: databricks-workload
92+
namespace: "<workflow-namespace>"
93+
labels:
94+
flyte.org/databricks-enabled: "true"
95+
annotations:
96+
flyte.org/databricks-client-id: "<service-principal-client-id>"
97+
flyte.org/databricks-audience: "databricks"
98+
```
99+
100+
If found, the connector creates a short-lived JWT for that ServiceAccount
101+
through the Kubernetes TokenRequest API and exchanges it for a Databricks
102+
token. If no matching ServiceAccount exists, connector-identity OIDC remains
103+
the fallback.
104+
105+
The connector ServiceAccount needs these additional Kubernetes permissions:
106+
107+
```yaml
108+
rules:
109+
- apiGroups: [""]
110+
resources: ["serviceaccounts"]
111+
verbs: ["get", "list"]
112+
- apiGroups: [""]
113+
resources: ["serviceaccounts/token"]
114+
verbs: ["create"]
115+
```
116+
117+
Configure exactly one Databricks-enabled ServiceAccount per workflow
118+
namespace. Multiple matching ServiceAccounts are treated as an error.

plugins/flytekit-spark/flytekitplugins/spark/connector.py

Lines changed: 39 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,7 @@ class DatabricksJobMetadata(ResourceMeta):
4343
client_id: Optional[str] = None
4444
oauth_secret_name: Optional[str] = None
4545
oidc_token_file: Optional[str] = None
46+
oidc_service_account: Optional[str] = None
4647
oidc_audience: Optional[str] = None
4748
namespace: Optional[str] = None
4849

@@ -307,15 +308,24 @@ async def create(
307308
raise RuntimeError(f"Failed to create databricks job with error: {response}")
308309

309310
logger.info(f"Successfully created Databricks job with run_id: {response['run_id']}")
311+
discovered = getattr(auth, "discovered", None)
312+
persisted_client_id = discovered.client_id if discovered is not None else auth.settings.client_id
313+
persisted_service_account = discovered.service_account if discovered is not None else None
314+
persisted_audience = (
315+
discovered.audience
316+
if discovered is not None
317+
else (auth.settings.oidc_audience if auth.auth_type == "oidc_federation" else None)
318+
)
310319
return DatabricksJobMetadata(
311320
databricks_instance=databricks_instance,
312321
run_id=str(response["run_id"]),
313322
auth_token=auth_token if auth.auth_type == "pat" else None,
314323
auth_type=auth.auth_type,
315-
client_id=auth.settings.client_id,
324+
client_id=persisted_client_id,
316325
oauth_secret_name=auth.settings.oauth_secret_name,
317326
oidc_token_file=(auth.settings.oidc_token_file if auth.auth_type == "oidc_federation" else None),
318-
oidc_audience=(auth.settings.oidc_audience if auth.auth_type == "oidc_federation" else None),
327+
oidc_service_account=persisted_service_account,
328+
oidc_audience=persisted_audience,
319329
namespace=namespace,
320330
)
321331

@@ -392,6 +402,7 @@ async def _request_with_auth(
392402
oauth_secret_name=resource_meta.oauth_secret_name,
393403
oidc_token_file=resource_meta.oidc_token_file,
394404
oidc_audience=resource_meta.oidc_audience,
405+
oidc_service_account=resource_meta.oidc_service_account,
395406
)
396407

397408
token = resource_meta.auth_token
@@ -437,6 +448,32 @@ def __init__(self):
437448
super(DatabricksConnector, self).__init__(task_type_name="databricks", metadata_type=DatabricksJobMetadata)
438449

439450

451+
def list_serviceaccounts_in_k8s(namespace: str, label_selector: Optional[str] = None) -> list:
452+
"""List labelled ServiceAccounts in a workflow namespace."""
453+
try:
454+
from kubernetes import client, config
455+
456+
try:
457+
config.load_incluster_config()
458+
except config.ConfigException:
459+
config.load_kube_config()
460+
461+
arguments = {"namespace": namespace}
462+
if label_selector:
463+
arguments["label_selector"] = label_selector
464+
response = client.CoreV1Api().list_namespaced_service_account(**arguments)
465+
return list(response.items or [])
466+
except ImportError:
467+
logger.warning("Kubernetes Python client is unavailable; skipping namespace " "ServiceAccount discovery")
468+
except Exception as error:
469+
logger.warning(
470+
"Unable to discover ServiceAccounts in namespace '%s': %s",
471+
namespace,
472+
error,
473+
)
474+
return []
475+
476+
440477
def get_secret_from_k8s(secret_name: str, secret_key: str, namespace: str) -> Optional[str]:
441478
"""Read a secret from Kubernetes using the Kubernetes Python client.
442479

0 commit comments

Comments
 (0)