Implement video feature in docker engine

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

@@ -13,6 +13,7 @@ import (
  "github.com/taskcluster/taskcluster-worker/engines/docker/network"
  "github.com/taskcluster/taskcluster-worker/runtime"
  "github.com/taskcluster/taskcluster-worker/runtime/util"
+ funk "github.com/thoas/go-funk"
 )
 
 type engine struct {
@@ -23,6 +24,7 @@ type engine struct {
  config configType
  networks *network.Pool
  imageCache *imagecache.ImageCache
+ video *videoDeviceManager
 }
 
 type engineProvider struct {
@@ -42,6 +44,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 +58,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 +93,15 @@ 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",
+ },
+ },
+ },
  },
  Required: []string{
  "image",
@@ -107,6 +130,13 @@ func (e *engine) NewSandboxBuilder(options engines.SandboxOptions) (engines.Sand
  var p payloadType
  schematypes.MustValidateAndMap(e.PayloadSchema(), options.Payload, &p)
 
+ p.Devices = funk.UniqString(p.Devices)
+
+ 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",