Merge pull request #2766 from coreos-inc/joseph.schorr/QUAY-634/buildlogsarchiver-data-interface

Change buildlogsarchiver to use a data model interface

conf/init/service/buildlogsarchiver/run

@@ -4,6 +4,6 @@ echo 'Starting build logs archiver worker'
 
 QUAYPATH=${QUAYPATH:-"."}
 cd ${QUAYDIR:-"/"}
-PYTHONPATH=$QUAYPATH venv/bin/python -m workers.buildlogsarchiver 2>&1
+PYTHONPATH=$QUAYPATH venv/bin/python -m workers.buildlogsarchiver.buildlogsarchiver 2>&1
 
 echo 'Diffs worker exited'

workers/buildlogsarchiver/buildlogsarchiver.py

@@ -1,13 +1,13 @@
 import logging
 
-from tempfile import SpooledTemporaryFile
 from gzip import GzipFile
+from tempfile import SpooledTemporaryFile
 
-from data import model
+from app import build_logs, log_archive, app
 from data.archivedlogs import JSON_MIMETYPE
 from data.database import CloseForLongOperation
-from app import build_logs, log_archive, app
 from util.streamingjsonencoder import StreamingJSONEncoder
+from workers.buildlogsarchiver.models_pre_oci import pre_oci_model as model
 from workers.worker import Worker
 
 POLL_PERIOD_SECONDS = 30
@@ -25,7 +25,7 @@ class ArchiveBuildLogsWorker(Worker):
     """ Archive a single build, choosing a candidate at random. This process must be idempotent to
         avoid needing two-phase commit. """
     # Get a random build to archive
-    to_archive = model.build.get_archivable_build()
+    to_archive = model.get_archivable_build()
     if to_archive is None:
       logger.debug('No more builds to archive')
       return
@@ -50,7 +50,7 @@ class ArchiveBuildLogsWorker(Worker):
           log_archive.store_file(tempfile, JSON_MIMETYPE, content_encoding='gzip',
                                  file_id=to_archive.uuid)
 
-    we_updated = model.build.mark_build_archived(to_archive.uuid)
+    we_updated = model.mark_build_archived(to_archive.uuid)
     if we_updated:
       build_logs.expire_status(to_archive.uuid)
       build_logs.delete_log_entries(to_archive.uuid)

workers/buildlogsarchiver/models_interface.py

@@ -0,0 +1,38 @@
+from abc import ABCMeta, abstractmethod
+from collections import namedtuple
+from six import add_metaclass
+
+
+class Build(namedtuple('Build', ['uuid', 'logs_archived'])):
+  """
+  Build represents a single build in the build system.
+  """
+
+
+@add_metaclass(ABCMeta)
+class BuildLogsArchiverWorkerDataInterface(object):
+  """
+  Interface that represents all data store interactions required by the build logs archiver worker.
+  """
+
+  @abstractmethod
+  def get_archivable_build(self):
+    """ Returns a build whose logs are available for archiving. If none, returns None. """
+    pass
+
+  @abstractmethod
+  def get_build(self, build_uuid):
+    """ Returns the build with the matching UUID or None if none. """
+    pass
+
+  @abstractmethod
+  def mark_build_archived(self, build_uuid):
+    """ Marks the build with the given UUID as having its logs archived. Returns False if
+        the build was already marked as archived.
+    """
+    pass
+
+  @abstractmethod
+  def create_build_for_testing(self):
+    """ Creates an unarchived build for testing of archiving. """
+    pass

workers/buildlogsarchiver/models_pre_oci.py

@@ -0,0 +1,32 @@
+from data import model
+from workers.buildlogsarchiver.models_interface import Build, BuildLogsArchiverWorkerDataInterface
+
+
+class PreOCIModel(BuildLogsArchiverWorkerDataInterface):
+  def get_archivable_build(self):
+    build = model.build.get_archivable_build()
+    if build is None:
+      return None
+
+    return Build(build.uuid, build.logs_archived)
+
+  def mark_build_archived(self, build_uuid):
+    return model.build.mark_build_archived(build_uuid)
+
+  def create_build_for_testing(self):
+    repo = model.repository.get_repository('devtable', 'simple')
+    access_token = model.token.create_access_token(repo, 'admin')
+    build = model.build.create_repository_build(repo, access_token, {}, None, 'foo')
+    build.phase = 'error'
+    build.save()
+    return Build(build.uuid, build.logs_archived)
+
+  def get_build(self, build_uuid):
+    build = model.build.get_repository_build(build_uuid)
+    if build is None:
+      return None
+
+    return Build(build.uuid, build.logs_archived)
+
+
+pre_oci_model = PreOCIModel()

workers/buildlogsarchiver/test/test_buildlogsarchiver.py

@@ -0,0 +1,30 @@
+from mock import patch, Mock
+
+from app import storage
+from workers.buildlogsarchiver.buildlogsarchiver import ArchiveBuildLogsWorker
+
+from test.fixtures import *
+
+from workers.buildlogsarchiver.models_pre_oci import pre_oci_model as model
+
+def test_logarchiving(app):
+  worker = ArchiveBuildLogsWorker()
+  logs_mock = Mock()
+  logs_mock.get_log_entries = Mock(return_value=(1, [{'some': 'entry'}]))
+
+  # Add a build that is ready for archiving.
+  build = model.create_build_for_testing()
+
+  with patch('workers.buildlogsarchiver.buildlogsarchiver.build_logs', logs_mock):
+    worker._archive_redis_buildlogs()
+
+  # Ensure the get method was called.
+  logs_mock.get_log_entries.assert_called_once()
+  logs_mock.expire_status.assert_called_once()
+  logs_mock.delete_log_entries.assert_called_once()
+
+  # Ensure the build was marked as archived.
+  assert model.get_build(build.uuid).logs_archived
+
+  # Ensure a file was written to storage.
+  assert storage.exists(['local_us'], 'logarchive/%s' % build.uuid)