o-nikolas commented on code in PR #75251:
URL: https://github.com/apache/airflow/pull/75251#discussion_r4361060051
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/cli/kubernetes_command.py:
##########
@@ -231,13 +252,50 @@ def cleanup_pods(args):
list_kwargs["_continue"] = continue_token
-def _delete_pod(name, namespace):
+def _get_team_executor_cls(team_name: str, executor: str | None) ->
type[KubernetesExecutor]:
+ """Return the class of the team's executor named ``executor``, or of its
default one."""
+ # Only the team's own executors: a global one doesn't connect with the
team's config.
+ names = [n for n in
ExecutorLoader.get_executor_names(validate_teams=False) if n.team_name ==
team_name]
+ if not names:
+ raise SystemExit(f"Team {team_name!r} has no executors configured.")
+ try:
+ if hasattr(ExecutorLoader, "resolve_executor_name"):
+ name = ExecutorLoader.resolve_executor_name(executor, team_name,
executor_names=names)
+ elif executor: # Airflow < 3.4
+ name = ExecutorLoader.lookup_executor_name_by_str(
+ executor, team_name=team_name, validate_teams=False
+ )
+ else:
+ name = names[0]
+ except UnknownExecutorException as e:
+ raise SystemExit(str(e))
+ executor_cls, _ = ExecutorLoader.import_executor_cls(name)
+ if not issubclass(executor_cls, KubernetesExecutor):
+ raise SystemExit(
+ f"Executor {name.module_path} of team {team_name!r} isn't a
KubernetesExecutor. "
+ "Pass --executor to pick one that is."
+ )
+ return executor_cls
+
+
+def _check_namespace_not_shared(team_name: str, namespace: str) -> None:
+ """Refuse a namespace another scope also uses, since their pods can't be
told apart."""
+ scopes = {n.team_name for n in
ExecutorLoader.get_executor_names(validate_teams=False)} - {team_name}
+ for other in sorted(scopes, key=lambda t: t or ""):
+ if conf.get("kubernetes_executor", "namespace", **_team_kwargs(other))
== namespace:
Review Comment:
This counts every scope that has any executor, then compares its
[kubernetes_executor] namespace. A scope with no Kubernetes executor never uses
that setting, but it still has the built-in value default. So with
CeleryExecutor;team_a=KubernetesExecutor and no namespace set, --team team_a
refuses ("also used by the global executors"), though Celery makes no pods
there. The "another cluster" hint is also wrong for that case.
Could we only count scopes that run a KubernetesExecutor? Something like:
```python
scopes = {
n.team_name
for n in ExecutorLoader.get_executor_names(validate_teams=False)
if n.team_name != team_name
and issubclass(ExecutorLoader.import_executor_cls(n)[0],
KubernetesExecutor)
}
```
The shared-with-global and shared-with-team tests would then need a
Kubernetes executor in the other scope, since the global setup and team_b only
run LocalExecutor there.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
