Tiered selector (#671)

Signed-off-by: David Chung <[email protected]>

Author: David Chung

docs/instance/selector/tiered/group.yml

@@ -0,0 +1,70 @@
+#
+#  Tiered provisioning example
+#
+#  Start up infrakit:
+#
+#  export INFRAKIT_SELECTOR_TIERED_PLUGINS='spot/compute;ondemand/compute'
+#
+#  In the env variable above we set the ordering of the plugins...
+#  First, we try spot/compute then ondemand/compute.  We will use the 'simulator'
+#  kind to simulate the different types of instances by starting the plugin at
+#  specific socket addresses: one as ondemand and the other as spot.
+#  In the infrakit CLI you will see ondemand/compute and spot/compute as available
+#  commands.  These will allow you to directly query the instance plugins.
+#
+#  infrakit plugin start manager group vanilla simulator:ondemand simulator:spot selector/tiered --log 5 --log-stack
+#
+#  * Note on the use of env variable...  when we support multiple plugins in a single
+#  spec and automatic activation, you will create sections of spec/metadata/Options
+#  to actually set the options for the tiered selector.
+#
+#  Commit this file:
+#
+#  infrakit group controller commit -y docs/instance/selector/tiered/group.yml
+#
+#  After a while, you can verify that 3 are created in spot/compute and 2 in ondemand/compute
+#  infrakit spot/compute describe
+#  infrakit ondemand/compute describe
+#
+#
+#
+kind: group
+metadata:
+    name: workers
+properties:
+    Allocation:
+      Size: 5
+    Flavor:
+      Plugin: vanilla
+      Properties:
+        Init:
+          - apt-get install curl
+          - curl -sSL https://get.docker.com | sh
+        Tags:
+          project: infrakit
+          tier: app
+    Instance:
+
+      # here we reference the tiered selector at the socket location, selector.
+      Plugin: selector/tiered
+      Properties:
+
+        spot/compute:
+          # This is a special property that the simulator understands to limit
+          # the number of instances it can provision.  So for cluster size > 3
+          # we expect to see the ondemand/compute instances getting added.
+          Cap: 3
+          instanceType: small
+          bid: 0.02
+          labels:
+            project: infrakit
+            billing: spot
+
+        # Properties for the ondemand/compute instances.  The section after the
+        # name of the plugin (ondemand/compute) is the properties blob to be fed
+        # to the plugin.
+        ondemand/compute:
+          instanceType: small
+          labels:
+            project: infrakit
+            billing: ondemand

pkg/discovery/local/dir.go

@@ -136,7 +136,7 @@ func (r *dirPluginDiscovery) List() (map[string]*plugin.Endpoint, error) {
 				continue
 			}
 
-			log.Debug("Discovered plugin", "address", instance.Address, "V", logutil.V(500))
+			log.Debug("Discovered plugin", "address", instance.Address, "V", logutil.V(1000))
 			plugins[instance.Name] = instance
 		}
 	}

pkg/plugin/instance/selector/internal/base.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"sort"
 	"strings"
+	"sync"
 
 	"github.com/docker/infrakit/pkg/discovery"
 	logutil "github.com/docker/infrakit/pkg/log"
@@ -14,25 +15,55 @@ import (
 	"github.com/docker/infrakit/pkg/types"
 )
 
-var log = logutil.New("module", "plugin/instance/selector")
+var log = logutil.New("module", "plugin/instance/selector/base")
 
 // Base is the base implementation of an instance plugin
 type Base struct {
 	Plugins          func() discovery.Plugins
 	Choices          []selector.Choice
 	SelectFunc       func(instance.Spec, []selector.Choice, func(selector.Choice) instance.Plugin) (selector.Choice, error)
-	PluginClientFunc func(map[string]*plugin.Endpoint, plugin.Name) instance.Plugin
+	PluginClientFunc func(plugin.Name) (instance.Plugin, error)
 }
 
