fix python & go function command (#147)

* fix python & go function command

* fix ci

* more output

* fix ci

* fix lint

Author: freeznet

.ci/examples/go-examples/exclamationFunc

@@ -24,7 +24,6 @@ import (
 	"strings"
 	"time"
 
-	"github.com/gogo/protobuf/jsonpb"
 	"github.com/streamnative/function-mesh/api/v1alpha1"
 	"github.com/streamnative/function-mesh/controllers/proto"
 
@@ -56,6 +55,8 @@ const (
 
 	AnnotationPrometheusScrape = "prometheus.io/scrape"
 	AnnotationPrometheusPort   = "prometheus.io/port"
+
+	EnvGoFunctionConfigs = "GO_FUNCTION_CONF"
 )
 
 var GRPCPort = corev1.ContainerPort{
@@ -259,7 +260,7 @@ func getProcessPythonRuntimeArgs(name string, packageName string, clusterName st
 		"python",
 		"/pulsar/instances/python-instance/python_instance_main.py",
 		"--py",
-		fmt.Sprintf("/pulsar/%s", packageName),
+		packageName,
 		"--logging_directory",
 		"logs/functions",
 		"--logging_file",
@@ -317,38 +318,35 @@ func getSharedArgs(details, clusterName string, authProvided bool) []string {
 	return args
 }
 
-func generateGoFunctionDetailsInJSON(function *v1alpha1.Function) string {
-	functionDetails := convertFunctionDetails(function)
-	marshaler := &jsonpb.Marshaler{}
-	json, err := marshaler.MarshalToString(functionDetails)
+func generateGoFunctionConf(function *v1alpha1.Function) string {
+	goFunctionConfs := convertGoFunctionConfs(function)
+	json, err := json.Marshal(goFunctionConfs)
 	if err != nil {
 		// TODO
 		panic(err)
 	}
-	return json
+	ret := string(json)
+	ret = strings.ReplaceAll(ret, "\"instanceID\":0", "\"instanceID\":${"+EnvShardID+"}")
+	return ret
 }
 
 func getProcessGoRuntimeArgs(goExecFilePath string, function *v1alpha1.Function) []string {
-	str := generateGoFunctionDetailsInJSON(function)
-	tmpStr := strings.TrimSuffix(str, "}")
-
-	inputTopic := function.Spec.Input.Topics[0]
-	outputTopic := function.Spec.Output.Topic
-
-	configContent := fmt.Sprintf("%s, \"pulsarServiceURL\": \"pulsar://test-pulsar-broker.default.svc.cluster.local:6650\", "+
-		"\"sourceSpecsTopic\": \"%s\", \"sinkSpecsTopic\": \"%s\"}", tmpStr, inputTopic, outputTopic)
-
-	goPath := fmt.Sprintf("/pulsar/%s", goExecFilePath)
-	conf := fmt.Sprintf("'%s'", configContent)
-
+	str := generateGoFunctionConf(function)
+	str = strings.ReplaceAll(str, "\"", "\\\"")
 	args := []string{
+		fmt.Sprintf("%s=%s", EnvGoFunctionConfigs, str),
+		"&&",
+		fmt.Sprintf("goFunctionConfigs=${%s}", EnvGoFunctionConfigs),
+		"&&",
+		"echo goFunctionConfigs=\"'${goFunctionConfigs}'\"",
+		"&&",
 		"chmod +x",
-		goPath,
+		goExecFilePath,
 		"&&",
 		"exec",
-		goPath,
+		goExecFilePath,
 		"-instance-conf",
-		conf,
+		"${goFunctionConfigs}",
 	}
 
 	return args

controllers/spec/common.go

@@ -18,8 +18,11 @@
 package spec
 
 import (
+	"strings"
 	"testing"
 
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
 	"github.com/streamnative/function-mesh/api/v1alpha1"
 
 	"github.com/stretchr/testify/assert"
@@ -170,3 +173,76 @@ func TestGetSourceRunnerImage(t *testing.T) {
 	image = getSourceRunnerImage(&spec)
 	assert.Equal(t, image, "streamnative/pulsar-io-test:2.7.1")
 }
+
+func TestMakeGoFunctionCommand(t *testing.T) {
+	function := makeGoFunctionSample(TestFunctionName)
+	commands := MakeGoFunctionCommand("", "/pulsar/go-func", function)
+	assert.Equal(t, commands[0], "sh")
+	assert.Equal(t, commands[1], "-c")
+	assert.True(t, strings.HasPrefix(commands[2], "SHARD_ID=${POD_NAME##*-} && echo shardId=${SHARD_ID}"))
+	innerCommands := strings.Split(commands[2], "&&")
+	assert.Equal(t, innerCommands[0], "SHARD_ID=${POD_NAME##*-} ")
+	assert.Equal(t, innerCommands[1], " echo shardId=${SHARD_ID} ")
+	assert.True(t, strings.HasPrefix(innerCommands[2], " GO_FUNCTION_CONF"))
+	assert.Equal(t, innerCommands[3], " goFunctionConfigs=${GO_FUNCTION_CONF} ")
+	assert.Equal(t, innerCommands[4], " echo goFunctionConfigs=\"'${goFunctionConfigs}'\" ")
+	assert.Equal(t, innerCommands[5], " chmod +x /pulsar/go-func ")
+	assert.Equal(t, innerCommands[6], " exec /pulsar/go-func -instance-conf ${goFunctionConfigs}")
+}
+
+const TestClusterName string = "test-pulsar"
+const TestFunctionName string = "test-function"
+const TestNameSpace string = "default"
+
+func makeSampleObjectMeta(name string) *metav1.ObjectMeta {
+	return &metav1.ObjectMeta{
+		Name:      name,
+		Namespace: TestNameSpace,
+		UID:       "dead-beef", // uid not generate automatically with fake k8s
+	}
+}
+
+func makeGoFunctionSample(functionName string) *v1alpha1.Function {
+	maxPending := int32(1000)
+	replicas := int32(1)
+	maxReplicas := int32(5)
+	trueVal := true
+	return &v1alpha1.Function{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "Function",
+			APIVersion: "compute.functionmesh.io/v1alpha1",
+		},
+		ObjectMeta: *makeSampleObjectMeta(functionName),
+		Spec: v1alpha1.FunctionSpec{
+			Name:        functionName,
+			Tenant:      "public",
+			ClusterName: TestClusterName,
+			Input: v1alpha1.InputConf{
+				Topics: []string{
+					"persistent://public/default/go-function-input-topic",
+				},
+			},
+			Output: v1alpha1.OutputConf{
+				Topic: "persistent://public/default/go-function-output-topic",
+			},
+			LogTopic:                     "persistent://public/default/go-function-logs",
+			Timeout:                      0,
+			MaxMessageRetry:              0,
+			ForwardSourceMessageProperty: &trueVal,
+			Replicas:                     &replicas,
+			MaxReplicas:                  &maxReplicas,
+			AutoAck:                      &trueVal,
+			MaxPendingAsyncRequests:      &maxPending,
+			Messaging: v1alpha1.Messaging{
+				Pulsar: &v1alpha1.PulsarMessaging{
+					PulsarConfig: TestClusterName,
+				},
+			},
+			Runtime: v1alpha1.Runtime{
+				Golang: &v1alpha1.GoRuntime{
+					Go: "/pulsar/go-func",
+				},
+			},
+		},
+	}
+}

controllers/spec/common_test.go

@@ -0,0 +1,68 @@
+// 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 "time"
+
+// copy from https://github.com/apache/pulsar/blob/master/pulsar-function-go/conf/conf.go
+// directly import Conf from pulsar-function-go/conf will break envtest
+
+type GoFunctionConf struct {
+	PulsarServiceURL string        `json:"pulsarServiceURL" yaml:"pulsarServiceURL"`
+	InstanceID       int           `json:"instanceID" yaml:"instanceID"`
+	FuncID           string        `json:"funcID" yaml:"funcID"`
+	FuncVersion      string        `json:"funcVersion" yaml:"funcVersion"`
+	MaxBufTuples     int           `json:"maxBufTuples" yaml:"maxBufTuples"`
+	Port             int           `json:"port" yaml:"port"`
+	ClusterName      string        `json:"clusterName" yaml:"clusterName"`
+	KillAfterIdle    time.Duration `json:"killAfterIdleMs" yaml:"killAfterIdleMs"`
+	// function details config
+	Tenant               string `json:"tenant" yaml:"tenant"`
+	NameSpace            string `json:"nameSpace" yaml:"nameSpace"`
+	Name                 string `json:"name" yaml:"name"`
+	LogTopic             string `json:"logTopic" yaml:"logTopic"`
+	ProcessingGuarantees int32  `json:"processingGuarantees" yaml:"processingGuarantees"`
+	SecretsMap           string `json:"secretsMap" yaml:"secretsMap"`
+	Runtime              int32  `json:"runtime" yaml:"runtime"`
+	AutoACK              bool   `json:"autoAck" yaml:"autoAck"`
+	Parallelism          int32  `json:"parallelism" yaml:"parallelism"`
+	//source config
+	SubscriptionType    int32  `json:"subscriptionType" yaml:"subscriptionType"`
+	TimeoutMs           uint64 `json:"timeoutMs" yaml:"timeoutMs"`
+	SubscriptionName    string `json:"subscriptionName" yaml:"subscriptionName"`
+	CleanupSubscription bool   `json:"cleanupSubscription"  yaml:"cleanupSubscription"`
+	//source input specs
+	SourceSpecTopic            string `json:"sourceSpecsTopic" yaml:"sourceSpecsTopic"`
+	SourceSchemaType           string `json:"sourceSchemaType" yaml:"sourceSchemaType"`
+	IsRegexPatternSubscription bool   `json:"isRegexPatternSubscription" yaml:"isRegexPatternSubscription"`
+	ReceiverQueueSize          int32  `json:"receiverQueueSize" yaml:"receiverQueueSize"`
+	//sink spec config
+	SinkSpecTopic  string `json:"sinkSpecsTopic" yaml:"sinkSpecsTopic"`
+	SinkSchemaType string `json:"sinkSchemaType" yaml:"sinkSchemaType"`
+	//resources config
+	CPU  float64 `json:"cpu" yaml:"cpu"`
+	RAM  int64   `json:"ram" yaml:"ram"`
+	Disk int64   `json:"disk" yaml:"disk"`
+	//retryDetails config
+	MaxMessageRetries           int32  `json:"maxMessageRetries" yaml:"maxMessageRetries"`
+	DeadLetterTopic             string `json:"deadLetterTopic" yaml:"deadLetterTopic"`
+	ExpectedHealthCheckInterval int32  `json:"expectedHealthCheckInterval" yaml:"expectedHealthCheckInterval"`
+	UserConfig                  string `json:"userConfig" yaml:"userConfig"`
+	//metrics config
+	MetricsPort int `json:"metricsPort" yaml:"metricsPort"`
+}

controllers/spec/functions_go_conf.go

@@ -19,8 +19,10 @@ package spec
 
 import (
 	"encoding/json"
+	"fmt"
 	"regexp"
 	"strings"
+	"time"
 
 	"k8s.io/apimachinery/pkg/util/validation"
 
@@ -55,6 +57,42 @@ func convertFunctionDetails(function *v1alpha1.Function) *proto.FunctionDetails
 	}
 }
 
+func convertGoFunctionConfs(function *v1alpha1.Function) *GoFunctionConf {
+	return &GoFunctionConf{
+		FuncID:                      fmt.Sprintf("${%s}-%d", EnvShardID, time.Now().Unix()),
+		PulsarServiceURL:            "${brokerServiceURL}",
+		FuncVersion:                 "0",
+		MaxBufTuples:                100, //TODO
+		Port:                        int(GRPCPort.ContainerPort),
+		ClusterName:                 function.Spec.ClusterName,
+		Tenant:                      function.Spec.Tenant,
+		NameSpace:                   function.Namespace,
+		Name:                        function.Spec.Name,
+		LogTopic:                    function.Spec.LogTopic,
+		ProcessingGuarantees:        int32(convertProcessingGuarantee(function.Spec.ProcessingGuarantee)),
+		SecretsMap:                  marshalSecretsMap(function.Spec.SecretsMap),
+		Runtime:                     int32(proto.FunctionDetails_GO),
+		AutoACK:                     *function.Spec.AutoAck,
+		Parallelism:                 *function.Spec.Replicas,
+		TimeoutMs:                   uint64(function.Spec.Timeout),
+		SubscriptionName:            function.Spec.SubscriptionName,
+		CleanupSubscription:         function.Spec.CleanupSubscription,
+		SourceSpecTopic:             function.Spec.Input.Topics[0],
+		SourceSchemaType:            "", // TODO: map schema type
+		IsRegexPatternSubscription:  function.Spec.Input.TopicPattern != "",
+		SinkSpecTopic:               function.Spec.Output.Topic,
+		SinkSchemaType:              "", // TODO: map schema type
+		CPU:                         float64(function.Spec.Resources.Requests.Cpu().Value()),
+		RAM:                         function.Spec.Resources.Requests.Memory().Value(),
+		Disk:                        function.Spec.Resources.Requests.Storage().Value(),
+		MaxMessageRetries:           function.Spec.MaxMessageRetry,
+		DeadLetterTopic:             function.Spec.DeadLetterTopic,
+		UserConfig:                  getUserConfig(function.Spec.FuncConfig),
+		MetricsPort:                 int(MetricsPort.ContainerPort),
+		ExpectedHealthCheckInterval: -1, // TurnOff BuiltIn HealthCheck to avoid instance exit
+	}
+}
+
 func generateInputSpec(sourceConf v1alpha1.InputConf) map[string]*proto.ConsumerSpec {
 	inputSpecs := make(map[string]*proto.ConsumerSpec)
 

controllers/spec/utils.go

@@ -63,7 +63,10 @@ var _ = BeforeSuite(func(done Done) {
 		}
 	} else {
 		testEnv = &envtest.Environment{
-			CRDDirectoryPaths: []string{filepath.Join("..", "config", "crd", "bases")},
+			CRDInstallOptions: envtest.CRDInstallOptions{
+				Paths: []string{filepath.Join("..", "config", "crd", "bases")},
+			},
+			AttachControlPlaneOutput: true,
 		}
 	}
 	var err error

controllers/suite_test.go

@@ -3,25 +3,20 @@ module github.com/streamnative/function-mesh
 go 1.13
 
 require (
+	github.com/apache/pulsar-client-go/oauth2 v0.0.0-20201120111947-b8bd55bc02bd // indirect
 	github.com/ghodss/yaml v1.0.0
-	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/go-logr/logr v0.1.0
 	github.com/gogo/protobuf v1.3.1
-	github.com/golang/protobuf v1.4.2
-	github.com/google/gofuzz v1.1.0 // indirect
+	github.com/golang/protobuf v1.4.3
+	github.com/kr/pretty v0.2.0 // indirect
 	github.com/onsi/ginkgo v1.14.2
 	github.com/onsi/gomega v1.10.4
-	github.com/pkg/errors v0.9.1 // indirect
-	github.com/prometheus/client_golang v1.1.0 // indirect
+	github.com/prometheus/client_golang v1.7.1 // indirect
 	github.com/streamnative/pulsarctl v0.5.0
 	github.com/stretchr/testify v1.6.1
-	golang.org/x/oauth2 v0.0.0-20200107190931-bf48bf16ab8d // indirect
-	google.golang.org/appengine v1.6.6 // indirect
 	google.golang.org/protobuf v1.25.0
-	gopkg.in/yaml.v2 v2.3.0
 	k8s.io/api v0.18.6
 	k8s.io/apimachinery v0.18.6
 	k8s.io/client-go v0.18.6
 	sigs.k8s.io/controller-runtime v0.6.2
-	sigs.k8s.io/controller-tools v0.2.4 // indirect
 )