(self)
Convert out-of-sync old root to branch and sync queue contents.
Convert out-of-sync old root to branch and sync queue contents.
Convert out - of - sync old root to branch and sync queue contents .
def cmd_resurrect(self):
        """Convert out-of-sync old root to branch and sync queue contents.
        """
        self.load_local_info()

        db = self.get_database(self.initial_db_name)
        curs = db.cursor()

        # stop if leaf
        if self.queue_info.local_node.type == 'leaf':
            self.log.info("Current node is leaf, nothing to do")
            return

        # stop if dump file exists
        if os.path.lexists(RESURRECT_DUMP_FILE):
            self.log.error("Dump file exists, cannot perform resurrection: %s", RESURRECT_DUMP_FILE)
            sys.exit(1)

        #
        # Find failover position
        #

        self.log.info("** Searching for gravestone **")

        # load subscribers
        sub_list = []
        q = "select * from pgq_node.get_subscriber_info(%s)"
        curs.execute(q, [self.queue_name])
        for row in curs.fetchall():
            sub_list.append(row['node_name'])
        db.commit()

        # find backup subscription
        this_node = self.queue_info.local_node.name
        failover_cons = self.failover_consumer_name(this_node)
        full_list = self.queue_info.member_map.keys()
        done_nodes = { this_node: 1 }
        prov_node = None
        root_node = None
        for node_name in sub_list + full_list:
            if node_name in done_nodes:
                continue
            done_nodes[node_name] = 1
            if not self.node_alive(node_name):
                self.log.info('Node %s is dead, skipping', node_name)
                continue
            self.log.info('Looking on node %s', node_name)
            node_db = None
            try:
                node_db = self.get_node_database(node_name)
                node_curs = node_db.cursor()
                node_curs.execute("select * from pgq.get_consumer_info(%s, %s)", [self.queue_name, failover_cons])
                cons_rows = node_curs.fetchall()
                node_curs.execute("select * from pgq_node.get_node_info(%s)", [self.queue_name])
                node_info = node_curs.fetchone()
                node_db.commit()
                if len(cons_rows) == 1:
                    if prov_node:
                        raise Exception('Unexpected situation: there are two gravestones - on nodes %s and %s' % (prov_node, node_name))
                    prov_node = node_name
                    failover_tick = cons_rows[0]['last_tick']
                    self.log.info("Found gravestone on node: %s", node_name)
                if node_info['node_type'] == 'root':
                    self.log.info("Found new root node: %s", node_name)
                    root_node = node_name
                self.close_node_database(node_name)
                node_db = None
                if root_node and prov_node:
                    break
            except skytools.DBError:
                self.log.warning("failed to check node %s", node_name)
                if node_db:
                    self.close_node_database(node_name)
                    node_db = None

        if not root_node:
            self.log.error("Cannot find new root node", failover_cons)
            sys.exit(1)
        if not prov_node:
            self.log.error("Cannot find failover position (%s)", failover_cons)
            sys.exit(1)

        # load worker state
        q = "select * from pgq_node.get_worker_state(%s)"
        rows = self.exec_cmd(db, q, [self.queue_name])
        state = rows[0]

        # demote & pause
        self.log.info("** Demote & pause local node **")
        if self.queue_info.local_node.type == 'root':
            self.log.info('Node %s is root, demoting', this_node)
            q = "select * from pgq_node.demote_root(%s, %s, %s)"
            self.exec_cmd(db, q, [self.queue_name, 1, prov_node])
            self.exec_cmd(db, q, [self.queue_name, 2, prov_node])

            # change node type and set worker paused in same TX
            curs = db.cursor()
            self.exec_cmd(curs, q, [self.queue_name, 3, prov_node])
            q = "select * from pgq_node.set_consumer_paused(%s, %s, true)"
            self.exec_cmd(curs, q, [self.queue_name, state['worker_name']])
            db.commit()
        elif not state['paused']:
            # pause worker, don't wait for reaction, as it may be dead
            self.log.info('Node %s is branch, pausing worker: %s', this_node, state['worker_name'])
            q = "select * from pgq_node.set_consumer_paused(%s, %s, true)"
            self.exec_cmd(db, q, [self.queue_name, state['worker_name']])
        else:
            self.log.info('Node %s is branch and worker is paused', this_node)

        #
        # Drop old consumers and subscribers
        #
        self.log.info("** Dropping old subscribers and consumers **")

        # unregister subscriber nodes
        q = "select pgq_node.unregister_subscriber(%s, %s)"
        for node_name in sub_list:
            self.log.info("Dropping old subscriber node: %s", node_name)
            curs.execute(q, [self.queue_name, node_name])

        # unregister consumers
        q = "select consumer_name from pgq.get_consumer_info(%s)"
        curs.execute(q, [self.queue_name])
        for row in curs.fetchall():
            cname = row['consumer_name']
            if cname[0] == '.':
                self.log.info("Keeping consumer: %s", cname)
                continue
            self.log.info("Dropping old consumer: %s", cname)
            q = "pgq.unregister_consumer(%s, %s)"
            curs.execute(q, [self.queue_name, cname])
        db.commit()

        # dump events
        self.log.info("** Dump & delete lost events **")
        stats = self.resurrect_process_lost_events(db, failover_tick)

        self.log.info("** Subscribing %s to %s **", this_node, prov_node)

        # set local position
        self.log.info("Reset local completed pos")
        q = "select * from pgq_node.set_consumer_completed(%s, %s, %s)"
        self.exec_cmd(db, q, [self.queue_name, state['worker_name'], failover_tick])

        # rename gravestone
        self.log.info("Rename gravestone to worker: %s", state['worker_name'])
        prov_db = self.get_node_database(prov_node)
        prov_curs = prov_db.cursor()
        q = "select * from pgq_node.unregister_subscriber(%s, %s)"
        self.exec_cmd(prov_curs, q, [self.queue_name, this_node], quiet = True)
        q = "select ret_code, ret_note, global_watermark"\
            " from pgq_node.register_subscriber(%s, %s, %s, %s)"
        res = self.exec_cmd(prov_curs, q, [self.queue_name, this_node, state['worker_name'], failover_tick], quiet = True)
        global_wm = res[0]['global_watermark']
        prov_db.commit()

        # import new global watermark
        self.log.info("Reset global watermark")
        q = "select * from pgq_node.set_global_watermark(%s, %s)"
        self.exec_cmd(db, q, [self.queue_name, global_wm], quiet = True)

        # show stats
        if stats:
            self.log.info("** Statistics **")
            klist = stats.keys()
            klist.sort()
            for k in klist:
                v = stats[k]
                self.log.info("  %s: %s", k, v)
        self.log.info("** Resurrection done, worker paused **")
