Merge pull request #4 from johanherman/archive-verify-remove

Update archive-db with support for verifications (and some ideas about removals)

Author: Stephan Lohse

.travis.yml

@@ -1,13 +1,17 @@
 language: python
+
 python:
-  - "3.5"
+        - "3.5"
+
+before_install:
+        - sudo python -m pip install pipenv
 
 install:
-  - pip install pipenv
-  - pipenv install --dev
+        - pipenv install --dev
 
 script:
-  - pipenv run nosetests tests/
+        - pipenv run nosetests tests/
 
 notifications:
-    email: false
+        email: false
+

README.md

@@ -1,15 +1,14 @@
-Arteria Archive DB
+SNPSEQ Archive DB
 ==================
 
-A self contained (Tornado) REST service that serves as a frontend for a simple SQL db that contains the state of our uploads, verifications and removals done by other Arteria archive services.  
+A self contained (Tornado) REST service that serves as a frontend for a simple SQL db that contains the state of our uploads, verifications and removals done by other SNPSEQ archive services.  
 
 Trying it out
 -------------
 
     python3 -m pip install pipenv
     pipenv install --deploy
 
-
 Try running it:
 
      pipenv run ./archive-db-ws --config=config/ --port=8888 --debug
@@ -28,4 +27,14 @@ Running tests
 REST endpoints
 --------------
 
-# FIXME: Update example
+Creating a new Upload (and associated Archive if none exists): 
+
+    curl -i -X "POST" -d '{"path": "/path/to/directory/", "host": "my-host", "description": "my-descr"}' http://localhost:8888/api/1.0/upload
+
+Creating a new Verification (and associated Archive if none exists):
+    
+    curl -i -X "POST" -d '{"path": "/path/to/directory/", "host": "my-host", "description": "my-descr"}' http://localhost:8888/api/1.0/verification
+
+Getting a randomly picked Archive that has been uploaded within a certain timespan, but never verified before: 
+
+    curl -i -X "GET" -d '{"age": "7", "safety_margin": "3"}' http://localhost:8888/api/1.0/randomarchive

archive_db/app.py

@@ -1,7 +1,7 @@
 import datetime
 
 from archive_db.models.Model import init_db, Archive, Upload, Verification, Removal
-from archive_db.handlers.DbHandlers import UploadHandler, VerificationHandler, RemovalHandler, VersionHandler
+from archive_db.handlers.DbHandlers import UploadHandler, VerificationHandler, RemovalHandler, VersionHandler, RandomUnverifiedArchiveHandler
 
 from arteria.web.app import AppService
 from peewee import *
@@ -19,9 +19,9 @@ def routes(**kwargs):
     return [
         url(r"/api/1.0/version", VersionHandler, name="version", kwargs=kwargs),
         url(r"/api/1.0/upload", UploadHandler, name="upload"),
-        url(r"/api/1.0/verifification/([\w_-]+)",
-            VerificationHandler, name="verification", kwargs=kwargs),
-        url(r"/api/1.0/removal/([\w_-]+)", RemovalHandler, name="removal", kwargs=kwargs)
+        url(r"/api/1.0/verification", VerificationHandler, name="verification"),
+        url(r"/api/1.0/randomarchive", RandomUnverifiedArchiveHandler, name="randomarchive"),
+        url(r"/api/1.0/removal", RemovalHandler, name="removal")
     ]
 
 

archive_db/handlers/DbHandlers.py

@@ -1,31 +1,16 @@
 import datetime as dt
+import os
 
 from arteria.web.handlers import BaseRestHandler
 
 from archive_db.models.Model import Archive, Upload, Verification, Removal
 from archive_db import __version__ as version
 
+from peewee import *
 from tornado import gen
 from tornado.web import RequestHandler, HTTPError
 from tornado.escape import json_decode, json_encode
 
-# TODO: Shall we implement a handler for something like:
-# "Has any package been verified since date `bar`?
-# At the moment this can be solved in the client by comparing
-# with `last verified date`.
-
-"""
-Our handlers are supposed to work as following:
-
-POST /upload/ - create a new archive entry + upload entry
-GET /upload/ - get last global upload
-GET /upload/<archive> - get last upload for <archive>
-POST /verification/ - create a new verification entry
-GET /verification/ - get last global verification
-GET /verification/<archive> - get last verification for <archive>
-POST /removal/<archive> - create a new removal entry for <archive>
-"""
-
 
 class BaseHandler(BaseRestHandler):
     # BaseRestHandler.body_as_object() does not work well
@@ -45,11 +30,12 @@ class UploadHandler(BaseHandler):
     @gen.coroutine
     def post(self):
         """
-        Archive `path` was just now uploaded (not?) OK.
+        Creates a new Upload object in the db, and the associated Archive if it doesn't already exist. 
 
         :param path: Path to archive uploaded
-        :param description: The TSM description of the archive
+        :param description: The unique TSM description of the archive
         :param host: From which host the archive was uploaded
+        :return Information about the created object
         """
 
         body = self.decode(required_members=["path", "description", "host"])
