Add Centralized Logger for Checks (#42775)

* add wip logging.py

* pass args to logging

* use logger in mypy, move logger to logging/init

* use env var

* minor log fix

* replace logging with logger

* replace prints with logs

* make log level args mutually exclusive

* fix arg precedence logic

* add unit test

* test fix

* test fix

* minor fix

* logic fix

* formatting fix

* merge

* clean test

* standardize log params

* copilot formatting fix

Co-authored-by: Copilot <[email protected]>

* remove levels arg

---------

Co-authored-by: Copilot <[email protected]>

Author: JennyPng

eng/tools/azure-sdk-tools/azpysdk/Check.py

@@ -11,6 +11,7 @@
 from ci_tools.functions import discover_targeted_packages, get_venv_call
 from ci_tools.variables import discover_repo_root
 from ci_tools.scenario import install_into_venv, get_venv_python
+from ci_tools.logging import logger
 
 # right now, we are assuming you HAVE to be in the azure-sdk-tools repo
 # we assume this because we don't know how a dev has installed this package, and might be
@@ -84,17 +85,17 @@ def get_targeted_directories(self, args: argparse.Namespace) -> List[ParsedSetup
             try:
                 targeted.append(ParsedSetup.from_path(targeted_dir))
             except Exception as e:
-                print("Error: Current directory does not appear to be a Python package (no setup.py or setup.cfg found). Remove '.' argument to run on child directories.")
-                print(f"Exception: {e}")
+                logger.error("Error: Current directory does not appear to be a Python package (no setup.py or setup.cfg found). Remove '.' argument to run on child directories.")
+                logger.error(f"Exception: {e}")
                 return []
         else:
             targeted_packages = discover_targeted_packages(args.target, targeted_dir)
             for pkg in targeted_packages:
                 try:
                     targeted.append(ParsedSetup.from_path(pkg))
                 except Exception as e:
-                    print(f"Unable to parse {pkg} as a Python package. Dumping exception detail and skipping.")
-                    print(f"Exception: {e}")
-                    print(traceback.format_exc())
+                    logger.error(f"Unable to parse {pkg} as a Python package. Dumping exception detail and skipping.")
+                    logger.error(f"Exception: {e}")
+                    logger.error(traceback.format_exc())
 
         return targeted

eng/tools/azure-sdk-tools/azpysdk/import_all.py

@@ -1,7 +1,6 @@
 import argparse
 import os
 import sys
-import logging
 import tempfile
 
 from typing import Optional,List
@@ -11,6 +10,7 @@
 from ci_tools.parsing import ParsedSetup
 from ci_tools.functions import discover_targeted_packages
 from ci_tools.scenario.generation import create_package_and_install
+from ci_tools.logging import logger
 
 # keyvault has dependency issue when loading private module _BearerTokenCredentialPolicyBase from azure.core.pipeline.policies
 # azure.core.tracing.opencensus and azure.eventhub.checkpointstoreblob.aio are skipped due to a known issue in loading azure.core.tracing.opencensus
@@ -42,7 +42,7 @@ def register(self, subparsers: "argparse._SubParsersAction", parent_parsers: Opt
     # todo: figure out venv abstraction mechanism via override
     def run(self, args: argparse.Namespace) -> int:
         """Run the import_all check command."""
-        print("Running import_all check in isolated venv...")
+        logger.info("Running import_all check in isolated venv...")
 
         targeted = self.get_targeted_directories(args)
 
@@ -69,7 +69,7 @@ def run(self, args: argparse.Namespace) -> int:
 
             if should_run_import_all(parsed.name):
                 # import all modules from current package
-                logging.info(
+                logger.info(
                     "Importing all modules from namespace [{0}] to verify dependency".format(
                         parsed.namespace
                     )
@@ -82,8 +82,8 @@ def run(self, args: argparse.Namespace) -> int:
                 ]
 
                 outcomes.append(check_call(commands))
-                logging.info("Verified module dependency, no issues found")
+                logger.info("Verified module dependency, no issues found")
             else:
-                logging.info("Package {} is excluded from dependency check".format(parsed.name))
+                logger.info("Package {} is excluded from dependency check".format(parsed.name))
 
         return max(outcomes) if outcomes else 0

eng/tools/azure-sdk-tools/azpysdk/main.py

@@ -16,6 +16,8 @@
 from .import_all import import_all
 from .mypy import mypy
 
+from ci_tools.logging import configure_logging, logger
+
 __all__ = ["main", "build_parser"]
 __version__ = "0.0.0"
 
@@ -29,6 +31,26 @@ def build_parser() -> argparse.ArgumentParser:
     parser.add_argument("--isolate", action="store_true", default=False,
                         help="If set, run in an isolated virtual environment.")
 
+    # mutually exclusive logging options
+    log_group = parser.add_mutually_exclusive_group()
+    log_group.add_argument(
+        "--quiet",
+        action="store_true",
+        default=False,
+        help="Enable quiet mode (only shows ERROR logs)"
+    )
+    log_group.add_argument(
+        "--verbose",
+        action="store_true",
+        default=False,
+        help="Enable verbose mode (shows DEBUG logs)"
+    )
+    log_group.add_argument(
+        "--log-level",
+        choices=["DEBUG", "INFO", "WARN", "ERROR", "FATAL"],
+        help="Set the logging level."
+    )
+
     common = argparse.ArgumentParser(add_help=False)
     common.add_argument(
         "target",
@@ -65,6 +87,8 @@ def main(argv: Optional[Sequence[str]] = None) -> int:
     parser = build_parser()
     args = parser.parse_args(argv)
 
+    configure_logging(args)
+
     if not hasattr(args, "func"):
         parser.print_help()
         return 1
@@ -73,10 +97,10 @@ def main(argv: Optional[Sequence[str]] = None) -> int:
         result = args.func(args)
         return int(result or 0)
     except KeyboardInterrupt:
-        print("Interrupted by user", file=sys.stderr)
+        logger.error("Interrupted by user")
         return 130
     except Exception as exc:  # pragma: no cover - simple top-level error handling
-        print(f"Error: {exc}", file=sys.stderr)
+        logger.error(f"Error: {exc}")
         return 2
 
 if __name__ == "__main__":

eng/tools/azure-sdk-tools/azpysdk/mypy.py

@@ -1,7 +1,6 @@
 import argparse
 import os
 import sys
-import logging
 import tempfile
 
 from typing import Optional, List
@@ -15,8 +14,7 @@
 from ci_tools.environment_exclusions import (
     is_check_enabled, is_typing_ignored
 )
-
-logging.getLogger().setLevel(logging.INFO)
+from ci_tools.logging import logger
 
 PYTHON_VERSION = "3.9"
 MYPY_VERSION = "1.14.1"
@@ -41,7 +39,7 @@ def register(self, subparsers: "argparse._SubParsersAction", parent_parsers: Opt
 
     def run(self, args: argparse.Namespace) -> int:
         """Run the mypy check command."""
-        print("Running mypy check in isolated venv...")
+        logger.info("Running mypy check in isolated venv...")
 
         set_envvar_defaults()
 
@@ -52,8 +50,10 @@ def run(self, args: argparse.Namespace) -> int:
         for parsed in targeted:
             package_dir = parsed.folder
             package_name = parsed.name
+            
             executable, staging_directory = self.get_executable(args.isolate, args.command, sys.executable, package_dir)
-            print(f"Processing {package_name} for mypy check")
+            logger.info(f"Processing {package_name} for mypy check")
+            
             create_package_and_install(
                 distribution_directory=staging_directory,
                 target_setup=package_dir,
@@ -74,14 +74,14 @@ def run(self, args: argparse.Namespace) -> int:
                 else:
                     pip_install([f"mypy=={MYPY_VERSION}"], True, executable, package_dir)
             except CalledProcessError as e:
-                print("Failed to install mypy:", e)
+                logger.error("Failed to install mypy:", e)
                 return e.returncode
 
-            logging.info(f"Running mypy against {package_name}")
+            logger.info(f"Running mypy against {package_name}")
 
             if not args.next and in_ci():
                 if not is_check_enabled(package_dir, "mypy", True) or is_typing_ignored(package_name):
-                    logging.info(
+                    logger.info(
                         f"Package {package_name} opts-out of mypy check. See https://aka.ms/python/typing-guide for information."
                     )
                     continue
@@ -101,17 +101,17 @@ def run(self, args: argparse.Namespace) -> int:
             src_code_error = None
             sample_code_error = None
             try:
-                logging.info(
+                logger.info(
                     f"Running mypy commands on src code: {src_code}"
                 )
                 results.append(check_call(src_code))
-                logging.info("Verified mypy, no issues found")
+                logger.info("Verified mypy, no issues found")
             except CalledProcessError as src_error:
                 src_code_error = src_error
                 results.append(src_error.returncode)
 
             if not args.next and in_ci() and not is_check_enabled(package_dir, "type_check_samples", True):
-                logging.info(
+                logger.info(
                     f"Package {package_name} opts-out of mypy check on samples."
                 )
                 continue
@@ -120,7 +120,7 @@ def run(self, args: argparse.Namespace) -> int:
                 samples = os.path.exists(os.path.join(package_dir, "samples"))
                 generated_samples = os.path.exists(os.path.join(package_dir, "generated_samples"))
                 if not samples and not generated_samples:
-                    logging.info(
+                    logger.info(
                         f"Package {package_name} does not have a samples directory."
                     )
                 else:
@@ -131,7 +131,7 @@ def run(self, args: argparse.Namespace) -> int:
                         os.path.join(package_dir, "samples" if samples else "generated_samples"),
                     ]
                     try:
-                        logging.info(
+                        logger.info(
                             f"Running mypy commands on sample code: {sample_code}"
                         )
                         results.append(check_call(sample_code))

eng/tools/azure-sdk-tools/azpysdk/whl.py

@@ -1,5 +1,4 @@
 import argparse
-import logging
 import tempfile
 import os
 from typing import Optional, List, Any
@@ -12,7 +11,7 @@
 from ci_tools.variables import set_envvar_defaults
 from ci_tools.parsing import ParsedSetup
 from ci_tools.scenario.generation import create_package_and_install
-
+from ci_tools.logging import logger
 
 class whl(Check):
     def __init__(self) -> None:
@@ -29,7 +28,7 @@ def register(self, subparsers: "argparse._SubParsersAction", parent_parsers: Opt
 
     def run(self, args: argparse.Namespace) -> int:
         """Run the whl check command."""
-        print("Running whl check...")
+        logger.info("Running whl check...")
 
         set_envvar_defaults()
 
@@ -64,7 +63,7 @@ def run(self, args: argparse.Namespace) -> int:
 
             # TODO: split sys.argv[1:] on -- and pass in everything after the -- as additional arguments
             # TODO: handle mark_args
-            logging.info(f"Invoke pytest for {pkg}")
+            logger.info(f"Invoke pytest for {pkg}")
             exit_code = run(
                 [executable, "-m", "pytest", "."] + [
                     "-rsfE",
@@ -83,9 +82,9 @@ def run(self, args: argparse.Namespace) -> int:
 
             if exit_code != 0:
                 if exit_code == 5 and is_error_code_5_allowed(parsed.folder, parsed.name):
-                    logging.info("Exit code 5 is allowed, continuing execution.")
+                    logger.info("Exit code 5 is allowed, continuing execution.")
                 else:
-                    logging.info(f"pytest failed with exit code {exit_code}.")
+                    logger.info(f"pytest failed with exit code {exit_code}.")
                     results.append(exit_code)
 
         # final result is the worst case of all the results

eng/tools/azure-sdk-tools/ci_tools/logging/__init__.py

@@ -4,9 +4,37 @@
 import os
 import datetime
 from subprocess import run
+import argparse
 
 LOGLEVEL = getattr(logging, os.environ.get("LOGLEVEL", "INFO").upper())
 
+logger = logging.getLogger("azure-sdk-tools")
+
+def configure_logging(
+    args: argparse.Namespace,
+    fmt: str = "%(asctime)s [%(levelname)s] %(name)s: %(message)s"
+) -> None:
+    """
+    Configures the shared logger. Should be called **once** at startup.
+    """
+    # use cli arg > log level arg > env var
+
+    if args.quiet:
+        numeric_level = logging.ERROR
+    elif args.verbose:
+        numeric_level = logging.DEBUG
+    elif not args.log_level:
+        numeric_level = getattr(logging, os.environ.get("LOGLEVEL", "INFO").upper())
+    else:
+        numeric_level = getattr(logging, args.log_level.upper(), None)
+
+    if not isinstance(numeric_level, int):
+        raise ValueError(f"Invalid log level: {numeric_level}")
+    logger.setLevel(numeric_level)
+
+    # Propagate logger config globally if needed
+    logging.basicConfig(level=numeric_level, format=fmt)
+
 
 def now() -> str:
     return datetime.datetime.now().strftime("%Y-%m-%dT%H.%M.%S")

eng/tools/azure-sdk-tools/tests/test_logging_config.py

@@ -0,0 +1,21 @@
+
+import logging
+from ci_tools.logging import configure_logging, logger
+from unittest.mock import patch
+import pytest
+import argparse
+import os
+
+@pytest.mark.parametrize("cli_args,level_env,expected_level", [
+    (argparse.Namespace(quiet=True, verbose=False, log_level=None), "INFO", logging.ERROR),  
+    (argparse.Namespace(quiet=False, verbose=True, log_level=None), "INFO", logging.DEBUG),  
+    (argparse.Namespace(quiet=False, verbose=False, log_level="ERROR"), "INFO", logging.ERROR),  
+    (argparse.Namespace(quiet=False, verbose=False, log_level=None), "WARN", logging.WARNING),  
+])
+@patch("logging.basicConfig")
+def test_configure_logging_various_levels(mock_basic_config, cli_args, level_env, expected_level, monkeypatch):
+    monkeypatch.setenv("LOGLEVEL", level_env)
+    assert os.environ["LOGLEVEL"] == level_env
+    configure_logging(cli_args)
+    assert logger.level == expected_level
+    mock_basic_config.assert_called_with(level=expected_level, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s")