Add nilmrun-ps, -kill, and -run commands

Multiprocessing and Apache's mod_wsgi don't play nicely.  Switch to
manually managing processes via subprocess.Popen etc instead.  When
running arbitrary code, we write it to an external file, and running
functions directly is no longer supported.

README.txt

@@ -6,10 +6,8 @@ Prerequisites:
   # Runtime and build environments
   sudo apt-get install python2.7 python-setuptools
 
-  # Base dependencies
-  sudo apt-get install python-numpy python-scipy
-
-  nilmdb (1.8.0+)
+  # Plus nilmdb and its dependencies
+  nilmdb (1.9.5+)
 
 Install:
 

docs/wsgi.md

@@ -21,13 +21,13 @@ arbitrary commands.
         SSLEngine On
 
         WSGIScriptAlias /nilmrun /home/nilm/nilmrun.wsgi
-        WSGIApplicationGroup nilmrun-appgroup
-        WSGIProcessGroup nilmrun-procgroup
         WSGIDaemonProcess nilmrun-procgroup threads=32 user=nilm group=nilm
+        <Location /nilmrun>
+            WSGIProcessGroup nilmrun-procgroup
+            WSGIApplicationGroup nilmrun-appgroup
+            SSLRequireSSL
 
             # Access control example:
-        <Location /nilmrun>
-            SSLRequireSSL
             Order deny,allow
             Deny from all
             Allow from 1.2.3.4

nilmrun/processmanager.py

@@ -3,7 +3,7 @@
 from nilmdb.utils.printf import *
 
 import threading
-import multiprocessing
+import subprocess
 import cStringIO
 import sys
 import os
@@ -11,8 +11,12 @@ import signal
 import time
 import uuid
 import psutil