@@ -65,60 +51,140 @@ def post(self):
                           "path": upload.archive.path,
                           "host": upload.archive.host}})
 
+
+class VerificationHandler(BaseHandler):
+
     @gen.coroutine
-    def get(self, archive):
+    def post(self):
+        """
+        Creates a new Verification object in the db, associated to a certain Archive object. 
+        If no Archive object matching the input parameters is found one will be created. 
+        This way we can take care of verifications done for archives uploaded to PDC before
+        this web service and db existed. 
+
+        :param description: The unique TSM description of the archive we've verified. 
+        :param path: The path to the archive that was uploaded 
+        :param host: The host from which the archive was uploaded
+        :return Information about the created object
         """
-        Archive `foo` was last uploaded OK at date `bar`.
+        body = self.decode(required_members=["description", "path", "host"])
 
-        :param archive: Path to archive uploaded
-        :param description: The TSM description of the archive
-        :param host: From which host the archive was uploaded
-        :return The `archive` when it was last uploaded. If no `archive` specified, then it will
-        return the last global upload archive.
+        archive, created = Archive.get_or_create(description=body["description"], host=body["host"], path=body["path"])
+
+        verification = Verification.create(archive=archive, timestamp=dt.datetime.utcnow())
+
+        self.write_json({"status": "created", "verification":
+                        {"id": verification.id,
+                         "timestamp": str(verification.timestamp),
+                         "description": verification.archive.description,
+                         "path": verification.archive.path,
+                         "host": verification.archive.host}})
+
+class RandomUnverifiedArchiveHandler(BaseHandler):
+
+    @gen.coroutine
+    def get(self):
         """
-        # Step 1 - get date when archive was last updated
-        pass
+        Returns an unverified Archive object that has an associated was Upload object 
+        within the interval [today - age - margin, today - margin]. The margin value is 
+        used as a safety buffer, to make sure that the archived data has been properly 
+        flushed to tape upstreams at PDC.
+
+        :param age: Number of days we should look back when picking an unverified archive
+        :param safety_margin: Number of days we should use as safety buffer
+        :return A randomly pickedunverified archive within the specified date interval
+        """
+        body = self.decode(required_members=["age", "safety_margin"])
+        age = int(body["age"])
+        margin = int(body["safety_margin"])
+
+        from_timestamp = dt.datetime.utcnow() - dt.timedelta(days=age+margin)
+        to_timestamp = dt.datetime.utcnow() - dt.timedelta(days=margin)
+
+        # "Give me a randomly chosen archive that was uploaded between from_timestamp and 
+        # to_timestamp, and has no previous verifications"
+        query = (Upload
+                .select()
+                .join(Verification, JOIN.LEFT_OUTER, on=(
+                    Verification.archive_id == Upload.archive_id))
+                .where(Upload.timestamp.between(from_timestamp, to_timestamp))
+                .group_by(Upload.archive_id)
+                .having(fn.Count(Verification.id) < 1)
+                .order_by(fn.Random())
+                .limit(1))
+
+        result_len = query.count()
+
+        if result_len > 0:
+            upload = next(query.execute())
+            archive_name = os.path.basename(os.path.normpath(upload.archive.path))
+            self.write_json({"status": "unverified", "archive":
+                            {"timestamp": str(upload.timestamp),
+                             "path": upload.archive.path,
+                             "description": upload.archive.description,
+                             "host": upload.archive.host,
+                             "archive": archive_name}})
+        else:
+            msg = "No unverified archives uploaded between {} and {} was found!".format(
+                    from_timestamp.strftime("%Y-%m-%d %H:%M:%S"), to_timestamp.strftime("%Y-%m-%d %H:%M:%S"))
+            raise HTTPError(500, msg)
+
 
 # TODO: We might have to add logic in some of the services
 # that adds a file with the description inside the archive,
 # so we can verify that we're operating on the correct
-# archive before verifying/removing.
-
+# archive before (verifying/)removing.
 
-class VerificationHandler(BaseHandler):
+class RemovalHandler(BaseHandler):
 
     @gen.coroutine
-    def post(self, archive):
+    def post(self):
         """
-        Archive `foo` was verified (not) OK at date `bar`.
-
-        :param archive: Path to archive verified
-        :param description: The TSM description of the archive we verified
+        Archive `foo` was either staged for removal or actually just physically removed from local disk, as well 
+        as all its associated files (e.g. runfolder etc). 
         """
         pass
-        # Step 1 - set date when archive was verified OK
+    
 
