Fix bug in nilmdb.stream_remove with max_removals

Instead of returning a single number at the end of N nilmdb calls, we
now use a generator that returns one line of text every time there's a
new count of rows removed.  This ensures that the connection will stay
alive for very long removals.

nilmdb/client/client.py

@@ -122,7 +122,10 @@ class Client(object):
             params["start"] = timestamp_to_string(start)
         if end is not None:
             params["end"] = timestamp_to_string(end)
-        return self.http.post("stream/remove", params)
+        total = 0
+        for count in self.http.post_gen("stream/remove", params):
+            total += int(count)
+        return total
 
     @contextlib.contextmanager
     def stream_insert_context(self, path, start = None, end = None):

nilmdb/client/httpclient.py

@@ -137,5 +137,14 @@ class HTTPClient(object):
         """Simple GET (parameters in URL) returning a generator"""
         return self._req_gen("GET", url, params, binary = binary)
 
+    def post_gen(self, url, params = None):
+        """Simple POST (parameters in body) returning a generator"""
+        if self.post_json:
+            return self._req_gen("POST", url, None,
+                                 json.dumps(params),
+                                 { 'Content-type': 'application/json' })
+        else:
+            return self._req_gen("POST", url, None, params)
+
     # Not much use for a POST or PUT generator, since they don't
     # return much data.

nilmdb/client/numpyclient.py

@@ -162,9 +162,12 @@ class StreamInserterNumpy(nilmdb.client.client.StreamInserter):
         elif array.ndim == 2:
             # Convert to structured array
             sarray = numpy.zeros(array.shape[0], dtype=self._dtype)
-            sarray['timestamp'] = array[:,0]
-            # Need the squeeze in case sarray['data'] is 1 dimensional
-            sarray['data'] = numpy.squeeze(array[:,1:])
+            try:
+                sarray['timestamp'] = array[:,0]
+                # Need the squeeze in case sarray['data'] is 1 dimensional
+                sarray['data'] = numpy.squeeze(array[:,1:])
+            except (IndexError, ValueError):
+                raise ValueError("wrong number of fields for this data type")
             array = sarray
         else:
             raise ValueError("wrong number of dimensions in array")

nilmdb/cmdline/cmdline.py

@@ -81,6 +81,12 @@ class Cmdline(object):
 
     def __init__(self, argv = None):
         self.argv = argv or sys.argv[1:]
+        try:
+            # Assume command line arguments are encoded with stdin's encoding,
+            # and reverse it.  Won't be needed in Python 3, but for now..
+            self.argv = [ x.decode(sys.stdin.encoding) for x in self.argv ]
+        except Exception: # pragma: no cover
+            pass
         self.client = None
         self.def_url = os.environ.get("NILMDB_URL", "http://localhost/nilmdb/")
         self.subcmd = {}

nilmdb/cmdline/destroy.py

@@ -1,5 +1,6 @@
 from nilmdb.utils.printf import *
 import nilmdb.client
+import fnmatch
 
 from argparse import ArgumentDefaultsHelpFormatter as def_form
 
@@ -10,25 +11,39 @@ def setup(self, sub):
                          Destroy the stream at the specified path.
                          The stream must be empty.  All metadata
                          related to the stream is permanently deleted.
+
+                         Wildcards and multiple paths are supported.
                          """)
     cmd.set_defaults(handler = cmd_destroy)
     group = cmd.add_argument_group("Options")
     group.add_argument("-R", "--remove", action="store_true",
                        help="Remove all data before destroying stream")
+    group.add_argument("-q", "--quiet", action="store_true",
+                       help="Don't display names when destroying "
+                       "multiple paths")
     group = cmd.add_argument_group("Required arguments")
-    group.add_argument("path",
-                       help="Path of the stream to delete, e.g. /foo/bar",
+    group.add_argument("path", nargs='+',
+                       help="Path of the stream to delete, e.g. /foo/bar/*",
                        ).completer = self.complete.path
     return cmd
 
 def cmd_destroy(self):
     """Destroy stream"""
-    if self.args.remove:
+    streams = [ s[0] for s in self.client.stream_list() ]
+    paths = []
+    for path in self.args.path:
+        new = fnmatch.filter(streams, path)
+        if not new:
+            self.die("error: no stream matched path: %s", path)
+        paths.extend(new)
+
+    for path in paths:
+        if not self.args.quiet and len(paths) > 1:
+            printf("Destroying %s\n", path)
+
         try:
-            count = self.client.stream_remove(self.args.path)
+            if self.args.remove:
+                count = self.client.stream_remove(path)
+            self.client.stream_destroy(path)
         except nilmdb.client.ClientError as e:
-            self.die("error removing data: %s", str(e))
-    try:
-        self.client.stream_destroy(self.args.path)
-    except nilmdb.client.ClientError as e:
-        self.die("error destroying stream: %s", str(e))
+            self.die("error destroying stream: %s", str(e))

nilmdb/cmdline/list.py

@@ -10,22 +10,16 @@ def setup(self, sub):
                          formatter_class = def_form,
                          description="""
                          List streams available in the database,
