Merge pull request #261 from helgi/k8s_tags

feat(tags): implement tags as k8s node selectors

Author: helgi

rootfs/api/models.py

@@ -121,6 +121,13 @@ class Meta:
         """Mark :class:`AuditedModel` as abstract."""
         abstract = True
 
+    @property
+    def _scheduler(self):
+        mod = importlib.import_module(settings.SCHEDULER_MODULE)
+        return mod.SchedulerClient(settings.SCHEDULER_URL,
+                                   settings.SCHEDULER_AUTH,
+                                   settings.SCHEDULER_OPTIONS)
+
 
 class UuidAuditedModel(AuditedModel):
     """Add a UUID primary key to an :class:`AuditedModel`."""
@@ -160,13 +167,6 @@ def select_app_name(self):
 
         return name
 
-    @property
-    def _scheduler(self):
-        mod = importlib.import_module(settings.SCHEDULER_MODULE)
-        return mod.SchedulerClient(settings.SCHEDULER_URL,
-                                   settings.SCHEDULER_AUTH,
-                                   settings.SCHEDULER_OPTIONS)
-
     def save(self, **kwargs):
         if not self.id:
             self.id = generate_app_name()
@@ -528,10 +528,6 @@ class Container(UuidAuditedModel):
     type = models.CharField(max_length=128, blank=False)
     num = models.PositiveIntegerField()
 
-    @property
-    def _scheduler(self):
-        return self.app._scheduler
-
     @property
     def state(self):
         return self._scheduler.state(self.job_id).name
@@ -778,6 +774,18 @@ def save(self, **kwargs):
         except Config.DoesNotExist:
             pass
 
+        # verify the tags exist on any nodes as labels
+        if self.tags:
+            # Get all nodes with label selectors
+            nodes = self._scheduler._get_nodes(labels=self.tags).json()
+            if not nodes['items']:
+                labels = ['{}={}'.format(key, value) for key, value in self.tags.items()]
+                raise EnvironmentError(
+                    'These tags do not match labels on kubernetes nodes: {}'.format(
+                        ', '.join(labels)
+                    )
+                )
+
         return super(Config, self).save(**kwargs)
 
 
@@ -952,13 +960,6 @@ class Domain(AuditedModel):
     app = models.ForeignKey('App')
     domain = models.TextField(blank=False, null=False, unique=True)
 
-    @property
-    def _scheduler(self):
-        mod = importlib.import_module(settings.SCHEDULER_MODULE)
-        return mod.SchedulerClient(settings.SCHEDULER_URL,
-                                   settings.SCHEDULER_AUTH,
-                                   settings.SCHEDULER_OPTIONS)
-
     def _fetch_service_config(self, app):
         # Get the service from k8s to attach the domain correctly
         svc = self._scheduler._get_service(app, app).json()

rootfs/api/views.py

@@ -274,6 +274,12 @@ class ConfigViewSet(ReleasableViewSet):
     model = models.Config
     serializer_class = serializers.ConfigSerializer
 
+    def create(self, request, **kwargs):
+        try:
+            return super(ConfigViewSet, self).create(request, **kwargs)
+        except EnvironmentError as e:
+            return Response({'detail': str(e)}, status=status.HTTP_400_BAD_REQUEST)
+
     def post_save(self, config):
         release = config.app.release_set.latest()
         self.release = release.new(self.request.user, config=config, build=release.build)

rootfs/scheduler/__init__.py

@@ -80,7 +80,8 @@
             }
             ]
           }
-        ]
+        ],
+        "nodeSelector": {}
       }
     }
   }
@@ -156,6 +157,7 @@
             ]
           }
         ],
+        "nodeSelector": {},
         "volumes":[
         {
             "name":"minio-user",
@@ -480,8 +482,8 @@ def _create_namespace(self, app_name):
             "apiVersion": self.apiversion,
             "metadata": {
                 "name": app_name
-                }
             }
+        }
         resp = self.session.post(url, json=data)
         if not resp.status_code == 201:
             error(resp, "create Namespace {}".format(app_name))
@@ -622,16 +624,24 @@ def _create_rc(self, name, image, command, **kwargs):  # noqa
         }
         template = string.Template(TEMPLATE).substitute(l)
         js_template = json.loads(template)
+
+        # apply tags as needed
+        tags = kwargs.get('tags', {})
+        js_template["spec"]["template"]["spec"]["nodeSelector"] = tags
+
+        # Deal with container information
         containers = js_template["spec"]["template"]["spec"]["containers"]
         containers[0]['args'] = args
         loc = locals().copy()
         loc.update(re.match(MATCH, container_fullname).groupdict())
         mem = kwargs.get('memory', {}).get(app_type)
         cpu = kwargs.get('cpu', {}).get(app_type)
         env = kwargs.get('envs', {})
+
         if env:
             for k, v in env.items():
                 containers[0]["env"].append({"name": k, "value": v})
+
         if mem or cpu:
             containers[0]["resources"] = {"limits": {}}
 
@@ -653,11 +663,13 @@ def _create_rc(self, name, image, command, **kwargs):  # noqa
         if unhealthy(resp.status_code):
             error(resp, 'create ReplicationController "{}" in Namespace "{}"',
                   name, app_name)
+
         create = False
         for _ in range(30):
             if not create and self._get_rc_status(name, app_name) == 404:
                 time.sleep(1)
                 continue
+
             create = True
             rc = self._get_rc(name, app_name)
             if (
@@ -835,4 +847,30 @@ def _pod_log(self, name, namespace):
 
         return resp.status_code, resp.text, resp.reason
 
+    # NODES #
+
+    def _get_nodes(self, **kwargs):
+        path = '/nodes'
+        query = {}
+
+        # labels and fields are encoded slightly differently than python-requests can do
+        labels = kwargs.get('labels', {})
+        if labels:
+            # http://kubernetes.io/v1.1/docs/user-guide/labels.html#list-and-watch-filtering
+            labels = ['{}={}'.format(key, value) for key, value in labels.items()]
+            query['labelSelector'] = ','.join(labels)
+
+        fields = kwargs.get('fields', {})
+        if fields:
+            fields = ['{}={}'.format(key, value) for key, value in fields.items()]
+            query['fieldSelector'] = ','.join(fields)
+
+        url = self._api(path)
+        response = self.session.get(url, params=query)
+        if unhealthy(response.status_code):
+            error(response, 'get Nodes')
+
+        return response
+
+
 SchedulerClient = KubeHTTPClient

rootfs/scheduler/mock.py

@@ -64,5 +64,11 @@ def _get_service(self, namespace, name):
     def _update_service(self, namespace, name, data):
         pass
 
+    def _get_nodes(self, **kwargs):
+        resp = requests.Response()
+        resp.status_code = 200
+        resp._content = b'{"items": [{"metadata": {"labels": {"env": "prod"}}}]}'
+        return resp
+
 
 SchedulerClient = MockSchedulerClient