chore: add timer to io and string manipulation code

refactors existing code to utilize timer codepaths:

- replace manual timer implementations in subp.py and sockets.py
- replace open() / read() calls with util.load_text_file() where appropriate

cloudinit/atomic_helper.py

@@ -7,25 +7,33 @@
 import tempfile
 from base64 import b64decode, b64encode
 
-from cloudinit import util
+from cloudinit import performance, util
 
 _DEF_PERMS = 0o644
 LOG = logging.getLogger(__name__)
 
 
+@performance.timed("Base64 decoding")
 def b64d(source):
-    # Base64 decode some data, accepting bytes or unicode/str, and returning
-    # str/unicode if the result is utf-8 compatible, otherwise returning bytes.
+    """base64 decode data
+
+    :param source: a bytes or str to decode
+    :return: base64 as a decoded str if utf-8 encoded, otherwise bytes
+    """
     decoded = b64decode(source)
     try:
         return decoded.decode("utf-8")
     except UnicodeDecodeError:
         return decoded
 
 
+@performance.timed("Base64 encoding")
 def b64e(source):
-    # Base64 encode some data, accepting bytes or unicode/str, and returning
-    # str/unicode if the result is utf-8 compatible, otherwise returning bytes.
+    """base64 encode data
+
+    :param source: a bytes or str to decode
+    :return: base64 encoded str
+    """
     if not isinstance(source, bytes):
         source = source.encode("utf-8")
     return b64encode(source).decode("utf-8")
@@ -34,8 +42,15 @@ def b64e(source):
 def write_file(
     filename, content, mode=_DEF_PERMS, omode="wb", preserve_mode=False
 ):
-    # open filename in mode 'omode', write content, set permissions to 'mode'
+    """open filename in mode omode, write content, set permissions to mode"""
+
+    with performance.Timed(f"Writing {filename}"):
+        return _write_file(filename, content, mode, omode, preserve_mode)
 
+
+def _write_file(
+    filename, content, mode=_DEF_PERMS, omode="wb", preserve_mode=False
+):
     if preserve_mode:
         try:
             file_stat = os.stat(filename)
@@ -75,6 +90,7 @@ def json_serialize_default(_obj):
         return "Warning: redacted unserializable type {0}".format(type(_obj))
 
 
+@performance.timed("Dumping json")
 def json_dumps(data):
     """Return data in nicely formatted json."""
     return json.dumps(

cloudinit/cmd/main.py

@@ -129,6 +129,7 @@ def welcome_format(action):
     )
 
 