-                         optionally filtering by layout or path.  Wildcards
-                         are accepted.
+                         optionally filtering by path.  Wildcards
+                         are accepted; non-matching paths or wildcards
+                         are ignored.
                          """)
     cmd.set_defaults(verify = cmd_list_verify,
                      handler = cmd_list)
 
     group = cmd.add_argument_group("Stream filtering")
-    group.add_argument("-p", "--path", metavar="PATH", default="*",
-                       help="Match only this path (-p can be omitted)",
+    group.add_argument("path", metavar="PATH", default=["*"], nargs='*',
                        ).completer = self.complete.path
-    group.add_argument("path_positional", default="*",
-                       nargs="?", help=argparse.SUPPRESS,
-                       ).completer = self.complete.path
-    group.add_argument("-l", "--layout", default="*",
-                       help="Match only this stream layout",
-                       ).completer = self.complete.layout
 
     group = cmd.add_argument_group("Interval info")
     group.add_argument("-E", "--ext", action="store_true",
@@ -49,20 +43,12 @@ def setup(self, sub):
     group = cmd.add_argument_group("Misc options")
     group.add_argument("-T", "--timestamp-raw", action="store_true",
                        help="Show raw timestamps when printing times")
+    group.add_argument("-l", "--layout", action="store_true",
+                       help="Show layout type next to path name")
 
     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.
-    got_opt = self.args.path != "*"
-    got_pos = self.args.path_positional != "*"
-    if got_pos:
-        if got_opt:
-            self.parser.error("too many paths specified")
-        else:
-            self.args.path = self.args.path_positional
-
     if self.args.start is not None and self.args.end is not None:
         if self.args.start >= self.args.end:
             self.parser.error("start must precede end")
@@ -80,29 +66,33 @@ def cmd_list(self):
     else:
         time_string = nilmdb.utils.time.timestamp_to_human
 
-    for stream in streams:
-        (path, layout, int_min, int_max, rows, time) = stream[:6]
-        if not (fnmatch.fnmatch(path, self.args.path) and
-                fnmatch.fnmatch(layout, self.args.layout)):
-            continue
+    for argpath in self.args.path:
+        for stream in streams:
+            (path, layout, int_min, int_max, rows, time) = stream[:6]
+            if not fnmatch.fnmatch(path, argpath):
+                continue
 
-        printf("%s %s\n", path, layout)
-
-        if self.args.ext:
-            if int_min is None or int_max is None:
-                printf("  interval extents: (no data)\n")
+            if self.args.layout:
+                printf("%s %s\n", path, layout)
             else:
-                printf("  interval extents: %s -> %s\n",
-                       time_string(int_min), time_string(int_max))
-            printf("        total data: %d rows, %.6f seconds\n",
-                   rows or 0,
-                   nilmdb.utils.time.timestamp_to_seconds(time or 0))
+                printf("%s\n", path)
 
-        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")
+            if self.args.ext:
+                if int_min is None or int_max is None:
+                    printf("  interval extents: (no data)\n")
+                else:
+                    printf("  interval extents: %s -> %s\n",
+                           time_string(int_min), time_string(int_max))
+                printf("        total data: %d rows, %.6f seconds\n",
+                       rows or 0,
+                       nilmdb.utils.time.timestamp_to_seconds(time or 0))
+
+            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/remove.py

@@ -1,17 +1,19 @@
 from nilmdb.utils.printf import *
 import nilmdb.client
+import fnmatch
 
 def setup(self, sub):
     cmd = sub.add_parser("remove", help="Remove data",
                          description="""
                          Remove all data from a specified time range within a
