Add remove_matching() method for metric label deletion

Signed-off-by: Hazel <[email protected]>

Author: Hazel

prometheus_client/metrics.py

@@ -203,6 +203,41 @@ def remove(self, *labelvalues: Any) -> None:
             if labelvalues in self._metrics:
                 del self._metrics[labelvalues]
 
+    def remove_matching(self, partial: dict[str, str]) -> int:
+        if 'prometheus_multiproc_dir' in os.environ or 'PROMETHEUS_MULTIPROC_DIR' in os.environ:
+            warnings.warn(
+                "Removal of labels has not been implemented in  multi-process mode yet.",
+                UserWarning
+            )
+
+        if not self._labelnames:
+            raise ValueError('No label names were set when constructing %s' % self)
+        
+        if not isinstance(partial, dict):
+            raise TypeError("partial must be a dict of {label_name: label_value}")
+    
+        if not partial:
+            return 0
+    
+        invalid = [k for k in partial.keys() if k not in self._labelnames]
+        if invalid:
+            raise ValueError(
+                'Unknown label names: %s; expected %s' % (invalid, self._labelnames)
+            )
+        
+        pos_filter = {self._labelnames.index(k): str(v) for k, v in partial.items()}
+
+        deleted = 0
+        with self._lock:
+            for lv in list(self._metrics.keys()):
+                if all(lv[pos] == want for pos, want in pos_filter.items()):
+                    if lv in self._metrics:
+                        del self._metrics[lv]
+                        deleted += 1
+        
+        return deleted
+        
+
     def clear(self) -> None:
         """Remove all labelsets from the metric"""
         if 'prometheus_multiproc_dir' in os.environ or 'PROMETHEUS_MULTIPROC_DIR' in os.environ:

tests/test_core.py

@@ -630,6 +630,36 @@ def test_labels_coerced_to_string(self):
         self.counter.remove(None)
         self.assertEqual(None, self.registry.get_sample_value('c_total', {'l': 'None'}))
 
+    def test_remove_matching(self):
+        from prometheus_client import Counter
+
+        c = Counter('c2', 'help', ['tenant', 'endpoint'], registry=self.registry)
+        c.labels('acme', '/').inc()
+        c.labels('acme', '/checkout').inc()
+        c.labels('globex', '/').inc()
+
+ 
+        deleted = c.remove_matching({'tenant': 'acme'})
+        self.assertEqual(2, deleted)
+
+        self.assertIsNone(self.registry.get_sample_value('c2_total', {'tenant': 'acme', 'endpoint': '/'}))
+        self.assertIsNone(self.registry.get_sample_value('c2_total', {'tenant': 'acme', 'endpoint': '/checkout'}))
+        self.assertEqual(1, self.registry.get_sample_value('c2_total', {'tenant': 'globex', 'endpoint': '/'}))
+
+    def test_remove_matching_invalid_label_name(self):
+        from prometheus_client import Counter
+        c = Counter('c3', 'help', ['tenant', 'endpoint'], registry=self.registry)
+        c.labels('acme', '/').inc()
+        with self.assertRaises(ValueError):
+            c.remove_matching({'badkey': 'x'})
+
+    def test_remove_matching_empty_is_noop(self):
+        from prometheus_client import Counter
+        c = Counter('c4', 'help', ['tenant', 'endpoint'], registry=self.registry)
+        c.labels('acme', '/').inc()
+        self.assertEqual(0, c.remove_matching({}))
+        self.assertEqual(1, self.registry.get_sample_value('c4_total', {'tenant': 'acme', 'endpoint': '/'}))
+
     def test_non_string_labels_raises(self):
         class Test:
             __str__ = None