Differential D4401

Remove conflict_update/policy_update option from BaseIndexer.run()
ClosedPublic
Actions

Authored by vlorentz on Nov 3 2020, 2:26 PM.

Details

Reviewers

ardumont

Group Reviewers

Reviewers

Commits

rDCIDX8e69633f639e: Remove conflict_update/policy_update option from BaseIndexer.run()

Summary

By removing the False/ignore-dups behavior, which we didn't use.

The main motivation is to make _add endpoints write to Kafka in a
future commit, as Kafka's semantics are closer to True/update-dups
than False/ignore-dups.

Diff Detail

Repository

rDCIDX Metadata indexer

Lint

Automatic diff as part of commit; lint not applicable.

Unit

Automatic diff as part of commit; unit tests not applicable.

Event Timeline

vlorentz created this revision.Nov 3 2020, 2:26 PM

Herald added a reviewer: Reviewers. · View Herald TranscriptNov 3 2020, 2:26 PM

vlorentz mentioned this in D4397: Remove conflict_update/policy_update option..Nov 3 2020, 2:27 PM

vlorentz added a child revision: D4400: Remove conflict_update option from _add() endpoints..

Build is green

Patch application report for D4401 (id=15589)

Rebasing onto 94c8259193...

Current branch diff-target is up to date.

Changes applied before test

commit 8e69633f639e31cf5efa3edd784397b351d3b2db
Author: Valentin Lorentz <vlorentz@softwareheritage.org>
Date:   Tue Nov 3 14:19:23 2020 +0100

    Remove conflict_update/policy_update option from BaseIndexer.run()
    
    By removing the False/ignore-dups behavior, which we didn't use.
    
    The main motivation is to make _add endpoints write to Kafka in a
    future commit, as Kafka's semantics are closer to True/update-dups
    than False/ignore-dups.

See https://jenkins.softwareheritage.org/job/DCIDX/job/tests-on-diff/109/ for more details.

Harbormaster completed remote builds in B16812: Diff 15589.Nov 3 2020, 2:28 PM

ardumont accepted this revision.Nov 3 2020, 2:34 PM

This revision is now accepted and ready to land.Nov 3 2020, 2:34 PM

Closed by commit rDCIDX8e69633f639e: Remove conflict_update/policy_update option from BaseIndexer.run() (authored by vlorentz). · Explain WhyNov 3 2020, 5:03 PM

This revision was automatically updated to reflect the committed changes.

vlorentz added a commit: rDCIDX8e69633f639e: Remove conflict_update/policy_update option from BaseIndexer.run().

Revision Contents
Changeset List

Path

Size

swh/

indexer/

2 lines

8 lines

9 lines

57 lines

1 line

31 lines

9 lines

4 lines

tests/

test_cli.py

2 lines

test_indexer.py

20 lines

test_journal_client.py

16 lines

test_metadata.py

6 lines

test_origin_head.py

2 lines

utils.py

7 lines

Diff 15611

View Options

swh/indexer/cli.py

Show First 20 Lines • Show All 199 Lines • ▼ Show 20 Lines	):
"""Schedules indexing tasks for origins that were already indexed."""		"""Schedules indexing tasks for origins that were already indexed."""
from swh.scheduler.cli_utils import schedule_origin_batches		from swh.scheduler.cli_utils import schedule_origin_batches

idx_storage = ctx.obj["indexer_storage"]		idx_storage = ctx.obj["indexer_storage"]
scheduler = ctx.obj["scheduler"]		scheduler = ctx.obj["scheduler"]

origins = list_origins_by_producer(idx_storage, mappings, tool_ids)		origins = list_origins_by_producer(idx_storage, mappings, tool_ids)

kwargs = {"policy_update": "update-dups", "retries_left": 1}		kwargs = {"retries_left": 1}
schedule_origin_batches(scheduler, task_type, origins, origin_batch_size, kwargs)		schedule_origin_batches(scheduler, task_type, origins, origin_batch_size, kwargs)


@indexer_cli_group.command("journal-client")		@indexer_cli_group.command("journal-client")
@click.option("--scheduler-url", "-s", default=None, help="URL of the scheduler API")		@click.option("--scheduler-url", "-s", default=None, help="URL of the scheduler API")
@click.option(		@click.option(
"--origin-metadata-task-type",		"--origin-metadata-task-type",
default="index-origin-metadata",		default="index-origin-metadata",
▲ Show 20 Lines • Show All 85 Lines • Show Last 20 Lines

View Options

swh/indexer/ctags.py

Show First 20 Lines • Show All 130 Lines • ▼ Show 20 Lines	) -> List[ContentCtagsRow]:
ContentCtagsRow(		ContentCtagsRow(
id=id, indexer_configuration_id=self.tool["id"], **ctag_kwargs,		id=id, indexer_configuration_id=self.tool["id"], **ctag_kwargs,
)		)
)		)

