quay/workers/logrotateworker.py

import logging
import json
import time

from datetime import timedelta, datetime

import features
from app import app, storage
from data.database import UseThenDisconnect
from data.model import db_transaction
from data.model.log import (get_stale_logs, get_stale_logs_start_id,
                            get_stale_logs_cutoff_id, delete_stale_logs)
from util.registry.gzipwrap import GzipWrap
from util.locking import GlobalLock
from workers.worker import Worker

logger = logging.getLogger(__name__)

STALE_AFTER = timedelta(days=30)
MIN_LOGS_PER_ROTATION = 10000

WORKER_FREQUENCY = app.config.get('ACTION_LOG_ROTATION_FREQUENCY', 3600 * 6)
SAVE_PATH = app.config.get('ACTION_LOG_ARCHIVE_PATH')
SAVE_LOCATION = app.config.get('ACTION_LOG_ARCHIVE_LOCATION')

class LogRotateWorker(Worker):
  """ Worker used to rotate old logs out the database and into storage. """
  def __init__(self):
    super(LogRotateWorker, self).__init__()
    self.add_operation(self._archive_logs, WORKER_FREQUENCY)

  def _archive_logs(self):
    logger.debug('Attempting to rotate log entries')
    while True:
      with GlobalLock('ACTION_LOG_ROTATION') as gl:
        if not gl:
          logger.debug('Could not acquire global lock; sleeping')
          return

        with UseThenDisconnect(app.config):
          with db_transaction():
            cutoff_date = datetime.now() - STALE_AFTER
            start_id = get_stale_logs_start_id()
            cutoff_id = get_stale_logs_cutoff_id(cutoff_date)

            if start_id is None or cutoff_id is None:
              logger.warning('No logs to be archived.')
              return

            logger.debug('Found starting ID %s and cutoff ID %s', start_id, cutoff_id)

            approx_count = cutoff_id - start_id
            if approx_count < MIN_LOGS_PER_ROTATION:
              logger.debug('Not enough stale logs to warrant rotation (approx %d)', approx_count)
              return

            end_id = start_id + MIN_LOGS_PER_ROTATION
            logs = (pretty_print_in_json(log)
                    for log in get_stale_logs(start_id, end_id))

        logger.debug('Archiving logs from IDs %s to %s', start_id, end_id)
        filename = '%s%d-%d.txt.gz' % (SAVE_PATH, start_id, end_id)
        storage.stream_write(SAVE_LOCATION, filename, GzipWrap(logs))

        with UseThenDisconnect(app.config):
          delete_stale_logs(start_id, end_id)


def pretty_print_in_json(log):
  """ Pretty prints a LogEntry in JSON. """
  return json.dumps({'kind_id': log.kind_id,
                     'account_id': log.account_id,
                     'performer_id': log.performer_id,
                     'repository_id': log.repository_id,
                     'datetime': str(log.datetime),
                     'ip': str(log.ip),
                     'metadata_json': json.loads(str(log.metadata_json))})


def main():
  logging.config.fileConfig('conf/logging_debug.conf', disable_existing_loggers=False)

  if not features.ACTION_LOG_ROTATION or None in [SAVE_PATH, SAVE_LOCATION]:
    logger.debug('Action log rotation worker not enabled; skipping')
    while True:
      time.sleep(100000)

  worker = LogRotateWorker()
  worker.start()


if __name__ == "__main__":
  main()