Show All 10 Lines	class ArchiverCopier():
""" This archiver copy some files into a remote objstorage		""" This archiver copy some files into a remote objstorage
in order to get a backup.		in order to get a backup.

Attributes:		Attributes:
content_ids: A list of sha1's that represents the content this copier		content_ids: A list of sha1's that represents the content this copier
has to archive.		has to archive.
server (RemoteArchive): The remote object storage that is used to		server (RemoteArchive): The remote object storage that is used to
backup content.		backup content.
master_storage (Storage): The master storage that contains the data		master_objstorage (ObjStorage): The master storage that contains the
the copier needs to archive.		data the copier needs to archive.
"""		"""
def __init__(self, destination, content, master_storage):		def __init__(self, destination, content, master_objstorage):
""" Create a Copier for the archiver		""" Create a Copier for the archiver

Args:		Args:
destination: A tuple (archive_name, archive_url) that represents a		destination: A tuple (archive_name, archive_url) that represents a
remote object storage as in the 'archive' table.		remote object storage as in the 'archive' table.
content: A list of sha1 that represents the content this copier		content: A list of sha1 that represents the content this copier
have to archive.		have to archive.
master_storage (Storage): The master storage of the system that		master_storage (Storage): The master storage of the system that
contains the data to archive.		contains the data to archive.
"""		"""
_name, self.url = destination		_name, self.url = destination
self.content_ids = content		self.content_ids = content
self.server = RemoteObjStorage(self.url)		self.server = RemoteObjStorage(self.url)
self.master_storage = master_storage		self.master_objstorage = master_objstorage

def run(self):		def run(self):
""" Do the copy on the backup storage.		""" Do the copy on the backup storage.

Run the archiver copier in order to copy the required content		Run the archiver copier in order to copy the required content
into the current destination.		into the current destination.
The content which corresponds to the sha1 in self.content_ids		The content which corresponds to the sha1 in self.content_ids
will be fetched from the master_storage and then copied into		will be fetched from the master_storage and then copied into
the backup object storage.		the backup object storage.

Returns:		Returns:
A boolean that indicates if the whole content have been copied.		A boolean that indicates if the whole content have been copied.
"""		"""
self.content_ids = list(map(lambda x: hashutil.hex_to_hash(x[2:]),		self.content_ids = map(lambda x: hashutil.hex_to_hash(x[2:]),
self.content_ids))		self.content_ids)
contents = self.master_storage.content_get(self.content_ids)
try:		try:
for content in contents:		for content_id in self.content_ids:
content_data = content['data']		content = self.master_objstorage.get(content_id)
self.server.content_add(content_data)		self.server.content_add(content, content_id)
except:		except:
return False		return False

return True		return True

View Options

swh/storage/archiver/director.py

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

import swh
import logging		import logging
import click		import click

from datetime import datetime		from datetime import datetime

from swh.core import hashutil, config		from swh.core import hashutil, config
		from swh.objstorage import PathSlicingObjStorage
		from swh.objstorage.api.client import RemoteObjStorage
from swh.scheduler.celery_backend.config import app		from swh.scheduler.celery_backend.config import app

from . import tasks # NOQA		from . import tasks # NOQA
from .storage import ArchiverStorage		from .storage import ArchiverStorage


DEFAULT_CONFIG = {		DEFAULT_CONFIG = {
		'objstorage_type': ('str', 'local_storage'),
'objstorage_path': ('str', '/tmp/swh-storage/objects'),		'objstorage_path': ('str', '/tmp/swh-storage/objects'),
		'objstorage_slicing': ('str', '0:2/2:4/4:6'),
		'objstorage_url': ('str', 'http://localhost:5003/'),

'batch_max_size': ('int', 50),		'batch_max_size': ('int', 50),
'archival_max_age': ('int', 3600),		'archival_max_age': ('int', 3600),
'retention_policy': ('int', 2),		'retention_policy': ('int', 2),
'asynchronous': ('bool', True),		'asynchronous': ('bool', True),

'dbconn': ('str', 'dbname=softwareheritage-archiver-dev user=guest'),		'dbconn': ('str', 'dbname=softwareheritage-archiver-dev user=guest')
'dbconn_storage': ('str', 'dbname=softwareheritage-dev user=guest')
}		}

