(self, node_type, args)
Generic node init.
Generic node init.
Generic node init .
def create_node(self, node_type, args):
        """Generic node init."""

        if node_type not in ('root', 'branch', 'leaf'):
            raise Exception('unknown node type')

        # load node name
        if len(args) > 0:
            node_name = args[0]
        else:
            node_name = self.cf.get('node_name', '')
        if not node_name:
            raise UsageError('Node name must be given either in command line or config')

        # load node public location
        if len(args) > 1:
            node_location = args[1]
        else:
            node_location = self.cf.get('public_node_location', '')
        if not node_location:
            raise UsageError('Node public location must be given either in command line or config')

        if len(args) > 2:
            raise UsageError('Too many args, only node name and public connect string allowed')

        # load provider
        provider_loc = self.options.provider
        if not provider_loc:
            provider_loc = self.cf.get('initial_provider_location', '')

        # check if sane
        ok = 0
        for k, v in skytools.parse_connect_string(node_location):
            if k in ('host', 'service'):
                ok = 1
                break
        if not ok:
            self.log.warning('No host= in public connect string, bad idea')

        # connect to database
        db = self.get_database(self.initial_db_name)

        # check if code is installed
        self.install_code(db)

        # query current status
        res = self.exec_query(db, "select * from pgq_node.get_node_info(%s)", [self.queue_name])
        info = res[0]
        if info['node_type'] is not None:
            self.log.info("Node is already initialized as %s", info['node_type'])
            return

        # check if public connstr is sane
        self.check_public_connstr(db, node_location)

        self.log.info("Initializing node")
        node_attrs = {}

        worker_name = self.options.worker
        if not worker_name:
            raise Exception('--worker required')
        combined_queue = self.options.merge
        if combined_queue and node_type != 'leaf':
            raise Exception('--merge can be used only for leafs')

        if self.options.sync_watermark:
            if node_type != 'branch':
                raise UsageError('--sync-watermark can be used only for branch nodes')
            node_attrs['sync_watermark'] = self.options.sync_watermark

        # register member
        if node_type == 'root':
            global_watermark = None
            combined_queue = None
            provider_name = None
            self.exec_cmd(db, "select * from pgq_node.register_location(%s, %s, %s, false)",
                          [self.queue_name, node_name, node_location])
            self.exec_cmd(db, "select * from pgq_node.create_node(%s, %s, %s, %s, %s, %s, %s)",
                          [self.queue_name, node_type, node_name, worker_name, provider_name, global_watermark, combined_queue])
            provider_db = None
        else:
            if not provider_loc:
                raise Exception('Please specify --provider')

            root_db = self.find_root_db(provider_loc)
            queue_info = self.load_queue_info(root_db)

            # check if member already exists
            if queue_info.get_member(node_name) is not None:
                self.log.error("Node '%s' already exists", node_name)
                sys.exit(1)

            combined_set = None

            provider_db = self.get_database('provider_db', connstr = provider_loc, profile = 'remote')
            q = "select node_type, node_name from pgq_node.get_node_info(%s)"
            res = self.exec_query(provider_db, q, [self.queue_name])
            row = res[0]
            if not row['node_name']:
                raise Exception("provider node not found")
            provider_name = row['node_name']

            # register member on root
            self.exec_cmd(root_db, "select * from pgq_node.register_location(%s, %s, %s, false)",
                          [self.queue_name, node_name, node_location])

            # lookup provider
            provider = queue_info.get_member(provider_name)
            if not provider:
                self.log.error("Node %s does not exist", provider_name)
                sys.exit(1)

            # register on provider
            self.exec_cmd(provider_db, "select * from pgq_node.register_location(%s, %s, %s, false)",
                          [self.queue_name, node_name, node_location])
            rows = self.exec_cmd(provider_db, "select * from pgq_node.register_subscriber(%s, %s, %s, null)",
                                 [self.queue_name, node_name, worker_name])
            global_watermark = rows[0]['global_watermark']

            # initialize node itself

            # insert members
            self.exec_cmd(db, "select * from pgq_node.register_location(%s, %s, %s, false)",
                          [self.queue_name, node_name, node_location])
            for m in queue_info.member_map.values():
                self.exec_cmd(db, "select * from pgq_node.register_location(%s, %s, %s, %s)",
                              [self.queue_name, m.name, m.location, m.dead])

            # real init
            self.exec_cmd(db, "select * from pgq_node.create_node(%s, %s, %s, %s, %s, %s, %s)",
                          [ self.queue_name, node_type, node_name, worker_name,
                            provider_name, global_watermark, combined_queue ])

        self.extra_init(node_type, db, provider_db)

        if node_attrs:
            s_attrs = skytools.db_urlencode(node_attrs)
            self.exec_cmd(db, "select * from pgq_node.set_node_attrs(%s, %s)",
                          [self.queue_name, s_attrs])

        self.log.info("Done")
