feat: support file path to pulsarconnection

Author: freeznet

api/v1alpha1/common.go

@@ -31,13 +31,18 @@ type SecretKeyRef struct {
 	Key  string `json:"key"`
 }
 
-// ValueOrSecretRef is a string or a secret reference of the authentication
+// ValueOrSecretRef is a string value, a secret reference, or a local file path for the authentication config.
 type ValueOrSecretRef struct {
 	// +optional
 	Value *string `json:"value,omitempty"`
 
 	// +optional
 	SecretRef *SecretKeyRef `json:"secretRef,omitempty"`
+
+	// File points to a local file path whose contents will be used as the value.
+	// This is useful when running the operator locally without creating Kubernetes secrets.
+	// +optional
+	File *string `json:"file,omitempty"`
 }
 
 // PulsarAuthentication defines the authentication configuration for Pulsar resources.

api/v1alpha1/zz_generated.deepcopy.go

@@ -143,6 +143,11 @@ spec:
                           For confidential clients, this would be the client secret.
                           For public clients using JWT authentication, this would be the path to the JSON credentials file.
                         properties:
+                          file:
+                            description: |-
+                              File points to a local file path whose contents will be used as the value.
+                              This is useful when running the operator locally without creating Kubernetes secrets.
+                            type: string
                           secretRef:
                             description: SecretKeyRef indicates a secret name and
                               key
@@ -187,6 +192,11 @@ spec:
                       This can be either a direct token value or a reference to a secret containing the token.
                       If using a secret, the token should be stored under the specified key in the secret.
                     properties:
+                      file:
+                        description: |-
+                          File points to a local file path whose contents will be used as the value.
+                          This is useful when running the operator locally without creating Kubernetes secrets.
+                        type: string
                       secretRef:
                         description: SecretKeyRef indicates a secret name and key
                         properties:

config/crd/bases/resource.streamnative.io_pulsarconnections.yaml

@@ -35,6 +35,8 @@ The `authentication` field supports two methods: JWT Token and OAuth2. Each meth
 |-------|-------------|------|----------|
 | `value` | Direct string value | `*string` | No |
 | `secretRef` | Reference to a Kubernetes Secret | `*SecretKeyRef` | No |
+| `file` | Local file path whose contents are used as the value | `*string` | No |
+| `file` | Local file path whose contents are used as the value | `*string` | No |
 
 #### SecretKeyRef
 
@@ -57,7 +59,7 @@ Note: Only one authentication method (either `token` or `oauth2`) should be spec
 
 ### JWT Token Authentication
 
-JWT Token authentication can be configured using either a direct value or a Kubernetes Secret reference.
+JWT Token authentication can be configured using a direct value, a local file path, or a Kubernetes Secret reference.
 
 #### Using a direct value
 
@@ -95,7 +97,7 @@ authentication:
 
 ### OAuth2 Authentication
 
-OAuth2 authentication can be configured using either a direct value or a Kubernetes Secret reference.
+OAuth2 authentication can be configured using a direct value, a local file path, or a Kubernetes Secret reference.
 
 #### Using a direct value
 
@@ -234,7 +236,7 @@ Note: Only one authentication method (either `token` or `oauth2`) should be spec
 
 ### JWT Token Authentication
 
-JWT Token authentication can be configured using either a direct value or a Kubernetes Secret reference.
+JWT Token authentication can be configured using a direct value, a local file path, or a Kubernetes Secret reference.
 
 #### Using a direct value
 
@@ -272,7 +274,7 @@ authentication:
 
 ### OAuth2 Authentication
 
-OAuth2 authentication can be configured using either a direct value or a Kubernetes Secret reference.
+OAuth2 authentication can be configured using a direct value, a local file path, or a Kubernetes Secret reference.
 
 #### Using a direct value
 
