def publish_counter()

in src/worker/publisher/metrics_publisher.py [0:0]

• 16 lines of code
• 5 McCabe index (conditional complexity)

    def publish_counter(self, metric):
        """
        Publish counter type metrics to Geneva
        Args:
            metric(dict): A metric dictionary:
            {'name': metric_name, 'value': metric_value, 'labels': metric_labels, 'type': metric_type}
        """
        metric_value = metric['value'] if metric['value'] is not None else 0
        key = self.get_key_from_metric(metric)
        if key not in self.metricKeytoPreviousValue:
            self.metricKeytoPreviousValue[key] = metric_value
            return
        previous_value = self.metricKeytoPreviousValue[key]
        self.metricKeytoPreviousValue[key] = metric_value
        delta = metric_value - previous_value
        # If the delta is negative, it means the counter has been reset
        if delta < 0:
            return
        if metric['name'] not in self.metricNametoCounter:
            self.metricNametoCounter[metric['name']] = self.meter.create_counter(metric['name'])
        counter = self.metricNametoCounter[metric['name']]
        tags = self.get_tags_from_metric(metric['labels'])
        counter.add(metric_value, tags)