Merge branch 'diffs' of https://bitbucket.org/yackob03/quay into diffs

README.md

@@ -31,3 +31,13 @@ start the workers:
 ```
 python -m workers.diffsworker -D
 ```
+
+bouncing the servers:
+
+```
+sudo kill -HUP <pid of nginx>
+kill -HUP <pid of gunicorn>
+
+kill <pids of worker daemons>
+restart daemons
+```

data/queue.py

@@ -52,4 +52,7 @@ class WorkQueue(object):
     return None
 
   def complete(self, completed_item):
-    item.delete_instance()
+    completed_item.delete_instance()
+
+
+image_diff_queue = WorkQueue('imagediff')

endpoints/registry.py

@@ -7,6 +7,7 @@ from datetime import datetime
 from time import time
 
 import storage
+from data.queue import image_diff_queue
 
 from app import app
 from auth.auth import process_auth, extract_namespace_repo_from_session
@@ -142,6 +143,16 @@ def put_image_layer(namespace, repository, image_id):
     abort(400)  # 'Checksum mismatch, ignoring the layer')
   # Checksum is ok, we remove the marker
   store.remove(mark_path)
+
+  # The layer is ready for download, send a job to the work queue to
+  # process it.
+  logger.debug('Queing diffs job for image: %s' % image_id)
+  image_diff_queue.put(json.dumps({
+    'namespace': namespace,
+    'repository': repository,
+    'image_id': image_id,
+  }))
+
   return make_response('true', 200)
 
 
@@ -173,6 +184,16 @@ def put_image_checksum(namespace, repository, image_id):
     abort(400)  # 'Checksum mismatch')
   # Checksum is ok, we remove the marker
   store.remove(mark_path)
+
+  # The layer is ready for download, send a job to the work queue to
+  # process it.
+  logger.debug('Queing diffs job for image: %s' % image_id)
+  image_diff_queue.put(json.dumps({
+    'namespace': namespace,
+    'repository': repository,
+    'image_id': image_id,
+  }))
+
   return make_response('true', 200)
 
 
@@ -357,6 +378,7 @@ def process_image_changes(namespace, repository, image_id):
   sections = ('added', 'changed', 'removed')
   for section, source_trie in zip(sections, new_metadata[1:]):
     diffs[section] = list(source_trie)
+    diffs[section].sort()
   store.put_content(image_diffs_path, json.dumps(diffs, indent=2))
 
   return new_trie_path

requirements-nover.txt

@@ -14,3 +14,4 @@ mixpanel-py
 beautifulsoup4
 marisa-trie
 apscheduler
+python-daemon

requirements.txt

@@ -1,25 +1,29 @@
+APScheduler==2.1.1
 Flask==0.10.1
 Flask-Login==0.2.7
 Flask-Mail==0.9.0
 Flask-Principal==0.4.0
 Jinja2==2.7.1
 MarkupSafe==0.18
-PyMySQL==0.5
+PyMySQL==0.6.1
 Werkzeug==0.9.4
 argparse==1.2.1
 beautifulsoup4==4.3.2
 blinker==1.3
-boto==2.13.3
+boto==2.15.0
 distribute==0.6.34
 eventlet==0.14.0
 greenlet==0.4.1
 gunicorn==18.0
 itsdangerous==0.23
+lockfile==0.9.1
+marisa-trie==0.5.1
 mixpanel-py==3.0.0
 peewee==2.1.4
 py-bcrypt==0.4
+python-daemon==1.6
 python-dateutil==2.1
 requests==2.0.0
 six==1.4.1
-stripe==1.9.5
+stripe==1.9.8
 wsgiref==0.1.2

test/data/registry/images/devtable/complex/d3acf10ce5ad8b208ba38e626ced2946/diffs.json

@@ -1,16 +1,16 @@
 {
   "removed": [], 
   "added": [
-    "dev/kmsg", 
+    "/dev/kmsg", 
-    "root/.bash_history", 
+    "/root/.bash_history", 
-    "usr/sbin/policy-rc.d"
+    "/usr/sbin/policy-rc.d"
   ], 
   "changed": [
-    "tmp", 
+    "/.", 
-    "dev", 
+    "/dev", 
-    ".", 
+    "/root", 
-    "usr", 
+    "/tmp", 
-    "root", 
+    "/usr", 
-    "usr/sbin"
+    "/usr/sbin"
   ]
 }