return ctags		return ctags

def persist_index_computations(		def persist_index_computations(
self, results: List[ContentCtagsRow], policy_update: str		self, results: List[ContentCtagsRow]
) -> Dict[str, int]:		) -> Dict[str, int]:
"""Persist the results in storage.		"""Persist the results in storage.

Args:		Args:
results: list of ctags returned by index()		results: list of ctags returned by index()
policy_update: either 'update-dups' or 'ignore-dups' to
respectively update duplicates or ignore them

"""		"""
return self.idx_storage.content_ctags_add(		return self.idx_storage.content_ctags_add(results)
results, conflict_update=(policy_update == "update-dups")
)

View Options

swh/indexer/fossology_license.py

Show First 20 Lines • Show All 108 Lines • ▼ Show 20 Lines	) -> List[ContentLicenseRow]:
return [		return [
ContentLicenseRow(		ContentLicenseRow(
id=id, indexer_configuration_id=self.tool["id"], license=license,		id=id, indexer_configuration_id=self.tool["id"], license=license,
)		)
for license in properties["licenses"]		for license in properties["licenses"]
]		]

def persist_index_computations(		def persist_index_computations(
self, results: List[ContentLicenseRow], policy_update: str		self, results: List[ContentLicenseRow]
) -> Dict[str, int]:		) -> Dict[str, int]:
"""Persist the results in storage.		"""Persist the results in storage.

Args:		Args:
results: list of content_license dict with the		results: list of content_license dict with the
following keys:		following keys:

- id (bytes): content's identifier (sha1)		- id (bytes): content's identifier (sha1)
- license (bytes): license in bytes		- license (bytes): license in bytes
- path (bytes): path		- path (bytes): path

policy_update: either 'update-dups' or 'ignore-dups' to
respectively update duplicates or ignore them

"""		"""
return self.idx_storage.content_fossology_license_add(		return self.idx_storage.content_fossology_license_add(results)
results, conflict_update=(policy_update == "update-dups")
)


class FossologyLicenseIndexer(		class FossologyLicenseIndexer(
MixinFossologyLicenseIndexer, ContentIndexer[ContentLicenseRow]		MixinFossologyLicenseIndexer, ContentIndexer[ContentLicenseRow]
):		):
"""Indexer in charge of:		"""Indexer in charge of:

- filtering out content already indexed		- filtering out content already indexed
▲ Show 20 Lines • Show All 47 Lines • Show Last 20 Lines

View Options

swh/indexer/indexer.py

# Copyright (C) 2016-2020 The Software Heritage developers		# Copyright (C) 2016-2020 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 abc		import abc
from contextlib import contextmanager		from contextlib import contextmanager
import logging		import logging
import os		import os
import shutil		import shutil
import tempfile		import tempfile
from typing import Any, Dict, Generic, Iterator, List, Optional, Set, TypeVar, Union		from typing import Any, Dict, Generic, Iterator, List, Optional, Set, TypeVar, Union
		import warnings

from swh.core import utils		from swh.core import utils
from swh.core.config import load_from_envvar, merge_configs		from swh.core.config import load_from_envvar, merge_configs
from swh.indexer.storage import INDEXER_CFG_KEY, PagedResult, Sha1, get_indexer_storage		from swh.indexer.storage import INDEXER_CFG_KEY, PagedResult, Sha1, get_indexer_storage
from swh.indexer.storage.interface import IndexerStorageInterface		from swh.indexer.storage.interface import IndexerStorageInterface
from swh.model import hashutil		from swh.model import hashutil
from swh.model.model import Revision, Sha1Git		from swh.model.model import Revision, Sha1Git
from swh.objstorage.exc import ObjNotFoundError		from swh.objstorage.exc import ObjNotFoundError
▲ Show 20 Lines • Show All 224 Lines • ▼ Show 20 Lines	def filter(self, ids: List[TId]) -> Iterator[TId]:

Yields:		Yields:
iterator of missing ids		iterator of missing ids

"""		"""
yield from ids		yield from ids

@abc.abstractmethod		@abc.abstractmethod
def persist_index_computations(		def persist_index_computations(self, results: List[TResult]) -> Dict[str, int]:
self, results: List[TResult], policy_update: str
) -> Dict[str, int]:
"""Persist the computation resulting from the index.		"""Persist the computation resulting from the index.

Args:		Args:

results: List of results. One result is the		results: List of results. One result is the
result of the index function.		result of the index function.
policy_update: either 'update-dups' or 'ignore-dups' to
respectively update duplicates or ignore them

Returns:		Returns:
a summary dict of what has been inserted in the storage		a summary dict of what has been inserted in the storage

"""		"""
return {}		return {}


