diff --git a/.ci/olm-tests/catalog.yml b/.ci/olm-tests/catalog.yml index 9836a3edc..cf441d95a 100644 --- a/.ci/olm-tests/catalog.yml +++ b/.ci/olm-tests/catalog.yml @@ -5,4 +5,4 @@ metadata: namespace: olm spec: sourceType: grpc - image: localhost:5000/streamnativeio/function-mesh-catalog:v0.27.0 + image: localhost:5000/streamnativeio/function-mesh-catalog:v0.27.2 diff --git a/.ci/olm-tests/subs.yml b/.ci/olm-tests/subs.yml index 4fabf104f..2cb6c0290 100644 --- a/.ci/olm-tests/subs.yml +++ b/.ci/olm-tests/subs.yml @@ -6,6 +6,6 @@ metadata: spec: channel: alpha name: function-mesh - startingCSV: function-mesh.v0.27.0 + startingCSV: function-mesh.v0.27.2 source: my-test-catalog sourceNamespace: olm diff --git a/.ci/tests/integration/cases/quoted-function-details/manifests.yaml b/.ci/tests/integration/cases/quoted-function-details/manifests.yaml new file mode 100644 index 000000000..6ae32f25c --- /dev/null +++ b/.ci/tests/integration/cases/quoted-function-details/manifests.yaml @@ -0,0 +1,46 @@ +apiVersion: compute.functionmesh.io/v1alpha1 +kind: Function +metadata: + name: function-details-quoted-sample + namespace: default +spec: + image: streamnative/pulsar-functions-java-sample:3.2.2.1 + className: org.apache.pulsar.functions.api.examples.ExclamationFunction + replicas: 1 + logTopic: persistent://public/default/logging-function-details-quoted-logs + funcConfig: + timePartitionPattern: "yyyy-MM-dd/HH'h'-mm'm'" + note: "quoted payload should survive shell parsing" + input: + topics: + - persistent://public/default/input-function-details-quoted-topic + typeClassName: java.lang.String + output: + topic: persistent://public/default/output-function-details-quoted-topic + typeClassName: java.lang.String + resources: + requests: + cpu: 50m + memory: 1G + limits: + memory: 1.1G + pulsar: + pulsarConfig: "test-pulsar" + tlsConfig: + enabled: false + allowInsecure: false + hostnameVerification: true + certSecretName: sn-platform-tls-broker + certSecretKey: "" + java: + jar: /pulsar/examples/api-examples.jar + clusterName: test + autoAck: true +--- +apiVersion: v1 +kind: ConfigMap +metadata: + name: test-pulsar +data: + webServiceURL: http://sn-platform-pulsar-broker.default.svc.cluster.local:8080 + brokerServiceURL: pulsar://sn-platform-pulsar-broker.default.svc.cluster.local:6650 diff --git a/.ci/tests/integration/cases/quoted-function-details/verify.sh b/.ci/tests/integration/cases/quoted-function-details/verify.sh new file mode 100644 index 000000000..78b19e50c --- /dev/null +++ b/.ci/tests/integration/cases/quoted-function-details/verify.sh @@ -0,0 +1,116 @@ +#!/usr/bin/env bash +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +# + +set -e + +E2E_DIR=$(dirname "$0") +BASE_DIR=$(cd "${E2E_DIR}"/../../../../..; pwd) +PULSAR_NAMESPACE=${PULSAR_NAMESPACE:-"default"} +PULSAR_RELEASE_NAME=${PULSAR_RELEASE_NAME:-"sn-platform"} +E2E_KUBECONFIG=${E2E_KUBECONFIG:-"/tmp/e2e-k8s.config"} +MANIFESTS_FILE="${BASE_DIR}"/.ci/tests/integration/cases/quoted-function-details/manifests.yaml + +source "${BASE_DIR}"/.ci/helm.sh + +FUNCTION_NAME=function-details-quoted-sample +STS_NAME=${FUNCTION_NAME}-function + +if [ ! "$KUBECONFIG" ]; then + export KUBECONFIG=${E2E_KUBECONFIG} +fi + +if [ -z "${FUNCTION_NAME}" ]; then + echo "function name is empty" + exit 1 +fi + +cleanup() { + kubectl delete -f "${MANIFESTS_FILE}" > /dev/null 2>&1 || true +} + +require_fragment() { + fragment=$1 + message=$2 + + printf '%s' "${START_COMMAND}" | grep -F -- "${fragment}" > /dev/null 2>&1 || { + echo "${message}" + echo "${START_COMMAND}" + exit 1 + } +} + +trap cleanup EXIT + +kubectl apply -f "${MANIFESTS_FILE}" > /dev/null 2>&1 + +START_COMMAND="" +for _ in $(seq 1 24); do + START_COMMAND=$(kubectl get statefulset "${STS_NAME}" -o jsonpath='{.spec.template.spec.containers[0].command[2]}' 2> /dev/null || true) + if [ -n "${START_COMMAND}" ]; then + break + fi + sleep 5 +done + +if [ -z "${START_COMMAND}" ]; then + echo "statefulset command is not available" + exit 1 +fi + +unsafe_pattern_fragment=$(cat <<'EOF' +timePartitionPattern\":\"yyyy-MM-dd/HH'h'-mm'm' +EOF +) +escaped_h_fragment=$(cat <<'EOF' +'"'"'h'"'"' +EOF +) +escaped_m_fragment=$(cat <<'EOF' +'"'"'m'"'"' +EOF +) + +require_fragment "timePartitionPattern" "function details command does not include the quoted config key" +if printf '%s' "${START_COMMAND}" | grep -F -- "${unsafe_pattern_fragment}" > /dev/null 2>&1; then + echo "function details command still contains raw single quotes" + echo "${START_COMMAND}" + exit 1 +fi +require_fragment "${escaped_h_fragment}" "function details command does not escape the quoted h segment" +require_fragment "${escaped_m_fragment}" "function details command does not escape the quoted m segment" + +verify_fm_result=$(ci::verify_function_mesh "${FUNCTION_NAME}" 2>&1) +if [ $? -ne 0 ]; then + echo "${verify_fm_result}" + exit 1 +fi + +verify_java_result=$(NAMESPACE=${PULSAR_NAMESPACE} CLUSTER=${PULSAR_RELEASE_NAME} ci::verify_exclamation_function \ + persistent://public/default/input-function-details-quoted-topic \ + persistent://public/default/output-function-details-quoted-topic \ + test-message \ + test-message! \ + 10 2>&1) +if [ $? -eq 0 ]; then + echo "e2e-test: ok" | yq eval - +else + echo "${verify_java_result}" + exit 1 +fi diff --git a/.ci/tests/integration/e2e.yaml b/.ci/tests/integration/e2e.yaml index 58bd413e5..fa87a1e30 100644 --- a/.ci/tests/integration/e2e.yaml +++ b/.ci/tests/integration/e2e.yaml @@ -133,6 +133,8 @@ verify: expected: expected.data.yaml - query: timeout 5m bash .ci/tests/integration/cases/java-function/verify.sh expected: expected.data.yaml + - query: timeout 5m bash .ci/tests/integration/cases/quoted-function-details/verify.sh + expected: expected.data.yaml - query: bash .ci/tests/integration/cases/java-function-vpa/verify.sh expected: expected.data.yaml - query: bash .ci/tests/integration/cases/reconciliation/verify.sh diff --git a/.gitignore b/.gitignore index 138f2ef41..ba204de7a 100644 --- a/.gitignore +++ b/.gitignore @@ -37,3 +37,4 @@ index_build_*/ config/crd/bases/vpa-v1-crd.yaml Listeners .tool-versions +/tmp diff --git a/.golangci.yml b/.golangci.yml index d876b5fa9..020cf2930 100644 --- a/.golangci.yml +++ b/.golangci.yml @@ -2,7 +2,7 @@ # https://github.com/golangci/golangci-lint#install # We can execute `golangci-lint run` for code checking. run: - deadline: 20m + timeout: 20m linters: disable-all: true enable: diff --git a/CHANGELOG.md b/CHANGELOG.md index 4676ae45c..525700f7d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,15 @@ +# Function Mesh v0.27.2 Release Notes + +## v0.27.2 What's New + +* [Runtime] fix: ensure proper shell quoting for parameters ([#837](https://github.com/streamnative/function-mesh/pull/837)) + +# Function Mesh v0.27.1 Release Notes + +## v0.27.1 What's New + +* [Runtime] feat: export DownloadServiceConfig and GeneratePythonLogConfigCommand ([#835](https://github.com/streamnative/function-mesh/pull/835)) + # Function Mesh v0.27.0 Release Notes ## v0.27.0 What's New diff --git a/Makefile b/Makefile index 5f000e7f4..340df68db 100644 --- a/Makefile +++ b/Makefile @@ -1,5 +1,5 @@ # Current Operator version -VERSION ?= 0.27.0 +VERSION ?= 0.27.2 # Default image tag DOCKER_REPO := $(if $(DOCKER_REPO),$(DOCKER_REPO),streamnative) OPERATOR_IMG ?= ${DOCKER_REPO}/function-mesh:v$(VERSION) diff --git a/README.md b/README.md index b4cc5d612..84e3f1a67 100644 --- a/README.md +++ b/README.md @@ -4,7 +4,7 @@ A Kubernetes-Native way to run pulsar functions, connectors and composed functio ## Install ```bash -curl -sSL https://github.com/streamnative/function-mesh/releases/download/v0.27.0/install.sh | bash +curl -sSL https://github.com/streamnative/function-mesh/releases/download/v0.27.2/install.sh | bash ``` The above command installs all the CRDs, required service account configuration, and all function-mesh operator components. Before you start running a function-mesh example, verify if Function Mesh is installed correctly. diff --git a/api/compute/v1alpha1/common.go b/api/compute/v1alpha1/common.go index 5d022c59b..998b21702 100644 --- a/api/compute/v1alpha1/common.go +++ b/api/compute/v1alpha1/common.go @@ -127,7 +127,7 @@ func (o *OAuth2Config) GetMountFile() string { } func (o *OAuth2Config) AuthenticationParameters() string { - return fmt.Sprintf(`'{"credentials_url":"file://%s","privateKey":"%s","private_key":"%s","issuerUrl":"%s","issuer_url":"%s","audience":"%s","scope":"%s"}'`, o.GetMountFile(), o.GetMountFile(), o.GetMountFile(), o.IssuerURL, o.IssuerURL, o.Audience, o.Scope) + return fmt.Sprintf(`{"credentials_url":"file://%s","privateKey":"%s","private_key":"%s","issuerUrl":"%s","issuer_url":"%s","audience":"%s","scope":"%s"}`, o.GetMountFile(), o.GetMountFile(), o.GetMountFile(), o.IssuerURL, o.IssuerURL, o.Audience, o.Scope) } type GenericAuth struct { diff --git a/api/go.mod b/api/go.mod index 69f920476..bf03c7f74 100644 --- a/api/go.mod +++ b/api/go.mod @@ -1,6 +1,6 @@ module github.com/streamnative/function-mesh/api -go 1.25.8 +go 1.25 require ( k8s.io/api v0.30.9 diff --git a/charts/function-mesh-operator/Chart.yaml b/charts/function-mesh-operator/Chart.yaml index bff8df6d2..26d51eaba 100644 --- a/charts/function-mesh-operator/Chart.yaml +++ b/charts/function-mesh-operator/Chart.yaml @@ -19,12 +19,12 @@ type: application # This is the chart version. This version number should be incremented each time you make changes # to the chart and its templates, including the app version. # Versions are expected to follow Semantic Versioning (https://semver.org/) -version: 0.2.40 +version: 0.2.42 # This is the version number of the application being deployed. This version number should be # incremented each time you make changes to the application. Versions are not expected to # follow Semantic Versioning. They should reflect the version the application is using. -appVersion: 0.27.0 +appVersion: 0.27.2 home: https://github.com/streamnative/function-mesh sources: @@ -33,5 +33,5 @@ sources: dependencies: - name: admission-webhook condition: admissionWebhook.enabled - version: 0.2.40 + version: 0.2.42 # repository: https://charts.functionmesh.io/ diff --git a/charts/function-mesh-operator/README.md b/charts/function-mesh-operator/README.md index 66cf0665c..b9d8c6ec5 100644 --- a/charts/function-mesh-operator/README.md +++ b/charts/function-mesh-operator/README.md @@ -1,6 +1,6 @@ # function-mesh-operator -![Version: 0.2.40](https://img.shields.io/badge/Version-0.2.40-informational?style=flat-square) ![Type: application](https://img.shields.io/badge/Type-application-informational?style=flat-square) ![AppVersion: 0.27.0](https://img.shields.io/badge/AppVersion-0.27.0-informational?style=flat-square) +![Version: 0.2.42](https://img.shields.io/badge/Version-0.2.42-informational?style=flat-square) ![Type: application](https://img.shields.io/badge/Type-application-informational?style=flat-square) ![AppVersion: 0.27.2](https://img.shields.io/badge/AppVersion-0.27.2-informational?style=flat-square) function mesh operator Helm chart for Kubernetes @@ -20,7 +20,7 @@ function mesh operator Helm chart for Kubernetes | Repository | Name | Version | |------------|------|---------| -| | admission-webhook | 0.2.40 | +| | admission-webhook | 0.2.42 | ## Values @@ -52,7 +52,7 @@ function mesh operator Helm chart for Kubernetes | imagePullPolicy | string | `"IfNotPresent"` | | | imagePullSecrets | list | `[]` | | | installation.namespace | string | `"function-mesh-system"` | | -| operatorImage | string | `"streamnative/function-mesh:v0.27.0"` | | +| operatorImage | string | `"streamnative/function-mesh:v0.27.2"` | | | rbac.create | bool | `true` | | ---------------------------------------------- diff --git a/charts/function-mesh-operator/charts/admission-webhook/Chart.yaml b/charts/function-mesh-operator/charts/admission-webhook/Chart.yaml index e65b92800..68d850777 100644 --- a/charts/function-mesh-operator/charts/admission-webhook/Chart.yaml +++ b/charts/function-mesh-operator/charts/admission-webhook/Chart.yaml @@ -19,13 +19,13 @@ type: application # This is the chart version. This version number should be incremented each time you make changes # to the chart and its templates, including the app version. # Versions are expected to follow Semantic Versioning (https://semver.org/) -version: 0.2.40 +version: 0.2.42 # This is the version number of the application being deployed. This version number should be # incremented each time you make changes to the application. Versions are not expected to # follow Semantic Versioning. They should reflect the version the application is using. # It is recommended to use it with quotes. -appVersion: 0.27.0 +appVersion: 0.27.2 home: https://github.com/streamnative/function-mesh sources: diff --git a/charts/function-mesh-operator/charts/admission-webhook/README.md b/charts/function-mesh-operator/charts/admission-webhook/README.md index c56b36e59..a4d1aeb40 100644 --- a/charts/function-mesh-operator/charts/admission-webhook/README.md +++ b/charts/function-mesh-operator/charts/admission-webhook/README.md @@ -1,6 +1,6 @@ # admission-webhook -![Version: 0.2.40](https://img.shields.io/badge/Version-0.2.40-informational?style=flat-square) ![Type: application](https://img.shields.io/badge/Type-application-informational?style=flat-square) ![AppVersion: 0.27.0](https://img.shields.io/badge/AppVersion-0.27.0-informational?style=flat-square) +![Version: 0.2.42](https://img.shields.io/badge/Version-0.2.42-informational?style=flat-square) ![Type: application](https://img.shields.io/badge/Type-application-informational?style=flat-square) ![AppVersion: 0.27.2](https://img.shields.io/badge/AppVersion-0.27.2-informational?style=flat-square) function mesh admission webhook Helm chart for Kubernetes diff --git a/charts/function-mesh-operator/charts/admission-webhook/templates/crd-compute.functionmesh.io-functionmeshes.yaml b/charts/function-mesh-operator/charts/admission-webhook/templates/crd-compute.functionmesh.io-functionmeshes.yaml index 0392a0e74..ced49f7eb 100644 --- a/charts/function-mesh-operator/charts/admission-webhook/templates/crd-compute.functionmesh.io-functionmeshes.yaml +++ b/charts/function-mesh-operator/charts/admission-webhook/templates/crd-compute.functionmesh.io-functionmeshes.yaml @@ -4110,8 +4110,6 @@ spec: required: - actualWindowFunctionClassName type: object - required: - - "" type: object type: array sinks: @@ -7871,8 +7869,6 @@ spec: - name type: object type: array - required: - - "" type: object type: array sources: @@ -11613,8 +11609,6 @@ spec: - name type: object type: array - required: - - "" type: object type: array type: object diff --git a/charts/function-mesh-operator/charts/admission-webhook/templates/crd-compute.functionmesh.io-functions.yaml b/charts/function-mesh-operator/charts/admission-webhook/templates/crd-compute.functionmesh.io-functions.yaml index 490a10b1d..06f529ba9 100644 --- a/charts/function-mesh-operator/charts/admission-webhook/templates/crd-compute.functionmesh.io-functions.yaml +++ b/charts/function-mesh-operator/charts/admission-webhook/templates/crd-compute.functionmesh.io-functions.yaml @@ -4129,8 +4129,6 @@ spec: required: - actualWindowFunctionClassName type: object - required: - - "" type: object status: properties: diff --git a/charts/function-mesh-operator/charts/admission-webhook/templates/crd-compute.functionmesh.io-sinks.yaml b/charts/function-mesh-operator/charts/admission-webhook/templates/crd-compute.functionmesh.io-sinks.yaml index 3b8c22fab..a3fce9cd4 100644 --- a/charts/function-mesh-operator/charts/admission-webhook/templates/crd-compute.functionmesh.io-sinks.yaml +++ b/charts/function-mesh-operator/charts/admission-webhook/templates/crd-compute.functionmesh.io-sinks.yaml @@ -3802,8 +3802,6 @@ spec: - name type: object type: array - required: - - "" type: object status: properties: diff --git a/charts/function-mesh-operator/charts/admission-webhook/templates/crd-compute.functionmesh.io-sources.yaml b/charts/function-mesh-operator/charts/admission-webhook/templates/crd-compute.functionmesh.io-sources.yaml index bcaa0f752..641e32dc6 100644 --- a/charts/function-mesh-operator/charts/admission-webhook/templates/crd-compute.functionmesh.io-sources.yaml +++ b/charts/function-mesh-operator/charts/admission-webhook/templates/crd-compute.functionmesh.io-sources.yaml @@ -3783,8 +3783,6 @@ spec: - name type: object type: array - required: - - "" type: object status: properties: diff --git a/charts/function-mesh-operator/values.yaml b/charts/function-mesh-operator/values.yaml index d0843e01e..4fcd9dc6a 100644 --- a/charts/function-mesh-operator/values.yaml +++ b/charts/function-mesh-operator/values.yaml @@ -11,7 +11,7 @@ installation: namespace: function-mesh-system # operatorImage is Function Mesh Operator image -operatorImage: streamnative/function-mesh:v0.27.0 +operatorImage: streamnative/function-mesh:v0.27.2 imagePullPolicy: IfNotPresent imagePullSecrets: [] diff --git a/controllers/spec/common.go b/controllers/spec/common.go index 8c799cc84..1060637ee 100644 --- a/controllers/spec/common.go +++ b/controllers/spec/common.go @@ -216,48 +216,48 @@ func prefixedDownloadCommandEnvNames(prefix string) downloadCommandEnvNames { } } -type downloadServiceConfig struct { +type DownloadServiceConfig struct { pulsar *v1alpha1.PulsarMessaging envPrefix string - envNames downloadCommandEnvNames - oauth2Mount string + EnvNames downloadCommandEnvNames + Oauth2Mount string tlsMount string } -func NewDownloadServiceConfig(packageService, messaging *v1alpha1.PulsarMessaging) downloadServiceConfig { +func NewDownloadServiceConfig(packageService, messaging *v1alpha1.PulsarMessaging) DownloadServiceConfig { if packageService != nil { - return downloadServiceConfig{ + return DownloadServiceConfig{ pulsar: packageService, envPrefix: PackageServiceEnvPrefix, - envNames: prefixedDownloadCommandEnvNames(PackageServiceEnvPrefix), - oauth2Mount: PackageOAuth2MountPath, + EnvNames: prefixedDownloadCommandEnvNames(PackageServiceEnvPrefix), + Oauth2Mount: PackageOAuth2MountPath, tlsMount: PackageTLSMountPath, } } - return downloadServiceConfig{ + return DownloadServiceConfig{ pulsar: messaging, - envNames: defaultDownloadCommandEnvNames(), - oauth2Mount: "", + EnvNames: defaultDownloadCommandEnvNames(), + Oauth2Mount: "", tlsMount: "", } } -func (d downloadServiceConfig) authProvided() bool { +func (d DownloadServiceConfig) AuthProvided() bool { return d.pulsar != nil && d.pulsar.AuthSecret != "" } -func (d downloadServiceConfig) tlsProvided() bool { +func (d DownloadServiceConfig) TLSProvided() bool { return d.pulsar != nil && d.pulsar.TLSSecret != "" } -func (d downloadServiceConfig) authConfig() *v1alpha1.AuthConfig { +func (d DownloadServiceConfig) AuthConfig() *v1alpha1.AuthConfig { if d.pulsar == nil { return nil } return d.pulsar.AuthConfig } -func (d downloadServiceConfig) tlsConfig() TLSConfig { +func (d DownloadServiceConfig) TLSConfig() TLSConfig { if d.pulsar == nil || d.pulsar.TLSConfig == nil { return nil } @@ -270,7 +270,7 @@ func (d downloadServiceConfig) tlsConfig() TLSConfig { } } -func (d downloadServiceConfig) envFrom() []corev1.EnvFromSource { +func (d DownloadServiceConfig) EnvFrom() []corev1.EnvFromSource { if d.pulsar == nil { return nil } @@ -352,7 +352,7 @@ func MakeHeadlessServiceName(serviceName string) string { func MakeStatefulSet(objectMeta *metav1.ObjectMeta, replicas *int32, downloaderImage string, container *corev1.Container, volumes []corev1.Volume, labels map[string]string, policy v1alpha1.PodPolicy, runtimeMessaging *v1alpha1.PulsarMessaging, - downloadConfig downloadServiceConfig, + downloadConfig DownloadServiceConfig, javaRuntime *v1alpha1.JavaRuntime, pythonRuntime *v1alpha1.PythonRuntime, goRuntime *v1alpha1.GoRuntime, env []corev1.EnvVar, logTopic, filebeatImage string, logTopicAgent v1alpha1.LogTopicAgent, definedVolumeMounts []corev1.VolumeMount, volumeClaimTemplates []corev1.PersistentVolumeClaim, @@ -390,16 +390,16 @@ func MakeStatefulSet(objectMeta *metav1.ObjectMeta, replicas *int32, downloaderI // mount auth and tls related VolumeMounts when download package from pulsar if !hasHTTPPrefix(downloadPath) { - if authConfig := downloadConfig.authConfig(); authConfig != nil && authConfig.OAuth2Config != nil { + if authConfig := downloadConfig.AuthConfig(); authConfig != nil && authConfig.OAuth2Config != nil { oauth2MountPath := authConfig.OAuth2Config.GetMountPath() - if downloadConfig.oauth2Mount != "" { - oauth2MountPath = downloadConfig.oauth2Mount + if downloadConfig.Oauth2Mount != "" { + oauth2MountPath = downloadConfig.Oauth2Mount } volumeMounts = appendVolumeMountIfNotExists(volumeMounts, generateVolumeMountFromOAuth2ConfigWithMountPath(authConfig.OAuth2Config, oauth2MountPath)) podVolumes = appendVolumeIfNotExists(podVolumes, generateVolumeFromOAuth2Config(authConfig.OAuth2Config)) } - tlsConfig := downloadConfig.tlsConfig() + tlsConfig := downloadConfig.TLSConfig() if !isNilTLSConfig(tlsConfig) && tlsConfig.HasSecretVolume() { volumeMounts = appendVolumeMountIfNotExists(volumeMounts, generateVolumeMountFromTLSConfig(tlsConfig)) podVolumes = appendVolumeIfNotExists(podVolumes, generateVolumeFromTLSConfig(tlsConfig)) @@ -419,15 +419,15 @@ func MakeStatefulSet(objectMeta *metav1.ObjectMeta, replicas *int32, downloaderI Image: image, Command: []string{"sh", "-c", strings.Join(GetDownloadCommandWithEnv(downloadPath, componentPackage, true, true, - downloadConfig.authProvided(), downloadConfig.tlsProvided(), downloadConfig.tlsConfig(), - downloadConfig.authConfig(), downloadConfig.envNames, downloadConfig.oauth2Mount), " ")}, + downloadConfig.AuthProvided(), downloadConfig.TLSProvided(), downloadConfig.TLSConfig(), + downloadConfig.AuthConfig(), downloadConfig.EnvNames, downloadConfig.Oauth2Mount), " ")}, VolumeMounts: volumeMounts, ImagePullPolicy: corev1.PullIfNotPresent, Env: []corev1.EnvVar{{ Name: "HOME", Value: "/tmp", }}, - EnvFrom: downloadConfig.envFrom(), + EnvFrom: downloadConfig.EnvFrom(), } podVolumes = appendVolumeIfNotExists(podVolumes, corev1.Volume{ Name: DownloaderVolume, @@ -670,7 +670,7 @@ func GenerateAffinity(affinity *corev1.Affinity, labels map[string]string, disab } func MakeJavaFunctionCommand(downloadPath, packageFile, name, clusterName, generateLogConfigCommand, logLevel, details, extraDependenciesDir, connectorsDirectory, uid string, - memory *resource.Quantity, javaOpts []string, hasPulsarctl, hasWget bool, downloadConfig downloadServiceConfig, + memory *resource.Quantity, javaOpts []string, hasPulsarctl, hasWget bool, downloadConfig DownloadServiceConfig, authProvided, tlsProvided bool, secretMaps map[string]v1alpha1.SecretRef, state *v1alpha1.Stateful, tlsConfig TLSConfig, authConfig *v1alpha1.AuthConfig, @@ -682,15 +682,15 @@ func MakeJavaFunctionCommand(downloadPath, packageFile, name, clusterName, gener if downloadPath != "" && !utils.EnableInitContainers { // prepend download command if the downPath is provided downloadCommand := strings.Join(GetDownloadCommandWithEnv(downloadPath, packageFile, hasPulsarctl, hasWget, - downloadConfig.authProvided(), downloadConfig.tlsProvided(), downloadConfig.tlsConfig(), - downloadConfig.authConfig(), downloadConfig.envNames, downloadConfig.oauth2Mount), " ") + downloadConfig.AuthProvided(), downloadConfig.TLSProvided(), downloadConfig.TLSConfig(), + downloadConfig.AuthConfig(), downloadConfig.EnvNames, downloadConfig.Oauth2Mount), " ") processCommand = downloadCommand + " && " + processCommand } return []string{"bash", "-c", processCommand} } func MakePythonFunctionCommand(downloadPath, packageFile, name, clusterName, generateLogConfigCommand, details, uid string, - hasPulsarctl, hasWget bool, downloadConfig downloadServiceConfig, authProvided, tlsProvided bool, secretMaps map[string]v1alpha1.SecretRef, + hasPulsarctl, hasWget bool, downloadConfig DownloadServiceConfig, authProvided, tlsProvided bool, secretMaps map[string]v1alpha1.SecretRef, state *v1alpha1.Stateful, tlsConfig TLSConfig, authConfig *v1alpha1.AuthConfig) []string { processCommand := setShardIDEnvironmentVariableCommand() + " && " + generateLogConfigCommand + @@ -699,15 +699,15 @@ func MakePythonFunctionCommand(downloadPath, packageFile, name, clusterName, gen if downloadPath != "" && !utils.EnableInitContainers { // prepend download command if the downPath is provided downloadCommand := strings.Join(GetDownloadCommandWithEnv(downloadPath, packageFile, hasPulsarctl, hasWget, - downloadConfig.authProvided(), - downloadConfig.tlsProvided(), downloadConfig.tlsConfig(), downloadConfig.authConfig(), - downloadConfig.envNames, downloadConfig.oauth2Mount), " ") + downloadConfig.AuthProvided(), + downloadConfig.TLSProvided(), downloadConfig.TLSConfig(), downloadConfig.AuthConfig(), + downloadConfig.EnvNames, downloadConfig.Oauth2Mount), " ") processCommand = downloadCommand + " && " + processCommand } return []string{"bash", "-c", processCommand} } -func MakeGoFunctionCommand(downloadPath, goExecFilePath string, function *v1alpha1.Function, downloadConfig downloadServiceConfig) []string { +func MakeGoFunctionCommand(downloadPath, goExecFilePath string, function *v1alpha1.Function, downloadConfig DownloadServiceConfig) []string { processCommand := setShardIDEnvironmentVariableCommand() + " && " + strings.Join(getProcessGoRuntimeArgs(goExecFilePath, function), " ") if downloadPath != "" && !utils.EnableInitContainers { @@ -719,15 +719,15 @@ func MakeGoFunctionCommand(downloadPath, goExecFilePath string, function *v1alph hasWget = true } downloadCommand := strings.Join(GetDownloadCommandWithEnv(downloadPath, goExecFilePath, - hasPulsarctl, hasWget, downloadConfig.authProvided(), - downloadConfig.tlsProvided(), downloadConfig.tlsConfig(), downloadConfig.authConfig(), - downloadConfig.envNames, downloadConfig.oauth2Mount), " ") + hasPulsarctl, hasWget, downloadConfig.AuthProvided(), + downloadConfig.TLSProvided(), downloadConfig.TLSConfig(), downloadConfig.AuthConfig(), + downloadConfig.EnvNames, downloadConfig.Oauth2Mount), " ") processCommand = downloadCommand + " && ls -al && pwd &&" + processCommand } return []string{"bash", "-c", processCommand} } -func MakeGenericFunctionCommand(downloadPath, functionFile, language, clusterName, details, uid string, downloadConfig downloadServiceConfig, authProvided, tlsProvided bool, secretMaps map[string]v1alpha1.SecretRef, +func MakeGenericFunctionCommand(downloadPath, functionFile, language, clusterName, details, uid string, downloadConfig DownloadServiceConfig, authProvided, tlsProvided bool, secretMaps map[string]v1alpha1.SecretRef, state *v1alpha1.Stateful, tlsConfig TLSConfig, authConfig *v1alpha1.AuthConfig) []string { processCommand := setShardIDEnvironmentVariableCommand() + " && " + @@ -736,9 +736,9 @@ func MakeGenericFunctionCommand(downloadPath, functionFile, language, clusterNam if downloadPath != "" && !utils.EnableInitContainers { // prepend download command if the downPath is provided downloadCommand := strings.Join(GetDownloadCommandWithEnv(downloadPath, functionFile, true, true, - downloadConfig.authProvided(), - downloadConfig.tlsProvided(), downloadConfig.tlsConfig(), downloadConfig.authConfig(), - downloadConfig.envNames, downloadConfig.oauth2Mount), " ") + downloadConfig.AuthProvided(), + downloadConfig.TLSProvided(), downloadConfig.TLSConfig(), downloadConfig.AuthConfig(), + downloadConfig.EnvNames, downloadConfig.Oauth2Mount), " ") processCommand = downloadCommand + " && " + processCommand } return []string{"sh", "-c", processCommand} @@ -831,12 +831,16 @@ func getOAuth2MountFile(authConfig *v1alpha1.OAuth2Config, mountPath string) str return fmt.Sprintf("%s/%s", mountPath, authConfig.KeySecretKey) } +func shellQuoteLiteral(value string) string { + return "'" + strings.ReplaceAll(value, "'", `'"'"'`) + "'" +} + func makeOAuth2AuthenticationParameters(authConfig *v1alpha1.OAuth2Config, mountPath string) string { if authConfig == nil { return "" } credentialsFile := getOAuth2MountFile(authConfig, mountPath) - return fmt.Sprintf(`'{"credentials_url":"file://%s","privateKey":"%s","private_key":"%s","issuerUrl":"%s","issuer_url":"%s","audience":"%s","scope":"%s"}'`, + return fmt.Sprintf(`{"credentials_url":"file://%s","privateKey":"%s","private_key":"%s","issuerUrl":"%s","issuer_url":"%s","audience":"%s","scope":"%s"}`, credentialsFile, credentialsFile, credentialsFile, authConfig.IssuerURL, authConfig.IssuerURL, authConfig.Audience, authConfig.Scope) } @@ -859,14 +863,14 @@ func getPulsarAdminCommandWithEnv(authProvided, tlsProvided bool, tlsConfig TLSC "--auth-plugin", OAuth2AuthenticationPlugin, "--auth-params", - makeOAuth2AuthenticationParameters(authConfig.OAuth2Config, oauth2MountPath), + shellQuoteLiteral(makeOAuth2AuthenticationParameters(authConfig.OAuth2Config, oauth2MountPath)), }...) } else if authConfig.GenericAuth != nil { args = append(args, []string{ "--auth-plugin", authConfig.GenericAuth.ClientAuthenticationPlugin, "--auth-params", - "'" + authConfig.GenericAuth.ClientAuthenticationParameters + "'", + shellQuoteLiteral(authConfig.GenericAuth.ClientAuthenticationParameters), }...) } } else if authProvided { @@ -960,13 +964,13 @@ func getPulsarctlCommandWithEnv(authProvided, tlsProvided bool, tlsConfig TLSCon "oauth2", "activate", "--auth-params", - "'" + authConfig.GenericAuth.ClientAuthenticationParameters + "'", + shellQuoteLiteral(authConfig.GenericAuth.ClientAuthenticationParameters), "|| true ) &&", PulsarctlExecutableFile, "--auth-plugin", authConfig.GenericAuth.ClientAuthenticationPlugin, "--auth-params", - "'" + authConfig.GenericAuth.ClientAuthenticationParameters + "'", + shellQuoteLiteral(authConfig.GenericAuth.ClientAuthenticationParameters), "--admin-service-url", "$" + envNames.webServiceURL, } @@ -1278,7 +1282,7 @@ func renderJavaInstanceLog4jXMLTemplate(runtime *v1alpha1.JavaRuntime, agent v1a return tpl.String(), nil } -func generatePythonLogConfigCommand(name string, runtime *v1alpha1.PythonRuntime, agent v1alpha1.LogTopicAgent) string { +func GeneratePythonLogConfigCommand(name string, runtime *v1alpha1.PythonRuntime, agent v1alpha1.LogTopicAgent) string { commands := "sed -i.bak 's/^ Log.setLevel/#&/' /pulsar/instances/python-instance/log.py && " if runtime == nil || (runtime.Log != nil && runtime.Log.LogConfig != nil) { return commands @@ -1610,7 +1614,7 @@ func getSharedArgs(details, clusterName, uid string, authProvided bool, tlsProvi "--function_version", "0", "--function_details", - "'" + details + "'", //in json format + shellQuoteLiteral(details), // in json format "--pulsar_serviceurl", "$brokerServiceURL", "--max_buffered_tuples", @@ -1631,13 +1635,13 @@ func getSharedArgs(details, clusterName, uid string, authProvided bool, tlsProvi "--client_auth_plugin", OAuth2AuthenticationPlugin, "--client_auth_params", - authConfig.OAuth2Config.AuthenticationParameters()}...) + shellQuoteLiteral(authConfig.OAuth2Config.AuthenticationParameters())}...) } else if authConfig.GenericAuth != nil { args = append(args, []string{ "--client_auth_plugin", authConfig.GenericAuth.ClientAuthenticationPlugin, "--client_auth_params", - "'" + authConfig.GenericAuth.ClientAuthenticationParameters + "'"}...) + shellQuoteLiteral(authConfig.GenericAuth.ClientAuthenticationParameters)}...) } } else if authProvided { args = append(args, []string{ @@ -2058,7 +2062,7 @@ func generateVolumeMountFromOAuth2ConfigWithMountPath(config *v1alpha1.OAuth2Con } } -func appendPackageServiceVolumes(volumes []corev1.Volume, packageService *v1alpha1.PulsarMessaging) []corev1.Volume { +func AppendPackageServiceVolumes(volumes []corev1.Volume, packageService *v1alpha1.PulsarMessaging) []corev1.Volume { if packageService == nil { return volumes } @@ -2077,7 +2081,7 @@ func appendPackageServiceVolumes(volumes []corev1.Volume, packageService *v1alph return volumes } -func appendPackageServiceVolumeMounts(mounts []corev1.VolumeMount, packageService *v1alpha1.PulsarMessaging) []corev1.VolumeMount { +func AppendPackageServiceVolumeMounts(mounts []corev1.VolumeMount, packageService *v1alpha1.PulsarMessaging) []corev1.VolumeMount { if packageService == nil { return mounts } diff --git a/controllers/spec/common_test.go b/controllers/spec/common_test.go index 9cfdd513f..a619d71aa 100644 --- a/controllers/spec/common_test.go +++ b/controllers/spec/common_test.go @@ -205,11 +205,11 @@ func TestGetDownloadCommand(t *testing.T) { "oauth2", "activate", "--auth-params", - "'auth-params'", + shellQuoteLiteral("auth-params"), "|| true ) &&", PulsarctlExecutableFile, "--auth-plugin", "auth-plugin", - "--auth-params", "'auth-params'", + "--auth-params", shellQuoteLiteral("auth-params"), "--admin-service-url", "$webServiceURL", "--tls-allow-insecure=true", "--tls-enable-hostname-verification=false", @@ -239,7 +239,7 @@ func TestGetDownloadCommand(t *testing.T) { PulsarAdminExecutableFile, "--admin-url", "$webServiceURL", "--auth-plugin", OAuth2AuthenticationPlugin, - "--auth-params", testOauth2.AuthenticationParameters(), + "--auth-params", shellQuoteLiteral(testOauth2.AuthenticationParameters()), "packages", "download", "function://public/default/test@v1", "--path", "function-package.jar", }, false, @@ -252,7 +252,7 @@ func TestGetDownloadCommand(t *testing.T) { PulsarAdminExecutableFile, "--admin-url", "$webServiceURL", "--auth-plugin", "auth-plugin", - "--auth-params", "'auth-params'", + "--auth-params", shellQuoteLiteral("auth-params"), "packages", "download", "sink://public/default/test@v1", "--path", "sink-package.jar", }, false, @@ -335,6 +335,67 @@ func TestGetDownloadCommand(t *testing.T) { } } +func TestShellQuoteLiteral(t *testing.T) { + testCases := []struct { + name string + input string + expected string + }{ + { + name: "empty", + input: "", + expected: "''", + }, + { + name: "plain", + input: "auth-params", + expected: "'auth-params'", + }, + { + name: "embedded single quote", + input: "a'b", + expected: `'a'"'"'b'`, + }, + { + name: "time partition pattern", + input: "yyyy-MM-dd/HH'h'-mm'm'", + expected: `'yyyy-MM-dd/HH'"'"'h'"'"'-mm'"'"'m'"'"''`, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + assert.Equal(t, tc.expected, shellQuoteLiteral(tc.input)) + }) + } +} + +func TestGetPulsarAdminCommandWithQuotedAuthParams(t *testing.T) { + authConfig := &v1alpha1.AuthConfig{ + GenericAuth: &v1alpha1.GenericAuth{ + ClientAuthenticationPlugin: "auth-plugin", + ClientAuthenticationParameters: `{"token":"a'b"}`, + }, + } + + command := getPulsarAdminCommand(false, false, nil, authConfig) + assert.Equal(t, "auth-plugin", command[4]) + assert.Equal(t, shellQuoteLiteral(`{"token":"a'b"}`), command[6]) +} + +func TestGetPulsarctlCommandWithQuotedAuthParams(t *testing.T) { + authConfig := &v1alpha1.AuthConfig{ + GenericAuth: &v1alpha1.GenericAuth{ + ClientAuthenticationPlugin: "auth-plugin", + ClientAuthenticationParameters: `{"token":"a'b"}`, + }, + } + + command := getPulsarctlCommand(false, false, nil, authConfig) + assert.Equal(t, shellQuoteLiteral(`{"token":"a'b"}`), command[5]) + assert.Equal(t, shellQuoteLiteral(`{"token":"a'b"}`), command[11]) +} + func TestGetFunctionRunnerImage(t *testing.T) { javaRuntime := v1alpha1.Runtime{Java: &v1alpha1.JavaRuntime{ Jar: "test.jar", diff --git a/controllers/spec/function.go b/controllers/spec/function.go index 29d54599f..5c6f83229 100644 --- a/controllers/spec/function.go +++ b/controllers/spec/function.go @@ -156,7 +156,7 @@ func makeFunctionVolumes(function *v1alpha1.Function, authConfig *v1alpha1.AuthC authConfig, GetRuntimeLogConfigNames(function.Spec.Java, function.Spec.Python, function.Spec.Golang), function.Spec.LogTopicAgent) - return appendPackageServiceVolumes(volumes, function.Spec.PulsarPackageService) + return AppendPackageServiceVolumes(volumes, function.Spec.PulsarPackageService) } func makeFunctionVolumeMounts(function *v1alpha1.Function, authConfig *v1alpha1.AuthConfig) []corev1.VolumeMount { @@ -177,7 +177,7 @@ func makeFunctionContainer(function *v1alpha1.Function) *corev1.Container { probe := MakeLivenessProbe(function.Spec.Pod.Liveness) allowPrivilegeEscalation := false mounts := makeFunctionVolumeMounts(function, function.Spec.Pulsar.AuthConfig) - mounts = appendPackageServiceVolumeMounts(mounts, function.Spec.PulsarPackageService) + mounts = AppendPackageServiceVolumeMounts(mounts, function.Spec.PulsarPackageService) if utils.EnableInitContainers { mounts = append(mounts, generateDownloaderVolumeMountsForRuntime(function.Spec.Java, function.Spec.Python, function.Spec.Golang, function.Spec.GenericRuntime)...) @@ -273,7 +273,7 @@ func makeFunctionCommand(function *v1alpha1.Function) []string { mountPath := extractMountPath(spec.Python.Py) return MakePythonFunctionCommand(spec.Python.PyLocation, mountPath, spec.Name, spec.ClusterName, - generatePythonLogConfigCommand(spec.Name, spec.Python, spec.LogTopicAgent), + GeneratePythonLogConfigCommand(spec.Name, spec.Python, spec.LogTopicAgent), generateFunctionDetailsInJSON(function), string(function.UID), hasPulsarctl, hasWget, downloadConfig, spec.Pulsar.AuthSecret != "", spec.Pulsar.TLSSecret != "", spec.SecretsMap, spec.StateConfig, spec.Pulsar.TLSConfig, spec.Pulsar.AuthConfig) diff --git a/controllers/spec/function_test.go b/controllers/spec/function_test.go index 14525fc13..ea1ca0b46 100644 --- a/controllers/spec/function_test.go +++ b/controllers/spec/function_test.go @@ -19,7 +19,6 @@ package spec import ( "encoding/json" - "regexp" "strings" "testing" @@ -198,11 +197,9 @@ func TestJavaFunctionCommandWithConnectorOverrides(t *testing.T) { startCommand := commands[2] assert.Assert(t, strings.Contains(startCommand, "--connectors_directory "+DefaultConnectorsDirectory), "start command should include connectors directory but got %s", startCommand) - re := regexp.MustCompile(`--function_details '([^']+)'`) - matches := re.FindStringSubmatch(startCommand) - assert.Assert(t, len(matches) == 2, "unable to locate function details in command: %s", startCommand) - - functionDetailsJSON := matches[1] + functionDetailsJSON := generateFunctionDetailsInJSON(function) + assert.Assert(t, strings.Contains(startCommand, "--function_details "+shellQuoteLiteral(functionDetailsJSON)), + "start command should include shell quoted function details but got %s", startCommand) details := &proto.FunctionDetails{} err := protojson.Unmarshal([]byte(functionDetailsJSON), details) assert.NilError(t, err) @@ -232,6 +229,116 @@ func TestJavaFunctionCommandWithConnectorOverrides(t *testing.T) { assert.Equal(t, producerConfig["enable.idempotence"], true) } +func TestSinkCommandShellQuotesFunctionDetails(t *testing.T) { + replicas := int32(1) + trueVal := true + sinkConfig := v1alpha1.NewConfig(map[string]interface{}{ + "partitionerType": "time", + "timePartitionDuration": "1m", + "timePartitionPattern": "yyyy-MM-dd/HH'h'-mm'm'", + }) + sink := &v1alpha1.Sink{ + TypeMeta: metav1.TypeMeta{ + Kind: "Sink", + APIVersion: "compute.functionmesh.io/v1alpha1", + }, + ObjectMeta: *makeSampleObjectMeta("time-pattern-sink"), + Spec: v1alpha1.SinkSpec{ + Name: "time-pattern-sink", + ClassName: "org.apache.pulsar.io.jcloud.sink.CloudStorageGenericRecordSink", + Tenant: "public", + Namespace: "default", + ClusterName: TestClusterName, + Input: v1alpha1.InputConf{ + Topics: []string{ + "persistent://public/default/input", + }, + TypeClassName: "org.apache.pulsar.client.api.schema.GenericRecord", + }, + SinkConfig: &sinkConfig, + Replicas: &replicas, + AutoAck: &trueVal, + Messaging: v1alpha1.Messaging{ + Pulsar: &v1alpha1.PulsarMessaging{ + PulsarConfig: TestClusterName, + }, + }, + Runtime: v1alpha1.Runtime{ + Java: &v1alpha1.JavaRuntime{ + Jar: "connectors/pulsar-io-cloud-storage.nar", + JarLocation: "", + }, + }, + Image: "streamnative/pulsar-io-cloud-storage:latest", + }, + } + + commands := MakeSinkCommand(sink) + assert.Assert(t, len(commands) == 3, "commands should be 3 but got %d", len(commands)) + + sinkDetailsJSON := generateSinkDetailsInJSON(sink) + assert.Assert(t, strings.Contains(commands[2], "--function_details "+shellQuoteLiteral(sinkDetailsJSON)), + "sink command should include shell quoted function details but got %s", commands[2]) + assert.Assert(t, strings.Contains(commands[2], `'"'"'`), + "sink command should escape embedded single quotes but got %s", commands[2]) +} + +func TestSinkCommandShellQuotesClientAuthParams(t *testing.T) { + replicas := int32(1) + trueVal := true + authParams := `{"token":"a'b"}` + sink := &v1alpha1.Sink{ + TypeMeta: metav1.TypeMeta{ + Kind: "Sink", + APIVersion: "compute.functionmesh.io/v1alpha1", + }, + ObjectMeta: *makeSampleObjectMeta("auth-config-sink"), + Spec: v1alpha1.SinkSpec{ + Name: "auth-config-sink", + ClassName: "org.apache.pulsar.io.elasticsearch.ElasticSearchSink", + Tenant: "public", + Namespace: "default", + ClusterName: TestClusterName, + Input: v1alpha1.InputConf{ + Topics: []string{ + "persistent://public/default/input", + }, + TypeClassName: "[B", + }, + SinkConfig: &v1alpha1.Config{ + Data: map[string]interface{}{ + "elasticSearchUrl": "http://localhost:9200", + }, + }, + Replicas: &replicas, + AutoAck: &trueVal, + Messaging: v1alpha1.Messaging{ + Pulsar: &v1alpha1.PulsarMessaging{ + PulsarConfig: TestClusterName, + AuthConfig: &v1alpha1.AuthConfig{ + GenericAuth: &v1alpha1.GenericAuth{ + ClientAuthenticationPlugin: "auth-plugin", + ClientAuthenticationParameters: authParams, + }, + }, + }, + }, + Runtime: v1alpha1.Runtime{ + Java: &v1alpha1.JavaRuntime{ + Jar: "connectors/pulsar-io-elastic-search.nar", + JarLocation: "", + }, + }, + Image: "streamnative/pulsar-io-elastic-search:latest", + }, + } + + commands := MakeSinkCommand(sink) + assert.Assert(t, len(commands) == 3, "commands should be 3 but got %d", len(commands)) + assert.Assert(t, strings.Contains(commands[2], "--client_auth_params "+shellQuoteLiteral(authParams)), + "sink command should shell quote client auth params but got %s", commands[2]) +} + func TestFunctionPulsarPackageServiceDownloadCommandAndPodWiring(t *testing.T) { previous := utils.EnableInitContainers defer func() { diff --git a/controllers/spec/sink.go b/controllers/spec/sink.go index 3630d9579..b240dd7f2 100644 --- a/controllers/spec/sink.go +++ b/controllers/spec/sink.go @@ -107,7 +107,7 @@ func makeSinkContainer(sink *v1alpha1.Sink) *corev1.Container { probe := MakeLivenessProbe(sink.Spec.Pod.Liveness) allowPrivilegeEscalation := false mounts := makeSinkVolumeMounts(sink, sink.Spec.Pulsar.AuthConfig) - mounts = appendPackageServiceVolumeMounts(mounts, sink.Spec.PulsarPackageService) + mounts = AppendPackageServiceVolumeMounts(mounts, sink.Spec.PulsarPackageService) if utils.EnableInitContainers { mounts = append(mounts, generateDownloaderVolumeMountsForRuntime(sink.Spec.Java, nil, nil, nil)...) } @@ -212,7 +212,7 @@ func makeSinkVolumes(sink *v1alpha1.Sink, authConfig *v1alpha1.AuthConfig) []cor authConfig, GetRuntimeLogConfigNames(sink.Spec.Java, sink.Spec.Python, sink.Spec.Golang), sink.Spec.LogTopicAgent) - return appendPackageServiceVolumes(volumes, sink.Spec.PulsarPackageService) + return AppendPackageServiceVolumes(volumes, sink.Spec.PulsarPackageService) } func makeSinkVolumeMounts(sink *v1alpha1.Sink, authConfig *v1alpha1.AuthConfig) []corev1.VolumeMount { diff --git a/controllers/spec/source.go b/controllers/spec/source.go index 4ce50ed98..ed245dd71 100644 --- a/controllers/spec/source.go +++ b/controllers/spec/source.go @@ -102,7 +102,7 @@ func makeSourceContainer(source *v1alpha1.Source) *corev1.Container { probe := MakeLivenessProbe(source.Spec.Pod.Liveness) allowPrivilegeEscalation := false mounts := makeSourceVolumeMounts(source, source.Spec.Pulsar.AuthConfig) - mounts = appendPackageServiceVolumeMounts(mounts, source.Spec.PulsarPackageService) + mounts = AppendPackageServiceVolumeMounts(mounts, source.Spec.PulsarPackageService) if utils.EnableInitContainers { mounts = append(mounts, generateDownloaderVolumeMountsForRuntime(source.Spec.Java, nil, nil, nil)...) } @@ -158,7 +158,7 @@ func makeSourceVolumes(source *v1alpha1.Source, authConfig *v1alpha1.AuthConfig) authConfig, GetRuntimeLogConfigNames(source.Spec.Java, source.Spec.Python, source.Spec.Golang), source.Spec.LogTopicAgent) - return appendPackageServiceVolumes(volumes, source.Spec.PulsarPackageService) + return AppendPackageServiceVolumes(volumes, source.Spec.PulsarPackageService) } func makeSourceVolumeMounts(source *v1alpha1.Source, authConfig *v1alpha1.AuthConfig) []corev1.VolumeMount { diff --git a/go.mod b/go.mod index a6ef85f94..f2ca44bf7 100644 --- a/go.mod +++ b/go.mod @@ -1,6 +1,6 @@ module github.com/streamnative/function-mesh -go 1.25.8 +go 1.25 require ( github.com/apache/pulsar-client-go v0.17.0 @@ -9,7 +9,7 @@ require ( github.com/onsi/ginkgo v1.16.5 github.com/onsi/gomega v1.35.1 github.com/prometheus/client_golang v1.20.5 - github.com/streamnative/function-mesh/api v0.27.0 + github.com/streamnative/function-mesh/api v0.27.2 github.com/streamnative/pulsarctl v0.6.0 github.com/stretchr/testify v1.11.1 google.golang.org/protobuf v1.36.10 diff --git a/install.sh b/install.sh index c090ce61f..62498b96d 100755 --- a/install.sh +++ b/install.sh @@ -56,7 +56,7 @@ EOF main() { local local_kube="kind" - local fm_version="v0.27.0" + local fm_version="v0.27.2" local kind_name="kind" local kind_version="v0.7.0" local node_num=2