Max resource lifetime

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

Author: David Chung

cmd/infrakit/main.go

@@ -32,6 +32,7 @@ import (
 	_ "github.com/docker/infrakit/cmd/infrakit/remote"
 	_ "github.com/docker/infrakit/cmd/infrakit/template"
 	_ "github.com/docker/infrakit/cmd/infrakit/util"
+	_ "github.com/docker/infrakit/cmd/infrakit/x"
 )
 
 func init() {

cmd/infrakit/util/util.go

@@ -21,7 +21,11 @@ func Command(plugins func() discovery.Plugins) *cobra.Command {
 		Short: "Utilties",
 	}
 
-	util.AddCommand(muxCommand(plugins), fileServerCommand(plugins), trackCommand(plugins))
+	util.AddCommand(
+		muxCommand(plugins),
+		fileServerCommand(plugins),
+		trackCommand(plugins),
+	)
 
 	return util
 }

cmd/infrakit/x/maxlife.go

@@ -0,0 +1,204 @@
+package x
+
+import (
+	"os"
+	"strings"
+	"time"
+
+	"github.com/docker/infrakit/pkg/discovery"
+	"github.com/docker/infrakit/pkg/plugin"
+	instance_rpc "github.com/docker/infrakit/pkg/rpc/instance"
+	"github.com/docker/infrakit/pkg/spi/instance"
+	"github.com/docker/infrakit/pkg/types"
+	"github.com/spf13/cobra"
+)
+
+func maxlifeCommand(plugins func() discovery.Plugins) *cobra.Command {
+
+	cmd := &cobra.Command{
+		Use:   "maxlife <instance plugin name>...",
+		Short: "Sets max life on the given instances",
+	}
+
+	//name := cmd.Flags().String("name", "", "Name to use as name of this plugin")
+	poll := cmd.Flags().DurationP("poll", "i", 10*time.Second, "Polling interval")
+	maxlife := cmd.Flags().DurationP("maxlife", "m", 10*time.Minute, "Max lifetime of the resource")
+	flagTags := cmd.Flags().StringSliceP("tag", "t", []string{}, "Tags to filter instance by")
+
+	cmd.RunE = func(c *cobra.Command, args []string) error {
+
+		if len(args) == 0 {
+			cmd.Usage()
+			os.Exit(-1)
+		}
+
+		tags := toTags(*flagTags)
+
+		// Now we have a list of instance plugins to maxlife
+		plugins, err := getInstancePlugins(plugins, args)
+		if err != nil {
+			return err
+		}
+
+		// For each we start a goroutine to poll and kill instances
+		stops := []chan struct{}{}
+
+		retry := false
+
+	loop:
+		for {
+			for name, plugin := range plugins {
+
+				stop := make(chan struct{})
+				stops = append(stops, stop)
+
+				described, err := plugin.DescribeInstances(tags, false)
+				if err != nil {
+					log.Warn("cannot get initial count", "name", name, "err", err)
+					retry = true
+				}
+
+				go ensureMaxlife(name, plugin, stop, *poll, *maxlife, tags, len(described))
+			}
+
+			if !retry {
+				break loop
+			}
+
+			// Wait a little bit before trying again -- use the same poll interval
+			<-time.After(*poll)
+		}
+
+		return nil
+	}
+
+	return cmd
+}
+
+func age(instance instance.Description, now time.Time) (age time.Duration) {
+	link := types.NewLinkFromMap(instance.Tags)
+	if link.Valid() {
+		age = now.Sub(link.Created())
+	}
+	return
+}
+
+func maxAge(instances []instance.Description, now time.Time) instance.Description {
+	// check to see if the tags of the instances have links.  Links have a creation date and
+	// we can use it to compute the age
+	var max time.Duration
+	var found = 0
+	for i, instance := range instances {
+		age := age(instance, now)
+		if age > max {
+			max = age
+			found = i
+		}
+	}
+	return instances[found]
+}
+
+func ensureMaxlife(name string, plugin instance.Plugin, stop chan struct{}, poll, maxlife time.Duration,
+	tags map[string]string, initialCount int) {
+
+	// Count is used to track the steady state...  we don't want to keep killing instances
+	// if the counts are steadily decreasing.  The idea here is that once we terminate a resource
+	// another one will be resurrected so we will be back to steady state.
+	// Of course it's possible that the size of the cluster actually is decreased.  So we'd
+	// wait for a few samples to get to steady state before we terminate another instance.
+	// Currently we assume damping == 1 or 1 successive samples of delta >= 0 is sufficient to terminate
+	// another instance.
+
+	last := initialCount
+	tick := time.Tick(poll)
+loop:
+	for {
+
+		select {
+
+		case now := <-tick:
+
+			described, err := plugin.DescribeInstances(tags, false)
+			if err != nil {
+				// Transient error?
+				log.Warn("error describing instances", "name", name, "err", err)
+				continue
+			}
+
+			// TODO -- we should compute the 2nd derivative wrt time to make sure we
+			// are truly in a steady state...
+
+			current := len(described)
+			delta := current - last
+			last = current
+
+			if current < 2 {
+				log.Info("there are less than 2 instances.  No actions.", "name", name)
+				continue
+			}
+
+			if delta < 0 {
+				// Don't do anything if there are fewer instances at this iteration
+				// than the last.  We want to wait until steady state
+				log.Info("fewer instances in this round.  No actions taken", "name", name)
+				continue
+			}
+
+			// Just pick a single oldest instance per polling cycle.  This is so
+			// that we don't end up destroying the cluster by taking down too many instances
+			// all at once.
+			oldest := maxAge(described, now)
+
+			// check to make sure the age is over the maxlife
+			if age(oldest, now) > maxlife {
+				// terminate it and hope the group controller restores with a new intance
+				err = plugin.Destroy(oldest.ID)
+				if err != nil {
+					log.Warn("cannot destroy instance", "name", name, "id", oldest.ID, "err", err)
+					continue
+				}
+			}
+
+		case <-stop:
+			log.Info("stop requested", "name", name)
+			break loop
+		}
+	}
+
+	log.Info("maxlife stopped", "name", name)
+	return
+}
+
+func toTags(slice []string) map[string]string {
+	tags := map[string]string{}
+
+	for _, tag := range slice {
+		kv := strings.SplitN(tag, "=", 2)
+		if len(kv) != 2 {
+			log.Warn("bad format tag", "input", tag)
+			continue
+		}
+		key := strings.TrimSpace(kv[0])
+		val := strings.TrimSpace(kv[1])
+		if key != "" && val != "" {
+			tags[key] = val
+		}
+	}
+	return tags
+}
+
+func getInstancePlugins(plugins func() discovery.Plugins, names []string) (map[string]instance.Plugin, error) {
+	targets := map[string]instance.Plugin{}
+	for _, target := range names {
+		endpoint, err := plugins().Find(plugin.Name(target))
+		if err != nil {
+			return nil, err
+		}
+		if p, err := instance_rpc.NewClient(plugin.Name(target), endpoint.Address); err == nil {
+			targets[target] = p
+		} else {
+			return nil, err
+		}
+	}
+	return targets, nil
+}

