(self)
The main entrypoint of the check.
The main entrypoint of the check.
The main entrypoint of the check .
def check(self):
        """The main entrypoint of the check."""
        self._consumer_offsets = {}  # Expected format: {(consumer_group, topic, partition): offset}
        self._highwater_offsets = {}  # Expected format: {(topic, partition): offset}

        # For calculating consumer lag, we have to fetch both the consumer offset and the broker highwater offset.
        # There's a potential race condition because whichever one we check first may be outdated by the time we check
        # the other. Better to check consumer offsets before checking broker offsets because worst case is that
        # overstates consumer lag a little. Doing it the other way can understate consumer lag to the point of having
        # negative consumer lag, which just creates confusion because it's theoretically impossible.

        # Fetch Kafka consumer offsets
        try:
            self._get_consumer_offsets()
        except Exception:
            self.log.exception("There was a problem collecting consumer offsets from Kafka.")
            # don't raise because we might get valid broker offsets

        # Fetch the broker highwater offsets
        try:
            if len(self._consumer_offsets) < self._context_limit:
                self._get_highwater_offsets()
            else:
                self.warning("Context limit reached. Skipping highwater offset collection.")
        except Exception:
            self.log.exception("There was a problem collecting the highwater mark offsets.")
            # Unlike consumer offsets, fail immediately because we can't calculate consumer lag w/o highwater_offsets
            raise

        total_contexts = len(self._consumer_offsets) + len(self._highwater_offsets)
        if total_contexts >= self._context_limit:
            self.warning(
                """Discovered %s metric contexts - this exceeds the maximum number of %s contexts permitted by the
                check. Please narrow your target by specifying in your kafka_consumer.yaml the consumer groups, topics
                and partitions you wish to monitor.""",
                total_contexts,
                self._context_limit,
            )

        # Report the metrics
        self._report_highwater_offsets(self._context_limit)
        self._report_consumer_offsets_and_lag(self._context_limit - len(self._highwater_offsets))

        self._collect_broker_metadata()
def check ( self ) : self . _consumer_offsets = { } # Expected format: {(consumer_group, topic, partition): offset} self . _highwater_offsets = { } # Expected format: {(topic, partition): offset} # For calculating consumer lag, we have to fetch both the consumer offset and the broker highwater offset. # There's a potential race condition because whichever one we check first may be outdated by the time we check # the other. Better to check consumer offsets before checking broker offsets because worst case is that # overstates consumer lag a little. Doing it the other way can understate consumer lag to the point of having # negative consumer lag, which just creates confusion because it's theoretically impossible. # Fetch Kafka consumer offsets try : self . _get_consumer_offsets ( ) except Exception : self . log . exception ( "There was a problem collecting consumer offsets from Kafka." ) # don't raise because we might get valid broker offsets # Fetch the broker highwater offsets try : if len ( self . _consumer_offsets ) < self . _context_limit : self . _get_highwater_offsets ( ) else : self . warning ( "Context limit reached. Skipping highwater offset collection." ) except Exception : self . log . exception ( "There was a problem collecting the highwater mark offsets." ) # Unlike consumer offsets, fail immediately because we can't calculate consumer lag w/o highwater_offsets raise total_contexts = len ( self . _consumer_offsets ) + len ( self . _highwater_offsets ) if total_contexts >= self . _context_limit : self . warning ( """Discovered %s metric contexts - this exceeds the maximum number of %s contexts permitted by the
                check. Please narrow your target by specifying in your kafka_consumer.yaml the consumer groups, topics
                and partitions you wish to monitor.""" , total_contexts , self . _context_limit , ) # Report the metrics self . _report_highwater_offsets ( self . _context_limit ) self . _report_consumer_offsets_and_lag ( self . _context_limit - len ( self . _highwater_offsets ) ) self . _collect_broker_metadata ( )