Differential D23 Diff 64 swh/storage/archiver/director.py

Changeset View

Standalone View

swh/storage/archiver/director.py

This file was added.

				# Copyright (C) 2015 The Software Heritage developers
				# See the AUTHORS file at the top-level directory of this distribution
				# License: GNU General Public License version 3, or any later version
				# See top-level LICENSE file for more information

				import swh

				from datetime import datetime

				from swh.core import hashutil, config
				from .worker import ArchiverWorker
				from ..db import cursor_to_bytes


				DEFAULT_CONFIG = {
				'objstorage_path': '/tmp/swh-storage/objects',
				'batch_max_size': 50,
				'archival_max_age': 3600,
				'retention_policy': 2,

				'dbname': 'softwareheritage-snapshot-20151001',
				'user': 'qcampos',
				}


				class ArchiverDirector():
				"""Process the files in order to know which one needs as backup.

				The archiver director process the files in the local storage in order
				ardumontUnsubmitted Done Inline Actions which one is `needed as` backup? process`es` ardumont: which one is `needed as` backup? process`es`
				to know which one needs archival and it delegates this task to
				archiver workers.

				Attributes:
				master_storage: the local storage of the master server.
				slave_storages: Iterable of remote obj storages to the slaves servers
				used for backup.
				batch_max_size: The number of content items that can be given
				to the same archiver worker.
				archival_max_age: Delay given to the worker to cpy all the files
				ardumontUnsubmitted Done Inline Actions c`o`py` ardumont: c`o`py`
				in a given batch.
				retention_policy: Required number of copies for the content to
				be considered safe.
				"""

				def __init__(self, db_conn, config):
				""" Constructor of the archiver director.

				Args:
				db_conn: db_conn: Either a libpq connection string,
				or a psycopg2 connection.
				config: Archiver_configuration. A dictionnary that must contains
				the following keys.
				objstorage_path (string): the path of the objstorage of the
				master.
				batch_max_size (int): The number of content items that can be
				given to the same archiver worker.
				archival_max_age (int): Delay given to the worker to cpy all
				the files in a given batch.
				retention_policy (int): Required number of copies for the
				content to be considered safe.
				"""
				# Get the local storage of the master and remote ones for the slaves.
				storage_args = [db_conn, config['objstorage_path']]
				master_storage = swh.storage.get_storage('local_storage', storage_args)
				slaves = {
				id: url
				for id, url
				in master_storage.db.archive_ls()
				}

				# TODO Database should be initialized somehow before going in
				# production. For now, assumes that the database contains
				# datas for all the current content.

				self.master_storage = master_storage
				self.slave_storages = slaves
				self.batch_max_size = config['batch_max_size']
				self.archival_max_age = config['archival_max_age']
				self.retention_policy = config['retention_policy']

				def run(self):
				""" Run the archiver director.

				The archiver director will check all the content of the archiver
				ardumontUnsubmitted Done Inline Actions content`s` ardumont: content`s`
				database and do the required backup jobs.
				"""
				for batch in self.get_unarchived_content():
				aw = ArchiverWorker(batch, self.master_storage,
				self.slave_storages, self.retention_policy)
				# TODO Run the task with celery.
				aw.run()

				def get_unarchived_content(self):
				""" get all the content that needs to be archived.
				ardumontUnsubmitted Done Inline Actions content`s` ardumont: content`s`

				Yields:
				A batch of content. Batches are dictionnaries wich associates
				a content id to the data about servers that conteins it or not.

				ardumontUnsubmitted Done Inline Actions content`s` w`h`ich cont`a`ins ardumont: content`s` w`h`ich cont`a`ins
				{'id1':
				{'present': [('slave1', 'slave1_url')],
				'missing': [('slave2', 'slave2_url'),
				('slave3', 'slave3_url')]
				},
				'id2':
				{'present': [],
				'missing': [
				('slave1', 'slave1_url'),
				('slave2', 'slave2_url'),
				('slave3', 'slave3_url')
				]
				}

				Where keys (idX) are sha1 of the content and (slaveX, slaveX_url)
				are ids and urls of the slaves storages.
				ardumontUnsubmitted Done Inline Actions storage slaves ardumont: storage slaves

				At least all the content that don't have enough copies on the
				backups servers are distributed into these batches.
				"""
				# Get the data about each content referenced into the archiver.
				missing_copy = {}
				for content_id in self.master_storage.db.content_archive_ls():
				# Do some initializations
				db_content_id = '\\x' + hashutil.hash_to_hex(content_id[0])

				# Query in order to know in which servers the content is saved.
				backups = self.master_storage.db.content_archive_get(
				content=db_content_id
				)
				for _content_id, server_id, status, mtime in backups:

				# If the content is ongoing but still have time, there is
				# another worker working on this content.
				if status == 'ongoing':
				mtime = mtime.replace(tzinfo=None)
				elapsed = (datetime.now() - mtime).total_seconds()
				if elapsed < self.archival_max_age:
				continue
				server_data = (server_id, self.slave_storages[server_id])
				missing_copy.setdefault(
				db_content_id,
				{'present': [], 'missing': []}
				).setdefault(status, []).append(server_data)

				# Check the content before archivage.
				ardumontUnsubmitted Done Inline Actions archiva`l` ardumont: archiva`l`
				# TODO catch exception and try to restore the file from an
				# archive?
				self.master_storage.objstorage.check(content_id[0])

				if len(missing_copy) >= self.batch_max_size:
				yield missing_copy
				missing_copy = {}

				if len(missing_copy) > 0:
				yield missing_copy


				def initialize_content_archive(db, sample_size, names=['Local']):
				""" Initialize the content_archive table with a sample.

				From the content table, get a sample of id, and fill the
				content_archive table with those id in order to create a test sample
				for the archiver.

				Args:
				db: The database of the storage.
				sample_size (int): The size of the sample to create.
				names: A list of archive names. Those archives must already exists.
				Archival status of the archives content will be erased on db.

				Returns:
				Tha amount of entry created.
				"""
				with db.transaction() as cur:
				cur.execute('DELETE FROM content_archive')

				with db.transaction() as cur:
				cur.execute('SELECT sha1 from content limit %d' % sample_size)
				ids = list(cursor_to_bytes(cur))

				for id, in ids:
				tid = r'\x' + hashutil.hash_to_hex(id)

				with db.transaction() as cur:
				for name in names:
				s = """INSERT INTO content_archive
				VALUES('%s'::sha1, '%s', 'missing', now())
				""" % (tid, name)
				cur.execute(s)

				print('Initialized database with', sample_size * len(names), 'items')
				return sample_size * len(names)


				def add_content_to_objstore(director, source, content_ids):
				""" Fill the objstore according to the database

				Get the current status of the database and fill the objstorage of the
				master storage according to these data.
				Content are fetched from the source, which is a storage.

				Args:
				director (ArchiverDirector): The archiver director containing
				the master storage to fill.
				source (Storage): A storage that contains the content for all the
				ids in content_ids.
				content_ids: A list of ids that should be added to the master object
				storage.
				"""
				for res in source.content_get(content_ids):
				content_data = res['data']
				director.master_storage.objstorage.add_bytes(content_data)
				qcamposAuthorUnsubmitted Done Inline Actions Please, ignore that as it is only used for debug purposes. However, that brings up a question. The table content_archive contains a field status that may be 'missing'. Does that means that everything in the content table will be in the content_data table as 'missing' at the start? The 'missing' status is something we can compute on the fly (in content but not in content_archive) in order to avoid this storage. I don't realize the size it takes, though (but I want to). It may be nothing compared to the whole database and not worth the extra-effort it would require to compute the missing on the fly. Is it? qcampos: Please, ignore that as it is only used for debug purposes. However, that brings up a question.
				ardumontUnsubmitted Done Inline Actions Please, ignore that as it is only used for debug purposes. Ok. Either, you could add a comment `NOT FOR PRODUCTION` in the function's docstring or get this code in a bin/ folder at the repository's root (the plus side is that it will be easier to reach from cli). Does that means that everything in the content table will be in the content_data table as 'missing' at the start? Possibly. I see 2 problematics: one is to adapt the existing code to deal with new content as soon as this module is ready. That is, when content_add is triggered, it also adds newly added contents in archival table with `missing` status. the other one being how to make the existing contents be archived (it's another task). But that is for later. I don't realize the size it takes, though (but I want to). To have an approximate size (because it's dayly updated) check https://archive.softwareheritage.org/api/1/stat/counters/. Right now, we are at 2349689600 contents ^^. It may be nothing compared to the whole database and not worth the extra-effort it would require to compute the missing on the fly. Is it? I think that bootstraping this and dealing with prior data are 2 separate tasks. ardumont: > Please, ignore that as it is only used for debug purposes. Ok. Either, you could add a…


				if __name__ == '__main__':
				import sys

				conf = config.read(sys.argv[1], DEFAULT_CONFIG)
				cstring = 'dbname={} user={}'.format(conf['dbname'], conf['user'])

				director = ArchiverDirector(cstring, conf)
				director.run()