feat: add grpc runtime (#135)

Signed-off-by: Zike Yang <[email protected]>

Author: RobertIndie

Makefile

@@ -41,5 +41,18 @@ gen_rest_client:
 	rm -r restclient/go.mod restclient/go.sum restclient/.travis.yml restclient/.openapi-generator-ignore \
 		restclient/git_push.sh restclient/.openapi-generator restclient/api restclient/test
 
+proto:
+	for PROTO_FILE in $$(find . -name '*.proto'); do \
+		echo "generating codes for $$PROTO_FILE"; \
+		protoc \
+			--go_out=. \
+			--go_opt paths=source_relative \
+			--plugin protoc-gen-go="${GOPATH}/bin/protoc-gen-go" \
+			--go-grpc_out=. \
+			--go-grpc_opt paths=source_relative \
+			--plugin protoc-gen-go-grpc="${GOPATH}/bin/protoc-gen-go-grpc" \
+			$$PROTO_FILE; \
+	done
+
 license:
 	./license-checker/license-checker.sh

benchmark/bench_test.go

@@ -35,7 +35,10 @@ import (
 )
 
 func BenchmarkStressForBasicFunc(b *testing.B) {
-	s := server.New(server.LoadConfigFromEnv())
+	s, err := server.NewServer(server.LoadConfigFromEnv())
+	if err != nil {
+		b.Fatal(err)
+	}
 	svrCtx, svrCancel := context.WithCancel(context.Background())
 	go s.Run(svrCtx)
 	defer func() {
@@ -103,14 +106,18 @@ func BenchmarkStressForBasicFunc(b *testing.B) {
 func BenchmarkStressForBasicFuncWithMemoryQueue(b *testing.B) {
 	memoryQueueFactory := contube.NewMemoryQueueFactory(context.Background())
 
-	svrConf := &fs.Config{
+	svrConf := &common.Config{
 		ListenAddr: common.DefaultAddr,
-		TubeBuilder: func(ctx context.Context, config *fs.Config) (contube.TubeFactory, error) {
-			return memoryQueueFactory, nil
-		},
 	}
 
-	s := server.New(svrConf)
+	fm, err := fs.NewFunctionManager(fs.WithDefaultTubeFactory(memoryQueueFactory))
+	if err != nil {
+		b.Fatal(err)
+	}
+	s, err := server.NewServer(svrConf, server.WithFunctionManager(fm))
+	if err != nil {
+		b.Fatal(err)
+	}
 	svrCtx, svrCancel := context.WithCancel(context.Background())
 	go s.Run(svrCtx)
 	defer func() {
@@ -130,7 +137,7 @@ func BenchmarkStressForBasicFuncWithMemoryQueue(b *testing.B) {
 			Output:   outputTopic,
 			Replicas: &replicas,
 		},
-		QueueBuilder: func(ctx context.Context, c *fs.Config) (contube.TubeFactory, error) {
+		QueueBuilder: func(ctx context.Context, c *common.Config) (contube.TubeFactory, error) {
 			return memoryQueueFactory, nil
 		},
 	}

cmd/server/cmd.go

@@ -35,7 +35,10 @@ var (
 
 func exec(*cobra.Command, []string) {
 	common.RunProcess(func() (io.Closer, error) {
-		s := server.New(server.LoadConfigFromEnv())
+		s, err := server.NewServer(server.LoadConfigFromEnv())
+		if err != nil {
+			return nil, err
+		}
 		go s.Run(context.Background())
 		return s, nil
 	})

cmd/standalone/cmd.go

@@ -35,7 +35,10 @@ var (
 
 func exec(*cobra.Command, []string) {
 	common.RunProcess(func() (io.Closer, error) {
-		s := server.New(server.LoadStandaloneConfigFromEnv())
+		s, err := server.NewServer(server.LoadStandaloneConfigFromEnv())
+		if err != nil {
+			return nil, err
+		}
 		go s.Run(context.Background())
 		return s, nil
 	})

fs/config.go common/config.gofs/config.go renamed to common/config.go

@@ -14,18 +14,11 @@
  * limitations under the License.
  */
 
-package fs
-
-import (
-	"context"
-	"github.com/functionstream/functionstream/fs/contube"
-)
-
-type TubeBuilder func(ctx context.Context, config *Config) (contube.TubeFactory, error)
+package common
 
 // Config is a struct that holds the configuration for a function stream.
 type Config struct {
-	ListenAddr  string      // ListenAddr is the address that the function stream REST service will listen on.
-	PulsarURL   string      // PulsarURL is the URL of the Pulsar service. It's used for the pulsar_tube
-	TubeBuilder TubeBuilder // TubeBuilder is a function that will be used to build the tube.
+	ListenAddr string // ListenAddr is the address that the function stream REST service will listen on.
+	PulsarURL  string // PulsarURL is the URL of the Pulsar service. It's used for the pulsar_tube
+	TubeType   string
 }

common/constants.go

@@ -18,6 +18,7 @@ package common
 
 const (
 	PulsarTubeType = "pulsar"
+	MemoryTubeType = "memory"
 
 	DefaultAddr      = "localhost:7300"
 	DefaultPulsarURL = "pulsar://localhost:6650"

common/model/function.go

@@ -17,9 +17,13 @@
 package model
 
 type Function struct {
-	Name     string
-	Archive  string
-	Inputs   []string
+	Name    string
+	Archive string
+	Source  map[string]any
+	Sink    map[string]any
+	// Deprecate
+	Inputs []string
+	// Deprecate
 	Output   string
 	Config   map[string]string
 	Replicas int32

fs/api/instance.go

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2024 Function Stream Org.
+ *
+ * Licensed 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 api
+
+import (
+	"github.com/functionstream/functionstream/common/model"
+	"github.com/functionstream/functionstream/fs/contube"
+	"golang.org/x/net/context"
+)
+
+type FunctionInstance interface {
+	Context() context.Context
+	Definition() *model.Function
+	Stop()
+	Run(factory FunctionRuntimeFactory)
+	WaitForReady() <-chan error
+}
+
+type FunctionInstanceFactory interface {
+	NewFunctionInstance(f *model.Function, queueFactory contube.TubeFactory, i int32) FunctionInstance
+}

fs/api/runtime.go

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2024 Function Stream Org.
+ *
+ * Licensed 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 api
+
+import (
+	"github.com/functionstream/functionstream/fs/contube"
+)
+
+type FunctionRuntime interface {
+	WaitForReady() <-chan error
+	Call(e contube.Record) (contube.Record, error)
+	Stop()
+}
+
+type FunctionRuntimeFactory interface {
+	NewFunctionRuntime(instance FunctionInstance) (FunctionRuntime, error)
+}