diff --git a/.github/workflows/bundle-release.yml b/.github/workflows/bundle-release.yml index 9ec02a49..e91a46da 100644 --- a/.github/workflows/bundle-release.yml +++ b/.github/workflows/bundle-release.yml @@ -49,10 +49,10 @@ jobs: username: ${{ secrets.DOCKER_USER }} password: ${{ secrets.DOCKER_PASSWORD }} - - name: Set up GO 1.25.11 + - name: Set up GO 1.25.12 uses: actions/setup-go@v5 with: - go-version: 1.25.11 + go-version: 1.25.12 id: go - name: InstallKubebuilder @@ -180,10 +180,10 @@ jobs: username: ${{ secrets.DOCKER_USER }} password: ${{ secrets.DOCKER_PASSWORD }} - - name: Set up GO 1.25.11 + - name: Set up GO 1.25.12 uses: actions/setup-go@v5 with: - go-version: 1.25.11 + go-version: 1.25.12 id: go - name: InstallKubebuilder diff --git a/.github/workflows/olm-verify.yml b/.github/workflows/olm-verify.yml index d43f1b09..0060643a 100644 --- a/.github/workflows/olm-verify.yml +++ b/.github/workflows/olm-verify.yml @@ -34,10 +34,10 @@ jobs: - name: checkout uses: actions/checkout@v2 - - name: Set up GO 1.25.11 + - name: Set up GO 1.25.12 uses: actions/setup-go@v5 with: - go-version: 1.25.11 + go-version: 1.25.12 id: go - name: InstallKubebuilder diff --git a/.github/workflows/project.yml b/.github/workflows/project.yml index 7aeb5b89..77021dc0 100644 --- a/.github/workflows/project.yml +++ b/.github/workflows/project.yml @@ -18,7 +18,7 @@ jobs: strategy: fail-fast: false matrix: - go-version: [1.25.11] + go-version: [1.25.12] steps: - name: Free Disk Space (Ubuntu) uses: jlumbroso/free-disk-space@v1.3.0 diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 9c8076b1..b4b1ce0d 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -40,10 +40,10 @@ jobs: - name: Set up Docker Buildx uses: docker/setup-buildx-action@v3 - - name: Set up GO 1.25.11 + - name: Set up GO 1.25.12 uses: actions/setup-go@v5 with: - go-version: 1.25.11 + go-version: 1.25.12 id: go - name: InstallKubebuilder diff --git a/.github/workflows/test-helm-charts.yml b/.github/workflows/test-helm-charts.yml index bf05ab6b..faade266 100644 --- a/.github/workflows/test-helm-charts.yml +++ b/.github/workflows/test-helm-charts.yml @@ -83,11 +83,11 @@ jobs: run: hack/kind-cluster-build.sh --name chart-testing -c 1 -v 10 --k8sVersion v1.23.17 if: steps.list-changed.outputs.changed == 'true' - - name: Set up GO 1.25.11 + - name: Set up GO 1.25.12 if: steps.list-changed.outputs.changed == 'true' uses: actions/setup-go@v5 with: - go-version: 1.25.11 + go-version: 1.25.12 id: go - name: setup kubebuilder 3.6.0 diff --git a/.github/workflows/test-integration-skywalking-e2e.yml b/.github/workflows/test-integration-skywalking-e2e.yml index 0b3cd2e6..78173f0c 100644 --- a/.github/workflows/test-integration-skywalking-e2e.yml +++ b/.github/workflows/test-integration-skywalking-e2e.yml @@ -64,10 +64,10 @@ jobs: - name: Report disk usage run: df -h - - name: Set up GO 1.25.11 + - name: Set up GO 1.25.12 uses: actions/setup-go@v5 with: - go-version: 1.25.11 + go-version: 1.25.12 - name: Checkout uses: actions/checkout@v3 diff --git a/.github/workflows/trivy.yml b/.github/workflows/trivy.yml index f26e312d..f2c58bec 100644 --- a/.github/workflows/trivy.yml +++ b/.github/workflows/trivy.yml @@ -57,10 +57,10 @@ jobs: repository: ${{github.event.pull_request.head.repo.full_name}} ref: ${{ github.event.pull_request.head.sha }} - - name: Set up GO 1.25.11 + - name: Set up GO 1.25.12 uses: actions/setup-go@v5 with: - go-version: 1.25.11 + go-version: 1.25.12 id: go - name: InstallKubebuilder diff --git a/.github/workflows/trivy_scheduled_master.yml b/.github/workflows/trivy_scheduled_master.yml index d376b2bb..084f39be 100644 --- a/.github/workflows/trivy_scheduled_master.yml +++ b/.github/workflows/trivy_scheduled_master.yml @@ -67,10 +67,10 @@ jobs: repository: ${{github.event.pull_request.head.repo.full_name}} ref: ${{ github.event.pull_request.head.sha }} - - name: Set up GO 1.25.11 + - name: Set up GO 1.25.12 uses: actions/setup-go@v5 with: - go-version: 1.25.11 + go-version: 1.25.12 id: go - name: InstallKubebuilder diff --git a/Dockerfile b/Dockerfile index b658d1b0..7ce9c730 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,5 +1,5 @@ # Build the manager binary -FROM --platform=$BUILDPLATFORM golang:1.25.11-trixie AS builder +FROM --platform=$BUILDPLATFORM golang:1.25.12-trixie AS builder ARG TARGETOS ARG TARGETARCH diff --git a/api/compute/v1alpha1/common.go b/api/compute/v1alpha1/common.go index 38d732cf..37aad430 100644 --- a/api/compute/v1alpha1/common.go +++ b/api/compute/v1alpha1/common.go @@ -95,6 +95,9 @@ type KafkaMessaging struct { // +kubebuilder:validation:Optional OutputSchemaConfig *KafkaSchemaConfig `json:"outputSchemaConfig,omitempty"` + + // +kubebuilder:validation:Optional + SchemaRegistry *KafkaSchemaRegistryConfig `json:"schemaRegistry,omitempty"` } type KafkaTLSConfig struct { @@ -126,6 +129,25 @@ type KafkaSchemaConfig struct { Version *int32 `json:"version,omitempty"` } +type KafkaSchemaRegistryConfig struct { + URL string `json:"url,omitempty"` + AuthConfig *KafkaSchemaRegistryAuthConfig `json:"authConfig,omitempty"` +} + +type KafkaSchemaRegistryAuthConfig struct { + OAuth2Config *OAuth2Config `json:"oauth2Config,omitempty"` + BasicAuthConfig *KafkaSchemaRegistryBasicAuthConfig `json:"basicAuthConfig,omitempty"` +} + +type KafkaSchemaRegistryBasicAuthConfig struct { + // The name of the k8s secret that contains the username and password for Schema Registry authentication. + // +kubebuilder:validation:Required + SecretName string `json:"secretName,omitempty"` + + UsernameKey string `json:"usernameKey,omitempty"` + PasswordKey string `json:"passwordKey,omitempty"` +} + func (c *PulsarTLSConfig) IsEnabled() bool { return c.Enabled } diff --git a/api/compute/v1alpha1/zz_generated.deepcopy.go b/api/compute/v1alpha1/zz_generated.deepcopy.go index 15b778f5..42546152 100644 --- a/api/compute/v1alpha1/zz_generated.deepcopy.go +++ b/api/compute/v1alpha1/zz_generated.deepcopy.go @@ -782,6 +782,11 @@ func (in *KafkaMessaging) DeepCopyInto(out *KafkaMessaging) { *out = new(KafkaSchemaConfig) (*in).DeepCopyInto(*out) } + if in.SchemaRegistry != nil { + in, out := &in.SchemaRegistry, &out.SchemaRegistry + *out = new(KafkaSchemaRegistryConfig) + (*in).DeepCopyInto(*out) + } } // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new KafkaMessaging. @@ -809,6 +814,46 @@ func (in *KafkaPlainAuthConfig) DeepCopy() *KafkaPlainAuthConfig { return out } +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *KafkaSchemaRegistryAuthConfig) DeepCopyInto(out *KafkaSchemaRegistryAuthConfig) { + *out = *in + if in.OAuth2Config != nil { + in, out := &in.OAuth2Config, &out.OAuth2Config + *out = new(OAuth2Config) + **out = **in + } + if in.BasicAuthConfig != nil { + in, out := &in.BasicAuthConfig, &out.BasicAuthConfig + *out = new(KafkaSchemaRegistryBasicAuthConfig) + **out = **in + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new KafkaSchemaRegistryAuthConfig. +func (in *KafkaSchemaRegistryAuthConfig) DeepCopy() *KafkaSchemaRegistryAuthConfig { + if in == nil { + return nil + } + out := new(KafkaSchemaRegistryAuthConfig) + in.DeepCopyInto(out) + return out +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *KafkaSchemaRegistryBasicAuthConfig) DeepCopyInto(out *KafkaSchemaRegistryBasicAuthConfig) { + *out = *in +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new KafkaSchemaRegistryBasicAuthConfig. +func (in *KafkaSchemaRegistryBasicAuthConfig) DeepCopy() *KafkaSchemaRegistryBasicAuthConfig { + if in == nil { + return nil + } + out := new(KafkaSchemaRegistryBasicAuthConfig) + in.DeepCopyInto(out) + return out +} + // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *KafkaSchemaConfig) DeepCopyInto(out *KafkaSchemaConfig) { *out = *in @@ -839,6 +884,26 @@ func (in *KafkaSchemaConfig) DeepCopy() *KafkaSchemaConfig { return out } +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *KafkaSchemaRegistryConfig) DeepCopyInto(out *KafkaSchemaRegistryConfig) { + *out = *in + if in.AuthConfig != nil { + in, out := &in.AuthConfig, &out.AuthConfig + *out = new(KafkaSchemaRegistryAuthConfig) + (*in).DeepCopyInto(*out) + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new KafkaSchemaRegistryConfig. +func (in *KafkaSchemaRegistryConfig) DeepCopy() *KafkaSchemaRegistryConfig { + if in == nil { + return nil + } + out := new(KafkaSchemaRegistryConfig) + in.DeepCopyInto(out) + return out +} + // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *KafkaTLSConfig) DeepCopyInto(out *KafkaTLSConfig) { *out = *in 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 2848614e..4ca67c64 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 @@ -334,6 +334,43 @@ spec: producerConfig: type: object x-kubernetes-preserve-unknown-fields: true + schemaRegistry: + properties: + authConfig: + properties: + basicAuthConfig: + properties: + passwordKey: + type: string + secretName: + type: string + usernameKey: + type: string + required: + - secretName + type: object + oauth2Config: + properties: + audience: + type: string + issuerUrl: + type: string + keySecretKey: + type: string + keySecretName: + type: string + scope: + type: string + required: + - audience + - issuerUrl + - keySecretKey + - keySecretName + type: object + type: object + url: + type: string + type: object tlsConfig: properties: enabled: @@ -4576,6 +4613,43 @@ spec: producerConfig: type: object x-kubernetes-preserve-unknown-fields: true + schemaRegistry: + properties: + authConfig: + properties: + basicAuthConfig: + properties: + passwordKey: + type: string + secretName: + type: string + usernameKey: + type: string + required: + - secretName + type: object + oauth2Config: + properties: + audience: + type: string + issuerUrl: + type: string + keySecretKey: + type: string + keySecretName: + type: string + scope: + type: string + required: + - audience + - issuerUrl + - keySecretKey + - keySecretName + type: object + type: object + url: + type: string + type: object tlsConfig: properties: enabled: @@ -8430,6 +8504,43 @@ spec: producerConfig: type: object x-kubernetes-preserve-unknown-fields: true + schemaRegistry: + properties: + authConfig: + properties: + basicAuthConfig: + properties: + passwordKey: + type: string + secretName: + type: string + usernameKey: + type: string + required: + - secretName + type: object + oauth2Config: + properties: + audience: + type: string + issuerUrl: + type: string + keySecretKey: + type: string + keySecretName: + type: string + scope: + type: string + required: + - audience + - issuerUrl + - keySecretKey + - keySecretName + type: object + type: object + url: + type: string + type: object tlsConfig: properties: enabled: 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 23012c3b..5b157857 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 @@ -353,6 +353,43 @@ spec: producerConfig: type: object x-kubernetes-preserve-unknown-fields: true + schemaRegistry: + properties: + authConfig: + properties: + basicAuthConfig: + properties: + passwordKey: + type: string + secretName: + type: string + usernameKey: + type: string + required: + - secretName + type: object + oauth2Config: + properties: + audience: + type: string + issuerUrl: + type: string + keySecretKey: + type: string + keySecretName: + type: string + scope: + type: string + required: + - audience + - issuerUrl + - keySecretKey + - keySecretName + type: object + type: object + url: + type: string + type: object tlsConfig: properties: enabled: 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 7707f258..2d6d60af 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 @@ -348,6 +348,43 @@ spec: producerConfig: type: object x-kubernetes-preserve-unknown-fields: true + schemaRegistry: + properties: + authConfig: + properties: + basicAuthConfig: + properties: + passwordKey: + type: string + secretName: + type: string + usernameKey: + type: string + required: + - secretName + type: object + oauth2Config: + properties: + audience: + type: string + issuerUrl: + type: string + keySecretKey: + type: string + keySecretName: + type: string + scope: + type: string + required: + - audience + - issuerUrl + - keySecretKey + - keySecretName + type: object + type: object + url: + type: string + type: object tlsConfig: properties: enabled: 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 b165164c..707abced 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 @@ -282,6 +282,43 @@ spec: producerConfig: type: object x-kubernetes-preserve-unknown-fields: true + schemaRegistry: + properties: + authConfig: + properties: + basicAuthConfig: + properties: + passwordKey: + type: string + secretName: + type: string + usernameKey: + type: string + required: + - secretName + type: object + oauth2Config: + properties: + audience: + type: string + issuerUrl: + type: string + keySecretKey: + type: string + keySecretName: + type: string + scope: + type: string + required: + - audience + - issuerUrl + - keySecretKey + - keySecretName + type: object + type: object + url: + type: string + type: object tlsConfig: properties: enabled: diff --git a/config/crd/bases/compute.functionmesh.io_functionmeshes.yaml b/config/crd/bases/compute.functionmesh.io_functionmeshes.yaml index 71bafee6..9740fa8e 100644 --- a/config/crd/bases/compute.functionmesh.io_functionmeshes.yaml +++ b/config/crd/bases/compute.functionmesh.io_functionmeshes.yaml @@ -334,6 +334,43 @@ spec: producerConfig: type: object x-kubernetes-preserve-unknown-fields: true + schemaRegistry: + properties: + authConfig: + properties: + basicAuthConfig: + properties: + passwordKey: + type: string + secretName: + type: string + usernameKey: + type: string + required: + - secretName + type: object + oauth2Config: + properties: + audience: + type: string + issuerUrl: + type: string + keySecretKey: + type: string + keySecretName: + type: string + scope: + type: string + required: + - audience + - issuerUrl + - keySecretKey + - keySecretName + type: object + type: object + url: + type: string + type: object tlsConfig: properties: enabled: @@ -4576,6 +4613,43 @@ spec: producerConfig: type: object x-kubernetes-preserve-unknown-fields: true + schemaRegistry: + properties: + authConfig: + properties: + basicAuthConfig: + properties: + passwordKey: + type: string + secretName: + type: string + usernameKey: + type: string + required: + - secretName + type: object + oauth2Config: + properties: + audience: + type: string + issuerUrl: + type: string + keySecretKey: + type: string + keySecretName: + type: string + scope: + type: string + required: + - audience + - issuerUrl + - keySecretKey + - keySecretName + type: object + type: object + url: + type: string + type: object tlsConfig: properties: enabled: @@ -8430,6 +8504,43 @@ spec: producerConfig: type: object x-kubernetes-preserve-unknown-fields: true + schemaRegistry: + properties: + authConfig: + properties: + basicAuthConfig: + properties: + passwordKey: + type: string + secretName: + type: string + usernameKey: + type: string + required: + - secretName + type: object + oauth2Config: + properties: + audience: + type: string + issuerUrl: + type: string + keySecretKey: + type: string + keySecretName: + type: string + scope: + type: string + required: + - audience + - issuerUrl + - keySecretKey + - keySecretName + type: object + type: object + url: + type: string + type: object tlsConfig: properties: enabled: diff --git a/config/crd/bases/compute.functionmesh.io_functions.yaml b/config/crd/bases/compute.functionmesh.io_functions.yaml index cc571c5b..90901e52 100644 --- a/config/crd/bases/compute.functionmesh.io_functions.yaml +++ b/config/crd/bases/compute.functionmesh.io_functions.yaml @@ -331,6 +331,43 @@ spec: producerConfig: type: object x-kubernetes-preserve-unknown-fields: true + schemaRegistry: + properties: + authConfig: + properties: + basicAuthConfig: + properties: + passwordKey: + type: string + secretName: + type: string + usernameKey: + type: string + required: + - secretName + type: object + oauth2Config: + properties: + audience: + type: string + issuerUrl: + type: string + keySecretKey: + type: string + keySecretName: + type: string + scope: + type: string + required: + - audience + - issuerUrl + - keySecretKey + - keySecretName + type: object + type: object + url: + type: string + type: object tlsConfig: properties: enabled: diff --git a/config/crd/bases/compute.functionmesh.io_sinks.yaml b/config/crd/bases/compute.functionmesh.io_sinks.yaml index 9be7ab14..a795d3f4 100644 --- a/config/crd/bases/compute.functionmesh.io_sinks.yaml +++ b/config/crd/bases/compute.functionmesh.io_sinks.yaml @@ -326,6 +326,43 @@ spec: producerConfig: type: object x-kubernetes-preserve-unknown-fields: true + schemaRegistry: + properties: + authConfig: + properties: + basicAuthConfig: + properties: + passwordKey: + type: string + secretName: + type: string + usernameKey: + type: string + required: + - secretName + type: object + oauth2Config: + properties: + audience: + type: string + issuerUrl: + type: string + keySecretKey: + type: string + keySecretName: + type: string + scope: + type: string + required: + - audience + - issuerUrl + - keySecretKey + - keySecretName + type: object + type: object + url: + type: string + type: object tlsConfig: properties: enabled: diff --git a/config/crd/bases/compute.functionmesh.io_sources.yaml b/config/crd/bases/compute.functionmesh.io_sources.yaml index b6acaa78..b91474fe 100644 --- a/config/crd/bases/compute.functionmesh.io_sources.yaml +++ b/config/crd/bases/compute.functionmesh.io_sources.yaml @@ -260,6 +260,43 @@ spec: producerConfig: type: object x-kubernetes-preserve-unknown-fields: true + schemaRegistry: + properties: + authConfig: + properties: + basicAuthConfig: + properties: + passwordKey: + type: string + secretName: + type: string + usernameKey: + type: string + required: + - secretName + type: object + oauth2Config: + properties: + audience: + type: string + issuerUrl: + type: string + keySecretKey: + type: string + keySecretName: + type: string + scope: + type: string + required: + - audience + - issuerUrl + - keySecretKey + - keySecretName + type: object + type: object + url: + type: string + type: object tlsConfig: properties: enabled: diff --git a/controllers/spec/common.go b/controllers/spec/common.go index f0343950..f7fd8dec 100644 --- a/controllers/spec/common.go +++ b/controllers/spec/common.go @@ -73,16 +73,19 @@ const ( RunnerImageHasPulsarctl = "pulsar-functions-(pulsarctl|sn|generic)-(java|python|go|nodejs|base)-runner" - PulsarctlExecutableFile = "pulsarctl" - DownloaderName = "downloader" - DownloaderVolume = "downloader-volume" - DownloaderImage = DefaultRunnerPrefix + "pulsarctl:2.10.2.3" - DownloadDir = "/pulsar/download" - PackageServiceEnvPrefix = "PACKAGE_" - PackageOAuth2MountPath = "/etc/oauth2-package-service" - PackageTLSMountPath = "/etc/tls/pulsar-functions-package-service" - KafkaAuthUsernameEnv = "KAFKA_AUTH_USERNAME" - KafkaAuthPasswordEnv = "KAFKA_AUTH_PASSWORD" + PulsarctlExecutableFile = "pulsarctl" + DownloaderName = "downloader" + DownloaderVolume = "downloader-volume" + DownloaderImage = DefaultRunnerPrefix + "pulsarctl:2.10.2.3" + DownloadDir = "/pulsar/download" + PackageServiceEnvPrefix = "PACKAGE_" + PackageOAuth2MountPath = "/etc/oauth2-package-service" + PackageTLSMountPath = "/etc/tls/pulsar-functions-package-service" + KafkaSchemaRegistryOAuth2MountPath = "/etc/oauth2-kafka-schema-registry" + KafkaAuthUsernameEnv = "KAFKA_AUTH_USERNAME" + KafkaAuthPasswordEnv = "KAFKA_AUTH_PASSWORD" + KafkaSchemaRegistryAuthUsernameEnv = "KAFKA_SCHEMA_REGISTRY_AUTH_USERNAME" + KafkaSchemaRegistryAuthPasswordEnv = "KAFKA_SCHEMA_REGISTRY_AUTH_PASSWORD" CleanupContainerName = "cleanup" diff --git a/controllers/spec/function.go b/controllers/spec/function.go index bf476a48..e0739ff0 100644 --- a/controllers/spec/function.go +++ b/controllers/spec/function.go @@ -163,18 +163,20 @@ func makeFunctionVolumes(function *v1alpha1.Function, authConfig *v1alpha1.AuthC authConfig, GetRuntimeLogConfigNames(function.Spec.Java, function.Spec.Python, function.Spec.Golang), function.Spec.LogTopicAgent) + volumes = appendKafkaSchemaRegistryOAuth2Volumes(volumes, function.Spec.Kafka) return AppendPackageServiceVolumes(volumes, function.Spec.PulsarPackageService) } func makeFunctionVolumeMounts(function *v1alpha1.Function, authConfig *v1alpha1.AuthConfig) []corev1.VolumeMount { tlsConfig := functionPulsarTLSConfig(function) - return GenerateContainerVolumeMounts(function.Spec.VolumeMounts, + mounts := GenerateContainerVolumeMounts(function.Spec.VolumeMounts, function.Spec.Output.ProducerConf, function.Spec.Input.SourceSpecs, tlsConfig, authConfig, GetRuntimeLogConfigNames(function.Spec.Java, function.Spec.Python, function.Spec.Golang), function.Spec.LogTopicAgent) + return appendKafkaSchemaRegistryOAuth2VolumeMounts(mounts, function.Spec.Kafka) } func makeFunctionContainer(function *v1alpha1.Function) *corev1.Container { @@ -198,6 +200,7 @@ func makeFunctionContainer(function *v1alpha1.Function) *corev1.Container { } env := generateContainerEnv(function) env = append(env, generateKafkaAuthEnv(function.Spec.Kafka)...) + env = append(env, generateKafkaSchemaRegistryAuthEnv(function.Spec.Kafka)...) return &corev1.Container{ // TODO new container to pull user code image and upload jars into bookkeeper Name: FunctionContainerName, @@ -464,6 +467,65 @@ func generateKafkaAuthEnv(kafka *v1alpha1.KafkaMessaging) []corev1.EnvVar { } } +func appendKafkaSchemaRegistryOAuth2Volumes(volumes []corev1.Volume, kafka *v1alpha1.KafkaMessaging) []corev1.Volume { + oauth2Config := kafkaSchemaRegistryOAuth2Config(kafka) + if oauth2Config == nil { + return volumes + } + return appendVolumeIfNotExists(volumes, generateVolumeFromOAuth2Config(oauth2Config)) +} + +func appendKafkaSchemaRegistryOAuth2VolumeMounts(mounts []corev1.VolumeMount, kafka *v1alpha1.KafkaMessaging) []corev1.VolumeMount { + oauth2Config := kafkaSchemaRegistryOAuth2Config(kafka) + if oauth2Config == nil { + return mounts + } + return appendVolumeMountIfNotExists(mounts, generateVolumeMountFromOAuth2ConfigWithMountPath(oauth2Config, KafkaSchemaRegistryOAuth2MountPath)) +} + +func kafkaSchemaRegistryOAuth2Config(kafka *v1alpha1.KafkaMessaging) *v1alpha1.OAuth2Config { + if kafka == nil || kafka.SchemaRegistry == nil || kafka.SchemaRegistry.AuthConfig == nil { + return nil + } + return kafka.SchemaRegistry.AuthConfig.OAuth2Config +} + +func generateKafkaSchemaRegistryAuthEnv(kafka *v1alpha1.KafkaMessaging) []corev1.EnvVar { + if kafka == nil || kafka.SchemaRegistry == nil || kafka.SchemaRegistry.AuthConfig == nil || + kafka.SchemaRegistry.AuthConfig.BasicAuthConfig == nil { + return nil + } + basicAuth := kafka.SchemaRegistry.AuthConfig.BasicAuthConfig + usernameKey := basicAuth.UsernameKey + if usernameKey == "" { + usernameKey = "username" + } + passwordKey := basicAuth.PasswordKey + if passwordKey == "" { + passwordKey = "password" + } + return []corev1.EnvVar{ + { + Name: KafkaSchemaRegistryAuthUsernameEnv, + ValueFrom: &corev1.EnvVarSource{ + SecretKeyRef: &corev1.SecretKeySelector{ + LocalObjectReference: corev1.LocalObjectReference{Name: basicAuth.SecretName}, + Key: usernameKey, + }, + }, + }, + { + Name: KafkaSchemaRegistryAuthPasswordEnv, + ValueFrom: &corev1.EnvVarSource{ + SecretKeyRef: &corev1.SecretKeySelector{ + LocalObjectReference: corev1.LocalObjectReference{Name: basicAuth.SecretName}, + Key: passwordKey, + }, + }, + }, + } +} + func mustJSON(value interface{}) string { data, err := json.Marshal(value) if err != nil { diff --git a/controllers/spec/function_test.go b/controllers/spec/function_test.go index f133448a..09f6d8d5 100644 --- a/controllers/spec/function_test.go +++ b/controllers/spec/function_test.go @@ -552,6 +552,78 @@ func TestGenericFunctionCommandUsesKafkaMessaging(t *testing.T) { assert.Assert(t, hasKafkaOAuthMount, "container should include kafka oauth2 mount") } +func TestGenericFunctionContainerUsesKafkaSchemaRegistryBasicAuthEnv(t *testing.T) { + function := makeFunctionSample("generic-kafka-schema-registry-basic") + function.Spec.Messaging = v1alpha1.Messaging{} + function.Spec.Kafka = &v1alpha1.KafkaMessaging{ + BootstrapServers: "kafka:9092", + SchemaRegistry: &v1alpha1.KafkaSchemaRegistryConfig{ + URL: "https://schema-registry:8081", + AuthConfig: &v1alpha1.KafkaSchemaRegistryAuthConfig{ + BasicAuthConfig: &v1alpha1.KafkaSchemaRegistryBasicAuthConfig{ + SecretName: "schema-registry-auth", + UsernameKey: "user", + PasswordKey: "pass", + }, + }, + }, + } + function.Spec.Runtime = v1alpha1.Runtime{ + GenericRuntime: &v1alpha1.GenericRuntime{ + FunctionFile: "/pulsar/function.py", + Language: "python", + }, + } + + container := makeFunctionContainer(function) + envByName := map[string]corev1.EnvVar{} + for _, env := range container.Env { + envByName[env.Name] = env + } + + usernameEnv := envByName[KafkaSchemaRegistryAuthUsernameEnv] + assert.Equal(t, "schema-registry-auth", usernameEnv.ValueFrom.SecretKeyRef.Name) + assert.Equal(t, "user", usernameEnv.ValueFrom.SecretKeyRef.Key) + passwordEnv := envByName[KafkaSchemaRegistryAuthPasswordEnv] + assert.Equal(t, "schema-registry-auth", passwordEnv.ValueFrom.SecretKeyRef.Name) + assert.Equal(t, "pass", passwordEnv.ValueFrom.SecretKeyRef.Key) +} + +func TestGenericFunctionContainerUsesKafkaSchemaRegistryOAuthMount(t *testing.T) { + function := makeFunctionSample("generic-kafka-schema-registry-oauth") + function.Spec.Messaging = v1alpha1.Messaging{} + function.Spec.Kafka = &v1alpha1.KafkaMessaging{ + BootstrapServers: "kafka:9092", + SchemaRegistry: &v1alpha1.KafkaSchemaRegistryConfig{ + URL: "https://schema-registry:8081", + AuthConfig: &v1alpha1.KafkaSchemaRegistryAuthConfig{ + OAuth2Config: &v1alpha1.OAuth2Config{ + Audience: "schema-registry", + IssuerURL: "https://issuer.example.com", + KeySecretName: "schema-registry-oauth", + KeySecretKey: "auth.json", + }, + }, + }, + } + function.Spec.Runtime = v1alpha1.Runtime{ + GenericRuntime: &v1alpha1.GenericRuntime{ + FunctionFile: "/pulsar/function.py", + Language: "python", + }, + } + + container := makeFunctionContainer(function) + hasSchemaRegistryOAuthMount := false + for _, mount := range container.VolumeMounts { + if mount.MountPath == KafkaSchemaRegistryOAuth2MountPath { + hasSchemaRegistryOAuthMount = true + } + } + + assert.Assert(t, hasSchemaRegistryOAuthMount, "container should include kafka schema registry oauth2 mount") +} + func TestFunctionCommandDoesNotBuildPulsarRuntimeCommandForKafkaJavaFunction(t *testing.T) { function := makeFunctionSample("java-kafka-command") function.Spec.Messaging = v1alpha1.Messaging{ diff --git a/controllers/spec/utils.go b/controllers/spec/utils.go index 723119c9..6966a081 100644 --- a/controllers/spec/utils.go +++ b/controllers/spec/utils.go @@ -201,6 +201,7 @@ func makeKafkaConfig(function *v1alpha1.Function) map[string]interface{} { "security.protocol": securityProtocol, } addConfigData(producerConfig, kafka.ProducerConfig) + addKafkaSchemaRegistryConfig(producerConfig, kafka.SchemaRegistry) return map[string]interface{}{ "messaging_type": "kafka", @@ -232,6 +233,34 @@ func kafkaAuthEnabled(auth *v1alpha1.KafkaAuthConfig) bool { auth.PlainAuthConfig != nil) } +func addKafkaSchemaRegistryConfig(config map[string]interface{}, schemaRegistry *v1alpha1.KafkaSchemaRegistryConfig) { + if schemaRegistry == nil { + return + } + if schemaRegistry.URL != "" { + config["schema.registry.url"] = schemaRegistry.URL + } + if schemaRegistry.AuthConfig == nil { + return + } + switch { + case schemaRegistry.AuthConfig.OAuth2Config != nil: + oauth2Config := schemaRegistry.AuthConfig.OAuth2Config + config["schema.registry.oauth2.config"] = mustJSON(map[string]string{ + "private_key": getOAuth2MountFile(oauth2Config, KafkaSchemaRegistryOAuth2MountPath), + "issuer_url": oauth2Config.IssuerURL, + "audience": oauth2Config.Audience, + "scope": oauth2Config.Scope, + }) + case schemaRegistry.AuthConfig.BasicAuthConfig != nil: + // The generic runtime expands these env refs before creating the Schema Registry client. + config["schema.registry.basic.auth.credentials.source"] = "USER_INFO" + config["schema.registry.basic.auth.user.info"] = fmt.Sprintf("${%s}:${%s}", + KafkaSchemaRegistryAuthUsernameEnv, + KafkaSchemaRegistryAuthPasswordEnv) + } +} + func makeKafkaInputSpecs(function *v1alpha1.Function) map[string]interface{} { kafka := function.Spec.Kafka if kafka == nil { diff --git a/controllers/spec/utils_test.go b/controllers/spec/utils_test.go index 238273ba..5c9c2c06 100644 --- a/controllers/spec/utils_test.go +++ b/controllers/spec/utils_test.go @@ -188,6 +188,14 @@ func TestConvertFunctionDetailsWithKafkaConfig(t *testing.T) { OutputSchemaConfig: &v1alpha1.KafkaSchemaConfig{ Type: stringPtr("json"), }, + SchemaRegistry: &v1alpha1.KafkaSchemaRegistryConfig{ + URL: "https://schema-registry:8081", + AuthConfig: &v1alpha1.KafkaSchemaRegistryAuthConfig{ + BasicAuthConfig: &v1alpha1.KafkaSchemaRegistryBasicAuthConfig{ + SecretName: "schema-registry-auth", + }, + }, + }, } details := convertFunctionDetails(function) @@ -200,12 +208,62 @@ func TestConvertFunctionDetailsWithKafkaConfig(t *testing.T) { assert.Equal(t, "earliest", kafkaConfig["consumer_config"].(map[string]interface{})["auto.offset.reset"]) assert.Equal(t, "kafka:9092", kafkaConfig["producer_config"].(map[string]interface{})["bootstrap.servers"]) assert.Equal(t, float64(5), kafkaConfig["producer_config"].(map[string]interface{})["linger.ms"]) + assert.Equal(t, "https://schema-registry:8081", kafkaConfig["producer_config"].(map[string]interface{})["schema.registry.url"]) + assert.Equal(t, "USER_INFO", + kafkaConfig["producer_config"].(map[string]interface{})["schema.registry.basic.auth.credentials.source"]) + assert.Equal(t, "${KAFKA_SCHEMA_REGISTRY_AUTH_USERNAME}:${KAFKA_SCHEMA_REGISTRY_AUTH_PASSWORD}", + kafkaConfig["producer_config"].(map[string]interface{})["schema.registry.basic.auth.user.info"]) inputSpecs := kafkaConfig["input_specs"].(map[string]interface{}) assert.Equal(t, "json", inputSpecs["orders"].(map[string]interface{})["kafka_schema"].(map[string]interface{})["type"]) assert.NotContains(t, inputSpecs, "unknown-topic") assert.Equal(t, "json", kafkaConfig["output_specs"].(map[string]interface{})["enriched-orders"].(map[string]interface{})["kafka_schema"].(map[string]interface{})["type"]) } +func TestConvertFunctionDetailsWithKafkaSchemaRegistryOAuth(t *testing.T) { + function := makeFunctionSample("generic-kafka") + function.Spec.Runtime = v1alpha1.Runtime{ + GenericRuntime: &v1alpha1.GenericRuntime{ + FunctionFile: "/pulsar/function.py", + Language: "python", + }, + } + function.Spec.Input = v1alpha1.InputConf{ + Topics: []string{"orders"}, + } + function.Spec.Output = v1alpha1.OutputConf{ + Topic: "enriched-orders", + } + function.Spec.Kafka = &v1alpha1.KafkaMessaging{ + BootstrapServers: "kafka:9092", + SchemaRegistry: &v1alpha1.KafkaSchemaRegistryConfig{ + URL: "https://schema-registry:8081", + AuthConfig: &v1alpha1.KafkaSchemaRegistryAuthConfig{ + OAuth2Config: &v1alpha1.OAuth2Config{ + IssuerURL: "https://issuer.example.com", + Audience: "schema-registry", + Scope: "produce", + KeySecretName: "schema-registry-oauth", + KeySecretKey: "auth.json", + }, + }, + }, + } + + details := convertFunctionDetails(function) + userConfig := map[string]interface{}{} + assert.NoError(t, json.Unmarshal([]byte(details.UserConfig), &userConfig)) + kafkaConfig := userConfig["_kafka_config"].(map[string]interface{}) + producerConfig := kafkaConfig["producer_config"].(map[string]interface{}) + assert.Equal(t, "https://schema-registry:8081", producerConfig["schema.registry.url"]) + + oauthConfig := map[string]interface{}{} + assert.NoError(t, json.Unmarshal([]byte(producerConfig["schema.registry.oauth2.config"].(string)), &oauthConfig)) + assert.Equal(t, "/etc/oauth2-kafka-schema-registry/auth.json", oauthConfig["private_key"]) + assert.Equal(t, "https://issuer.example.com", oauthConfig["issuer_url"]) + assert.Equal(t, "schema-registry", oauthConfig["audience"]) + assert.Equal(t, "produce", oauthConfig["scope"]) +} + func TestConvertFunctionDetailsKafkaInputSchemaConfigKeepsDeclaredTopicType(t *testing.T) { function := makeFunctionSample("generic-kafka") function.Spec.Runtime = v1alpha1.Runtime{ diff --git a/go.mod b/go.mod index 592878d9..e8f5626e 100644 --- a/go.mod +++ b/go.mod @@ -95,12 +95,12 @@ require ( go.uber.org/multierr v1.11.0 // indirect go.uber.org/zap v1.27.0 // indirect golang.org/x/exp v0.0.0-20220722155223-a9213eeb770e // indirect - golang.org/x/net v0.48.0 // indirect + golang.org/x/net v0.55.0 // indirect golang.org/x/oauth2 v0.34.0 // indirect - golang.org/x/sync v0.19.0 // indirect - golang.org/x/sys v0.42.0 // indirect - golang.org/x/term v0.38.0 // indirect - golang.org/x/text v0.32.0 // indirect + golang.org/x/sync v0.20.0 // indirect + golang.org/x/sys v0.45.0 // indirect + golang.org/x/term v0.43.0 // indirect + golang.org/x/text v0.37.0 // indirect golang.org/x/time v0.10.0 // indirect gomodules.xyz/jsonpatch/v2 v2.4.0 // indirect google.golang.org/genproto/googleapis/api v0.0.0-20251202230838-ff82c1b0f217 // indirect diff --git a/go.sum b/go.sum index a411dd85..3cf61150 100644 --- a/go.sum +++ b/go.sum @@ -260,24 +260,24 @@ golang.org/x/exp v0.0.0-20220722155223-a9213eeb770e h1:+WEEuIdZHnUeJJmEUjyYC2gfU golang.org/x/exp v0.0.0-20220722155223-a9213eeb770e/go.mod h1:Kr81I6Kryrl9sr8s2FK3vxD90NdsKWRuOIl2O4CvYbA= golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= -golang.org/x/mod v0.30.0 h1:fDEXFVZ/fmCKProc/yAXXUijritrDzahmwwefnjoPFk= -golang.org/x/mod v0.30.0/go.mod h1:lAsf5O2EvJeSFMiBxXDki7sCgAxEUcZHXoXMKT4GJKc= +golang.org/x/mod v0.35.0 h1:Ww1D637e6Pg+Zb2KrWfHQUnH2dQRLBQyAtpr/haaJeM= +golang.org/x/mod v0.35.0/go.mod h1:+GwiRhIInF8wPm+4AoT6L0FA1QWAad3OMdTRx4tFYlU= golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= golang.org/x/net v0.0.0-20200520004742-59133d7f0dd7/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A= golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU= -golang.org/x/net v0.48.0 h1:zyQRTTrjc33Lhh0fBgT/H3oZq9WuvRR5gPC70xpDiQU= -golang.org/x/net v0.48.0/go.mod h1:+ndRgGjkh8FGtu1w1FGbEC31if4VrNVMuKTgcAAnQRY= +golang.org/x/net v0.55.0 h1:bcvxaJn3e1U6InsFWt1JUq1aSjnRxLzT2rtD2KfkDF8= +golang.org/x/net v0.55.0/go.mod h1:L5U2KuzuOe1lY7Z+aWVIKK6qEeJXnXV9yzGA+WCHJww= golang.org/x/oauth2 v0.34.0 h1:hqK/t4AKgbqWkdkcAeI8XLmbK+4m4G5YeQRrmiotGlw= golang.org/x/oauth2 v0.34.0/go.mod h1:lzm5WQJQwKZ3nwavOZ3IS5Aulzxi68dUSgRHujetwEA= golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= -golang.org/x/sync v0.19.0 h1:vV+1eWNmZ5geRlYjzm2adRgW2/mcpevXNg50YZtPCE4= -golang.org/x/sync v0.19.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= +golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4= +golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190222072716-a9d3bda3a223/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= @@ -289,14 +289,14 @@ golang.org/x/sys v0.0.0-20200323222414-85ca7c5b95cd/go.mod h1:h1NjWce9XRLGQEsW7w golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20210112080510-489259a85091/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20210819135213-f52c844e1c1c/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo= -golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= -golang.org/x/term v0.38.0 h1:PQ5pkm/rLO6HnxFR7N2lJHOZX6Kez5Y1gDSJla6jo7Q= -golang.org/x/term v0.38.0/go.mod h1:bSEAKrOT1W+VSu9TSCMtoGEOUcKxOKgl3LE5QEF/xVg= +golang.org/x/sys v0.45.0 h1:dO4czNzziLiiXplLQgBCEpCvXQ3dnkn0SdaZSYdQ+FY= +golang.org/x/sys v0.45.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/term v0.43.0 h1:S4RLU2sB31O/NCl+zFN9Aru9A/Cq2aqKpTZJ6B+DwT4= +golang.org/x/term v0.43.0/go.mod h1:lrhlHNdQJHO+1qVYiHfFKVuVioJIheAc3fBSMFYEIsk= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= -golang.org/x/text v0.32.0 h1:ZD01bjUt1FQ9WJ0ClOL5vxgxOI/sVCNgX1YtKwcY0mU= -golang.org/x/text v0.32.0/go.mod h1:o/rUWzghvpD5TXrTIBuJU77MTaN0ljMWE47kxGJQ7jY= +golang.org/x/text v0.37.0 h1:Cqjiwd9eSg8e0QAkyCaQTNHFIIzWtidPahFWR83rTrc= +golang.org/x/text v0.37.0/go.mod h1:a5sjxXGs9hsn/AJVwuElvCAo9v8QYLzvavO5z2PiM38= golang.org/x/time v0.10.0 h1:3usCWA8tQn0L8+hFJQNgzpWbd89begxN66o1Ojdn5L4= golang.org/x/time v0.10.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM= golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= @@ -304,8 +304,8 @@ golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtn golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE= golang.org/x/tools v0.0.0-20201224043029-2b0845dc783e/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= -golang.org/x/tools v0.39.0 h1:ik4ho21kwuQln40uelmciQPp9SipgNDdrafrYA4TmQQ= -golang.org/x/tools v0.39.0/go.mod h1:JnefbkDPyD8UU2kI5fuf8ZX4/yUeh9W877ZeBONxUqQ= +golang.org/x/tools v0.44.0 h1:UP4ajHPIcuMjT1GqzDWRlalUEoY+uzoZKnhOjbIPD2c= +golang.org/x/tools v0.44.0/go.mod h1:KA0AfVErSdxRZIsOVipbv3rQhVXTnlU6UhKxHd1seDI= golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= diff --git a/images/samples/go-function-samples/Dockerfile b/images/samples/go-function-samples/Dockerfile index 64d8a10b..98133b6e 100644 --- a/images/samples/go-function-samples/Dockerfile +++ b/images/samples/go-function-samples/Dockerfile @@ -1,5 +1,5 @@ ARG PULSAR_IMAGE_TAG -FROM --platform=$BUILDPLATFORM golang:1.25.11-trixie as builder +FROM --platform=$BUILDPLATFORM golang:1.25.12-trixie as builder ARG TARGETOS ARG TARGETARCH diff --git a/images/samples/go-function-samples/func/go.mod b/images/samples/go-function-samples/func/go.mod index 2c75b99a..0496f773 100644 --- a/images/samples/go-function-samples/func/go.mod +++ b/images/samples/go-function-samples/func/go.mod @@ -1,6 +1,6 @@ module github.com/apache/pulsar/pulsar-function-go/examples -go 1.25.11 +go 1.25.12 require github.com/apache/pulsar/pulsar-function-go v0.0.0-20250430085326-611dc3f360b5 diff --git a/operator.Dockerfile b/operator.Dockerfile index a0d316fa..052a0a73 100644 --- a/operator.Dockerfile +++ b/operator.Dockerfile @@ -1,4 +1,4 @@ -FROM --platform=$BUILDPLATFORM golang:1.25.11-trixie AS builder +FROM --platform=$BUILDPLATFORM golang:1.25.12-trixie AS builder ARG TARGETOS ARG TARGETARCH diff --git a/pkg/webhook/validate.go b/pkg/webhook/validate.go index b98256c3..8b754c24 100644 --- a/pkg/webhook/validate.go +++ b/pkg/webhook/validate.go @@ -438,6 +438,9 @@ func validateFunctionMessaging(spec *v1alpha1.FunctionSpec) *field.Error { spec.Kafka.AuthConfig.PlainAuthConfig.SecretName, "kafka.authConfig.plainAuthConfig.secretName needs to be set") } + if fieldErr := validateKafkaSchemaRegistry(spec.Kafka.SchemaRegistry); fieldErr != nil { + return fieldErr + } if spec.CleanupSubscription { return field.Invalid(field.NewPath("spec").Child("cleanupSubscription"), spec.CleanupSubscription, "cleanupSubscription only supports pulsar messaging") @@ -447,6 +450,49 @@ func validateFunctionMessaging(spec *v1alpha1.FunctionSpec) *field.Error { return validateMessaging(&spec.Messaging) } +func validateKafkaSchemaRegistry(schemaRegistry *v1alpha1.KafkaSchemaRegistryConfig) *field.Error { + if schemaRegistry == nil { + return nil + } + if schemaRegistry.URL == "" { + return field.Invalid(field.NewPath("spec").Child("kafka", "schemaRegistry", "url"), + schemaRegistry.URL, "kafka.schemaRegistry.url needs to be set") + } + if schemaRegistry.AuthConfig == nil { + return nil + } + if schemaRegistry.AuthConfig.BasicAuthConfig != nil && schemaRegistry.AuthConfig.OAuth2Config != nil { + return field.Invalid(field.NewPath("spec").Child("kafka", "schemaRegistry", "authConfig"), + schemaRegistry.AuthConfig, "only one kafka.schemaRegistry auth config can be set") + } + if schemaRegistry.AuthConfig.BasicAuthConfig != nil && + schemaRegistry.AuthConfig.BasicAuthConfig.SecretName == "" { + return field.Invalid(field.NewPath("spec").Child("kafka", "schemaRegistry", "authConfig", "basicAuthConfig", "secretName"), + schemaRegistry.AuthConfig.BasicAuthConfig.SecretName, + "kafka.schemaRegistry.authConfig.basicAuthConfig.secretName needs to be set") + } + if oauth2Config := schemaRegistry.AuthConfig.OAuth2Config; oauth2Config != nil { + oauth2Path := field.NewPath("spec").Child("kafka", "schemaRegistry", "authConfig", "oauth2Config") + if oauth2Config.IssuerURL == "" { + return field.Invalid(oauth2Path.Child("issuerUrl"), oauth2Config.IssuerURL, + "kafka.schemaRegistry.authConfig.oauth2Config.issuerUrl needs to be set") + } + if oauth2Config.Audience == "" { + return field.Invalid(oauth2Path.Child("audience"), oauth2Config.Audience, + "kafka.schemaRegistry.authConfig.oauth2Config.audience needs to be set") + } + if oauth2Config.KeySecretName == "" { + return field.Invalid(oauth2Path.Child("keySecretName"), oauth2Config.KeySecretName, + "kafka.schemaRegistry.authConfig.oauth2Config.keySecretName needs to be set") + } + if oauth2Config.KeySecretKey == "" { + return field.Invalid(oauth2Path.Child("keySecretKey"), oauth2Config.KeySecretKey, + "kafka.schemaRegistry.authConfig.oauth2Config.keySecretKey needs to be set") + } + } + return nil +} + func validateKafkaMessagingUnsupported(component string, messaging *v1alpha1.Messaging) *field.Error { if messaging == nil || messaging.Kafka == nil { return nil diff --git a/pkg/webhook/validate_test.go b/pkg/webhook/validate_test.go index 7cd79715..d81b9968 100644 --- a/pkg/webhook/validate_test.go +++ b/pkg/webhook/validate_test.go @@ -62,6 +62,130 @@ func TestValidateFunctionMessagingRejectsMissingKafkaPlainAuthSecretName(t *test } } +func TestValidateFunctionMessagingRejectsMissingKafkaSchemaRegistryURL(t *testing.T) { + err := validateFunctionMessaging(&v1alpha1.FunctionSpec{ + Messaging: v1alpha1.Messaging{ + Kafka: &v1alpha1.KafkaMessaging{ + BootstrapServers: "kafka:9092", + SchemaRegistry: &v1alpha1.KafkaSchemaRegistryConfig{}, + }, + }, + }) + if err == nil || !strings.Contains(err.Error(), "kafka.schemaRegistry.url needs to be set") { + t.Fatalf("expected missing kafka schema registry url error, got %v", err) + } +} + +func TestValidateFunctionMessagingRejectsMissingKafkaSchemaRegistryBasicAuthSecretName(t *testing.T) { + err := validateFunctionMessaging(&v1alpha1.FunctionSpec{ + Messaging: v1alpha1.Messaging{ + Kafka: &v1alpha1.KafkaMessaging{ + BootstrapServers: "kafka:9092", + SchemaRegistry: &v1alpha1.KafkaSchemaRegistryConfig{ + URL: "https://schema-registry:8081", + AuthConfig: &v1alpha1.KafkaSchemaRegistryAuthConfig{ + BasicAuthConfig: &v1alpha1.KafkaSchemaRegistryBasicAuthConfig{}, + }, + }, + }, + }, + }) + if err == nil || !strings.Contains(err.Error(), "kafka.schemaRegistry.authConfig.basicAuthConfig.secretName needs to be set") { + t.Fatalf("expected missing kafka schema registry basic auth secretName error, got %v", err) + } +} + +func TestValidateFunctionMessagingRejectsMissingKafkaSchemaRegistryOAuth2RequiredFields(t *testing.T) { + tests := []struct { + name string + mutate func(*v1alpha1.OAuth2Config) + want string + }{ + { + name: "issuer url", + mutate: func(config *v1alpha1.OAuth2Config) { + config.IssuerURL = "" + }, + want: "kafka.schemaRegistry.authConfig.oauth2Config.issuerUrl needs to be set", + }, + { + name: "audience", + mutate: func(config *v1alpha1.OAuth2Config) { + config.Audience = "" + }, + want: "kafka.schemaRegistry.authConfig.oauth2Config.audience needs to be set", + }, + { + name: "key secret name", + mutate: func(config *v1alpha1.OAuth2Config) { + config.KeySecretName = "" + }, + want: "kafka.schemaRegistry.authConfig.oauth2Config.keySecretName needs to be set", + }, + { + name: "key secret key", + mutate: func(config *v1alpha1.OAuth2Config) { + config.KeySecretKey = "" + }, + want: "kafka.schemaRegistry.authConfig.oauth2Config.keySecretKey needs to be set", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + oauth2Config := &v1alpha1.OAuth2Config{ + Audience: "schema-registry", + IssuerURL: "https://issuer.example.com", + KeySecretName: "sr-oauth", + KeySecretKey: "auth.json", + } + tt.mutate(oauth2Config) + + err := validateFunctionMessaging(&v1alpha1.FunctionSpec{ + Messaging: v1alpha1.Messaging{ + Kafka: &v1alpha1.KafkaMessaging{ + BootstrapServers: "kafka:9092", + SchemaRegistry: &v1alpha1.KafkaSchemaRegistryConfig{ + URL: "https://schema-registry:8081", + AuthConfig: &v1alpha1.KafkaSchemaRegistryAuthConfig{ + OAuth2Config: oauth2Config, + }, + }, + }, + }, + }) + if err == nil || !strings.Contains(err.Error(), tt.want) { + t.Fatalf("expected %q error, got %v", tt.want, err) + } + }) + } +} + +func TestValidateFunctionMessagingRejectsMultipleKafkaSchemaRegistryAuthConfigs(t *testing.T) { + err := validateFunctionMessaging(&v1alpha1.FunctionSpec{ + Messaging: v1alpha1.Messaging{ + Kafka: &v1alpha1.KafkaMessaging{ + BootstrapServers: "kafka:9092", + SchemaRegistry: &v1alpha1.KafkaSchemaRegistryConfig{ + URL: "https://schema-registry:8081", + AuthConfig: &v1alpha1.KafkaSchemaRegistryAuthConfig{ + BasicAuthConfig: &v1alpha1.KafkaSchemaRegistryBasicAuthConfig{SecretName: "sr-basic"}, + OAuth2Config: &v1alpha1.OAuth2Config{ + Audience: "schema-registry", + IssuerURL: "https://issuer.example.com", + KeySecretName: "sr-oauth", + KeySecretKey: "auth.json", + }, + }, + }, + }, + }, + }) + if err == nil || !strings.Contains(err.Error(), "only one kafka.schemaRegistry auth config can be set") { + t.Fatalf("expected multiple kafka schema registry auth configs error, got %v", err) + } +} + func TestValidateFunctionMessagingRejectsKafkaCleanupSubscription(t *testing.T) { err := validateFunctionMessaging(&v1alpha1.FunctionSpec{ CleanupSubscription: true, diff --git a/redhat.Dockerfile b/redhat.Dockerfile index 3a6eb1f2..7a9fa64c 100644 --- a/redhat.Dockerfile +++ b/redhat.Dockerfile @@ -1,5 +1,5 @@ # Build the manager binary -FROM --platform=$BUILDPLATFORM golang:1.25.11-trixie as builder +FROM --platform=$BUILDPLATFORM golang:1.25.12-trixie as builder ARG TARGETOS ARG TARGETARCH