fix: TDE-392 stop processing when an image transform fails (#1013)

* fix: stop processing if image transform fails and log error

* fix: metadata filtering logging

* fix: stop processing if image transform fails and log error

* fix: metadata filtering logging

* fix: remove testing variable

* fix: fix exception and logging messages

Author: amfage

topo_processor/cog/execution.py

@@ -18,9 +18,9 @@ def run(cmd: "Command") -> Tuple[int, str, str]:
 
         proc = subprocess.run(cmd.to_full_command(), stdout=subprocess.PIPE, stderr=subprocess.PIPE)
         if proc.returncode != 0:
-            get_log().trace("Ran command failed", command=cmd.redacted_command(), duration=time_in_ms() - start_time)
+            get_log().error("Run command failed", command=cmd.redacted_command(), duration=time_in_ms() - start_time)
             raise Exception(proc.stderr.decode())
-        get_log().trace("Ran command succeeded", command=cmd.redacted_command(), duration=time_in_ms() - start_time)
+        get_log().trace("Run command succeeded", command=cmd.redacted_command(), duration=time_in_ms() - start_time)
         return proc.returncode, proc.stdout.decode(), proc.stderr.decode()
 
 

topo_processor/data/data_transformers/data_transformer_imagery_historic.py

@@ -38,7 +38,12 @@ def transform_data(self, item: Item) -> None:
                 return
             output_path = os.path.join(item.collection.get_temp_dir(), f"{ulid.ULID()}.tiff")
 
-            create_cog(asset.source_path, output_path).run()
+            try:
+                create_cog(asset.source_path, output_path).run()
+            except Exception as e:
+                raise Exception(
+                    f"COG creation failed for item {item.id} with source path {asset.source_path} and output path {output_path}."
+                ) from e
 
             get_log().debug("Created COG", output_path=output_path, duration=time_in_ms() - start_time)
 

topo_processor/data/data_transformers/data_transformer_repo.py

@@ -26,8 +26,8 @@ def transform_data(self, item: Item) -> None:
                     transformer.transform_data(item)
                 except Exception as e:
                     item.add_error(str(e), transformer.name, e)
-                    get_log().warning(f"Data Transform Failed: {e}", transformers=transformer.name)
-                    return
+                    get_log().error("Data Transform Failed. Process is stopped.", transformers=transformer.name, error=e)
+                    raise Exception(e)
                 get_log().debug(
                     "Data Transformed",
                     duration=time_in_ms() - start_time,

topo_processor/metadata/lds_cache/lds_cache.py

@@ -72,7 +72,7 @@ def get_metadata(
 
 
 def filter_metadata(metadata_to_filter: Dict[str, Any], criteria: Dict[str, Any]) -> Dict[str, Any]:
-    get_log().debug("filter_metadata_start", criteria=criteria)
+    get_log().debug("filter_metadata", criteria=criteria)
     filtered_dict: Dict[str, Any] = {}
     is_found = False
 
@@ -86,5 +86,4 @@ def filter_metadata(metadata_to_filter: Dict[str, Any], criteria: Dict[str, Any]
                     break
         if is_found:
             filtered_dict[metadata_key] = metadata_value
-    get_log().debug("filter_metadata_end", metadata=filtered_dict)
     return filtered_dict

topo_processor/stac/item_factory.py

@@ -34,7 +34,7 @@ def _create_assets(source: str, data_type: str, metadata_path: str) -> None:
     for asset in assets:
         metadata_loader_rep.load_metadata(asset)
         if not asset.item:
-            raise Exception(f"No item set for asset. Process is stopped.")
+            raise Exception(f"No item set for asset {asset.source_path}. Process is stopped.")
 
 
 def _create_items(force: bool = False) -> None: