refactor logger

Previously, we would invoke logger before initilizing it, which was a
bug.

Additionally, we had multiple instances of logger spread out throughout
the code. It seems cleaner to define this once, and provide a single
method to update the verbosity.

Signed-off-by: Salvatore Daniele <[email protected]>

dpu-tools

@@ -4,7 +4,7 @@ from __future__ import annotations
 from abc import ABC, abstractmethod
 import argparse
 import sys
-import logging
+from logger import logger
 from typing import Optional
 from utils.fwutils import BFFirmware, IPUFirmware, cx_fwup
 from utils.common_ipu import (
@@ -18,13 +18,10 @@ from utils.common import (
     DPUType,
     detect_dpu_type,
     scan_for_dpus,
-    setup_logging,
     run,
 )
 from utils.pxeboot import Pxeboot
 
-logger = logging.getLogger(__name__)
-
 
 class DPUTools(ABC):
     def __init__(self, parser: argparse.ArgumentParser) -> None:
@@ -70,9 +67,6 @@ class DPUTools(ABC):
         for i, (k, (d, kind)) in enumerate(devs.items()):
             print(f"{i: 5d}  {k.ljust(8)}  {d.ljust(12)}  {kind}")
 
-    def setup_logging(self) -> None:
-        setup_logging(self.args.verbose)
-
     def setup_arguments(self, parser: argparse.ArgumentParser) -> argparse.Namespace:
         """Add common arguments and subcommands."""
         # Add global arguments
@@ -300,11 +294,11 @@ class IPUTools(DPUTools):
         fw.reflash_ipu()
 
     def firmware_reset(self) -> None:
-        result = get_current_version(self.args.imc_address, logger=logger)
+        result = get_current_version(self.args.imc_address)
         if result.returncode:
             logger.debug("Failed with ssh, trying minicom!")
             try:
-                minicom_get_version(logger=logger)
+                minicom_get_version()
             except Exception as e:
                 logger.error(f"Error ssh try: {result.err}")
                 logger.error(f"Exception with minicom: {e}")
@@ -320,11 +314,11 @@ class IPUTools(DPUTools):
         fw.reflash_ipu()
 
     def firmware_version(self) -> None:
-        result = get_current_version(self.args.imc_address, logger=logger)
+        result = get_current_version(self.args.imc_address)
         if result.returncode:
             logger.debug("Failed with ssh, trying minicom!")
             try:
-                minicom_get_version(logger=logger)
+                minicom_get_version()
             except Exception as e:
                 logger.error(f"Error ssh try: {result.err}")
                 logger.error(f"Exception with minicom: {e}")
@@ -388,7 +382,7 @@ def main() -> None:
         logger.debug(f"Dpu_type was not given, detecting it...")
         detected = detect_dpu_type()
         if detected.returncode != 0:
-            logging.error(f"Couldn't detect DPU type: {detected.err}")
+            logger.error(f"Couldn't detect DPU type: {detected.err}")
             sys.exit(detected.returncode)
         dpu_type = detected.out.upper()
         logger.debug(f"Detected dpu_type: {dpu_type}")
@@ -405,7 +399,7 @@ def main() -> None:
         exit(1)
     assert dpu_tools is not None
 
-    dpu_tools.setup_logging()
+    logger.setup_logging(dpu_tools.args.verbose)
     dpu_tools.dispatch()
 
 

logger.py

@@ -0,0 +1,19 @@
+import logging
+import sys
+
+
+def setup_logging(verbose: bool = False) -> None:
+    log_level = logging.DEBUG if verbose else logging.INFO
+
+    logging.basicConfig(
+        level=log_level,
+        format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+        handlers=[
+            logging.StreamHandler(sys.stdout),
+        ],
+    )
+
+
+setup_logging(verbose=False)
+
+logger = logging.getLogger(__name__)

utils/common.py

@@ -1,8 +1,7 @@
 import subprocess
-import logging
+from logger import logger
 from typing import IO
 import requests
-import sys
 import tarfile
 import os
 import re
@@ -25,24 +24,6 @@ class Result:
     returncode: int
 
 
-def setup_logging(verbose: bool) -> None:
-    if verbose:
-        log_level = logging.DEBUG
-    else:
-        log_level = logging.INFO
-
-    logging.basicConfig(
-        level=log_level,
-        format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
-        handlers=[
-            logging.StreamHandler(sys.stdout),  # Log to stdout
-        ],
-    )
-
-
-logger = logging.getLogger(__name__)
-
-
 def run(command: str, capture_output: bool = True, dry_run: bool = False) -> Result:
     """
     This run command is able to both output to the screen and capture its respective stream into a Result, using multithreading

utils/common_bf.py

@@ -1,5 +1,5 @@
 import dataclasses
-import logging
+from logger import logger
 import os
 import sys
 import argparse
@@ -16,9 +16,6 @@ class Result:
     returncode: int
 
 
-logger = logging.getLogger(__name__)
-
-
 def all_interfaces() -> dict[str, str]:
     out = run("lshw -c network -businfo").out
     ret = {}

utils/common_ipu.py

@@ -1,4 +1,4 @@
-import logging
+from logger import logger
 import os
 import re
 import pexpect
@@ -9,8 +9,6 @@
 
 VERSIONS = ["1.2.0.7550", "1.6.2.9418", "1.8.0.10052"]
 
-logger = logging.getLogger(__name__)
-
 
 def find_image(
     extracted_files: list[str], bin_file_prefix: str, identifier: str = ""
@@ -27,9 +25,7 @@ def find_image(
     )
 
 
-def get_current_version(
-    imc_address: str, logger: logging.Logger, dry_run: bool = False
-) -> Result:
+def get_current_version(imc_address: str, dry_run: bool = False) -> Result:
     logger.debug("Getting Version via SSH")
     version = ""
     # Execute the commands over SSH with dry_run handling
@@ -49,7 +45,7 @@ def get_current_version(
     return Result(version, result.err, result.returncode)
 
 
-def minicom_get_version(logger: logging.Logger) -> str:
+def minicom_get_version() -> str:
     version = ""
     run("pkill -9 minicom")
     logger.debug("Configuring minicom")