fix: charmcraft fetch-lib

lib/charms/grafana_agent/v0/cos_agent.py

@@ -211,7 +211,9 @@ def __init__(self, *args):
 ```
 """
 
+import copy
 import enum
+import hashlib
 import json
 import logging
 import socket
@@ -254,7 +256,7 @@ class _MetricsEndpointDict(TypedDict):
 
 LIBID = "dc15fa84cef84ce58155fb84f6c6213a"
 LIBAPI = 0
-LIBPATCH = 22
+LIBPATCH = 25
 
 PYDEPS = ["cosl >= 0.0.50", "pydantic"]
 
@@ -264,12 +266,6 @@ class _MetricsEndpointDict(TypedDict):
 logger = logging.getLogger(__name__)
 SnapEndpoint = namedtuple("SnapEndpoint", "owner, name")
 
-# Note: MutableMapping is imported from the typing module and not collections.abc
-# because subscripting collections.abc.MutableMapping was added in python 3.9, but
-# most of our charms are based on 20.04, which has python 3.8.
-
-_RawDatabag = MutableMapping[str, str]
-
 
 class TransportProtocolType(str, enum.Enum):
     """Receiver Type."""
@@ -305,6 +301,22 @@ class TransportProtocolType(str, enum.Enum):
 ReceiverProtocol = Literal["otlp_grpc", "otlp_http", "zipkin", "jaeger_thrift_http", "jaeger_grpc"]
 
 
+def _dedupe_list(items: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+    """Deduplicate items in the list via object identity."""
+    unique_items = []
+    for item in items:
+        if item not in unique_items:
+            unique_items.append(item)
+    return unique_items
+
+
+def _dict_hash_except_key(scrape_config: Dict[str, Any], key: Optional[str]):
+    """Get a hash of the scrape_config dict, except for the specified key."""
+    cfg_for_hash = {k: v for k, v in scrape_config.items() if k != key}
+    serialized = json.dumps(cfg_for_hash, sort_keys=True)
+    return hashlib.blake2b(serialized.encode(), digest_size=4).hexdigest()
+
+
 class TracingError(Exception):
     """Base class for custom errors raised by tracing."""
 
@@ -619,7 +631,8 @@ def __init__(
         refresh_events: Optional[List] = None,
         tracing_protocols: Optional[List[str]] = None,
         *,
-        scrape_configs: Optional[Union[List[dict], Callable]] = None,
+        scrape_configs: Optional[Union[List[dict], Callable[[], List[Dict[str, Any]]]]] = None,
+        extra_alert_groups: Optional[Callable[[], Dict[str, Any]]] = None,
     ):
         """Create a COSAgentProvider instance.
 
@@ -640,6 +653,9 @@ def __init__(
             scrape_configs: List of standard scrape_configs dicts or a callable
                 that returns the list in case the configs need to be generated dynamically.
                 The contents of this list will be merged with the contents of `metrics_endpoints`.
+            extra_alert_groups: A callable that returns a dict of alert rule groups in case the
+                alerts need to be generated dynamically. The contents of this dict will be merged
+                with generic and bundled alert rules.
         """
         super().__init__(charm, relation_name)
         dashboard_dirs = dashboard_dirs or ["./src/grafana_dashboards"]
@@ -648,6 +664,7 @@ def __init__(
         self._relation_name = relation_name
         self._metrics_endpoints = metrics_endpoints or []
         self._scrape_configs = scrape_configs or []
+        self._extra_alert_groups = extra_alert_groups or {}
         self._metrics_rules = metrics_rules_dir
         self._logs_rules = logs_rules_dir
         self._recursive = recurse_rules_dirs
@@ -689,12 +706,34 @@ def _on_refresh(self, event):
                 ) as e:
                     logger.error("Invalid relation data provided: %s", e)
 
+    def _deterministic_scrape_configs(
+        self, scrape_configs: List[Dict[str, Any]]
+    ) -> List[Dict[str, Any]]:
+        """Get deterministic scrape_configs with stable job names.
+
+        For stability across serializations, compute a short per-config hash
+        and append it to the existing job name (or 'default'). Keep the app
+        name as a prefix: <app>_<job_or_default>_<8hex-hash>.
+
+        Hash the whole scrape_config (except any existing job_name) so the
+        suffix is sensitive to all stable fields. Use deterministic JSON
+        serialization.
+        """
+        local_scrape_configs = copy.deepcopy(scrape_configs)
+        for scrape_config in local_scrape_configs:
+            name = scrape_config.get("job_name", "default")
+            short_id = _dict_hash_except_key(scrape_config, "job_name")
+            scrape_config["job_name"] = f"{self._charm.app.name}_{name}_{short_id}"
+
+        return sorted(local_scrape_configs, key=lambda c: c.get("job_name", ""))
+
     @property
     def _scrape_jobs(self) -> List[Dict]:
-        """Return a prometheus_scrape-like data structure for jobs.
+        """Return a list of scrape_configs.
 
         https://prometheus.io/docs/prometheus/latest/configuration/configuration/#scrape_config
         """
+        # Optionally allow the charm to set the scrape_configs
         if callable(self._scrape_configs):
             scrape_configs = self._scrape_configs()
         else:
@@ -712,26 +751,30 @@ def _scrape_jobs(self) -> List[Dict]:
 
         scrape_configs = scrape_configs or []
 
-        # Augment job name to include the app name and a unique id (index)
-        for idx, scrape_config in enumerate(scrape_configs):
-            scrape_config["job_name"] = "_".join(
-                [self._charm.app.name, str(idx), scrape_config.get("job_name", "default")]
-            )
-
-        return scrape_configs
+        return self._deterministic_scrape_configs(scrape_configs)
 
     @property
     def _metrics_alert_rules(self) -> Dict:
-        """Use (for now) the prometheus_scrape AlertRules to initialize this."""
+        """Return a dict of alert rule groups."""
+        # Optionally allow the charm to add the metrics_alert_rules
+        if callable(self._extra_alert_groups):
+            rules = self._extra_alert_groups()
+        else:
+            rules = {"groups": []}
+
         alert_rules = AlertRules(
             query_type="promql", topology=JujuTopology.from_charm(self._charm)
         )
         alert_rules.add_path(self._metrics_rules, recursive=self._recursive)
         alert_rules.add(
-            generic_alert_groups.application_rules,
+            copy.deepcopy(generic_alert_groups.application_rules),
             group_name_prefix=JujuTopology.from_charm(self._charm).identifier,
         )
-        return alert_rules.as_dict()
+
+        # NOTE: The charm could supply rules we implement in this method, so we deduplicate
+        rules["groups"] = _dedupe_list(rules["groups"] + alert_rules.as_dict()["groups"])
+
+        return rules
 
     @property
     def _log_alert_rules(self) -> Dict: