Merge pull request #1708 from bacongobbler/publisher

ref(scheduler): introduce deis/publisher

Author: Matthew Fisher

Makefile

@@ -4,8 +4,8 @@
 
 include includes.mk
 
-COMPONENTS=builder cache controller database logger registry router
-START_ORDER=logger database cache registry controller builder router
+COMPONENTS=builder cache controller database logger publisher registry router
+START_ORDER=publisher logger database cache registry controller builder router
 CLIENTS=client deisctl
 
 all: build run

controller/api/models.py

@@ -436,9 +436,6 @@ def _get_command(self):
 
     _command = property(_get_command)
 
-    def _command_announceable(self):
-        return self._command.lower() in ['start web', '']
-
     def clone(self, release):
         c = Container.objects.create(owner=self.owner,
                                      app=self.app,
@@ -459,7 +456,7 @@ def create(self):
                 name=job_id,
                 image=image,
                 command=self._command,
-                use_announcer=self._command_announceable(), **kwargs)
+                **kwargs)
         except Exception as e:
             err = '{} (create): {}'.format(job_id, e)
             log_event(self.app, err, logging.ERROR)
@@ -469,7 +466,7 @@ def create(self):
     def start(self):
         job_id = self._job_id
         try:
-            self._scheduler.start(job_id, self._command_announceable())
+            self._scheduler.start(job_id)
         except Exception as e:
             err = '{} (start): {}'.format(job_id, e)
             log_event(self.app, err, logging.WARNING)
@@ -479,7 +476,7 @@ def start(self):
     def stop(self):
         job_id = self._job_id
         try:
-            self._scheduler.stop(job_id, self._command_announceable())
+            self._scheduler.stop(job_id)
         except Exception as e:
             err = '{} (stop): {}'.format(job_id, e)
             log_event(self.app, err, logging.ERROR)
@@ -489,7 +486,7 @@ def stop(self):
     def destroy(self):
         job_id = self._job_id
         try:
-            self._scheduler.destroy(job_id, self._command_announceable())
+            self._scheduler.destroy(job_id)
         except Exception as e:
             err = '{} (destroy): {}'.format(job_id, e)
             log_event(self.app, err, logging.ERROR)

controller/scheduler/chaos.py

@@ -25,28 +25,28 @@ def tearDown(self):
 
     # job api
 
-    def create(self, name, image, command, use_announcer, **kwargs):
+    def create(self, name, image, command, **kwargs):
         if random.random() < CREATE_ERROR_RATE:
             raise RuntimeError
         return True
 
-    def start(self, name, use_announcer):
+    def start(self, name):
         """
         Start an idle job
         """
         if random.random() < START_ERROR_RATE:
             raise RuntimeError
         return True
 
-    def stop(self, name, use_announcer):
+    def stop(self, name):
         """
         Stop a running job
         """
         if random.random() < STOP_ERROR_RATE:
             raise RuntimeError
         return True
 
-    def destroy(self, name, use_announcer):
+    def destroy(self, name):
         """
         Destroy an existing job
         """

controller/scheduler/coreos.py

@@ -101,15 +101,12 @@ def _get_machines(self):
 
     # container api
 
-    def create(self, name, image, command='', template=None, use_announcer=True, **kwargs):
+    def create(self, name, image, command='', template=None, **kwargs):
         """Create a container"""
         self._create_container(name, image, command,
                                template or copy.deepcopy(CONTAINER_TEMPLATE), **kwargs)
         self._create_log(name, image, command, copy.deepcopy(LOG_TEMPLATE))
 
-        if use_announcer:
-            self._create_announcer(name, image, command, copy.deepcopy(ANNOUNCE_TEMPLATE))
-
     def _create_container(self, name, image, command, unit, **kwargs):
         l = locals().copy()
         l.update(re.match(MATCH, name).groupdict())
