what is working to get rid of this issue , add some checks after agent revived in "fetch_and_sync_all_routers" method @neutron/agent/l3/agent.py
if r.get('_ha_state') == 'unknown':
LOG.debug("Processing router with l3_agent in " "unknown state")
self._process_added_router(r)
and entire method looks like this :
def fetch_and_sync_all_routers(self, context, ns_manager): prev_router_ids = set(self.router_info) curr_router_ids = set()
timestamp = timeutils.utcnow()
router_ids = []
chunk = [] is_snat_agent = (self.conf.agent_mode == lib_const.L3_AGENT_MODE_DVR_SNAT)
try: router_ids = self.plugin_rpc.get_router_ids(context) LOG.debug("Router IDs from Neutron API: %s", router_ids) self._remove_orphan_routers_config( ha_confs_path=self.ha_confs_path, router_ids=set(router_ids),
)
# fetch routers by chunks to reduce the load on server and to
# start router processing earlier
for i in range(0, len(router_ids), self.sync_routers_chunk_size): chunk = router_ids[i:i + self.sync_routers_chunk_size] routers = self.plugin_rpc.get_routers(context, chunk) LOG.debug('Processing :%r', routers)
for r in routers: curr_router_ids.add(r['id']) ns_manager.keep_router(r['id']) if r.get('distributed'): # need to keep fip namespaces as well ext_net_id = (r['external_gateway_info'] or {}).get( 'network_id') if ext_net_id: ns_manager.keep_ext_net(ext_net_id) elif is_snat_agent and not r.get('ha'): ns_manager.ensure_snat_cleanup(r['id']) if r.get('_ha_state') == 'unknown': LOG.debug("Processing router with l3_agent in " "unknown state") self._process_added_router(r) update = queue.ResourceUpdate( r['id'], PRIORITY_SYNC_ROUTERS_TASK, resource=r, action=ADD_UPDATE_ROUTER, timestamp=timestamp) self._queue.add(update)
except oslo_messaging.MessagingTimeout:
if self.sync_routers_chunk_size > SYNC_ROUTERS_MIN_CHUNK_SIZE: self.sync_routers_chunk_size = max( self.sync_routers_chunk_size // 2, SYNC_ROUTERS_MIN_CHUNK_SIZE) LOG.error('Server failed to return info for routers in ' 'required time, decreasing chunk size to: %s', self.sync_routers_chunk_size)
else: LOG.error('Server failed to return info for routers in ' 'required time even with min chunk size: %s. ' 'It might be under very high load or ' 'just inoperable', self.sync_routers_chunk_size)
raise
except oslo_messaging.MessagingException: failed_routers = chunk or router_ids LOG.exception("Failed synchronizing routers '%s' " "due to RPC error", failed_routers)
raise l3_exc.AbortSyncRouters() self.fullsync = False LOG.debug("periodic_sync_routers_task successfully completed")
# adjust chunk size after successful sync
if self.sync_routers_chunk_size < SYNC_ROUTERS_MAX_CHUNK_SIZE: self.sync_routers_chunk_size = min( self.sync_routers_chunk_size + SYNC_ROUTERS_MIN_CHUNK_SIZE, SYNC_ROUTERS_MAX_CHUNK_SIZE)
# Delete routers that have disappeared since the last sync
for router_id in prev_router_ids - curr_router_ids: ns_manager.keep_router(router_id)
update = queue.ResourceUpdate(router_id, PRIORITY_SYNC_ROUTERS_TASK, timestamp=timestamp, action=DELETE_ROUTER) self._queue.add(update)
what is working to get rid of this issue , add some checks after agent revived in "fetch_ and_sync_ all_routers" method @neutron/ agent/l3/ agent.py
if r.get('_ha_state') == 'unknown': debug(" Processing router with l3_agent in "
"unknown state") _process_ added_router( r)
LOG.
self.
and entire method looks like this :
def fetch_and_ sync_all_ routers( self, context, ns_manager):
prev_router_ ids = set(self. router_ info)
curr_router_ ids = set()
is_snat_ agent = (self.conf. agent_mode ==
lib_ const.L3_ AGENT_MODE_ DVR_SNAT)
router_ ids = self.plugin_ rpc.get_ router_ ids(context)
LOG. debug(" Router IDs from Neutron API: %s", router_ids)
self. _remove_ orphan_ routers_ config(
ha_confs_ path=self. ha_confs_ path,
router_ ids=set( router_ ids), routers_ chunk_size) :
chunk = router_ids[i:i + self.sync_ routers_ chunk_size]
routers = self.plugin_ rpc.get_ routers( context, chunk)
LOG.debug( 'Processing :%r', routers)
curr_ router_ ids.add( r['id'] )
ns_ manager. keep_router( r['id'] )
if r.get(' distributed' ):
# need to keep fip namespaces as well
ext_ net_id = (r['external_ gateway_ info'] or {}).get(
'network_ id')
if ext_net_id:
ns_manager. keep_ext_ net(ext_ net_id)
elif is_snat_agent and not r.get('ha'):
ns_manager. ensure_ snat_cleanup( r['id'] )
if r.get('_ha_state') == 'unknown':
LOG.debug( "Processing router with l3_agent in "
"unknown state")
self._process_ added_router( r)
update = queue.ResourceU pdate(
r['id' ],
PRIORITY_ SYNC_ROUTERS_ TASK,
resource= r,
action= ADD_UPDATE_ ROUTER,
timestamp= timestamp)
self. _queue. add(update) MessagingTimeou t: routers_ chunk_size > SYNC_ROUTERS_ MIN_CHUNK_ SIZE:
self. sync_routers_ chunk_size = max(
self. sync_routers_ chunk_size // 2,
SYNC_ ROUTERS_ MIN_CHUNK_ SIZE)
LOG.error( 'Server failed to return info for routers in '
'required time, decreasing chunk size to: %s',
self. sync_routers_ chunk_size)
LOG.error( 'Server failed to return info for routers in '
'required time even with min chunk size: %s. '
'It might be under very high load or '
'just inoperable',
self. sync_routers_ chunk_size) MessagingExcept ion:
failed_ routers = chunk or router_ids
LOG. exception( "Failed synchronizing routers '%s' "
"due to RPC error", failed_routers) AbortSyncRouter s()
self.fullsync = False
LOG.debug( "periodic_ sync_routers_ task successfully completed") routers_ chunk_size < SYNC_ROUTERS_ MAX_CHUNK_ SIZE:
self. sync_routers_ chunk_size = min(
self. sync_routers_ chunk_size + SYNC_ROUTERS_ MIN_CHUNK_ SIZE,
SYNC_ ROUTERS_ MAX_CHUNK_ SIZE)
ns_ manager. keep_router( router_ id) pdate(router_ id,
PRIORITY_ SYNC_ROUTERS_ TASK,
timestamp= timestamp,
action= DELETE_ ROUTER)
self. _queue. add(update)
timestamp = timeutils.utcnow()
router_ids = []
chunk = []
try:
)
# fetch routers by chunks to reduce the load on server and to
# start router processing earlier
for i in range(0, len(router_ids), self.sync_
for r in routers:
except oslo_messaging.
if self.sync_
else:
raise
except oslo_messaging.
raise l3_exc.
# adjust chunk size after successful sync
if self.sync_
# Delete routers that have disappeared since the last sync
for router_id in prev_router_ids - curr_router_ids:
update = queue.ResourceU