[A] Workaround: Manifest content hash computation times out (#6123)

Author: dsotirho-ucsc

environment.py

@@ -943,5 +943,9 @@ def env() -> Mapping[str, Optional[str]]:
         # $1 per one million requests above ten million requests. The blocking
         # only applies to URLs disallowed via robots.txt.
         #
-        'azul_waf_bot_control': '0'
+        'azul_waf_bot_control': '0',
+
+        # Whether to enable bundle notifications for incremental index changes
+        #
+        'AZUL_ENABLE_BUNDLE_NOTIFICATIONS': '0'
     }

scripts/generate_openapi_document.py

@@ -52,7 +52,8 @@ def main():
         patch_config(f'{lambda_name}_function_name', f'azul-{lambda_name}-dev'),
         patch_config('enable_log_forwarding', False),
         patch_config('enable_replicas', True),
-        patch_config('monitoring_email', '[email protected]')
+        patch_config('monitoring_email', '[email protected]'),
+        patch_config('enable_bundle_notifications', True)
     ):
         lambda_endpoint = furl('http://localhost')
         with patch.object(target=AzulChaliceApp,

scripts/reindex.py

@@ -45,15 +45,22 @@
 parser.add_argument('--local',
                     default=False,
                     action='store_true',
-                    help='Do not offload the listing of subgraphs to the indexer Lambda function. When this option is '
-                         'used, this script queries the repository without partitioning, and the indexer notification '
-                         'endpoint is invoked for each subgraph individually and concurrently using worker threads. '
-                         'This is magnitudes slower than remote (i.e. partitioned) indexing. If this option is not '
-                         'used (the default), the set of subgraphs matching the query is partitioned using the '
-                         'partition prefix length configured for each of the catalog sources being reindexed. Each '
-                         'query partition is processed independently and remotely by the indexer lambda. The index '
-                         'Lambda function queries the repository for each partition and queues a notification for each '
-                         'matching subgraph in the partition.')
+                    help=(
+                             '' if config.enable_bundle_notifications else '**DISABLED** '
+                         ) + (
+                             'Do not offload the listing of subgraphs to the indexer Lambda function. When '
+                             'this option is used, this script queries the repository without partitioning, '
+                             'and the indexer notification endpoint is invoked for each subgraph '
+                             'individually and concurrently using worker threads. This is magnitudes slower '
+                             'than remote (i.e. partitioned) indexing. If this option is not used (the '
+                             'default), the set of subgraphs matching the query is partitioned using the '
+                             'partition prefix length configured for each of the catalog sources being '
+                             'reindexed. Each query partition is processed independently and remotely by '
+                             'the indexer lambda. The index Lambda function queries the repository for each '
+                             'partition and queues a notification for each matching subgraph in the '
+                             'partition.'
+                         )
+                    )
 parser.add_argument('--catalogs',
                     nargs='+',
                     metavar='NAME',
@@ -133,6 +140,10 @@ def main(argv: list[str]):
     deindex = args.deindex or (args.delete and not every_source)
     delete = args.delete and every_source
 
+    if args.local and not config.enable_bundle_notifications:
+        parser.error('Local reindexing is not available while bundle '
+                     'notifications are disabled.')
+
     if every_source:
         if deindex:
             parser.error('--deindex is incompatible with source `*`. Use --delete instead.')

src/azul/__init__.py

@@ -1858,6 +1858,10 @@ def enable_mirroring(self) -> bool:
     def mirror_bucket(self) -> str | None:
         return self.environ.get('AZUL_MIRROR_BUCKET')
 
+    @property
+    def enable_bundle_notifications(self):
+        return self._boolean(self.environ['AZUL_ENABLE_BUNDLE_NOTIFICATIONS'])
+
 
 config: Config = Config()  # yes, the type hint does help PyCharm
 

src/azul/azulclient.py

@@ -228,7 +228,7 @@ def handle_future(future: Future) -> None:
             log.error('Unsent notifications and their HTTP status code:\n%s',
                       printer.pformat(missing))
         if errors or missing:
-            raise AzulClientNotificationError
+            raise AzulClientNotificationError(set(errors.keys()))
 
     def matching_sources(self,
                          catalogs: Iterable[CatalogName],
@@ -431,5 +431,5 @@ class AzulClientError(RuntimeError):
 
 class AzulClientNotificationError(AzulClientError):
 
-    def __init__(self) -> None:
-        super().__init__('Some notifications could not be sent')
+    def __init__(self, *args, **kwargs) -> None:
+        super().__init__('Some notifications could not be sent', *args, **kwargs)

src/azul/indexer/index_controller.py

@@ -74,6 +74,7 @@ def handlers(self) -> dict[str, Any]:
         @self.app.route(
             '/{catalog}/{action}',
             methods=['POST'],
+            enabled=config.enable_bundle_notifications,
             spec={
                 'tags': ['Indexing'],
                 'summary': 'Notify the indexer to perform an action on a bundle',
@@ -213,6 +214,8 @@ def handle_notification(self, catalog: CatalogName, action: str):
                     raise R.propagate(e, chalice.BadRequestError)
             notification = request.json_body
             log.info('Received notification %r for catalog %r', notification, catalog)
+            assert config.enable_bundle_notifications, R(
+                'Bundle notifications are disabled')
             self._validate_notification(notification)
             service = self.index_queue_service
             message = service.index_bundle_message(self._load_action(action),

src/azul/service/manifest_service.py

@@ -72,6 +72,7 @@
 from azul import (
     CatalogName,
     R,
+    cache,
     cached_property,
     config,
     mutable_furl,
@@ -982,7 +983,9 @@ def manifest_key(self) -> ManifestKey:
         """
         git_commit = config.lambda_git_status['commit']
         filter_string = repr(sort_frozen(freeze(self.filters.explicit)))
-        content_hash = str(self.manifest_content_hash)
+        # If incremental index changes are disabled, we don't need to worry
+        # about individual bundles, only sources.
+        content_hash = str(self.manifest_hash(config.enable_bundle_notifications))
         catalog = self.catalog
         format = self.format()
         manifest_hash_input = [
@@ -1032,9 +1035,11 @@ def file_name(self,
             file_name = atlas + '-manifest-' + self.s3_object_key_base(manifest_key)
         return file_name
 
-    def _create_request(self) -> Search:
+    def _create_request(self, entity_type: str | None = None) -> Search:
+        if entity_type is None:
+            entity_type = self.entity_type
         pipeline = self._create_pipeline()
-        request = self.service.create_request(self.catalog, self.entity_type)
+        request = self.service.create_request(self.catalog, entity_type)
         request = pipeline.prepare_request(request)
         # The response is processed by the generator, not the pipeline
         return request
@@ -1144,22 +1149,58 @@ def _azul_file_url(self,
                                           fetch=False,
                                           **args))
 
-    @cached_property
-    def manifest_content_hash(self) -> int:
-        log.debug('Computing content hash for manifest using filters %r ...', self.filters)
+    @cache
+    def manifest_hash(self, by_bundle: bool) -> int:
+        """
+        Return a content hash for the manifest.
+
+        If `by_bundle` is True, the hash is computed from the fully-qualified
+        identifiers of all bundles containing files that match the current
+        filter. The return value approximates a hash of the content of the
+        manifest because a change of the file data requires a change to the file
+        metadata which requires a new bundle or bundle version.
+
+        If `by_bundle` is False, the hash is computed from the identifiers of
+        the sources from which projects/datasets containing files matching the
+        current filters were indexed. It's worth noting that a filter may match
+        a project/dataset but none of the project's files. For example, if a
+        project contains only files derived from either mouse brains or lion
+        hearts, the project will match the filter `species=lion and
+        organ=brain`, but none of its files will. If such a project/dataset is
+        added/removed to/from the index, the manifest hash returned for a given
+        filter will be different even though the contents of the manifest hasn't
+        changed, as no matching files were added or removed.
+
+        So while the hash computed from the sources is less sensitive than the
+        one computed from the bundles, it can be computed much more quickly.
+        """
+        log.debug('Computing content hash for manifest from %s using %r ...',
+                  'bundles' if by_bundle else 'sources', self.filters)
         start_time = time.time()
-        request = self._create_request()
+        if by_bundle:
+            request = self._create_request()
+        else:
+            root_entity_type = self.metadata_plugin.root_entity_type
+            request = self._create_request(entity_type=root_entity_type)
         request.aggs.metric(
             'hash',
             'scripted_metric',
             init_script='''
                 state.fields = 0
             ''',
-            map_script='''
-                for (bundle in params._source.bundles) {
-                    state.fields += (bundle.uuid + bundle.version).hashCode()
-                }
-            ''',
+            map_script=(
+                '''
+                    for (bundle in params._source.bundles) {
+                        state.fields += (bundle.uuid + bundle.version).hashCode()
+                    }
+                '''
+                if by_bundle else
+                '''
+                    for (source in params._source.sources) {
+                        state.fields += source.id.hashCode()
+                    }
+                '''
+            ),
             combine_script='''
                 return state.fields.hashCode()
             ''',

test/azul_test_case.py

@@ -242,6 +242,7 @@ def setUpClass(cls) -> None:
         cls._patch_dss_query_prefix()
         cls._patch_lambda_env()
         cls._patch_valid_schema_domains()
+        cls._patch_enable_bundle_notifications()
 
     def setUp(self) -> None:
         super().setUp()
@@ -344,6 +345,13 @@ def _patch_valid_schema_domains(cls):
                                        attribute='valid_schema_domains',
                                        new=cls.valid_schema_domains))
 
+    @classmethod
+    def _patch_enable_bundle_notifications(cls):
+        cls.addClassPatch(patch.object(target=type(config),
+                                       attribute='enable_bundle_notifications',
+                                       new_callable=PropertyMock,
+                                       return_value=True))
+
 
 class CatalogTestCase(AzulUnitTestCase, metaclass=ABCMeta):
     catalog: CatalogName = 'test'

test/integration_test.py

@@ -1746,10 +1746,12 @@ class AzulClientIntegrationTest(IntegrationTestCase):
     def test_azul_client_error_handling(self):
         invalid_notification = {}
         notifications = [invalid_notification]
-        self.assertRaises(AzulClientNotificationError,
-                          self.azul_client.index,
-                          first(config.integration_test_catalogs),
-                          notifications)
+        with self.assertRaises(AzulClientNotificationError) as cm:
+            self.azul_client.index(catalog=first(config.integration_test_catalogs),
+                                   notifications=notifications)
+        self.assertEqual('Some notifications could not be sent', cm.exception.args[0])
+        expected = 400 if config.enable_bundle_notifications else 403
+        self.assertEqual({expected}, cm.exception.args[1])
 
 
 class OpenAPIIntegrationTest(AzulTestCase):

test/service/test_manifest.py

@@ -902,37 +902,57 @@ def test_manifest_filter_validation(self):
         response = requests.put(str(url))
         self.assertEqual(400, response.status_code, response.content)
 
-    def test_manifest_content_disposition_header(self):
+    @patch.object(type(config),
+                  'enable_bundle_notifications',
+                  new=PropertyMock(return_value=True))
+    def test_content_disposition_header_with_notifications_enabled(self) -> None:
+        self._test_content_disposition_header()
+
+    @patch.object(type(config),
+                  'enable_bundle_notifications',
+                  new=PropertyMock(return_value=False))
+    def test_content_disposition_header_with_notifications_disabled(self) -> None:
+        self._test_content_disposition_header()
+
+    def _test_content_disposition_header(self):
         bundle_fqid = self.bundle_fqid(uuid='f79257a7-dfc6-46d6-ae00-ba4b25313c10',
                                        version='2018-09-14T13:33:14.453337Z')
         self._index_canned_bundle(bundle_fqid)
         with patch.object(manifest_service, 'datetime') as mock_datetime:
             mock_datetime.now.return_value = datetime(1985, 10, 25, 1, 21)
             for format in [ManifestFormat.compact]:
-                for filters, expected_name in [
+                source_hash = '4bc67e84-4873-591f-b524-a5fe4ec215eb'
+                for filters, name_with_notif_enabled, name_with_notif_disabled in [
                     # For a single project, the content disposition file name should
                     # be the project name followed by the date and time
                     (
                         {'project': {'is': ['Single of human pancreas']}},
+                        'Single of human pancreas 1985-10-25 01.21',
                         'Single of human pancreas 1985-10-25 01.21'
                     ),
                     # In all other cases, the standard content disposition file name
                     # should be "hca-manifest-" followed by the manifest key,
                     # a pair of deterministically derived v5 UUIDs.
                     (
                         {'project': {'is': ['Single of human pancreas', 'Mouse Melanoma']}},
-                        'hca-manifest-20d97863-d8cf-54f3-8575-0f9593d3d7ef.4bc67e84-4873-591f-b524-a5fe4ec215eb'
+                        'hca-manifest-20d97863-d8cf-54f3-8575-0f9593d3d7ef.' + source_hash,
+                        'hca-manifest-9f00706c-aa4f-5ae3-80ee-5328c80b6fce.' + source_hash
                     ),
                     (
                         {},
-                        'hca-manifest-c3cf398e-1927-5aae-ba2a-81d8d1800b2d.4bc67e84-4873-591f-b524-a5fe4ec215eb'
+                        'hca-manifest-c3cf398e-1927-5aae-ba2a-81d8d1800b2d.' + source_hash,
+                        'hca-manifest-f2adc097-6300-5d33-9fa4-34d97d9bb39f.' + source_hash
                     )
                 ]:
                     with self.subTest(filters=filters, format=format):
                         manifest, num_partitions = self._get_manifest_object(format, filters)
                         self.assertFalse(manifest.was_cached)
                         self.assertEqual(1, num_partitions)
                         url = furl(self._service.get_manifest_url(manifest))
+                        if config.enable_bundle_notifications:
+                            expected_name = name_with_notif_enabled
+                        else:
+                            expected_name = name_with_notif_disabled
                         expected_cd = f'attachment;filename="{expected_name}.tsv"'
                         actual_cd = url.args['response-content-disposition']
                         self.assertEqual(expected_cd, actual_cd)
@@ -1057,9 +1077,18 @@ def test_compact_metadata_cache(self, _time_until_object_expires: MagicMock):
                 self.assertEqual([2, 2], list(map(len, file_names.values())))
                 self.assertEqual([1, 1], list(map(len, map(set, file_names.values()))))
 
+    @patch.object(type(config),
+                  'enable_bundle_notifications',
+                  new=PropertyMock(return_value=True))
     def test_hash_validity_with_notifications_enabled(self) -> None:
         self._test_hash_validity()
 
+    @patch.object(type(config),
+                  'enable_bundle_notifications',
+                  new=PropertyMock(return_value=False))
+    def test_hash_validity_with_notifications_disabled(self) -> None:
+        self._test_hash_validity()
+
     def _test_hash_validity(self):
         self.maxDiff = None
         bundles_by_project = {