chore: add grpc handler and proto file

pkg/kb_agent/handlers/grpc_handler.go

@@ -22,6 +22,9 @@ package handlers
 import (
 	"context"
 
+	"github.com/apecloud/kubeblocks/pkg/constant"
+	"github.com/apecloud/kubeblocks/pkg/kb_agent/plugin"
+	viper "github.com/apecloud/kubeblocks/pkg/viperx"
 	"github.com/go-logr/logr"
 	"github.com/pkg/errors"
 	ctrl "sigs.k8s.io/controller-runtime"
@@ -30,24 +33,53 @@ import (
 )
 
 type GRPCHandler struct {
-	Logger logr.Logger
+	Logger     logr.Logger
+	grpcClient plugin.GrpcClient
 }
 
 var _ Handler = &GRPCHandler{}
 
 func NewGRPCHandler(properties map[string]string) (*GRPCHandler, error) {
 	logger := ctrl.Log.WithName("GRPC handler")
+	host := viper.GetString(constant.KBEnvPodIP)
+	if h, ok := properties["host"]; ok && h != "" {
+		host = h
+	}
+	port, ok := properties["port"]
+	if !ok || port == "" {
+		return &GRPCHandler{logger, nil}, nil
+	}
+	grpcClient, err := plugin.NewGRPCClient(host, port)
+	if err != nil {
+		return nil, errors.Wrap(err, "new client failed")
+	}
 	h := &GRPCHandler{
-		Logger: logger,
+		Logger:     logger,
+		grpcClient: grpcClient,
 	}
-
 	return h, nil
 }
 
 func (h *GRPCHandler) Do(ctx context.Context, setting util.HandlerSpec, args map[string]any) (*Response, error) {
 	if setting.GPRC == nil {
 		return nil, errors.New("grpc setting is nil")
 	}
-	// TODO: implement grpc handler
-	return nil, ErrNotImplemented
+	if args == nil {
+		return nil, errors.New("args is nil")
+	}
+	parameters, err := util.WrapperArgs(args)
+	if err != nil {
+		return nil, errors.Wrap(err, "wrapper args failed")
+	}
+	request := &plugin.Request{
+		MethodName: args["methodName"].(string),
+		Parameters: parameters,
+	}
+	client := h.grpcClient
+	result, err := client.Call(ctx, request)
+	if err != nil {
+		return nil, errors.Wrap(err, "grpc call failed")
+	}
+	resp := &Response{Message: result.Message}
+	return resp, nil
 }

pkg/kb_agent/handlers/grpc_handler_test.go

@@ -21,39 +21,112 @@ package handlers
 
 import (
 	"context"
+	"flag"
+	"log"
+	"net"
 	"testing"
-
+
+	"github.com/apecloud/kubeblocks/pkg/kb_agent/plugin"
+	"google.golang.org/grpc"
+
 	"github.com/apecloud/kubeblocks/pkg/kb_agent/util"
 	"github.com/pkg/errors"
 	"github.com/stretchr/testify/assert"
 )
 
 func TestNewGRPCHandler(t *testing.T) {
-	handler, err := NewGRPCHandler(nil)
+	handler, err := NewGRPCHandler(map[string]string{
+		"host": "localhost",
+		"port": "50051",
+	})
 	assert.NotNil(t, handler)
 	assert.Nil(t, err)
 }
 
 func TestGRPCHandlerDo(t *testing.T) {
 	ctx := context.Background()
-	handler := &GRPCHandler{}
+	go func() {
+		flag.Parse()
+		lis, err := net.Listen("tcp", ":50051")
+		if err != nil {
+			log.Fatalf("failed to listen: %v", err)
+		}
+		s := grpc.NewServer()
+		plugin.RegisterGrpcServer(s, &MockGrpcServer{})
+		log.Printf("server listening at %v", lis.Addr())
+		if err := s.Serve(lis); err != nil {
+			log.Fatalf("failed to serve: %v", err)
+		}
+	}()
+	setting := util.HandlerSpec{
+		GPRC: map[string]string{
+			"host": "localhost",
+			"port": "50051",
+		},
+	}
 	t.Run("grpc handler is nil", func(t *testing.T) {
 		setting := util.HandlerSpec{
 			GPRC: nil,
 		}
+		handler := &GRPCHandler{}
 		do, err := handler.Do(ctx, setting, nil)
 		assert.Nil(t, do)
 		assert.NotNil(t, err)
 		assert.Error(t, err, errors.New("grpc setting is nil"))
 	})
 
-	t.Run("grpc handler is not nil but not implemented", func(t *testing.T) {
-		setting := util.HandlerSpec{
-			GPRC: map[string]string{"test": "test"},
-		}
+	t.Run("grpc args is nil", func(t *testing.T) {
+		handler, err := NewGRPCHandler(setting.GPRC)
+		assert.Nil(t, err)
 		do, err := handler.Do(ctx, setting, nil)
 		assert.Nil(t, do)
 		assert.NotNil(t, err)
-		assert.Error(t, err, ErrNotImplemented)
+		assert.Error(t, err, errors.New("args is nil"))
+	})
+
+	t.Run("grpc do success", func(t *testing.T) {
+		handler, err := NewGRPCHandler(setting.GPRC)
+		assert.Nil(t, err)
+		args := map[string]interface{}{
+			"methodName": "test",
+			"username":   "admin",
+			"password":   "admin",
+		}
+		result, err := handler.Do(ctx, setting, args)
+		assert.NotNil(t, result)
+		assert.Nil(t, err)
+		assert.Equal(t, "methodName : test", result.Message)
 	})
+
+	t.Run("grpc do not implemented", func(t *testing.T) {
+		handler, err := NewGRPCHandler(setting.GPRC)
+		assert.Nil(t, err)
+		args := map[string]interface{}{
+			"methodName": "notImplemented",
+			"username":   "admin",
+		}
+		result, err := handler.Do(ctx, setting, args)
+		assert.Nil(t, result)
+		assert.NotNil(t, err)
+		assert.Error(t, err, errors.New("not implemented"))
+	})
+}
+
+type MockGrpcServer struct {
+	*plugin.UnimplementedGrpcServer
+}
+
+func (s *MockGrpcServer) Call(ctx context.Context, in *plugin.Request) (*plugin.Response, error) {
+	methodName := in.MethodName
+	parameters := in.Parameters
+	m, _ := util.ParseArgs(parameters)
+	for k, v := range m {
+		log.Printf("key: %s, value: %s", k, v)
+	}
+	switch methodName {
+	case "test":
+		return &plugin.Response{Message: "methodName : " + methodName}, nil
+	default:
+		return nil, errors.New("not implemented")
+	}
 }

pkg/kb_agent/httpserver/apis_test.go

@@ -71,16 +71,6 @@ func TestActionHandler(t *testing.T) {
 		assert.JSONEq(t, `{"errorCode":"ERR_MALFORMED_REQUEST_DATA","message":"no action in request"}`, string(reqCtx.Response.Body()))
 	})
 
-	t.Run("action not implemented", func(t *testing.T) {
-		reqCtx := &fasthttp.RequestCtx{}
-		reqCtx.Request.Header.SetMethod(fasthttp.MethodPost)
-		reqCtx.Request.Header.SetContentType("application/json")
-		reqCtx.Request.SetBody([]byte(`{"action":"test"}`))
-		actionHandler(reqCtx)
-		assert.Equal(t, fasthttp.StatusNotImplemented, reqCtx.Response.StatusCode())
-		assert.JSONEq(t, `{"errorCode":"ERR_ACTION_FAILED","message":"action exec failed: not implemented"}`, string(reqCtx.Response.Body()))
-	})
-
 	t.Run("action exec failed", func(t *testing.T) {
 		msg := "action exec failed"
 		reqCtx := &fasthttp.RequestCtx{}