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

Author: nadove-ucsc

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
@@ -42,8 +43,11 @@ def mksrc(source_type: Literal['bigquery', 'parquet'],
             google_project,
             snapshot,
         ]),
-        {}
+        {
+            'mirror': not (flags & no_mirror),
+        }
     )
+
     return project, source
 
 

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
@@ -40,8 +41,11 @@ def mksrc(source_type: Literal['bigquery', 'parquet'],
             google_project,
             snapshot,
         ]),
-        {}
+        {
+            'mirror': not (flags & no_mirror),
+        }
     )
+
     return project, source
 
 

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
@@ -43,8 +44,11 @@ def mksrc(source_type: Literal['bigquery', 'parquet'],
             google_project,
             snapshot,
         ]),
-        {}
+        {
+            'mirror': not (flags & no_mirror),
+        }
     )
+
     return project, source
 
 

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
@@ -29,8 +30,11 @@ def mksrc(source_type: Literal['bigquery', 'parquet'],
             google_project,
             snapshot,
         ]),
-        {}
+        {
+            'mirror': not (flags & no_mirror),
+        }
     )
+
     return project, source
 
 

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
@@ -45,8 +46,11 @@ def mksrc(source_type: Literal['bigquery', 'parquet'],
             google_project,
             snapshot,
         ]),
-        {}
+        {
+            'mirror': not (flags & no_mirror),
+        }
     )
+
     return project, source
 
 

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
@@ -31,8 +32,11 @@ def mksrc(source_type: Literal['bigquery', 'parquet'],
             google_project,
             snapshot,
         ]),
-        {}
+        {
+            'mirror': not (flags & no_mirror),
+        }
     )
+
     return project, source
 
 

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
@@ -31,8 +32,11 @@ def mksrc(source_type: Literal['bigquery', 'parquet'],
             google_project,
             snapshot,
         ]),
-        {}
+        {
+            'mirror': not (flags & no_mirror),
+        }
     )
+
     return project, source
 
 

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
@@ -29,8 +30,11 @@ def mksrc(source_type: Literal['bigquery', 'parquet'],
             google_project,
             snapshot,
         ]),
-        {}
+        {
+            'mirror': not (flags & no_mirror),
+        }
     )
+
     return project, source
 
 

src/azul/azulclient.py

@@ -1,3 +1,6 @@
+import fnmatch
+import logging
+import uuid
 from collections import (
     defaultdict,
 )
@@ -11,18 +14,15 @@
 from enum import (
     auto,
 )
-import fnmatch
 from functools import (
     partial,
 )
-import logging
+from itertools import starmap
 from pprint import (
     PrettyPrinter,
 )
 from typing import (
-    AbstractSet,
-)
-import uuid
+    AbstractSet, )
 
 import attrs
 import requests
@@ -389,15 +389,22 @@ def is_queue_empty(self, queue_name: str) -> bool:
         length, _ = self.queues.get_queue_lengths(queues)
         return length == 0
 
-    def remote_mirror(self, catalog: CatalogName, sources: Iterable[SourceRef]):
+    def remote_mirror(self,
+                      catalog: CatalogName,
+                      sources: Iterable[tuple[SourceRef, SourceConfig]]):
 
-        def message(source: SourceRef):
-            log.info('Mirroring files in source %r from catalog %r',
-                     str(source.spec), catalog)
-            return self.mirror_source_message(catalog, source)
+        def messages():
+            for source, cfg in sources:
+                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)
 
-        messages = map(message, sources)
-        self.queue_mirror_messages(messages)
+        self.queue_mirror_messages(messages())
 
     def _get_non_empty_fail_queues(self) -> set[str]:
         return {

src/azul/indexer/__init__.py

@@ -412,7 +412,7 @@ class SourceConfig(SerializableAttrs):
     """
     Configuration
     """
-    pass
+    mirror: bool
 
 
 @attrs.frozen(kw_only=True, order=True)