Merge pull request #1279 from deis/1275_skip_announcers

fix(scheduler/coreos): skip announcer containers for start, stop, destroy

Author: carmstrong

controller/api/models.py

@@ -304,19 +304,25 @@ def _get_command(self):
 
     _command = property(_get_command)
 
+    def _command_announceable(self):
+        return self._command.lower() in ['start web', '']
+
     @close_db_connections
     @transition(field=state, source=INITIALIZED, target=CREATED)
     def create(self):
         image = self.release.image
         c_type = self.type
-        self._scheduler.create(self._job_id, image, self._command.format(**locals()))
+        self._scheduler.create(name=self._job_id,
+                               image=image,
+                               command=self._command.format(**locals()),
+                               use_announcer=self._command_announceable())
 
     @close_db_connections
     @transition(field=state,
                 source=[CREATED, UP, DOWN],
                 target=UP, crashed=DOWN)
     def start(self):
-        self._scheduler.start(self._job_id)
+        self._scheduler.start(self._job_id, self._command_announceable())
 
     @close_db_connections
     @transition(field=state,
@@ -332,23 +338,26 @@ def deploy(self, release):
         new_job_id = self._job_id
         image = self.release.image
         c_type = self.type
-        self._scheduler.create(new_job_id, image, self._command.format(**locals()))
-        self._scheduler.start(new_job_id)
+        self._scheduler.create(name=new_job_id,
+                               image=image,
+                               command=self._command.format(**locals()),
+                               use_announcer=self._command_announceable())
+        self._scheduler.start(new_job_id, self._command_announceable())
         # destroy old container
-        self._scheduler.destroy(old_job_id)
+        self._scheduler.destroy(old_job_id, self._command_announceable())
 
     @close_db_connections
     @transition(field=state, source=UP, target=DOWN)
     def stop(self):
-        self._scheduler.stop(self._job_id)
+        self._scheduler.stop(self._job_id, self._command_announceable())
 
     @close_db_connections
     @transition(field=state,
                 source=[INITIALIZED, CREATED, UP, DOWN],
                 target=DESTROYED)
     def destroy(self):
         # TODO: add check for active connections before killing
-        self._scheduler.destroy(self._job_id)
+        self._scheduler.destroy(self._job_id, self._command_announceable())
 
     @transition(field=state,
                 source=[INITIALIZED, CREATED, DESTROYED],

controller/scheduler/coreos.py

@@ -46,9 +46,17 @@ def tearDown(self):
         """
         return
 
+    # announcer helpers
+
+    def _log_skipped_announcer(self, action, name):
+        """
+        Logs a message stating that this operation doesn't require an announcer
+        """
+        print "-- skipping announcer {} for {}".format(action, name)
+
     # job api
 
-    def create(self, name, image, command='', template=None):
+    def create(self, name, image, command='', template=None, use_announcer=True):
         """
         Create a new job
         """
@@ -57,11 +65,10 @@ def create(self, name, image, command='', template=None):
         self._create_container(name, image, command, template or CONTAINER_TEMPLATE, env)
         self._create_log(name, image, command, LOG_TEMPLATE, env)
 
-        # only announce web and cmd processes
-        if command.lower() in ['start web', '']:
+        if use_announcer:
             self._create_announcer(name, image, command, ANNOUNCE_TEMPLATE, env)
         else:
-            print "-- skipping announcer for {} - cmd type is {}".format(name,command)
+            self._log_skipped_announcer('create', name)
 
     def _create_container(self, name, image, command, template, env):
         l = locals().copy()
@@ -87,16 +94,20 @@ def _create_log(self, name, image, command, template, env):
         return subprocess.check_call('fleetctl.sh submit {name}-log.service'.format(**locals()),  # noqa
                                      shell=True, env=env)
 
-    def start(self, name):
+    def start(self, name, use_announcer=True):
         """
         Start an idle job
         """
         print 'Starting {name}'.format(**locals())
         env = self.env.copy()
         self._start_container(name, env)
         self._start_log(name, env)
-        self._start_announcer(name, env)
-        self._wait_for_announcer(name, env)
+
+        if use_announcer:
+            self._start_announcer(name, env)
+            self._wait_for_announcer(name, env)
+        else:
+            self._log_skipped_announcer('start', name)
 
     def _start_log(self, name, env):
         subprocess.check_call(
@@ -125,13 +136,18 @@ def _wait_for_announcer(self, name, env):
         else:
             raise RuntimeError('Container failed to start')
 
-    def stop(self, name):
+    def stop(self, name, use_announcer=True):
         """
         Stop a running job
         """
         print 'Stopping {name}'.format(**locals())
         env = self.env.copy()
-        self._stop_announcer(name, env)
+
+        if use_announcer:
+            self._stop_announcer(name, env)
+        else:
+            self._log_skipped_announcer('stop', name)
+
         self._stop_container(name, env)
         self._stop_log(name, env)
 
@@ -150,13 +166,18 @@ def _stop_log(self, name, env):
             'fleetctl.sh stop -block-attempts=600 {name}-log.service'.format(**locals()),
             shell=True, env=env)
 
-    def destroy(self, name):
+    def destroy(self, name, use_announcer=True):
         """
         Destroy an existing job
         """
         print 'Destroying {name}'.format(**locals())
         env = self.env.copy()
-        self._destroy_announcer(name, env)
+
+        if use_announcer:
+            self._destroy_announcer(name, env)
+        else:
+            self._log_skipped_announcer('destroy', name)
+
         self._destroy_container(name, env)
         self._destroy_log(name, env)
 

controller/scheduler/mock.py

@@ -26,25 +26,25 @@ def tearDown(self):
 
     # job api
 
-    def create(self, name, image, command):
+    def create(self, name, image, command, use_announcer):
         """
         Create a new job
         """
         return {'state': 'inactive'}
 
-    def start(self, name):
+    def start(self, name, use_announcer):
         """
         Start an idle job
         """
         return {'state': 'active'}
 
-    def stop(self, name):
+    def stop(self, name, use_announcer):
         """
         Stop a running job
         """
         return {'state': 'inactive'}
 
-    def destroy(self, name):
+    def destroy(self, name, use_announcer):
         """
         Destroy an existing job
         """