[ubuntu/noble]: patch out breaking changes

.github/labeler.yml

@@ -1,2 +1,7 @@
 documentation:
-- doc/*
+- all:
+  - changed-files:
+    - any-glob-to-any-file:
+      - 'doc/*'
+      - 'cloudinit/config/schemas/*'
+  - base-branch: ['main']

.github/workflows/labeler.yaml

@@ -0,0 +1,9 @@
+name: "Pull Request Labeler"
+on:
+- pull_request_target
+
+jobs:
+  labeler:
+    runs-on: ubuntu-latest
+    steps:
+    - uses: actions/labeler@v5

cloudinit/cmd/devel/logs.py

@@ -7,14 +7,16 @@
 """Define 'collect-logs' utility and handler to include in cloud-init cmd."""
 
 import argparse
+import logging
 import os
+import pathlib
 import shutil
 import subprocess
 import sys
 from datetime import datetime, timezone
-from pathlib import Path
-from typing import NamedTuple, Optional
+from typing import List, NamedTuple, Optional, cast
 
+from cloudinit import log
 from cloudinit.cmd.devel import read_cfg_paths
 from cloudinit.stages import Init
 from cloudinit.subp import ProcessExecutionError, subp
@@ -27,6 +29,8 @@
     write_file,
 )
 
+LOG = cast(log.CustomLoggerType, logging.getLogger(__name__))
+
 
 class LogPaths(NamedTuple):
     userdata_raw: str
@@ -96,7 +100,9 @@ class ApportFile(NamedTuple):
 ]
 
 
-def get_parser(parser=None):
+def get_parser(
+    parser: Optional[argparse.ArgumentParser] = None,
+) -> argparse.ArgumentParser:
     """Build or extend and arg parser for collect-logs utility.
 
     @param parser: Optional existing ArgumentParser instance representing the
@@ -141,7 +147,7 @@ def get_parser(parser=None):
     return parser
 
 
-def _get_copytree_ignore_files(paths: LogPaths):
+def _get_copytree_ignore_files(paths: LogPaths) -> List[str]:
     """Return a list of files to ignore for /run/cloud-init directory"""
     ignored_files = [
         "hook-hotplug-cmd",  # named pipe for hotplug
@@ -152,111 +158,109 @@ def _get_copytree_ignore_files(paths: LogPaths):
     return ignored_files
 
 
-def _write_command_output_to_file(cmd, filename, msg, verbosity):
+def _write_command_output_to_file(
+    cmd: List[str],
+    file_path: pathlib.Path,
+    msg: str,
+) -> Optional[str]:
     """Helper which runs a command and writes output or error to filename."""
-    ensure_dir(os.path.dirname(filename))
+    file_path.parent.mkdir(parents=True, exist_ok=True)
     try:
         output = subp(cmd).stdout
     except ProcessExecutionError as e:
-        write_file(filename, str(e))
-        _debug("collecting %s failed.\n" % msg, 1, verbosity)
+        write_file(file_path, str(e))
+        LOG.debug("collecting %s failed.", msg)
+        output = None
     else:
-        write_file(filename, output)
-        _debug("collected %s\n" % msg, 1, verbosity)
-        return output
+        write_file(file_path, output)
+        LOG.debug("collected %s to file '%s'", msg, file_path.stem)
+    return output
 
 
-def _stream_command_output_to_file(cmd, filename, msg, verbosity):
-    """Helper which runs a command and writes output or error to filename."""
-    ensure_dir(os.path.dirname(filename))
+def _stream_command_output_to_file(
+    cmd: List[str], file_path: pathlib.Path, msg: str
+) -> None:
+    """Helper which runs a command and writes output or error to filename.
+
+    `subprocess.call` is invoked directly here to stream output to the file.
+    Otherwise memory usage can be high for large outputs.
+    """
+    file_path.parent.mkdir(parents=True, exist_ok=True)
     try:
-        with open(filename, "w") as f:
+        with file_path.open("w") as f:
             subprocess.call(cmd, stdout=f, stderr=f)  # nosec B603
     except OSError as e:
-        write_file(filename, str(e))
-        _debug("collecting %s failed.\n" % msg, 1, verbosity)
+        write_file(file_path, str(e))
+        LOG.debug("collecting %s failed.", msg)
     else:
-        _debug("collected %s\n" % msg, 1, verbosity)
+        LOG.debug("collected %s to file '%s'", msg, file_path.stem)
 
 
-def _debug(msg, level, verbosity):
-    if level <= verbosity:
-        sys.stderr.write(msg)
-
-
-def _collect_file(path, out_dir, verbosity):
+def _collect_file(path: str, out_dir: str) -> None:
     if os.path.isfile(path):
         copy(path, out_dir)
-        _debug("collected file: %s\n" % path, 1, verbosity)
+        LOG.debug("collected file: %s", path)
     else:
-        _debug("file %s did not exist\n" % path, 2, verbosity)
+        LOG.trace("file %s did not exist", path)
 
 
-def _collect_installer_logs(
-    log_dir: str, include_userdata: bool, verbosity: int
-):
+def _collect_installer_logs(log_dir: str, include_userdata: bool) -> None:
     """Obtain subiquity logs and config files."""
     for src_file in INSTALLER_APPORT_FILES:
-        destination_dir = Path(log_dir + src_file.path).parent
+        destination_dir = pathlib.Path(log_dir + src_file.path).parent
         if not destination_dir.exists():
             ensure_dir(str(destination_dir))
-        _collect_file(src_file.path, str(destination_dir), verbosity)
+        _collect_file(src_file.path, str(destination_dir))
     if include_userdata:
         for src_file in INSTALLER_APPORT_SENSITIVE_FILES:
-            destination_dir = Path(log_dir + src_file.path).parent
+            destination_dir = pathlib.Path(log_dir + src_file.path).parent
             if not destination_dir.exists():
                 ensure_dir(str(destination_dir))
-            _collect_file(src_file.path, str(destination_dir), verbosity)
+            _collect_file(src_file.path, str(destination_dir))
 
 
-def _collect_version_info(log_dir: str, verbosity: int):
+def _collect_version_info(log_dir: str) -> None:
     version = _write_command_output_to_file(
         cmd=["cloud-init", "--version"],
-        filename=os.path.join(log_dir, "version"),
+        file_path=pathlib.Path(log_dir, "version"),
         msg="cloud-init --version",
-        verbosity=verbosity,
     )
     dpkg_ver = _write_command_output_to_file(
         cmd=["dpkg-query", "--show", "-f=${Version}\n", "cloud-init"],
-        filename=os.path.join(log_dir, "dpkg-version"),
+        file_path=pathlib.Path(log_dir, "dpkg-version"),
         msg="dpkg version",
-        verbosity=verbosity,
     )
     if not version:
-        version = dpkg_ver if dpkg_ver else "not-available"
-    _debug("collected cloud-init version: %s\n" % version, 1, verbosity)
+        version = dpkg_ver or "not-available"
 
 
-def _collect_system_logs(log_dir: str, verbosity: int):
+def _collect_system_logs(log_dir: str) -> None:
     _stream_command_output_to_file(
         cmd=["dmesg"],
-        filename=os.path.join(log_dir, "dmesg.txt"),
+        file_path=pathlib.Path(log_dir, "dmesg.txt"),
         msg="dmesg output",
-        verbosity=verbosity,
     )
     _stream_command_output_to_file(
         cmd=["journalctl", "--boot=0", "-o", "short-precise"],
-        filename=os.path.join(log_dir, "journal.txt"),
+        file_path=pathlib.Path(log_dir, "journal.txt"),
         msg="systemd journal of current boot",
-        verbosity=verbosity,
     )
 
 
 def _collect_cloudinit_logs(
     log_dir: str,
-    verbosity: int,
     init: Init,
     paths: LogPaths,
     include_userdata: bool,
-):
-    for log in get_config_logfiles(init.cfg):
-        _collect_file(log, log_dir, verbosity)
+) -> None:
+    for logfile in get_config_logfiles(init.cfg):
+        _collect_file(logfile, log_dir)
     if include_userdata:
         user_data_file = paths.userdata_raw
-        _collect_file(user_data_file, log_dir, verbosity)
+        _collect_file(user_data_file, log_dir)
 
 
-def _collect_run_dir(log_dir: str, verbosity: int, paths: LogPaths):
+def _collect_run_dir(log_dir: str, paths: LogPaths) -> None:
     run_dir = os.path.join(log_dir, "run")
     ensure_dir(run_dir)
     if os.path.exists(paths.run_dir):
@@ -267,38 +271,31 @@ def _collect_run_dir(log_dir: str, verbosity: int, paths: LogPaths):
                 ignore=lambda _, __: _get_copytree_ignore_files(paths),
             )
         except shutil.Error as e:
-            sys.stderr.write("Failed collecting file(s) due to error:\n")
-            sys.stderr.write(str(e) + "\n")
-        _debug("collected dir %s\n" % paths.run_dir, 1, verbosity)
+            LOG.warning("Failed collecting file(s) due to error: %s", e)
+        LOG.debug("collected directory: %s", paths.run_dir)
     else:
-        _debug(
-            "directory '%s' did not exist\n" % paths.run_dir,
-            1,
-            verbosity,
-        )
+        LOG.debug("directory '%s' did not exist", paths.run_dir)
     if os.path.exists(os.path.join(paths.run_dir, "disabled")):
         # Fallback to grab previous cloud/data
-        cloud_data_dir = Path(paths.cloud_data)
+        cloud_data_dir = pathlib.Path(paths.cloud_data)
         if cloud_data_dir.exists():
             shutil.copytree(
                 str(cloud_data_dir),
-                Path(log_dir + str(cloud_data_dir)),
+                pathlib.Path(log_dir + str(cloud_data_dir)),
             )
 
 
-def collect_logs(
-    tarfile: str, include_userdata: bool, verbosity: int = 0
-) -> int:
+def collect_logs(tarfile: str, include_userdata: bool) -> int:
     """Collect all cloud-init logs and tar them up into the provided tarfile.
 
     @param tarfile: The path of the tar-gzipped file to create.
     @param include_userdata: Boolean, true means include user-data.
     @return: 0 on success, 1 on failure.
     """
     if include_userdata and os.getuid() != 0:
-        sys.stderr.write(
-            "To include userdata, root user is required."
-            " Try sudo cloud-init collect-logs\n"
+        LOG.error(
+            "To include userdata, root user is required. "
+            "Try sudo cloud-init collect-logs"
         )
         return 1
 
@@ -312,33 +309,39 @@ def collect_logs(
         init.read_cfg()
         paths = get_log_paths(init)
 
-        _collect_version_info(log_dir, verbosity)
-        _collect_system_logs(log_dir, verbosity)
-        _collect_cloudinit_logs(
-            log_dir, verbosity, init, paths, include_userdata
-        )
-        _collect_installer_logs(log_dir, include_userdata, verbosity)
-        _collect_run_dir(log_dir, verbosity, paths)
+        _collect_version_info(log_dir)
+        _collect_system_logs(log_dir)
+        _collect_cloudinit_logs(log_dir, init, paths, include_userdata)
+        _collect_installer_logs(log_dir, include_userdata)
+        _collect_run_dir(log_dir, paths)
         with chdir(tmp_dir):
             subp(["tar", "czvf", tarfile, log_dir.replace(f"{tmp_dir}/", "")])
-    sys.stderr.write("Wrote %s\n" % tarfile)
+    LOG.info("Wrote %s", tarfile)
     return 0
 
 
-def handle_collect_logs_args(name, args):
+def _setup_logger(verbosity: int) -> None:
+    log.reset_logging()
+    if verbosity == 0:
+        level = logging.INFO
+    elif verbosity == 1:
+        level = logging.DEBUG
+    else:
+        level = log.TRACE
+    LOG.setLevel(level)
+    handler = logging.StreamHandler()
+    handler.setFormatter(logging.Formatter("%(message)s"))
+    LOG.addHandler(handler)
+
+
+def handle_collect_logs_args(_name: str, args: argparse.Namespace) -> int:
     """Handle calls to 'cloud-init collect-logs' as a subcommand."""
+    _setup_logger(args.verbosity)
     return collect_logs(
         tarfile=args.tarfile,
         include_userdata=args.userdata,
-        verbosity=args.verbosity,
     )
 
 
-def main():
-    """Tool to collect and tar all cloud-init related logs."""
-    parser = get_parser()
-    return handle_collect_logs_args("collect-logs", parser.parse_args())
-
-
 if __name__ == "__main__":
-    sys.exit(main())
+    sys.exit(handle_collect_logs_args("", get_parser().parse_args()))