task_name = 'swh.storage.archiver.tasks.SWHArchiverTask'		task_name = 'swh.storage.archiver.tasks.SWHArchiverTask'

logger = logging.getLogger()		logger = logging.getLogger()


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

The archiver director processes the files in the local storage in order		The archiver director processes the files in the local storage in order
to know which one needs archival and it delegates this task to		to know which one needs archival and it delegates this task to
archiver workers.		archiver workers.
Attributes:		Attributes:
master_storage: the local storage of the master server.		master_objstorage: the local storage of the master server.
slave_storages: Iterable of remote obj storages to the slaves servers		master_objstorage_args (dict): arguments of the master objstorage
used for backup.		initialization.

		archiver_storage: a wrapper for archiver db operations.
		db_conn_archiver: Either a libpq connection string,
		or a psycopg2 connection for the archiver db.

		slave_objstorages: Iterable of remote obj storages to the slaves
		servers used for backup.
config: Archiver_configuration. A dictionary that must contain		config: Archiver_configuration. A dictionary that must contain
the following keys.		the following keys:
objstorage_path (string): master's objstorage path
		objstorage_type (str): type of objstorage used (local_storage
		or remote_storage).
		If the storage is local, the arguments keys must be present
		objstorage_path (str): master's objstorage path
		objstorage_slicing (str): masters's objstorage slicing
		Otherwise, if it's a remote objstorage, the keys must be:
		objstorage_url (str): url of the remote objstorage

batch_max_size (int): The number of content items that can be		batch_max_size (int): The number of content items that can be
given to the same archiver worker.		given to the same archiver worker.
archival_max_age (int): Delay given to the worker to copy all		archival_max_age (int): Delay given to the worker to copy all
the files in a given batch.		the files in a given batch.
retention_policy (int): Required number of copies for the		retention_policy (int): Required number of copies for the
content to be considered safe.		content to be considered safe.
asynchronous (boolean): Indicate whenever the archival should		asynchronous (boolean): Indicate whenever the archival should
run in asynchronous mode or not.		run in asynchronous mode or not.
"""		"""

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

Args:		Args:
db_conn_archiver: Either a libpq connection string,		db_conn_archiver: Either a libpq connection string,
or a psycopg2 connection for the archiver db connection.		or a psycopg2 connection for the archiver db.
db_conn_storage: Either a libpq connection string,
or a psycopg2 connection for the db storage connection.
config: Archiver_configuration. A dictionary that must contain		config: Archiver_configuration. A dictionary that must contain
the following keys.		the following keys:
objstorage_path (string): master's objstorage path
		objstorage_type (str): type of objstorage used
		(local_objstorage or remote_objstorage).
		If the storage is local, the arguments keys must be present
		objstorage_path (str): master's objstorage path
		objstorage_slicing (str): masters's objstorage slicing
		Otherwise, if it's a remote objstorage, the keys must be:
		objstorage_url (str): url of the remote objstorage

batch_max_size (int): The number of content items that can be		batch_max_size (int): The number of content items that can be
given to the same archiver worker.		given to the same archiver worker.
archival_max_age (int): Delay given to the worker to copy all		archival_max_age (int): Delay given to the worker to copy all
the files in a given batch.		the files in a given batch.
retention_policy (int): Required number of copies for the		retention_policy (int): Required number of copies for the
content to be considered safe.		content to be considered safe.
asynchronous (boolean): Indicate whenever the archival should		asynchronous (boolean): Indicate whenever the archival should
run in asynchronous mode or not.		run in asynchronous mode or not.
"""		"""
# Get the local storage of the master and remote ones for the slaves.		# Get the slave storages
self.db_conn_archiver = db_conn_archiver		self.db_conn_archiver = db_conn_archiver
self.archiver_storage = ArchiverStorage(db_conn_archiver)		self.archiver_storage = ArchiverStorage(db_conn_archiver)
		self.slave_objstorages = {
self.master_storage_args = [db_conn_storage, config['objstorage_path']]
master_storage = swh.storage.get_storage('local_storage',
self.master_storage_args)
slaves = {
id: url		id: url
for id, url		for id, url
in self.archiver_storage.archive_ls()		in self.archiver_storage.archive_ls()
}		}
		# Check that there is enough backup servers for the retention policy
		if config['retention_policy'] > len(self.slave_objstorages) + 1:
		raise ValueError(
		"Can't have a retention policy of %d with %d backup servers"
		% (config['retention_policy'], len(self.slave_objstorages))
		)