-                         stream.
+                         stream.  If multiple streams or wildcards are provided,
+                         the same time range is removed from all streams.
                          """)
     cmd.set_defaults(handler = cmd_remove)
 
     group = cmd.add_argument_group("Data selection")
-    group.add_argument("path",
-                       help="Path of stream, e.g. /foo/bar",
+    group.add_argument("path", nargs='+',
+                       help="Path of stream, e.g. /foo/bar/*",
                        ).completer = self.complete.path
     group.add_argument("-s", "--start", required=True,
                        metavar="TIME", type=self.arg_time,
@@ -23,18 +25,31 @@ def setup(self, sub):
                        ).completer = self.complete.time
 
     group = cmd.add_argument_group("Output format")
+    group.add_argument("-q", "--quiet", action="store_true",
+                       help="Don't display names when removing "
+                       "from multiple paths")
     group.add_argument("-c", "--count", action="store_true",
                        help="Output number of data points removed")
     return cmd
 
 def cmd_remove(self):
+    streams = [ s[0] for s in self.client.stream_list() ]
+    paths = []
+    for path in self.args.path:
+        new = fnmatch.filter(streams, path)
+        if not new:
+            self.die("error: no stream matched path: %s", path)
+        paths.extend(new)
+
     try:
-        count = self.client.stream_remove(self.args.path,
-                                          self.args.start, self.args.end)
+        for path in paths:
+            if not self.args.quiet and len(paths) > 1:
+                printf("Removing from %s\n", path)
+            count = self.client.stream_remove(path,
+                                              self.args.start, self.args.end)
+            if self.args.count:
+                printf("%d\n", count);
     except nilmdb.client.ClientError as e:
         self.die("error removing data: %s", str(e))
 
-    if self.args.count:
-        printf("%d\n", count)
-
     return 0

nilmdb/server/nilmdb.py

@@ -675,6 +675,7 @@ class NilmDB(object):
 
             # Count how many were removed
             removed += row_end - row_start
+            remaining -= row_end - row_start
 
             if restart is not None:
                 break

nilmdb/server/server.py

@@ -347,24 +347,34 @@ class Stream(NilmApp):
     # /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.CORS_allow(methods = ["POST"])
+    @chunked_response
+    @response_type("application/x-json-stream")
     def remove(self, path, start = None, end = None):
         """
         Remove data from the backend database.  Removes all data in
