Better handling of JSON requests

Requires fallocate(2) support with FALLOC_FL_PUNCH_HOLE, as
well as a filesystem that supports it (in Linux 3.7,
tmpfs, btrfs, xfs, or ext4)

nilmdb/client/client.py

@@ -21,8 +21,12 @@ def extract_timestamp(line):
 class Client(object):
     """Main client interface to the Nilm database."""
 
-    def __init__(self, url):
+    def __init__(self, url, post_json = False):
-        self.http = nilmdb.client.httpclient.HTTPClient(url)
+        """Initialize client with given URL.  If post_json is true,
+        POST requests are sent with Content-Type 'application/json'
+        instead of the default 'x-www-form-urlencoded'."""
+        self.http = nilmdb.client.httpclient.HTTPClient(url, post_json)
+        self.post_json = post_json
 
     # __enter__/__exit__ allow this class to be a context manager
     def __enter__(self):
@@ -31,8 +35,11 @@ class Client(object):
     def __exit__(self, exc_type, exc_value, traceback):
         self.close()
 
-    def _json_param(self, data):
+    def _json_post_param(self, data):
         """Return compact json-encoded version of parameter"""
+        if self.post_json:
+            # If we're posting as JSON, we don't need to encode it further here
+            return data
         return json.dumps(data, separators=(',',':'))
 
     def close(self):
@@ -52,12 +59,14 @@ class Client(object):
         as a dictionary."""
         return self.http.get("dbinfo")
 
-    def stream_list(self, path = None, layout = None):
+    def stream_list(self, path = None, layout = None, extent = False):
         params = {}
         if path is not None:
             params["path"] = path
         if layout is not None:
             params["layout"] = layout
+        if extent:
+            params["extent"] = 1
         return self.http.get("stream/list", params)
 
     def stream_get_metadata(self, path, keys = None):
@@ -71,7 +80,7 @@ class Client(object):
         metadata."""
         params = {
             "path": path,
-            "data": self._json_param(data)
+            "data": self._json_post_param(data)
             }
         return self.http.post("stream/set_metadata", params)
 
@@ -79,7 +88,7 @@ class Client(object):
         """Update stream metadata from a dictionary"""
         params = {
             "path": path,
-            "data": self._json_param(data)
+            "data": self._json_post_param(data)
             }
         return self.http.post("stream/update_metadata", params)
 

nilmdb/client/httpclient.py

@@ -10,7 +10,7 @@ import requests
 
 class HTTPClient(object):
     """Class to manage and perform HTTP requests from the client"""
-    def __init__(self, baseurl = ""):
+    def __init__(self, baseurl = "", post_json = False):
         """If baseurl is supplied, all other functions that take
         a URL can be given a relative URL instead."""
         # Verify / clean up URL
@@ -26,6 +26,10 @@ class HTTPClient(object):
         # Saved response, so that tests can verify a few things.
         self._last_response = {}
 
+        # Whether to send application/json POST bodies (versus
+        # x-www-form-urlencoded)
+        self.post_json = post_json
+
     def _handle_error(self, url, code, body):
         # Default variables for exception.  We use the entire body as
         # the default message, in case we can't extract it from a JSON
@@ -57,12 +61,14 @@ class HTTPClient(object):
     def close(self):
         self.session.close()
 
-    def _do_req(self, method, url, query_data, body_data, stream):
+    def _do_req(self, method, url, query_data, body_data, stream, headers):
         url = urlparse.urljoin(self.baseurl, url)
         try:
             response = self.session.request(method, url,
                                             params = query_data,
-                                            data = body_data)
+                                            data = body_data,
+                                            stream = stream,
+                                            headers = headers)
         except requests.RequestException as e:
             raise ServerError(status = "502 Error", url = url,
                               message = str(e.message))
@@ -76,12 +82,13 @@ class HTTPClient(object):
             return (response, False)
 
     # Normal versions that return data directly
-    def _req(self, method, url, query = None, body = None):
+    def _req(self, method, url, query = None, body = None, headers = None):
         """
         Make a request and return the body data as a string or parsed
         JSON object, or raise an error if it contained an error.
         """
-        (response, isjson) = self._do_req(method, url, query, body, False)
+        (response, isjson) = self._do_req(method, url, query, body,
+                                          stream = False, headers = headers)
         if isjson:
             return json.loads(response.content)
         return response.content
@@ -92,20 +99,26 @@ class HTTPClient(object):
 
     def post(self, url, params = None):
         """Simple POST (parameters in body)"""
-        return self._req("POST", url, None, params)
+        if self.post_json:
+            return self._req("POST", url, None,
+                             json.dumps(params),
+                             { 'Content-type': 'application/json' })
+        else:
+            return self._req("POST", url, None, params)
 
     def put(self, url, data, params = None):
         """Simple PUT (parameters in URL, data in body)"""
         return self._req("PUT", url, params, data)
 
     # Generator versions that return data one line at a time.
-    def _req_gen(self, method, url, query = None, body = None):
+    def _req_gen(self, method, url, query = None, body = None, headers = None):
         """
         Make a request and return a generator that gives back strings
         or JSON decoded lines of the body data, or raise an error if
         it contained an eror.
         """
-        (response, isjson) = self._do_req(method, url, query, body, True)
+        (response, isjson) = self._do_req(method, url, query, body,
+                                          stream = True, headers = headers)
         for line in response.iter_lines():
             if isjson:
                 yield json.loads(line)

nilmdb/cmdline/cmdline.py

@@ -6,13 +6,14 @@ from nilmdb.utils import datetime_tz
 import nilmdb.utils.time
 
 import sys
