performa/shaker
Code Issues Proposed changes

Ignore messages from unexpected agents

Browse Source 
Change-Id: Ie701a5cb19a44b8d091883737f91dd7fa1b7bfab
This commit is contained in:
Ilya Shakhat 2015-03-24 17:28:18 +03:00
parent 3ed919553a
commit 5d158d7d03
1 changed files with 24 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

40
shaker/engine/server.py
View File

@ -44,15 +44,20 @@ class Quorum(object):
LOG.debug('Waiting for quorum of agents: %s', agent_ids) LOG.debug('Waiting for quorum of agents: %s', agent_ids)
alive_agents = set() alive_agents = set()
for message, reply_handler in self.message_queue: for message, reply_handler in self.message_queue:
agent_id = message.get('agent_id') msg_agent_id = message.get('agent_id')
alive_agents.add(agent_id)
if msg_agent_id not in agent_ids:
reply_handler(dict(operation='none'))
continue
alive_agents.add(msg_agent_id)
reply_handler(dict(operation='configure', reply_handler(dict(operation='configure',
polling_interval=self.polling_interval)) polling_interval=self.polling_interval))
LOG.debug('Alive agents: %s', alive_agents) LOG.debug('Alive agents: %s', alive_agents)
if alive_agents >= agent_ids: if alive_agents == agent_ids:
LOG.info('All expected agents are alive') LOG.info('All expected agents are alive')
break break
@ -68,21 +73,24 @@ class Quorum(object):
operation = message.get('operation') operation = message.get('operation')
reply = {'operation': 'none'} reply = {'operation': 'none'}
if agent_id in test_case:
# message from a known agent
test = test_case[agent_id] if agent_id not in test_case:
reply_handler(reply)
continue
if operation == 'poll': # message from a known agent
reply = { test = test_case[agent_id]
'operation': 'execute',
'start_at': start_at, if operation == 'poll':
'command': test.get_command(), reply = {
} 'operation': 'execute',
working_agents.add(agent_id) 'start_at': start_at,
elif operation == 'reply': 'command': test.get_command(),
replied_agents.add(agent_id) }
result[agent_id] = test.process_reply(message) working_agents.add(agent_id)
elif operation == 'reply':
replied_agents.add(agent_id)
result[agent_id] = test.process_reply(message)
reply_handler(reply) reply_handler(reply)