def __init__()

in luigi/contrib/prometheus_metric.py [0:0]

• 39 lines of code
• 3 McCabe index (conditional complexity)

    def __init__(self, *args, **kwargs):
        super(PrometheusMetricsCollector, self).__init__()
        self.registry = CollectorRegistry()
        config = prometheus(**kwargs)
        self.labels = list(config.task_parameters_to_use_in_labels)
        if config.use_task_family_in_labels:
            self.labels += ["family"]
        if not self.labels:
            raise ValueError("Prometheus labels cannot be empty (see prometheus configuration)")
        self.task_started_counter = Counter(
            'luigi_task_started_total',
            'number of started luigi tasks',
            self.labels,
            registry=self.registry
        )
        self.task_failed_counter = Counter(
            'luigi_task_failed_total',
            'number of failed luigi tasks',
            self.labels,
            registry=self.registry
        )
        self.task_disabled_counter = Counter(
            'luigi_task_disabled_total',
            'number of disabled luigi tasks',
            self.labels,
            registry=self.registry
        )
        self.task_done_counter = Counter(
            'luigi_task_done_total',
            'number of done luigi tasks',
            self.labels,
            registry=self.registry
        )
        self.task_execution_time = Gauge(
            'luigi_task_execution_time_seconds',
            'luigi task execution time in seconds',
            self.labels,
            registry=self.registry
        )