+import os
 import argparse
 from argparse import ArgumentDefaultsHelpFormatter as def_form
 
 # Valid subcommands.  Defined in separate files just to break
 # things up -- they're still called with Cmdline as self.
-subcommands = [ "info", "create", "list", "metadata", "insert", "extract",
+subcommands = [ "help", "info", "create", "list", "metadata",
-                "remove", "destroy" ]
+                "insert", "extract", "remove", "destroy" ]
 
 # Import the subcommand modules
 subcmd_mods = {}
@@ -29,6 +30,8 @@ class Cmdline(object):
     def __init__(self, argv = None):
         self.argv = argv or sys.argv[1:]
         self.client = None
+        self.def_url = os.environ.get("NILMDB_URL", "http://localhost:12380")
+        self.subcmd = {}
 
     def arg_time(self, toparse):
         """Parse a time string argument"""
@@ -50,18 +53,17 @@ class Cmdline(object):
 
         group = self.parser.add_argument_group("Server")
         group.add_argument("-u", "--url", action="store",
-                           default="http://localhost:12380/",
+                           default=self.def_url,
                            help="NilmDB server URL (default: %(default)s)")
 
-        sub = self.parser.add_subparsers(title="Commands",
+        sub = self.parser.add_subparsers(
-                                         dest="command",
+            title="Commands", dest="command",
-                                         description="Specify --help after "
+            description="Use 'help command' or 'command --help' for more "
-                                         "the command for command-specific "
+            "details on a particular command.")
-                                         "options.")
 
         # Set up subcommands (defined in separate files)
         for cmd in subcommands:
-            subcmd_mods[cmd].setup(self, sub)
+            self.subcmd[cmd] = subcmd_mods[cmd].setup(self, sub)
 
     def die(self, formatstr, *args):
         fprintf(sys.stderr, formatstr + "\n", *args)
@@ -84,11 +86,13 @@ class Cmdline(object):
 
         self.client = nilmdb.Client(self.args.url)
 
-        # Make a test connection to make sure things work
+        # Make a test connection to make sure things work,
-        try:
+        # unless the particular command requests that we don't.
-            server_version = self.client.version()
+        if "no_test_connect" not in self.args:
-        except nilmdb.client.Error as e:
+            try:
-            self.die("error connecting to server: %s", str(e))
+                server_version = self.client.version()
+            except nilmdb.client.Error as e:
+                self.die("error connecting to server: %s", str(e))
 
         # Now dispatch client request to appropriate function.  Parser
         # should have ensured that we don't have any unknown commands

nilmdb/cmdline/create.py

@@ -26,6 +26,7 @@ Layout types are of the format: type_count
                        help="Path (in database) of new stream, e.g. /foo/bar")
     group.add_argument("layout",
                        help="Layout type for new stream, e.g. float32_8")
+    return cmd
 
 def cmd_create(self):
     """Create new stream"""

nilmdb/cmdline/destroy.py

@@ -16,6 +16,7 @@ def setup(self, sub):
     group = cmd.add_argument_group("Required arguments")
     group.add_argument("path",
                        help="Path of the stream to delete, e.g. /foo/bar")
+    return cmd
 
 def cmd_destroy(self):
     """Destroy stream"""

nilmdb/cmdline/extract.py

@@ -30,6 +30,7 @@ def setup(self, sub):
                        help="Show raw timestamps in annotated information")
     group.add_argument("-c", "--count", action="store_true",
                        help="Just output a count of matched data points")
+    return cmd
 
 def cmd_extract_verify(self):
     if self.args.start is not None and self.args.end is not None:

nilmdb/cmdline/help.py

@@ -0,0 +1,26 @@
+from nilmdb.utils.printf import *
+
+import argparse
+import sys
+
+def setup(self, sub):
+    cmd = sub.add_parser("help", help="Show detailed help for a command",
+                         description="""
+                         Show help for a command. 'help command' is
+                         the same as 'command --help'.
+                         """)
+    cmd.set_defaults(handler = cmd_help)
+    cmd.set_defaults(no_test_connect = True)
+    cmd.add_argument("command", nargs="?",
+                     help="Command to get help about")
+    cmd.add_argument("rest", nargs=argparse.REMAINDER,
+                     help=argparse.SUPPRESS)
+    return cmd
+
+def cmd_help(self):
+    if self.args.command in self.subcmd:
+        self.subcmd[self.args.command].print_help()
+    else:
+        self.parser.print_help()
+
+    return

nilmdb/cmdline/info.py