cmd/infrakit/x/x.go

@@ -0,0 +1,29 @@
+package x
+
+import (
+	"github.com/docker/infrakit/cmd/infrakit/base"
+	"github.com/docker/infrakit/pkg/discovery"
+	logutil "github.com/docker/infrakit/pkg/log"
+	"github.com/spf13/cobra"
+)
+
+var log = logutil.New("module", "cli/x")
+
+func init() {
+	base.Register(Command)
+}
+
+// Command is the head of this module
+func Command(plugins func() discovery.Plugins) *cobra.Command {
+
+	experimental := &cobra.Command{
+		Use:   "x",
+		Short: "Experimental features",
+	}
+
+	experimental.AddCommand(
+		maxlifeCommand(plugins),
+	)
+
+	return experimental
+}

pkg/types/link.go

@@ -20,25 +20,42 @@ func init() {
 type Link struct {
 	value   string
 	context string
+	created time.Time
 }
 
 // NewLink creates a link
 func NewLink() *Link {
 	return &Link{
-		value: randomAlphaNumericString(16),
+		value:   randomAlphaNumericString(16),
+		created: time.Now(),
 	}
 }
 
-// NewLinkFromMap constructs a link from data in the map
+// Link related labels
+const (
+	LinkLabel        = "infrakit-link"
+	LinkContextLabel = "infrakit-link-context"
+	LinkCreatedLabel = "infrakit-link-created"
+)
+
+// NewLinkFromMap constructs a link from data in the map. The link will have missing data
+// if the input does not contain the attribute labels.
 func NewLinkFromMap(m map[string]string) *Link {
 	l := &Link{}
-	if v, has := m["infrakit-link"]; has {
+	if v, has := m[LinkLabel]; has {
 		l.value = v
 	}
 
-	if v, has := m["infrakit-link-context"]; has {
+	if v, has := m[LinkContextLabel]; has {
 		l.context = v
 	}
+
+	if v, has := m[LinkCreatedLabel]; has {
+		t, err := time.Parse(time.RFC3339, v)
+		if err == nil {
+			l.created = t
+		}
+	}
 	return l
 }
 
@@ -52,9 +69,14 @@ func (l Link) Value() string {
 	return l.value
 }
 
+// Created returns the creation time of the link
+func (l Link) Created() time.Time {
+	return l.created
+}
+
 // Label returns the label to look for the link
 func (l Link) Label() string {
-	return "infrakit-link"
+	return LinkLabel
 }
 
 // Context returns the context of the link
@@ -80,8 +102,9 @@ func (l *Link) KVPairs() []string {
 // Map returns a representation that is easily converted to JSON or YAML
 func (l *Link) Map() map[string]string {
 	return map[string]string{
-		"infrakit-link":         l.value,
-		"infrakit-link-context": l.context,
+		LinkLabel:        l.value,
+		LinkContextLabel: l.context,
+		LinkCreatedLabel: l.created.Format(time.RFC3339),
 	}
 }
 
@@ -94,15 +117,15 @@ func (l *Link) WriteMap(target map[string]string) {
 
 // InMap returns true if the link is contained in the map
 func (l *Link) InMap(m map[string]string) bool {
-	c, has := m["infrakit-link-context"]
+	c, has := m[LinkContextLabel]
 	if !has {
 		return false
 	}
 	if c != l.context {
 		return false
 	}
 
-	v, has := m["infrakit-link"]
+	v, has := m[LinkLabel]
 	if !has {
 		return false
 	}