Add nilmdb-fsck tool to check database consistency

.coveragerc

@@ -7,4 +7,4 @@
 exclude_lines =
 	pragma: no cover
 	if 0:
-omit = nilmdb/utils/datetime_tz*,nilmdb/scripts,nilmdb/_version.py
+omit = nilmdb/utils/datetime_tz*,nilmdb/scripts,nilmdb/_version.py,nilmdb/fsck

Makefile

@@ -23,6 +23,10 @@ docs:
 lint:
 	pylint --rcfile=.pylintrc nilmdb
 
+fscktest:
+#	python -c "import nilmdb.fsck; nilmdb.fsck.Fsck('/home/jim/wsgi/db').check()"
+	python -c "import nilmdb.fsck; nilmdb.fsck.Fsck('/home/jim/mnt/bucket/mnt/sharon/data/db').check()"
+
 test:
 ifeq ($(INSIDE_EMACS), t)
 # Use the slightly more flexible script

nilmdb/cmdline/cmdline.py

@@ -29,6 +29,14 @@ for cmd in subcommands:
     subcmd_mods[cmd] = __import__("nilmdb.cmdline." + cmd, fromlist = [ cmd ])
 
 class JimArgumentParser(argparse.ArgumentParser):
+    def parse_args(self, args=None, namespace=None):
+        # Look for --version anywhere and change it to just "nilmtool
+        # --version".  This makes "nilmtool cmd --version" work, which
+        # is needed by help2man.
+        if "--version" in (args or sys.argv[1:]):
+            args = [ "--version" ]
+        return argparse.ArgumentParser.parse_args(self, args, namespace)
+
     def error(self, message):
         self.print_usage(sys.stderr)
         self.exit(2, sprintf("error: %s\n", message))

nilmdb/fsck/.#fsck.py

@@ -0,0 +1 @@
+jim@pilot.lees.18066:1373305995

nilmdb/fsck/__init__.py

@@ -0,0 +1,5 @@
+"""nilmdb.fsck"""
+
+from __future__ import absolute_import
+
+from nilmdb.fsck.fsck import Fsck

nilmdb/fsck/fsck.py

