Kubeflow 1.3多用户环境下，GCP存储桶文件上传触发流水线的实现方法及权限拒绝错误排查

我来帮你解决这两个问题，先从你遇到的权限错误说起，再讲存储桶触发流水线的实现步骤：

一、解决list_experiments的权限拒绝错误 #

从你给出的错误信息来看，服务账号myServiceAccount@appspot.gserviceaccount.com在namespace1命名空间下没有权限执行list操作访问experiments资源。在Kubeflow 1.3的多用户环境中，权限通过Kubernetes RBAC和Kubeflow Profile机制管控，你可以按以下步骤修复：

1. 为服务账号分配Kubeflow Pipeline相关权限 #

你需要在namespace1命名空间中创建一个Role，授予list experiments的权限，再通过RoleBinding把这个角色绑定到你的服务账号上。

创建kfp-experiment-viewer-role.yaml文件：

apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  name: kfp-experiment-viewer
  namespace: namespace1
rules:
- apiGroups: ["pipelines.kubeflow.org"]
  resources: ["experiments"]
  verbs: ["list", "get"]

创建bind-kfp-experiment-viewer-to-sa.yaml文件：

apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
  name: bind-kfp-experiment-viewer-to-sa
  namespace: namespace1
subjects:
- kind: ServiceAccount
  name: myServiceAccount  # 注意：这里是K8s集群内的SA名称，需与GCP SA关联的K8s SA对应
  namespace: namespace1
roleRef:
  kind: Role
  name: kfp-experiment-viewer
  apiGroup: rbac.authorization.k8s.io

执行命令应用配置：

kubectl apply -f kfp-experiment-viewer-role.yaml
kubectl apply -f bind-kfp-experiment-viewer-to-sa.yaml

2. 关联GCP服务账号到Kubernetes服务账号（若使用Workload Identity） #

如果你的Cloud Function通过Workload Identity访问Kubernetes集群，需要完成GCP SA与K8s SA的绑定：

gcloud iam service-accounts add-iam-policy-binding myServiceAccount@appspot.gserviceaccount.com \
  --member="serviceAccount:你的GCP项目ID.svc.id.goog[namespace1/myServiceAccount]" \
  --role="roles/iam.workloadIdentityUser"

给K8s服务账号添加注释：

kubectl annotate serviceaccount myServiceAccount \
  --namespace namespace1 \
  iam.gke.io/gcp-service-account=myServiceAccount@appspot.gserviceaccount.com

二、GCP存储桶上传触发Kubeflow流水线运行 #

结合你给出的Cloud Function代码框架，我们可以实现文件上传时自动触发流水线，具体步骤如下：

1. 配置Cloud Function监听存储桶事件 #

在GCP控制台创建Cloud Function时，选择触发器类型为「Cloud Storage」，事件类型为「对象创建（最终）」，指定你要监听的目标存储桶。

2. 修改Cloud Function代码实现触发逻辑 #

在main函数中添加流水线触发逻辑，示例代码如下：

import kfp
import logging

# 配置你的Kubeflow参数
HOST = "https://你的Kubeflow Pipelines端点"
CLIENT_ID = "你的客户端ID"
NAMESPACE = "namespace1"
PIPELINE_ID = "要触发的流水线ID"
EXPERIMENT_NAME = "自动触发实验"  # 可指定现有实验或创建新实验

def main(data, context):
    # 初始化KFP客户端
    client = kfp.Client(host=HOST, client_id=CLIENT_ID, namespace=NAMESPACE)
    
    # 验证权限（需先解决前面的权限问题）
    try:
        experiments = client.list_experiments(namespace=NAMESPACE)
        logging.info(f"成功列出实验: {experiments}")
    except Exception as e:
        logging.error(f"列出实验失败: {str(e)}")
        raise e
    
    # 获取上传文件信息
    file_name = data['name']
    bucket_name = data['bucket']
    file_path = f"gs://{bucket_name}/{file_name}"
    logging.info(f"检测到新文件上传: {file_path}")
    
    # 触发流水线运行
    try:
        run = client.create_run_from_pipeline_id(
            pipeline_id=PIPELINE_ID,
            arguments={
                # 将文件路径作为参数传递给流水线
                'input_file': file_path
            },
            namespace=NAMESPACE,
            experiment_name=EXPERIMENT_NAME
        )
        logging.info(f"流水线运行触发成功，运行ID: {run.run_id}")
    except Exception as e:
        logging.error(f"触发流水线运行失败: {str(e)}")
        raise e

3. 确保Cloud Function服务账号的额外权限 #

• 赋予服务账号storage.objects.get权限，允许它读取存储桶中的文件（若流水线需要访问该文件）。
• 确保服务账号已具备Kubeflow流水线的运行权限（可通过添加kubeflow-pipelines-editor角色到RoleBinding中实现）。

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