-        the interval [start, end).  Returns the number of data points
-        removed.
+        the interval [start, end).
+
+        Returns the number of data points removed.  Since this is a potentially
+        long-running operation, multiple numbers may be returned as the
+        data gets removed from the backend database.  The total number of
+        points removed is the sum of all of these numbers.
         """
         (start, end) = self._get_times(start, end)
-        total_removed = 0
-        while True:
-            (removed, restart) = self.db.stream_remove(path, start, end)
-            total_removed += removed
-            if restart is None:
-                break
-            start = restart
-        return total_removed
+
+        if len(self.db.stream_list(path = path)) != 1:
+            raise cherrypy.HTTPError("404", "No such stream: " + path)
+
+        @workaround_cp_bug_1200
+        def content(start, end):
+            # Note: disable chunked responses to see tracebacks from here.
+            while True:
+                (removed, restart) = self.db.stream_remove(path, start, end)
+                yield json.dumps(removed) + "\r\n"
+                if restart is None:
+                    break
+                start = restart
+        return content(start, end)
 
     # /stream/intervals?path=/newton/prep
     # /stream/intervals?path=/newton/prep&start=1234567890.0&end=1234567899.0

tests/test_cmdline.py

@@ -300,38 +300,19 @@ class TestCmdline(object):
 
         # Verify we got those 3 streams and they're returned in
         # alphabetical order.
-        self.ok("list")
+        self.ok("list -l")
         self.match("/newton/prep float32_8\n"
                    "/newton/raw uint16_6\n"
                    "/newton/zzz/rawnotch uint16_9\n")
 
         # Match just one type or one path.  Also check
         # that --path is optional
-        self.ok("list --path /newton/raw")
-        self.match("/newton/raw uint16_6\n")
-
-        self.ok("list /newton/raw")
-        self.match("/newton/raw uint16_6\n")
-
-        self.fail("list -p /newton/raw /newton/raw")
-        self.contain("too many paths")
-
-        self.ok("list --layout uint16_6")
+        self.ok("list --layout /newton/raw")
         self.match("/newton/raw uint16_6\n")
 
         # Wildcard matches
-        self.ok("list --layout uint16*")
-        self.match("/newton/raw uint16_6\n"
-                   "/newton/zzz/rawnotch uint16_9\n")
-
-        self.ok("list --path *zzz* --layout uint16*")
-        self.match("/newton/zzz/rawnotch uint16_9\n")
-
-        self.ok("list *zzz* --layout uint16*")
-        self.match("/newton/zzz/rawnotch uint16_9\n")
-
-        self.ok("list --path *zzz* --layout float32*")
-        self.match("")
+        self.ok("list *zzz*")
+        self.match("/newton/zzz/rawnotch\n")
 
         # reversed range
         self.fail("list /newton/prep --start 2020-01-01 --end 2000-01-01")
@@ -497,28 +478,28 @@ class TestCmdline(object):
         self.ok("list --detail")
         lines_(self.captured, 8)
 
-        self.ok("list --detail --path *prep")
+        self.ok("list --detail *prep")
         lines_(self.captured, 4)
 
-        self.ok("list --detail --path *prep --start='23 Mar 2012 10:02'")
+        self.ok("list --detail *prep --start='23 Mar 2012 10:02'")
         lines_(self.captured, 3)
 
-        self.ok("list --detail --path *prep --start='23 Mar 2012 10:05'")
+        self.ok("list --detail *prep --start='23 Mar 2012 10:05'")
         lines_(self.captured, 2)
 
-        self.ok("list --detail --path *prep --start='23 Mar 2012 10:05:15'")
+        self.ok("list --detail *prep --start='23 Mar 2012 10:05:15'")
         lines_(self.captured, 2)
         self.contain("10:05:15.000")
 
-        self.ok("list --detail --path *prep --start='23 Mar 2012 10:05:15.50'")
+        self.ok("list --detail *prep --start='23 Mar 2012 10:05:15.50'")
         lines_(self.captured, 2)
         self.contain("10:05:15.500")
 
-        self.ok("list --detail --path *prep --start='23 Mar 2012 19:05:15.50'")
+        self.ok("list --detail *prep --start='23 Mar 2012 19:05:15.50'")
         lines_(self.captured, 2)
         self.contain("no intervals")
 
-        self.ok("list --detail --path *prep --start='23 Mar 2012 10:05:15.50'"
+        self.ok("list --detail *prep --start='23 Mar 2012 10:05:15.50'"
                 + " --end='23 Mar 2012 10:05:15.51'")
         lines_(self.captured, 2)
         self.contain("10:05:15.500")
@@ -527,15 +508,15 @@ class TestCmdline(object):
         lines_(self.captured, 8)
 
         # Verify the "raw timestamp" output
-        self.ok("list --detail --path *prep --timestamp-raw "
+        self.ok("list --detail *prep --timestamp-raw "
                 "--start='23 Mar 2012 10:05:15.50'")
         lines_(self.captured, 2)
         self.contain("[ 1332497115500000 -> 1332497160000000 ]")
 
         # bad time
-        self.fail("list --detail --path *prep -T --start='9332497115.612'")
+        self.fail("list --detail *prep -T --start='9332497115.612'")
         # good time
-        self.ok("list --detail --path *prep -T --start='1332497115.612'")
+        self.ok("list --detail *prep -T --start='1332497115.612'")
         lines_(self.captured, 2)
         self.contain("[ 1332497115612000 -> 1332497160000000 ]")
 
@@ -639,7 +620,7 @@ class TestCmdline(object):
 
         # Try nonexistent stream
         self.fail("remove /no/such/foo --start 2000-01-01 --end 2020-01-01")
-        self.contain("No stream at path")
+        self.contain("no stream matched path")
 
         # empty or backward ranges return errors
         self.fail("remove /newton/prep --start 2020-01-01 --end 2000-01-01")
@@ -667,9 +648,14 @@ class TestCmdline(object):
                 "--start '23 Mar 2022 20:00:30' " +
                 "--end '23 Mar 2022 20:00:31'")
         self.match("0\n")
+        self.ok("remove -c /newton/prep /newton/pre* " +
+                "--start '23 Mar 2022 20:00:30' " +
+                "--end '23 Mar 2022 20:00:31'")
+        self.match("Removing from /newton/prep\n0\n" +
+                   "Removing from /newton/prep\n0\n")
 
         # Make sure we have the data we expect
-        self.ok("list --detail /newton/prep")
+        self.ok("list -l --detail /newton/prep")
         self.match("/newton/prep float32_8\n" +
                    "  [ Fri, 23 Mar 2012 10:00:00.000000 +0000"
                    " -> Fri, 23 Mar 2012 10:01:59.991668 +0000 ]\n"
@@ -704,7 +690,7 @@ class TestCmdline(object):
         self.match("24000\n")
 
         # See the missing chunks in list output
-        self.ok("list --detail /newton/prep")
+        self.ok("list --layout --detail /newton/prep")
         self.match("/newton/prep float32_8\n" +
                    "  [ Fri, 23 Mar 2012 10:00:00.000000 +0000"
                    " -> Fri, 23 Mar 2012 10:00:05.000000 +0000 ]\n"
@@ -718,7 +704,7 @@ class TestCmdline(object):
         # Remove all data, verify it's missing
         self.ok("remove /newton/prep --start 2000-01-01 --end 2020-01-01")
         self.match("")  # no count requested this time
-        self.ok("list --detail /newton/prep")
+        self.ok("list -l --detail /newton/prep")
         self.match("/newton/prep float32_8\n" +
                    "  (no intervals)\n")
 
@@ -736,16 +722,16 @@ class TestCmdline(object):
         self.contain("too few arguments")
 
         self.fail("destroy /no/such/stream")
-        self.contain("No stream at path")
+        self.contain("no stream matched path")
 
         self.fail("destroy -R /no/such/stream")
-        self.contain("No stream at path")
+        self.contain("no stream matched path")
 
         self.fail("destroy asdfasdf")
-        self.contain("No stream at path")
+        self.contain("no stream matched path")
 
         # From previous tests, we have:
-        self.ok("list")
+        self.ok("list -l")
         self.match("/newton/prep float32_8\n"
                    "/newton/raw uint16_6\n"
                    "/newton/zzz/rawnotch uint16_9\n")
@@ -761,13 +747,13 @@ class TestCmdline(object):
         lines_(self.captured, 7)
 
         # Destroy for real
-        self.ok("destroy -R /newton/prep")
-        self.ok("list")
+        self.ok("destroy -R /n*/prep")
+        self.ok("list -l")
         self.match("/newton/raw uint16_6\n"
                    "/newton/zzz/rawnotch uint16_9\n")
 
         self.ok("destroy /newton/zzz/rawnotch")
-        self.ok("list")
+        self.ok("list -l")
         self.match("/newton/raw uint16_6\n")
 
         self.ok("destroy /newton/raw")
@@ -786,18 +772,17 @@ class TestCmdline(object):
             self.ok("list")
             self.contain(path)
             # Make sure it was created empty
-            self.ok("list --detail --path " + path)
+            self.ok("list --detail " + path)
             self.contain("(no intervals)")
 
     def test_12_unicode(self):
         # Unicode paths.
         self.ok("destroy /newton/asdf/qwer")
-        self.ok("destroy /newton/prep")
-        self.ok("destroy /newton/raw")
+        self.ok("destroy /newton/prep /newton/raw")
         self.ok("destroy /newton/zzz")
 
         self.ok(u"create /düsseldorf/raw uint16_6")
-        self.ok("list --detail")
+        self.ok("list -l --detail")
         self.contain(u"/düsseldorf/raw uint16_6")
         self.contain("(no intervals)")
 
@@ -883,7 +868,7 @@ class TestCmdline(object):
         du_before = nilmdb.utils.diskusage.du(testdb)
 
         # Make sure we have the data we expect
-        self.ok("list --detail")
+        self.ok("list -l --detail")
         self.match("/newton/prep float32_8\n" +
                    "  [ Fri, 23 Mar 2012 10:00:00.000000 +0000"
                    " -> Fri, 23 Mar 2012 10:01:59.991668 +0000 ]\n"
@@ -919,7 +904,7 @@ class TestCmdline(object):
         self.match("3600\n")
 
         # See the missing chunks in list output
-        self.ok("list --detail")
+        self.ok("list -l --detail")
         self.match("/newton/prep float32_8\n" +
                    "  [ Fri, 23 Mar 2012 10:00:00.000000 +0000"
                    " -> Fri, 23 Mar 2012 10:00:05.000000 +0000 ]\n"
@@ -1043,7 +1028,7 @@ class TestCmdline(object):
             else:
                 raise AssertionError("data not found at " + seek)
             # Verify "list" output
-            self.ok("list")
+            self.ok("list -l")
             self.match("/" + "/".join(components) + " float32_8\n")
 
         # Lots of renames

tests/test_numpyclient.py

@@ -130,6 +130,15 @@ class TestNumpyClient(object):
                                                   [4, 5]]]))
         in_("wrong number of dimensions", str(e.exception))
 
+        # Wrong number of fields
+        with assert_raises(ValueError) as e:
+            client.stream_insert_numpy("/test/1",
+                                       np.array([[0, 1, 2],
+                                                 [1, 2, 3],
+                                                 [3, 4, 5],
+                                                 [4, 5, 6]]))
+        in_("wrong number of fields", str(e.exception))
+
         # Unstructured
         client.stream_create("/test/2", "float32_8")
         client.stream_insert_numpy(