Implementation:Kubeflow Pipelines Secret Access Sample

Knowledge Sources	Kubeflow_Pipelines
Domains	Pipeline_Sample, Security, GCP
Last Updated	2026-02-13 14:00 GMT

Overview

Sample pipeline demonstrating how to access cloud resources (GCS) from pipeline tasks using Kubernetes secrets for authentication.

Description

This sample (69 lines) shows two authentication patterns: (1) CLI-based access using gsutil with service account activation from `GOOGLE_APPLICATION_CREDENTIALS`, and (2) Python library-based access using google-cloud-storage with Application Default Credentials. Both components access GCS and rely on Kubernetes secrets mounted to provide GCP credentials.

Usage

Reference this sample when building pipelines that need to authenticate with cloud services using Kubernetes secrets.

Code Reference

Source Location

• Repository: Kubeflow_Pipelines
• File: samples/core/secret/secret.py
• Lines: 1-69

Signature

@dsl.component(packages_to_install=['google-cloud-storage'])
def gcs_list_buckets():
    """Lists all GCS buckets using google-cloud-storage library."""

@dsl.pipeline
def secret_op_pipeline():
    """Pipeline: gsutil CLI access + Python library access to GCS."""

Import

from kfp import dsl

I/O Contract

Inputs

Name	Type	Required	Description
GOOGLE_APPLICATION_CREDENTIALS	env var (from Secret)	Yes	Path to GCP service account key file
GCS URL	pipeline parameter	No	GCS path for gsutil listing

Outputs

Name	Type	Description
Compiled YAML	file	Pipeline IR YAML for submission to KFP

Usage Examples

Cloud Authentication Patterns

from kfp import dsl

# Pattern 1: CLI-based (gsutil)
gcs_list_items_op = dsl.load_component_from_text('''
implementation:
  container:
    image: google/cloud-sdk
    command: [gsutil, ls, {inputValue: url}]
''')

# Pattern 2: Python library-based
@dsl.component(packages_to_install=['google-cloud-storage'])
def gcs_list_buckets():
    from google.cloud import storage
    client = storage.Client()
    for bucket in client.list_buckets():
        print(bucket.name)

@dsl.pipeline
def secret_op_pipeline():
    task1 = gcs_list_items_op(url="gs://my-bucket/")
    task2 = gcs_list_buckets()

Related Pages

• Environment:Kubeflow_Pipelines_Python_SDK