DockerIM should use operation mechanism properly

Author: 0405ysj

pkg/app/instances/docker.go

@@ -35,6 +35,7 @@ import (
 	"github.com/docker/docker/api/types/mount"
 	"github.com/docker/docker/api/types/volume"
 	"github.com/docker/docker/client"
+	"github.com/google/uuid"
 )
 
 const DockerIMType IMType = "docker"
@@ -54,18 +55,12 @@ const uaMountTarget = "/var/lib/cuttlefish-common/userartifacts"
 
 // Docker implementation of the instance manager.
 type DockerInstanceManager struct {
-	Config  Config
-	Client  *client.Client
-	mutexes sync.Map
+	Config     Config
+	Client     *client.Client
+	mutexes    sync.Map
+	operations sync.Map
 }
 
-type OPType string
-
-const (
-	CreateHostOPType OPType = "createhost"
-	DeleteHostOPType OPType = "deletehost"
-)
-
 func NewDockerInstanceManager(cfg Config, cli *client.Client) *DockerInstanceManager {
 	return &DockerInstanceManager{
 		Config: cfg,
@@ -85,26 +80,14 @@ func (m *DockerInstanceManager) CreateHost(zone string, _ *apiv1.CreateHostReque
 	if zone != "local" {
 		return nil, errors.NewBadRequestError("Invalid zone. It should be 'local'.", nil)
 	}
-	mu := m.getRWMutex(user)
-	mu.RLock()
-	defer mu.RUnlock()
-	ctx := context.TODO()
-	if err := m.downloadDockerImageIfNeeded(ctx); err != nil {
-		return nil, fmt.Errorf("failed to retrieve docker image name: %w", err)
-	}
-	// A docker volume is shared across all hosts under each user. If no volume
-	// exists for given user, create it.
-	if err := m.createDockerVolumeIfNeeded(ctx, user); err != nil {
-		return nil, fmt.Errorf("failed to prepare docker volume: %w", err)
-	}
-	host, err := m.createDockerContainer(ctx, user)
-	if err != nil {
-		return nil, fmt.Errorf("failed to prepare docker container: %w", err)
-	}
-	return &apiv1.Operation{
-		Name: EncodeOperationName(CreateHostOPType, host),
-		Done: true,
-	}, nil
+	op := m.newOperation()
+	go func() {
+		val, err := m.createHost(user)
+		if opErr := m.completeOperation(op.Name, &operationResult{Error: err, Value: val}); opErr != nil {
+			log.Printf("error completing operation %q: %v\n", op.Name, opErr)
+		}
+	}()
+	return &op, nil
 }
 
 func (m *DockerInstanceManager) ListHosts(zone string, user accounts.User, _ *ListHostsRequest) (*apiv1.ListHostsResponse, error) {
@@ -141,88 +124,25 @@ func (m *DockerInstanceManager) DeleteHost(zone string, user accounts.User, host
 	if zone != "local" {
 		return nil, errors.NewBadRequestError("Invalid zone. It should be 'local'.", nil)
 	}
-	ctx := context.TODO()
-	if err := m.deleteDockerContainer(ctx, user, host); err != nil {
-		return nil, fmt.Errorf("failed to delete docker container: %w", err)
-	}
-	// A docker volume is shared across all hosts under each user. If no host
-	// exists for given user, delete volume afterwards to cleanup.
-	if err := m.deleteDockerVolumeIfNeeded(ctx, user); err != nil {
-		return nil, fmt.Errorf("failed to cleanup docker volume: %w", err)
-	}
-	return &apiv1.Operation{
-		Name: EncodeOperationName(DeleteHostOPType, host),
-		Done: true,
-	}, nil
-}
-
-func EncodeOperationName(opType OPType, host string) string {
-	return string(opType) + "_" + host
-}
-
-func DecodeOperationName(name string) (OPType, string, error) {
-	words := strings.SplitN(name, "_", 2)
-	if len(words) == 2 {
-		return OPType(words[0]), words[1], nil
-	} else {
-		return "", "", errors.NewBadRequestError(fmt.Sprintf("cannot parse operation from %q", name), nil)
-	}
-}
-
-func (m *DockerInstanceManager) waitCreateHostOperation(host string) (*apiv1.HostInstance, error) {
-	ctx, cancel := context.WithTimeout(context.TODO(), 3*time.Minute)
-	defer cancel()
-	for {
-		select {
-		case <-ctx.Done():
-			return nil, errors.NewServiceUnavailableError("Wait for operation timed out", nil)
-		default:
-			res, err := m.Client.ContainerInspect(ctx, host)
-			if err != nil {
-				return nil, fmt.Errorf("failed to inspect docker container: %w", err)
-			}
-			if res.State.Running {
-				return &apiv1.HostInstance{
-					Name: host,
-				}, nil
-			}
-			time.Sleep(time.Second)
+	op := m.newOperation()
+	go func() {
+		val, err := m.deleteHost(user, host)
+		if opErr := m.completeOperation(op.Name, &operationResult{Error: err, Value: val}); opErr != nil {
+			log.Printf("error completing operation %q: %v\n", op.Name, opErr)
 		}
-	}
-}
-
-func (m *DockerInstanceManager) waitDeleteHostOperation(host string) (*apiv1.HostInstance, error) {
-	ctx, cancel := context.WithTimeout(context.TODO(), 3*time.Minute)
-	defer cancel()
-	resCh, errCh := m.Client.ContainerWait(ctx, host, "")
-	select {
-	case <-ctx.Done():
-		return nil, errors.NewServiceUnavailableError("Wait for operation timed out", nil)
-	case err := <-errCh:
-		return nil, fmt.Errorf("error is thrown while waiting for deleting host: %w", err)
-	case <-resCh:
-		return &apiv1.HostInstance{
-			Name: host,
-		}, nil
-	}
+	}()
+	return &op, nil
 }
 
 func (m *DockerInstanceManager) WaitOperation(zone string, _ accounts.User, name string) (any, error) {
 	if zone != "local" {
 		return nil, errors.NewBadRequestError("Invalid zone. It should be 'local'.", nil)
 	}
-	opType, host, err := DecodeOperationName(name)
+	val, err := m.waitOperation(name, 3*time.Minute)
 	if err != nil {
 		return nil, err
 	}
-	switch opType {
-	case CreateHostOPType:
-		return m.waitCreateHostOperation(host)
-	case DeleteHostOPType:
-		return m.waitDeleteHostOperation(host)
-	default:
-		return nil, errors.NewBadRequestError(fmt.Sprintf("operation type %s not found.", opType), nil)
-	}
+	return val.Value, val.Error
 }
 
 func (m *DockerInstanceManager) getIpAddr(container *types.Container) (string, error) {
@@ -318,6 +238,28 @@ func (m *DockerInstanceManager) getContainerLabel(host string, key string) (stri
 	return value, nil
 }
 
+func (m *DockerInstanceManager) createHost(user accounts.User) (*apiv1.HostInstance, error) {
+	mu := m.getRWMutex(user)
+	mu.RLock()
+	defer mu.RUnlock()
+	ctx := context.TODO()
+	if err := m.downloadDockerImageIfNeeded(ctx); err != nil {
+		return nil, fmt.Errorf("failed to retrieve docker image name: %w", err)
+	}
+	// A docker volume is shared across all hosts under each user. If no volume
+	// exists for given user, create it.
+	if err := m.createDockerVolumeIfNeeded(ctx, user); err != nil {
+		return nil, fmt.Errorf("failed to prepare docker volume: %w", err)
+	}
+	host, err := m.createDockerContainer(ctx, user)
+	if err != nil {
+		return nil, fmt.Errorf("failed to prepare docker container: %w", err)
+	}
+	return &apiv1.HostInstance{
+		Name: host,
+	}, nil
+}
+
 func (m *DockerInstanceManager) downloadDockerImageIfNeeded(ctx context.Context) error {
 	listRes, err := m.Client.ImageList(ctx, image.ListOptions{})
 	if err != nil {
@@ -364,6 +306,8 @@ func (m *DockerInstanceManager) createDockerVolumeIfNeeded(ctx context.Context,
 	return nil
 }
 
+const containerInspectRetryLimit = 5
+
 func (m *DockerInstanceManager) createDockerContainer(ctx context.Context, user accounts.User) (string, error) {
 	config := &container.Config{
 		AttachStdin: true,
@@ -401,7 +345,31 @@ func (m *DockerInstanceManager) createDockerContainer(ctx context.Context, user
 	if err := m.Client.ContainerExecStart(ctx, execRes.ID, container.ExecStartOptions{}); err != nil {
 		return "", fmt.Errorf("failed to start container execution %q: %w", strings.Join(execConfig.Cmd, " "), err)
 	}
-	return createRes.ID, nil
+	for i := 0; ; i++ {
+		res, err := m.Client.ContainerInspect(ctx, createRes.ID)
+		if err == nil && res.State.Running {
+			return createRes.ID, nil
+		}
+		if i >= containerInspectRetryLimit {
+			return "", fmt.Errorf("failed to inspect docker container: %w", err)
+		}
+		time.Sleep(time.Second)
+	}
+}
+
+func (m *DockerInstanceManager) deleteHost(user accounts.User, host string) (*apiv1.HostInstance, error) {
+	ctx := context.TODO()
+	if err := m.deleteDockerContainer(ctx, user, host); err != nil {
+		return nil, fmt.Errorf("failed to delete docker container: %w", err)
+	}
+	// A docker volume is shared across all hosts under each user. If no host
+	// exists for given user, delete volume afterwards to cleanup.
+	if err := m.deleteDockerVolumeIfNeeded(ctx, user); err != nil {
+		return nil, fmt.Errorf("failed to cleanup docker volume: %w", err)
+	}
+	return &apiv1.HostInstance{
+		Name: host,
+	}, nil
 }
 
 func (m *DockerInstanceManager) deleteDockerContainer(ctx context.Context, user accounts.User, host string) error {
@@ -454,3 +422,71 @@ func (m *DockerInstanceManager) getRWMutex(user accounts.User) *sync.RWMutex {
 	mu, _ := m.mutexes.LoadOrStore(user.Username(), &sync.RWMutex{})
 	return mu.(*sync.RWMutex)
 }
+
+type operationResult struct {
+	Error error
+	Value interface{}
+}
+
+type operationEntry struct {
+	op     apiv1.Operation
+	result *operationResult
+	mutex  sync.RWMutex
+	done   chan struct{}
+}
+
+const newOperationRetryLimit = 100
+
+func (m *DockerInstanceManager) newOperation() apiv1.Operation {
+	for i := 0; i < newOperationRetryLimit; i++ {
+		name := uuid.New().String()
+		newEntry := &operationEntry{
+			op: apiv1.Operation{
+				Name: name,
+				Done: false,
+			},
+			mutex: sync.RWMutex{},
+			done:  make(chan struct{}),
+		}
+		entry, loaded := m.operations.LoadOrStore(name, newEntry)
+		if !loaded {
+			// It succeeded to store a new operation entry.
+			return entry.(*operationEntry).op
+		}
+	}
+	panic("Reached newOperationRetryLimit")
+}
+
+func (m *DockerInstanceManager) completeOperation(name string, result *operationResult) error {
+	val, loaded := m.operations.Load(name)
+	if !loaded {
+		return fmt.Errorf("Operation not found for %q", name)
+	}
+	entry := val.(*operationEntry)
+
+	entry.mutex.Lock()
+	defer entry.mutex.Unlock()
+	entry.op.Done = true
+	entry.result = result
+	close(entry.done)
+	return nil
+}
+
+func (m *DockerInstanceManager) waitOperation(name string, dt time.Duration) (*operationResult, error) {
+	val, loaded := m.operations.Load(name)
+	if !loaded {
+		return nil, fmt.Errorf("Operation not found for %q", name)
+	}
+	entry := val.(*operationEntry)
+	ctx, cancel := context.WithTimeout(context.TODO(), 3*time.Minute)
+	defer cancel()
+	select {
+	case <-entry.done:
+		entry.mutex.RLock()
+		result := entry.result
+		entry.mutex.RUnlock()
+		return result, nil
+	case <-ctx.Done():
+		return nil, fmt.Errorf("Reached timeout for %q", name)
+	}
+}