feat: replace local CosTool fork with cosl implementation

[Copilot]

Issue

v1/loki_push_api.py maintained a local fork of CosTool that diverged from the upstream cosl implementation, contributing to bugs from the deviation. AlertRules had already been migrated to cosl.rules in a prior change; CosTool remained local.

Solution

• Import CosTool from cosl instead of defining it locally: from cosl import CosTool, JujuTopology
• Update instantiation in LokiPushApiProvider.__init__: CosTool("logql") — cosl's CosTool takes a default_query_type instead of a charm reference; "logql" is correct for all Loki query expressions
• Remove the local CosTool class (~120 lines of duplicated/drifted code)
• Remove orphaned helpers _is_official_alert_rule_format and _is_single_alert_rule_format — unused in v1 since AlertRules was moved to cosl
• Drop subprocess and tempfile imports — only needed by the removed local class
• Bump LIBPATCH 23 → 25
• Pin cosl>=1.9.1 in pyproject.toml; regenerate uv.lock
• Guard app-level relation data writes in LokiPushApiProvider.alerts with a leader check — only the leader may write to the app data bag; non-leaders attempting this write would fail with "permission denied"
• Guard _has_alert_rule_errors() in charm.py with an early if not self.unit.is_leader(): return False — non-leader units running non-relation hooks (e.g. pebble-ready) cannot read their own app's relation data and would crash with "permission denied"

InvalidAlertRulePathError and _resolve_dir_against_charm_path are retained — they are charm-specific (resolve paths against charm.charm_dir) with no cosl equivalent.

Per the issue guidance, v0/loki_push_api.py is unchanged.

Context

The cosl CosTool differs from the previous local fork in two key ways:

• Constructor: takes default_query_type: Optional[QueryType] instead of a charm object; query type flows to all method calls automatically
• Uses platform.machine() (correct) instead of platform.processor() (unreliable in containers/CI) to locate the cos-tool binary

In Juju, non-leader units running non-relation hooks (like pebble-ready) get "permission denied" when calling relation-get --app to access their own application's relation data bag. Any read or write of relation.data[self.app] outside a relation hook must be guarded by self.unit.is_leader().

Testing Instructions

tox -e unit

Note: tests that validate topology label injection into LogQL expressions (test_rules_have_correct_labels) require the cos-tool binary, which tox -e unit downloads automatically via curl. These tests pass in CI; they fail locally without the binary — this is pre-existing behaviour unrelated to this change (those tests target v0, which is unchanged).

Upgrade Notes

Users of LokiPushApiProvider from charms.loki_k8s.v1.loki_push_api should ensure cosl>=1.9.1 is available in their charm's environment. No API-level changes to the public interface.

[lucabello]

@copilot Can you figure out why those integration tests are failing and provide a fix?

[Copilot]

@copilot Can you figure out why those integration tests are failing and provide a fix?

The failures were caused by two related bugs introduced in the commit that added _has_alert_rule_errors:

1. _has_alert_rule_errors in charm.py: Accessing relation.data[self.app] (our own app's relation data bag) fails with "permission denied" for non-leader units when running outside a relation hook (e.g. pebble-ready). Fixed by returning False early when not self.unit.is_leader().

2. LokiPushApiProvider.alerts in v1/loki_push_api.py: The write relation.data[self._charm.app]["event"] = ... (error reporting) also has no leader guard. If cos-tool were present and a validation error occurred, this would similarly crash on non-leaders. Fixed by guarding the write with if self._charm.unit.is_leader().

Commit: e554754

[lucabello]

Diff: CosTool class — loki-k8s-operator (main) vs cos-lib (main)

Left (-): lib/charms/loki_k8s/v0/loki_push_api.py in canonical/loki-k8s-operator
Right (+): src/cosl/cos_tool.py in canonical/cos-lib

--- costool_loki.py	2026-05-07 14:48:03.287869146 +0200
+++ costool_coslib.py	2026-05-07 14:48:03.292869183 +0200
@@ -1,11 +1,18 @@
 class CosTool:
-    """Uses cos-tool to inject label matchers into alert rule expressions and validate rules."""
+    """Uses cos-tool to inject label matchers into alert rule expressions and validate rules.
+
+    Args:
+        default_query_type: an optional querytype to use for all invocations of this class, if
+          not specified per-method. Either :default_query_type: or per-method :query_type:
+          **must** be used, or a :TypeError: will be raised.
+    """
 
     _path = None
     _disabled = False
+    query_type: Union[QueryType, None] = None
 
-    def __init__(self, charm):
-        self._charm = charm
+    def __init__(self, default_query_type: Optional[QueryType] = None):
+        self.query_type = default_query_type
 
     @property
     def path(self):
@@ -19,16 +26,21 @@
                 self._disabled = True
         return self._path
 
-    def apply_label_matchers(self, rules) -> dict:
+    @ensure_querytype
+    def apply_label_matchers(
+        self, rules: OfficialRuleFileFormat, query_type: Optional[QueryType] = None
+    ) -> OfficialRuleFileFormat:
         """Will apply label matchers to the expression of all alerts in all supplied groups."""
+        query_type = query_type or self.query_type
         if not self.path:
             return rules
-        for group in rules["groups"]:
+        for group in rules.get("groups", []):
             rules_in_group = group.get("rules", [])
             for rule in rules_in_group:
                 topology = {}
                 # if the user for some reason has provided juju_unit, we'll need to honor it
                 # in most cases, however, this will be empty
+                labels = rule.get("labels", {})
                 for label in [
                     "juju_model",
                     "juju_model_uuid",
@@ -36,14 +48,18 @@
                     "juju_charm",
                     "juju_unit",
                 ]:
-                    if label in rule["labels"]:
-                        topology[label] = rule["labels"][label]
+                    if label in labels:
+                        topology[label] = labels[label]
 
-                rule["expr"] = self.inject_label_matchers(rule["expr"], topology)
+                rule["expr"] = self.inject_label_matchers(rule["expr"], topology, query_type)  # type: ignore
         return rules
 
-    def validate_alert_rules(self, rules: dict) -> Tuple[bool, str]:
+    @ensure_querytype
+    def validate_alert_rules(
+        self, rules: OfficialRuleFileFormat, query_type: Optional[QueryType] = None
+    ) -> Tuple[bool, str]:
         """Will validate correctness of alert rules, returning a boolean and any errors."""
+        query_type = query_type or self.query_type
         if not self.path:
             logger.debug("`cos-tool` unavailable. Not validating alert correctness.")
             return True, ""
@@ -51,7 +67,8 @@
         with tempfile.TemporaryDirectory() as tmpdir:
             rule_path = Path(tmpdir + "/validate_rule.yaml")
 
-            # Smash "our" rules format into what upstream actually uses, which is more like:
+            # Smash "our" rules format into what upstream actually uses for Loki,
+            # which is more like:
             #
             # groups:
             #   - name: foo
@@ -60,43 +77,74 @@
             #         expr: up
             #       - alert: OtherAlert
             #         expr: up
-            transformed_rules = {"groups": []}  # type: ignore
-            for rule in rules["groups"]:
-                transformed_rules["groups"].append(rule)
+            if query_type == "logql":
+                transformed_rules = OfficialRuleFileFormat(groups=[])
+                for rule in rules.get("groups", []):
+                    transformed_rules.get("groups", []).append(rule)
+
+                rules = transformed_rules
 
-            rule_path.write_text(yaml.dump(transformed_rules))
-            args = [str(self.path), "--format", "logql", "validate", str(rule_path)]
+            rule_path.write_text(yaml.dump(rules))
+
+            args = [str(self.path), "--format", query_type, "validate", str(rule_path)]
             # noinspection PyBroadException
             try:
-                self._exec(args)
+                self._exec(args)  # type: ignore
                 return True, ""
             except subprocess.CalledProcessError as e:
-                logger.debug("Validating the rules failed: %s", e.output)
-                return False, ", ".join([line for line in e.output if "error validating" in line])
-
-    def inject_label_matchers(self, expression, topology) -> str:
+                logger.debug("Validating the rules failed: %s", e.output.decode("utf-8"))
+                return False, ", ".join(
+                    [
+                        line
+                        for line in e.output.decode("utf-8").splitlines()
+                        if "error validating" in line
+                    ]
+                )
+
+    @ensure_querytype
+    def inject_label_matchers(
+        self,
+        expression: str,
+        topology: Dict[str, str],
+        query_type: Optional[QueryType] = None,
+        dashboard_variable: Optional[bool] = False,
+    ) -> str:
         """Add label matchers to an expression."""
+        query_type = query_type or self.query_type
+
         if not topology:
             return expression
         if not self.path:
             logger.debug("`cos-tool` unavailable. Leaving expression unchanged: %s", expression)
             return expression
-        args = [str(self.path), "--format", "logql", "transform"]
+        args = [str(self.path), "--format", query_type, "transform"]
+
+        value_tmpl = r"${}" if dashboard_variable else "{}"
+
+        variable_topology = {k: value_tmpl.format(topology[k]) for k in topology.keys()}
         args.extend(
-            ["--label-matcher={}={}".format(key, value) for key, value in topology.items()]
+            [
+                "--label-matcher={}={}".format(key, value)
+                for key, value in variable_topology.items()
+            ]
         )
 
-        args.extend(["{}".format(expression)])
+        # Pass a leading "--" so expressions with a negation or subtraction aren't interpreted as
+        # flags
+        args.extend(["--", "{}".format(expression)])
         # noinspection PyBroadException
         try:
-            return self._exec(args)
+            return (
+                re.sub(r'="\$juju', r'=~"$juju', self._exec(args))  # type: ignore
+                if dashboard_variable
+                else self._exec(args)  # type: ignore
+            )
         except subprocess.CalledProcessError as e:
             logger.debug('Applying the expression failed: "%s", falling back to the original', e)
-            print('Applying the expression failed: "{}", falling back to the original'.format(e))
             return expression
 
     def _get_tool_path(self) -> Optional[Path]:
-        arch = platform.processor()
+        arch = platform.machine()
         arch = "amd64" if arch == "x86_64" else arch
         res = "cos-tool-{}".format(arch)
         try:
@@ -106,7 +154,6 @@
             logger.debug('Could not locate cos-tool at: "{}"'.format(res))
         return None
 
-    def _exec(self, cmd) -> str:
-        result = subprocess.run(cmd, check=True, stdout=subprocess.PIPE)
-        output = result.stdout.decode("utf-8").strip()
-        return output
+    def _exec(self, cmd: List[str]) -> str:
+        result = subprocess.run(cmd, check=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+        return result.stdout.decode("utf-8").strip()

Key differences summary:

• cos-lib is generic (supports promql/logql via query_type param); loki-k8s-operator hardcodes "logql"
• cos-lib uses @ensure_querytype decorator for runtime validation
• cos-lib uses typed signatures (OfficialRuleFileFormat, Dict[str, str], etc.)
• cos-lib adds dashboard_variable support in inject_label_matchers
• cos-lib uses platform.machine() (correct); loki uses platform.processor() (can return empty string)
• cos-lib captures stderr in _exec; loki does not
• cos-lib properly decodes error output (e.output.decode("utf-8").splitlines()); loki iterates raw bytes
• cos-lib passes "--" separator before expressions to avoid flag misinterpretation
• cos-lib removes print() call (uses only logger.debug)
• cos-lib constructor takes default_query_type instead of charm