-func instancePlugin(plugins map[string]*plugin.Endpoint, name plugin.Name) instance.Plugin {
+var (
+	clients = map[string]instance.Plugin{}
+	lock    sync.RWMutex
+)
+
+func (b *Base) instancePlugin(name plugin.Name) (instance.Plugin, error) {
+
 	lookup, _ := name.GetLookupAndType()
+
+	lock.RLock()
+	if p, has := clients[lookup]; has {
+		lock.RUnlock()
+		return p, nil
+	}
+	lock.RUnlock()
+
+	lock.Lock()
+	defer lock.Unlock()
+
+	plugins, err := b.Plugins().List()
+	if err != nil {
+		return nil, err
+	}
+
 	if endpoint, has := plugins[lookup]; has {
 		if p, err := instance_rpc.NewClient(name, endpoint.Address); err == nil {
-			return p
+			return p, nil
 		}
 		log.Warn("not an instance plugin", "name", name, "endpoint", endpoint)
 	}
-	return nil
+	return nil, nil
+}
+
+// Init initializes the base by setting any unset properties with defaults
+func (b *Base) Init() *Base {
+	if b.PluginClientFunc == nil {
+		b.PluginClientFunc = b.instancePlugin
+	}
+	return b
 }
 
 func (b *Base) selectOne(spec instance.Spec) (match selector.Choice, p instance.Plugin, err error) {
@@ -60,21 +91,44 @@ func (b *Base) selectOne(spec instance.Spec) (match selector.Choice, p instance.
 	return
 }
 
-func (b *Base) visit(f func(selector.Choice, instance.Plugin) error) error {
-	if b.PluginClientFunc == nil {
-		b.PluginClientFunc = instancePlugin
-	}
+// VisitChoices visits all the choices linearly one by one.  If the work function returns
+// false or error, the visit stops.
+func (b *Base) VisitChoices(visit func(selector.Choice, instance.Plugin) (bool, error)) error {
 
-	plugins, err := b.Plugins().List()
-	if err != nil {
-		return err
+	for _, choice := range b.Choices {
+		instancePlugin, err := b.PluginClientFunc(choice.Name)
+		if err != nil {
+			return err
+		}
+
+		if instancePlugin == nil {
+			// TODO -- implement retry??
+			log.Warn("cannot contact plugin", "name", choice.Name)
+			continue
+
+		}
+		if continueRun, err := visit(choice, instancePlugin); err != nil {
+			return err
+		} else if !continueRun {
+			return nil
+		}
 	}
+	return nil
+}
 
+func (b *Base) visit(f func(selector.Choice, instance.Plugin) error) error {
 	for _, choice := range b.Choices {
-		instancePlugin := b.PluginClientFunc(plugins, choice.Name)
+		log.Debug("checking choice", "choice", choice)
+
+		instancePlugin, err := b.PluginClientFunc(choice.Name)
+		if err != nil {
+			return err
+		}
+
 		if instancePlugin == nil {
 			continue
 		}
+		log.Debug("found instance plugin", "name", choice.Name, "client", instancePlugin)
 		if err := f(choice, instancePlugin); err != nil {
 			return err
 		}
@@ -83,7 +137,6 @@ func (b *Base) visit(f func(selector.Choice, instance.Plugin) error) error {
 }
 
 func (b *Base) doAll(count int, work func(instance.Plugin) error) error {
-
 	errs := make(chan error, len(b.Choices))
 	success := make(chan interface{}, len(b.Choices))
 	err := b.visit(func(c selector.Choice, p instance.Plugin) error {
@@ -165,17 +218,23 @@ func (b *Base) Provision(spec instance.Spec) (*instance.ID, error) {
 
 // DescribeInstances returns descriptions of all instances matching all of the provided tags.
 func (b *Base) DescribeInstances(tags map[string]string, properties bool) ([]instance.Description, error) {
+
+	log.Debug("DescribeInstances", "tags", tags, "properties", properties)
+
 	// Loop through all the choices and aggregate all the instances
 
 	keys := []string{}
 	uniques := map[string]instance.Description{}
 
 	err := b.visit(func(c selector.Choice, p instance.Plugin) error {
 		instances, err := p.DescribeInstances(tags, properties)
+
+		log.Debug("describing instances", "choice", c, "instances", instances, "err", err)
 		if err != nil {
 			// It's important to fail at this point if we can't get an accurate list of instances
 			// across the zones. This way, other controllers won't be fooled into thinking that
 			// they need reconcile state by provisioning more instances.
+			log.Error("describing instances", "choice", c, "err", err)
 			return err
 		}
 		for _, instance := range instances {
@@ -193,6 +252,7 @@ func (b *Base) DescribeInstances(tags map[string]string, properties bool) ([]ins
 	for _, k := range keys {
 		result = append(result, uniques[k])
 	}
+
 	return result, nil
 }
 

pkg/plugin/instance/selector/internal/base_test.go

@@ -13,6 +13,13 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+func mustPlugin(p instance.Plugin, err error) instance.Plugin {
+	if err != nil {
+		panic(err)
+	}
+	return p
+}
+
 type testDiscovery map[string]*plugin.Endpoint
 
 func (td testDiscovery) List() (map[string]*plugin.Endpoint, error) {
@@ -59,21 +66,21 @@ func TestVisit(t *testing.T) {
 			}
 		},
 		Choices: options,
-		PluginClientFunc: func(m map[string]*plugin.Endpoint, n plugin.Name) instance.Plugin {
+		PluginClientFunc: func(n plugin.Name) (instance.Plugin, error) {
 			switch n {
 			case n1:
-				return p1
+				return p1, nil
 			case n2:
-				return p2
+				return p2, nil
 			}
-			return nil
+			return nil, nil
 		},
 	}
 
-	m, err := b.Plugins().List()
+	_, err := b.Plugins().List()
 	require.NoError(t, err)
-	require.Equal(t, p1, b.PluginClientFunc(m, options[0].Name))
-	require.Equal(t, p2, b.PluginClientFunc(m, options[1].Name))
+	require.Equal(t, p1, mustPlugin(b.PluginClientFunc(options[0].Name)))
+	require.Equal(t, p2, mustPlugin(b.PluginClientFunc(options[1].Name)))
 
 	// Check error handling
 	require.Error(t, b.visit(
@@ -128,14 +135,14 @@ func TestDoAll(t *testing.T) {
 			}
 		},
 		Choices: options,
-		PluginClientFunc: func(m map[string]*plugin.Endpoint, n plugin.Name) instance.Plugin {
+		PluginClientFunc: func(n plugin.Name) (instance.Plugin, error) {
 			switch n {
 			case n1:
-				return p1
+				return p1, nil
 			case n2:
-				return p2
+				return p2, nil
 			}
-			return nil
+			return nil, nil
 		},
 	}
 
@@ -186,14 +193,14 @@ func TestDescribeInstances(t *testing.T) {
 			}
 		},
 		Choices: options,
-		PluginClientFunc: func(m map[string]*plugin.Endpoint, n plugin.Name) instance.Plugin {
+		PluginClientFunc: func(n plugin.Name) (instance.Plugin, error) {
 			switch n {
 			case n1:
-				return p1
+				return p1, nil
 			case n2:
-				return p2
+				return p2, nil
 			}
-			return nil
+			return nil, nil
 		},
 	}
 
@@ -255,14 +262,14 @@ func TestValidate(t *testing.T) {
 			}
 		},
 		Choices: options,
-		PluginClientFunc: func(m map[string]*plugin.Endpoint, n plugin.Name) instance.Plugin {
+		PluginClientFunc: func(n plugin.Name) (instance.Plugin, error) {
 			switch n {
 			case n1:
-				return p1
+				return p1, nil
 			case n2:
-				return p2
+				return p2, nil
 			}
-			return nil
+			return nil, nil
 		},
 	}
 
@@ -315,14 +322,14 @@ func TestLabel(t *testing.T) {
 			}
 		},
 		Choices: options,
-		PluginClientFunc: func(m map[string]*plugin.Endpoint, n plugin.Name) instance.Plugin {
+		PluginClientFunc: func(n plugin.Name) (instance.Plugin, error) {
 			switch n {
 			case n1:
-				return p1
+				return p1, nil
 			case n2:
-				return p2
+				return p2, nil
 			}
-			return nil
+			return nil, nil
 		},
 	}
 
@@ -381,14 +388,14 @@ func TestDestroy(t *testing.T) {
 			}
 		},
 		Choices: options,
-		PluginClientFunc: func(m map[string]*plugin.Endpoint, n plugin.Name) instance.Plugin {
+		PluginClientFunc: func(n plugin.Name) (instance.Plugin, error) {
 			switch n {
 			case n1:
-				return p1
+				return p1, nil
 			case n2:
-				return p2
+				return p2, nil
 			}
-			return nil
+			return nil, nil
 		},
 	}
 
@@ -444,14 +451,14 @@ func TestProvision(t *testing.T) {
 			}
 		},
 		Choices: options,
-		PluginClientFunc: func(m map[string]*plugin.Endpoint, n plugin.Name) instance.Plugin {
+		PluginClientFunc: func(n plugin.Name) (instance.Plugin, error) {
 			switch n {
 			case n1:
-				return p1
+				return p1, nil
 			case n2:
-				return p2
+				return p2, nil
 			}
-			return nil
+			return nil, nil
 		},
 	}