@@ -146,22 +143,10 @@ def _create_log(self, name, image, command, unit):
         # post unit to fleet
         self._put_unit(name+'-log', {"desiredState": "launched", "options": unit})
 
-    def _create_announcer(self, name, image, command, unit):
-        l = locals().copy()
-        l.update(re.match(MATCH, name).groupdict())
-        # construct unit from template
-        for f in unit:
-            f['value'] = f['value'].format(**l)
-        # post unit to fleet
-        self._put_unit(name+'-announce', {"desiredState": "launched", "options": unit})
-
-    def start(self, name, use_announcer=True):
+    def start(self, name):
         """Start a container"""
         self._wait_for_container(name)
 
-        if use_announcer:
-            self._wait_for_announcer(name)
-
     def _wait_for_container(self, name):
         # we bump to 20 minutes here to match the timeout on the router and in the app unit files
         for _ in range(1200):
@@ -177,24 +162,6 @@ def _wait_for_container(self, name):
         else:
             raise RuntimeError('container failed to start')
 
-    def _wait_for_announcer(self, name):
-        # wait a bit for the announcer to come up, otherwise we may have hit
-        # https://github.com/docker/docker/issues/8022
-        for _ in range(30):
-            states = self._get_state(name)
-            if states and len(states.get('states', [])) == 1:
-                state = states.get('states')[0]
-                subState = state.get('systemdSubState')
-                if subState == 'running':
-                    # wait for the router to be reconfigured
-                    time.sleep(10)
-                    break
-                elif subState == 'failed':
-                    raise RuntimeError('announcer failed to start')
-            time.sleep(1)
-        else:
-            raise RuntimeError('announcer timeout on start')
-
     def _wait_for_destroy(self, name):
         for _ in range(30):
             states = self._get_state(name)
@@ -204,15 +171,13 @@ def _wait_for_destroy(self, name):
         else:
             raise RuntimeError('timeout on container destroy')
 
-    def stop(self, name, use_announcer=True):
+    def stop(self, name):
         """Stop a container"""
         raise NotImplementedError
 
-    def destroy(self, name, use_announcer=True):
+    def destroy(self, name):
         """Destroy a container"""
         funcs = []
-        if use_announcer:
-            funcs.append(functools.partial(self._destroy_announcer, name))
         funcs.append(functools.partial(self._destroy_container, name))
         funcs.append(functools.partial(self._destroy_log, name))
         # call all destroy functions, ignoring any errors
@@ -226,9 +191,6 @@ def destroy(self, name, use_announcer=True):
     def _destroy_container(self, name):
         return self._delete_unit(name)
 
-    def _destroy_announcer(self, name):
-        return self._delete_unit(name+'-announce')
-
     def _destroy_log(self, name):
         return self._delete_unit(name+'-log')
 
@@ -350,18 +312,6 @@ def attach(self, name):
 ]
 
 
