fixed minor issues with ApplianceMonitor and service/job requests; ad…

…ded Chronos and Zookeeper Dockerfiles

appliance/handler.py

@@ -3,7 +3,7 @@
 
 from tornado.web import RequestHandler
 
-from appliance.manager import ApplianceManager, ApplianceMonitor
+from appliance.manager import ApplianceManager
 from container.manager import ContainerManager
 from util import message, error
 from util import Loggable
@@ -14,15 +14,12 @@ class AppliancesHandler(RequestHandler, Loggable):
   def initialize(self, config):
     self.__app_mgr = ApplianceManager(config)
     self.__contr_mgr = ContainerManager(config)
-    self.__app_monitor = ApplianceMonitor(self.__app_mgr, self.__contr_mgr)
 
   async def post(self):
     data = tornado.escape.json_decode(self.request.body)
     status, app, err = await self.__app_mgr.create_appliance(data)
     self.set_status(status)
     self.write(json.dumps(app.to_render() if status == 201 else error(err)))
-    if status == 201:
-      self.__app_monitor.spawn(app)
 
 
 class ApplianceHandler(RequestHandler, Loggable):

appliance/manager.py

@@ -13,10 +13,11 @@ class ApplianceManager(Loggable, metaclass=Singleton):
 
   def __init__(self, config):
     self.__config = config
+    self.__http_cli = SecureAsyncHttpClient(config)
     self.__app_col = MotorClient().requester.appliance
     self.__contr_col = MotorClient().requester.container
     self.__contr_mgr = ContainerManager(config)
-    self.__http_cli = SecureAsyncHttpClient(config)
+    self.__app_monitor = ApplianceMonitor(self, self.__contr_mgr)
 
   async def get_appliance(self, app_id):
     status, app, err = await self._get_appliance_from_db(app_id)
@@ -44,13 +45,17 @@ async def create_appliance(self, data):
       self.logger.error(err)
       return status, None, err
     self.logger.info(msg)
+    self.logger.info("Start monitoring appliance '%s'"%app)
+    self.__app_monitor.spawn(app)
     return 201, app, None
 
   async def delete_appliance(self, app_id):
     status, app, err = await self._get_appliance_from_db(app_id)
     if err:
       self.logger.error(err)
       return status, None, err
+    self.logger.info("Stop monitoring appliance '%s'"%app_id)
+    self.__app_monitor.stop(app_id)
     status, msg, err = await self.__contr_mgr.delete_containers(appliance=app_id,
                                                                 type=ContainerType.JOB.value)
     if err:
@@ -61,7 +66,6 @@ async def delete_appliance(self, app_id):
     if err and status != 404:
       self.logger.error(err)
       return 400, None, "Failed to deprovision appliance '%s'"%app_id
-    self.logger.info(msg)
     status, msg, err = await self._delete_appliance_from_db(app_id)
     if err:
       self.logger.error(err)
@@ -124,6 +128,11 @@ def spawn(self, app):
     self.__callbacks[app.id] = cb
     cb.start()
 
+  def stop(self, app_id):
+    cb = self.__callbacks.pop(app_id, None)
+    if cb:
+      cb.stop()
+
   async def _monitor_appliance(self, app):
     self.logger.info('Containers left: %s'%list(app.dag.parent_map.keys()))
     cb = self.__callbacks.get(app.id, None)
@@ -132,6 +141,14 @@ async def _monitor_appliance(self, app):
       cb.stop()
       self.__callbacks.pop(app.id, None)
       return
+    free_contrs = [c.id for c in app.dag.get_free_containers()]
+    self.logger.info('Launch free containers: %s'%free_contrs)
+    for c in app.dag.get_free_containers():
+      if c.state == ContainerState.SUBMITTED:
+        _, _, err = await self.__contr_mgr.provision_container(c)
+        if err:
+          self.logger.error("Failed to launch container '%s'"%c)
+          self.logger.error(err)
     self.logger.info('Update DAG')
     for c in app.dag.get_free_containers():
       _, c, err = await self.__contr_mgr.get_container(app.id, c.id)
@@ -144,13 +161,6 @@ async def _monitor_appliance(self, app):
         app.dag.remove_container(c.id)
       else:
         app.dag.update_container(c)
-    self.logger.info('Launch free containers')
-    for c in app.dag.get_free_containers():
-      if c.state == ContainerState.SUBMITTED:
-        _, _, err = await self.__contr_mgr.provision_container(c)
-        if err:
-          self.logger.error("Failed to launch container '%s'"%c)
-          self.logger.error(err)
     status, msg, err = await self.__app_mgr.save_appliance(app, False)
     if err:
       self.logger.error(err)

container/job.py