@@ -614,4 +616,4 @@ kubectl apply -f connection.yaml
 kubectl -n test delete pulsarconnection.resource.streamnative.io test-pulsar-connection
 ```
 
-Please be noticed, because the Pulsar Resources Operator are using the connection to manage pulsar resources, If you delete the pulsar connection, it will only be deleted after the resources CRs are deleted
+Please be noticed, because the Pulsar Resources Operator are using the connection to manage pulsar resources, If you delete the pulsar connection, it will only be deleted after the resources CRs are deleted

docs/pulsar_connection.md

@@ -353,13 +353,17 @@ type PulsarAdminConfig struct {
 	// Either Token or OAuth2 configuration must be provided
 	// The Token used for authentication.
 	Token string
+	// TokenFilePath points to a local file that contains the token.
+	TokenFilePath string
 
 	// OAuth2 related configuration used for authentication.
 	IssuerEndpoint string
 	ClientID       string
 	Audience       string
 	Key            string
-	Scope          string
+	// KeyFilePath points to a local file that contains the OAuth2 credentials JSON.
+	KeyFilePath string
+	Scope       string
 
 	// TLS Authentication related configuration
 	ClientCertificatePath    string
@@ -375,7 +379,7 @@ func NewPulsarAdmin(conf PulsarAdminConfig) (PulsarAdmin, error) {
 	var err error
 	var adminClient admin.Client
 
-	config := &config.Config{
+	cfg := &config.Config{
 		WebServiceURL:                 conf.WebServiceURL,
 		TLSAllowInsecureConnection:    conf.TLSAllowInsecureConnection,
 		TLSEnableHostnameVerification: conf.TLSEnableHostnameVerification,
@@ -385,10 +389,11 @@ func NewPulsarAdmin(conf PulsarAdminConfig) (PulsarAdmin, error) {
 	}
 
 	if conf.PulsarAPIVersion != nil {
-		config.PulsarAPIVersion = *conf.PulsarAPIVersion
+		cfg.PulsarAPIVersion = *conf.PulsarAPIVersion
 	}
 
-	if conf.Key != "" {
+	keyFilePath = conf.KeyFilePath
+	if keyFilePath == "" && conf.Key != "" {
 		keyFile, err = os.CreateTemp("", "oauth2-key-")
 		if err != nil {
 			return nil, err
@@ -398,12 +403,14 @@ func NewPulsarAdmin(conf PulsarAdminConfig) (PulsarAdmin, error) {
 		if err != nil {
 			return nil, err
 		}
+	}
 
-		config.IssuerEndpoint = conf.IssuerEndpoint
-		config.ClientID = conf.ClientID
-		config.Audience = conf.Audience
-		config.KeyFile = keyFilePath
-		config.Scope = conf.Scope
+	if keyFilePath != "" {
+		cfg.IssuerEndpoint = conf.IssuerEndpoint
+		cfg.ClientID = conf.ClientID
+		cfg.Audience = conf.Audience
+		cfg.KeyFile = keyFilePath
+		cfg.Scope = conf.Scope
 
 		oauthProvider, err := auth.NewAuthenticationOAuth2WithFlow(oauth2.Issuer{
 			IssuerEndpoint: conf.IssuerEndpoint,
@@ -417,27 +424,34 @@ func NewPulsarAdmin(conf PulsarAdminConfig) (PulsarAdmin, error) {
 		if err != nil {
 			return nil, err
 		}
-		adminClient, err = admin.NewPulsarClientWithAuthProvider(config, oauthProvider)
+		adminClient, err = admin.NewPulsarClientWithAuthProvider(cfg, oauthProvider)
+		if err != nil {
+			return nil, err
+		}
+	} else if conf.TokenFilePath != "" {
+		cfg.TokenFile = conf.TokenFilePath
+
+		adminClient, err = admin.New(cfg)
 		if err != nil {
 			return nil, err
 		}
 	} else if conf.Token != "" {
-		config.Token = conf.Token
+		cfg.Token = conf.Token
 
-		adminClient, err = admin.New(config)
+		adminClient, err = admin.New(cfg)
 		if err != nil {
 			return nil, err
 		}
 	} else if conf.ClientCertificatePath != "" {
-		config.AuthPlugin = auth.TLSPluginName
-		config.AuthParams = fmt.Sprintf("{\"tlsCertFile\": %q, \"tlsKeyFile\": %q}", conf.ClientCertificatePath, conf.ClientCertificateKeyPath)
+		cfg.AuthPlugin = auth.TLSPluginName
+		cfg.AuthParams = fmt.Sprintf("{\"tlsCertFile\": %q, \"tlsKeyFile\": %q}", conf.ClientCertificatePath, conf.ClientCertificateKeyPath)
 
-		adminClient, err = admin.New(config)
+		adminClient, err = admin.New(cfg)
 		if err != nil {
 			return nil, err
 		}
 	} else {
-		adminClient, err = admin.New(config)
+		adminClient, err = admin.New(cfg)
 		if err != nil {
 			return nil, err
 		}

pkg/admin/interface.go

@@ -319,10 +319,16 @@ func createParams(ctx context.Context, destConnection *resourcev1alpha1.PulsarCo
 		hasAuth := false
 		if auth := destConnection.Spec.Authentication; auth != nil {
 			if auth.Token != nil {
-				value, err := GetValue(ctx, client, destConnection.Namespace, auth.Token)
+				value, filePath, err := GetValue(ctx, client, destConnection.Namespace, auth.Token)
 				if err != nil {
 					return nil, err
 				}
+				if value == nil && filePath != nil {
+					value, err = valueFromFile(*filePath)
+					if err != nil {
+						return nil, err
+					}
+				}
 				if value != nil {
 					clusterParam.AuthPlugin = resourcev1alpha1.AuthPluginToken
 					clusterParam.AuthParameters = "token:" + *value
@@ -337,10 +343,19 @@ func createParams(ctx context.Context, destConnection *resourcev1alpha1.PulsarCo
 					ClientID:  oauth2.ClientID,
 				}
 				if oauth2.Key != nil {
-					value, err := GetValue(ctx, client, destConnection.Namespace, oauth2.Key)
+					value, filePath, err := GetValue(ctx, client, destConnection.Namespace, oauth2.Key)
 					if err != nil {
 						return nil, err
 					}
+					if value == nil && filePath != nil {
+						value, err = valueFromFile(*filePath)
+						if err != nil {
+							return nil, err
+						}
+					}
+					if value == nil && filePath == nil {
+						return nil, fmt.Errorf("OAuth2 key is empty")
+					}
 					if value != nil {
 						paramsJSON.PrivateKey = "data:application/json;base64," + base64.StdEncoding.EncodeToString([]byte(*value))
 						clusterParam.AuthPlugin = resourcev1alpha1.AuthPluginOAuth2

pkg/connection/reconcile_geo_replication.go

@@ -17,6 +17,7 @@ package connection
 import (
 	"context"
 	"fmt"
+	"os"
 
 	"github.com/apache/pulsar-client-go/pulsaradmin/pkg/admin/config"
 
@@ -270,21 +271,38 @@ func NewErrorCondition(generation int64, msg string) *metav1.Condition {
 	}
 }
 
-// GetValue get the authentication token value or secret
+// GetValue resolves the authentication value from direct input, secret, or file reference.
+// It returns either the resolved string value or a file path when provided.
 func GetValue(ctx context.Context, k8sClient client.Client, namespace string,
-	vRef *resourcev1alpha1.ValueOrSecretRef) (*string, error) {
+	vRef *resourcev1alpha1.ValueOrSecretRef) (*string, *string, error) {
+	if vRef == nil {
+		return nil, nil, nil
+	}
 	if value := vRef.Value; value != nil {
-		return value, nil
-	} else if ref := vRef.SecretRef; ref != nil {
+		return value, nil, nil
+	}
+	if filePath := vRef.File; filePath != nil {
+		return nil, filePath, nil
+	}
+	if ref := vRef.SecretRef; ref != nil {
 		secret := &corev1.Secret{}
 		if err := k8sClient.Get(ctx, types.NamespacedName{Namespace: namespace, Name: ref.Name}, secret); err != nil {
-			return nil, err
+			return nil, nil, err
 		}
 		if value, exists := secret.Data[ref.Key]; exists {
-			return ptr.To(string(value)), nil
+			return ptr.To(string(value)), nil, nil
 		}
 	}
-	return nil, nil
+	return nil, nil, nil
+}
+
+func valueFromFile(filePath string) (*string, error) {
+	content, err := os.ReadFile(filePath)
+	if err != nil {
+		return nil, err
+	}
+	value := string(content)
+	return &value, nil
 }
 
 // MakePulsarAdminConfig create pulsar admin configuration
@@ -329,10 +347,14 @@ func MakePulsarAdminConfig(ctx context.Context, connection *resourcev1alpha1.Pul
 	hasAuth := false
 	if authn := connection.Spec.Authentication; authn != nil {
 		if token := authn.Token; token != nil {
-			value, err := GetValue(ctx, k8sClient, connection.Namespace, token)
+			value, filePath, err := GetValue(ctx, k8sClient, connection.Namespace, token)
 			if err != nil {
 				return nil, err
 			}
+			if filePath != nil {
+				cfg.TokenFilePath = *filePath
+				hasAuth = true
+			}
 			if value != nil {
 				cfg.Token = *value
 				hasAuth = true
@@ -346,12 +368,20 @@ func MakePulsarAdminConfig(ctx context.Context, connection *resourcev1alpha1.Pul
 			if oauth2.Key == nil {
 				return nil, fmt.Errorf("oauth2 key must not be empty")
 			}
-			value, err := GetValue(ctx, k8sClient, connection.Namespace, oauth2.Key)
+			value, filePath, err := GetValue(ctx, k8sClient, connection.Namespace, oauth2.Key)
 			if err != nil {
 				return nil, err
 			}
+			if value == nil && filePath == nil {
+				return nil, fmt.Errorf("oauth2 key must not be empty")
+			}
 			if value != nil {
 				cfg.Key = *value
+				hasAuth = true
+			}
+			if filePath != nil {
+				cfg.KeyFilePath = *filePath
+				hasAuth = true
 			}
 		}
 		if tls := authn.TLS; tls != nil {