[Carry #1517] Support v2 plugins; re-enable scheduler plugin filter

agent/exec/container/executor.go

@@ -8,6 +8,7 @@ import (
 	"github.com/docker/swarmkit/agent/exec"
 	"github.com/docker/swarmkit/agent/secrets"
 	"github.com/docker/swarmkit/api"
+	"github.com/docker/swarmkit/log"
 	"golang.org/x/net/context"
 )
 
@@ -41,12 +42,41 @@ func (e *executor) Describe(ctx context.Context) (*api.NodeDescription, error) {
 		}
 	}
 
+	// add v1 plugins to 'plugins'
 	addPlugins("Volume", info.Plugins.Volume)
 	// Add builtin driver "overlay" (the only builtin multi-host driver) to
 	// the plugin list by default.
 	addPlugins("Network", append([]string{"overlay"}, info.Plugins.Network...))
 	addPlugins("Authorization", info.Plugins.Authorization)
 
+	// retrieve v2 plugins
+	v2plugins, err := e.client.PluginList(ctx)
+	if err != nil {
+		log.L.WithError(err).Warning("PluginList operation failed")
+	} else {
+		// add v2 plugins to 'plugins'
+		for _, plgn := range v2plugins {
+			for _, typ := range plgn.Config.Interface.Types {
+				if typ.Prefix == "docker" && plgn.Enabled {
+					plgnTyp := typ.Capability
+					if typ.Capability == "volumedriver" {
+						plgnTyp = "Volume"
+					} else if typ.Capability == "networkdriver" {
+						plgnTyp = "Network"
+					}
+					plgnName := plgn.Name
+					if plgn.Tag != "" {
+						plgnName += ":" + plgn.Tag
+					}
+					plugins[api.PluginDescription{
+						Type: plgnTyp,
+						Name: plgnName,
+					}] = struct{}{}
+				}
+			}
+		}
+	}
+
 	pluginFields := make([]api.PluginDescription, 0, len(plugins))
 	for k := range plugins {
 		pluginFields = append(pluginFields, k)

manager/scheduler/filter.go

@@ -2,6 +2,7 @@ package scheduler
 
 import (
 	"fmt"
+	"strings"
 
 	"github.com/docker/swarmkit/api"
 	"github.com/docker/swarmkit/manager/constraint"
@@ -93,19 +94,25 @@ type PluginFilter struct {
 	t *api.Task
 }
 
+func referencesVolumePlugin(mount api.Mount) bool {
+	return mount.Type == api.MountTypeVolume &&
+		mount.VolumeOptions != nil &&
+		mount.VolumeOptions.DriverConfig != nil &&
+		mount.VolumeOptions.DriverConfig.Name != "" &&
+		mount.VolumeOptions.DriverConfig.Name != "local"
+
+}
+
 // SetTask returns true when the filter is enabled for a given task.
 func (f *PluginFilter) SetTask(t *api.Task) bool {
 	c := t.Spec.GetContainer()
 
 	var volumeTemplates bool
 	if c != nil {
 		for _, mount := range c.Mounts {
-			if mount.Type == api.MountTypeVolume &&
-				mount.VolumeOptions != nil &&
-				mount.VolumeOptions.DriverConfig != nil &&
-				mount.VolumeOptions.DriverConfig.Name != "" &&
-				mount.VolumeOptions.DriverConfig.Name != "local" {
+			if referencesVolumePlugin(mount) {
 				volumeTemplates = true
+				break
 			}
 		}
 	}
@@ -128,7 +135,7 @@ func (f *PluginFilter) Check(n *NodeInfo) bool {
 	container := f.t.Spec.GetContainer()
 	if container != nil {
 		for _, mount := range container.Mounts {
-			if mount.VolumeOptions != nil && mount.VolumeOptions.DriverConfig != nil {
+			if referencesVolumePlugin(mount) {
 				if !f.pluginExistsOnNode("Volume", mount.VolumeOptions.DriverConfig.Name, nodePlugins) {
 					return false
 				}
@@ -138,16 +145,30 @@ func (f *PluginFilter) Check(n *NodeInfo) bool {
 
 	// Check if all network plugins required by task are installed on node
 	for _, tn := range f.t.Networks {
-		if !f.pluginExistsOnNode("Network", tn.Network.DriverState.Name, nodePlugins) {
-			return false
+		if tn.Network != nil && tn.Network.DriverState != nil && tn.Network.DriverState.Name != "" {
+			if !f.pluginExistsOnNode("Network", tn.Network.DriverState.Name, nodePlugins) {
+				return false
+			}
 		}
 	}
 	return true
 }
 
+// pluginExistsOnNode returns true if the (pluginName, pluginType) pair is present in nodePlugins
 func (f *PluginFilter) pluginExistsOnNode(pluginType string, pluginName string, nodePlugins []api.PluginDescription) bool {
 	for _, np := range nodePlugins {
-		if pluginType == np.Type && pluginName == np.Name {
+		if pluginType != np.Type {
+			continue
+		}
+		if pluginName == np.Name {
+			return true
+		}
+		// This does not use the reference package to avoid the
+		// overhead of parsing references as part of the scheduling
+		// loop. This is okay only because plugin names are a very
+		// strict subset of the reference grammar that is always
+		// name:tag.
+		if strings.HasPrefix(np.Name, pluginName) && np.Name[len(pluginName):] == ":latest" {
 			return true
 		}
 	}

manager/scheduler/pipeline.go

@@ -11,12 +11,7 @@ var (
 		// Always check for readiness first.
 		&ReadyFilter{},
 		&ResourceFilter{},
-
-		// TODO(stevvooe): Do not filter based on plugins since they are lazy
-		// loaded in the engine. We can add this back when we can schedule
-		// plugins in the future.
-		// &PluginFilter{},
-
+		&PluginFilter{},
 		&ConstraintFilter{},
 	}
 )

manager/scheduler/scheduler_test.go

@@ -1387,7 +1387,6 @@ func watchAssignment(t *testing.T, watch chan events.Event) *api.Task {
 }
 
 func TestSchedulerPluginConstraint(t *testing.T) {
-	t.Skip("plugin filtering disabled since plugins on the engine are lazy loaded")
 	ctx := context.Background()
 
 	// Node1: vol plugin1