class ContentIndexer(BaseIndexer[Sha1, bytes, TResult], Generic[TResult]):		class ContentIndexer(BaseIndexer[Sha1, bytes, TResult], Generic[TResult]):
"""A content indexer working on a list of ids directly.		"""A content indexer working on a list of ids directly.

To work on indexer partition, use the :class:`ContentPartitionIndexer`		To work on indexer partition, use the :class:`ContentPartitionIndexer`
instead.		instead.

Note: :class:`ContentIndexer` is not an instantiable object. To		Note: :class:`ContentIndexer` is not an instantiable object. To
use it, one should inherit from this class and override the		use it, one should inherit from this class and override the
methods mentioned in the :class:`BaseIndexer` class.		methods mentioned in the :class:`BaseIndexer` class.

"""		"""

def run(self, ids: List[Sha1], policy_update: str, **kwargs) -> Dict:		def run(self, ids: List[Sha1], **kwargs) -> Dict:
"""Given a list of ids:		"""Given a list of ids:

- retrieve the content from the storage		- retrieve the content from the storage
- execute the indexing computations		- execute the indexing computations
- store the results (according to policy_update)		- store the results

Args:		Args:
ids (Iterable[Union[bytes, str]]): sha1's identifier list		ids (Iterable[Union[bytes, str]]): sha1's identifier list
policy_update (str): either 'update-dups' or 'ignore-dups' to
respectively update duplicates or ignore
them
**kwargs: passed to the `index` method		**kwargs: passed to the `index` method

Returns:		Returns:
A summary Dict of the task's status		A summary Dict of the task's status

"""		"""
		if "policy_update" in kwargs:
		warnings.warn(
		"'policy_update' argument is deprecated and ignored.",
		DeprecationWarning,
		)
		del kwargs["policy_update"]

sha1s = [		sha1s = [
hashutil.hash_to_bytes(id_) if isinstance(id_, str) else id_ for id_ in ids		hashutil.hash_to_bytes(id_) if isinstance(id_, str) else id_ for id_ in ids
]		]
results = []		results = []
summary: Dict = {"status": "uneventful"}		summary: Dict = {"status": "uneventful"}
try:		try:
for sha1 in sha1s:		for sha1 in sha1s:
try:		try:
raw_content = self.objstorage.get(sha1)		raw_content = self.objstorage.get(sha1)
except ObjNotFoundError:		except ObjNotFoundError:
self.log.warning(		self.log.warning(
"Content %s not found in objstorage"		"Content %s not found in objstorage"
% hashutil.hash_to_hex(sha1)		% hashutil.hash_to_hex(sha1)
)		)
continue		continue
res = self.index(sha1, raw_content, **kwargs)		res = self.index(sha1, raw_content, **kwargs)
if res: # If no results, skip it		if res: # If no results, skip it
results.extend(res)		results.extend(res)
summary["status"] = "eventful"		summary["status"] = "eventful"
summary = self.persist_index_computations(results, policy_update)		summary = self.persist_index_computations(results)
self.results = results		self.results = results
except Exception:		except Exception:
if not self.catch_exceptions:		if not self.catch_exceptions:
raise		raise
self.log.exception("Problem when reading contents metadata.")		self.log.exception("Problem when reading contents metadata.")
summary["status"] = "failed"		summary["status"] = "failed"
return summary		return summary

▲ Show 20 Lines • Show All 143 Lines • ▼ Show 20 Lines	) -> Dict:
partition_id, nb_partitions		partition_id, nb_partitions
)		)
else:		else:
gen = self._index_contents(partition_id, nb_partitions, indexed=set([]))		gen = self._index_contents(partition_id, nb_partitions, indexed=set([]))

count_object_added_key: Optional[str] = None		count_object_added_key: Optional[str] = None

for contents in utils.grouper(gen, n=self.config["write_batch_size"]):		for contents in utils.grouper(gen, n=self.config["write_batch_size"]):
res = self.persist_index_computations(		res = self.persist_index_computations(list(contents))
list(contents), policy_update="update-dups"
)
if not count_object_added_key:		if not count_object_added_key:
count_object_added_key = list(res.keys())[0]		count_object_added_key = list(res.keys())[0]
count += res[count_object_added_key]		count += res[count_object_added_key]
if count > 0:		if count > 0:
summary["status"] = "eventful"		summary["status"] = "eventful"
except Exception:		except Exception:
if not self.catch_exceptions:		if not self.catch_exceptions:
raise		raise
Show All 11 Lines	class OriginIndexer(BaseIndexer[str, None, TResult], Generic[TResult]):

Note: the :class:`OriginIndexer` is not an instantiable object.		Note: the :class:`OriginIndexer` is not an instantiable object.
To use it in another context one should inherit from this class		To use it in another context one should inherit from this class
and override the methods mentioned in the :class:`BaseIndexer`		and override the methods mentioned in the :class:`BaseIndexer`
class.		class.

