Implement 'pool': An attribute & constraint that lets you split a clu…

…ster into multiple parts, without having to use Mesos's roles.

paasta_tools/chronos_tools.py

@@ -221,7 +221,7 @@ def get_env(self):
         return [{"name": key, "value": value} for key, value in original_env.iteritems()]
 
     def get_constraints(self):
-        return self.config_dict.get('constraints')
+        return self.config_dict.get('constraints', [['pool', 'EQUALS', self.get_pool()]])
 
     def check_bounce_method(self):
         bounce_method = self.get_bounce_method()

paasta_tools/marathon_tools.py

@@ -259,9 +259,12 @@ def get_constraints(self, service_namespace_config):
             discover_level = service_namespace_config.get_discover()
             locations = get_mesos_slaves_grouped_by_attribute(
                 attribute=discover_level, blacklist=self.get_deploy_blacklist())
+            pool = self.get_pool()
+
             deploy_constraints = deploy_blacklist_to_constraints(self.get_deploy_blacklist())
             routing_constraints = [[discover_level, "GROUP_BY", str(len(locations))]]
-            return routing_constraints + deploy_constraints
+            pool_constraints = [["pool", "LIKE", pool]]
+            return routing_constraints + deploy_constraints + pool_constraints
 
     def format_marathon_app_dict(self, app_id, docker_url, docker_volumes, service_namespace_config):
         """Create the configuration that will be passed to the Marathon REST API.

paasta_tools/utils.py

@@ -218,6 +218,18 @@ def get_extra_volumes(self):
         <https://mesosphere.github.io/marathon/docs/native-docker.html>`_"""
         return self.config_dict.get('extra_volumes', [])
 
+    def get_pool(self):
+        """Which pool of nodes this job should run on. This can be used to mitigate noisy neighbors, by putting
+        particularly noisy or noise-sensitive jobs into different pools.
+
+        This is implemented with an attribute "pool" on each mesos slave and by adding a constraint to Marathon/Chronos
+        application defined by this instance config.
+
+        Eventually this may be implemented with Mesos roles, once a framework can register under multiple roles.
+
+        :returns: the "pool" attribute in your config dict, or the string "default" if not specified."""
+        return self.config_dict.get('pool', 'default')
+
 
 def validate_service_instance(service, instance, cluster, soa_dir):
     marathon_services = get_services_for_cluster(cluster=cluster, instance_type='marathon', soa_dir=soa_dir)

tests/test_chronos_tools.py

@@ -385,6 +385,18 @@ def test_get_constraints(self):
         actual = fake_conf.get_constraints()
         assert actual == fake_constraints
 