# TODO Database should be initialized somehow before going in		# Get the master storage that contains content to be archived
# production. For now, assumes that the database contains		if config['objstorage_type'] == 'local_objstorage':
# data for all the current content.		master_objstorage_args = {
		'root': config['objstorage_path'],
		'slicing': config['objstorage_slicing']
		}
		master_objstorage = PathSlicingObjStorage(
		**master_objstorage_args
		)
		elif config['objstorage_type'] == 'remote_objstorage':
		master_objstorage_args = {'base_url': config['objstorage_url']}
		master_objstorage = RemoteObjStorage(**master_objstorage_args)
		else:
		raise ValueError(
		'Unknow objstorage class `%s`' % config['objstorage_type']
		)
		self.master_objstorage = master_objstorage
		self.master_objstorage_args = master_objstorage_args

self.master_storage = master_storage		# Keep the full configuration
self.slave_storages = slaves
self.config = config		self.config = config

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

The archiver director will check all the contents of the archiver		The archiver director will check all the contents of the archiver
database and do the required backup jobs.		database and do the required backup jobs.
"""		"""
if self.config['asynchronous']:		if self.config['asynchronous']:
run_fn = self.run_async_worker		run_fn = self.run_async_worker
else:		else:
run_fn = self.run_sync_worker		run_fn = self.run_sync_worker

for batch in self.get_unarchived_content():		for batch in self.get_unarchived_content():
run_fn(batch)		run_fn(batch)

def run_async_worker(self, batch):		def run_async_worker(self, batch):
""" Produce a worker that will be added to the task queue.		""" Produce a worker that will be added to the task queue.
"""		"""
task = app.tasks[task_name]		task = app.tasks[task_name]
task.delay(batch,		task.delay(batch,
archiver_args=self.db_conn_archiver,		archiver_args=self.db_conn_archiver,
master_storage_args=self.master_storage_args,		master_objstorage_args=self.master_objstorage_args,
slave_storages=self.slave_storages,		slave_objstorages=self.slave_objstorages,
config=self.config)		config=self.config)

def run_sync_worker(self, batch):		def run_sync_worker(self, batch):
""" Run synchronously a worker on the given batch.		""" Run synchronously a worker on the given batch.
"""		"""
task = app.tasks[task_name]		task = app.tasks[task_name]
task(batch,		task(batch,
archiver_args=self.db_conn_archiver,		archiver_args=self.db_conn_archiver,
master_storage_args=self.master_storage_args,		master_objstorage_args=self.master_objstorage_args,
slave_storages=self.slave_storages,		slave_objstorages=self.slave_objstorages,
config=self.config)		config=self.config)

def get_unarchived_content(self):		def get_unarchived_content(self):
""" Get contents that need to be archived.		""" Get contents that need to be archived.

Yields:		Yields:
A batch of contents. Batches are dictionaries which associates		A batch of contents. Batches are dictionaries which associates
a content id to the data about servers that contains it or not.		a content id to the data about servers that contains it or not.
Show All 24 Lines	def get_unarchived_content(self):
db_content_id = '\\x' + hashutil.hash_to_hex(content_id[0])		db_content_id = '\\x' + hashutil.hash_to_hex(content_id[0])

# Fetch the datas about archival status of the content		# Fetch the datas about archival status of the content
backups = self.archiver_storage.content_archive_get(		backups = self.archiver_storage.content_archive_get(
content=db_content_id		content=db_content_id
)		)
for _content_id, server_id, status, mtime in backups:		for _content_id, server_id, status, mtime in backups:
virtual_status = self.get_virtual_status(status, mtime)		virtual_status = self.get_virtual_status(status, mtime)
server_data = (server_id, self.slave_storages[server_id])		server_data = (server_id, self.slave_objstorages[server_id])

