[u 7/7] Configure mirroring per source (#7066)

nadove-ucsc · nadove-ucsc · commit 73b51e6aa945 · 2025-11-07T18:33:28.000-08:00
diff --git a/UPGRADING.rst b/UPGRADING.rst
@@ -20,6 +20,27 @@ reverted. This is all fairly informal and loosely defined. Hopefully we won't
 have too many entries in this file.
 
 
+#7066 Configure mirroring per catalog and source
+================================================
+
+In your personal deployments' ``environment.py`` file(s):
+
+1. Rename the loop variable ``internal`` to ``is_it``
+
+2. Update the value of the ``pop`` flag
+
+2. Replace ``mklist`` with ``mksrcs`` and update the call site
+
+3. Update the type annotations and definitions for ``bqsrc``, ``mksrc``,
+   ``mkdelta``, and ``mkdict``
+
+4. Insert the ``mirror_limit`` property in the definition of
+   ``AZUL_CATALOGS``
+
+As always, use the sandbox deployment's ``environment.py`` as a model when
+upgrading personal deployments.
+
+
 #7468 Update Swagger to v5.29.4
 ===============================
 
diff --git a/deployments/anvilbox/environment.py b/deployments/anvilbox/environment.py
@@ -9,6 +9,7 @@
 is_sandbox = True
 
 pop = 1  # remove snapshot
+no_mirror = 2  # do not mirror files from snapshot (redundant for managed access snapshots)
 
 type ProjectName = str
 type SourceSpec = str
@@ -33,7 +34,7 @@ def mksrc(source_type: Literal['bigquery', 'parquet'],
           flags: int = 0
           ) -> tuple[ProjectName, SourceItem | None]:
     project = '_'.join(snapshot.split('_')[1:-3])
-    assert flags <= pop
+    assert flags <= pop | no_mirror
     source = None if flags & pop else (
         ':'.join([
             'tdr',
@@ -42,7 +43,9 @@ def mksrc(source_type: Literal['bigquery', 'parquet'],
             google_project,
             snapshot,
         ]),
-        {}
+        {
+            'mirror': not (flags & no_mirror),
+        }
     )
     return project, source
 
diff --git a/deployments/anvildev/environment.py b/deployments/anvildev/environment.py
@@ -7,6 +7,7 @@
 )
 
 pop = 1  # remove snapshot
+no_mirror = 2  # do not mirror files from snapshot (redundant for managed access snapshots)
 
 type ProjectName = str
 type SourceSpec = str
@@ -31,7 +32,7 @@ def mksrc(source_type: Literal['bigquery', 'parquet'],
           flags: int = 0
           ) -> tuple[ProjectName, SourceItem | None]:
     project = '_'.join(snapshot.split('_')[1:-3])
-    assert flags <= pop
+    assert flags <= pop | no_mirror
     source = None if flags & pop else (
         ':'.join([
             'tdr',
@@ -40,7 +41,9 @@ def mksrc(source_type: Literal['bigquery', 'parquet'],
             google_project,
             snapshot,
         ]),
-        {}
+        {
+            'mirror': not (flags & no_mirror),
+        }
     )
     return project, source
 
diff --git a/deployments/anvilprod/environment.py b/deployments/anvilprod/environment.py
@@ -9,6 +9,7 @@
 )
 
 pop = 1  # remove snapshot
+no_mirror = 2  # do not mirror files from snapshot (redundant for managed access snapshots)
 
 type ProjectName = str
 type SourceSpec = str
@@ -34,7 +35,7 @@ def mksrc(source_type: Literal['bigquery', 'parquet'],
           flags: int = 0
           ) -> tuple[ProjectName, SourceItem | None]:
     project = '_'.join(snapshot.split('_')[1:-3])
-    assert flags <= pop
+    assert flags <= pop | no_mirror
     source = None if flags & pop else (
         ':'.join([
             'tdr',
@@ -43,7 +44,9 @@ def mksrc(source_type: Literal['bigquery', 'parquet'],
             google_project,
             snapshot,
         ]),
-        {}
+        {
+            'mirror': not (flags & no_mirror),
+        }
     )
     return project, source
 
diff --git a/deployments/dev/environment.py b/deployments/dev/environment.py
@@ -7,6 +7,7 @@
 )
 
 pop = 1  # remove snapshot
+no_mirror = 2  # do not mirror files from snapshot (redundant for managed access snapshots)
 
 type ProjectName = str
 type SourceSpec = str
@@ -20,7 +21,7 @@ def mksrc(source_type: Literal['bigquery', 'parquet'],
           flags: int = 0
           ) -> tuple[ProjectName, SourceItem | None]:
     _, env, project, _ = snapshot.split('_', 3)
-    assert flags <= pop
+    assert flags <= pop | no_mirror
     source = None if flags & pop else (
         ':'.join([
             'tdr',
@@ -29,7 +30,9 @@ def mksrc(source_type: Literal['bigquery', 'parquet'],
             google_project,
             snapshot,
         ]),
-        {}
+        {
+            'mirror': not (flags & no_mirror),
+        }
     )
     return project, source
 
diff --git a/deployments/hammerbox/environment.py b/deployments/hammerbox/environment.py
@@ -11,6 +11,7 @@
 is_sandbox = True
 
 pop = 1  # remove snapshot
+no_mirror = 2  # do not mirror files from snapshot (redundant for managed access snapshots)
 
 type ProjectName = str
 type SourceSpec = str
@@ -36,7 +37,7 @@ def mksrc(source_type: Literal['bigquery', 'parquet'],
           flags: int = 0
           ) -> tuple[ProjectName, SourceItem | None]:
     project = '_'.join(snapshot.split('_')[1:-3])
-    assert flags <= pop
+    assert flags <= pop | no_mirror
     source = None if flags & pop else (
         ':'.join([
             'tdr',
@@ -45,7 +46,9 @@ def mksrc(source_type: Literal['bigquery', 'parquet'],
             google_project,
             snapshot,
         ]),
-        {}
+        {
+            'mirror': not (flags & no_mirror),
+        }
     )
     return project, source
 
diff --git a/deployments/prod/environment.py b/deployments/prod/environment.py
@@ -9,6 +9,7 @@
 )
 
 pop = 1  # remove snapshot
+no_mirror = 2  # do not mirror files from snapshot (redundant for managed access snapshots)
 
 type ProjectName = str
 type SourceSpec = str
@@ -22,7 +23,7 @@ def mksrc(source_type: Literal['bigquery', 'parquet'],
           flags: int = 0
           ) -> tuple[ProjectName, SourceItem | None]:
     _, env, project, _ = snapshot.split('_', 3)
-    assert flags <= pop
+    assert flags <= pop | no_mirror
     source = None if flags & pop else (
         ':'.join([
             'tdr',
@@ -31,7 +32,9 @@ def mksrc(source_type: Literal['bigquery', 'parquet'],
             google_project,
             snapshot,
         ]),
-        {}
+        {
+            'mirror': not (flags & no_mirror),
+        }
     )
     return project, source
 
diff --git a/deployments/sandbox/environment.py b/deployments/sandbox/environment.py
@@ -9,6 +9,7 @@
 is_sandbox = True
 
 pop = 1  # remove snapshot
+no_mirror = 2  # do not mirror files from snapshot (redundant for managed access snapshots)
 
 type ProjectName = str
 type SourceSpec = str
@@ -22,7 +23,7 @@ def mksrc(source_type: Literal['bigquery', 'parquet'],
           flags: int = 0
           ) -> tuple[ProjectName, SourceItem | None]:
     _, env, project, _ = snapshot.split('_', 3)
-    assert flags <= pop
+    assert flags <= pop | no_mirror
     source = None if flags & pop else (
         ':'.join([
             'tdr',
@@ -31,7 +32,9 @@ def mksrc(source_type: Literal['bigquery', 'parquet'],
             google_project,
             snapshot,
         ]),
-        {}
+        {
+            'mirror': not (flags & no_mirror),
+        }
     )
     return project, source
 
diff --git a/deployments/tempdev/environment.py b/deployments/tempdev/environment.py
@@ -7,6 +7,7 @@
 )
 
 pop = 1  # remove snapshot
+no_mirror = 2  # do not mirror files from snapshot (redundant for managed access snapshots)
 
 type ProjectName = str
 type SourceSpec = str
@@ -31,7 +32,7 @@ def mksrc(source_type: Literal['bigquery', 'parquet'],
           flags: int = 0,
           ) -> tuple[ProjectName, SourceItem | None]:
     project = '_'.join(snapshot.split('_')[1:-3])
-    assert flags <= pop
+    assert flags <= pop | no_mirror
     source = None if flags & pop else (
         ':'.join([
             'tdr',
@@ -40,7 +41,9 @@ def mksrc(source_type: Literal['bigquery', 'parquet'],
             google_project,
             snapshot,
         ]),
-        {}
+        {
+            'mirror': not (flags & no_mirror),
+        }
     )
     return project, source
 
diff --git a/src/azul/azulclient.py b/src/azul/azulclient.py
@@ -401,9 +401,14 @@ def remote_mirror(self,
 
             def messages():
                 for source, cfg in sources:
-                    log.info('Mirroring files in source %r from catalog %r',
-                             str(source.spec), catalog)
-                    yield self.mirror_source_message(catalog, source)
+                    if cfg.mirror:
+                        log.info('Mirroring files in source %r from catalog %r',
+                                 str(source.spec), catalog)
+                        yield self.mirror_source_message(catalog, source)
+                    else:
+                        log.info('Not mirroring any files in source %r from catalog %r because '
+                                 'mirroring is explicitly disabled',
+                                 str(source.spec), catalog)
 
             self.queue_mirror_messages(messages())
 
diff --git a/src/azul/indexer/__init__.py b/src/azul/indexer/__init__.py
@@ -410,9 +410,9 @@ def __contains__(self, partition_prefix: str) -> bool:
 @attrs.frozen(kw_only=True)
 class SourceConfig(SerializableAttrs):
     """
-    Configuration
+    Configuration on how to index or mirror a specific source.
     """
-    pass
+    mirror: bool
 
 
 @attrs.frozen(kw_only=True, order=True)
diff --git a/test/azul_test_case.py b/test/azul_test_case.py
@@ -456,7 +456,7 @@ def catalog_config(cls) -> dict[CatalogName, Config.Catalog]:
                                         mirror_limit=None,
                                         plugins=dict(metadata=config.Catalog.Plugin(name='hca'),
                                                      repository=config.Catalog.Plugin(name='dss')),
-                                        sources={str(cls.source.spec): {}})
+                                        sources={str(cls.source.spec): {'mirror': True}})
         }
 
 
@@ -480,7 +480,7 @@ def _patch_tdr_service_url(cls):
 
     @classmethod
     def _sources(cls):
-        return {str(cls.source.spec): {}}
+        return {str(cls.source.spec): {'mirror': True}}
 
     @classmethod
     def _patch_source_cache(cls):
@@ -522,5 +522,5 @@ def catalog_config(cls) -> dict[CatalogName, Config.Catalog]:
                                         mirror_limit=None,
                                         plugins=dict(metadata=config.Catalog.Plugin(name='anvil'),
                                                      repository=config.Catalog.Plugin(name='tdr_anvil')),
-                                        sources={str(cls.source.spec): {}})
+                                        sources={str(cls.source.spec): {'mirror': True}})
         }