+@performance.timed("Closing stdin")
 def close_stdin(logger: Callable[[str], None] = LOG.debug):
     """
     reopen stdin as /dev/null to ensure no side effects
@@ -311,7 +312,7 @@ def purge_cache_on_python_version_change(init):
         init.paths.get_cpath("data"), "python-version"
     )
     if os.path.exists(python_version_path):
-        cached_python_version = open(python_version_path).read()
+        cached_python_version = util.load_text_file(python_version_path)
         # The Python version has changed out from under us, anything that was
         # pickled previously is likely useless due to API changes.
         if cached_python_version != current_python_version:

cloudinit/config/cc_growpart.py

@@ -20,7 +20,7 @@
 from pathlib import Path
 from typing import Optional, Tuple
 
-from cloudinit import lifecycle, subp, temp_utils, util
+from cloudinit import lifecycle, performance, subp, temp_utils, util
 from cloudinit.cloud import Cloud
 from cloudinit.config import Config
 from cloudinit.config.schema import MetaSchema
@@ -318,7 +318,9 @@ def resize_encrypted(blockdev, partition) -> Tuple[str, str]:
     if not KEYDATA_PATH.exists():
         return (RESIZE.SKIPPED, "No encryption keyfile found")
     try:
-        with KEYDATA_PATH.open() as f:
+        with performance.Timed(
+            f"Reading {KEYDATA_PATH}"
+        ), KEYDATA_PATH.open() as f:
             keydata = json.load(f)
         key = keydata["key"]
         decoded_key = base64.b64decode(key)

cloudinit/config/modules.py

@@ -12,7 +12,14 @@
 from types import ModuleType
 from typing import Dict, List, NamedTuple, Optional
 
-from cloudinit import config, importer, lifecycle, type_utils, util
+from cloudinit import (
+    config,
+    importer,
+    lifecycle,
+    performance,
+    type_utils,
+    util,
+)
 from cloudinit.distros import ALL_DISTROS
 from cloudinit.helpers import ConfigMerger
 from cloudinit.reporting.events import ReportEventStack
@@ -283,11 +290,15 @@ def _run_modules(self, mostly_mods: List[ModuleDetails]):
                             deprecated_version="23.2",
                         )
                         func_args.update({"log": LOG})
-                    ran, _r = cc.run(
-                        run_name, mod.handle, func_args, freq=freq
-                    )
+
+                    with performance.Timed("and", log_mode="skip") as timer:
+                        ran, _r = cc.run(
+                            run_name, mod.handle, func_args, freq=freq
+                        )
                     if ran:
-                        myrep.message = "%s ran successfully" % run_name
+                        myrep.message = (
+                            f"{run_name} ran successfully {timer.output}"
+                        )
                     else:
                         myrep.message = "%s previously ran" % run_name
 

cloudinit/config/schema.py

@@ -31,7 +31,7 @@
 
 import yaml
 
-from cloudinit import features, importer, lifecycle, safeyaml
+from cloudinit import features, importer, lifecycle, performance, safeyaml
 from cloudinit.cmd.devel import read_cfg_paths
 from cloudinit.handlers import INCLUSION_TYPES_MAP, type_from_starts_with
 from cloudinit.helpers import Paths
@@ -697,6 +697,7 @@ def netplan_validate_network_schema(
     return True
 
 
+@performance.timed("Validating schema")
 def validate_cloudconfig_schema(
     config: dict,
     schema: Optional[dict] = None,

cloudinit/dmi.py

@@ -5,7 +5,7 @@
 from collections import namedtuple
 from typing import Optional
 
-from cloudinit import subp
+from cloudinit import performance, subp
 from cloudinit.util import (
     is_container,
     is_DragonFlyBSD,
@@ -91,7 +91,9 @@ def _read_dmi_syspath(key: str) -> Optional[str]:
         return None
 
     try:
-        with open(dmi_key_path, "rb") as fp:
+        with performance.Timed(f"Reading {dmi_key_path}"), open(
+            dmi_key_path, "rb"
+        ) as fp:
             key_data = fp.read()
     except PermissionError:
         LOG.debug("Could not read %s", dmi_key_path)

cloudinit/log/log_util.py

@@ -2,7 +2,9 @@
 import os
 import sys
 
-from cloudinit.performance import Timed
+from cloudinit.performance import timed
+
+LOG = logging.getLogger(__name__)
 
 
 def logexc(
@@ -12,6 +14,7 @@ def logexc(
     log.debug(msg, exc_info=exc_info, *args)
 
 
+@timed("Writing to console")
 def write_to_console(conpath, text):
     with open(conpath, "w") as wfh:
         wfh.write(text)

cloudinit/net/eni.py

@@ -7,7 +7,7 @@
 import re
 from typing import Optional
 
-from cloudinit import subp, util
+from cloudinit import performance, subp, util
 from cloudinit.net import ParserError, renderer, subnet_is_ipv6
 from cloudinit.net.network_state import NetworkState
 
@@ -208,8 +208,7 @@ def _parse_deb_config_data(ifaces, contents, src_dir, src_path):
                     )
                 ]
                 for entry in dir_contents:
-                    with open(entry, "r") as fp:
-                        src_data = fp.read().strip()
+                    src_data = util.load_text_file(entry).strip()
                     abs_entry = os.path.abspath(entry)
                     _parse_deb_config_data(
                         ifaces, src_data, os.path.dirname(abs_entry), abs_entry
@@ -308,8 +307,9 @@ def _parse_deb_config_data(ifaces, contents, src_dir, src_path):
             ifaces[iface]["auto"] = False
 
 
+@performance.timed("Converting eni data")
 def convert_eni_data(eni_data):
-    # return a network config representation of what is in eni_data
+    """Return a network config representation of what is in eni_data"""
     ifaces = {}
     _parse_deb_config_data(ifaces, eni_data, src_dir=None, src_path=None)
     return _ifaces_to_net_config_data(ifaces)

cloudinit/reporting/events.py

@@ -9,10 +9,12 @@
 They can be published to registered handlers with report_event.
 """
 import base64
+import logging
 import os.path
 import time
 from typing import List
 