-ANNOUNCE_TEMPLATE = [
-    {"section": "Unit", "name": "Description", "value": "{name} announce"},
-    {"section": "Unit", "name": "BindsTo", "value": "{name}.service"},
-    {"section": "Service", "name": "EnvironmentFile", "value": "/etc/environment"},
-    {"section": "Service", "name": "ExecStartPre", "value": '''/bin/sh -c "until docker inspect -f '{{{{range $i, $e := .NetworkSettings.Ports }}}}{{{{$p := index $e 0}}}}{{{{$p.HostPort}}}}{{{{end}}}}' {name} >/dev/null 2>&1; do sleep 2; done; port=$(docker inspect -f '{{{{range $i, $e := .NetworkSettings.Ports }}}}{{{{$p := index $e 0}}}}{{{{$p.HostPort}}}}{{{{end}}}}' {name}); if [[ -z $port ]]; then echo We have no port...; exit 1; fi; echo Waiting for $port/tcp...; until netstat -lnt | grep :$port >/dev/null; do sleep 1; done"'''},  # noqa
-    {"section": "Service", "name": "ExecStart", "value": '''/bin/sh -c "port=$(docker inspect -f '{{{{range $i, $e := .NetworkSettings.Ports }}}}{{{{$p := index $e 0}}}}{{{{$p.HostPort}}}}{{{{end}}}}' {name}); echo Connected to $COREOS_PRIVATE_IPV4:$port/tcp, publishing to etcd...; while netstat -lnt | grep :$port >/dev/null; do etcdctl set /deis/services/{app}/{name} $COREOS_PRIVATE_IPV4:$port --ttl 60 >/dev/null; sleep 45; done"'''},  # noqa
-    {"section": "Service", "name": "ExecStop", "value": "/usr/bin/etcdctl rm --recursive /deis/services/{app}/{name}"},  # noqa
-    {"section": "Service", "name": "TimeoutStartSec", "value": "20m"},
-    {"section": "X-Fleet", "name": "MachineOf", "value": "{name}.service"},
-]
-
-
 RUN_TEMPLATE = [
     {"section": "Unit", "name": "Description", "value": "{name} admin command"},
     {"section": "Service", "name": "ExecStartPre", "value": '''/bin/sh -c "IMAGE=$(etcdctl get /deis/registry/host 2>&1):$(etcdctl get /deis/registry/port 2>&1)/{image}; docker pull $IMAGE"'''},  # noqa

controller/scheduler/mock.py

@@ -20,25 +20,25 @@ def tearDown(self):
 
     # container api
 
-    def create(self, name, image, command, use_announcer, **kwargs):
+    def create(self, name, image, command, **kwargs):
         """
         Create a new container
         """
         return
 
-    def start(self, name, use_announcer):
+    def start(self, name):
         """
         Start a container
         """
         return
 
-    def stop(self, name, use_announcer):
+    def stop(self, name):
         """
         Stop a container
         """
         return
 
-    def destroy(self, name, use_announcer):
+    def destroy(self, name):
         """
         Destroy a container
         """

deisctl/backend/fleet/create.go

@@ -13,6 +13,7 @@ import (
 // and blocks until the unit is loaded
 func (c *FleetClient) Create(targets []string) error {
 	units := make([]*schema.Unit, len(targets))
+	desiredState := string(job.JobStateLoaded)
 	for i, target := range targets {
 		unitName, unitFile, err := c.createUnitFile(target)
 		if err != nil {
@@ -31,13 +32,11 @@ func (c *FleetClient) Create(targets []string) error {
 				return fmt.Errorf("failed creating job %s: %v", unit.Name, err)
 			}
 		}
-		desiredState := string(job.JobStateLoaded)
 		if err := c.Fleet.SetUnitTargetState(unit.Name, desiredState); err != nil {
 			return err
 		}
 	}
 	for _, unit := range units {
-		desiredState := string(job.JobStateLoaded)
 		outchan, errchan := waitForUnitStates([]string{unit.Name}, desiredState)
 		if err := printUnitState(unit.Name, outchan, errchan); err != nil {
 			return err

deisctl/backend/fleet/destroy.go

@@ -11,18 +11,14 @@ import (
 func (c *FleetClient) Destroy(targets []string) error {
 	for _, target := range targets {
 		// check if the unit exists
-		units, err := c.Units(target)
+		_, err := c.Units(target)
 		if err != nil {
 			return err
 		}
 		component, num, err := splitTarget(target)
 		if err != nil {
 			return err
 		}
-		// if no number is specified, destroy ALL THE UNITS!
-		if num == 0 {
-			num = len(units)
-		}
 		if strings.HasSuffix(component, "-data") {
 			err = c.destroyDataUnit(component)
 		} else {
@@ -35,7 +31,7 @@ func (c *FleetClient) Destroy(targets []string) error {
 	return nil
 }
 
-func (c *FleetClient) destroyServiceUnit(component string, num int) (err error) {
+func (c *FleetClient) destroyServiceUnit(component string, num int) error {
 	name, err := formatUnitName(component, num)
 	if err != nil {
 		return err
@@ -46,34 +42,30 @@ func (c *FleetClient) destroyServiceUnit(component string, num int) (err error)
 		return err
 	}
 	outchan, errchan := waitForUnitStates([]string{name}, desiredState)
-	err = printUnitState(name, outchan, errchan)
-	if err != nil {
+	if err := printUnitState(name, outchan, errchan); err != nil {
 		return err
 	}
 	if err = c.Fleet.DestroyUnit(name); err != nil {
 		return fmt.Errorf("failed destroying job %s: %v", name, err)
 	}
-	return err
+	return nil
 }
 
-func (c *FleetClient) destroyDataUnit(component string) (err error) {
+func (c *FleetClient) destroyDataUnit(component string) error {
 	name, err := formatUnitName(component, 0)
+	desiredState := string(job.JobStateInactive)
 	if err != nil {
 		return err
 	}
-	desiredState := string(job.JobStateInactive)
-	err = c.Fleet.SetUnitTargetState(name, desiredState)
-	if err != nil {
+	if err := c.Fleet.SetUnitTargetState(name, desiredState); err != nil {
 		return err
 	}
 	outchan, errchan := waitForUnitStates([]string{name}, desiredState)
-	err = printUnitState(name, outchan, errchan)
-	if err != nil {
+	if err := printUnitState(name, outchan, errchan); err != nil {
 		return err
 	}
-	if err = c.Fleet.DestroyUnit(name); err != nil {
+	if err := c.Fleet.DestroyUnit(name); err != nil {
 		return fmt.Errorf("failed destroying job %s: %v", name, err)
 	}
-	return err
-
+	return nil
 }

deisctl/backend/fleet/state.go

@@ -126,6 +126,11 @@ func assertUnitState(name string, desiredState string, outchan chan *schema.Unit
 	// send unit across the output channel
 	outchan <- u
 
+	// HACK: global units have no soul
+	if u.CurrentState == "" {
+		return true
+	}
+
 	if u.DesiredState == u.CurrentState {
 		return true
 	}
@@ -182,6 +187,10 @@ func printUnitState(name string, outchan chan *schema.Unit, errchan chan error)
 			if u.Name != name {
 				continue
 			}
+			// HACK: global units have no soul
+			if u.CurrentState == "" {
+				continue
+			}
 			// otherwise print output
 			if u.CurrentState != u.DesiredState {
 				fmt.Printf("\033[0;33m%v:\033[0m %v (pending)                       \r",

deisctl/backend/fleet/unit.go

@@ -5,7 +5,6 @@ import (
 	"io/ioutil"
 	"os"
 	"path"
-	"regexp"
 	"strconv"
 	"strings"
 
@@ -25,18 +24,18 @@ func (c *FleetClient) Units(target string) (units []string, err error) {
 	if err != nil {
 		return
 	}
-	var r *regexp.Regexp
 	if strings.HasSuffix(target, "-data") {
-		r = regexp.MustCompile(`deis\-(` + target + `)\.service`)
-	} else if strings.Contains(target, "@") {
-		r = regexp.MustCompile(`deis\-(` + target + `)\.service`)
+		for _, u := range allUnits {
+			if strings.Contains(u.Name, target) {
+				units = []string{u.Name}
+				return
+			}
+		}
 	} else {
-		r = regexp.MustCompile(`deis\-(` + target + `)@([\d]+)\.service`)
-	}
-	for _, u := range allUnits {
-		match := r.MatchString(u.Name)
-		if match {
-			units = append(units, u.Name)
+		for _, u := range allUnits {
+			if strings.Contains(u.Name, target) && !strings.HasSuffix(u.Name, "-data.service") {
+				units = append(units, u.Name)
+			}
 		}
 	}
 	if len(units) == 0 {