@@ -70,15 +70,14 @@ def to_request(self):
     r = dict(name=str(self),
              schedule='R%d/%s/P%s'%(self.repeats, self.start_time, self.interval),
              cpus=self.resources.cpus, mem=self.resources.mem, disk=self.resources.disk,
-             shell=not self.args,
-             environmentVariables=self.env,
+             shell=self.args is None,
+             command = self.cmd if self.cmd else '',
+             environmentVariables=[dict(name=k, value=v) for k, v in self.env.items()],
              container=dict(type='DOCKER',
                             image=self.image,
                             network=self.network_mode.value,
                             volumes=[v.to_request() for v in self.volumes],
                             forcePullImage=self.force_pull_image))
-    if self.cmd:
-      r['command'] = self.cmd
     if self.args:
       r['arguments'] = self.args
     parameters = [dict(key='privileged', value=self.is_privileged)]

container/manager.py

@@ -196,12 +196,12 @@ async def _get_update_job(self, job):
     return status, job, None
 
   async def _provision_service(self, service):
-    url = '%s/apps'%self.__config.url.service_scheduler
+    url = '%s/apps?force=true'%self.__config.url.service_scheduler
     body = dict(service.to_request())
     return await self.__http_cli.post(url, body)
 
   async def _delete_service(self, contr):
-    return await self.__http_cli.delete('%s/apps%s'%(self.__config.url.service_scheduler, contr))
+    return await self.__http_cli.delete('%s/apps%s?force=true'%(self.__config.url.service_scheduler, contr))
 
   async def _provision_job(self, job):
     return await self.__http_cli.post('%s/iso8601'%self.__config.url.job_scheduler, job.to_request())

container/service.py

@@ -151,7 +151,8 @@ def to_save(self):
   def to_request(self):
     r = dict(id=str(self), instances=self.instances,
              **self.resources.to_request(),
-             env=self.env, labels=self.labels,
+             env={k: str(v) for k, v in self.env.items()},
+             labels=self.labels,
              requirePorts=len(self.ports) > 0,
              acceptedResourceRoles=[ "slave_public", "*" ],
              container=dict(type='DOCKER',
@@ -176,7 +177,7 @@ def to_request(self):
       r['networks'] = [dict(mode='container', name='dcos')]
     # set port definitions
     if self.network_mode == NetworkMode.HOST:
-      r['portDefinitions'] = [dict(protocol=p.protocol, port=p.host_port,
+      r['portDefinitions'] = [dict(protocol=p.protocol, port=p.container_port,
                                    labels={"VIP_%d"%i: "/:%d"%p.load_balanced_port})
                               for i, p in enumerate(self.ports)]
     else:

docker/chronos/Dockerfile

@@ -0,0 +1,21 @@
+FROM ubuntu:xenial
+MAINTAINER Fan Jiang <dcvan@renci.org>
+
+RUN apt-key adv --keyserver keyserver.ubuntu.com --recv E56151BF \
+    && echo "deb http://repos.mesosphere.io/debian jessie main" >> /etc/apt/sources.list.d/mesosphere.list \
+    && apt-get update \
+    && apt-get install --no-install-recommends -yq mesos git openjdk-8-jdk python-dev curl iproute2 maven
+
+RUN curl -sL https://deb.nodesource.com/setup_8.x -o nodesource_setup.sh
+RUN bash nodesource_setup.sh && rm -rf nodesource_setup.sh
+RUN apt-get install -yq nodejs
+RUN git clone https://github.com/dcvan24/chronos.git
+RUN cd chronos && mvn package -DskipTests 
+RUN cd .. && cp -a chronos/target/chronos*.jar $HOME && rm -rf chronos
+RUN rm -rf $HOME/.m2
+RUN apt-get purge -y maven git && apt-get autoremove -y && apt-get autoclean
+
+COPY run.sh /usr/bin/run.sh
+RUN chmod +x /usr/bin/run.sh
+
+ENTRYPOINT ["/usr/bin/run.sh"]

docker/chronos/run.sh

@@ -0,0 +1,4 @@
+#!/bin/bash 
+export LIBPROCESS_IP=$(ip route get 8.8.8.8 | head -1 | cut -d' ' -f8)
+env MESOS_NATIVE_JAVA_LIBRARY=/usr/lib/libmesos.so \
+    java -cp $HOME/chronos*.jar org.apache.mesos.chronos.scheduler.Main $@

docker/zookeeper/Dockerfile

@@ -0,0 +1,9 @@
+FROM ubuntu:xenial
+MAINTAINER Fan Jiang <dcvan@renci.org>
+
+RUN apt-get update \
+    && apt-get install --no-install-recommends -yq zookeeperd
+
+EXPOSE 2181
+
+ENTRYPOINT ["/usr/share/zookeeper/bin/zkServer.sh", "start-foreground"]