missing_copy.setdefault(		missing_copy.setdefault(
db_content_id,		db_content_id,
{'present': [], 'missing': []}		{'present': [], 'missing': []}
).setdefault(virtual_status, []).append(server_data)		).setdefault(virtual_status, []).append(server_data)

# Check the content before archival.		# Check the content before archival.
try:		try:
self.master_storage.objstorage.check(content_id[0])		self.master_objstorage.check(content_id[0])
except Exception as e:		except Exception as e:
# Exception can be Error or ObjNotFoundError.		# Exception can be Error or ObjNotFoundError.
logger.error(e)		logger.error(e)
# TODO Do something to restore the content?		# TODO Do something to restore the content?

if len(missing_copy) >= self.config['batch_max_size']:		if len(missing_copy) >= self.config['batch_max_size']:
yield missing_copy		yield missing_copy
missing_copy = {}		missing_copy = {}
Show All 40 Lines	def get_virtual_status(self, status, mtime):
raise ValueError("status must be either 'present', 'missing' "		raise ValueError("status must be either 'present', 'missing' "
"or 'ongoing'")		"or 'ongoing'")


@click.command()		@click.command()
@click.argument('config-path', required=1)		@click.argument('config-path', required=1)
@click.option('--dbconn', default=DEFAULT_CONFIG['dbconn'][1],		@click.option('--dbconn', default=DEFAULT_CONFIG['dbconn'][1],
help="Connection string for the archiver database")		help="Connection string for the archiver database")
@click.option('--dbconn-storage', default=DEFAULT_CONFIG['dbconn_storage'][1],
help="Connection string for the storage database")
@click.option('--async/--sync', default=DEFAULT_CONFIG['asynchronous'][1],		@click.option('--async/--sync', default=DEFAULT_CONFIG['asynchronous'][1],
help="Indicates if the archiver should run asynchronously")		help="Indicates if the archiver should run asynchronously")
def launch(config_path, dbconn, dbconn_storage, async):		def launch(config_path, dbconn, dbconn_storage, async):
		ardumontAuthorUnsubmitted Not Done Inline Actions You forgot to remove the argument dbconn_storage here. ardumont: You forgot to remove the argument dbconn_storage here.
# The configuration have following priority :		# The configuration have following priority :
# command line > file config > default config		# command line > file config > default config
cl_config = {		cl_config = {
'dbconn': dbconn,		'dbconn': dbconn,
'dbconn_storage': dbconn_storage,
'asynchronous': async		'asynchronous': async
}		}
conf = config.read(config_path, DEFAULT_CONFIG)		conf = config.read(config_path, DEFAULT_CONFIG)
conf.update(cl_config)		conf.update(cl_config)
# Create connection data and run the archiver.		# Create connection data and run the archiver.
archiver = ArchiverDirector(conf['dbconn'], conf['dbconn_storage'], conf)		archiver = ArchiverDirector(conf['dbconn'], conf)
logger.info("Starting an archival at", datetime.now())		logger.info("Starting an archival at", datetime.now())
archiver.run()		archiver.run()


if __name__ == '__main__':		if __name__ == '__main__':
launch()		launch()

View Options

swh/storage/archiver/tasks.py

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

	from swh.scheduler.task import Task			from swh.scheduler.task import Task
	from .worker import ArchiverWorker			from .worker import ArchiverWorker


	class SWHArchiverTask(Task):			class SWHArchiverTask(Task):
	""" Main task that archive a batch of content.			""" Main task that archive a batch of content.
	"""			"""
	task_queue = 'swh_storage_archive_worker'			task_queue = 'swh_storage_archive_worker'

	def run(self, batch, archiver_args, master_storage_args, slave_storages,			def run(self, batch, archiver_args, master_objstorage_args,
	config):			slave_objstorages, config):
	aw = ArchiverWorker(batch, archiver_args, master_storage_args,			aw = ArchiverWorker(batch, archiver_args, master_objstorage_args,
	slave_storages, config)			slave_objstorages, config)
	if aw.run():			if aw.run():
	self.log("Successful backup for a batch of size %s" % len(batch))			self.log("Successful backup for a batch of size %s" % len(batch))