-    @gen.coroutine
-    def get(self, archive):
         """
-        Give me the date for when any archive was last verified (OK).
-
-        :param archive: Path to archive we want to check
-        :return The `archive` when it was last verified. If no `archive` specified, then it will
-        return the last globally verified archive.
+        # This is an example for how one could start implementing the handler that first schedules archives for 
+        # removal. 
+
+        body = self.decode(required_members=["description", "action"])
+
+        try:
+            archive = Archive.get(description=body["description"])
+        except Archive.DoesNotExist:
+            msg = "No archive with the unique description {} exists in the database!".format(body["description"])
+            self.set_status(500, msg)
+            self.write_json({"status": msg})
+
+        if body["action"] == "set_removable":
+            removal = Removal.create(archive=archive, timestamp_scheduled=dt.datetime.utcnow())
+
+            self.write_json({"status": "scheduled", "removal":
+                            {"id": removal.id,
+                             "timestamp_scheduled": str(removal.timestamp_scheduled),
+                             "description": removal.archive.description,
+                             "path": removal.archive.path,
+                             "host": removal.archive.host,
+                             "done": removal.done}})
+        elif body["action"] == "set_removed":
+            pass
+        else:
+            msg = "Expecting parameter 'action' to be 'set_removable' or set_removed'."
+            raise HTTPError(400, msg)
         """
-        pass
-
-
-class RemovalHandler(BaseHandler):
 
     @gen.coroutine
-    def post(self, archive):
+    def get(self):
         """
-        Archive `foo` was removed from disk at date `bar`.
+        HTTP GET /removal is in this imagined implementation supposed to return those Archive objects
+        that are removable and are verified. One could probably do this by e.g. 
 
-        :param archive: Path to archive removed from disk
-        :param description: The TSM description of the archive we removed
+            - fetch latest date from Verify, which has done == False, and call this X
+            - fetch all Uploads that have has a timestamp older or equal to X
+            - the set of Archives belonging to those Uploads should be OK to remove 
         """
         pass
 

archive_db/models/Model.py

@@ -1,7 +1,10 @@
 from peewee import *
 
-# TODO: Shall we log failed operations? (not uploaded OK, not verified OK)
-# TODO: Shall we have anything to do with staging operations?
+# For schema migrations, see http://docs.peewee-orm.com/en/latest/peewee/database.html#schema-migrations
+# and http://docs.peewee-orm.com/en/latest/peewee/playhouse.html#migrate
+#
+# Make sure that we *always*, as extra security, take a backup of the previous
+# db before doing a migration. We should also take continous backups
 
 db_proxy = Proxy()
 
@@ -48,8 +51,13 @@ class Removal(ChildModel):
     archive = ForeignKeyField(Archive, related_name="removals")
     timestamp = DateTimeField()
 
-# For schema migrations, see http://docs.peewee-orm.com/en/latest/peewee/database.html#schema-migrations
-# and http://docs.peewee-orm.com/en/latest/peewee/playhouse.html#migrate
-#
-# Make sure that we *always*, as extra security, take a backup of the previous
-# db before doing a migration. We should also take continous backups
+    """
+    To let archive-remove better support archives staged/marked/scheduled for removal I envision that
+    one could modify this to something like the following instead: 
+
+        archive = ForeignKeyField(Archive, related_name="removals")
+        done = BooleanField(default=False)      # False = archive has been scheduled for removal; True = archive has been removed.
+        timestamp_scheduled = DateTimeField()   # Or one can just let the queries look to see which timestamp has been filled with a value.
+        timestamp_done = DateTimeField()
+    """
+

config/app.config

@@ -9,3 +9,4 @@ archive_db_log_directory: /var/arteria/archive-db/
 
 # Path to the Sqlite db
 archive_db_path: /var/db/arteria/archive.db
+

tests/test_models.py

@@ -27,7 +27,7 @@ def get_app(self):
         return Application(routes())
 
     def go(self, target, method, body=""):
-        return self.fetch(self.API_BASE + target, method=method, body=json_encode(body), headers={"Content-Type": "application/json"})
+        return self.fetch(self.API_BASE + target, method=method, body=json_encode(body), headers={"Content-Type": "application/json"}, allow_nonstandard_methods=True)
 
     def create_data(self):
         for i in range(self.num_archives):
@@ -89,3 +89,14 @@ def test_create_upload_for_existing_archive(self):
         resp = json_decode(resp.body)
         self.assertEqual(resp["status"], "created")
         self.assertEqual(resp["upload"]["id"], upload_two)
+
+    # Populating the db in a similar way as in self.create_data() does not make the data available for 
+    # the handlers, as they seem to live in an other in-memory instance of the db. Therefore a 
+    # failing test will have to do for now. 
+    def test_failing_fetch_random_unverified_archive(self):
+        # I.e. our lookback window is [today - 5 - 1, today - 1] days. 
+        body = {"age": "5", "safety_margin": "1"}
+        resp = self.go("/randomarchive", method="GET", body=body)
+        self.assertEqual(resp.code, 500)
+    
+