Merge pull request #158 from ibuildthecloud/serial2

Make compute lock be more fine grained for start

core/compute/compute.go

@@ -15,6 +15,7 @@ import (
 	"github.com/rancher/agent/core/storage"
 	"github.com/rancher/agent/model"
 	"github.com/rancher/agent/utilities/constants"
+	dutils "github.com/rancher/agent/utilities/docker"
 	"github.com/rancher/agent/utilities/utils"
 	"golang.org/x/net/context"
 )
@@ -115,7 +116,10 @@ func DoInstanceActivate(instance model.Instance, host model.Host, progress *prog
 		created = true
 	}
 
-	if startErr := dockerClient.ContainerStart(context.Background(), containerID, types.ContainerStartOptions{}); startErr != nil {
+	startErr := dutils.Serialize(func() error {
+		return dockerClient.ContainerStart(context.Background(), containerID, types.ContainerStartOptions{})
+	})
+	if startErr != nil {
 		if created {
 			if err := utils.RemoveContainer(dockerClient, containerID); err != nil {
 				return errors.Wrap(err, constants.DoInstanceActivateError+"failed to remove container")

core/compute/compute_unix.go

@@ -21,6 +21,7 @@ import (
 	"github.com/rancher/agent/core/hostInfo"
 	"github.com/rancher/agent/model"
 	"github.com/rancher/agent/utilities/constants"
+	dutils "github.com/rancher/agent/utilities/docker"
 	"github.com/rancher/agent/utilities/utils"
 )
 
@@ -474,10 +475,18 @@ func setupDeviceOptions(hostConfig *container.HostConfig, instance model.Instanc
 }
 
 func dockerContainerCreate(ctx context.Context, dockerClient *client.Client, config *container.Config, hostConfig *container.HostConfig, networkingConfig *network.NetworkingConfig, containerName string) (types.ContainerCreateResponse, error) {
+	var (
+		ret types.ContainerCreateResponse
+		err error
+	)
 	if err := modifyForCNI(dockerClient, config, hostConfig); err != nil {
 		return types.ContainerCreateResponse{}, err
 	}
-	return dockerClient.ContainerCreate(context.Background(), config, hostConfig, networkingConfig, containerName)
+	dutils.Serialize(func() error {
+		ret, err = dockerClient.ContainerCreate(context.Background(), config, hostConfig, networkingConfig, containerName)
+		return err
+	})
+	return ret, err
 }
 
 func getContainerName(instance model.Instance) string {

handlers/common.go

@@ -3,7 +3,6 @@ package handlers
 import (
 	"fmt"
 	"os"
-	"sync"
 	"time"
 
 	"github.com/Sirupsen/logrus"
@@ -19,11 +18,6 @@ import (
 	"golang.org/x/net/context"
 )
 
-var (
-	serializeCompute = false
-	computeLock      = sync.Mutex{}
-)
-
 type Handler struct {
 	compute      *ComputeHandler
 	storage      *StorageHandler
@@ -38,16 +32,16 @@ func GetHandlers() (map[string]revents.EventHandler, error) {
 		return nil, err
 	}
 
-	serializeCompute = (info.Driver == "devicemapper")
-	logrus.Infof("Serialize compute requests: %v, driver: %s", serializeCompute, info.Driver)
+	docker.SerializeCompute = (info.Driver == "devicemapper")
+	logrus.Infof("Serialize compute requests: %v, driver: %s", docker.SerializeCompute, info.Driver)
 
 	return map[string]revents.EventHandler{
-		"compute.instance.activate":   serialize(cleanLog(logRequest(handler.compute.InstanceActivate))),
-		"compute.instance.deactivate": serialize(cleanLog(logRequest(handler.compute.InstanceDeactivate))),
-		"compute.instance.force.stop": serialize(cleanLog(logRequest(handler.compute.InstanceForceStop))),
+		"compute.instance.activate":   cleanLog(logRequest(handler.compute.InstanceActivate)),
+		"compute.instance.deactivate": docker.SerializeHandler(cleanLog(logRequest(handler.compute.InstanceDeactivate))),
+		"compute.instance.force.stop": docker.SerializeHandler(cleanLog(logRequest(handler.compute.InstanceForceStop))),
 		"compute.instance.inspect":    cleanLog(logRequest(handler.compute.InstanceInspect)),
 		"compute.instance.pull":       cleanLog(logRequest(handler.compute.InstancePull)),
-		"compute.instance.remove":     serialize(cleanLog(logRequest(handler.compute.InstanceRemove))),
+		"compute.instance.remove":     docker.SerializeHandler(cleanLog(logRequest(handler.compute.InstanceRemove))),
 		"storage.image.activate":      cleanLog(logRequest(handler.storage.ImageActivate)),
 		"storage.volume.activate":     cleanLog(logRequest(handler.storage.VolumeActivate)),
 		"storage.volume.remove":       cleanLog(logRequest(handler.storage.VolumeRemove)),
@@ -56,23 +50,6 @@ func GetHandlers() (map[string]revents.EventHandler, error) {
 	}, nil
 }
 
-func serialize(f revents.EventHandler) revents.EventHandler {
-	return func(event *revents.Event, cli *client.RancherClient) error {
-		if !serializeCompute {
-			return f(event, cli)
-		}
-
-		computeLock.Lock()
-		logrus.Info("Compute lock")
-		defer func() {
-			logrus.Info("Compute unlock")
-			computeLock.Unlock()
-		}()
-
-		return f(event, cli)
-	}
-}
-
 func logRequest(f revents.EventHandler) revents.EventHandler {
 	return func(event *revents.Event, cli *client.RancherClient) error {
 		logrus.Infof("Received event: Name: %s, Event Id: %s, Resource Id: %s", event.Name, event.ID, event.ResourceID)

utilities/docker/compute_lock.go

@@ -0,0 +1,37 @@
+package docker
+
+import (
+	"sync"
+
+	"github.com/Sirupsen/logrus"
+	revents "github.com/rancher/event-subscriber/events"
+	"github.com/rancher/go-rancher/v2"
+)
+
+var (
+	SerializeCompute = false
+	computeLock      = sync.Mutex{}
+)
+
+func SerializeHandler(f revents.EventHandler) revents.EventHandler {
+	return func(event *revents.Event, cli *client.RancherClient) error {
+		return Serialize(func() error {
+			return f(event, cli)
+		})
+	}
+}
+
+func Serialize(f func() error) error {
+	if !SerializeCompute {
+		return f()
+	}
+
+	computeLock.Lock()
+	logrus.Info("Compute lock")
+	defer func() {
+		logrus.Info("Compute unlock")
+		computeLock.Unlock()
+	}()
+
+	return f()
+}