-import imp
-import traceback
+import tempfile
+import atexit
+import shutil
+
+class ProcessError(Exception):
+    pass
 
 class LogReceiver(object):
     """Spawn a thread that listens to a pipe for log messages,
@@ -38,75 +42,48 @@ class LogReceiver(object):
         self.log = cStringIO.StringIO()
 
 class Process(object):
-    """Spawn and manage a process that calls a Python function"""
-    def __init__(self, name, function, parameters):
-        self.parameters = parameters
+    """Spawn and manage a subprocess, and capture its output."""
+    def __init__(self, argv, tempfile = None):
         self.start_time = None
-        self.name = name
 
         # Use a pipe for communicating log data
         (rpipe, wpipe) = os.pipe()
         self._log = LogReceiver(rpipe)
 
-        # Start the function in a new process
-        self._process = multiprocessing.Process(
-            target = self._trampoline, name = name,
-            args = (function, rpipe, wpipe, parameters))
-        self._process.daemon = True
-        self._process.start()
+        # Stdin is null
+        nullfd = os.open(os.devnull, os.O_RDONLY)
 
-        # Close the writer end of the pipe, get process info
+        # Spawn the new process
+        try:
+            self._process = subprocess.Popen(args = argv, stdin = nullfd,
+                                             stdout = wpipe, stderr = wpipe,
+                                             close_fds = True, cwd = "/tmp")
+        except (OSError, TypeError) as e:
+            raise ProcessError(str(e))
+        finally:
+            # Close the FDs we don't need
             os.close(wpipe)
+            os.close(nullfd)
+
+        # Get process info
         self.start_time = time.time()
         self.pid = str(uuid.uuid1(self._process.pid or 0))
 
-    def _trampoline(self, func, rpipe, wpipe, param): # pragma: no cover
-        # No coverage report for this, because it's executed in a subprocess
-        """Trampoline function to set up stdio and call the real function."""
-        # Close the reader end of the pipe
-        os.close(rpipe)
-
-        # Like os.close() but ignores errors
-        def tryclose(fd):
-            try:
-                os.close(fd)
-            except OSError:
-                pass
-
-        # Remap stdio to go to the pipe.  We do this at the OS level,
-        # replacing FDs, so that future spawned processes do the right thing.
-
-        # stdin
-        sys.stdin.close()
-        tryclose(0)
-        fd = os.open(os.devnull, os.O_RDONLY) # 0
-        sys.stdin = os.fdopen(fd, 'r', 0)
-
-        # stdout
-        sys.stdout.close()
-        tryclose(1)
-        fd = os.dup(wpipe) # 1
-        sys.stdout = os.fdopen(fd, 'w', 0)
-
-        # stdout
-        sys.stderr.close()
-        tryclose(2)
-        fd = os.dup(wpipe) # 2
-        sys.stderr = os.fdopen(fd, 'w', 0)
-
-        # Don't need this extra fd
-        os.close(wpipe)
-
-        # Ready to go -- call the function, exit when it's done
-        func(param)
-        sys.exit(0)
+    def _join(self, timeout = 1.0):
+        start = time.time()
+        while True:
+            if self._process.poll() is not None:
+                return True
+            if (time.time() - start) >= timeout:
+                return False
+            time.sleep(0.1)
 
     def terminate(self, timeout = 1.0):
         """Terminate a process, and all of its children that are in the same
         process group."""
+        try:
             # First give it some time to die on its own
-        self._process.join(timeout)
-        if not self.alive:
+            if self._join(timeout):
                 return True
 
             def getpgid(pid):
@@ -115,6 +92,12 @@ class Process(object):
                 except OSError: # pragma: no cover
                     return None
 
+            def kill(pid, sig):
+                try:
+                    return os.kill(pid, sig)
+                except OSError: # pragma: no cover
+                    return
+
             # Find all children
             group = getpgid(self._process.pid)
             main = psutil.Process(self._process.pid)
@@ -123,21 +106,21 @@ class Process(object):
             # Kill with SIGTERM, if they're still in this process group
             for proc in allproc:
                 if getpgid(proc.pid) == group:
-                os.kill(proc.pid, signal.SIGTERM)
+                    kill(proc.pid, signal.SIGTERM)
 
             # Wait for it to die again
-        self._process.join(timeout)
-        if not self.alive:
+            if self._join(timeout):
                 return True
 
             # One more try with SIGKILL
             for proc in allproc:
                 if getpgid(proc.pid) == group:
-                os.kill(proc.pid, signal.SIGKILL)
+                    kill(proc.pid, signal.SIGKILL)
 
             # See if it worked
-        self._process.join(timeout)
-        return not self.alive
+            return self._join(timeout)
+        except psutil.Error: # pragma: no cover (race condition)
+            return True
 
     def clear_log(self):
         self._log.clear()
@@ -148,64 +131,80 @@ class Process(object):
 
     @property
     def alive(self):
-        return self._process.is_alive()
+        return self._process.poll() is None
 
     @property
     def exitcode(self):
-        return self._process.exitcode
+        return self._process.returncode
 
-def _exec_user_code(codeargs): # pragma: no cover (runs in subprocess)
-    """Execute 'code' as if it were placed into a file and executed"""
-    (code, args) = codeargs
-    # This is split off into a separate function because the Python3
-    # syntax of "exec" triggers a SyntaxError in Python2, if it's within
-    # a nested function.
-    imp.acquire_lock()
+    def get_info_prepare(self):
+        """Prepare the process list and measurement for .get_info.
+        Call .get_info() about a second later."""
         try:
-        module = imp.new_module("__main__")
-    finally:
-        imp.release_lock()
-    module.__file__ = "<user-code>"
-    sys.argv = [''] + args
-    # Wrap the compile and exec in a try/except so we can format the
-    # exception more nicely.
+            main = psutil.Process(self._process.pid)
+            self._process_list = [ main ] + main.get_children(recursive = True)
+            for proc in self._process_list:
+                proc.get_cpu_percent(0)
+        except psutil.Error: # pragma: no cover (race condition)
+            self._process_list = [ ]
+
+    @staticmethod
+    def get_empty_info():
+        return { "cpu_percent": 0,
+                 "cpu_user": 0,
+                 "cpu_sys": 0,
+                 "mem_phys": 0,
+                 "mem_virt": 0,
+                 "io_read": 0,
+                 "io_write": 0,
+                 "procs": 0 }
+
+    def get_info(self):
+        """Return a dictionary with info about the process CPU and memory
+        usage.  Call .get_info_prepare() about a second before this."""
+        d = self.get_empty_info()
+        for proc in self._process_list:
             try:
-        codeobj = compile(code, '<user-code>', 'exec',
-                          flags = 0, dont_inherit = 1)
-        exec(codeobj, module.__dict__, {})
-    except:
-        try:
-            # Pull out the exception
-            info = sys.exc_info()
-            tblist = traceback.extract_tb(info[2])
-
-            # First entry is probably this code; get rid of it
-            if len(tblist) and tblist[0][2] == '_exec_user_code':
-                tblist = tblist[1:]
-
-            # Add the user's source code to every line that's missing it
-            lines = code.splitlines()
-            maxline = len(lines)
-            for (n, (name, line, func, text)) in enumerate(tblist):
-                if name == '<user-code>' and text is None and line <= maxline:
-                    tblist[n] = (name, line, func, lines[line-1].strip())
-
-            # Format it in the usual manner
-            out = ['Traceback (most recent call last):\n']
-            out.extend(traceback.format_list(tblist))
-            out.extend(traceback.format_exception_only(info[0], info[1]))
-        finally:
-            # Need to explicitly delete traceback object to avoid ref cycle
-            del info
-        sys.stderr.write("".join(out))
-        sys.stderr.flush()
-        sys.exit(1)
-    sys.exit(0)
+                d["cpu_percent"] += proc.get_cpu_percent(0)
+                cpuinfo = proc.get_cpu_times()
+                d["cpu_user"] += cpuinfo.user
+                d["cpu_sys"] += cpuinfo.system
+                meminfo = proc.get_memory_info()
+                d["mem_phys"] += meminfo.rss
+                d["mem_virt"] += meminfo.vms
+                ioinfo = proc.get_io_counters()
+                d["io_read"] += ioinfo.read_bytes
+                d["io_write"] += ioinfo.write_bytes
+                d["procs"] += 1
+            except psutil.Error:
+                pass
+        return d
 
 class ProcessManager(object):
     """Track and manage a collection of Process objects"""
     def __init__(self):
         self.processes = {}
+        self.tmpdirs = {}
+        atexit.register(self._atexit)
+
+    def _cleanup_tmpdir(self, pid):
+        if pid in self.tmpdirs:
+            try:
+                shutil.rmtree(self.tmpdirs[pid])
+            except OSError: # pragma: no cover
+                pass
+            del self.tmpdirs[pid]
+
+    def _atexit(self):
+        # Kill remaining processes, remove their dirs
+        for pid in self.processes.keys():
+            try:
+                self.processes[pid].terminate()
+                del self.processes[pid]
+                shutil.rmtree(self.tmpdirs[pid])
+                del self.tmpdirs[pid]
+            except Exception: # pragma: no cover
+                pass
 
     def __iter__(self):
         return iter(self.processes.keys())
@@ -213,35 +212,82 @@ class ProcessManager(object):
     def __getitem__(self, key):
         return self.processes[key]
 
-    def run_function(self, procname, function, parameters):
-        """Run a Python function that already exists"""
-        new = Process(procname, function, parameters)
+    def run_code(self, code, args):
+        """Evaluate 'code' as if it were placed into a Python file and
+        executed.  The arguments, which must be strings, will be
+        accessible in the code as sys.argv[1:]."""
+        # The easiest way to do this, by far, is to just write the
+        # code to a file.  Make a directory to put it in.
+        tmpdir = tempfile.mkdtemp(prefix = "nilmrun-usercode-")
+        try:
+            # Write the code
+            codepath = os.path.join(tmpdir, "usercode.py")
+            with open(codepath, "w") as f:
+                f.write(code)
+            # Save the args too, for debugging purposes
+            with open(os.path.join(tmpdir, "args.txt"), "w") as f:
+                f.write(repr(args))
+
+            # Run the code
+            argv = [ sys.executable, "-B", "-s", "-u", codepath ] + args
+            pid = self.run_command(argv)
+
+            # Save the temp dir
+            self.tmpdirs[pid] = tmpdir
+            tmpdir = None  # Don't need to remove it anymore
+
+            return pid
+        finally:
+            # Clean up tempdir if we didn't finish
+            if tmpdir is not None:
+                try:
+                    shutil.rmtree(tmpdir)
+                except OSError: # pragma: no cover
+                    pass
+
+    def run_command(self, argv):
+        """Execute a command line program"""
+        new = Process(argv)
         self.processes[new.pid] = new
         return new.pid
 
-    def run_code(self, procname, code, args):
-        """Evaluate 'code' as if it were placed into a Python file and
-        executed.  The arguments will be accessible in the code as
-        sys.argv[1:]."""
-        return self.run_function(procname, _exec_user_code, (code, args))
-
-    def run_command(self, procname, argv):
-        """Execute a command line program"""
-        def spwan_user_command(argv): # pragma: no cover (runs in subprocess)
-            try:
-                maxfd = os.sysconf("SC_OPEN_MAX")
-            except Exception:
-                maxfd = 256
-            os.closerange(3, maxfd)
-            try:
-                os.chdir("/tmp")
-            except OSError:
-                pass
-            os.execvp(argv[0], argv)
-        return self.run_function(procname, spwan_user_command, argv)
-
     def terminate(self, pid):
         return self.processes[pid].terminate()
 
     def remove(self, pid):
+        self._cleanup_tmpdir(pid)
         del self.processes[pid]
+
+    def get_info(self):
+        """Get info about all running PIDs"""
+        info = { "total" : Process.get_empty_info(),
+                 "pids" : {},
+                 "system" : {}
+                 }
+
+        # Trigger CPU usage collection
+        for pid in self:
+            self[pid].get_info_prepare()
+        psutil.cpu_percent(0, percpu = True)
+
+        # Give it some time
+        time.sleep(1)
+
+        # Retrieve info for system
+        info["system"]["cpu_percent"] = sum(psutil.cpu_percent(0, percpu=True))
+        info["system"]["cpu_max"] = 100.0 * psutil.NUM_CPUS
+        info["system"]["procs"] = len(psutil.get_pid_list())
+        # psutil > 0.6.0's psutil.virtual_memory() would be better here,
+        # but this should give the same info.
+        meminfo = psutil.phymem_usage()
+        info["system"]["mem_total"] = meminfo.total
+        info["system"]["mem_used"] = int(meminfo.total * meminfo.percent / 100)
+
+        # Retrieve info for each PID
+        for pid in self:
+            info["pids"][pid] = self[pid].get_info()
+            # Update totals
+            for key in info["total"]:
+                info["total"][key] += info["pids"][pid][key]
+
+        return info

nilmrun/server.py

@@ -5,10 +5,7 @@ import sys
 import os
 import socket
 import simplejson as json
-import decorator
-import psutil
 import traceback
-import argparse
 import time
 
 import nilmdb
@@ -23,7 +20,9 @@ from nilmdb.server.serverutil import (
         json_error_page,
         cherrypy_start,
         cherrypy_stop,
+        bool_param,
         )
+from nilmdb.utils import serializer_proxy
 import nilmrun
 import nilmrun.testfilter
 
@@ -62,14 +61,16 @@ class AppProcess(object):
         self.manager = manager
 
     def process_status(self, pid):
+        # We need to convert the log (which is bytes) to Unicode
+        # characters, in order to send it via JSON.  Treat it as UTF-8
+        # but replace invalid characters with markers.
+        log = self.manager[pid].log.decode('utf-8', errors='replace')
         return {
             "pid": pid,
             "alive": self.manager[pid].alive,
             "exitcode": self.manager[pid].exitcode,
-            "name": self.manager[pid].name,
             "start_time": self.manager[pid].start_time,
-            "parameters": self.manager[pid].parameters,
-            "log": self.manager[pid].log,
+            "log": log
             }
 
     # /process/status
@@ -78,6 +79,7 @@ class AppProcess(object):
     def status(self, pid, clear = False):
         """Return status about a process.  If clear = True, also clear
         the log."""
+        clear = bool_param(clear)
         if pid not in self.manager:
             raise cherrypy.HTTPError("404 Not Found", "No such PID")
         status = self.process_status(pid)
@@ -92,6 +94,14 @@ class AppProcess(object):
         """Return a list of processes in the manager."""
         return list(self.manager)
 
+    # /process/info
+    @cherrypy.expose
+    @cherrypy.tools.json_out()
+    def info(self):
+        """Return detailed CPU and memory info about the system and
+        all processes"""
+        return self.manager.get_info()
+
     # /process/remove
     @cherrypy.expose
     @cherrypy.tools.json_in()
@@ -116,34 +126,35 @@ class AppRun(object):
     @cherrypy.expose
     @cherrypy.tools.json_in()
     @cherrypy.tools.json_out()
+    @exception_to_httperror(nilmrun.processmanager.ProcessError)
     @cherrypy.tools.CORS_allow(methods = ["POST"])
     def command(self, argv):
         """Execute an arbitrary program on the server.  argv is a
         list of the program and its arguments: 'argv[0]' is the program
         and 'argv[1:]' are arguments"""
-        return self.manager.run_command("command", argv)
+        if not isinstance(argv, list):
+            raise cherrypy.HTTPError("400 Bad Request",
+                                     "argv must be a list of strings")
+        return self.manager.run_command(argv)
 
     # /run/code
     @cherrypy.expose
     @cherrypy.tools.json_in()
     @cherrypy.tools.json_out()
+    @exception_to_httperror(nilmrun.processmanager.ProcessError)
     @cherrypy.tools.CORS_allow(methods = ["POST"])
-    def code(self, code, args):
+    def code(self, code, args = None):
         """Execute arbitrary Python code.  'code' is a formatted string.
         It will be run as if it were written into a Python file and
