Merge pull request deis#3718 from krancour/plane-placement

feat(platform): support placement options for each plane and router mesh

Makefile

@@ -27,6 +27,7 @@ dev-cluster: discovery-url
 	ssh-add ~/.vagrant.d/insecure_private_key
 	deisctl config platform set sshPrivateKey=$(HOME)/.vagrant.d/insecure_private_key
 	deisctl config platform set domain=local3.deisapp.com
+	deisctl config platform set enablePlacementOptions=true
 	deisctl install platform
 
 discovery-url:

contrib/coreos/user-data.example

@@ -7,6 +7,7 @@ coreos:
     public-ip: $private_ipv4
     # allow etcd to slow down at times
     etcd_request_timeout: 3.0
+    metadata: controlPlane=true,dataPlane=true,routerMesh=true
   units:
   - name: etcd.service
     command: start

controller/scheduler/fleet.py

@@ -150,10 +150,10 @@ def _create_container(self, name, image, command, unit, **kwargs):
             f['value'] = f['value'].format(**l)
         # prepare tags only if one was provided
         tags = kwargs.get('tags', {})
-        if tags:
-            tagset = ' '.join(['"{}={}"'.format(k, v) for k, v in tags.viewitems()])
+        tagset = ' '.join(['"{}={}"'.format(k, v) for k, v in tags.viewitems()])
+        if settings.ENABLE_PLACEMENT_OPTIONS in ['true', 'True', 'TRUE', '1']:
             unit.append({"section": "X-Fleet", "name": "MachineMetadata",
-                         "value": tagset})
+                         "value": tagset + ' "dataPlane=true"'})
         # post unit to fleet
         self._put_unit(name, {"desiredState": "loaded", "options": unit})
 

controller/templates/confd_settings.py

@@ -24,6 +24,8 @@
 # platform domain must be provided
 DEIS_DOMAIN = '{{ getv "/deis/platform/domain" }}'
 
+ENABLE_PLACEMENT_OPTIONS = """{{ if exists "/deis/platform/enablePlacementOptions" }}{{ getv "/deis/platform/enablePlacementOptions" }}{{ else }}false{{end}}"""
+
 # use the private registry module
 REGISTRY_MODULE = 'registry.private'
 REGISTRY_URL = '{{ getv "/deis/registry/protocol" }}://{{ getv "/deis/registry/host" }}:{{ getv "/deis/registry/port" }}'  # noqa

deisctl/backend/fleet/create.go

