无需安装kubelogin，能否在Airflow中像EKSHook操作EKS那样操作Azure AKS Pod？

方案1：使用Airflow官方Azure Kubernetes Hook #

这是最贴合Airflow生态的方案，无需额外安装系统级工具，直接通过Airflow的Azure Provider完成认证与集群操作。

• 依赖准备：安装Airflow Azure Provider包

  pip install apache-airflow-providers-microsoft-azure
  

• 配置Airflow Azure Connection：
  • 创建类型为Azure Kubernetes Service的连接
  • 认证方式二选一：
    • Managed Identity：若Airflow部署在Azure环境（如AKS集群、Azure VM、App Service），直接启用系统/用户分配的托管标识，无需密钥。需确保该标识拥有AKS集群的访问权限（如Kubernetes Cluster Admin Role）
    • Service Principal：输入客户端ID、客户端密钥、租户ID，同样需为服务主体分配AKS的对应权限
• 示例操作代码：

  from airflow.providers.microsoft.azure.hooks.azure_kubernetes import AzureKubernetesHook
  from kubernetes.client import CoreV1Api
  
  def list_aks_pods():
      # 引用已配置的Airflow连接ID
      hook = AzureKubernetesHook(azure_conn_id="azure_aks_conn")
      api_client = hook.get_conn()
      v1_api = CoreV1Api(api_client)
      
      # 列出default命名空间下的Pod
      pods = v1_api.list_namespaced_pod(namespace="default")
      for pod in pods.items:
          print(f"Pod名称: {pod.metadata.name}, 状态: {pod.status.phase}")
  
  def create_aks_pod():
      hook = AzureKubernetesHook(azure_conn_id="azure_aks_conn")
      api_client = hook.get_conn()
      v1_api = CoreV1Api(api_client)
      
      # 构造Pod定义
      pod_manifest = {
          "apiVersion": "v1",
          "kind": "Pod",
          "metadata": {"name": "test-pod"},
          "spec": {
              "containers": [
                  {
                      "name": "test-container",
                      "image": "nginx:alpine",
                      "ports": [{"containerPort": 80}]
                  }
              ]
          }
      }
      
      # 在default命名空间创建Pod
      v1_api.create_namespaced_pod(namespace="default", body=pod_manifest)
  

方案2：直接用Kubernetes Python客户端+Azure Identity认证 #

若不想依赖Airflow的Hook，可通过Azure Identity库直接获取AKS API的访问令牌，手动构造kubeconfig后操作集群，完全脱离az cli和kubelogin。

• 依赖准备：

  pip install kubernetes azure-identity
  

• 示例操作代码：

  from kubernetes import client, config
  from azure.identity import DefaultAzureCredential
  
  def get_aks_api_client(aks_cluster_server, tenant_id, ca_cert_data):
      # 获取Azure AD令牌（针对AKS API的权限范围）
      credential = DefaultAzureCredential(tenant_id=tenant_id)
      token = credential.get_token("https://management.azure.com/.default").token
      
      # 构造kubeconfig并加载
      kubeconfig = {
          "apiVersion": "v1",
          "kind": "Config",
          "clusters": [{"name": "aks-cluster", "cluster": {"server": aks_cluster_server, "certificate-authority-data": ca_cert_data}}],
          "users": [{"name": "azure-user", "user": {"auth-provider": {"name": "azure", "config": {"access-token": token, "tenant-id": tenant_id}}}}],
          "contexts": [{"name": "aks-context", "context": {"cluster": "aks-cluster", "user": "azure-user"}}],
          "current-context": "aks-context"
      }
      config.load_kube_config_from_dict(kubeconfig)
      return client.CoreV1Api()
  
  def operate_aks_pods():
      # 替换为你的AKS API服务器地址、租户ID、集群CA证书内容（可从Azure Portal获取）
      v1_api = get_aks_api_client("https://my-aks-cluster.hcp.eastus.azmk8s.io", "xxxx-xxxx-xxxx-xxxx", "xxxxxx")
      
      # 列出Pod
      pods = v1_api.list_namespaced_pod("default")
      # 创建Pod（可复用方案1中的pod_manifest结构）
      # v1_api.create_namespaced_pod("default", body=pod_manifest)
  

关键注意事项 #

• 权限配置：无论是托管标识还是服务主体，必须被授予AKS集群的访问权限，推荐遵循最小权限原则，按需分配Kubernetes Cluster Admin Role或自定义角色
• 网络访问：Airflow部署环境需能访问AKS的API服务器端点，需确认NSG、防火墙等网络规则允许通信
• 版本兼容性：确保Airflow Providers Azure、kubernetes、azure-identity库的版本兼容，避免依赖冲突

内容的提问来源于stack exchange，提问作者Jun Kim