(self, setup_db, dst_db)
Before locking anything check if consumer is working ok.
Before locking anything check if consumer is working ok.
Before locking anything check if consumer is working ok .
def check_consumer(self, setup_db, dst_db):
        """Before locking anything check if consumer is working ok."""

        setup_curs = setup_db.cursor()
        dst_curs = dst_db.cursor()
        c = 0
        while 1:
            q = "select * from pgq_node.get_consumer_state(%s, %s)"
            res = self.exec_cmd(dst_db, q, [self.queue_name, self.consumer_name])
            completed_tick = res[0]['completed_tick']

            q = "select extract(epoch from ticker_lag) from pgq.get_queue_info(%s)"
            setup_curs.execute(q, [self.queue_name])
            ticker_lag = setup_curs.fetchone()[0]

            q = "select extract(epoch from (now() - t.tick_time)) as lag"\
                " from pgq.tick t, pgq.queue q"\
                " where q.queue_name = %s"\
                "   and t.tick_queue = q.queue_id"\
                "   and t.tick_id = %s"
            setup_curs.execute(q, [self.queue_name, completed_tick])
            res = setup_curs.fetchall()

            if len(res) == 0:
                self.log.warning('Consumer completed_tick (%d) to not exists on provider (%s), too big lag?',
                                 completed_tick, self.provider_info['node_name'])
                self.sleep(10)
                continue

            consumer_lag = res[0][0]
            if consumer_lag < ticker_lag + 5:
                break

            lag_msg = 'Consumer lag: %s, ticker_lag %s, too big difference, waiting'
            if c % 30 == 0:
                self.log.warning(lag_msg, consumer_lag, ticker_lag)
            else:
                self.log.debug(lag_msg, consumer_lag, ticker_lag)
            c += 1
            time.sleep(1)
def check_consumer ( self , setup_db , dst_db ) : setup_curs = setup_db . cursor ( ) dst_curs = dst_db . cursor ( ) c = 0 while 1 : q = "select * from pgq_node.get_consumer_state(%s, %s)" res = self . exec_cmd ( dst_db , q , [ self . queue_name , self . consumer_name ] ) completed_tick = res [ 0 ] [ 'completed_tick' ] q = "select extract(epoch from ticker_lag) from pgq.get_queue_info(%s)" setup_curs . execute ( q , [ self . queue_name ] ) ticker_lag = setup_curs . fetchone ( ) [ 0 ] q = "select extract(epoch from (now() - t.tick_time)) as lag" " from pgq.tick t, pgq.queue q" " where q.queue_name = %s" "   and t.tick_queue = q.queue_id" "   and t.tick_id = %s" setup_curs . execute ( q , [ self . queue_name , completed_tick ] ) res = setup_curs . fetchall ( ) if len ( res ) == 0 : self . log . warning ( 'Consumer completed_tick (%d) to not exists on provider (%s), too big lag?' , completed_tick , self . provider_info [ 'node_name' ] ) self . sleep ( 10 ) continue consumer_lag = res [ 0 ] [ 0 ] if consumer_lag < ticker_lag + 5 : break lag_msg = 'Consumer lag: %s, ticker_lag %s, too big difference, waiting' if c % 30 == 0 : self . log . warning ( lag_msg , consumer_lag , ticker_lag ) else : self . log . debug ( lag_msg , consumer_lag , ticker_lag ) c += 1 time . sleep ( 1 )