Add more logging

Author: neomatamune

cosmotech/coal/azure/adx/auth.py

@@ -98,7 +98,7 @@ def initialize_clients(adx_uri: str, adx_ingest_uri: str) -> Tuple[KustoClient,
     Returns:
         tuple: (kusto_client, ingest_client)
     """
-    LOGGER.debug("Initializing clients")
+    LOGGER.debug(T("coal.logs.adx.auth.initializing_clients"))
     kusto_client = create_kusto_client(adx_uri)
     ingest_client = create_ingest_client(adx_ingest_uri)
     return kusto_client, ingest_client

cosmotech/coal/azure/adx/ingestion.py

@@ -237,7 +237,7 @@ def monitor_ingestion(
     has_failures = False
     source_ids_copy = source_ids.copy()
 
-    LOGGER.info("Waiting for ingestion of data to finish")
+    LOGGER.info(T("coal.logs.adx.waiting_ingestion"))
 
     with tqdm.tqdm(desc="Ingestion status", total=len(source_ids_copy)) as pbar:
         while any(
@@ -252,7 +252,9 @@ def monitor_ingestion(
             for ingestion_id, ingestion_status in results:
                 if ingestion_status == IngestionStatus.FAILURE:
                     LOGGER.error(
-                        f"Ingestion {ingestion_id} failed for table {table_ingestion_id_mapping.get(ingestion_id)}"
+                        T("coal.logs.adx.ingestion_failed").format(
+                            ingestion_id=ingestion_id, table=table_ingestion_id_mapping.get(ingestion_id)
+                        )
                     )
                     has_failures = True
 
@@ -271,12 +273,14 @@ def monitor_ingestion(
             for ingestion_id, ingestion_status in results:
                 if ingestion_status == IngestionStatus.FAILURE:
                     LOGGER.error(
-                        f"Ingestion {ingestion_id} failed for table {table_ingestion_id_mapping.get(ingestion_id)}"
+                        T("coal.logs.adx.ingestion_failed").format(
+                            ingestion_id=ingestion_id, table=table_ingestion_id_mapping.get(ingestion_id)
+                        )
                     )
                     has_failures = True
         pbar.update(len(source_ids_copy))
 
-    LOGGER.info("All data ingestion attempts completed")
+    LOGGER.info(T("coal.logs.adx.ingestion_completed"))
     return has_failures
 
 
@@ -294,7 +298,7 @@ def handle_failures(kusto_client: KustoClient, database: str, operation_tag: str
         bool: True if the process should abort, False otherwise
     """
     if has_failures:
-        LOGGER.warning(f"Failures detected during ingestion - dropping data with tag: {operation_tag}")
+        LOGGER.warning(T("coal.logs.adx.failures_detected").format(operation_tag=operation_tag))
         _drop_by_tag(kusto_client, database, operation_tag)
         return True
     return False

cosmotech/coal/azure/adx/runner.py

@@ -57,7 +57,7 @@ def prepare_csv_content(folder_path: str) -> Dict[str, Dict[str, Any]]:
         cols = {k.strip(): "string" for k in headers}
         csv_datas = {"filename": _file.name.removesuffix(".csv"), "headers": cols}
         content[str(_file)] = csv_datas
-    LOGGER.debug(content)
+    LOGGER.debug(T("coal.logs.adx.runner.content_debug").format(content=content))
 
     return content
 
@@ -211,7 +211,7 @@ def send_runner_data(
             LOGGER.info(T("coal.logs.ingestion.table_created").format(table=k))
         else:
             LOGGER.error(T("coal.logs.ingestion.table_creation_failed").format(table=k))
-            LOGGER.error(r.get_exceptions())
+            LOGGER.error(T("coal.logs.ingestion.exceptions").format(exceptions=r.get_exceptions()))
             raise RuntimeError(f"Failed to create table {k}")
     insert_csv_files(
         files_data=csv_data,

cosmotech/coal/azure/adx/store.py

@@ -45,7 +45,7 @@ def send_table_data(
     Returns:
         tuple: (source_id, table_name)
     """
-    LOGGER.debug(f"Sending data to the table {table_name}")
+    LOGGER.debug(T("coal.logs.adx.store.sending_data").format(table_name=table_name))
     result = send_pyarrow_table_to_adx(ingest_client, database, table_name, data, operation_tag)
     return result.source_id, table_name
 
@@ -69,15 +69,15 @@ def process_tables(
     source_ids = []
     table_ingestion_id_mapping = dict()
 
-    LOGGER.debug("Listing tables")
+    LOGGER.debug(T("coal.logs.adx.store.listing_tables"))
     table_list = list(store.list_tables())
 
     for target_table_name in table_list:
-        LOGGER.info(f"Working on table: {target_table_name}")
+        LOGGER.info(T("coal.logs.adx.store.working_on_table").format(table_name=target_table_name))
         data = store.get_table(target_table_name)
 
         if data.num_rows < 1:
-            LOGGER.warning(f"Table {target_table_name} has no rows - skipping it")
+            LOGGER.warning(T("coal.logs.adx.store.table_empty").format(table_name=target_table_name))
             continue
 
         check_and_create_table(kusto_client, database, target_table_name, data)
@@ -140,14 +140,14 @@ def send_store_to_adx(
     """
     # Generate a unique operation tag if none provided
     operation_tag = tag or f"op-{str(uuid.uuid4())}"
-    LOGGER.debug(f"Starting ingestion operation with tag: {operation_tag}")
+    LOGGER.debug(T("coal.logs.adx.store.starting_ingestion").format(operation_tag=operation_tag))
 
     # Initialize clients
     kusto_client, ingest_client = initialize_clients(adx_uri, adx_ingest_uri)
     database = database_name
 
     # Load datastore
-    LOGGER.debug("Loading datastore")
+    LOGGER.debug(T("coal.logs.adx.store.loading_datastore"))
     store = Store(store_location=store_location)
 
     try:
@@ -156,7 +156,7 @@ def send_store_to_adx(
             store, kusto_client, ingest_client, database, operation_tag
         )
 
-        LOGGER.info("Store data was sent for ADX ingestion")
+        LOGGER.info(T("coal.logs.adx.store.data_sent"))
 
         # Monitor ingestion if wait is True
         has_failures = False
@@ -171,9 +171,9 @@ def send_store_to_adx(
         return True
 
     except Exception as e:
-        LOGGER.exception("Error during ingestion process")
+        LOGGER.exception(T("coal.logs.adx.store.ingestion_error"))
         # Perform rollback using the tag
-        LOGGER.warning(f"Dropping data with tag: {operation_tag}")
+        LOGGER.warning(T("coal.logs.adx.store.dropping_data").format(operation_tag=operation_tag))
         _drop_by_tag(kusto_client, database, operation_tag)
         raise e
 

cosmotech/coal/azure/adx/tables.py

@@ -53,12 +53,12 @@ def check_and_create_table(kusto_client: KustoClient, database: str, table_name:
     Returns:
         bool: True if the table was created, False if it already existed
     """
-    LOGGER.debug("  - Checking if table exists")
+    LOGGER.debug(T("coal.logs.adx.checking_table_exists"))
     if not table_exists(kusto_client, database, table_name):
         from cosmotech.coal.azure.adx.utils import create_column_mapping
 
         mapping = create_column_mapping(data)
-        LOGGER.debug("  - Does not exist, creating it")
+        LOGGER.debug(T("coal.logs.adx.creating_nonexistent_table"))
         create_table(kusto_client, database, table_name, mapping)
         return True
     return False
@@ -73,16 +73,16 @@ def _drop_by_tag(kusto_client: KustoClient, database: str, tag: str) -> None:
         database: The database name
         tag: The tag to drop data by
     """
-    LOGGER.info(f"Dropping data with tag: {tag}")
+    LOGGER.info(T("coal.logs.adx.dropping_data_by_tag").format(tag=tag))
 
     try:
         # Execute the drop by tag command
         drop_command = f'.drop extents <| .show database extents where tags has "drop-by:{tag}"'
         kusto_client.execute_mgmt(database, drop_command)
-        LOGGER.info("Drop by tag operation completed")
+        LOGGER.info(T("coal.logs.adx.drop_completed"))
     except Exception as e:
-        LOGGER.error(f"Error during drop by tag operation: {str(e)}")
-        LOGGER.exception("Drop by tag details")
+        LOGGER.error(T("coal.logs.adx.drop_error").format(error=str(e)))
+        LOGGER.exception(T("coal.logs.adx.drop_details"))
 
 
 def create_table(client: KustoClient, database: str, table_name: str, schema: Dict[str, str]) -> bool:

cosmotech/coal/azure/adx/utils.py

@@ -30,7 +30,7 @@ def create_column_mapping(data: pyarrow.Table) -> Dict[str, str]:
         try:
             ex = next(v for v in column.to_pylist() if v is not None)
         except StopIteration:
-            LOGGER.error(f"Column {column_name} has no content, defaulting it to string")
+            LOGGER.error(T("coal.logs.adx.utils.empty_column").format(column_name=column_name))
             mapping[column_name] = type_mapping(column_name, "string")
             continue
         else:

cosmotech/coal/cosmotech_api/parameters.py

@@ -18,6 +18,7 @@
 from typing import List, Dict, Any
 
 from cosmotech.coal.utils.logger import LOGGER
+from cosmotech.orchestrator.utils.translate import T
 
 
 def write_parameters(
@@ -34,14 +35,14 @@ def write_parameters(
     """
     if write_csv:
         tmp_parameter_file = os.path.join(parameter_folder, "parameters.csv")
-        LOGGER.info(f"Generating {tmp_parameter_file}")
+        LOGGER.info(T("coal.logs.runner.generating_file").format(file=tmp_parameter_file))
         with open(tmp_parameter_file, "w") as _file:
             _w = DictWriter(_file, fieldnames=["parameterId", "value", "varType", "isInherited"])
             _w.writeheader()
             _w.writerows(parameters)
 
     if write_json:
         tmp_parameter_file = os.path.join(parameter_folder, "parameters.json")
-        LOGGER.info(f"Generating {tmp_parameter_file}")
+        LOGGER.info(T("coal.logs.runner.generating_file").format(file=tmp_parameter_file))
         with open(tmp_parameter_file, "w") as _file:
             json.dump(parameters, _file, indent=2)

cosmotech/coal/cosmotech_api/run_data.py

@@ -47,7 +47,7 @@ def send_csv_to_run_data(
     source_dir = pathlib.Path(source_folder)
 
     if not source_dir.exists():
-        LOGGER.error(f"{source_dir} does not exists")
+        LOGGER.error(T("coal.errors.file_system.file_not_found").format(source_folder=source_dir))
         raise FileNotFoundError(f"{source_dir} does not exist")
 
     with get_api_client()[0] as api_client:
@@ -56,8 +56,8 @@ def send_csv_to_run_data(
             with open(csv_path) as _f:
                 dr = DictReader(_f)
                 table_name = csv_path.name.replace(".csv", "")
-                LOGGER.info(f"Sending data to table CD_{table_name}")
-                LOGGER.debug(f"  - Column list: {dr.fieldnames}")
+                LOGGER.info(T("coal.logs.run_data.sending_to_table").format(table_name=f"CD_{table_name}"))
+                LOGGER.debug(T("coal.logs.database.column_list").format(columns=dr.fieldnames))
                 data = []
 
                 for row in dr:
@@ -72,7 +72,7 @@ def send_csv_to_run_data(
                             n_row[k] = v
                     data.append(n_row)
 
-                LOGGER.info(f"  - Sending {len(data)} rows")
+                LOGGER.info(T("coal.logs.database.row_count").format(count=len(data)))
                 api_run.send_run_data(
                     organization_id,
                     workspace_id,
@@ -102,25 +102,25 @@ def send_store_to_run_data(
     source_dir = pathlib.Path(store_folder)
 
     if not source_dir.exists():
-        LOGGER.error(f"{source_dir} does not exists")
+        LOGGER.error(T("coal.errors.file_system.file_not_found").format(source_folder=source_dir))
         raise FileNotFoundError(f"{source_dir} does not exist")
 
     with get_api_client()[0] as api_client:
         api_run = RunApi(api_client)
         _s = Store()
         for table_name in _s.list_tables():
-            LOGGER.info(f"Sending data to table CD_{table_name}")
+            LOGGER.info(T("coal.logs.run_data.sending_to_table").format(table_name=f"CD_{table_name}"))
             data = convert_table_as_pylist(table_name)
             if not len(data):
-                LOGGER.info("  - No rows : skipping")
+                LOGGER.info(T("coal.logs.database.no_rows"))
                 continue
             fieldnames = _s.get_table_schema(table_name).names
             for row in data:
                 for field in fieldnames:
                     if row[field] is None:
                         del row[field]
-            LOGGER.debug(f"  - Column list: {fieldnames}")
-            LOGGER.info(f"  - Sending {len(data)} rows")
+            LOGGER.debug(T("coal.logs.database.column_list").format(columns=fieldnames))
+            LOGGER.info(T("coal.logs.database.row_count").format(count=len(data)))
             api_run.send_run_data(
                 organization_id,
                 workspace_id,
@@ -160,14 +160,14 @@ def load_csv_from_run_data(
             organization_id, workspace_id, runner_id, run_id, RunDataQuery(query=query)
         )
         if query_result.result:
-            LOGGER.info(f"Query returned {len(query_result.result)} rows")
+            LOGGER.info(T("coal.logs.database.query_results").format(count=len(query_result.result)))
             with open(target_dir / (file_name + ".csv"), "w") as _f:
                 headers = set()
                 for r in query_result.result:
                     headers = headers | set(r.keys())
                 dw = DictWriter(_f, fieldnames=sorted(headers))
                 dw.writeheader()
                 dw.writerows(query_result.result)
-            LOGGER.info(f"Results saved as {target_dir / file_name}.csv")
+            LOGGER.info(T("coal.logs.database.saved_results").format(file=f"{target_dir / file_name}.csv"))
         else:
-            LOGGER.info("No results returned by the query")
+            LOGGER.info(T("coal.logs.database.no_results"))

cosmotech/coal/cosmotech_api/run_template.py

@@ -58,7 +58,7 @@ def load_run_template_handlers(
             LOGGER.error(
                 T("coal.errors.workspace.not_found").format(workspace_id=workspace_id, organization_id=organization_id)
             )
-            LOGGER.debug(e.body)
+            LOGGER.debug(T("coal.logs.orchestrator.error_details").format(details=e.body))
             raise ValueError(f"Workspace {workspace_id} not found in organization {organization_id}")
         solution_id = r_data.solution.solution_id
 
@@ -86,7 +86,7 @@ def load_run_template_handlers(
                         solution=solution_id,
                     )
                 )
-                LOGGER.debug(e.body)
+                LOGGER.debug(T("coal.logs.orchestrator.error_details").format(details=e.body))
                 has_errors = True
                 continue
             LOGGER.info(T("coal.logs.orchestrator.extracting_handler").format(path=handler_path.absolute()))

cosmotech/coal/cosmotech_api/twin_data_layer.py

@@ -160,7 +160,7 @@ def get_dataset_id_from_runner(organization_id: str, workspace_id: str, runner_i
 
     if (datasets_len := len(runner_info.dataset_list)) != 1:
         LOGGER.error(T("coal.logs.runner.not_single_dataset").format(runner_id=runner_info.id, count=datasets_len))
-        LOGGER.debug(runner_info)
+        LOGGER.debug(T("coal.logs.runner.runner_info").format(info=runner_info))
         raise ValueError(f"Runner {runner_info.id} does not have exactly one dataset")
 
     return runner_info.dataset_list[0]
@@ -309,7 +309,7 @@ def _process_csv_file(
     if len(errors):
         LOGGER.error(T("coal.logs.storage.import_errors").format(count=len(errors)))
         for _err in errors:
-            LOGGER.error(str(_err))
+            LOGGER.error(T("coal.logs.storage.error_detail").format(error=str(_err)))
         raise ValueError(f"Error importing data from {file_path}")
 
 
@@ -340,7 +340,7 @@ def load_files_from_tdl(
         LOGGER.error(
             T("coal.logs.runner.dataset_state").format(dataset_id=dataset_id, status=dataset_info.ingestion_status)
         )
-        LOGGER.debug(dataset_info)
+        LOGGER.debug(T("coal.logs.runner.dataset_info").format(info=dataset_info))
         raise ValueError(f"Dataset {dataset_id} is not in SUCCESS state")
 
     # Create directory