test/data/registry/images/devtable/gargantuan/0ccb6d859cbdf08f5c71132707e144fc/diffs.json

@@ -1,16 +1,16 @@
 {
   "removed": [], 
   "added": [
-    "dev/kmsg", 
+    "/dev/kmsg", 
-    "root/.bash_history", 
+    "/root/.bash_history", 
-    "usr/sbin/policy-rc.d"
+    "/usr/sbin/policy-rc.d"
   ], 
   "changed": [
-    "tmp", 
+    "/.", 
-    "dev", 
+    "/dev", 
-    ".", 
+    "/root", 
-    "usr", 
+    "/tmp", 
-    "root", 
+    "/usr", 
-    "usr/sbin"
+    "/usr/sbin"
   ]
 }

test/data/registry/images/devtable/gargantuan/1dcf485dfbdf2ee735d39f74d4ec9132/diffs.json

@@ -1,16 +1,16 @@
 {
   "removed": [], 
   "added": [
-    "dev/kmsg", 
+    "/dev/kmsg", 
-    "root/.bash_history", 
+    "/root/.bash_history", 
-    "usr/sbin/policy-rc.d"
+    "/usr/sbin/policy-rc.d"
   ], 
   "changed": [
-    "tmp", 
+    "/.", 
-    "dev", 
+    "/dev", 
-    ".", 
+    "/root", 
-    "usr", 
+    "/tmp", 
-    "root", 
+    "/usr", 
-    "usr/sbin"
+    "/usr/sbin"
   ]
 }

test/data/registry/images/devtable/gargantuan/1f9c4e12e13735668f2f2b5ebed92a27/diffs.json

@@ -1,16 +1,16 @@
 {
   "removed": [], 
   "added": [
-    "dev/kmsg", 
+    "/dev/kmsg", 
-    "root/.bash_history", 
+    "/root/.bash_history", 
-    "usr/sbin/policy-rc.d"
+    "/usr/sbin/policy-rc.d"
   ], 
   "changed": [
-    "tmp", 
+    "/.", 
-    "dev", 
+    "/dev", 
-    ".", 
+    "/root", 
-    "usr", 
+    "/tmp", 
-    "root", 
+    "/usr", 
-    "usr/sbin"
+    "/usr/sbin"
   ]
 }

test/data/registry/images/devtable/gargantuan/22d54e9c626a2aa258e5185dc315d5a0/diffs.json

@@ -1,16 +1,16 @@
 {
   "removed": [], 
   "added": [
-    "dev/kmsg", 
+    "/dev/kmsg", 
-    "root/.bash_history", 
+    "/root/.bash_history", 
-    "usr/sbin/policy-rc.d"
+    "/usr/sbin/policy-rc.d"
   ], 
   "changed": [
-    "tmp", 
+    "/.", 
-    "dev", 
+    "/dev", 
-    ".", 
+    "/root", 
-    "usr", 
+    "/tmp", 
-    "root", 
+    "/usr", 
-    "usr/sbin"
+    "/usr/sbin"
   ]
 }

test/data/registry/images/devtable/gargantuan/2561f1d3a276103635c2cde5a7dc7759/diffs.json

@@ -1,16 +1,16 @@
 {
   "removed": [], 
   "added": [
-    "dev/kmsg", 
+    "/dev/kmsg", 
-    "root/.bash_history", 
+    "/root/.bash_history", 
-    "usr/sbin/policy-rc.d"
+    "/usr/sbin/policy-rc.d"
   ], 
   "changed": [
-    "tmp", 
+    "/.", 
-    "dev", 
+    "/dev", 
-    ".", 
+    "/root", 
-    "usr", 
+    "/tmp", 
-    "root", 
+    "/usr", 
-    "usr/sbin"
+    "/usr/sbin"
   ]
 }

test/data/registry/images/devtable/gargantuan/8324a62cde97d7a0ebfc011068186a05/diffs.json

@@ -1,16 +1,16 @@
 {
   "removed": [], 
   "added": [
-    "dev/kmsg", 
+    "/dev/kmsg", 
-    "root/.bash_history", 
+    "/root/.bash_history", 
-    "usr/sbin/policy-rc.d"
+    "/usr/sbin/policy-rc.d"
   ], 
   "changed": [
-    "tmp", 
+    "/.", 
-    "dev", 
+    "/dev", 
-    ".", 
+    "/root", 
-    "usr", 
+    "/tmp", 
-    "root", 
+    "/usr", 
-    "usr/sbin"
+    "/usr/sbin"
   ]
 }

