Skip to content
This repository was archived by the owner on Jul 8, 2026. It is now read-only.
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
54 changes: 43 additions & 11 deletions lib/charms/operator_libs_linux/v2/snap.py
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,10 @@
except snap.SnapError as e:
logger.error("An exception occurred when installing snaps. Reason: %s" % e.message)
```

Dependencies:
Note that this module requires `opentelemetry-api`, which is already included into
your charm's virtual environment via `ops >= 2.21`.
"""

from __future__ import annotations
Expand Down Expand Up @@ -85,6 +89,8 @@
TypeVar,
)

import opentelemetry.trace

if typing.TYPE_CHECKING:
# avoid typing_extensions import at runtime
from typing_extensions import NotRequired, ParamSpec, Required, Self, TypeAlias, Unpack
Expand All @@ -93,6 +99,7 @@
_T = TypeVar("_T")

logger = logging.getLogger(__name__)
tracer = opentelemetry.trace.get_tracer(__name__)

# The unique Charmhub library identifier, never change it
LIBID = "05394e5893f94f2d90feb7cbe6b633cd"
Expand All @@ -102,7 +109,9 @@

# Increment this PATCH version before using `charmcraft publish-lib` or reset
# to 0 if you are raising the major API version
LIBPATCH = 12
LIBPATCH = 13

PYDEPS = ["opentelemetry-api"]


# Regex to locate 7-bit C1 ANSI sequences
Expand Down Expand Up @@ -277,7 +286,9 @@ def _from_called_process_error(cls, msg: str, error: CalledProcessError) -> Self
lines.extend(['Stderr:', error.stderr])
try:
cmd = ['journalctl', '--unit', 'snapd', '--lines', '20']
logs = subprocess.check_output(cmd, text=True)
with tracer.start_as_current_span(cmd[0]) as span:
span.set_attribute("argv", cmd)
logs = subprocess.check_output(cmd, text=True)
except Exception as e:
lines.extend(['Error fetching logs:', str(e)])
else:
Expand Down Expand Up @@ -356,7 +367,9 @@ 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, stderr=subprocess.PIPE)
with tracer.start_as_current_span(args[0]) as span:
span.set_attribute("argv", args)
return subprocess.check_output(args, text=True, stderr=subprocess.PIPE)
except CalledProcessError as e:
msg = f'Snap: {self._name!r} -- command {args!r} failed!'
raise SnapError._from_called_process_error(msg=msg, error=e) from e
Expand Down Expand Up @@ -384,7 +397,9 @@ def _snap_daemons(
args = ["snap", *command, *services]

try:
return subprocess.run(args, text=True, check=True, capture_output=True)
with tracer.start_as_current_span(args[0]) as span:
span.set_attribute("argv", args)
return subprocess.run(args, text=True, check=True, capture_output=True)
except CalledProcessError as e:
msg = f'Snap: {self._name!r} -- command {args!r} failed!'
raise SnapError._from_called_process_error(msg=msg, error=e) from e
Expand Down Expand Up @@ -491,7 +506,9 @@ def connect(self, plug: str, service: str | None = None, slot: str | None = None

args = ["snap", *command]
try:
subprocess.run(args, text=True, check=True, capture_output=True)
with tracer.start_as_current_span(args[0]) as span:
span.set_attribute("argv", args)
subprocess.run(args, text=True, check=True, capture_output=True)
except CalledProcessError as e:
msg = f'Snap: {self._name!r} -- command {args!r} failed!'
raise SnapError._from_called_process_error(msg=msg, error=e) from e
Expand Down Expand Up @@ -523,7 +540,9 @@ def alias(self, application: str, alias: str | None = None) -> None:
alias = application
args = ["snap", "alias", f"{self.name}.{application}", alias]
try:
subprocess.run(args, text=True, check=True, capture_output=True)
with tracer.start_as_current_span(args[0]) as span:
span.set_attribute("argv", args)
subprocess.run(args, text=True, check=True, capture_output=True)
except CalledProcessError as e:
msg = f'Snap: {self._name!r} -- command {args!r} failed!'
raise SnapError._from_called_process_error(msg=msg, error=e) from e
Expand Down Expand Up @@ -932,15 +951,20 @@ def _request_raw(

def get_installed_snaps(self) -> list[dict[str, JSONType]]:
"""Get information about currently installed snaps."""
return self._request("GET", "snaps") # type: ignore
with tracer.start_as_current_span("get_installed_snaps"):
return self._request("GET", "snaps") # type: ignore

def get_snap_information(self, name: str) -> dict[str, JSONType]:
"""Query the snap server for information about single snap."""
return self._request("GET", "find", {"name": name})[0] # type: ignore
with tracer.start_as_current_span("get_snap_information") as span:
span.set_attribute("name", name)
return self._request("GET", "find", {"name": name})[0] # type: ignore

def get_installed_snap_apps(self, name: str) -> list[dict[str, JSONType]]:
"""Query the snap server for apps belonging to a named, currently installed snap."""
return self._request("GET", "apps", {"names": name, "select": "service"}) # type: ignore
with tracer.start_as_current_span("get_installed_snap_apps") as span:
span.set_attribute("name", name)
return self._request("GET", "apps", {"names": name, "select": "service"}) # type: ignore

def _put_snap_conf(self, name: str, conf: dict[str, JSONAble]) -> None:
"""Set the configuration details for an installed snap."""
Expand Down Expand Up @@ -1280,7 +1304,13 @@ def install_local(
if dangerous:
args.append("--dangerous")
try:
result = subprocess.check_output(args, text=True, stderr=subprocess.PIPE).splitlines()[-1]
with tracer.start_as_current_span(args[0]) as span:
span.set_attribute("argv", args)
result = subprocess.check_output(
args,
text=True,
stderr=subprocess.PIPE,
).splitlines()[-1]
snap_name, _ = result.split(" ", 1)
snap_name = ansi_filter.sub("", snap_name)

Expand Down Expand Up @@ -1309,7 +1339,9 @@ def _system_set(config_item: str, value: str) -> None:
"""
args = ["snap", "set", "system", f"{config_item}={value}"]
try:
subprocess.run(args, text=True, check=True, capture_output=True)
with tracer.start_as_current_span(args[0]) as span:
span.set_attribute("argv", args)
subprocess.run(args, text=True, check=True, capture_output=True)
except CalledProcessError as e:
msg = f"Failed setting system config '{config_item}' to '{value}'"
raise SnapError._from_called_process_error(msg=msg, error=e) from e
Expand Down
Loading