Implement video feature in docker engine

Author: Wander Lairson Costa

engines/docker/config.go

@@ -8,8 +8,9 @@ import (
 )
 
 type configType struct {
-	DockerSocket string `json:"dockerSocket"`
-	Privileged   string `json:"privileged"`
+	DockerSocket  string `json:"dockerSocket"`
+	Privileged    string `json:"privileged"`
+	EnableDevices bool   `json:"enableDevices"`
 }
 
 const (
@@ -49,6 +50,13 @@ var configSchema = schematypes.Object{
 				privilegedNever,
 			},
 		},
+		"enableDevices": schematypes.Boolean{
+			Title: "Enable host devices",
+			Description: util.Markdown(`
+				When true, this enables the support for host devices inside the container,
+				such as video and sound.
+			`),
+		},
 	},
 	Required: []string{
 		"privileged",

engines/docker/engine.go

@@ -23,6 +23,7 @@ type engine struct {
 	config      configType
 	networks    *network.Pool
 	imageCache  *imagecache.ImageCache
+	video       *videoDeviceManager
 }
 
 type engineProvider struct {
@@ -42,6 +43,8 @@ func (p engineProvider) NewEngine(options engines.EngineOptions) (engines.Engine
 	var c configType
 	schematypes.MustValidateAndMap(configSchema, options.Config, &c)
 
+	debug(fmt.Sprintf("Devices enabled = %v", c.EnableDevices))
+
 	if c.DockerSocket == "" {
 		c.DockerSocket = "unix:///var/run/docker.sock" // default docker socket
 	}
@@ -54,20 +57,30 @@ func (p engineProvider) NewEngine(options engines.EngineOptions) (engines.Engine
 
 	env := options.Environment
 	monitor := options.Monitor
+	var video *videoDeviceManager
+	if c.EnableDevices {
+		video, err = newVideoDeviceManager()
+		if err != nil {
+			return nil, err
+		}
+	}
+
 	return &engine{
 		config:      c,
 		docker:      client,
 		Environment: env,
 		monitor:     monitor,
 		networks:    network.NewPool(client, monitor.WithPrefix("network-pool")),
 		imageCache:  imagecache.New(client, env.GarbageCollector, monitor.WithPrefix("image-cache")),
+		video:       video,
 	}, nil
 }
 
 type payloadType struct {
 	Image      interface{} `json:"image"`
 	Command    []string    `json:"command"`
 	Privileged bool        `json:"privileged"`
+	Devices    []string    `json:"devices"`
 }
 
 func (e *engine) PayloadSchema() schematypes.Object {
@@ -79,6 +92,16 @@ func (e *engine) PayloadSchema() schematypes.Object {
 				Description: "Command to run inside the container.",
 				Items:       schematypes.String{},
 			},
+			"devices": schematypes.Array{
+				Title:       "Devices",
+				Description: "List of host devices required.",
+				Items: schematypes.StringEnum{
+					Options: []string{
+						"video",
+					},
+				},
+				Unique: true,
+			},
 		},
 		Required: []string{
 			"image",
@@ -107,6 +130,11 @@ func (e *engine) NewSandboxBuilder(options engines.SandboxOptions) (engines.Sand
 	var p payloadType
 	schematypes.MustValidateAndMap(e.PayloadSchema(), options.Payload, &p)
 
+	if len(p.Devices) > 0 && e.config.EnableDevices == false {
+		options.TaskContext.LogError(fmt.Sprintf("Task requests device %v, but device support is not enabled", p.Devices))
+		return nil, runtime.NewMalformedPayloadError("Devices feature is not enabled")
+	}
+
 	// Check if privileged == true is allowed
 	switch e.config.Privileged {
 	case privilegedAllow: // Check scope if p.Privileged is true

engines/docker/sandbox.go

@@ -15,6 +15,7 @@ import (
 	"github.com/taskcluster/taskcluster-worker/runtime"
 	"github.com/taskcluster/taskcluster-worker/runtime/atomics"
 	"github.com/taskcluster/taskcluster-worker/runtime/ioext"
+	funk "github.com/thoas/go-funk"
 )
 
 const dockerEngineKillTimeout = 5 * time.Second
@@ -32,6 +33,7 @@ type sandbox struct {
 	taskCtx       *runtime.TaskContext
 	networkHandle *network.Handle
 	imageHandle   *imagecache.ImageHandle
+	videoDev      *device
 }
 
 func newSandbox(sb *sandboxBuilder) (*sandbox, error) {
@@ -52,6 +54,21 @@ func newSandbox(sb *sandboxBuilder) (*sandbox, error) {
 		return nil, errors.Wrap(err, "docker.CreateNetwork failed")
 	}
 
+	devices := []docker.Device{}
+	var dev *device
+	if funk.InStrings(sb.payload.Devices, "video") {
+		dev = sb.e.video.claim()
+		if dev == nil {
+			return nil, errors.New("No video device available")
+		}
+		debug(fmt.Sprintf("Claimed %s", dev.path))
+		devices = append(devices, docker.Device{
+			PathOnHost:        dev.path,
+			PathInContainer:   dev.path,
+			CgroupPermissions: "rwm",
+		})
+	}
+
 	// Create the container
 	container, err := sb.e.docker.CreateContainer(docker.CreateContainerOptions{
 		Config: &docker.Config{
@@ -70,6 +87,7 @@ func newSandbox(sb *sandboxBuilder) (*sandbox, error) {
 			// to the proxies added to proxyMux above..
 			ExtraHosts: []string{fmt.Sprintf("taskcluster:%s", networkHandle.Gateway())},
 			Mounts:     sb.mounts,
+			Devices:    devices,
 		},
 		NetworkingConfig: &docker.NetworkingConfig{
 			EndpointsConfig: map[string]*docker.EndpointConfig{
@@ -79,6 +97,7 @@ func newSandbox(sb *sandboxBuilder) (*sandbox, error) {
 	})
 	if err != nil {
 		imageHandle.Release()
+		sb.e.video.release(dev)
 		return nil, runtime.NewMalformedPayloadError(
 			"could not create container: " + err.Error())
 	}
@@ -87,6 +106,7 @@ func newSandbox(sb *sandboxBuilder) (*sandbox, error) {
 	storage, err := sb.e.Environment.TemporaryStorage.NewFolder()
 	if err != nil {
 		imageHandle.Release()
+		sb.e.video.release(dev)
 		monitor.ReportError(err, "failed to create temporary folder")
 		return nil, runtime.ErrFatalInternalError
 	}
@@ -101,6 +121,7 @@ func newSandbox(sb *sandboxBuilder) (*sandbox, error) {
 			"containerId": container.ID,
 			"networkId":   networkHandle.NetworkID(),
 		}),
+		videoDev: dev,
 	}
 
 	// attach to the container before starting so that we get all the logs
@@ -277,5 +298,9 @@ func (s *sandbox) dispose() error {
 	if hasErr {
 		return runtime.ErrNonFatalInternalError
 	}
+
+	if s.videoDev != nil {
+		s.videoDev.claimed = false
+	}
 	return nil
 }

engines/docker/video.go

@@ -0,0 +1,51 @@
+package dockerengine
+
+import (
+	"path/filepath"
+	"regexp"
+
+	"github.com/pkg/errors"
+	funk "github.com/thoas/go-funk"
+)
+
+type device struct {
+	path    string
+	claimed bool
+}
+
+type videoDeviceManager struct {
+	devices []device
+}
+
+func newVideoDeviceManager() (*videoDeviceManager, error) {
+	matches, err := filepath.Glob("/dev/video*")
+	if err != nil {
+		return nil, errors.Wrap(err, "Failed to call filepath.Glob function")
+	}
+
+	r := regexp.MustCompile("/dev/video[0-9]+")
+	matches = funk.FilterString(matches, r.MatchString)
+
+	devices := make([]device, len(matches))
+	for i := range devices {
+		devices[i].path = matches[i]
+	}
+
+	return &videoDeviceManager{
+		devices: devices,
+	}, nil
+}
+
+func (d *videoDeviceManager) claim() *device {
+	for i := range d.devices {
+		if !d.devices[i].claimed {
+			return &d.devices[i]
+		}
+	}
+
+	return nil
+}
+
+func (d *videoDeviceManager) release(dev *device) {
+	dev.claimed = false
+}

engines/docker/video_test.go

@@ -0,0 +1,36 @@
+// +build dockervideo
+
+package dockerengine
+
+import (
+	"testing"
+
+	"github.com/taskcluster/taskcluster-worker/engines/enginetest"
+)
+
+// Image and tag used in test cases below
+const (
+	videoDockerImageName = "alpine:3.6"
+)
+
+var videoProvider = &enginetest.EngineProvider{
+	Engine: "docker",
+	Config: `{
+		"privileged": "allow",
+		"enableDevices": true
+	}`,
+}
+
+func TestVideo(t *testing.T) {
+	c := enginetest.LoggingTestCase{
+		EngineProvider: videoProvider,
+		Target:         "/dev/video0",
+		TargetPayload: `{
+			"command": ["sh", "-c", "ls /dev/video0"],
+			"devices": ["video"],
+			"image": "` + videoDockerImageName + `"
+		}`,
+	}
+
+	c.Test()
+}

engines/enginetest/logging.go

@@ -71,7 +71,13 @@ func (c *LoggingTestCase) TestSilentTask() {
 
 // Test will run all logging tests
 func (c *LoggingTestCase) Test() {
-	c.TestLogTarget()
-	c.TestLogTargetWhenFailing()
-	c.TestSilentTask()
+	if len(c.TargetPayload) > 0 {
+		c.TestLogTarget()
+	}
+	if len(c.FailingPayload) > 0 {
+		c.TestLogTargetWhenFailing()
+	}
+	if len(c.SilentPayload) > 0 {
+		c.TestSilentTask()
+	}
 }

vendor/vendor.json

@@ -677,6 +677,12 @@
 			"revision": "d4fa08268f573f25df477fedc813f26bfb833761",
 			"revisionTime": "2016-08-05T02:05:43Z"
 		},
+		{
+			"checksumSHA1": "UHPK5Fi61Zqvf/ctF4s9hwjHFGU=",
+			"path": "github.com/thoas/go-funk",
+			"revision": "d2deeb5709c1da54d5da8c76b2f65421f5ff8de4",
+			"revisionTime": "2018-05-05T20:14:24Z"
+		},
 		{
 			"checksumSHA1": "HN3pLd5cC+QXkX8j8FsCCB3FzSI=",
 			"path": "github.com/tinylib/msgp/msgp",