View Options

swh/storage/archiver/worker.py

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

import random		import random
import logging		import logging

		from datetime import datetime

		from swh.objstorage import PathSlicingObjStorage
		from swh.objstorage.api.client import RemoteObjStorage

from .storage import ArchiverStorage		from .storage import ArchiverStorage
from .copier import ArchiverCopier		from .copier import ArchiverCopier
from .. import get_storage

from datetime import datetime

logger = logging.getLogger()		logger = logging.getLogger()


class ArchiverWorker():		class ArchiverWorker():
""" Do the required backups on a given batch of contents.		""" Do the required backups on a given batch of contents.

Process the content of a content batch in order to do the needed backups on		Process the content of a content batch in order to do the needed backups on
the slaves servers.		the slaves servers.

Attributes:		Attributes:
batch: The content this worker has to archive, which is a dictionary		batch: The content this worker has to archive, which is a dictionary
that associates a content's sha1 id to the list of servers where		that associates a content's sha1 id to the list of servers where
the content is present or missing		the content is present or missing
(see ArchiverDirector::get_unarchived_content).		(see ArchiverDirector::get_unarchived_content).
master_storage_args: The connection argument to initialize the		master_objstorage_args: The connection argument to initialize the
master storage with the db connection url & the object storage		master storage with the db connection url & the object storage
path.		path.
slave_storages: A map that associates server_id to the remote server.		slave_objstorages: A map that associates server_id to the remote server
config: Archiver_configuration. A dictionary that must contains		config: Archiver_configuration. A dictionary that must contains
the following keys.		the following keys.
objstorage_path (string): the path of the objstorage of the		objstorage_path (string): the path of the objstorage of the
master.		master.
batch_max_size (int): The number of content items that can be		batch_max_size (int): The number of content items that can be
given to the same archiver worker.		given to the same archiver worker.
archival_max_age (int): Delay given to the worker to copy all		archival_max_age (int): Delay given to the worker to copy all
the files in a given batch.		the files in a given batch.
retention_policy (int): Required number of copies for the		retention_policy (int): Required number of copies for the
content to be considered safe.		content to be considered safe.
asynchronous (boolean): Indicate whenever the archival should		asynchronous (boolean): Indicate whenever the archival should
run in asynchronous mode or not.		run in asynchronous mode or not.
"""		"""
def __init__(self, batch, archiver_args, master_storage_args,		def __init__(self, batch, archiver_args, master_objstorage_args,
slave_storages, config):		slave_objstorages, config):
""" Constructor of the ArchiverWorker class.		""" Constructor of the ArchiverWorker class.

Args:		Args:
batch: A batch of content, which is a dictionary that associates		batch: A batch of content, which is a dictionary that associates
a content's sha1 id to the list of servers where the content		a content's sha1 id to the list of servers where the content
is present.		is present.
archiver_args: The archiver's arguments to establish connection to		archiver_args: The archiver's arguments to establish connection to
db.		db.
master_storage_args: The master storage arguments.		master_objstorage_args: The master storage arguments.
slave_storages: A map that associates server_id to the remote		slave_objstorages: A map that associates server_id to the remote
server.		server.
config: Archiver_configuration. A dictionary that must contains		config: Archiver_configuration. A dictionary that must contains
the following keys.		the following keys.
objstorage_path (string): the path of the objstorage of the		objstorage_path (string): the path of the objstorage of the
master.		master.
batch_max_size (int): The number of content items that can be		batch_max_size (int): The number of content items that can be
given to the same archiver worker.		given to the same archiver worker.
archival_max_age (int): Delay given to the worker to copy all		archival_max_age (int): Delay given to the worker to copy all
the files in a given batch.		the files in a given batch.
retention_policy (int): Required number of copies for the		retention_policy (int): Required number of copies for the
content to be considered safe.		content to be considered safe.
asynchronous (boolean): Indicate whenever the archival should		asynchronous (boolean): Indicate whenever the archival should
run in asynchronous mode or not.		run in asynchronous mode or not.
"""		"""
self.batch = batch		self.batch = batch
self.archiver_storage = ArchiverStorage(archiver_args)		self.archiver_storage = ArchiverStorage(archiver_args)
self.master_storage = get_storage('local_storage', master_storage_args)		self.slave_objstorages = slave_objstorages
self.slave_storages = slave_storages
self.config = config		self.config = config