def cmd_resurrect ( self ) : self . load_local_info ( ) db = self . get_database ( self . initial_db_name ) curs = db . cursor ( ) # stop if leaf if self . queue_info . local_node . type == 'leaf' : self . log . info ( "Current node is leaf, nothing to do" ) return # stop if dump file exists if os . path . lexists ( RESURRECT_DUMP_FILE ) : self . log . error ( "Dump file exists, cannot perform resurrection: %s" , RESURRECT_DUMP_FILE ) sys . exit ( 1 ) # # Find failover position # self . log . info ( "** Searching for gravestone **" ) # load subscribers sub_list = [ ] q = "select * from pgq_node.get_subscriber_info(%s)" curs . execute ( q , [ self . queue_name ] ) for row in curs . fetchall ( ) : sub_list . append ( row [ 'node_name' ] ) db . commit ( ) # find backup subscription this_node = self . queue_info . local_node . name failover_cons = self . failover_consumer_name ( this_node ) full_list = self . queue_info . member_map . keys ( ) done_nodes = { this_node : 1 } prov_node = None root_node = None for node_name in sub_list + full_list : if node_name in done_nodes : continue done_nodes [ node_name ] = 1 if not self . node_alive ( node_name ) : self . log . info ( 'Node %s is dead, skipping' , node_name ) continue self . log . info ( 'Looking on node %s' , node_name ) node_db = None try : node_db = self . get_node_database ( node_name ) node_curs = node_db . cursor ( ) node_curs . execute ( "select * from pgq.get_consumer_info(%s, %s)" , [ self . queue_name , failover_cons ] ) cons_rows = node_curs . fetchall ( ) node_curs . execute ( "select * from pgq_node.get_node_info(%s)" , [ self . queue_name ] ) node_info = node_curs . fetchone ( ) node_db . commit ( ) if len ( cons_rows ) == 1 : if prov_node : raise Exception ( 'Unexpected situation: there are two gravestones - on nodes %s and %s' % ( prov_node , node_name ) ) prov_node = node_name failover_tick = cons_rows [ 0 ] [ 'last_tick' ] self . log . info ( "Found gravestone on node: %s" , node_name ) if node_info [ 'node_type' ] == 'root' : self . log . info ( "Found new root node: %s" , node_name ) root_node = node_name self . close_node_database ( node_name ) node_db = None if root_node and prov_node : break except skytools . DBError : self . log . warning ( "failed to check node %s" , node_name ) if node_db : self . close_node_database ( node_name ) node_db = None if not root_node : self . log . error ( "Cannot find new root node" , failover_cons ) sys . exit ( 1 ) if not prov_node : self . log . error ( "Cannot find failover position (%s)" , failover_cons ) sys . exit ( 1 ) # load worker state q = "select * from pgq_node.get_worker_state(%s)" rows = self . exec_cmd ( db , q , [ self . queue_name ] ) state = rows [ 0 ] # demote & pause self . log . info ( "** Demote & pause local node **" ) if self . queue_info . local_node . type == 'root' : self . log . info ( 'Node %s is root, demoting' , this_node ) q = "select * from pgq_node.demote_root(%s, %s, %s)" self . exec_cmd ( db , q , [ self . queue_name , 1 , prov_node ] ) self . exec_cmd ( db , q , [ self . queue_name , 2 , prov_node ] ) # change node type and set worker paused in same TX curs = db . cursor ( ) self . exec_cmd ( curs , q , [ self . queue_name , 3 , prov_node ] ) q = "select * from pgq_node.set_consumer_paused(%s, %s, true)" self . exec_cmd ( curs , q , [ self . queue_name , state [ 'worker_name' ] ] ) db . commit ( ) elif not state [ 'paused' ] : # pause worker, don't wait for reaction, as it may be dead self . log . info ( 'Node %s is branch, pausing worker: %s' , this_node , state [ 'worker_name' ] ) q = "select * from pgq_node.set_consumer_paused(%s, %s, true)" self . exec_cmd ( db , q , [ self . queue_name , state [ 'worker_name' ] ] ) else : self . log . info ( 'Node %s is branch and worker is paused' , this_node ) # # Drop old consumers and subscribers # self . log . info ( "** Dropping old subscribers and consumers **" ) # unregister subscriber nodes q = "select pgq_node.unregister_subscriber(%s, %s)" for node_name in sub_list : self . log . info ( "Dropping old subscriber node: %s" , node_name ) curs . execute ( q , [ self . queue_name , node_name ] ) # unregister consumers q = "select consumer_name from pgq.get_consumer_info(%s)" curs . execute ( q , [ self . queue_name ] ) for row in curs . fetchall ( ) : cname = row [ 'consumer_name' ] if cname [ 0 ] == '.' : self . log . info ( "Keeping consumer: %s" , cname ) continue self . log . info ( "Dropping old consumer: %s" , cname ) q = "pgq.unregister_consumer(%s, %s)" curs . execute ( q , [ self . queue_name , cname ] ) db . commit ( ) # dump events self . log . info ( "** Dump & delete lost events **" ) stats = self . resurrect_process_lost_events ( db , failover_tick ) self . log . info ( "** Subscribing %s to %s **" , this_node , prov_node ) # set local position self . log . info ( "Reset local completed pos" ) q = "select * from pgq_node.set_consumer_completed(%s, %s, %s)" self . exec_cmd ( db , q , [ self . queue_name , state [ 'worker_name' ] , failover_tick ] ) # rename gravestone self . log . info ( "Rename gravestone to worker: %s" , state [ 'worker_name' ] ) prov_db = self . get_node_database ( prov_node ) prov_curs = prov_db . cursor ( ) q = "select * from pgq_node.unregister_subscriber(%s, %s)" self . exec_cmd ( prov_curs , q , [ self . queue_name , this_node ] , quiet = True ) q = "select ret_code, ret_note, global_watermark" " from pgq_node.register_subscriber(%s, %s, %s, %s)" res = self . exec_cmd ( prov_curs , q , [ self . queue_name , this_node , state [ 'worker_name' ] , failover_tick ] , quiet = True ) global_wm = res [ 0 ] [ 'global_watermark' ] prov_db . commit ( ) # import new global watermark self . log . info ( "Reset global watermark" ) q = "select * from pgq_node.set_global_watermark(%s, %s)" self . exec_cmd ( db , q , [ self . queue_name , global_wm ] , quiet = True ) # show stats if stats : self . log . info ( "** Statistics **" ) klist = stats . keys ( ) klist . sort ( ) for k in klist : v = stats [ k ] self . log . info ( "  %s: %s" , k , v ) self . log . info ( "** Resurrection done, worker paused **" )