+from cloudinit import performance
 from cloudinit.reporting import (
     available_handlers,
     instantiated_handler_registry,
@@ -23,6 +25,7 @@
 START_EVENT_TYPE = "start"
 
 DEFAULT_EVENT_ORIGIN = "cloudinit"
+LOG = logging.getLogger(__name__)
 
 
 class _nameset(set):
@@ -301,7 +304,9 @@ def _collect_file_info(files):
         if not os.path.isfile(fname):
             content = None
         else:
-            with open(fname, "rb") as fp:
+            with performance.Timed(f"Reading {fname}"), open(
+                fname, "rb"
+            ) as fp:
                 content = base64.b64encode(fp.read()).decode()
         ret.append({"path": fname, "content": content, "encoding": "base64"})
     return ret

cloudinit/reporting/handlers.py

@@ -14,7 +14,7 @@
 from threading import Event
 from typing import Union
 
-from cloudinit import url_helper, util
+from cloudinit import performance, url_helper, util
 from cloudinit.registry import DictRegistry
 
 LOG = logging.getLogger(__name__)
@@ -310,7 +310,9 @@ def _decode_kvp_item(self, record_data):
         return {"key": k, "value": v}
 
     def _append_kvp_item(self, record_data):
-        with open(self._kvp_file_path, "ab") as f:
+        with performance.Timed(f"Appending {self._kvp_file_path}"), open(
+            self._kvp_file_path, "ab"
+        ) as f:
             fcntl.flock(f, fcntl.LOCK_EX)
             for data in record_data:
                 f.write(data)

cloudinit/safeyaml.py

@@ -4,12 +4,17 @@
 #
 # This file is part of cloud-init. See LICENSE file for license information.
 
+import logging
 from collections import defaultdict
 from itertools import chain
 from typing import Any, Dict, List, Tuple
 
 import yaml
 
+from cloudinit import performance
+
+LOG = logging.getLogger(__name__)
+
 
 # SchemaPathMarks track the path to an element within a loaded YAML file.
 # The start_mark and end_mark contain the row and column indicators
@@ -237,6 +242,7 @@ def ignore_aliases(self, data):
         return True
 
 
+@performance.timed("Loading yaml")
 def load_with_marks(blob) -> Tuple[Any, Dict[str, int]]:
     """Perform YAML SafeLoad and track start and end marks during parse.
 
@@ -261,12 +267,13 @@ def load_with_marks(blob) -> Tuple[Any, Dict[str, int]]:
 def dumps(obj, explicit_start=True, explicit_end=True, noalias=False):
     """Return data in nicely formatted yaml."""
 
-    return yaml.dump(
-        obj,
-        line_break="\n",
-        indent=4,
-        explicit_start=explicit_start,
-        explicit_end=explicit_end,
-        default_flow_style=False,
-        Dumper=(NoAliasSafeDumper if noalias else yaml.dumper.SafeDumper),
-    )
+    with performance.Timed("Dumping yaml"):
+        return yaml.dump(
+            obj,
+            line_break="\n",
+            indent=4,
+            explicit_start=explicit_start,
+            explicit_end=explicit_end,
+            default_flow_style=False,
+            Dumper=(NoAliasSafeDumper if noalias else yaml.dumper.SafeDumper),
+        )

cloudinit/socket.py

@@ -4,9 +4,9 @@
 import os
 import socket
 import sys
-import time
 from contextlib import suppress
 
+from cloudinit import performance
 from cloudinit.settings import DEFAULT_RUN_DIR
 
 LOG = logging.getLogger(__name__)
@@ -109,14 +109,14 @@ def __enter__(self):
             "STATUS=Waiting on external services to "
             f"complete before starting the {self.stage} stage."
         )
-        start_time = time.monotonic()
         # block until init system sends us data
         # the first value returned contains a message from the init system
         #     (should be "start")
         # the second value contains the path to a unix socket on which to
         #     reply, which is expected to be /path/to/{self.stage}-return.sock
         sock = self.sockets[self.stage]
-        chunk, self.remote = sock.recvfrom(5)
+        with performance.Timed(f"Waiting to start stage {self.stage}"):
+            chunk, self.remote = sock.recvfrom(5)
 
         if b"start" != chunk:
             # The protocol expects to receive a command "start"
@@ -130,10 +130,7 @@ def __enter__(self):
             self.__exit__(None, None, None)
             raise ValueError(f"Unexpected path to unix socket: {self.remote}")
 
-        total = time.monotonic() - start_time
-        time_msg = f"took {total: .3f}s to " if total > 0.01 else ""
         sd_notify(f"STATUS=Running ({self.stage} stage)")
-        LOG.debug("sync(%s): synchronization %scomplete", self.stage, time_msg)
         return self
 
     def __exit__(self, exc_type, exc_val, exc_tb):

cloudinit/sources/DataSourceAzure.py

@@ -19,7 +19,7 @@
 
 import requests
 
-from cloudinit import net, sources, ssh_util, subp, util
+from cloudinit import net, performance, sources, ssh_util, subp, util
 from cloudinit.event import EventScope, EventType
 from cloudinit.net import device_driver
 from cloudinit.net.dhcp import (
@@ -1995,7 +1995,7 @@ def load_azure_ds_dir(source_dir):
     if not os.path.isfile(ovf_file):
         raise NonAzureDataSource("No ovf-env file found")
 
-    with open(ovf_file, "rb") as fp:
+    with performance.Timed("Reading ovf-env.xml"), open(ovf_file, "rb") as fp:
         contents = fp.read()
 
     md, ud, cfg = read_azure_ovf(contents)

cloudinit/sources/DataSourceVMware.py

@@ -267,9 +267,10 @@ def get_instance_id(self):
         # read the file /sys/class/dmi/id/product_uuid for the instance ID.
         if self.metadata and "instance-id" in self.metadata:
             return self.metadata["instance-id"]
-        with open(PRODUCT_UUID_FILE_PATH, "r") as id_file:
-            self.metadata["instance-id"] = str(id_file.read()).rstrip().lower()
-            return self.metadata["instance-id"]
+        self.metadata["instance-id"] = (
+            util.load_text_file(PRODUCT_UUID_FILE_PATH).rstrip().lower()
+        )
+        return self.metadata["instance-id"]
 
     def check_if_fallback_is_allowed(self):
         if (