@@ -0,0 +1,194 @@
+# -*- coding: utf-8 -*-
+
+"""Check database consistency"""
+
+import nilmdb.utils
+import nilmdb.server
+from nilmdb.utils.interval import IntervalError
+from nilmdb.server.interval import Interval, IntervalSet
+from nilmdb.utils.printf import *
+from nilmdb.utils.time import timestamp_to_string
+
+from collections import defaultdict
+import sqlite3
+import os
+import progressbar
+import time
+import cPickle as pickle
+
+class FsckError(Exception):
+    def __init__(self, format, *args):
+        Exception.__init__(self, sprintf(format, *args))
+
+def log(format, *args):
+    printf(format, *args)
+
+def err(format, *args):
+    fprintf(sys.stderr, format, *args)
+
+class Progress(object):
+    def __init__(self, maxval):
+        self.bar = progressbar.ProgressBar(maxval = maxval)
+        if self.bar.term_width == 0:
+            self.bar.term_width = 75
+    def __enter__(self):
+        self.bar.start()
+        self.last_update = 0
+        return self
+    def __exit__(self, exc_type, exc_value, traceback):
+        if exc_type is None:
+            self.bar.finish()
+        else:
+            printf("\n")
+    def update(self, val):
+        self.bar.update(val)
+        #now = time.time()
+        #if now - self.last_update < 0.005:
+        #   time.sleep(0.005)
+        #self.last_update = now
+
+class Fsck(object):
+
+    def __init__(self, path):
+        self.basepath = path
+        self.sqlpath = os.path.join(path, "data.sql")
+        self.bulkpath = os.path.join(path, "data")
+        self.bulklock = os.path.join(path, "data.lock")
+
+    def check(self):
+        self.check_paths()
+        self.check_sql()
+        self.check_streams()
+        log("ok\n")
+
+    def check_paths(self):
+        log("checking paths\n")
+        if not os.path.isfile(self.sqlpath):
+            raise FsckError("SQL database missing")
+        if not os.path.isdir(self.bulkpath):
+            raise FsckError("Bulk data directory missing")
+        with open(self.bulklock, "w") as lockfile:
+            if not nilmdb.utils.lock.exclusive_lock(lockfile):
+                raise FsckError('database already locked by another process')
+        self.bulk = nilmdb.server.bulkdata.BulkData(self.basepath)
+        # override must_close warning
+        if "_must_close" in dir(self.bulk):
+            del self.bulk._must_close
+
+    def check_sql(self):
+        log("checking sqlite database\n")
+
+        self.sql = sqlite3.connect(self.sqlpath)
+        with self.sql as con:
+            ver = con.execute("PRAGMA user_version").fetchone()[0]
+            good = max(nilmdb.server.nilmdb._sql_schema_updates.keys())
+            if ver != good:
+                raise FsckError("database version %d too old, should be %d",
+                                ver, good)
+            self.stream_path = {}
+            self.stream_layout = {}
+            log("  loading paths\n")
+            result = con.execute("SELECT id, path, layout FROM streams")
+            for r in result:
+                if r[0] in self.stream_path:
+                    raise FsckError("duplicated ID %d in stream IDs", r[0])
+                self.stream_path[r[0]] = r[1]
+                self.stream_layout[r[0]] = r[2]
+
+            log("  loading intervals\n")
+            self.stream_interval = defaultdict(list)
+            result = con.execute("SELECT stream_id, start_time, end_time, "
+                                 "start_pos, end_pos FROM ranges")
+            for r in result:
+                if r[0] not in self.stream_path:
+                    raise FsckError("interval ID %d not in streams", k)
+                self.stream_interval[r[0]].append((r[1], r[2], r[3], r[4]))
+
+            log("  loading metadata\n")
+            self.stream_meta = defaultdict(dict)
+            result = con.execute("SELECT stream_id, key, value FROM metadata")
+            for r in result:
+                if r[0] not in self.stream_path:
+                    raise FsckError("metadata ID %d not in streams", k)
+                if r[1] in self.stream_meta[r[0]]:
+                    raise FsckError("duplicate metadata key '%s' for stream %d",
+                                    r[1], r[0])
+                self.stream_meta[r[0]][r[1]] = r[2]
+
+    def check_streams(self):
+        log("checking streams\n")
+        ids = self.stream_path.keys()
+        with Progress(len(ids)) as pbar:
+            for i, sid in enumerate(ids):
+                pbar.update(i)
+                path = self.stream_path[sid]
+
+                # unique path, valid layout
+                if self.stream_path.values().count(path) != 1:
+                    raise FsckError("duplicated path %s", path)
+                layout = self.stream_layout[sid].split('_')[0]
+                if layout not in ('int8', 'int16', 'int32', 'int64',
+                                  'uint8', 'uint16', 'uint32', 'uint64',
+                                  'float32', 'float64'):
+                    raise FsckError("bad layout %s for %s", layout, path)
+                count = int(self.stream_layout[sid].split('_')[1])
+                if count < 1 or count > 1024:
+                    raise FsckError("bad count %d for %s", count, path)
+
+                # must exist in bulkdata
+                bulk = self.bulkpath + path
+                if not os.path.isdir(bulk):
+                    raise FsckError("%s: missing bulkdata dir", path)
+                if not nilmdb.server.bulkdata.Table.exists(bulk):
+                    raise FsckError("%s: bad bulkdata table", path)
+
+                # intervals don't overlap.  Abuse IntervalSet to check
+                # for intervals in file positions, too.
+                timeiset = IntervalSet()
+                posiset = IntervalSet()
+                for (stime, etime, spos, epos) in self.stream_interval[sid]:
+                    new = Interval(stime, etime)
+                    try:
+                        timeiset += new
+                    except IntervalError:
+                        raise FsckError("%s: overlap in intervals:\n"
+                                        "set: %s\nnew: %s\n",
+                                        path, str(timeiset), str(new))
+                    if spos != epos:
+                        new = Interval(spos, epos)
+                        try:
+                            posiset += new
+                        except IntervalError:
+                            raise FsckError("%s: overlap in file offsets:\n"
+                                            "set: %s\nnew: %s\n",
+                                            path, str(posiset), str(new))
+
+                # check bulkdata
+                self.check_bulkdata(sid, path, bulk)
+
+                continue
+                # verify we can can open it with bulkdata
+                try:
+                    tab = None
+                    try:
+                        tab = nilmdb.server.bulkdata.Table(bulk)
+                    except Exception as e:
+                        raise FsckError("%s: can't open bulkdata: %s",
+                                        path, str(e))
+                    self.check_bulkdata(path, tab)
+                finally:
+                    if tab:
+                        tab.close()
+
+    def check_bulkdata(self, sid, path, bulk):
+        with open(os.path.join(bulk, "_format"), "rb") as f:
+            fmt = pickle.load(f)
+        if fmt["version"] != 3:
+            raise FsckError("%s: bad or unsupported bulkdata version %d",
+                            path, fmt["version"])
+        row_per_file = int(fmt["rows_per_file"])
+        files_per_dir = int(fmt["files_per_dir"])
+        layout = fmt["layout"]
+        if layout != self.stream_layout[sid]:
+            raise FsckError("%s: layout mismatch %s != %s", path,
+                            layout, self.stream_layout[sid])

