Merge pull request #437 from dulek/fix-metrics-cpu

Bug 1920481: Decrease CPU usage of Prometheus exporter

kuryr_kubernetes/cni/daemon/service.py

@@ -17,6 +17,7 @@
 from http import client as httplib
 import multiprocessing
 import os
+import queue
 import socket
 import sys
 import threading
@@ -79,18 +80,8 @@ def _prepare_request(self):
 
     def _update_metrics(self, command, error, duration):
         """Add a new metric value to the shared metrics dict"""
-        params = {}
-        try:
-            params = self._prepare_request()
-        except Exception:
-            LOG.exception('Exception when reading CNI params.')
-            return
-        namespace = params.args.K8S_POD_NAMESPACE
-        name = params.args.K8S_POD_NAME
-        name = f'export-{namespace}/{name}'
         labels = {'command': command, 'error': error}
-        with lockutils.lock(name):
-            self.metrics[name] = {'labels': labels, 'duration': duration}
+        self.metrics.put({'labels': labels, 'duration': duration})
 
     def _error(self, error_code, message, details=""):
         template = {
@@ -359,13 +350,13 @@ def __init__(self, worker_id, metrics):
 
     def _start_metric_updater(self):
         while self.is_running:
-            if self.metrics:
-                pod_name = list(self.metrics.keys())[0]
-                with lockutils.lock(pod_name):
-                    labels = self.metrics[pod_name]['labels']
-                    duration = self.metrics[pod_name]['duration']
-                    self.prometheus_exporter.update_metric(labels, duration)
-                    del self.metrics[pod_name]
+            try:
+                metric = self.metrics.get(timeout=1)
+            except queue.Empty:
+                continue
+            labels = metric['labels']
+            duration = metric['duration']
+            self.prometheus_exporter.update_metric(labels, duration)
 
     def terminate(self):
         self.is_running = False
@@ -393,7 +384,7 @@ def __init__(self):
         self.manager = multiprocessing.Manager()
         registry = self.manager.dict()  # For Watcher->Server communication.
         healthy = multiprocessing.Value(c_bool, True)
-        metrics = self.manager.dict()
+        metrics = self.manager.Queue()
         self.add(CNIDaemonWatcherService, workers=1, args=(registry, healthy,))
         self.add(CNIDaemonServerService, workers=1, args=(registry, healthy,
                                                           metrics,))

kuryr_kubernetes/tests/unit/cni/test_service.py

@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import queue
 from unittest import mock
 
 from oslo_serialization import jsonutils
@@ -31,7 +32,7 @@ def setUp(self):
         self.k8s_mock = self.useFixture(kuryr_fixtures.MockK8sClient())
         self.plugin = k8s_cni_registry.K8sCNIRegistryPlugin({}, healthy)
         self.health_registry = mock.Mock()
-        self.metrics = dict()
+        self.metrics = queue.Queue()
         self.srv = service.DaemonServer(
             self.plugin, self.health_registry, self.metrics)