@@ -12,6 +12,7 @@ def setup(self, sub):
                          version.
                          """)
     cmd.set_defaults(handler = cmd_info)
+    return cmd
 
 def cmd_info(self):
     """Print info about the server"""

nilmdb/cmdline/insert.py

@@ -47,6 +47,7 @@ def setup(self, sub):
                        help="Path of stream, e.g. /foo/bar")
     group.add_argument("file", nargs="*", default=['-'],
                        help="File(s) to insert (default: - (stdin))")
+    return cmd
 
 def cmd_insert(self):
     # Find requested stream

nilmdb/cmdline/list.py

@@ -24,11 +24,13 @@ def setup(self, sub):
     group.add_argument("-l", "--layout", default="*",
                        help="Match only this stream layout")
 
+    group = cmd.add_argument_group("Interval extent")
+    group.add_argument("-E", "--extent", action="store_true",
+                       help="Show min/max timestamps in this stream")
+
     group = cmd.add_argument_group("Interval details")
     group.add_argument("-d", "--detail", action="store_true",
                        help="Show available data time intervals")
-    group.add_argument("-T", "--timestamp-raw", action="store_true",
-                       help="Show raw timestamps in time intervals")
     group.add_argument("-s", "--start",
                        metavar="TIME", type=self.arg_time,
                        help="Starting timestamp (free-form, inclusive)")
@@ -36,6 +38,12 @@ def setup(self, sub):
                        metavar="TIME", type=self.arg_time,
                        help="Ending timestamp (free-form, noninclusive)")
 
+    group = cmd.add_argument_group("Misc options")
+    group.add_argument("-T", "--timestamp-raw", action="store_true",
+                       help="Show raw timestamps in time intervals or extents")
+
+    return cmd
+
 def cmd_list_verify(self):
     # A hidden "path_positional" argument lets the user leave off the
     # "-p" when specifying the path.  Handle it here.
@@ -51,28 +59,38 @@ def cmd_list_verify(self):
         if self.args.start >= self.args.end:
             self.parser.error("start must precede end")
 
+    if self.args.start is not None or self.args.end is not None:
+        if not self.args.detail:
+            self.parser.error("--start and --end only make sense with --detail")
+
 def cmd_list(self):
     """List available streams"""
-    streams = self.client.stream_list()
+    streams = self.client.stream_list(extent = True)
 
     if self.args.timestamp_raw:
         time_string = repr
     else:
         time_string = nilmdb.utils.time.format_time
 
-    for (path, layout) in streams:
+    for (path, layout, extent_min, extent_max) in streams:
         if not (fnmatch.fnmatch(path, self.args.path) and
                 fnmatch.fnmatch(layout, self.args.layout)):
             continue
 
         printf("%s %s\n", path, layout)
-        if not self.args.detail:
-            continue
 
-        printed = False
+        if self.args.extent:
-        for (start, end) in self.client.stream_intervals(path, self.args.start,
+            if extent_min is None or extent_max is None:
-                                                         self.args.end):
+                printf("  extent: (no data)\n")
-            printf("  [ %s -> %s ]\n", time_string(start), time_string(end))
+            else:
-            printed = True
+                printf("  extent: %s -> %s\n",
-        if not printed:
+                       time_string(extent_min), time_string(extent_max))
-            printf("  (no intervals)\n")
+
+        if self.args.detail:
+            printed = False
+            for (start, end) in self.client.stream_intervals(
+                path, self.args.start, self.args.end):
+                printf("  [ %s -> %s ]\n", time_string(start), time_string(end))
+                printed = True
+            if not printed:
+                printf("  (no intervals)\n")

nilmdb/cmdline/metadata.py

@@ -26,6 +26,7 @@ def setup(self, sub):
     exc.add_argument("-u", "--update", nargs="+", metavar="key=value",
                      help="Update metadata using provided "
                      "key=value pairs")
+    return cmd
 
 def cmd_metadata(self):
     """Manipulate metadata"""

nilmdb/cmdline/remove.py

@@ -23,6 +23,7 @@ def setup(self, sub):
     group = cmd.add_argument_group("Output format")
     group.add_argument("-c", "--count", action="store_true",
                        help="Output number of data points removed")
+    return cmd
 
 def cmd_remove(self):
     try:

nilmdb/scripts/nilmdb_server.py

@@ -28,6 +28,9 @@ def main():
     group.add_argument('-n', '--nosync', help = 'Use asynchronous '
                        'commits for sqlite transactions',
                        action = 'store_true', default = False)
+    group.add_argument('-t', '--traceback',
+                       help = 'Provide tracebacks in client errors',
+                       action = 'store_true', default = False)
 
     group = parser.add_argument_group("Debug options")
     group.add_argument('-y', '--yappi', help = 'Run under yappi profiler and '
@@ -49,10 +52,12 @@ def main():
     server = nilmdb.server.Server(db,
                                   host = args.address,
                                   port = args.port,
-                                  embedded = embedded)
+                                  embedded = embedded,
+                                  force_traceback = args.traceback)
 
     # Print info
     if not args.quiet:
+        print "Version: %s" % nilmdb.__version__
         print "Database: %s" % (os.path.realpath(args.database))
         if args.address == '0.0.0.0' or args.address == '::':
             host = socket.getfqdn()

nilmdb/server/bulkdata.py

@@ -221,9 +221,11 @@ class File(object):
         # An optimized verison of append, to avoid flushing the file
         # and resizing the mmap after each data point.
         try:
+            rows = []
             for i in xrange(count):
                 row = dataiter.next()
-                self._f.write(packer(*row))
+                rows.append(packer(*row))
+            self._f.write("".join(rows))
         finally:
             self._f.flush()
             self.size = self._f.tell()
@@ -408,8 +410,16 @@ class Table(object):
 
     def _remove_rows(self, subdir, filename, start, stop):
         """Helper to mark specific rows as being removed from a
-        file, and potentially removing or truncating the file itself."""
+        file, and potentially remove or truncate the file itself."""
-        # Import an existing list of deleted rows for this file
+        # Close potentially open file in file_open LRU cache
+        self.file_open.cache_remove(self, subdir, filename)
+
+        # We keep a file like 0000.removed that contains a list of
+        # which rows have been "removed".  Note that we never have to
+        # remove entries from this list, because we never decrease
+        # self.nrows, and so we will never overwrite those locations in the
+        # file.  Only when the list covers the entire extent of the
+        # file will that file be removed.
         datafile = os.path.join(self.root, subdir, filename)
         cachefile = datafile + ".removed"
         try:
@@ -463,6 +473,14 @@ class Table(object):
             except:
                 pass
         else:
+            # File needs to stick around.  This means we can get
+            # degenerate cases where we have large files containing as
+            # little as one row.  Try to punch a hole in the file,
+            # so that this region doesn't take up filesystem space.
+            offset = start * self.packer.size
+            count = (stop - start) * self.packer.size
+            nilmdb.utils.fallocate.punch_hole(datafile, offset, count)
+
             # Update cache.  Try to do it atomically.
             nilmdb.utils.atomic.replace_file(cachefile,
                                              pickle.dumps(merged, 2))

nilmdb/server/nilmdb.py

@@ -269,28 +269,39 @@ class NilmDB(object):
 
         return
 
-    def stream_list(self, path = None, layout = None):
+    def stream_list(self, path = None, layout = None, extent = False):
-        """Return list of [path, layout] lists of all streams
+        """Return list of lists of all streams in the database.
-        in the database.
 
         If path is specified, include only streams with a path that
         matches the given string.
 
         If layout is specified, include only streams with a layout
         that matches the given string.
-        """
-        where = "WHERE 1=1"
-        params = ()
-        if layout:
-            where += " AND layout=?"
-            params += (layout,)
-        if path:
-            where += " AND path=?"
-            params += (path,)
-        result = self.con.execute("SELECT path, layout "
-                                  "FROM streams " + where, params).fetchall()
 
-        return sorted(list(x) for x in result)
+        If extent = False, returns a list of lists containing
+        the path and layout: [ path, layout ]
+
+        If extent = True, returns a list of lists containing the
+        path, layout, and min/max extent of the data:
+        [ path, layout, extent_min, extent_max ]
+        """
+        params = ()
+        query = "SELECT streams.path, streams.layout"
+        if extent:
+            query += ", min(ranges.start_time), max(ranges.end_time)"
+        query += " FROM streams"
+        if extent:
+            query += " LEFT JOIN ranges ON streams.id = ranges.stream_id"
+        query += " WHERE 1=1"
+        if layout is not None:
+            query += " AND streams.layout=?"
+            params += (layout,)
+        if path is not None:
+            query += " AND streams.path=?"
+            params += (path,)
+        query += " GROUP BY streams.id ORDER BY streams.path"
+        result = self.con.execute(query, params).fetchall()
+        return [ list(x) for x in result ]
 
     def stream_intervals(self, path, start = None, end = None):
         """

nilmdb/server/server.py

@@ -71,16 +71,59 @@ def exception_to_httperror(*expected):
     # care of that.
     return decorator.decorator(wrapper)
 
-# Custom Cherrypy tools
+# Custom CherryPy tools
-def allow_methods(methods):
+
-    method = cherrypy.request.method.upper()
+def CORS_allow(methods):
-    if method not in methods:
+    """This does several things:
-        if method in cherrypy.request.methods_with_bodies:
+
-            cherrypy.request.body.read()
+    Handles CORS preflight requests.
-        allowed = ', '.join(methods)
+    Adds Allow: header to all requests.
-        cherrypy.response.headers['Allow'] = allowed
+    Raise 405 if request.method not in method.
-        raise cherrypy.HTTPError(405, method + " not allowed; use " + allowed)
+
-cherrypy.tools.allow_methods = cherrypy.Tool('before_handler', allow_methods)
+    It is similar to cherrypy.tools.allow, with the CORS stuff added.
+    """
+    request = cherrypy.request.headers
+    response = cherrypy.response.headers
+
+    if not isinstance(methods, (tuple, list)): # pragma: no cover
+        methods = [ methods ]
+    methods = [ m.upper() for m in methods if m ]
+    if not methods: # pragma: no cover
+        methods = [ 'GET', 'HEAD' ]
+    elif 'GET' in methods and 'HEAD' not in methods: # pragma: no cover
+        methods.append('HEAD')
+    response['Allow'] = ', '.join(methods)
+
+    # Allow all origins
+    if 'Origin' in request:
+        response['Access-Control-Allow-Origin'] = request['Origin']
+
+    # If it's a CORS request, send response.
+    request_method = request.get("Access-Control-Request-Method", None)
+    request_headers = request.get("Access-Control-Request-Headers", None)
+    if (cherrypy.request.method == "OPTIONS" and
+        request_method and request_headers):
+        response['Access-Control-Allow-Headers'] = request_headers
+        response['Access-Control-Allow-Methods'] = ', '.join(methods)
+        # Try to stop further processing and return a 200 OK
+        cherrypy.response.status = "200 OK"
+        cherrypy.response.body = ""
+        cherrypy.request.handler = lambda: ""
+        return
+
+    # Reject methods that were not explicitly allowed
+    if cherrypy.request.method not in methods:
+        raise cherrypy.HTTPError(405)
+
+cherrypy.tools.CORS_allow = cherrypy.Tool('on_start_resource', CORS_allow)
+
+# Helper for json_in tool to process JSON data into normal request
+# parameters.
+def json_to_request_params(body):
+    cherrypy.lib.jsontools.json_processor(body)
+    if not isinstance(cherrypy.request.json, dict):
+        raise cherrypy.HTTPError(415)
+    cherrypy.request.params.update(cherrypy.request.json)
 
 # CherryPy apps
 class Root(NilmApp):
@@ -121,20 +164,29 @@ class Stream(NilmApp):
 
     # /stream/list
     # /stream/list?layout=PrepData
-    # /stream/list?path=/newton/prep
+    # /stream/list?path=/newton/prep&extent=1
     @cherrypy.expose
     @cherrypy.tools.json_out()
-    def list(self, path = None, layout = None):
+    def list(self, path = None, layout = None, extent = None):
         """List all streams in the database.  With optional path or
         layout parameter, just list streams that match the given path
