(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.log.debug("Running legacy Kafka Consumer check.")
        self._zk_consumer_offsets = {}  # Expected format: {(consumer_group, topic, partition): offset}
        self._kafka_consumer_offsets = {}  # Expected format: {(consumer_group, topic, partition): offset}
        self._highwater_offsets = {}  # Expected format: {(topic, partition): offset}
        contexts_limit = self._context_limit

        # 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 consumer group offsets from Zookeeper
        if self._zk_hosts_ports is not None:
            try:
                self._get_zk_consumer_offsets(contexts_limit)
                contexts_limit -= len(self._zk_consumer_offsets)
            except Exception:
                self.log.exception("There was a problem collecting consumer offsets from Zookeeper.")
                # don't raise because we might get valid broker offsets

        # Fetch consumer group offsets from Kafka
        # Support for storing offsets in Kafka not available until Kafka 0.8.2. Also, for legacy reasons, this check
        # only fetches consumer offsets from Kafka if Zookeeper is omitted or kafka_consumer_offsets is True.
        if self.kafka_client.config.get('api_version') >= (0, 8, 2) and is_affirmative(
            self.instance.get('kafka_consumer_offsets', self._zk_hosts_ports is None)
        ):
            try:
                self.log.debug('Collecting consumer offsets')
                self._get_kafka_consumer_offsets(contexts_limit)
                contexts_limit -= len(self._kafka_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
        else:
            self.log.debug(
                'Identified api_version: %s, kafka_consumer_offsets: %s, zk_connection_string: %s.'
                ' Skipping consumer offset collection',
                str(self.kafka_client.config.get('api_version')),
                str(self.instance.get('kafka_consumer_offsets')),
                str(self._zk_hosts_ports),
            )

        # Fetch the broker highwater offsets
        try:
            self._get_highwater_offsets(contexts_limit)
        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 = sum(
            [len(self._zk_consumer_offsets), len(self._kafka_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._report_consumer_offsets_and_lag(self._kafka_consumer_offsets)
        # if someone is in the middle of migrating their offset storage from zookeeper to kafka,
        # they need to identify which source is reporting which offsets. So we tag zookeeper with 'source:zk'
        self._report_consumer_offsets_and_lag(self._zk_consumer_offsets, source='zk')
def check ( self ) : self . log . debug ( "Running legacy Kafka Consumer check." ) self . _zk_consumer_offsets = { } # Expected format: {(consumer_group, topic, partition): offset} self . _kafka_consumer_offsets = { } # Expected format: {(consumer_group, topic, partition): offset} self . _highwater_offsets = { } # Expected format: {(topic, partition): offset} contexts_limit = self . _context_limit # 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 consumer group offsets from Zookeeper if self . _zk_hosts_ports is not None : try : self . _get_zk_consumer_offsets ( contexts_limit ) contexts_limit -= len ( self . _zk_consumer_offsets ) except Exception : self . log . exception ( "There was a problem collecting consumer offsets from Zookeeper." ) # don't raise because we might get valid broker offsets # Fetch consumer group offsets from Kafka # Support for storing offsets in Kafka not available until Kafka 0.8.2. Also, for legacy reasons, this check # only fetches consumer offsets from Kafka if Zookeeper is omitted or kafka_consumer_offsets is True. if self . kafka_client . config . get ( 'api_version' ) >= ( 0 , 8 , 2 ) and is_affirmative ( self . instance . get ( 'kafka_consumer_offsets' , self . _zk_hosts_ports is None ) ) : try : self . log . debug ( 'Collecting consumer offsets' ) self . _get_kafka_consumer_offsets ( contexts_limit ) contexts_limit -= len ( self . _kafka_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 else : self . log . debug ( 'Identified api_version: %s, kafka_consumer_offsets: %s, zk_connection_string: %s.' ' Skipping consumer offset collection' , str ( self . kafka_client . config . get ( 'api_version' ) ) , str ( self . instance . get ( 'kafka_consumer_offsets' ) ) , str ( self . _zk_hosts_ports ) , ) # Fetch the broker highwater offsets try : self . _get_highwater_offsets ( contexts_limit ) 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 = sum ( [ len ( self . _zk_consumer_offsets ) , len ( self . _kafka_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 . _report_consumer_offsets_and_lag ( self . _kafka_consumer_offsets ) # if someone is in the middle of migrating their offset storage from zookeeper to kafka, # they need to identify which source is reporting which offsets. So we tag zookeeper with 'source:zk' self . _report_consumer_offsets_and_lag ( self . _zk_consumer_offsets , source = 'zk' )