+    def test_get_constraints_pool(self):
+        fake_pool = 'poolname'
+        fake_conf = chronos_tools.ChronosJobConfig(
+            service='fake_name',
+            cluster='fake_cluster',
+            instance='fake_instance',
+            config_dict={'pool': fake_pool},
+            branch_dict={},
+        )
+        actual = fake_conf.get_constraints()
+        assert actual == [['pool', 'EQUALS', 'poolname']]
+
     def test_get_retries_default(self):
         fake_conf = chronos_tools.ChronosJobConfig(
             service='fake_name',
@@ -828,7 +840,7 @@ def test_format_chronos_job_dict(self):
             'scheduleTimeZone': None,
             'environmentVariables': [],
             'arguments': None,
-            'constraints': None,
+            'constraints': [['pool', 'EQUALS', 'default']],
             'retries': 2,
             'epsilon': fake_epsilon,
             'name': 'test_job',
@@ -1161,7 +1173,7 @@ def test_create_complete_config(self):
             actual = chronos_tools.create_complete_config('fake-service', 'fake-job')
             expected = {
                 'arguments': None,
-                'constraints': None,
+                'constraints': [['pool', 'EQUALS', 'default']],
                 'schedule': 'R/2015-03-25T19:36:35Z/PT5M',
                 'async': False,
                 'cpus': 5.5,
@@ -1216,7 +1228,7 @@ def test_create_complete_config_desired_state_start(self):
             actual = chronos_tools.create_complete_config('fake_service', 'fake_job')
             expected = {
                 'arguments': None,
-                'constraints': None,
+                'constraints': [['pool', 'EQUALS', 'default']],
                 'schedule': 'R/2015-03-25T19:36:35Z/PT5M',
                 'async': False,
                 'cpus': 5.5,
@@ -1271,7 +1283,7 @@ def test_create_complete_config_desired_state_stop(self):
             actual = chronos_tools.create_complete_config('fake_service', 'fake_job')
             expected = {
                 'arguments': None,
-                'constraints': None,
+                'constraints': [['pool', 'EQUALS', 'default']],
                 'schedule': 'R/2015-03-25T19:36:35Z/PT5M',
                 'async': False,
                 'cpus': 5.5,
@@ -1342,7 +1354,7 @@ def test_create_complete_config_respects_extra_volumes(self):
             actual = chronos_tools.create_complete_config('fake_service', 'fake_job')
             expected = {
                 'arguments': None,
-                'constraints': None,
+                'constraints': [['pool', 'EQUALS', 'default']],
                 'schedule': 'R/2015-03-25T19:36:35Z/PT5M',
                 'async': False,
                 'cpus': 5.5,

tests/test_marathon_tools.py

@@ -812,7 +812,7 @@ def test_format_marathon_app_dict(self):
                 'type': 'DOCKER',
                 'volumes': fake_volumes,
             },
-            'constraints': [["habitat", "GROUP_BY", "1"]],
+            'constraints': [["habitat", "GROUP_BY", "1"], ["pool", "LIKE", "default"]],
             'uris': ['file:///root/.dockercfg', ],
             'mem': fake_mem,
             'env': fake_env,
@@ -1045,7 +1045,11 @@ def test_get_constraints_default(self):
             autospec=True,
         ) as get_slaves_patch:
             get_slaves_patch.return_value = {'fake_region': {}}
-            assert fake_conf.get_constraints(fake_service_namespace_config) == [["region", "GROUP_BY", "1"]]
+            expected_constraints = [
+                ["region", "GROUP_BY", "1"],
+                ["pool", "LIKE", "default"],
+            ]
+            assert fake_conf.get_constraints(fake_service_namespace_config) == expected_constraints
 
     def test_get_constraints_from_discover(self):
         fake_service_namespace_config = marathon_tools.ServiceNamespaceConfig({
@@ -1065,7 +1069,11 @@ def test_get_constraints_from_discover(self):
             autospec=True,
         ) as get_slaves_patch:
             get_slaves_patch.return_value = {'fake_region': {}, 'fake_other_region': {}}
-            assert fake_conf.get_constraints(fake_service_namespace_config) == [["habitat", "GROUP_BY", "2"]]
+            expected_constraints = [
+                ["habitat", "GROUP_BY", "2"],
+                ["pool", "LIKE", "default"],
+            ]
+            assert fake_conf.get_constraints(fake_service_namespace_config) == expected_constraints
             get_slaves_patch.assert_called_once_with(attribute='habitat', blacklist=[])
 
     def test_get_constraints_respects_deploy_blacklist(self):
@@ -1078,7 +1086,11 @@ def test_get_constraints_respects_deploy_blacklist(self):
             config_dict={'deploy_blacklist': fake_deploy_blacklist},
             branch_dict={},
         )
-        expected_constraints = [["region", "GROUP_BY", "1"], ["region", "UNLIKE", "fake_blacklisted_region"]]
+        expected_constraints = [
+            ["region", "GROUP_BY", "1"],
+            ["region", "UNLIKE", "fake_blacklisted_region"],
+            ["pool", "LIKE", "default"],
+        ]
         with mock.patch(
             'paasta_tools.marathon_tools.get_mesos_slaves_grouped_by_attribute',
             autospec=True,
@@ -1974,7 +1986,7 @@ def test_create_complete_config_no_smartstack():
             'health_checks': [],
             'env': {},
             'id': fake_job_id,
-            'constraints': [["region", "GROUP_BY", "1"]],
+            'constraints': [["region", "GROUP_BY", "1"], ["pool", "LIKE", "default"]],
         }
         assert actual == expected
 
@@ -2054,7 +2066,7 @@ def test_create_complete_config_with_smartstack():
             ],
             'env': {},
             'id': fake_job_id,
-            'constraints': [["region", "GROUP_BY", "1"]],
+            'constraints': [["region", "GROUP_BY", "1"], ["pool", "LIKE", "default"]],
         }
         assert actual == expected
 
@@ -2138,7 +2150,7 @@ def test_create_complete_config_utilizes_extra_volumes():
             'health_checks': [],
             'env': {},
             'id': fake_job_id,
-            'constraints': [["region", "GROUP_BY", "1"]],
+            'constraints': [["region", "GROUP_BY", "1"], ["pool", "LIKE", "default"]],
         }
         assert actual == expected
 

tests/test_utils.py

@@ -927,6 +927,27 @@ def test_extra_volumes_normal(self):
         )
         assert fake_conf.get_extra_volumes() == fake_extra_volumes
 
+    def test_get_pool(self):
+        pool = "poolname"
+        fake_conf = utils.InstanceConfig(
+            service='',
+            cluster='',
+            instance='',
+            config_dict={'pool': pool},
+            branch_dict={},
+        )
+        assert fake_conf.get_pool() == pool
+
+    def test_get_pool_default(self):
+        fake_conf = utils.InstanceConfig(
+            service='',
+            cluster='',
+            instance='',
+            config_dict={},
+            branch_dict={},
+        )
+        assert fake_conf.get_pool() == 'default'
+
 
 def test_is_under_replicated_ok():
     num_available = 1