-        or layout"""
+        or layout.
-        return self.db.stream_list(path, layout)
+
+        If extent is not given, returns a list of lists containing
+        the path and layout: [ path, layout ]
+
+        If extent is provided, returns a list of lists containing the
+        path, layout, and min/max extent of the data:
+        [ path, layout, extent_min, extent_max ]
+        """
+        return self.db.stream_list(path, layout, bool(extent))
 
     # /stream/create?path=/newton/prep&layout=PrepData
     @cherrypy.expose
+    @cherrypy.tools.json_in()
     @cherrypy.tools.json_out()
     @exception_to_httperror(NilmDBError, ValueError)
-    @cherrypy.tools.allow_methods(methods = ["POST"])
+    @cherrypy.tools.CORS_allow(methods = ["POST"])
     def create(self, path, layout):
         """Create a new stream in the database.  Provide path
         and one of the nilmdb.layout.layouts keys.
@@ -143,9 +195,10 @@ class Stream(NilmApp):
 
     # /stream/destroy?path=/newton/prep
     @cherrypy.expose
+    @cherrypy.tools.json_in()
     @cherrypy.tools.json_out()
     @exception_to_httperror(NilmDBError)
-    @cherrypy.tools.allow_methods(methods = ["POST"])
+    @cherrypy.tools.CORS_allow(methods = ["POST"])
     def destroy(self, path):
         """Delete a stream and its associated data."""
         return self.db.stream_destroy(path)
@@ -176,31 +229,41 @@ class Stream(NilmApp):
 
     # /stream/set_metadata?path=/newton/prep&data=<json>
     @cherrypy.expose
+    @cherrypy.tools.json_in()
     @cherrypy.tools.json_out()
-    @exception_to_httperror(NilmDBError, LookupError, TypeError)
+    @exception_to_httperror(NilmDBError, LookupError)
-    @cherrypy.tools.allow_methods(methods = ["POST"])
+    @cherrypy.tools.CORS_allow(methods = ["POST"])
     def set_metadata(self, path, data):
-        """Set metadata for the named stream, replacing any
+        """Set metadata for the named stream, replacing any existing
-        existing metadata.  Data should be a json-encoded
+        metadata.  Data can be json-encoded or a plain dictionary (if
-        dictionary"""
+        it was sent as application/json to begin with)"""
-        data_dict = json.loads(data)
+        if not isinstance(data, dict):
-        self.db.stream_set_metadata(path, data_dict)
+            try:
+                data = dict(json.loads(data))
+            except TypeError as e:
+                raise NilmDBError("can't parse 'data' parameter: " + e.message)
+        self.db.stream_set_metadata(path, data)
 
     # /stream/update_metadata?path=/newton/prep&data=<json>
     @cherrypy.expose
+    @cherrypy.tools.json_in()
     @cherrypy.tools.json_out()
-    @exception_to_httperror(NilmDBError, LookupError, TypeError)
+    @exception_to_httperror(NilmDBError, LookupError)
-    @cherrypy.tools.allow_methods(methods = ["POST"])
+    @cherrypy.tools.CORS_allow(methods = ["POST"])
     def update_metadata(self, path, data):
         """Update metadata for the named stream.  Data
         should be a json-encoded dictionary"""
-        data_dict = json.loads(data)
+        if not isinstance(data, dict):
-        self.db.stream_update_metadata(path, data_dict)
+            try:
+                data = dict(json.loads(data))
+            except TypeError as e:
+                raise NilmDBError("can't parse 'data' parameter: " + e.message)
+        self.db.stream_update_metadata(path, data)
 
     # /stream/insert?path=/newton/prep
     @cherrypy.expose
     @cherrypy.tools.json_out()
-    @cherrypy.tools.allow_methods(methods = ["PUT"])
+    @cherrypy.tools.CORS_allow(methods = ["PUT"])
     def insert(self, path, start, end):
         """
         Insert new data into the database.  Provide textual data
@@ -254,9 +317,10 @@ class Stream(NilmApp):
     # /stream/remove?path=/newton/prep
     # /stream/remove?path=/newton/prep&start=1234567890.0&end=1234567899.0
     @cherrypy.expose
+    @cherrypy.tools.json_in()
     @cherrypy.tools.json_out()
     @exception_to_httperror(NilmDBError)