test/data/registry/images/devtable/gargantuan/8de22bdae0bcfee17a4b54c46ccca70a/diffs.json

@@ -1,16 +1,16 @@
 {
   "removed": [], 
   "added": [
-    "dev/kmsg", 
+    "/dev/kmsg", 
-    "root/.bash_history", 
+    "/root/.bash_history", 
-    "usr/sbin/policy-rc.d"
+    "/usr/sbin/policy-rc.d"
   ], 
   "changed": [
-    "tmp", 
+    "/.", 
-    "dev", 
+    "/dev", 
-    ".", 
+    "/root", 
-    "usr", 
+    "/tmp", 
-    "root", 
+    "/usr", 
-    "usr/sbin"
+    "/usr/sbin"
   ]
 }

test/data/registry/images/devtable/gargantuan/c1d29009854393d48c3c9a64891e2ba6/diffs.json

@@ -1,16 +1,16 @@
 {
   "removed": [], 
   "added": [
-    "dev/kmsg", 
+    "/dev/kmsg", 
-    "root/.bash_history", 
+    "/root/.bash_history", 
-    "usr/sbin/policy-rc.d"
+    "/usr/sbin/policy-rc.d"
   ], 
   "changed": [
-    "tmp", 
+    "/.", 
-    "dev", 
+    "/dev", 
-    ".", 
+    "/root", 
-    "usr", 
+    "/tmp", 
-    "root", 
+    "/usr", 
-    "usr/sbin"
+    "/usr/sbin"
   ]
 }

test/data/registry/images/devtable/shared/fb820a41a1b0f94cf779305c2e23d4d1/diffs.json

@@ -1,16 +1,16 @@
 {
   "removed": [], 
   "added": [
-    "dev/kmsg", 
+    "/dev/kmsg", 
-    "root/.bash_history", 
+    "/root/.bash_history", 
-    "usr/sbin/policy-rc.d"
+    "/usr/sbin/policy-rc.d"
   ], 
   "changed": [
-    "tmp", 
+    "/.", 
-    "dev", 
+    "/dev", 
-    ".", 
+    "/root", 
-    "usr", 
+    "/tmp", 
-    "root", 
+    "/usr", 
-    "usr/sbin"
+    "/usr/sbin"
   ]
 }

test/data/registry/images/devtable/simple/44c28f16388dcfba12cbde037563982c/diffs.json

@@ -1,16 +1,16 @@
 {
   "removed": [], 
   "added": [
-    "dev/kmsg", 
+    "/dev/kmsg", 
-    "root/.bash_history", 
+    "/root/.bash_history", 
-    "usr/sbin/policy-rc.d"
+    "/usr/sbin/policy-rc.d"
   ], 
   "changed": [
-    "tmp", 
+    "/.", 
-    "dev", 
+    "/dev", 
-    ".", 
+    "/root", 
-    "usr", 
+    "/tmp", 
-    "root", 
+    "/usr", 
-    "usr/sbin"
+    "/usr/sbin"
   ]
 }

test/data/registry/images/public/publicrepo/030581d6f421978b637479b73f862b63/diffs.json

@@ -1,16 +1,16 @@
 {
   "removed": [], 
   "added": [
-    "dev/kmsg", 
+    "/dev/kmsg", 
-    "root/.bash_history", 
+    "/root/.bash_history", 
-    "usr/sbin/policy-rc.d"
+    "/usr/sbin/policy-rc.d"
   ], 
   "changed": [
-    "tmp", 
+    "/.", 
-    "dev", 
+    "/dev", 
-    ".", 
+    "/root", 
-    "usr", 
+    "/tmp", 
-    "root", 
+    "/usr", 
-    "usr/sbin"
+    "/usr/sbin"
   ]
 }

test/data/registry/images/public/publicrepo/1b4bc84b74c71ac837ff19eae76e76c0/diffs.json

@@ -1,16 +1,16 @@
 {
   "removed": [], 
   "added": [
-    "dev/kmsg", 
+    "/dev/kmsg", 
-    "root/.bash_history", 
+    "/root/.bash_history", 
-    "usr/sbin/policy-rc.d"
+    "/usr/sbin/policy-rc.d"
   ], 
   "changed": [
-    "tmp", 
+    "/.", 
-    "dev", 
+    "/dev", 
-    ".", 
+    "/root", 
-    "usr", 
+    "/tmp", 
-    "root", 
+    "/usr", 
-    "usr/sbin"
+    "/usr/sbin"
   ]
 }