nilmdb/scripts/nilmdb_fsck.py

@@ -0,0 +1,23 @@
+#!/usr/bin/python
+
+import nilmdb.fsck
+import argparse
+import os
+import sys
+
+def main():
+    """Main entry point for the 'nilmdb-fsck' command line script"""
+
+    parser = argparse.ArgumentParser(
+        description = 'Check database consistency',
+        formatter_class = argparse.ArgumentDefaultsHelpFormatter)
+    parser.add_argument("-V", "--version", action="version",
+                        version = nilmdb.__version__)
+    parser.add_argument('-d', '--database', help = 'Database directory',
+                        default = "./db")
+    args = parser.parse_args()
+
+    nilmdb.fsck.Fsck(args.database).check()
+
+if __name__ == "__main__":
+    main()

nilmdb/server/bulkdata.py

@@ -330,7 +330,8 @@ class Table(object):
 
         # Find the last directory.  We sort and loop through all of them,
         # starting with the numerically greatest, because the dirs could be
-        # empty if something was deleted.
+        # empty if something was deleted but the directory was unexpectedly
+        # not deleted.
         subdirs = sorted(filter(regex.search, os.listdir(self.root)),
                          key = lambda x: int(x, 16), reverse = True)
 

nilmdb/server/server.py

@@ -84,10 +84,18 @@ class Stream(NilmApp):
     # Helpers
     def _get_times(self, start_param, end_param):
         (start, end) = (None, None)
-        if start_param is not None:
-            start = string_to_timestamp(start_param)
-        if end_param is not None:
-            end = string_to_timestamp(end_param)
+        try:
+            if start_param is not None:
+                start = string_to_timestamp(start_param)
+        except Exception:
+            raise cherrypy.HTTPError("400 Bad Request", sprintf(
+                "invalid start (%s): must be a numeric timestamp", start_param))
+        try:
+            if end_param is not None:
+                end = string_to_timestamp(end_param)
+        except Exception:
+            raise cherrypy.HTTPError("400 Bad Request", sprintf(
+                "invalid end (%s): must be a numeric timestamp", end_param))
         if start is not None and end is not None:
             if start >= end:
                 raise cherrypy.HTTPError(

setup.py

@@ -126,11 +126,13 @@ setup(name='nilmdb',
                    'nilmdb.client',
                    'nilmdb.cmdline',
                    'nilmdb.scripts',
+                   'nilmdb.fsck',
                    ],
       entry_points = {
           'console_scripts': [
               'nilmtool = nilmdb.scripts.nilmtool:main',
               'nilmdb-server = nilmdb.scripts.nilmdb_server:main',
+              'nilmdb-fsck = nilmdb.scripts.nilmdb_fsck:main',
               ],
           },
       ext_modules = ext_modules,

tests/test_client.py

@@ -242,6 +242,19 @@ class TestClient(object):
         in_("400 Bad Request", str(e.exception))
         in_("start must precede end", str(e.exception))
 
+        # Invalid times in HTTP request
+        with assert_raises(ClientError) as e:
+            client.http.put("stream/insert", "", { "path": "/newton/prep",
+                                                   "start": "asdf", "end": 0 })
+        in_("400 Bad Request", str(e.exception))
+        in_("invalid start", str(e.exception))
+
+        with assert_raises(ClientError) as e:
+            client.http.put("stream/insert", "", { "path": "/newton/prep",
+                                                   "start": 0, "end": "asdf" })
+        in_("400 Bad Request", str(e.exception))
+        in_("invalid end", str(e.exception))
+
         # Good content type
         with assert_raises(ClientError) as e:
             client.http.put("stream/insert", "",

tests/test_cmdline.py

@@ -59,8 +59,7 @@ class TestCmdline(object):
 
     def run(self, arg_string, infile=None, outfile=None):
         """Run a cmdline client with the specified argument string,
-        passing the given input.  Returns a tuple with the output and
-        exit code"""
+        passing the given input.  Save the output and exit code."""
         # printf("TZ=UTC ./nilmtool.py %s\n", arg_string)
         os.environ['NILMDB_URL'] = "http://localhost:32180/"
         class stdio_wrapper:
@@ -160,6 +159,12 @@ class TestCmdline(object):
         self.ok("--help")
         self.contain("usage:")
 
+        # help
+        self.ok("--version")
+        ver = self.captured
+        self.ok("list --version")
+        eq_(self.captured, ver)
+
         # fail for no args
         self.fail("")