"""		"""

def run(		def run(self, origin_urls: List[str], **kwargs) -> Dict:
self, origin_urls: List[str], policy_update: str = "update-dups", **kwargs
) -> Dict:
"""Given a list of origin urls:		"""Given a list of origin urls:

- retrieve origins from storage		- retrieve origins from storage
- execute the indexing computations		- execute the indexing computations
- store the results (according to policy_update)		- store the results

Args:		Args:
origin_urls: list of origin urls.		origin_urls: list of origin urls.
policy_update: either 'update-dups' or 'ignore-dups' to
respectively update duplicates (default) or ignore them
**kwargs: passed to the `index` method		**kwargs: passed to the `index` method

"""		"""
		if "policy_update" in kwargs:
		warnings.warn(
		"'policy_update' argument is deprecated and ignored.",
		DeprecationWarning,
		)
		del kwargs["policy_update"]
summary: Dict[str, Any] = {"status": "uneventful"}		summary: Dict[str, Any] = {"status": "uneventful"}
try:		try:
results = self.index_list(origin_urls, **kwargs)		results = self.index_list(origin_urls, **kwargs)
except Exception:		except Exception:
if not self.catch_exceptions:		if not self.catch_exceptions:
raise		raise
summary["status"] = "failed"		summary["status"] = "failed"
return summary		return summary

summary_persist = self.persist_index_computations(results, policy_update)		summary_persist = self.persist_index_computations(results)
self.results = results		self.results = results
if summary_persist:		if summary_persist:
for value in summary_persist.values():		for value in summary_persist.values():
if value > 0:		if value > 0:
summary["status"] = "eventful"		summary["status"] = "eventful"
summary.update(summary_persist)		summary.update(summary_persist)
return summary		return summary

Show All 14 Lines	class RevisionIndexer(BaseIndexer[Sha1Git, Revision, TResult], Generic[TResult]):

Note: the :class:`RevisionIndexer` is not an instantiable object.		Note: the :class:`RevisionIndexer` is not an instantiable object.
To use it in another context one should inherit from this class		To use it in another context one should inherit from this class
and override the methods mentioned in the :class:`BaseIndexer`		and override the methods mentioned in the :class:`BaseIndexer`
class.		class.

"""		"""

def run(self, ids: List[Sha1Git], policy_update: str) -> Dict:		def run(self, ids: List[Sha1Git], **kwargs) -> Dict:
"""Given a list of sha1_gits:		"""Given a list of sha1_gits:

- retrieve revisions from storage		- retrieve revisions from storage
- execute the indexing computations		- execute the indexing computations
- store the results (according to policy_update)		- store the results

Args:		Args:
ids: sha1_git's identifier list		ids: sha1_git's identifier list
policy_update: either 'update-dups' or 'ignore-dups' to
respectively update duplicates or ignore them

"""		"""
		if "policy_update" in kwargs:
		warnings.warn(
		"'policy_update' argument is deprecated and ignored.",
		DeprecationWarning,
		)
		del kwargs["policy_update"]
summary: Dict[str, Any] = {"status": "uneventful"}		summary: Dict[str, Any] = {"status": "uneventful"}
results = []		results = []

revision_ids = [		revision_ids = [
hashutil.hash_to_bytes(id_) if isinstance(id_, str) else id_ for id_ in ids		hashutil.hash_to_bytes(id_) if isinstance(id_, str) else id_ for id_ in ids
]		]
for (rev_id, rev) in zip(revision_ids, self.storage.revision_get(revision_ids)):		for (rev_id, rev) in zip(revision_ids, self.storage.revision_get(revision_ids)):
if not rev:		if not rev:
# TODO: call self.index() with rev=None?		# TODO: call self.index() with rev=None?
self.log.warning(		self.log.warning(
"Revision %s not found in storage", hashutil.hash_to_hex(rev_id)		"Revision %s not found in storage", hashutil.hash_to_hex(rev_id)
)		)
continue		continue
try:		try:
results.extend(self.index(rev_id, rev))		results.extend(self.index(rev_id, rev))
except Exception:		except Exception:
if not self.catch_exceptions:		if not self.catch_exceptions:
raise		raise
self.log.exception("Problem when processing revision")		self.log.exception("Problem when processing revision")
summary["status"] = "failed"		summary["status"] = "failed"
return summary		return summary

summary_persist = self.persist_index_computations(results, policy_update)		summary_persist = self.persist_index_computations(results)
if summary_persist:		if summary_persist:
for value in summary_persist.values():		for value in summary_persist.values():
if value > 0:		if value > 0:
summary["status"] = "eventful"		summary["status"] = "eventful"
summary.update(summary_persist)		summary.update(summary_persist)
self.results = results		self.results = results
return summary		return summary