-    @cherrypy.tools.allow_methods(methods = ["POST"])
+    @cherrypy.tools.CORS_allow(methods = ["POST"])
     def remove(self, path, start = None, end = None):
         """
         Remove data from the backend database.  Removes all data in
@@ -411,16 +475,20 @@ class Server(object):
             'error_page.default': self.json_error_page,
             })
 
-        # Send a permissive Access-Control-Allow-Origin (CORS) header
+        # Some default headers to just help identify that things are working
-        # with all responses so that browsers can send cross-domain
+        app_config.update({ 'response.headers.X-Jim-Is-Awesome': 'yeah' })
-        # requests to this server.
-        app_config.update({ 'response.headers.Access-Control-Allow-Origin':
-                            '*' })
 
-        # Only allow GET and HEAD by default.  Individual handlers
+        # Set up Cross-Origin Resource Sharing (CORS) handler so we
-        # can override.
+        # can correctly respond to browsers' CORS preflight requests.
-        app_config.update({ 'tools.allow_methods.on': True,
+        # This also limits verbs to GET and HEAD by default.
-                            'tools.allow_methods.methods': ['GET', 'HEAD'] })
+        app_config.update({ 'tools.CORS_allow.on': True,
+                            'tools.CORS_allow.methods': ['GET', 'HEAD'] })
+
+        # Configure the 'json_in' tool to also allow other content-types
+        # (like x-www-form-urlencoded), and to treat JSON as a dict that
+        # fills requests.param.
+        app_config.update({ 'tools.json_in.force': False,
+                            'tools.json_in.processor': json_to_request_params })
 
         # Send tracebacks in error responses.  They're hidden by the
         # error_page function for client errors (code 400-499).

nilmdb/utils/__init__.py

@@ -8,3 +8,4 @@ from nilmdb.utils.diskusage import du, human_size
 from nilmdb.utils.mustclose import must_close
 from nilmdb.utils import atomic
 import nilmdb.utils.threadsafety
+import nilmdb.utils.fallocate

nilmdb/utils/fallocate.py

@@ -0,0 +1,49 @@
+# Implementation of hole punching via fallocate, if the OS
+# and filesystem support it.
+
+try:
+    import os
+    import ctypes
+    import ctypes.util
+
+    def make_fallocate():
+        libc_name = ctypes.util.find_library('c')
+        libc = ctypes.CDLL(libc_name, use_errno=True)
+
+        _fallocate = libc.fallocate
+        _fallocate.restype = ctypes.c_int
+        _fallocate.argtypes = [ ctypes.c_int, ctypes.c_int,
+                                ctypes.c_int64, ctypes.c_int64 ]
+
+        del libc
+        del libc_name
+
+        def fallocate(fd, mode, offset, len_):
+            res = _fallocate(fd, mode, offset, len_)
+            if res != 0: # pragma: no cover
+                errno = ctypes.get_errno()
+                raise IOError(errno, os.strerror(errno))
+        return fallocate
+
+    fallocate = make_fallocate()
+    del make_fallocate
+except Exception: # pragma: no cover
+    fallocate = None
+
+FALLOC_FL_KEEP_SIZE = 0x01
+FALLOC_FL_PUNCH_HOLE = 0x02
+
+def punch_hole(filename, offset, length, ignore_errors = True):
+    """Punch a hole in the file.  This isn't well supported, so errors
+    are ignored by default."""
+    try:
+        if fallocate is None: # pragma: no cover
+            raise IOError("fallocate not available")
+        with open(filename, "r+") as f:
+            fallocate(f.fileno(),
+                      FALLOC_FL_KEEP_SIZE | FALLOC_FL_PUNCH_HOLE,
+                      offset, length)
+    except IOError: # pragma: no cover
+        if ignore_errors:
+            return
+        raise

tests/test_client.py

@@ -25,7 +25,7 @@ import re
 from testutil.helpers import *
 
 testdb = "tests/client-testdb"
-testurl = "http://localhost:12380/"
+testurl = "http://localhost:32180/"
 
 def setup_module():
     global test_server, test_db
@@ -35,7 +35,7 @@ def setup_module():
     # Start web app on a custom port
     test_db = nilmdb.utils.serializer_proxy(nilmdb.NilmDB)(testdb, sync = False)
     test_server = nilmdb.Server(test_db, host = "127.0.0.1",
-                                port = 12380, stoppable = False,
+                                port = 32180, stoppable = False,
                                 fast_shutdown = True,
                                 force_traceback = False)
     test_server.start(blocking = False)
@@ -55,20 +55,14 @@ class TestClient(object):
             client.version()
         client.close()
 
-        # Trigger same error with a PUT request
-        client = nilmdb.Client(url = "http://localhost:1/")
-        with assert_raises(nilmdb.client.ServerError):
-            client.version()
-        client.close()
-
         # Then a fake URL on a real host
-        client = nilmdb.Client(url = "http://localhost:12380/fake/")
+        client = nilmdb.Client(url = "http://localhost:32180/fake/")
         with assert_raises(nilmdb.client.ClientError):
             client.version()
         client.close()
 
         # Now a real URL with no http:// prefix
-        client = nilmdb.Client(url = "localhost:12380")
+        client = nilmdb.Client(url = "localhost:32180")
         version = client.version()
         client.close()
 
@@ -360,52 +354,48 @@ class TestClient(object):
             raise AssertionError("/stream/extract is not text/plain:\n" +
                                  headers())
 
-        # Make sure Access-Control-Allow-Origin gets set
-        if "access-control-allow-origin: " not in headers():
-            raise AssertionError("No Access-Control-Allow-Origin (CORS) "
-                                 "header in /stream/extract response:\n" +
-                                 headers())
-
         client.close()
 
     def test_client_08_unicode(self):
-        # Basic Unicode tests
+        # Try both with and without posting JSON
-        client = nilmdb.Client(url = testurl)
+        for post_json in (False, True):
+            # Basic Unicode tests
+            client = nilmdb.Client(url = testurl, post_json = post_json)
 
-        # Delete streams that exist
+            # Delete streams that exist
-        for stream in client.stream_list():
+            for stream in client.stream_list():
-            client.stream_destroy(stream[0])
+                client.stream_destroy(stream[0])
 
-        # Database is empty
+            # Database is empty
-        eq_(client.stream_list(), [])
+            eq_(client.stream_list(), [])
 
-        # Create Unicode stream, match it
+            # Create Unicode stream, match it
-        raw = [ u"/düsseldorf/raw", u"uint16_6" ]
+            raw = [ u"/düsseldorf/raw", u"uint16_6" ]
-        prep = [ u"/düsseldorf/prep", u"uint16_6" ]
+            prep = [ u"/düsseldorf/prep", u"uint16_6" ]
-        client.stream_create(*raw)
+            client.stream_create(*raw)
-        eq_(client.stream_list(), [raw])
+            eq_(client.stream_list(), [raw])
-        eq_(client.stream_list(layout=raw[1]), [raw])
+            eq_(client.stream_list(layout=raw[1]), [raw])
-        eq_(client.stream_list(path=raw[0]), [raw])
+            eq_(client.stream_list(path=raw[0]), [raw])
-        client.stream_create(*prep)
+            client.stream_create(*prep)
-        eq_(client.stream_list(), [prep, raw])
+            eq_(client.stream_list(), [prep, raw])
 
-        # Set / get metadata with Unicode keys and values
+            # Set / get metadata with Unicode keys and values
-        eq_(client.stream_get_metadata(raw[0]), {})
+            eq_(client.stream_get_metadata(raw[0]), {})
-        eq_(client.stream_get_metadata(prep[0]), {})
+            eq_(client.stream_get_metadata(prep[0]), {})
-        meta1 = { u"alpha": u"α",
+            meta1 = { u"alpha": u"α",
-                  u"β": u"beta" }
+                      u"β": u"beta" }
-        meta2 = { u"alpha": u"α" }
+            meta2 = { u"alpha": u"α" }
-        meta3 = { u"β": u"beta" }
+            meta3 = { u"β": u"beta" }
-        client.stream_set_metadata(prep[0], meta1)
+            client.stream_set_metadata(prep[0], meta1)
-        client.stream_update_metadata(prep[0], {})
+            client.stream_update_metadata(prep[0], {})
-        client.stream_update_metadata(raw[0], meta2)
+            client.stream_update_metadata(raw[0], meta2)
-        client.stream_update_metadata(raw[0], meta3)
+            client.stream_update_metadata(raw[0], meta3)
-        eq_(client.stream_get_metadata(prep[0]), meta1)
+            eq_(client.stream_get_metadata(prep[0]), meta1)
-        eq_(client.stream_get_metadata(raw[0]), meta1)
+            eq_(client.stream_get_metadata(raw[0]), meta1)
-        eq_(client.stream_get_metadata(raw[0], [ "alpha" ]), meta2)
+            eq_(client.stream_get_metadata(raw[0], [ "alpha" ]), meta2)
-        eq_(client.stream_get_metadata(raw[0], [ "alpha", "β" ]), meta1)
+            eq_(client.stream_get_metadata(raw[0], [ "alpha", "β" ]), meta1)
 
-        client.close()
+            client.close()
 
     def test_client_09_closing(self):
         # Make sure we actually close sockets correctly.  New
@@ -584,7 +574,7 @@ class TestClient(object):
             def connections():
                 try:
                     poolmanager = c.http._last_response.connection.poolmanager
-                    pool = poolmanager.pools[('http','localhost',12380)]
+                    pool = poolmanager.pools[('http','localhost',32180)]
                     return (pool.num_connections, pool.num_requests)
                 except:
                     raise SkipTest("can't get connection info")

tests/test_cmdline.py

@@ -11,12 +11,7 @@ from nose.tools import assert_raises
 import itertools
 import os
 import re
-import shutil
 import sys
-import threading
-import urllib2
-from urllib2 import urlopen, HTTPError
-import Queue
 import StringIO
 import shlex
 
@@ -32,7 +27,7 @@ def server_start(max_results = None, bulkdata_args = {}):
         max_results = max_results,
         bulkdata_args = bulkdata_args)
     test_server = nilmdb.Server(test_db, host = "127.0.0.1",
-                                port = 12380, stoppable = False,
+                                port = 32180, stoppable = False,
                                 fast_shutdown = True,
                                 force_traceback = False)
     test_server.start(blocking = False)
@@ -64,6 +59,7 @@ class TestCmdline(object):
         passing the given input.  Returns a tuple with the output and
         exit code"""
         # printf("TZ=UTC ./nilmtool.py %s\n", arg_string)