test/data/registry/images/public/publicrepo/7025cb49ee8cd5da3054e4b69210fdbe/diffs.json

@@ -1,16 +1,16 @@
 {
   "removed": [], 
   "added": [
-    "dev/kmsg", 
+    "/dev/kmsg", 
-    "root/.bash_history", 
+    "/root/.bash_history", 
-    "usr/sbin/policy-rc.d"
+    "/usr/sbin/policy-rc.d"
   ], 
   "changed": [
-    "tmp", 
+    "/.", 
-    "dev", 
+    "/dev", 
-    ".", 
+    "/root", 
-    "usr", 
+    "/tmp", 
-    "root", 
+    "/usr", 
-    "usr/sbin"
+    "/usr/sbin"
   ]
 }

test/data/sample/diffs/diffs3.json

@@ -1,16 +1,16 @@
 {
   "removed": [], 
   "added": [
-    "dev/kmsg", 
+    "/dev/kmsg", 
-    "root/.bash_history", 
+    "/root/.bash_history", 
-    "usr/sbin/policy-rc.d"
+    "/usr/sbin/policy-rc.d"
   ], 
   "changed": [
-    "tmp", 
+    "/.", 
-    "dev", 
+    "/dev", 
-    ".", 
+    "/root", 
-    "usr", 
+    "/tmp", 
-    "root", 
+    "/usr", 
-    "usr/sbin"
+    "/usr/sbin"
   ]
 }

util/changes.py

@@ -27,7 +27,7 @@ def files_and_dirs_from_tar(source_stream, removed_prefix_collector):
       continue
 
     else:
-      yield absolute
+      yield "/" + absolute
 
 
 def __compute_removed(base_trie, removed_prefixes):

workers/diffsworker.py

@@ -2,17 +2,16 @@ import logging
 import json
 import daemon
 import time
+import argparse
 
 from apscheduler.scheduler import Scheduler
 
-from data.queue import WorkQueue
+from data.queue import image_diff_queue
 from endpoints.registry import process_image_changes
 
 
 logger = logging.getLogger(__name__)
 
-image_diff_queue = WorkQueue('imagediff')
-
 
 def process_work_items():
   logger.debug('Getting work item from queue.')
@@ -23,18 +22,41 @@ def process_work_items():
     logger.debug('Queue gave us some work: %s' % item.body)
 
     request = json.loads(item.body)
-    process_image_changes(request['namepspace'], request['repository'],
+    process_image_changes(request['namespace'], request['repository'],
                           request['image_id'])
+
+    image_diff_queue.complete(item)
+
   else:
     logger.debug('No work today.')
 
-FORMAT = '%(asctime)-15s - %(levelname)s - %(pathname)s - %(funcName)s - %(message)s'
-logging.basicConfig(format=FORMAT, level=logging.DEBUG)
 
-sched = Scheduler()
+parser = argparse.ArgumentParser(description='Worker daemon to compute diffs')
-sched.start()
+parser.add_argument('-D', action='store_true', default=False,
+                    help='Run the worker in daemon mode.')
+parser.add_argument('--log', default='diffsworker.log',
+                    help='Specify the log file for the worker as a daemon.')
+args = parser.parse_args()
 
-sched.add_interval_job(process_work_items, seconds=10)
+def start_worker(args):
+  FORMAT = '%(asctime)-15s - %(levelname)s - %(pathname)s - %(funcName)s - %(message)s'
 
-while True:
+  if args.D:
-  time.sleep(60 * 60 * 24)  # sleep one day, basically forever
+    logging.basicConfig(filename=args.log, format=FORMAT, level=logging.DEBUG)
+  else:
+    logging.basicConfig(format=FORMAT, level=logging.DEBUG)    
+
+  sched = Scheduler()
+  sched.start()
+
+  sched.add_interval_job(process_work_items, seconds=10)
+
+  while True:
+    time.sleep(60 * 60 * 24)  # sleep one day, basically forever
+
+if args.D:
+  with daemon.DaemonContext():
+    start_worker(args)
+
+else:
+  start_worker(args)