refactor!: gateway (#314)

* refactor!: gateway

* update gateway owner to JK

* fix due to review

Author: wenfengwang

.golangci.yaml

@@ -425,6 +425,7 @@ issues:
         - cyclop
         - gochecknoinits
         - gochecknoglobals
+        - forbidigo
     - text: "unexported-return: exported func (NewController|NewGateway) returns unexported type"
       linters:
         - revive

CODEOWNERS

@@ -13,7 +13,7 @@
 
 /internal/controller/ @xdlbdy
 /internal/controller/eventbus/ @ifplusor
-/internal/gateway/ @ifplusor
+/internal/gateway/ @hwjiangkai
 /internal/kv/ @xdlbdy
 /internal/raft/ @ifplusor
 /internal/store/ @ifplusor

client/client.go

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+//go:generate mockgen -source=client.go  -destination=mock_client.go -package=client
 package client
 
 import (

client/pkg/api/client.go

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+//go:generate mockgen -source=client.go  -destination=mock_client.go -package=api
 package api
 
 import (

client/pkg/api/option.go

@@ -25,10 +25,16 @@ type WriteOptions struct {
 	Oneway bool
 }
 
-func (opts *WriteOptions) Copy() *WriteOptions {
+func (wo *WriteOptions) Apply(opts ...WriteOption) {
+	for i := range opts {
+		opts[i](wo)
+	}
+}
+
+func (wo *WriteOptions) Copy() *WriteOptions {
 	return &WriteOptions{
-		Oneway: opts.Oneway,
-		Policy: opts.Policy,
+		Oneway: wo.Oneway,
+		Policy: wo.Policy,
 	}
 }
 
@@ -40,11 +46,17 @@ type ReadOptions struct {
 	Policy         ReadPolicy
 }
 
-func (opts *ReadOptions) Copy() *ReadOptions {
+func (ro *ReadOptions) Apply(opts ...ReadOption) {
+	for i := range opts {
+		opts[i](ro)
+	}
+}
+
+func (ro *ReadOptions) Copy() *ReadOptions {
 	return &ReadOptions{
-		BatchSize:      opts.BatchSize,
-		PollingTimeout: opts.PollingTimeout,
-		Policy:         opts.Policy,
+		BatchSize:      ro.BatchSize,
+		PollingTimeout: ro.PollingTimeout,
+		Policy:         ro.Policy,
 	}
 }
 

cmd/gateway/main.go

@@ -22,6 +22,7 @@ import (
 	"github.com/linkall-labs/vanus/internal/gateway"
 	"github.com/linkall-labs/vanus/observability/log"
 	"github.com/linkall-labs/vanus/observability/tracing"
+	"github.com/linkall-labs/vanus/pkg/util/signal"
 )
 
 var (
@@ -41,22 +42,20 @@ func main() {
 
 	tracing.Init("Vanus-Gateway")
 
-	go gateway.NewHTTPServer(*cfg).MustStartHTTP()
-
+	ctx := signal.SetupSignalContext()
 	ga := gateway.NewGateway(*cfg)
-	err = ga.StartCtrlProxy(context.Background())
-	if err != nil {
-		log.Error(context.Background(), "start controller proxy failed", map[string]interface{}{
-			log.KeyError: err,
-		})
-		os.Exit(-1)
-	}
 
-	err = ga.StartReceive(context.Background())
-	if err != nil {
-		log.Error(context.Background(), "start CloudEvents gateway failed", map[string]interface{}{
+	if err = ga.Start(ctx); err != nil {
+		log.Error(context.Background(), "start gateway failed", map[string]interface{}{
 			log.KeyError: err,
 		})
 		os.Exit(-1)
 	}
+	log.Info(ctx, "Gateway has started", nil)
+	select {
+	case <-ctx.Done():
+		log.Info(ctx, "received system signal, preparing exit", nil)
+	}
+	ga.Stop()
+	log.Info(ctx, "the gateway has been shutdown gracefully", nil)
 }

deploy/yaml/gateway.yaml

@@ -11,15 +11,11 @@ spec:
     - port: 8080
       targetPort: 8080
       nodePort: 30001
-      name: put
+      name: proxy
     - port: 8081
       targetPort: 8081
       nodePort: 30002
-      name: get
-    - name: ctrl-proxy
-      nodePort: 30003
-      port: 8082
-      targetPort: 8082
+      name: cloudevents
 ---
 apiVersion: v1
 kind: ConfigMap
@@ -58,12 +54,10 @@ spec:
           image: public.ecr.aws/vanus/gateway:v0.4.0
           imagePullPolicy: IfNotPresent
           ports:
-            - name: httpput
+            - name: proxy
               containerPort: 8080
-            - name: httpget
+            - name: cloudevents
               containerPort: 8081
-            - name: ctrl-proxy
-              containerPort: 8082
           volumeMounts:
             - name: config-gateway
               mountPath: /vanus/config

internal/controller/snowflake/snowflake.go

@@ -112,7 +112,7 @@ func (sf *snowflake) RegisterNode(ctx context.Context, in *wrapperspb.UInt32Valu
 	//
 	// if exist {
 	//	return nil, errors.New("node has been register")
-	//}
+	// }
 
 	n := &node{
 		ID:      id,

internal/gateway/config.go

@@ -14,12 +14,31 @@
 
 package gateway
 
-import "github.com/linkall-labs/vanus/internal/primitive"
+import (
+	"github.com/linkall-labs/vanus/internal/gateway/proxy"
+	"github.com/linkall-labs/vanus/internal/primitive"
+	"google.golang.org/grpc/credentials/insecure"
+)
 
 type Config struct {
-	Port           int      `yaml:"port"`
-	ControllerAddr []string `yaml:"controllers"`
-	TracingURL     string   `yaml:"tracing_url"`
+	Port                 int      `yaml:"port"`
+	ControllerAddr       []string `yaml:"controllers"`
+	TracingURL           string   `yaml:"tracing_url"`
+	GRPCReflectionEnable bool     `yaml:"grpc_reflection_enable"`
+}
+
+func (c Config) GetProxyConfig() proxy.Config {
+	return proxy.Config{
+		Endpoints:              c.ControllerAddr,
+		ProxyPort:              c.Port,
+		CloudEventReceiverPort: c.GetCloudEventReceiverPort(),
+		GRPCReflectionEnable:   c.GRPCReflectionEnable,
+		Credentials:            insecure.NewCredentials(),
+	}
+}
+
+func (c Config) GetCloudEventReceiverPort() int {
+	return c.Port + 1
 }
 
 func InitConfig(filename string) (*Config, error) {

internal/gateway/gateway.go

@@ -30,30 +30,15 @@ import (
 	"github.com/google/uuid"
 	eb "github.com/linkall-labs/vanus/client"
 	"github.com/linkall-labs/vanus/client/pkg/api"
+	"github.com/linkall-labs/vanus/internal/gateway/proxy"
 	"github.com/linkall-labs/vanus/internal/primitive"
 	"github.com/linkall-labs/vanus/observability/log"
 	"github.com/linkall-labs/vanus/observability/tracing"
 	"go.opentelemetry.io/otel/trace"
 )
 
 const (
-	httpRequestPrefix  = "/gateway"
-	ctrlProxyPortShift = 2
-)
-
-var (
-	allowCtrlProxyList = map[string]string{
-		"/linkall.vanus.controller.PingServer/Ping":                      "ALLOW",
-		"/linkall.vanus.controller.EventBusController/ListEventBus":      "ALLOW",
-		"/linkall.vanus.controller.EventBusController/CreateEventBus":    "ALLOW",
-		"/linkall.vanus.controller.EventBusController/DeleteEventBus":    "ALLOW",
-		"/linkall.vanus.controller.EventBusController/GetEventBus":       "ALLOW",
-		"/linkall.vanus.controller.EventLogController/ListSegment":       "ALLOW",
-		"/linkall.vanus.controller.TriggerController/CreateSubscription": "ALLOW",
-		"/linkall.vanus.controller.TriggerController/DeleteSubscription": "ALLOW",
-		"/linkall.vanus.controller.TriggerController/GetSubscription":    "ALLOW",
-		"/linkall.vanus.controller.TriggerController/ListSubscription":   "ALLOW",
-	}
+	httpRequestPrefix = "/gateway"
 )
 
 var (
@@ -67,28 +52,44 @@ type EventData struct {
 
 type ceGateway struct {
 	// ceClient  v2.Client
-	busWriter sync.Map
-	config    Config
-	client    eb.Client
-	cp        *ctrlProxy
-	tracer    *tracing.Tracer
+	busWriter  sync.Map
+	config     Config
+	client     eb.Client
+	proxySrv   *proxy.ControllerProxy
+	tracer     *tracing.Tracer
+	ceListener net.Listener
 }
 
 func NewGateway(config Config) *ceGateway {
 	return &ceGateway{
-		config: config,
-		client: eb.Connect(config.ControllerAddr),
-		cp:     newCtrlProxy(config.Port+ctrlProxyPortShift, allowCtrlProxyList, config.ControllerAddr),
-		tracer: tracing.NewTracer("cloudevents", trace.SpanKindServer),
+		config:   config,
+		client:   eb.Connect(config.ControllerAddr),
+		proxySrv: proxy.NewControllerProxy(config.GetProxyConfig()),
+		tracer:   tracing.NewTracer("cloudevents", trace.SpanKindServer),
 	}
 }
 
-func (ga *ceGateway) StartCtrlProxy(ctx context.Context) error {
-	return ga.cp.start(ctx)
+func (ga *ceGateway) Start(ctx context.Context) error {
+	if err := ga.startCloudEventsReceiver(ctx); err != nil {
+		return err
+	}
+	if err := ga.proxySrv.Start(); err != nil {
+		return err
+	}
+	return nil
 }
 
-func (ga *ceGateway) StartReceive(ctx context.Context) error {
-	ls, err := net.Listen("tcp", fmt.Sprintf(":%d", ga.config.Port))
+func (ga *ceGateway) Stop() {
+	ga.proxySrv.Stop()
+	if err := ga.ceListener.Close(); err != nil {
+		log.Warning(context.Background(), "close CloudEvents listener error", map[string]interface{}{
+			log.KeyError: err,
+		})
+	}
+}
+
+func (ga *ceGateway) startCloudEventsReceiver(ctx context.Context) error {
+	ls, err := net.Listen("tcp", fmt.Sprintf(":%d", ga.config.GetCloudEventReceiverPort()))
 	if err != nil {
 		return err
 	}
@@ -97,7 +98,14 @@ func (ga *ceGateway) StartReceive(ctx context.Context) error {
 	if err != nil {
 		return err
 	}
-	return c.StartReceiver(ctx, ga.receive)
+
+	ga.ceListener = ls
+	go func() {
+		if err := c.StartReceiver(ctx, ga.receive); err != nil {
+			panic(fmt.Sprintf("start CloudEvents receiver failed: %s", err.Error()))
+		}
+	}()
+	return nil
 }
 
 func (ga *ceGateway) receive(ctx context.Context, event v2.Event) (*v2.Event, protocol.Result) {