+        os.environ['NILMDB_URL'] = "http://localhost:32180/"
         class stdio_wrapper:
             def __init__(self, stdin, stdout, stderr):
                 self.io = (stdin, stdout, stderr)
@@ -174,7 +170,7 @@ class TestCmdline(object):
         self.fail("-u localhost:1 info")
         self.contain("error connecting to server")
 
-        self.ok("-u localhost:12380 info")
+        self.ok("-u localhost:32180 info")
         self.ok("info")
 
         # Duplicated arguments should fail, but this isn't implemented
@@ -192,6 +188,20 @@ class TestCmdline(object):
             self.fail("extract --start 2000-01-01 --start 2001-01-02")
             self.contain("duplicated argument")
 
+        # Verify that "help command" and "command --help" are identical
+        # for all commands.
+        self.fail("")
+        m = re.search(r"{(.*)}", self.captured)
+        for command in [""] + m.group(1).split(','):
+            self.ok(command + " --help")
+            cap1 = self.captured
+            self.ok("help " + command)
+            cap2 = self.captured
+            self.ok("help " + command + " asdf --url --zxcv -")
+            cap3 = self.captured
+            eq_(cap1, cap2)
+            eq_(cap2, cap3)
+
     def test_02_parsetime(self):
         os.environ['TZ'] = "America/New_York"
         test = datetime_tz.datetime_tz.now()
@@ -210,7 +220,7 @@ class TestCmdline(object):
 
     def test_03_info(self):
         self.ok("info")
-        self.contain("Server URL: http://localhost:12380/")
+        self.contain("Server URL: http://localhost:32180/")
         self.contain("Client version: " + nilmdb.__version__)
         self.contain("Server version: " + test_server.version)
         self.contain("Server database path")