@@ -3,6 +3,7 @@ package fleet
 import (
 	"fmt"
 	"io"
+	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -96,7 +97,15 @@ func (c *FleetClient) createServiceUnit(component string, num int) (name string,
 	if err != nil {
 		return "", nil, err
 	}
-	uf, err = NewUnit(component, c.templatePaths)
+	decorateStr, err := c.configBackend.GetWithDefault("/deis/platform/enablePlacementOptions", "false")
+	if err != nil {
+		return "", nil, err
+	}
+	decorate, err := strconv.ParseBool(decorateStr)
+	if err != nil {
+		return "", nil, err
+	}
+	uf, err = NewUnit(component, c.templatePaths, decorate)
 	if err != nil {
 		return
 	}

deisctl/backend/fleet/create_test.go

@@ -6,6 +6,9 @@ import (
 	"sync"
 	"testing"
 
+	"github.com/deis/deis/deisctl/config/model"
+	"github.com/deis/deis/deisctl/test/mock"
+
 	"github.com/coreos/fleet/schema"
 )
 
@@ -30,7 +33,9 @@ func TestCreate(t *testing.T) {
 	testFleetClient := stubFleetClient{testUnits: []*schema.Unit{}, unitsMutex: &sync.Mutex{},
 		unitStatesMutex: &sync.Mutex{}}
 
-	c := &FleetClient{templatePaths: []string{name}, Fleet: &testFleetClient}
+	testConfigBackend := mock.ConfigBackend{Expected: []*model.ConfigNode{{Key: "/deis/platform/enablePlacementOptions", Value: "true"}}}
+
+	c := &FleetClient{templatePaths: []string{name}, Fleet: &testFleetClient, configBackend: testConfigBackend}
 
 	var errOutput string
 	var wg sync.WaitGroup

deisctl/backend/fleet/fleet.go

@@ -6,13 +6,16 @@ import (
 	"path"
 	"text/tabwriter"
 
+	"github.com/deis/deis/deisctl/config"
+
 	"github.com/coreos/fleet/client"
 	"github.com/coreos/fleet/machine"
 )
 
 // FleetClient used to wrap Fleet API calls
 type FleetClient struct {
-	Fleet client.API
+	Fleet         client.API
+	configBackend config.Backend
 
 	// used to cache MachineStates
 	machineStates map[string]*machine.MachineState
@@ -25,7 +28,7 @@ type FleetClient struct {
 
 // NewClient returns a client used to communicate with Fleet
 // using the Registry API
-func NewClient() (*FleetClient, error) {
+func NewClient(cb config.Backend) (*FleetClient, error) {
 	client, err := getRegistryClient()
 	if err != nil {
 		return nil, err
@@ -41,6 +44,6 @@ func NewClient() (*FleetClient, error) {
 	out := new(tabwriter.Writer)
 	out.Init(os.Stdout, 0, 8, 1, '\t', 0)
 
-	return &FleetClient{Fleet: client, templatePaths: templatePaths, runner: sshCommandRunner{},
+	return &FleetClient{Fleet: client, configBackend: cb, templatePaths: templatePaths, runner: sshCommandRunner{},
 		out: out, errWriter: os.Stderr}, nil
 }

deisctl/backend/fleet/fleet_test.go

@@ -6,6 +6,9 @@ import (
 	"sync"
 	"testing"
 
+	"github.com/deis/deis/deisctl/config/model"
+	"github.com/deis/deis/deisctl/test/mock"
+
 	"github.com/coreos/fleet/machine"
 	"github.com/coreos/fleet/schema"
 )
@@ -176,8 +179,10 @@ func TestNewClient(t *testing.T) {
 	// set required flags
 	Flags.Endpoint = "http://127.0.0.1:4001"
 
+	testConfigBackend := mock.ConfigBackend{Expected: []*model.ConfigNode{}}
+
 	// instantiate client
-	_, err := NewClient()
+	_, err := NewClient(testConfigBackend)
 	if err != nil {
 		t.Fatal(err)
 	}

deisctl/backend/fleet/scale_test.go

@@ -7,6 +7,9 @@ import (
 	"sync"
 	"testing"
 
+	"github.com/deis/deis/deisctl/config/model"
+	"github.com/deis/deis/deisctl/test/mock"
+
 	"github.com/coreos/fleet/schema"
 )
 
@@ -30,7 +33,9 @@ func TestScaleUp(t *testing.T) {
 
 	testFleetClient := stubFleetClient{testUnits: testUnits, testUnitStates: []*schema.UnitState{}, unitsMutex: &sync.Mutex{}, unitStatesMutex: &sync.Mutex{}}
 
-	c := &FleetClient{templatePaths: []string{name}, Fleet: &testFleetClient}
+	testConfigBackend := mock.ConfigBackend{Expected: []*model.ConfigNode{{Key: "/deis/platform/enablePlacementOptions", Value: "true"}}}
+
+	c := &FleetClient{templatePaths: []string{name}, Fleet: &testFleetClient, configBackend: testConfigBackend}
 
 	var errOutput string
 	var wg sync.WaitGroup

deisctl/backend/fleet/unit.go

@@ -11,6 +11,21 @@ import (
 	"github.com/coreos/fleet/unit"
 )
 
+// path hierarchy for finding systemd service templates
+var templatePaths = []string{
+	os.Getenv("DEISCTL_UNITS"),
+	path.Join(os.Getenv("HOME"), ".deis", "units"),
+	"/var/lib/deis/units",
+}
+
+// and the same for systemd service "decorators" for optionally isolating
+// control plane, data plane, and router mesh
+var decoratorPaths = []string{
+	path.Join(os.Getenv("DEISCTL_UNITS"), "decorators"),
+	path.Join(os.Getenv("HOME"), ".deis", "units", "decorators"),
+	"/var/lib/deis/units/decorators",
+}
+
 // Units returns a list of units filtered by target
 func (c *FleetClient) Units(target string) (units []string, err error) {
 	allUnits, err := c.Fleet.Units()
@@ -56,14 +71,19 @@ func (c *FleetClient) lastUnit(component string) (num int, err error) {
 
 // NewUnit takes a component type and returns a Fleet unit
 // that includes the relevant systemd service template
-func NewUnit(component string, templatePaths []string) (uf *unit.UnitFile, err error) {
+func NewUnit(component string, templatePaths []string, decorate bool) (uf *unit.UnitFile, err error) {
 	template, err := readTemplate(component, templatePaths)
 	if err != nil {
 		return
 	}
-	uf, err = unit.NewUnitFile(string(template))
-	if err != nil {
-		return
+	if decorate {
+		decorator, err := readDecorator(component)
+		if err != nil {
+			return nil, err
+		}
+		uf, err = unit.NewUnitFile(string(template) + "\n" + string(decorator))
+	} else {
+		uf, err = unit.NewUnitFile(string(template))
 	}
 	return
 }
@@ -104,3 +124,26 @@ func readTemplate(component string, templatePaths []string) (out []byte, err err
 	}
 	return
 }
+
+// readDecorator returns the contents of a file containing a snippet that can
+// optionally be grafted on to the end of a corresponding systemd unit to
+// achieve isolation of the control plane, data plane, and router mesh
+func readDecorator(component string) (out []byte, err error) {
+	decoratorName := "deis-" + component + ".service.decorator"
+	var decoratorFile string
+
+	// look in $DEISCTL_UNITS env var, then the local and global root paths
+	for _, p := range decoratorPaths {
+		filename := path.Join(p, decoratorName)
+		if _, err := os.Stat(filename); err == nil {
+			decoratorFile = filename
+			break
+		}
+	}
+
+	if decoratorFile == "" {
+		return
+	}
+	out, err = ioutil.ReadFile(decoratorFile)
+	return
+}

deisctl/backend/fleet/unit_test.go

@@ -119,7 +119,7 @@ Description=deis-controller`
 
 	ioutil.WriteFile(path.Join(name, unit+".service"), []byte(unitFile), 777)
 
-	uf, err := NewUnit(unit[5:], []string{name})
+	uf, err := NewUnit(unit[5:], []string{name}, false)
 
 	if err != nil {
 		t.Fatal(err)

deisctl/client/client.go

@@ -46,13 +46,18 @@ type Client struct {
 func NewClient(requestedBackend string) (*Client, error) {
 	var backend backend.Backend
 
+	cb, err := etcd.NewConfigBackend()
+	if err != nil {
+		return nil, err
+	}
+
 	if requestedBackend == "" {
 		requestedBackend = "fleet"
 	}
 
 	switch requestedBackend {
 	case "fleet":
-		b, err := fleet.NewClient()
+		b, err := fleet.NewClient(cb)
 		if err != nil {
 			return nil, err
 		}
@@ -61,11 +66,6 @@ func NewClient(requestedBackend string) (*Client, error) {
 		return nil, errors.New("invalid backend")
 	}
 
-	cb, err := etcd.NewConfigBackend()
-	if err != nil {
-		return nil, err
-	}
-
 	return &Client{Backend: backend, configBackend: cb}, nil
 }
 

deisctl/cmd/cmd.go

@@ -2,11 +2,8 @@ package cmd
 
 import (
 	"bytes"
-	"errors"
 	"fmt"
 	"io"
-	"io/ioutil"
-	"net/http"
 	"os"
 	"path/filepath"
 	"regexp"
@@ -18,6 +15,7 @@ import (
 	"github.com/deis/deis/deisctl/config"
 	"github.com/deis/deis/deisctl/units"
 	"github.com/deis/deis/deisctl/utils"
+	"github.com/deis/deis/deisctl/utils/net"
 )
 
 const (
@@ -452,32 +450,32 @@ func Config(target string, action string, key []string, cb config.Backend) error
 // RefreshUnits overwrites local unit files with those requested.
 // Downloading from the Deis project GitHub URL by tag or SHA is the only mechanism
 // currently supported.
-func RefreshUnits(dir, tag, url string) error {
-	dir = utils.ResolvePath(dir)
+func RefreshUnits(unitDir, tag, rootURL string) error {
+	unitDir = utils.ResolvePath(unitDir)
+	decoratorDir := filepath.Join(unitDir, "decorators")
 	// create the target dir if necessary
-	if err := os.MkdirAll(dir, 0755); err != nil {
+	if err := os.MkdirAll(decoratorDir, 0755); err != nil {
 		return err
 	}
 	// download and save the unit files to the specified path
 	for _, unit := range units.Names {
-		src := fmt.Sprintf(url, tag, unit)
-		dest := filepath.Join(dir, unit+".service")
-		res, err := http.Get(src)
-		if err != nil {
-			return err
-		}
-		if res.StatusCode != 200 {
-			return errors.New(res.Status)
-		}
-		defer res.Body.Close()
-		data, err := ioutil.ReadAll(res.Body)
-		if err != nil {
+		unitSrc := rootURL + tag + "/deisctl/units/" + unit + ".service"
+		unitDest := filepath.Join(unitDir, unit+".service")
+		if err := net.Download(unitSrc, unitDest); err != nil {
 			return err
 		}
-		if err = ioutil.WriteFile(dest, data, 0644); err != nil {
-			return err
+		fmt.Printf("Refreshed %s unit from %s\n", unit, tag)
+		decoratorSrc := rootURL + tag + "/deisctl/units/decorators/" + unit + ".service.decorator"
+		decoratorDest := filepath.Join(decoratorDir, unit+".service.decorator")
+		if err := net.Download(decoratorSrc, decoratorDest); err != nil {
+			if err.Error() == "404 Not Found" {
+				fmt.Printf("Decorator for %s not found in %s\n", unit, tag)
+			} else {
+				return err
+			}
+		} else {
+			fmt.Printf("Refreshed %s decorator from %s\n", unit, tag)
 		}
-		fmt.Printf("Refreshed %s from %s\n", unit, tag)
 	}
 	return nil
 }

deisctl/cmd/cmd_test.go

@@ -115,16 +115,20 @@ func TestRefreshUnits(t *testing.T) {
 	server := httptest.NewServer(handler)
 	defer server.Close()
 
-	err = RefreshUnits(name, "v1.7.2", server.URL+"/%s/%s.service")
+	err = RefreshUnits(name, "v1.7.2", server.URL+"/")
 
 	if err != nil {
 		t.Error(err)
 	}
 
 	files, err := ioutil.ReadDir(name)
 
-	if len(units.Names) != len(files) {
-		t.Error(fmt.Errorf("Expected %d units, Got %d", len(units.Names), len(files)))
+	// There will be a "decorators" subdirectory and that shouldn't be
+	// counted as a unit when making the upcoming assertion.
+	numFiles := len(files) - 1
+
+	if len(units.Names) != numFiles {
+		t.Error(fmt.Errorf("Expected %d units, Got %d", len(units.Names), numFiles))
 	}
 
 	for _, unit := range units.Names {
@@ -155,7 +159,7 @@ func TestRefreshUnitsError(t *testing.T) {
 	server := httptest.NewServer(handler)
 	defer server.Close()
 
-	err = RefreshUnits(name, "foo", server.URL+"/%s/%s.service")
+	err = RefreshUnits(name, "foo", server.URL+"/")
 	result := err.Error()
 	expected := "404 Not Found"
 

deisctl/config/backend.go

@@ -5,6 +5,7 @@ import "github.com/deis/deis/deisctl/config/model"
 // Backend is an interface for any sort of underlying key/value config store
 type Backend interface {
 	Get(string) (string, error)
+	GetWithDefault(string, string) (string, error)
 	Set(string, string) (string, error)
 	SetWithTTL(string, string, uint64) (string, error)
 	Delete(string) error