def __choose_backup_servers(self, allowed_storage, backup_number):		if config['objstorage_type'] == 'local_objstorage':
		master_objstorage = PathSlicingObjStorage(**master_objstorage_args)
		else:
		master_objstorage = RemoteObjStorage(**master_objstorage_args)
		self.master_objstorage = master_objstorage

		def _choose_backup_servers(self, allowed_storage, backup_number):
""" Choose the slave servers for archival.		""" Choose the slave servers for archival.

Choose the given amount of servers among those which don't already		Choose the given amount of servers among those which don't already
contain a copy of the content.		contain a copy of the content.

Args:		Args:
allowed_storage: servers when the content is not already present.		allowed_storage: servers when the content is not already present.
backup_number (int): The number of servers we have to choose in		backup_number (int): The number of servers we have to choose in
order to fullfill the objective.		order to fullfill the objective.
"""		"""
# In case there is not enough backup servers to get all the backups		# In case there is not enough backup servers to get all the backups
# we need, just do our best.		# we need, just do our best.
# TODO such situation can only be caused by an incorrect configuration		# Such situation should not happen.
# setting. Do a verification previously.
backup_number = min(backup_number, len(allowed_storage))		backup_number = min(backup_number, len(allowed_storage))

# TODO Find a better (or a good) policy to choose the backup servers.		# TODO Find a better (or a good) policy to choose the backup servers.
# The random choice should be equivalently distributed between		# The random choice should be equivalently distributed between
# servers for a great amount of data, but don't take care of servers		# servers for a great amount of data, but don't take care of servers
# capacities.		# capacities.
return random.sample(allowed_storage, backup_number)		return random.sample(allowed_storage, backup_number)

def __get_archival_status(self, content_id, server):		def _get_archival_status(self, content_id, server):
""" Get the archival status of the required content.		""" Get the archival status of the required content.

Attributes:		Attributes:
content_id (string): Sha1 of the content.		content_id (string): Sha1 of the content.
server: Tuple (archive_id, archive_url) of the archive server.		server: Tuple (archive_id, archive_url) of the archive server.
Returns:		Returns:
A dictionary that contains all the required data : 'content_id',		A dictionary that contains all the required data : 'content_id',
'archive_id', 'status', and 'mtime'		'archive_id', 'status', and 'mtime'
"""		"""
t, = list(		t, = list(
self.archiver_storage.content_archive_get(content_id, server[0])		self.archiver_storage.content_archive_get(content_id, server[0])
)		)
return {		return {
'content_id': t[0],		'content_id': t[0],
'archive_id': t[1],		'archive_id': t[1],
'status': t[2],		'status': t[2],
'mtime': t[3]		'mtime': t[3]
}		}

def __content_archive_update(self, content_id, archive_id,		def _content_archive_update(self, content_id, archive_id,
new_status=None):		new_status=None):
""" Update the status of a archive content and set it's mtime to now()		""" Update the status of a archive content and set it's mtime to now()

Change the last modification time of an archived content and change		Change the last modification time of an archived content and change
its status to the given one.		its status to the given one.

Args:		Args:
content_id (string): The content id.		content_id (string): The content id.
archive_id (string): The id of the concerned archive.		archive_id (string): The id of the concerned archive.
Show All 12 Lines	def need_archival(self, content, destination):

Filter function that returns True if a given content		Filter function that returns True if a given content
still require to be archived.		still require to be archived.