diff --git a/test/indexer/test_mirror_controller.py b/test/indexer/test_mirror_controller.py
@@ -130,8 +130,9 @@ def mirror_controller(self) -> MirrorController:
     def _mirror_event(self, body: JSON) -> list[SQSRecord]:
         return [self._mock_sqs_record(body, fifo=True)]
 
-    def _remote_mirror(self) -> MutableJSONs:
-        self.client.remote_mirror(self.catalog, [(self.source, SourceConfig())])
+    def _remote_mirror(self, mirror_source_cfg: bool = True) -> MutableJSONs:
+        cfg = SourceConfig(mirror=mirror_source_cfg)
+        self.client.remote_mirror(self.catalog, [(self.source, cfg)])
         return self._read_queue(self.client.mirror_queue())
 
     def _test_remote_mirror(self):
@@ -208,6 +209,11 @@ def test_info_schema(self):
 
     def test_files_not_mirrored(self):
         self._create_mock_queues(config.mirror_queue_names)
+
+        with self.subTest(no_mirror=True):
+            messages = self._remote_mirror(mirror_source_cfg=False)
+            self.assertEqual([], messages)
+
         catalog = config.catalogs[self.catalog]
 
         def patch_max_file_size(size):
diff --git a/test/integration_test.py b/test/integration_test.py
@@ -297,7 +297,8 @@ def managed_access_sources_by_catalog(self
     def _select_source(self,
                        catalog: CatalogName,
                        *,
-                       public: bool | None = None
+                       public: bool | None = None,
+                       mirror: bool = False,
                        ) -> tuple[SourceRef, SourceConfig] | None:
         """
         Choose an indexed source at random.
@@ -310,6 +311,11 @@ def _select_source(self,
                        public sources. If false, choose a non-public source, or
                        return `None` if the catalog contains no non-public
                        sources.
+
+        :param mirror: If true, choose a source where the `no_mirror` flag is
+                       not present, or return `None` if the catalog contains no
+                       such source. If false, choose a source regardless of
+                       whether this flag is present.
         """
         plugin = self.repository_plugin(catalog)
         sources = plugin.sources
@@ -335,6 +341,8 @@ def _filter(source: tuple[SourceSpec, SourceConfig]) -> bool:
                 valid = source[0] in ma_sources
             else:
                 assert False, public
+            if mirror:
+                valid &= source[1].mirror
             return valid
 
         sources = dict(filter(_filter, sources.items()))
@@ -1722,7 +1730,7 @@ def _test_mirroring(self, *, delete: bool):
                 if c.is_integration_test_catalog and c.mirror_limit >= 0
             ]
             sources_by_catalog = {
-                catalog: [self._select_source(catalog, public=True)]
+                catalog: [self._select_source(catalog, public=True, mirror=True)]
                 for catalog in catalogs
             }
 
@@ -1897,7 +1905,9 @@ def test_can_bundle_canned_repository(self):
                                           'repository': config.Catalog.Plugin(name='canned'),
                                       },
                                       sources={
-                                          'https://github.com/HumanCellAtlas/schema-test-data/tree/master/tests': {}
+                                          'https://github.com/HumanCellAtlas/schema-test-data/tree/master/tests': {
+                                              'mirror': False
+                                          },
                                       })
         with mock.patch.object(Config,
                                'catalogs',
diff --git a/test/service/test_repository_files.py b/test/service/test_repository_files.py
@@ -195,7 +195,7 @@ def test_repository_files(self, mock_get_cached_sources):
 
     @classmethod
     def _sources(cls):
-        return {cls.make_mock_source_spec(n): {} for n in cls.mock_source_names}
+        return {cls.make_mock_source_spec(n): {'mirror': True} for n in cls.mock_source_names}
 
     @mock.patch.object(TDRClient, 'snapshot_names_by_id')
     @mock.patch.object(TDRClient, 'validate', new=MagicMock())
