[vm] delete vm intance cache (#544)

Author: huangzhiran

vm/server.go renamed to vm/instance.go

@@ -14,10 +14,9 @@ import (
 
 type instance struct {
 	conn *grpc.ClientConn
-	resp *proto.CreateResponse
 }
 
-func newInstance(ctx context.Context, endpoint string, projectID uint64, executeBinary string, expParam string) (*instance, error) {
+func newInstance(ctx context.Context, projectID uint64, endpoint, executeBinary, expParam string) (*instance, error) {
 	conn, err := grpc.Dial(endpoint, grpc.WithTransportCredentials(insecure.NewCredentials()))
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to dial vm server")
@@ -29,11 +28,10 @@ func newInstance(ctx context.Context, endpoint string, projectID uint64, execute
 		Content:   executeBinary,
 		ExpParam:  expParam,
 	}
-	resp, err := cli.Create(ctx, req)
-	if err != nil {
+	if _, err := cli.Create(ctx, req); err != nil {
 		return nil, errors.Wrap(err, "failed to create vm instance")
 	}
-	return &instance{conn: conn, resp: resp}, nil
+	return &instance{conn: conn}, nil
 }
 
 func (i *instance) execute(ctx context.Context, task *task.Task) ([]byte, error) {
@@ -51,12 +49,13 @@ func (i *instance) execute(ctx context.Context, task *task.Task) ([]byte, error)
 	cli := proto.NewVmRuntimeClient(i.conn)
 	resp, err := cli.ExecuteOperator(ctx, req)
 	if err != nil {
-		slog.Debug("request", "body", req)
 		return nil, errors.Wrap(err, "failed to execute vm instance")
 	}
 	return resp.Result, nil
 }
 
 func (i *instance) release() {
-	i.conn.Close()
+	if err := i.conn.Close(); err != nil {
+		slog.Error("failed to close grpc conn", "error", err)
+	}
 }

vm/server_test.go renamed to vm/instance_test.go

@@ -31,7 +31,7 @@ func TestNewInstance(t *testing.T) {
 		defer p.Reset()
 
 		p.ApplyFuncReturn(grpc.Dial, nil, errors.New(t.Name()))
-		_, err := newInstance(context.Background(), "any", 100, "any", "any")
+		_, err := newInstance(context.Background(), 100, "any", "any", "any")
 		r.ErrorContains(err, t.Name())
 	})
 
@@ -43,7 +43,7 @@ func TestNewInstance(t *testing.T) {
 		p.ApplyFuncReturn(proto.NewVmRuntimeClient, &MockClient{})
 		p.ApplyMethodReturn(&MockClient{}, "Create", nil, errors.New(t.Name()))
 
-		_, err := newInstance(context.Background(), "any", 100, "any", "any")
+		_, err := newInstance(context.Background(), 100, "any", "any", "any")
 		r.ErrorContains(err, t.Name())
 	})
 
@@ -56,7 +56,7 @@ func TestNewInstance(t *testing.T) {
 		p.ApplyMethodReturn(&MockClient{}, "Create", &proto.CreateResponse{}, nil)
 		p.ApplyMethodReturn(&grpc.ClientConn{}, "Close", nil)
 
-		i, err := newInstance(context.Background(), "any", 100, "any", "any")
+		i, err := newInstance(context.Background(), 100, "any", "any", "any")
 		r.NoError(err, t.Name())
 		r.NotNil(i)
 		i.release()

vm/manager.go

@@ -2,7 +2,6 @@ package vm
 
 import (
 	"context"
-	"fmt"
 	"log/slog"
 
 	"github.com/pkg/errors"
@@ -21,7 +20,6 @@ const (
 
 type Handler struct {
 	vmServerEndpoints map[Type]string
-	instanceManager   *manager
 }
 
 func (r *Handler) Handle(task *task.Task, vmtype Type, code string, expParam string) ([]byte, error) {
@@ -30,12 +28,12 @@ func (r *Handler) Handle(task *task.Task, vmtype Type, code string, expParam str
 		return nil, errors.New("unsupported vm type")
 	}
 
-	ins, err := r.instanceManager.acquire(task.ProjectID, endpoint, code, expParam)
+	ins, err := newInstance(context.Background(), task.ProjectID, endpoint, code, expParam)
 	if err != nil {
-		return nil, errors.Wrap(err, "failed to get instance")
+		return nil, errors.Wrap(err, "failed to new instance")
 	}
-	slog.Debug(fmt.Sprintf("acquire %s instance success", vmtype))
-	defer r.instanceManager.release(task.ProjectID, ins)
+	defer ins.release()
+	slog.Debug("acquire vm instance success", "vm_type", vmtype)
 
 	res, err := ins.execute(context.Background(), task)
 	if err != nil {
@@ -47,6 +45,5 @@ func (r *Handler) Handle(task *task.Task, vmtype Type, code string, expParam str
 func NewHandler(vmServerEndpoints map[Type]string) *Handler {
 	return &Handler{
 		vmServerEndpoints: vmServerEndpoints,
-		instanceManager:   newManager(),
 	}
 }

vm/manager_test.go

@@ -33,13 +33,11 @@ func TestHandler_Handle(t *testing.T) {
 		r.Error(err)
 	})
 
-	t.Run("FailedToAcquireVmInstance", func(t *testing.T) {
+	t.Run("FailedToNewVmInstance", func(t *testing.T) {
 		p := gomonkey.NewPatches()
 		defer p.Reset()
 
-		p.ApplyPrivateMethod(&manager{}, "acquire", func(uint64, string, string, string) (*instance, error) {
-			return nil, errors.New(t.Name())
-		})
+		p.ApplyFuncReturn(newInstance, nil, errors.New(t.Name()))
 		_, err := h.Handle(&task.Task{}, ZKwasm, "any", "any")
 		r.ErrorContains(err, t.Name())
 	})
@@ -48,10 +46,8 @@ func TestHandler_Handle(t *testing.T) {
 		p := gomonkey.NewPatches()
 		defer p.Reset()
 
-		p.ApplyPrivateMethod(&manager{}, "acquire", func(uint64, string, string, string) (*instance, error) {
-			return &instance{}, nil
-		})
-		p.ApplyPrivateMethod(&manager{}, "release", func(uint64, *instance) {})
+		p.ApplyFuncReturn(newInstance, &instance{}, nil)
+		p.ApplyPrivateMethod(&instance{}, "release", func() {})
 		p.ApplyPrivateMethod(&instance{}, "execute", func(context.Context, *task.Task) ([]byte, error) {
 			return nil, errors.New(t.Name())
 		})
@@ -64,10 +60,8 @@ func TestHandler_Handle(t *testing.T) {
 		p := gomonkey.NewPatches()
 		defer p.Reset()
 
-		p.ApplyPrivateMethod(&manager{}, "acquire", func(uint64, string, string, string) (*instance, error) {
-			return &instance{}, nil
-		})
-		p.ApplyPrivateMethod(&manager{}, "release", func(uint64, *instance) {})
+		p.ApplyFuncReturn(newInstance, &instance{}, nil)
+		p.ApplyPrivateMethod(&instance{}, "release", func() {})
 		p.ApplyPrivateMethod(&instance{}, "execute", func(context.Context, *task.Task) ([]byte, error) {
 			return []byte("any"), nil
 		})