def get_metrics()

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

• 17 lines of code
• 4 McCabe index (conditional complexity)

    def get_metrics(self):
        """
        Get a list of metrics from the specified metrics ports
        Returns:
            metrics(list): A list of metrics:
            [{'name': metric_name, 'value': metric_value, 'labels': metric_labels, 'type': metric_type}]
        """
        metrics = []
        metrics_port_list = self.metrics_ports.split(',')
        for port in metrics_port_list:
            metrics_url = f"http://localhost:{port}/metrics"
            response = urllib.request.urlopen(metrics_url)
            content = response.read().decode('utf-8')
            metric_families = text_string_to_metric_families(content)
            for metric_family in metric_families:
                metric_type = metric_family.type
                for sample in metric_family.samples:
                    metric_name = sample.name
                    metric_value = sample.value
                    metric_labels = sample.labels
                    metrics.append({'name': metric_name, 'value': metric_value,
                                    'labels': metric_labels, 'type': metric_type})
        return metrics