-        executed, with the arguments in 'args' passed on the command line
-        (i.e., they end up in sys.argv[1:])"""
-        return self.manager.run_code("usercode", code, args)
-
-    # /run/testfilter
-    @cherrypy.expose
-    @cherrypy.tools.json_in()
-    @cherrypy.tools.json_out()
-    @exception_to_httperror(KeyError, ValueError)
-    @cherrypy.tools.CORS_allow(methods = ["POST"])
-    def testfilter(self, data):
-        return self.manager.run_function(
-            "dummy", nilmrun.testfilter.test, data)
+        executed.  'args' is a list of strings, and they are passed
+        on the command line as additional arguments (i.e., they end up
+        in sys.argv[1:])"""
+        if args is None:
+            args = []
+        if not isinstance(args, list):
+            raise cherrypy.HTTPError("400 Bad Request",
+                                     "args must be a list of strings")
+        return self.manager.run_code(code, args)
 
 class Server(object):
     def __init__(self, host = '127.0.0.1', port = 8080,
@@ -194,8 +205,12 @@ class Server(object):
         # error messages.
         cherrypy._cperror._ie_friendly_error_sizes = {}
 
+        # The manager maintains internal state and isn't necessarily
+        # thread-safe, so wrap it in the serializer.
+        manager = serializer_proxy(nilmrun.processmanager.ProcessManager)()
+
         # Build up the application and mount it
-        manager = nilmrun.processmanager.ProcessManager()
+        self._manager = manager
         root = App()
         root.process = AppProcess(manager)
         root.run = AppRun(manager)

scripts/kill.py

@@ -0,0 +1,51 @@
+#!/usr/bin/python
+
+from nilmdb.client.httpclient import HTTPClient, ClientError, ServerError
+from nilmdb.utils.printf import *
+import nilmrun
+
+import argparse
+import os
+import sys
+
+def main():
+    """Kill/remove a process from the NilmRun server"""
+    def_url = os.environ.get("NILMRUN_URL", "http://localhost/nilmrun/")
+    parser = argparse.ArgumentParser(
+        description = 'Kill/remove a process from the NilmRun server',
+        formatter_class = argparse.ArgumentDefaultsHelpFormatter)
+    parser.add_argument("-V", "--version", action="version",
+                        version = nilmrun.__version__)
+    group = parser.add_argument_group("Standard options")
+    group.add_argument('-u', '--url',
+                       help = 'NilmRun server URL', default = def_url)
+    group.add_argument('-n', '--noverify', action="store_true",
+                       help = 'Disable SSL certificate verification')
+    group = parser.add_argument_group("Program")
+    group.add_argument('-q', '--quiet', action="store_true",
+                       help = "Don't print out the final log contents")
+    group.add_argument('pid', nargs='+', help="PIDs to kill")
+    args = parser.parse_args()
+
+    client = HTTPClient(baseurl = args.url, verify_ssl = not args.noverify)
+
+    # Kill or remove process
+    all_failed = True
+    for pid in args.pid:
+        try:
+            s = client.post("process/remove", { "pid": pid })
+            if not args.quiet:
+                sys.stdout.write(s['log'])
+            all_failed = False
+        except ClientError as e:
+            if "404" in e.status:
+                fprintf(sys.stderr, "no such pid: %s\n", pid)
+            else:
+                raise
+
+    # Return error if we failed to remove any of them
+    if all_failed:
+        raise SystemExit(1)
+
+if __name__ == "__main__":
+    main()

scripts/ps.py

@@ -0,0 +1,66 @@
+#!/usr/bin/python
+
+from nilmdb.client.httpclient import HTTPClient, ClientError, ServerError
+from nilmdb.utils.printf import *
+from nilmdb.utils import datetime_tz
+import nilmrun
+
+import argparse
+import os
+
+def main():
+    """List NilmRun processes"""
+    def_url = os.environ.get("NILMRUN_URL", "http://localhost/nilmrun/")
+    parser = argparse.ArgumentParser(
+        description = 'List NilmRun processes',
+        formatter_class = argparse.ArgumentDefaultsHelpFormatter)
+    parser.add_argument("-V", "--version", action="version",
+                        version = nilmrun.__version__)
+    group = parser.add_argument_group("Standard options")
+    group.add_argument('-u', '--url',
+                       help = 'NilmRun server URL', default = def_url)
+    group.add_argument('-n', '--noverify', action="store_true",
+                       help = 'Disable SSL certificate verification')
+    args = parser.parse_args()
+
+    client = HTTPClient(baseurl = args.url, verify_ssl = not args.noverify)
+    # Print overall system info
+    info = client.get("process/info")
+    total = info['total']
+    system = info['system']
+    printf(" procs: %d nilm, %d other\n", info['total']['procs'],
+           info['system']['procs'] - info['total']['procs'])
+    printf("   cpu: %d%% nilm, %d%% other, %d%% max\n",
+           round(info['total']['cpu_percent']),
+           round(info['system']['cpu_percent'] - info['total']['cpu_percent']),
+           round(info['system']['cpu_max']))
+    printf("   mem: %d MiB used, %d MiB total, %d%%\n",
+           round(info['system']['mem_used'] / 1048576.0),
+           round(info['system']['mem_total'] / 1048576.0),
+           round(info['system']['mem_used'] * 100.0
+                 / info['system']['mem_total']))
+
+    if len(info['pids']) == 0:
+        printf("No running processes\n")
+        raise SystemExit(0)
+
+    # Print process detail for each managed process
+    fmt = "%-36s %-6s %-15s %-4s %-3s %-5s\n"
+    printf(fmt, "PID", "STATE", "SINCE", "PROC", "CPU", "LOG")
+    for pid in sorted(info['pids'].keys()):
+        pidinfo = client.get("process/status", { "pid": pid })
+        if pidinfo['alive']:
+            status = "alive"
+        else:
+            if pidinfo['exitcode']:
+                status = "error"
+            else:
+                status = "done"
+        dt = datetime_tz.datetime_tz.fromtimestamp(pidinfo['start_time'])
+        since = dt.strftime("%m/%d-%H:%M:%S")
+        printf(fmt, pid, status, since, info['pids'][pid]['procs'],
+               str(int(round(info['pids'][pid]['cpu_percent']))),
+               len(pidinfo['log']))
+
+if __name__ == "__main__":
+    main()

scripts/run.py

@@ -0,0 +1,60 @@
+#!/usr/bin/python
+
+from nilmdb.client.httpclient import HTTPClient, ClientError, ServerError
+from nilmdb.utils.printf import *
+import nilmrun
+
+import argparse
+import os
+import time
+import sys
+
+def main():
+    """Run a command on the NilmRun server"""
+    def_url = os.environ.get("NILMRUN_URL", "http://localhost/nilmrun/")
+    parser = argparse.ArgumentParser(
+        description = 'Run a command on the NilmRun server',
+        formatter_class = argparse.ArgumentDefaultsHelpFormatter)
+    parser.add_argument("-V", "--version", action="version",
+                        version = nilmrun.__version__)
+    group = parser.add_argument_group("Standard options")
+    group.add_argument('-u', '--url',
+                       help = 'NilmRun server URL', default = def_url)
+    group.add_argument('-n', '--noverify', action="store_true",
+                       help = 'Disable SSL certificate verification')
+    group = parser.add_argument_group("Program")
+    group.add_argument('-d', '--detach', action="store_true",
+                       help = 'Run process and return immediately without '
+                       'printing its output')
+    group.add_argument('cmd', help="Command to run")
+    group.add_argument('arg', nargs=argparse.REMAINDER,
+                       help="Arguments for command")
+    args = parser.parse_args()
+
+    client = HTTPClient(baseurl = args.url, verify_ssl = not args.noverify)
+
+    # Run command
+    pid = client.post("run/command", { "argv": [ args.cmd ] + args.arg })
+
+    # If we're detaching, just print the PID
+    if args.detach:
+        print pid
+        raise SystemExit(0)
+
+    # Otherwise, watch the log output, and kill the process when it's done
+    # or when this script terminates.
+    try:
+        while True:
+            s = client.get("process/status", { "pid": pid, "clear": 1 })
+            sys.stdout.write(s['log'])
+            sys.stdout.flush()
+            if not s['alive']:
+                break
+            time.sleep(1)
+    finally:
+        s = client.post("process/remove", { "pid": pid })
+
+    raise SystemExit(s['exitcode'])
+
+if __name__ == "__main__":
+    main()

setup.py

@@ -61,10 +61,10 @@ setup(name='nilmrun',
       long_description = "NILM Database Filter Runner",
       license = "Proprietary",
       author_email = 'jim@jtan.com',
-      install_requires = [ 'nilmdb >= 1.8.0',
-                           'nilmtools >= 1.2.2',
-                           'numpy',
-                           'scipy',
+      install_requires = [ 'nilmdb >= 1.9.5',
+                           'psutil >= 0.3.0',
+                           'cherrypy >= 3.2',
+                           'simplejson',
                            ],
       packages = [ 'nilmrun',
                    'nilmrun.scripts',
@@ -75,7 +75,9 @@ setup(name='nilmrun',
       entry_points = {
           'console_scripts': [
               'nilmrun-server = nilmrun.scripts.nilmrun_server:main',
-              'nilm-trainola = nilmrun.trainola:main',
+              'nilmrun-ps = nilmrun.scripts.ps:main',
+              'nilmrun-run = nilmrun.scripts.run:main',
+              'nilmrun-kill = nilmrun.scripts.kill:main',
               ],
           },
       zip_safe = False,

tests/test_nilmrun.py

@@ -30,6 +30,7 @@ import textwrap
 from testutil.helpers import *
 
 testurl = "http://localhost:32181/"
+#testurl = "http://bucket.mit.edu/nilmrun/"
 
 def setup_module():
     global test_server
@@ -49,10 +50,10 @@ class TestClient(object):
 
     def wait_kill(self, client, pid, timeout = 1):
         time.sleep(timeout)
-        status = client.get("/process/status", { "pid": pid })
+        status = client.get("process/status", { "pid": pid })
         if not status["alive"]:
             raise AssertionError("died before we could kill it")
-        status = client.post("/process/remove", { "pid": pid })
+        status = client.post("process/remove", { "pid": pid })
         if status["alive"]:
             raise AssertionError("didn't get killed")
         return status
@@ -61,181 +62,47 @@ class TestClient(object):
         start = time.time()
         status = None
         while (time.time() - start) < timeout:
-            status = client.get("/process/status", { "pid": pid })
+            status = client.get("process/status", { "pid": pid })
             if status["alive"] == False:
                 break
+            time.sleep(0.1)
         else:
             raise AssertionError("process " + str(pid) + " didn't die in " +
                                  str(timeout) + " seconds: " + repr(status))
         if remove:
-            status = client.post("/process/remove", { "pid": pid })
+            status = client.post("process/remove", { "pid": pid })
         return status
 
     def test_client_01_basic(self):
         client = HTTPClient(baseurl = testurl)
-        version = client.get("/version")
+        version = client.get("version")
         eq_(distutils.version.LooseVersion(version),
             distutils.version.LooseVersion(nilmrun.__version__))
 
-        in_("This is NilmRun", client.get("/"))
+        in_("This is NilmRun", client.get(""))
 
         with assert_raises(ClientError):
-            client.get("/favicon.ico")
+            client.get("favicon.ico")
 
     def test_client_02_manager(self):
         client = HTTPClient(baseurl = testurl)
 
-        eq_(client.get("/process/list"), [])
+        eq_(client.get("process/list"), [])
 
         with assert_raises(ClientError) as e:
-            client.get("/process/status", { "pid": 12345 })
+            client.get("process/status", { "pid": 12345 })
         in_("No such PID", str(e.exception))
         with assert_raises(ClientError):
-            client.get("/process/remove", { "pid": 12345 })
+            client.get("process/remove", { "pid": 12345 })
         in_("No such PID", str(e.exception))
 
-    def test_client_03_process_basic(self):
+    def test_client_03_run_command(self):
         client = HTTPClient(baseurl = testurl, post_json = True)
-
-        # start dummy filter
-        pid = client.post("/run/testfilter", { "data": 30 })
-        eq_(client.get("/process/list"), [pid])
-        time.sleep(1)
-
-        # Verify that status looks OK
-        status = client.get("/process/status", { "pid": pid, "clear": True })
-        for x in [ "pid", "alive", "exitcode", "name",
-                   "start_time", "parameters", "log" ]:
-            in_(x, status)
-        in_("dummy 0\ndummy 1\ndummy 2\ndummy 3\n", status["log"])
-        eq_(status["alive"], True)
-        eq_(status["exitcode"], None)
-
-        # Check that the log got cleared
-        status = client.get("/process/status", { "pid": pid })
-        nin_("dummy 0\ndummy 1\ndummy 2\ndummy 3\n", status["log"])
-
-        # See that it ended properly
-        status = self.wait_end(client, pid, remove = False)
-        in_("dummy 27\ndummy 28\ndummy 29\n", status["log"])
-        eq_(status["exitcode"], 0)
-
-        # Remove it
-        killstatus = client.post("/process/remove", { "pid": pid })
-        eq_(status, killstatus)
-        eq_(client.get("/process/list"), [])
-        with assert_raises(ClientError) as e:
-            client.post("/process/remove", { "pid": pid })
-        in_("No such PID", str(e.exception))
-
-    def test_client_04_process_terminate(self):
-        client = HTTPClient(baseurl = testurl, post_json = True)
-
-        # Trigger exception in filter
-        pid = client.post("/run/testfilter", { "data": -1 })
-        time.sleep(0.5)
-        status = client.get("/process/status", { "pid": pid })
-        eq_(status["alive"], False)
-        eq_(status["exitcode"], 1)
-        in_("Exception: test exception", status["log"])
-        client.post("/process/remove", { "pid": pid })
-
-        # Kill a running filter by removing it early
-        newpid = client.post("/run/testfilter", { "data": 50 })
-        ne_(newpid, pid)
-        time.sleep(0.5)
-        start = time.time()
-        status = client.post("/process/remove", { "pid": newpid })
-        elapsed = time.time() - start
-        # Should have died in slightly over 1 second
-        assert(0.5 < elapsed < 2)
-        eq_(status["alive"], False)
-        ne_(status["exitcode"], 0)
-
-        # No more
-        eq_(client.get("/process/list"), [])
-
-        # Try to remove a running filter that ignored SIGTERM
-        pid = client.post("/run/testfilter", { "data": 0 })
-        start = time.time()
-        status = client.post("/process/remove", { "pid": pid })
-        elapsed = time.time() - start
-        # Should have died in slightly over 2 seconds
-        assert(1.5 < elapsed < 3)
-        eq_(status["alive"], False)
-        ne_(status["exitcode"], 0)
-
-    @unittest.skip("trainola moving to nilmtools")
-    def test_client_05_trainola_simple(self):
-        client = HTTPClient(baseurl = testurl, post_json = True)
-        pid = client.post("/run/trainola", { "data": {} })
-        status = self.wait_end(client, pid, remove = False)
-        ne_(status["exitcode"], 0)
-        status = client.post("/process/remove", { "pid": pid })
-
-    @unittest.skip("needs a running nilmdb; trainola moved to nilmtools")
-    def test_client_06_trainola(self):
-        client = HTTPClient(baseurl = testurl, post_json = True)
-
-        data = { "url": "http://bucket.mit.edu/nilmdb",
-                 "stream": "/sharon/prep-a",
-                 "start": 1366111383280463,
-                 "end": 1366126163457797,
-                 "columns": [ { "name": "P1", "index": 0 },
-                              { "name": "Q1", "index": 1 },
-                              { "name": "P3", "index": 2 } ],
-                 "exemplars": [
-                     { "name": "Boiler Pump ON",
-                       "url": "http://bucket.mit.edu/nilmdb",
-                       "stream": "/sharon/prep-a",
-                       "start": 1366260494269078,
-                       "end": 1366260608185031,
-                       "columns": [ { "name": "P1", "index": 0 },
-                                    { "name": "Q1", "index": 1 }
-                                    ]
-                       },
-                     { "name": "Boiler Pump OFF",
-                       "url": "http://bucket.mit.edu/nilmdb",
-                       "stream": "/sharon/prep-a",
-                       "start": 1366260864215764,
-                       "end": 1366260870882998,
-                       "columns": [ { "name": "P1", "index": 0 },
-                                    { "name": "Q1", "index": 1 }
-                                    ]
-                       }
-                     ]
-                 }
-
-        # start trainola
-        pid = client.post("/run/trainola", { "data": data })
-
-        # wait for it to finish
-        for i in range(60):
-            time.sleep(1)
-            if i == 2:
-                status = client.get("/process/status", { "pid": pid,
-                                                         "clear": True })
-                in_("Loading stream data", status['log'])
-            elif i == 3:
-                status = client.get("/process/status", { "pid": pid })
-                nin_("Loading stream data", status['log'])
-            else:
-                status = client.get("/process/status", { "pid": pid })
-            if status["alive"] == False:
-                break
-        else:
-            client.post("/process/remove", {"pid": pid })
-            raise AssertionError("took too long")
-        if i < 3:
-            raise AssertionError("too fast?")
-
-    def test_client_07_run_command(self):
-        client = HTTPClient(baseurl = testurl, post_json = True)
-        eq_(client.get("/process/list"), [])
+        eq_(client.get("process/list"), [])
 
         def do(argv, kill):
-            pid = client.post("/run/command", { "argv": argv } )
-            eq_(client.get("/process/list"), [pid])
+            pid = client.post("run/command", { "argv": argv } )
+            eq_(client.get("process/list"), [pid])
             if kill:
                 return self.wait_kill(client, pid)
             return self.wait_end(client, pid)
@@ -251,20 +118,151 @@ class TestClient(object):
         eq_("3\n", status["log"])
 
         # Missing command
-        status = do(["/no-such-command-blah-blah"], False)
-        ne_(status["exitcode"], 0)
+        with assert_raises(ClientError) as e:
+            do(["/no-such-command-blah-blah"], False)
+        in_("No such file or directory", str(e.exception))
 
         # Kill a slow command
         status = do(["sleep", "60"], True)
         ne_(status["exitcode"], 0)
 
-    def test_client_08_run_code(self):
+    def _run_testfilter(self, client, args):
+        code = textwrap.dedent("""
+        import nilmrun.testfilter
+        import simplejson as json
+        import sys
+        nilmrun.testfilter.test(json.loads(sys.argv[1]))
+        """)
+        jsonargs = json.dumps(args)
+        return client.post("run/code", { "code": code, "args": [ jsonargs ] })
+
+    def test_client_04_process_basic(self):
         client = HTTPClient(baseurl = testurl, post_json = True)
-        eq_(client.get("/process/list"), [])
+
+        # start dummy filter
+        pid = self._run_testfilter(client, 30)
+        eq_(client.get("process/list"), [pid])
+        time.sleep(1)
+
+        # Verify that status looks OK
+        status = client.get("process/status", { "pid": pid, "clear": True })
+        for x in [ "pid", "alive", "exitcode", "start_time", "log" ]:
+            in_(x, status)
+        in_("dummy 0\ndummy 1\ndummy 2\ndummy 3\n", status["log"])
+        eq_(status["alive"], True)
+        eq_(status["exitcode"], None)
+
+        # Check that the log got cleared
+        status = client.get("process/status", { "pid": pid })
+        nin_("dummy 0\ndummy 1\ndummy 2\ndummy 3\n", status["log"])
+
+        # See that it ended properly
+        status = self.wait_end(client, pid, remove = False)
+        in_("dummy 27\ndummy 28\ndummy 29\n", status["log"])
+        eq_(status["exitcode"], 0)
+
+        # Remove it
+        killstatus = client.post("process/remove", { "pid": pid })
+        eq_(status, killstatus)
+        eq_(client.get("process/list"), [])
+        with assert_raises(ClientError) as e:
+            client.post("process/remove", { "pid": pid })
+        in_("No such PID", str(e.exception))
+
+    def test_client_05_process_terminate(self):
+        client = HTTPClient(baseurl = testurl, post_json = True)
+
+        # Trigger exception in filter
+        pid = self._run_testfilter(client, -1)
+        time.sleep(0.5)
+        status = client.get("process/status", { "pid": pid })
+        eq_(status["alive"], False)
+        eq_(status["exitcode"], 1)
+        in_("Exception: test exception", status["log"])
+        client.post("process/remove", { "pid": pid })
+
+        # Kill a running filter by removing it early
+        newpid = self._run_testfilter(client, 50)
+        ne_(newpid, pid)
+        time.sleep(0.5)
+        start = time.time()
+        status = client.post("process/remove", { "pid": newpid })
+        elapsed = time.time() - start
+        # Should have died in slightly over 1 second
+        assert(0.5 < elapsed < 2)
+        eq_(status["alive"], False)
+        ne_(status["exitcode"], 0)
+
+        # No more
+        eq_(client.get("process/list"), [])
+
+        # Try to remove a running filter that ignored SIGTERM
+        pid = self._run_testfilter(client, 0)
+        start = time.time()
+        status = client.post("process/remove", { "pid": pid })
+        elapsed = time.time() - start
+        # Should have died in slightly over 2 seconds
+        assert(1.5 < elapsed < 3)
+        eq_(status["alive"], False)
+        ne_(status["exitcode"], 0)
+
+    @unittest.skip("needs a running nilmdb; trainola moved to nilmtools")
+    def test_client_06_trainola(self):
+        client = HTTPClient(baseurl = testurl, post_json = True)
+        data = { "url": "http://bucket.mit.edu/nilmdb",
+                 "dest_stream": "/sharon/prep-a-matches",
+                 "stream": "/sharon/prep-a",
+                 "start": 1366111383280463,
+                 "end": 1366126163457797,
+                 "columns": [ { "name": "P1", "index": 0 },
+                              { "name": "Q1", "index": 1 },
+                              { "name": "P3", "index": 2 } ],
+                 "exemplars": [
+                     { "name": "Boiler Pump ON",
+                       "url": "http://bucket.mit.edu/nilmdb",
+                       "stream": "/sharon/prep-a",
+                       "start": 1366260494269078,
+                       "end": 1366260608185031,
+                       "dest_column": 0,
+                       "columns": [ { "name": "P1", "index": 0 },
+                                    { "name": "Q1", "index": 1 }
+                                    ]
+                       },
+                     { "name": "Boiler Pump OFF",
+                       "url": "http://bucket.mit.edu/nilmdb",
+                       "stream": "/sharon/prep-a",
+                       "start": 1366260864215764,
+                       "end": 1366260870882998,
+                       "dest_column": 1,
+                       "columns": [ { "name": "P1", "index": 0 },
+                                    { "name": "Q1", "index": 1 }
+                                    ]
+                       }
+                     ]
+                 }
+        pid = client.post("run/code", { "code": "import nilmtools.trainola\n" +
+                                         "nilmtools.trainola.main()",
+                                         "args": [ data ] })
+        while True:
+            status = client.get("process/status", { "pid": pid, "clear": 1 })
+            sys.stdout.write(status["log"])
+            sys.stdout.flush()
+            if status["alive"] == False:
+                break
+            time.sleep(0.1)
+        status = client.post("process/remove", { "pid": pid })
+        os._exit(int(status["exitcode"]))
+
+    def test_client_07_run_code(self):
+        client = HTTPClient(baseurl = testurl, post_json = True)
+        eq_(client.get("process/list"), [])
 
         def do(code, args, kill):
-            pid = client.post("/run/code", { "code": code, "args": args } )
-            eq_(client.get("/process/list"), [pid])
+            if args is not None:
+                pid = client.post("run/code", { "code": code, "args": args } )
+            else:
+                pid = client.post("run/code", { "code": code } )
+            eq_(client.get("process/list"), [pid])
             if kill:
                 return self.wait_kill(client, pid)
             return self.wait_end(client, pid)
@@ -295,20 +293,25 @@ class TestClient(object):
         foo(123)
         """)
         status = do(code, [], False)
+        cleaned_log = re.sub('File "[^"]*",', 'File "",', status["log"])
         eq_('Traceback (most recent call last):\n' +
-        '  File "<user-code>", line 4, in <module>\n' +
+        '  File "", line 4, in <module>\n' +
         '    foo(123)\n' +
-        '  File "<user-code>", line 3, in foo\n' +
+        '  File "", line 3, in foo\n' +
         '    raise Exception(arg)\n' +
-        'Exception: 123\n', status["log"])
+        'Exception: 123\n', cleaned_log)
         eq_(status["exitcode"], 1)
 
         # argument handling (strings come in as unicode)
         code = textwrap.dedent("""
         import sys
         print sys.argv[1].encode('ascii'), sys.argv[2]
+        sys.exit(0)  # also test raising SystemExit
         """)
-        status = do(code, ["hello", 123], False)
+        with assert_raises(ClientError) as e:
+	    do(code, ["hello", 123], False)
+        in_("400 Bad Request", str(e.exception))
+        status = do(code, ["hello", "123"], False)
         eq_(status["log"], "hello 123\n")
         eq_(status["exitcode"], 0)
 
@@ -322,3 +325,85 @@ class TestClient(object):
         status = do(code, [], True)
         eq_(status["log"], "hello\n")
         ne_(status["exitcode"], 0)
+
+        # default arguments are empty
+        code = textwrap.dedent("""
+        import sys
+        print 'args:', len(sys.argv[1:])
+        """)
+        status = do(code, None, False)
+        eq_(status["log"], "args: 0\n")
+        eq_(status["exitcode"], 0)
+
+    def test_client_08_bad_types(self):
+        client = HTTPClient(baseurl = testurl, post_json = True)
+
+        with assert_raises(ClientError) as e:
+            client.post("run/code", { "code": "asdf", "args": "qwer" })
+        in_("must be a list", str(e.exception))
+
+        with assert_raises(ClientError) as e:
+            client.post("run/command", { "argv": "asdf" })
+        in_("must be a list", str(e.exception))
+
+    def test_client_09_info(self):
+        client = HTTPClient(baseurl = testurl, post_json = True)
+
+        # start some processes
+        a = client.post("run/command", { "argv": ["sleep","60"] } )
+        b = client.post("run/command", { "argv": ["sh","-c","sleep 2;true"] } )
+        c = client.post("run/command", { "argv": ["sh","-c","burnP5;true"] } )
+        d = client.post("run/command", { "argv": ["burnP5" ] } )
+
+        info = client.get("process/info")
+        eq_(info["pids"][a]["procs"], 1)
+        eq_(info["pids"][b]["procs"], 2)
+        eq_(info["pids"][c]["procs"], 2)
+        eq_(info["pids"][d]["procs"], 1)
+        eq_(info["total"]["procs"], 6)
+        lt_(info["pids"][a]["cpu_percent"], 50)
+        lt_(20, info["pids"][c]["cpu_percent"])
+        lt_(80, info["system"]["cpu_percent"])
+
+        time.sleep(2)
+        info = client.get("process/info")
+        eq_(info["pids"][b]["procs"], 0)
+
+        # kill all processes
+        for pid in client.get("process/list"):
+            client.post("process/remove", { "pid": pid })
+
+    def test_client_10_unicode(self):
+        client = HTTPClient(baseurl = testurl, post_json = True)
+        eq_(client.get("process/list"), [])
+        def verify(cmd, result):
+            pid = client.post("run/command", { "argv": [ "sh", "-c", cmd ] })
+            eq_(client.get("process/list"), [pid])
+            status = self.wait_end(client, pid)
+            eq_(result, status["log"])
+
+        # Unicode should work
+        verify("echo -n hello", "hello")
+        verify(u"echo -n ☠", u"☠")
+        verify("echo -ne \\\\xe2\\\\x98\\\\xa0", u"☠")
+
+        # Programs that spit out invalid UTF-8 should get replacement
+        # markers
+        verify("echo -ne \\\\xae", u"\ufffd")
+
+    def test_client_11_atexit(self):
+        # Leave a directory and running process behind, for the atexit
+        # handler to clean up.  Here we trigger the atexit manually,
+        # since it's hard to trigger it as part of the test suite.
+        client = HTTPClient(baseurl = testurl, post_json = True)
+        code = textwrap.dedent("""
+        import time
+        time.sleep(10)
+        """)
+        client.post("run/code", { "code": code, "args": [ "hello"] })
+
+        # Trigger atexit function
+        test_server._manager._atexit()
+
+        # Ensure no processes exit
+        eq_(client.get("process/list"), [])