def create_node ( self , node_type , args ) : if node_type not in ( 'root' , 'branch' , 'leaf' ) : raise Exception ( 'unknown node type' ) # load node name if len ( args ) > 0 : node_name = args [ 0 ] else : node_name = self . cf . get ( 'node_name' , '' ) if not node_name : raise UsageError ( 'Node name must be given either in command line or config' ) # load node public location if len ( args ) > 1 : node_location = args [ 1 ] else : node_location = self . cf . get ( 'public_node_location' , '' ) if not node_location : raise UsageError ( 'Node public location must be given either in command line or config' ) if len ( args ) > 2 : raise UsageError ( 'Too many args, only node name and public connect string allowed' ) # load provider provider_loc = self . options . provider if not provider_loc : provider_loc = self . cf . get ( 'initial_provider_location' , '' ) # check if sane ok = 0 for k , v in skytools . parse_connect_string ( node_location ) : if k in ( 'host' , 'service' ) : ok = 1 break if not ok : self . log . warning ( 'No host= in public connect string, bad idea' ) # connect to database db = self . get_database ( self . initial_db_name ) # check if code is installed self . install_code ( db ) # query current status res = self . exec_query ( db , "select * from pgq_node.get_node_info(%s)" , [ self . queue_name ] ) info = res [ 0 ] if info [ 'node_type' ] is not None : self . log . info ( "Node is already initialized as %s" , info [ 'node_type' ] ) return # check if public connstr is sane self . check_public_connstr ( db , node_location ) self . log . info ( "Initializing node" ) node_attrs = { } worker_name = self . options . worker if not worker_name : raise Exception ( '--worker required' ) combined_queue = self . options . merge if combined_queue and node_type != 'leaf' : raise Exception ( '--merge can be used only for leafs' ) if self . options . sync_watermark : if node_type != 'branch' : raise UsageError ( '--sync-watermark can be used only for branch nodes' ) node_attrs [ 'sync_watermark' ] = self . options . sync_watermark # register member if node_type == 'root' : global_watermark = None combined_queue = None provider_name = None self . exec_cmd ( db , "select * from pgq_node.register_location(%s, %s, %s, false)" , [ self . queue_name , node_name , node_location ] ) self . exec_cmd ( db , "select * from pgq_node.create_node(%s, %s, %s, %s, %s, %s, %s)" , [ self . queue_name , node_type , node_name , worker_name , provider_name , global_watermark , combined_queue ] ) provider_db = None else : if not provider_loc : raise Exception ( 'Please specify --provider' ) root_db = self . find_root_db ( provider_loc ) queue_info = self . load_queue_info ( root_db ) # check if member already exists if queue_info . get_member ( node_name ) is not None : self . log . error ( "Node '%s' already exists" , node_name ) sys . exit ( 1 ) combined_set = None provider_db = self . get_database ( 'provider_db' , connstr = provider_loc , profile = 'remote' ) q = "select node_type, node_name from pgq_node.get_node_info(%s)" res = self . exec_query ( provider_db , q , [ self . queue_name ] ) row = res [ 0 ] if not row [ 'node_name' ] : raise Exception ( "provider node not found" ) provider_name = row [ 'node_name' ] # register member on root self . exec_cmd ( root_db , "select * from pgq_node.register_location(%s, %s, %s, false)" , [ self . queue_name , node_name , node_location ] ) # lookup provider provider = queue_info . get_member ( provider_name ) if not provider : self . log . error ( "Node %s does not exist" , provider_name ) sys . exit ( 1 ) # register on provider self . exec_cmd ( provider_db , "select * from pgq_node.register_location(%s, %s, %s, false)" , [ self . queue_name , node_name , node_location ] ) rows = self . exec_cmd ( provider_db , "select * from pgq_node.register_subscriber(%s, %s, %s, null)" , [ self . queue_name , node_name , worker_name ] ) global_watermark = rows [ 0 ] [ 'global_watermark' ] # initialize node itself # insert members self . exec_cmd ( db , "select * from pgq_node.register_location(%s, %s, %s, false)" , [ self . queue_name , node_name , node_location ] ) for m in queue_info . member_map . values ( ) : self . exec_cmd ( db , "select * from pgq_node.register_location(%s, %s, %s, %s)" , [ self . queue_name , m . name , m . location , m . dead ] ) # real init self . exec_cmd ( db , "select * from pgq_node.create_node(%s, %s, %s, %s, %s, %s, %s)" , [ self . queue_name , node_type , node_name , worker_name , provider_name , global_watermark , combined_queue ] ) self . extra_init ( node_type , db , provider_db ) if node_attrs : s_attrs = skytools . db_urlencode ( node_attrs ) self . exec_cmd ( db , "select * from pgq_node.set_node_attrs(%s, %s)" , [ self . queue_name , s_attrs ] ) self . log . info ( "Done" )