@@ -418,7 +428,7 @@ class TestCmdline(object):
         # bad start time
         self.fail("insert --rate 120 --start 'whatever' /newton/prep /dev/null")
 
-    def test_07_detail(self):
+    def test_07_detail_extent(self):
         # Just count the number of lines, it's probably fine
         self.ok("list --detail")
         lines_(self.captured, 8)
@@ -463,6 +473,18 @@ class TestCmdline(object):
         lines_(self.captured, 2)
         self.contain("[ 1332497115.612 -> 1332497159.991668 ]")
 
+        # Check --extent output
+        self.ok("list --extent")
+        lines_(self.captured, 6)
+
+        self.ok("list -E -T")
+        self.contain("  extent: 1332496800 -> 1332497159.991668")
+        self.contain("  extent: (no data)")
+
+        # Misc
+        self.fail("list --extent --start='23 Mar 2012 10:05:15.50'")
+        self.contain("--start and --end only make sense with --detail")
+
     def test_08_extract(self):
         # nonexistent stream
         self.fail("extract /no/such/foo --start 2000-01-01 --end 2020-01-01")

tests/test_layout.py

@@ -9,14 +9,7 @@ from nose.tools import assert_raises
 import distutils.version
 import itertools
 import os
-import shutil
 import sys
-import cherrypy
-import threading
-import urllib2
-from urllib2 import urlopen, HTTPError
-import Queue
-import cStringIO
 import random
 import unittest
 

tests/test_nilmdb.py

@@ -6,15 +6,13 @@ import distutils.version
 import simplejson as json
 import itertools
 import os
-import shutil
 import sys
-import cherrypy
 import threading
 import urllib2
 from urllib2 import urlopen, HTTPError
-import Queue
 import cStringIO
 import time
+import requests
 
 from nilmdb.utils import serializer_proxy
 
@@ -119,7 +117,7 @@ class TestBlockingServer(object):
 
         # Start web app on a custom port
         self.server = nilmdb.Server(self.db, host = "127.0.0.1",
-                                    port = 12380, stoppable = True)
+                                    port = 32180, stoppable = True)
 
         # Run it
         event = threading.Event()
@@ -131,13 +129,13 @@ class TestBlockingServer(object):
             raise AssertionError("server didn't start in 10 seconds")
 
         # Send request to exit.
-        req = urlopen("http://127.0.0.1:12380/exit/", timeout = 1)
+        req = urlopen("http://127.0.0.1:32180/exit/", timeout = 1)
 
         # Wait for it
         thread.join()
 
 def geturl(path):
-    req = urlopen("http://127.0.0.1:12380" + path, timeout = 10)
+    req = urlopen("http://127.0.0.1:32180" + path, timeout = 10)
     return req.read()
 
 def getjson(path):
@@ -149,7 +147,7 @@ class TestServer(object):
         # Start web app on a custom port
         self.db = serializer_proxy(nilmdb.NilmDB)(testdb, sync=False)
         self.server = nilmdb.Server(self.db, host = "127.0.0.1",
-                                    port = 12380, stoppable = False)
+                                    port = 32180, stoppable = False)
         self.server.start(blocking = False)
 
     def tearDown(self):
@@ -208,3 +206,51 @@ class TestServer(object):
         data = getjson("/stream/get_metadata?path=/newton/prep"
                        "&key=foo")
         eq_(data, {'foo': None})
+
+    def test_cors_headers(self):
+        # Test that CORS headers are being set correctly
+
+        # Normal GET should send simple response
+        url = "http://127.0.0.1:32180/stream/list"
+        r = requests.get(url, headers = { "Origin": "http://google.com/" })
+        eq_(r.status_code, 200)
+        if "access-control-allow-origin" not in r.headers:
+            raise AssertionError("No Access-Control-Allow-Origin (CORS) "
+                                 "header in response:\n", r.headers)
+        eq_(r.headers["access-control-allow-origin"], "http://google.com/")
+
+        # OPTIONS without CORS preflight headers should result in 405
+        r = requests.options(url, headers = {
+            "Origin": "http://google.com/",
+            })
+        eq_(r.status_code, 405)
+
+        # OPTIONS with preflight headers should give preflight response
+        r = requests.options(url, headers = {
+            "Origin": "http://google.com/",
+            "Access-Control-Request-Method": "POST",
+            "Access-Control-Request-Headers": "X-Custom",
+            })
+        eq_(r.status_code, 200)
+        if "access-control-allow-origin" not in r.headers:
+            raise AssertionError("No Access-Control-Allow-Origin (CORS) "
+                                 "header in response:\n", r.headers)
+        eq_(r.headers["access-control-allow-methods"], "GET, HEAD")
+        eq_(r.headers["access-control-allow-headers"], "X-Custom")
+
+    def test_post_bodies(self):
+        # Test JSON post bodies
+        r = requests.post("http://127.0.0.1:32180/stream/set_metadata",
+                          headers = { "Content-Type": "application/json" },
+                          data = '{"hello": 1}')
+        eq_(r.status_code, 404) # wrong parameters
+
+        r = requests.post("http://127.0.0.1:32180/stream/set_metadata",
+                          headers = { "Content-Type": "application/json" },
+                          data = '["hello"]')
+        eq_(r.status_code, 415) # not a dict
+
+        r = requests.post("http://127.0.0.1:32180/stream/set_metadata",
+                          headers = { "Content-Type": "application/json" },
+                          data = '[hello]')
+        eq_(r.status_code, 400) # badly formatted JSON