feat: improve logging on snap error

lib/charms/operator_libs_linux/v2/snap.py

@@ -87,7 +87,7 @@
 
 if typing.TYPE_CHECKING:
     # avoid typing_extensions import at runtime
-    from typing_extensions import NotRequired, ParamSpec, Required, TypeAlias, Unpack
+    from typing_extensions import NotRequired, ParamSpec, Required, Self, TypeAlias, Unpack
 
     _P = ParamSpec("_P")
     _T = TypeVar("_T")
@@ -102,7 +102,7 @@
 
 # Increment this PATCH version before using `charmcraft publish-lib` or reset
 # to 0 if you are raising the major API version
-LIBPATCH = 11
+LIBPATCH = 12
 
 
 # Regex to locate 7-bit C1 ANSI sequences
@@ -268,6 +268,22 @@ class SnapState(Enum):
 class SnapError(Error):
     """Raised when there's an error running snap control commands."""
 
+    @classmethod
+    def _from_called_process_error(cls, msg: str, error: CalledProcessError) -> Self:
+        lines = [msg]
+        if error.stdout:
+            lines.extend(['Stdout:', error.stdout])
+        if error.stderr:
+            lines.extend(['Stderr:', error.stderr])
+        try:
+            cmd = ['journalctl', '--unit', 'snapd', '--lines', '20']
+            logs = subprocess.check_output(cmd, text=True)
+        except Exception as e:
+            lines.extend(['Error fetching logs:', str(e)])
+        else:
+            lines.extend(['Latest logs:', logs])
+        return cls('\n'.join(lines))
+
 
 class SnapNotFoundError(Error):
     """Raised when a requested snap is not known to the system."""
@@ -340,11 +356,10 @@ def _snap(self, command: str, optargs: Iterable[str] | None = None) -> str:
         optargs = optargs or []
         args = ["snap", command, self._name, *optargs]
         try:
-            return subprocess.check_output(args, text=True)
+            return subprocess.check_output(args, text=True, stderr=subprocess.PIPE)
         except CalledProcessError as e:
-            raise SnapError(
-                f"Snap: {self._name!r}; command {args!r} failed with output = {e.output!r}"
-            ) from e
+            msg = f'Snap: {self._name!r} -- command {args!r} failed!'
+            raise SnapError._from_called_process_error(msg=msg, error=e) from e
 
     def _snap_daemons(
         self,
@@ -371,7 +386,8 @@ def _snap_daemons(
         try:
             return subprocess.run(args, text=True, check=True, capture_output=True)
         except CalledProcessError as e:
-            raise SnapError(f"Could not {args} for snap [{self._name}]: {e.stderr}") from e
+            msg = f'Snap: {self._name!r} -- command {args!r} failed!'
+            raise SnapError._from_called_process_error(msg=msg, error=e) from e
 
     @typing.overload
     def get(self, key: None | Literal[""], *, typed: Literal[False] = False) -> NoReturn: ...
@@ -477,7 +493,8 @@ def connect(self, plug: str, service: str | None = None, slot: str | None = None
         try:
             subprocess.run(args, text=True, check=True, capture_output=True)
         except CalledProcessError as e:
-            raise SnapError(f"Could not {args} for snap [{self._name}]: {e.stderr}") from e
+            msg = f'Snap: {self._name!r} -- command {args!r} failed!'
+            raise SnapError._from_called_process_error(msg=msg, error=e) from e
 
     def hold(self, duration: timedelta | None = None) -> None:
         """Add a refresh hold to a snap.
@@ -506,11 +523,10 @@ def alias(self, application: str, alias: str | None = None) -> None:
             alias = application
         args = ["snap", "alias", f"{self.name}.{application}", alias]
         try:
-            subprocess.check_output(args, text=True)
+            subprocess.run(args, text=True, check=True, capture_output=True)
         except CalledProcessError as e:
-            raise SnapError(
-                f"Snap: {self._name!r}; command {args!r} failed with output = {e.output!r}"
-            ) from e
+            msg = f'Snap: {self._name!r} -- command {args!r} failed!'
+            raise SnapError._from_called_process_error(msg=msg, error=e) from e
 
     def restart(self, services: list[str] | None = None, reload: bool = False) -> None:
         """Restarts a snap's services.
@@ -1264,7 +1280,7 @@ def install_local(
     if dangerous:
         args.append("--dangerous")
     try:
-        result = subprocess.check_output(args, text=True).splitlines()[-1]
+        result = subprocess.check_output(args, text=True, stderr=subprocess.PIPE).splitlines()[-1]
         snap_name, _ = result.split(" ", 1)
         snap_name = ansi_filter.sub("", snap_name)
 
@@ -1280,7 +1296,8 @@ def install_local(
             )
             raise SnapError(f"Failed to find snap {snap_name} in Snap cache") from e
     except CalledProcessError as e:
-        raise SnapError(f"Could not install snap {filename}: {e.output}") from e
+        msg = f'Cound not install snap {filename}!'
+        raise SnapError._from_called_process_error(msg=msg, error=e) from e
 
 
 def _system_set(config_item: str, value: str) -> None:
@@ -1292,9 +1309,10 @@ def _system_set(config_item: str, value: str) -> None:
     """
     args = ["snap", "set", "system", f"{config_item}={value}"]
     try:
-        subprocess.check_call(args, text=True)
+        subprocess.run(args, text=True, check=True, capture_output=True)
     except CalledProcessError as e:
-        raise SnapError(f"Failed setting system config '{config_item}' to '{value}'") from e
+        msg = f"Failed setting system config '{config_item}' to '{value}'"
+        raise SnapError._from_called_process_error(msg=msg, error=e) from e
 
 
 def hold_refresh(days: int = 90, forever: bool = False) -> None:

tests/integration/test_snap.py

@@ -167,12 +167,10 @@ def test_unset_key_raises_snap_error():
 
     # Verify that the correct exception gets raised in the case of an unset key.
     key = "keythatdoesntexist01"
-    try:
+    with pytest.raises(snap.SnapError) as ctx:
         lxd.get(key)
-    except snap.SnapError:
-        pass
-    else:
-        logger.error("Getting an unset key should result in a SnapError.")
+    assert key in ctx.value.message
+    assert "\nLatest logs:\n" in ctx.value.message  # journalctl log retrieval on SnapError
 
     # We can make the above work w/ arbitrary config.
     lxd.set({key: "true"})