Args:		Args:
content (str): Sha1 of a content.		content (str): Sha1 of a content.
destination: Tuple (archive id, archive url).		destination: Tuple (archive id, archive url).
"""		"""
archival_status = self.__get_archival_status(		archival_status = self._get_archival_status(
content,		content,
destination		destination
)		)
status = archival_status['status']		status = archival_status['status']
mtime = archival_status['mtime']		mtime = archival_status['mtime']
# If the archive is already present, no need to backup.		# If the archive is already present, no need to backup.
if status == 'present':		if status == 'present':
return False		return False
Show All 22 Lines	def sort_content_by_archive(self):
The created mapping.		The created mapping.
"""		"""
slaves_copy = {}		slaves_copy = {}
for content_id in self.batch:		for content_id in self.batch:
# Choose some servers to upload the content among the missing ones.		# Choose some servers to upload the content among the missing ones.
server_data = self.batch[content_id]		server_data = self.batch[content_id]
nb_present = len(server_data['present'])		nb_present = len(server_data['present'])
nb_backup = self.config['retention_policy'] - nb_present		nb_backup = self.config['retention_policy'] - nb_present
backup_servers = self.__choose_backup_servers(		backup_servers = self._choose_backup_servers(
server_data['missing'],		server_data['missing'],
nb_backup		nb_backup
)		)
# Fill the map destination -> content to upload		# Fill the map destination -> content to upload
for server in backup_servers:		for server in backup_servers:
slaves_copy.setdefault(server, []).append(content_id)		slaves_copy.setdefault(server, []).append(content_id)
return slaves_copy		return slaves_copy

Show All 10 Lines	def run(self):
# job have been done by another worker.		# job have been done by another worker.
for destination in slaves_copy:		for destination in slaves_copy:
# list() is needed because filter's result will be consumed twice.		# list() is needed because filter's result will be consumed twice.
slaves_copy[destination] = list(filter(		slaves_copy[destination] = list(filter(
lambda content_id: self.need_archival(content_id, destination),		lambda content_id: self.need_archival(content_id, destination),
slaves_copy[destination]		slaves_copy[destination]
))		))
for content_id in slaves_copy[destination]:		for content_id in slaves_copy[destination]:
self.__content_archive_update(content_id, destination[0],		self._content_archive_update(content_id, destination[0],
new_status='ongoing')		new_status='ongoing')

# Spawn a copier for each destination		# Spawn a copier for each destination
for destination in slaves_copy:		for destination in slaves_copy:
try:		try:
self.run_copier(destination, slaves_copy[destination])		self.run_copier(destination, slaves_copy[destination])
except:		except:
logger.error('Unable to copy a batch to %s' % destination)		logger.error('Unable to copy a batch to %s' % destination)

def run_copier(self, destination, contents):		def run_copier(self, destination, contents):
""" Run a copier in order to archive the given contents		""" Run a copier in order to archive the given contents

Upload the given contents to the given archive.		Upload the given contents to the given archive.
If the process fail, the whole content is considered uncopied		If the process fail, the whole content is considered uncopied
and remains 'ongoing', waiting to be rescheduled as there is		and remains 'ongoing', waiting to be rescheduled as there is
a delay.		a delay.

Attributes:		Attributes:
destination: Tuple (archive_id, archive_url) of the destination.		destination: Tuple (archive_id, archive_url) of the destination.
contents: List of contents to archive.		contents: List of contents to archive.
"""		"""
ac = ArchiverCopier(destination, contents, self.master_storage)		ac = ArchiverCopier(destination, contents, self.master_objstorage)
if ac.run():		if ac.run():
# Once the archival complete, update the database.		# Once the archival complete, update the database.
for content_id in contents:		for content_id in contents:
self.__content_archive_update(content_id, destination[0],		self._content_archive_update(content_id, destination[0],
new_status='present')		new_status='present')

View Options

swh/storage/tests/test_archiver.py

archiver: change the master's storage to an object storageClosedPublicActions

Active Operations

Details

Diff Detail

Event Timeline

Revision ContentsChangeset List

Diff 256

swh/storage/archiver/copier.py

swh/storage/archiver/director.py

swh/storage/archiver/tasks.py

swh/storage/archiver/worker.py

swh/storage/tests/test_archiver.py

archiver: change the master's storage to an object storage
ClosedPublic
Actions

Revision Contents
Changeset List