feat: add proxy for controller in gateway

Author: wenfengwang

internal/gateway/grpc_proxy.go

@@ -0,0 +1,97 @@
+// Copyright 2022 Linkall Inc.
+//
+// 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 gateway
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"sync"
+	"time"
+
+	"github.com/linkall-labs/vanus/observability/log"
+	"github.com/mwitkow/grpc-proxy/proxy"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/metadata"
+	"google.golang.org/grpc/status"
+)
+
+func startGRPCProxy(ctx context.Context, port int, ctrlList []string) error {
+	cp := &ctrlProxy{ctrlLists: ctrlList}
+	grpcServer := grpc.NewServer(
+		//grpc.CustomCodec(proxy.Codec()),
+		grpc.UnknownServiceHandler(proxy.TransparentHandler(cp.director)),
+	)
+	listen, err := net.Listen("tcp", fmt.Sprintf(":%d", port))
+	if err != nil {
+		return err
+	}
+	go func() {
+		if err := grpcServer.Serve(listen); err != nil {
+			panic("start grpc server failed: " + err.Error())
+		}
+		log.Info(ctx, "the grpc server shutdown", nil)
+	}()
+
+	go func() {
+		for {
+			select {
+			case <-ctx.Done():
+				grpcServer.GracefulStop()
+			}
+		}
+	}()
+	return nil
+}
+
+type ctrlProxy struct {
+	ctrlLists  []string
+	rwMutex    sync.RWMutex
+	leaderConn *grpc.ClientConn
+	haveLeader bool
+}
+
+func (cp *ctrlProxy) director(ctx context.Context, fullMethodName string) (context.Context, *grpc.ClientConn, error) {
+	md, _ := metadata.FromIncomingContext(ctx)
+	outCtx := metadata.NewOutgoingContext(ctx, md.Copy())
+
+	if !cp.haveLeader {
+		return outCtx, nil, status.Errorf(codes.Internal, "No leader founded")
+	}
+
+	switch fullMethodName {
+	case "xxx":
+	default:
+		return outCtx, nil, status.Errorf(codes.Unimplemented, "Unknown method")
+	}
+	cp.rwMutex.RLock()
+	defer cp.rwMutex.RUnlock()
+	return outCtx, cp.leaderConn, nil
+}
+
+func (cp *ctrlProxy) updateLeader(ctx context.Context) {
+	ticker := time.NewTicker(time.Second)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			cp.rwMutex.Lock()
+			cp.rwMutex.Unlock()
+		}
+	}
+}

internal/gateway/grpc_proxy_test.go

@@ -0,0 +1 @@
+package gateway