feat: improve log (#181)

Author: RobertIndie

benchmark/bench_test.go

@@ -18,7 +18,7 @@ package benchmark
 
 import (
 	"context"
-	"github.com/functionstream/function-stream/fs"
+	"github.com/functionstream/function-stream/fs/api"
 	"github.com/functionstream/function-stream/fs/runtime/wazero"
 	"math/rand"
 	"os"
@@ -114,8 +114,14 @@ func BenchmarkStressForBasicFunc(b *testing.B) {
 func BenchmarkStressForBasicFuncWithMemoryQueue(b *testing.B) {
 	memoryQueueFactory := contube.NewMemoryQueueFactory(context.Background())
 
-	s, err := server.NewServer(server.WithFunctionManager(fs.WithTubeFactory(common.MemoryTubeType, memoryQueueFactory),
-		fs.WithRuntimeFactory(common.WASMRuntime, wazero.NewWazeroFunctionRuntimeFactory())))
+	s, err := server.NewServer(
+		server.WithRuntimeFactoryBuilder(common.WASMRuntime, func(configMap common.ConfigMap) (api.FunctionRuntimeFactory, error) {
+			return wazero.NewWazeroFunctionRuntimeFactory(), nil
+		}),
+		server.WithTubeFactoryBuilder(common.MemoryTubeType, func(configMap common.ConfigMap) (contube.TubeFactory, error) {
+			return memoryQueueFactory, nil
+		}),
+	)
 	if err != nil {
 		b.Fatal(err)
 	}

common/errors.go

@@ -16,11 +16,11 @@
 
 package common
 
-import "errors"
+import "fmt"
 
 var (
-	ErrorFunctionNotFound       = errors.New("function not found")
-	ErrorFunctionExists         = errors.New("function already exists")
-	ErrorTubeFactoryNotFound    = errors.New("tube factory not found")
-	ErrorRuntimeFactoryNotFound = errors.New("runtime factory not found")
+	ErrorFunctionNotFound       = fmt.Errorf("function not found")
+	ErrorFunctionExists         = fmt.Errorf("function already exists")
+	ErrorRuntimeFactoryNotFound = fmt.Errorf("runtime factory not found")
+	ErrorTubeFactoryNotFound    = fmt.Errorf("tube factory not found")
 )

common/log.go

@@ -0,0 +1,59 @@
+/*
+ * 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 common
+
+import (
+	"github.com/go-logr/logr"
+	"github.com/go-logr/zapr"
+	"go.uber.org/zap"
+)
+
+const (
+	DebugLevel int = 4
+)
+
+type Logger struct {
+	*logr.Logger
+}
+
+func NewDefaultLogger() *Logger {
+	zapLogger, err := zap.NewDevelopment()
+	if err != nil {
+		panic("failed to create zap logger:" + err.Error())
+	}
+	zaprLog := zapr.NewLogger(zapLogger)
+	return NewLogger(&zaprLog)
+}
+
+func NewLogger(logger *logr.Logger) *Logger {
+	return &Logger{logger}
+}
+
+func (l *Logger) DebugEnabled() bool {
+	return l.GetV() >= DebugLevel
+}
+
+func (l *Logger) Debug(msg string, keysAndValues ...interface{}) {
+	if l.DebugEnabled() {
+		l.V(DebugLevel).Info(msg, keysAndValues...)
+	}
+}
+
+func (l *Logger) SubLogger(keysAndValues ...any) *Logger {
+	internalLogger := l.WithValues(keysAndValues...)
+	return &Logger{&internalLogger}
+}

common/model/function.go

@@ -24,7 +24,7 @@ import (
 )
 
 type TubeConfig struct {
-	Type   string            `json:"type"` // Default to `default`
+	Type   string            `json:"type"`
 	Config contube.ConfigMap `json:"config,omitempty"`
 }
 

fs/api/instance.go

@@ -17,8 +17,7 @@
 package api
 
 import (
-	"log/slog"
-
+	"github.com/functionstream/function-stream/common"
 	"github.com/functionstream/function-stream/common/model"
 	"github.com/functionstream/function-stream/fs/contube"
 	"golang.org/x/net/context"
@@ -30,11 +29,10 @@ type FunctionInstance interface {
 	Definition() *model.Function
 	Index() int32
 	Stop()
-	Run(factory FunctionRuntimeFactory, sources []<-chan contube.Record, sink chan<- contube.Record)
-	WaitForReady() <-chan error
-	Logger() *slog.Logger
+	Run(runtime FunctionRuntime, sources []<-chan contube.Record, sink chan<- contube.Record)
+	Logger() *common.Logger
 }
 
 type FunctionInstanceFactory interface {
-	NewFunctionInstance(f *model.Function, funcCtx FunctionContext, i int32, logger *slog.Logger) FunctionInstance
+	NewFunctionInstance(f *model.Function, funcCtx FunctionContext, i int32, logger *common.Logger) FunctionInstance
 }

fs/api/runtime.go

@@ -21,7 +21,6 @@ import (
 )
 
 type FunctionRuntime interface {
-	WaitForReady() <-chan error
 	Call(e contube.Record) (contube.Record, error)
 	Stop()
 }

fs/contube/http.go

@@ -17,8 +17,8 @@
 package contube
 
 import (
+	"github.com/functionstream/function-stream/common"
 	"io"
-	"log/slog"
 	"net/http"
 	"sync"
 	"sync/atomic"
@@ -30,9 +30,10 @@ import (
 type state int
 
 const (
-	EndpointKey = "endpoint"
+	EndpointKey     = "endpoint"
+	IncludeMetadata = "includeMetadata"
 
-	stateReady  state = iota
+	stateReady  state = iota // TODO: Why do we need this? Maybe we can remove it.
 	stateClosed state = iota
 )
 
@@ -43,6 +44,11 @@ var (
 )
 
 type EndpointHandler func(ctx context.Context, endpoint string, payload []byte) error
+type HttpHandler func(w http.ResponseWriter, r *http.Request, payload []byte) Record
+
+func DefaultHttpHandler(_ http.ResponseWriter, _ *http.Request, payload []byte) Record {
+	return NewRecordImpl(payload, func() {})
+}
 
 type endpointHandler struct {
 	ctx context.Context
@@ -55,32 +61,42 @@ type HttpTubeFactory struct {
 	ctx       context.Context
 	mu        sync.RWMutex
 	endpoints map[string]*endpointHandler
+	handler   HttpHandler
 }
 
 func NewHttpTubeFactory(ctx context.Context) *HttpTubeFactory {
+	return NewHttpTubeFactoryWithIntercept(ctx, DefaultHttpHandler)
+}
+
+func NewHttpTubeFactoryWithIntercept(ctx context.Context, handler HttpHandler) *HttpTubeFactory {
 	return &HttpTubeFactory{
 		ctx:       ctx,
 		endpoints: make(map[string]*endpointHandler),
+		handler:   handler,
 	}
 }
 
 type httpSourceTubeConfig struct {
 	Endpoint string `json:"endpoint" validate:"required"`
 }
 
-func (f *HttpTubeFactory) Handle(ctx context.Context, endpoint string, payload []byte) error {
+func (f *HttpTubeFactory) getEndpoint(endpoint string) (*endpointHandler, bool) {
 	f.mu.RLock()
+	defer f.mu.RUnlock()
 	e, ok := f.endpoints[endpoint]
+	return e, ok
+}
+
+func (f *HttpTubeFactory) Handle(ctx context.Context, endpoint string, record Record) error {
+	e, ok := f.getEndpoint(endpoint)
 	if !ok {
-		f.mu.RUnlock()
 		return ErrEndpointNotFound
 	}
-	f.mu.RUnlock()
 	if e.s.Load() == stateClosed {
 		return ErrEndpointClosed
 	}
 	select {
-	case e.c <- NewRecordImpl(payload, func() {}):
+	case e.c <- record:
 		return nil
 	case <-ctx.Done():
 		return ctx.Err()
@@ -125,28 +141,33 @@ func (f *HttpTubeFactory) NewSinkTube(_ context.Context, _ ConfigMap) (chan<- Re
 }
 
 func (f *HttpTubeFactory) GetHandleFunc(getEndpoint func(r *http.Request) (string, error),
-	logger *slog.Logger) func(http.ResponseWriter, *http.Request) {
+	logger *common.Logger) func(http.ResponseWriter, *http.Request) {
 	return func(w http.ResponseWriter, r *http.Request) {
 		endpoint, err := getEndpoint(r)
 		if err != nil {
-			logger.Error("Failed to get endpoint", "error", err)
+			logger.Error(err, "Failed to get endpoint")
 			http.Error(w, errors.Wrap(err, "Failed to get endpoint").Error(), http.StatusBadRequest)
 			return
 		}
-		log := logger.With(slog.String("endpoint", endpoint), slog.String("component", "http-tube"))
-		log.Info("Handle records from http request")
+		log := logger.SubLogger("endpoint", endpoint, "component", "http-tube")
+		if log.DebugEnabled() {
+			log.Debug("Handle records from http request")
+		}
 		content, err := io.ReadAll(r.Body)
 		if err != nil {
-			log.Error("Failed to read body", "error", err)
+			log.Error(err, "Failed to read body")
 			http.Error(w, errors.Wrap(err, "Failed to read body").Error(), http.StatusBadRequest)
 			return
 		}
-		err = f.Handle(r.Context(), endpoint, content)
+		record := f.handler(w, r, content)
+		err = f.Handle(r.Context(), endpoint, record)
 		if err != nil {
-			log.Error("Failed to handle record", "error", err)
+			log.Error(err, "Failed to handle record")
 			http.Error(w, err.Error(), http.StatusInternalServerError)
 			return
 		}
-		log.Info("Handled records from http request")
+		if log.DebugEnabled() {
+			log.Debug("Handled records from http request")
+		}
 	}
 }

fs/contube/http_test.go

@@ -29,7 +29,7 @@ func TestHttpTubeHandleRecord(t *testing.T) {
 	f := NewHttpTubeFactory(ctx)
 
 	endpoint := "test"
-	err := f.Handle(ctx, "test", []byte("test"))
+	err := f.Handle(ctx, "test", NewRecordImpl([]byte("test"), nil))
 	assert.ErrorIs(t, err, ErrEndpointNotFound)
 
 	config := make(ConfigMap)
@@ -39,7 +39,7 @@ func TestHttpTubeHandleRecord(t *testing.T) {
 	_, err = f.NewSourceTube(ctx, config)
 	assert.ErrorIs(t, err, ErrorEndpointAlreadyExists)
 
-	err = f.Handle(ctx, endpoint, []byte("test"))
+	err = f.Handle(ctx, endpoint, NewRecordImpl([]byte("test"), nil))
 	assert.Nil(t, err)
 
 	record := <-source
@@ -48,7 +48,7 @@ func TestHttpTubeHandleRecord(t *testing.T) {
 	cancel()
 
 	assert.Nil(t, <-source)
-	err = f.Handle(ctx, endpoint, []byte("test"))
+	err = f.Handle(ctx, endpoint, NewRecordImpl([]byte("test"), nil))
 	assert.Error(t, err, ErrEndpointNotFound)
 }