[#302] add controller-configs (#322)

* add config files

* resolve import cycle

* resolve import cycle

* fix style

* fix tests

* add more tests and support helm

* fix empty test case

* fix ci

* fix test

* update dep

* more logs

* always remember to re-init

* add resourceAnnotations

Author: freeznet

.github/workflows/project.yml

@@ -6,7 +6,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        go-version: [1.13, 1.14]
+        go-version: [1.13, 1.14, 1.15, 1.16, 1.17]
     steps:
       - name: clean disk
         run: |

charts/function-mesh-operator/templates/controller-manager-configmap.yaml

@@ -0,0 +1,24 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: function-mesh-controller-manager-configs
+  labels:
+    app.kubernetes.io/name: { { template "function-mesh-operator.name" . } }
+    app.kubernetes.io/managed-by: { { .Release.Service } }
+    app.kubernetes.io/instance: { { .Release.Name } }
+    app.kubernetes.io/component: controller-manager
+    helm.sh/chart: { { .Chart.Name } }-{{ .Chart.Version | replace "+"  "_" }}
+data:
+  configs.yaml: |
+    {{- if .Values.controllerManager.runnerImages }}
+    runnerImages:
+{{ toYaml .Values.controllerManager.runnerImages | indent 6 }}
+    {{- end }}
+    {{- if .Values.controllerManager.resourceLabels }}
+    resourceLabels:
+{{ toYaml .Values.controllerManager.resourceLabels | indent 6 }}
+    {{- end }}
+    {{- if .Values.controllerManager.resourceAnnotations }}
+    resourceAnnotations:
+{{ toYaml .Values.controllerManager.resourceAnnotations | indent 6 }}
+    {{- end }}

charts/function-mesh-operator/templates/controller-manager-deployment.yaml

@@ -39,11 +39,18 @@ spec:
         {{- end }}
         command:
           - /manager
+        args:
+          - --enable-leader-election
+          - --config-file=/etc/config/config.yaml
         env:
           - name: NAMESPACE
             valueFrom:
               fieldRef:
                 fieldPath: metadata.namespace
+        volumeMounts:
+          - name: cfg
+            mountPath: /etc/config/config.yaml
+            subPath: config.yaml
       {{- with .Values.controllerManager.nodeSelector }}
       nodeSelector:
 {{ toYaml . | indent 8 }}
@@ -59,3 +66,8 @@ spec:
       {{- if .Values.controllerManager.priorityClassName }}
       priorityClassName: {{ .Values.controllerManager.priorityClassName }}
       {{- end }}
+      volumes:
+        - name: cfg
+          configMap:
+            name: function-mesh-controller-manager-configs
+            defaultMode: 420

charts/function-mesh-operator/values.yaml

@@ -46,3 +46,12 @@ controllerManager:
   selector: []
   # - k1==v1
   # - k2!=v2
+  # default runner images for different language runtime
+  # runnerImages:
+  #  java: streamnative/pulsar-functions-java-runner:2.9.0.0-rc-4
+  #  python: streamnative/pulsar-functions-python-runner:2.9.0.0-rc-4
+  #  go: streamnative/pulsar-functions-go-runner:2.9.0.0-rc-4
+  # resource labels applied to each function/connector managed by this controller
+  # resourceLabels: {}
+  # resource annotations applied to each function/connector managed by this controller
+  # resourceAnnotations: {}

controllers/spec/common.go

@@ -142,8 +142,8 @@ func MakePodTemplate(container *corev1.Container, volumes []corev1.Volume,
 	}
 	return &corev1.PodTemplateSpec{
 		ObjectMeta: metav1.ObjectMeta{
-			Labels:      mergeLabels(labels, policy.Labels),
-			Annotations: generateAnnotations(policy.Annotations),
+			Labels:      mergeLabels(labels, Configs.ResourceLabels, policy.Labels),
+			Annotations: generateAnnotations(Configs.ResourceAnnotations, policy.Annotations),
 		},
 		Spec: corev1.PodSpec{
 			InitContainers:                policy.InitContainers,
@@ -191,17 +191,13 @@ func MakeGoFunctionCommand(downloadPath, goExecFilePath string, function *v1alph
 		strings.Join(getProcessGoRuntimeArgs(goExecFilePath, function), " ")
 	if downloadPath != "" {
 		// prepend download command if the downPath is provided
-		downloadCommand := strings.Join(getDownloadCommand(downloadPath, goExecFilePath, function.Spec.Pulsar.AuthSecret != "", function.Spec.Pulsar.TLSSecret != ""), " ")
+		downloadCommand := strings.Join(getDownloadCommand(downloadPath, goExecFilePath,
+			function.Spec.Pulsar.AuthSecret != "", function.Spec.Pulsar.TLSSecret != ""), " ")
 		processCommand = downloadCommand + " && ls -al && pwd &&" + processCommand
 	}
 	return []string{"sh", "-c", processCommand}
 }
 
-func ComputeConfigHash(config map[string]interface{}) (string, error) {
-
-	return "", nil
-}
-
 func getDownloadCommand(downloadPath, componentPackage string, authProvided, tlsProvided bool) []string {
 	// The download path is the path that the package saved in the pulsar.
 	// By default, it's the path that the package saved in the pulsar, we can use package name
@@ -599,30 +595,30 @@ func generatePodVolumes(podVolumes []corev1.Volume, producerConf *v1alpha1.Produ
 	return volumes
 }
 
-func mergeLabels(label1, label2 map[string]string) map[string]string {
-	label := make(map[string]string)
+func mergeLabels(labels ...map[string]string) map[string]string {
+	merged := make(map[string]string)
 
-	for k, v := range label1 {
-		label[k] = v
-	}
-
-	for k, v := range label2 {
-		label[k] = v
+	for _, m := range labels {
+		for k, v := range m {
+			merged[k] = v
+		}
 	}
 
-	return label
+	return merged
 }
 
-func generateAnnotations(customAnnotations map[string]string) map[string]string {
+func generateAnnotations(customAnnotations ...map[string]string) map[string]string {
 	annotations := make(map[string]string)
 
 	// controlled annotations
 	annotations[AnnotationPrometheusScrape] = "true"
 	annotations[AnnotationPrometheusPort] = strconv.Itoa(int(MetricsPort.ContainerPort))
 
 	// customized annotations which may override any previous set annotations
-	for k, v := range customAnnotations {
-		annotations[k] = v
+	for _, custom := range customAnnotations {
+		for k, v := range custom {
+			annotations[k] = v
+		}
 	}
 
 	return annotations
@@ -634,11 +630,11 @@ func getFunctionRunnerImage(spec *v1alpha1.FunctionSpec) string {
 	if img != "" {
 		return img
 	} else if runtime.Java != nil && runtime.Java.Jar != "" {
-		return DefaultJavaRunnerImage
+		return Configs.RunnerImages.Java
 	} else if runtime.Python != nil && runtime.Python.Py != "" {
-		return DefaultPythonRunnerImage
+		return Configs.RunnerImages.Python
 	} else if runtime.Golang != nil && runtime.Golang.Go != "" {
-		return DefaultGoRunnerImage
+		return Configs.RunnerImages.Go
 	}
 	return DefaultRunnerImage
 }
@@ -650,7 +646,7 @@ func getSinkRunnerImage(spec *v1alpha1.SinkSpec) string {
 	}
 	if spec.Runtime.Java.Jar != "" && spec.Runtime.Java.JarLocation != "" &&
 		hasPackageNamePrefix(spec.Runtime.Java.JarLocation) {
-		return DefaultJavaRunnerImage
+		return Configs.RunnerImages.Java
 	}
 	return DefaultRunnerImage
 }
@@ -662,7 +658,7 @@ func getSourceRunnerImage(spec *v1alpha1.SourceSpec) string {
 	}
 	if spec.Runtime.Java.Jar != "" && spec.Runtime.Java.JarLocation != "" &&
 		hasPackageNamePrefix(spec.Runtime.Java.JarLocation) {
-		return DefaultJavaRunnerImage
+		return Configs.RunnerImages.Java
 	}
 	return DefaultRunnerImage
 }

controllers/spec/controller_configs.go

@@ -0,0 +1,64 @@
+// 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.
+
+package spec
+
+import (
+	"io/ioutil"
+
+	"gopkg.in/yaml.v3"
+)
+
+type RunnerImages struct {
+	Java   string `yaml:"java,omitempty"`
+	Python string `yaml:"python,omitempty"`
+	Go     string `yaml:"go,omitempty"`
+}
+
+type ControllerConfigs struct {
+	RunnerImages        RunnerImages      `yaml:"runnerImages,omitempty"`
+	ResourceLabels      map[string]string `yaml:"resourceLabels,omitempty"`
+	ResourceAnnotations map[string]string `yaml:"resourceAnnotations,omitempty"`
+}
+
+var Configs = DefaultConfigs()
+
+func DefaultConfigs() *ControllerConfigs {
+	return &ControllerConfigs{
+		RunnerImages: RunnerImages{
+			Java:   DefaultJavaRunnerImage,
+			Python: DefaultPythonRunnerImage,
+			Go:     DefaultGoRunnerImage,
+		},
+	}
+}
+
+func ParseControllerConfigs(configFilePath string) error {
+	yamlFile, err := ioutil.ReadFile(configFilePath)
+	if err != nil {
+		return err
+	}
+	if len(yamlFile) == 0 {
+		return nil
+	}
+	err = yaml.Unmarshal(yamlFile, Configs)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}

controllers/spec/controller_configs_test.go

@@ -0,0 +1,56 @@
+// 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.
+
+package spec
+
+import (
+	"testing"
+
+	"gotest.tools/assert"
+)
+
+func TestParseConfigFiles(t *testing.T) {
+	Configs = DefaultConfigs()
+	err := ParseControllerConfigs("../../testdata/controller_configs.yaml")
+	if err != nil {
+		t.Errorf("ParseControllerConfigs failed: %v", err)
+	}
+	assert.Assert(t, Configs != nil)
+	assert.Assert(t, Configs.RunnerImages.Java == "streamnative/pulsar-functions-java-runner:latest")
+	assert.Assert(t, Configs.RunnerImages.Python == "streamnative/pulsar-functions-python-runner:latest")
+	assert.Assert(t, Configs.RunnerImages.Go == "streamnative/pulsar-functions-go-runner:latest")
+	assert.Assert(t, len(Configs.ResourceLabels) == 2)
+	assert.Assert(t, len(Configs.ResourceAnnotations) == 1)
+	assert.Assert(t, Configs.ResourceLabels["functionmesh.io/managedBy"] == "function-mesh")
+	assert.Assert(t, Configs.ResourceLabels["foo"] == "bar")
+	assert.Assert(t, Configs.ResourceAnnotations["fooAnnotation"] == "barAnnotation")
+}
+
+func TestParseEmptyConfigFiles(t *testing.T) {
+	Configs = DefaultConfigs()
+	err := ParseControllerConfigs("../../testdata/empty_controller_configs.yaml")
+	if err != nil {
+		t.Errorf("ParseControllerConfigs failed: %v", err)
+	}
+	assert.Assert(t, Configs != nil)
+	t.Log("Configs", Configs)
+	assert.Assert(t, Configs.RunnerImages.Java == DefaultJavaRunnerImage)
+	assert.Assert(t, Configs.RunnerImages.Python == DefaultPythonRunnerImage)
+	assert.Assert(t, Configs.RunnerImages.Go == DefaultGoRunnerImage)
+	assert.Assert(t, len(Configs.ResourceLabels) == 0)
+	assert.Assert(t, len(Configs.ResourceAnnotations) == 0)
+}

go.mod

@@ -13,6 +13,8 @@ require (
 	github.com/streamnative/pulsarctl v0.4.3-0.20220104092115-5af28d815290
 	github.com/stretchr/testify v1.6.1
 	google.golang.org/protobuf v1.25.0
+	gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c
+	gotest.tools v2.2.0+incompatible
 	k8s.io/api v0.18.6
 	k8s.io/apimachinery v0.18.6
 	k8s.io/client-go v0.18.6

go.sum

@@ -784,6 +784,7 @@ gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gotest.tools v0.0.0-20181223230014-1083505acf35/go.mod h1:R//lfYlUuTOTfblYI3lGoAAAebUdzjvbmQsuB7Ykd90=
+gotest.tools v2.2.0+incompatible h1:VsBPFP1AI068pPrMxtb/S8Zkgf9xEmTLJjfM+P5UIEo=
 gotest.tools v2.2.0+incompatible/go.mod h1:DsYFclhRJ6vuDpmuTbkuFWG+y2sxOXAzmJt81HFBacw=
 honnef.co/go/tools v0.0.0-20180728063816-88497007e858/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=

main.go

@@ -21,6 +21,8 @@ import (
 	"flag"
 	"os"
 
+	"github.com/streamnative/function-mesh/controllers/spec"
+
 	"k8s.io/apimachinery/pkg/runtime"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
@@ -50,6 +52,7 @@ func main() {
 	var leaderElectionID string
 	var certDir string
 	var enableLeaderElection bool
+	var configFile string
 	flag.StringVar(&metricsAddr, "metrics-addr", ":8080", "The address the metric endpoint binds to.")
 	flag.StringVar(&leaderElectionID, "leader-election-id", "a3f45fce.functionmesh.io",
 		"the name of the configmap that leader election will use for holding the leader lock.")
@@ -58,10 +61,20 @@ func main() {
 			"Enabling this will ensure there is only one active controller manager.")
 	flag.StringVar(&certDir, "cert-dir", "",
 		"CertDir is the directory that contains the server key and certificate.\n\tif not set, webhook server would look up the server key and certificate in\n\t{TempDir}/k8s-webhook-server/serving-certs. The server key and certificate\n\tmust be named tls.key and tls.crt, respectively.")
+	flag.StringVar(&configFile, "config-file", "",
+		"config file path for controller manager")
 	flag.Parse()
 
 	ctrl.SetLogger(zap.New(zap.UseDevMode(true)))
 
+	if configFile != "" {
+		err := spec.ParseControllerConfigs(configFile)
+		if err != nil {
+			setupLog.Error(err, "unable to parse the controller configs")
+			os.Exit(1)
+		}
+	}
+
 	mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
 		Scheme:             scheme,
 		MetricsBindAddress: metricsAddr,