You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@slider.apache.org by st...@apache.org on 2014/08/20 11:26:19 UTC
[21/50] [abbrv] git commit: SLIDER-323. Allocated ports must be
reported back upon AM restart SLIDER-262. Slider agent should provide process
supervision such as auto-restart
SLIDER-323. Allocated ports must be reported back upon AM restart
SLIDER-262. Slider agent should provide process supervision such as auto-restart
Project: http://git-wip-us.apache.org/repos/asf/incubator-slider/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-slider/commit/47822dff
Tree: http://git-wip-us.apache.org/repos/asf/incubator-slider/tree/47822dff
Diff: http://git-wip-us.apache.org/repos/asf/incubator-slider/diff/47822dff
Branch: refs/heads/master
Commit: 47822dffa3e086c198fed820ec1e4b099a453659
Parents: 18b0545
Author: Sumit Mohanty <sm...@hortonworks.com>
Authored: Thu Aug 14 17:36:59 2014 -0700
Committer: Sumit Mohanty <sm...@hortonworks.com>
Committed: Thu Aug 14 17:57:43 2014 -0700
----------------------------------------------------------------------
.../src/main/python/agent/ActionQueue.py | 11 ++-
.../src/main/python/agent/CommandStatusDict.py | 5 +-
.../src/main/python/agent/Controller.py | 27 ++++---
.../python/agent/CustomServiceOrchestrator.py | 8 +-
slider-agent/src/main/python/agent/Heartbeat.py | 9 ++-
slider-agent/src/main/python/agent/Register.py | 7 +-
.../src/test/python/agent/TestActionQueue.py | 2 -
.../src/test/python/agent/TestController.py | 29 +++++--
.../agent/TestCustomServiceOrchestrator.py | 2 +
.../src/test/python/agent/TestHeartbeat.py | 82 +++++++++++++++++++-
.../src/test/python/agent/TestRegistration.py | 8 +-
.../providers/agent/AgentProviderService.java | 51 +++++++++---
.../appmaster/web/rest/agent/HeartBeat.java | 10 ---
.../appmaster/web/rest/agent/Register.java | 58 ++++++++++----
.../agent/TestAgentProviderService.java | 17 ++++
15 files changed, 254 insertions(+), 72 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/47822dff/slider-agent/src/main/python/agent/ActionQueue.py
----------------------------------------------------------------------
diff --git a/slider-agent/src/main/python/agent/ActionQueue.py b/slider-agent/src/main/python/agent/ActionQueue.py
index fba9aa2..4c45a76 100644
--- a/slider-agent/src/main/python/agent/ActionQueue.py
+++ b/slider-agent/src/main/python/agent/ActionQueue.py
@@ -122,7 +122,8 @@ class ActionQueue(threading.Thread):
taskId = command['taskId']
- reportResult = not command[Constants.AUTO_GENERATED]
+ # if auto generated then do not report result
+ reportResult = CommandStatusDict.shouldReportResult(command)
# Preparing 'IN_PROGRESS' report
in_progress_status = self.commandStatuses.generate_report_template(command)
@@ -140,7 +141,8 @@ class ActionQueue(threading.Thread):
if ActionQueue.STORE_APPLIED_CONFIG in command['commandParams']:
store_config = 'true' == command['commandParams'][ActionQueue.STORE_APPLIED_CONFIG]
store_command = False
- if ActionQueue.AUTO_RESTART in command['roleParams']:
+ if 'roleParams' in command and ActionQueue.AUTO_RESTART in command['roleParams']:
+ logger.info("Component has indicated auto-restart. Saving details from START command.")
store_command = 'true' == command['roleParams'][ActionQueue.AUTO_RESTART]
@@ -195,10 +197,7 @@ class ActionQueue(threading.Thread):
cluster = command['clusterName']
service = command['serviceName']
component = command['componentName']
- reportResult = True
- if Constants.AUTO_GENERATED in command:
- reportResult = not command[Constants.AUTO_GENERATED]
-
+ reportResult = CommandStatusDict.shouldReportResult(command)
component_status = self.customServiceOrchestrator.requestComponentStatus(command)
result = {"componentName": component,
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/47822dff/slider-agent/src/main/python/agent/CommandStatusDict.py
----------------------------------------------------------------------
diff --git a/slider-agent/src/main/python/agent/CommandStatusDict.py b/slider-agent/src/main/python/agent/CommandStatusDict.py
index f20e0fa..bcbce9e 100644
--- a/slider-agent/src/main/python/agent/CommandStatusDict.py
+++ b/slider-agent/src/main/python/agent/CommandStatusDict.py
@@ -114,7 +114,7 @@ class CommandStatusDict():
grep = Grep()
output = grep.tail(tmpout, Grep.OUTPUT_LAST_LINES)
inprogress = self.generate_report_template(command)
- reportResult = not command[Constants.AUTO_GENERATED]
+ reportResult = CommandStatusDict.shouldReportResult(command)
inprogress.update({
'stdout': grep.filterMarkup(output),
'stderr': tmperr,
@@ -142,3 +142,6 @@ class CommandStatusDict():
return stub
+ @staticmethod
+ def shouldReportResult(command):
+ return not (Constants.AUTO_GENERATED in command and command[Constants.AUTO_GENERATED])
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/47822dff/slider-agent/src/main/python/agent/Controller.py
----------------------------------------------------------------------
diff --git a/slider-agent/src/main/python/agent/Controller.py b/slider-agent/src/main/python/agent/Controller.py
index 6bdfc26..1e27efa 100644
--- a/slider-agent/src/main/python/agent/Controller.py
+++ b/slider-agent/src/main/python/agent/Controller.py
@@ -118,7 +118,11 @@ class Controller(threading.Thread):
while not self.isRegistered:
try:
- data = json.dumps(self.register.build(id))
+ data = json.dumps(self.register.build(
+ self.componentActualState,
+ self.componentExpectedState,
+ self.actionQueue.customServiceOrchestrator.allocated_ports,
+ id))
logger.info("Registering with the server at " + self.registerUrl +
" with data " + pprint.pformat(data))
response = self.sendRequest(self.registerUrl, data)
@@ -211,7 +215,7 @@ class Controller(threading.Thread):
try:
if not retry:
data = json.dumps(
- self.heartbeat.build(commandResult, self.componentActualState,
+ self.heartbeat.build(commandResult,
self.responseId, self.hasMappedComponents))
self.updateStateBasedOnResult(commandResult)
logger.debug("Sending request: " + data)
@@ -225,8 +229,11 @@ class Controller(threading.Thread):
serverId = int(response['responseId'])
+ restartEnabled = False
if 'restartEnabled' in response:
- restartEnabled = 'true' == response['restartEnabled']
+ restartEnabled = response['restartEnabled']
+ if restartEnabled:
+ logger.info("Component auto-restart is enabled.")
if 'hasMappedComponents' in response.keys():
self.hasMappedComponents = response['hasMappedComponents'] != False
@@ -262,15 +269,16 @@ class Controller(threading.Thread):
pass
# Add a start command
- if self.componentActualState == State.INSTALLED and \
+ if self.componentActualState == State.FAILED and \
self.componentExpectedState == State.STARTED and restartEnabled:
stored_command = self.actionQueue.customServiceOrchestrator.stored_command
if len(stored_command) > 0:
- auto_start_command = self.create_start_command(stored_command)
- if auto_start_command:
- logger.info("Automatically adding a start command.")
- self.updateStateBasedOnCommand([auto_start_command], False)
- self.addToQueue([auto_start_command])
+ auto_start_command = self.create_start_command(stored_command)
+ if auto_start_command:
+ logger.info("Automatically adding a start command.")
+ logger.debug("Auto start command: " + pprint.pformat(auto_start_command))
+ self.updateStateBasedOnCommand([auto_start_command], False)
+ self.addToQueue([auto_start_command])
pass
# Add a status command
@@ -388,6 +396,7 @@ class Controller(threading.Thread):
if "healthStatus" in commandResult:
if commandResult["healthStatus"] == "INSTALLED":
+ # Mark it FAILED as its a failure remedied by auto-start or container restart
self.componentActualState = State.FAILED
self.failureCount += 1
self.logStates()
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/47822dff/slider-agent/src/main/python/agent/CustomServiceOrchestrator.py
----------------------------------------------------------------------
diff --git a/slider-agent/src/main/python/agent/CustomServiceOrchestrator.py b/slider-agent/src/main/python/agent/CustomServiceOrchestrator.py
index 6b2ace5..15f1664 100644
--- a/slider-agent/src/main/python/agent/CustomServiceOrchestrator.py
+++ b/slider-agent/src/main/python/agent/CustomServiceOrchestrator.py
@@ -57,6 +57,7 @@ class CustomServiceOrchestrator():
'status_command_stderr.txt'))
self.public_fqdn = hostname.public_hostname()
self.stored_command = {}
+ self.allocated_ports = {}
# Clean up old status command files if any
try:
os.unlink(self.status_commands_stdout)
@@ -69,7 +70,7 @@ class CustomServiceOrchestrator():
def runCommand(self, command, tmpoutfile, tmperrfile,
override_output_files=True, store_command=False):
- allocated_port = {}
+ allocated_ports = {}
try:
script_type = command['commandParams']['script_type']
script = command['commandParams']['script']
@@ -86,7 +87,7 @@ class CustomServiceOrchestrator():
# We don't support anything else yet
message = "Unknown script type {0}".format(script_type)
raise AgentException(message)
- json_path = self.dump_command_to_json(command, allocated_port, store_command)
+ json_path = self.dump_command_to_json(command, allocated_ports, store_command)
py_file_list = [script_tuple]
# filter None values
filtered_py_file_list = [i for i in py_file_list if i]
@@ -132,7 +133,8 @@ class CustomServiceOrchestrator():
}
if Constants.EXIT_CODE in ret and ret[Constants.EXIT_CODE] == 0:
- ret[Constants.ALLOCATED_PORTS] = allocated_port
+ ret[Constants.ALLOCATED_PORTS] = allocated_ports
+ self.allocated_ports = allocated_ports
# Irrespective of the outcome report the folder paths
if command_name == 'INSTALL':
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/47822dff/slider-agent/src/main/python/agent/Heartbeat.py
----------------------------------------------------------------------
diff --git a/slider-agent/src/main/python/agent/Heartbeat.py b/slider-agent/src/main/python/agent/Heartbeat.py
index aa403d4..b107d92 100644
--- a/slider-agent/src/main/python/agent/Heartbeat.py
+++ b/slider-agent/src/main/python/agent/Heartbeat.py
@@ -36,7 +36,7 @@ class Heartbeat:
self.config = config
self.reports = []
- def build(self, commandResult, componentActualState, id='-1',
+ def build(self, commandResult, id='-1',
componentsMapped=False):
timestamp = int(time.time() * 1000)
queueResult = self.actionQueue.result()
@@ -49,8 +49,7 @@ class Heartbeat:
'timestamp': timestamp,
'hostname': self.config.getLabel(),
'nodeStatus': nodeStatus,
- 'fqdn': hostname.public_hostname(),
- 'agentState': componentActualState
+ 'fqdn': hostname.public_hostname()
}
commandsInProgress = False
@@ -62,6 +61,10 @@ class Heartbeat:
if report['reportResult']:
del report['reportResult']
heartbeat['reports'].append(report)
+ else:
+ # dropping the result but only recording the status
+ commandResult["commandStatus"] = report["status"]
+ pass
if len(heartbeat['reports']) > 0:
# There may be IN_PROGRESS tasks
commandsInProgress = True
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/47822dff/slider-agent/src/main/python/agent/Register.py
----------------------------------------------------------------------
diff --git a/slider-agent/src/main/python/agent/Register.py b/slider-agent/src/main/python/agent/Register.py
index 7c7ff06..b59154f 100644
--- a/slider-agent/src/main/python/agent/Register.py
+++ b/slider-agent/src/main/python/agent/Register.py
@@ -29,7 +29,7 @@ class Register:
def __init__(self, config):
self.config = config
- def build(self, id='-1'):
+ def build(self, actualState, expectedState, allocated_ports, id='-1'):
timestamp = int(time.time() * 1000)
version = self.read_agent_version()
@@ -38,7 +38,10 @@ class Register:
'timestamp': timestamp,
'hostname': self.config.getLabel(),
'publicHostname': hostname.public_hostname(),
- 'agentVersion': version
+ 'agentVersion': version,
+ 'actualState': actualState,
+ 'expectedState': expectedState,
+ 'allocatedPorts': allocated_ports
}
return register
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/47822dff/slider-agent/src/test/python/agent/TestActionQueue.py
----------------------------------------------------------------------
diff --git a/slider-agent/src/test/python/agent/TestActionQueue.py b/slider-agent/src/test/python/agent/TestActionQueue.py
index 4a53c2e..8071ee8 100644
--- a/slider-agent/src/test/python/agent/TestActionQueue.py
+++ b/slider-agent/src/test/python/agent/TestActionQueue.py
@@ -61,8 +61,6 @@ class TestActionQueue(TestCase):
'serviceName': u'HBASE',
'configurations': {'global': {}},
'configurationTags': {'global': {'tag': 'v1'}},
- 'auto_generated': False,
- 'roleParams': {'auto_restart':'false'},
'commandParams': {'script_type': 'PYTHON',
'script': 'scripts/abc.py',
'command_timeout': '600'}
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/47822dff/slider-agent/src/test/python/agent/TestController.py
----------------------------------------------------------------------
diff --git a/slider-agent/src/test/python/agent/TestController.py b/slider-agent/src/test/python/agent/TestController.py
index 91e12f1..401d69a 100644
--- a/slider-agent/src/test/python/agent/TestController.py
+++ b/slider-agent/src/test/python/agent/TestController.py
@@ -55,6 +55,7 @@ class TestController(unittest.TestCase):
self.controller = Controller.Controller(config)
self.controller.netutil.MINIMUM_INTERVAL_BETWEEN_HEARTBEATS = 0.1
self.controller.netutil.HEARTBEAT_NOT_IDDLE_INTERVAL_SEC = 0.1
+ self.controller.actionQueue = ActionQueue.ActionQueue(config, self.controller)
@patch("json.dumps")
@@ -289,7 +290,7 @@ class TestController(unittest.TestCase):
self.controller.sendRequest = sendRequest
self.controller.responseId = 1
- response = {"responseId":"2", "restartAgent":"false"}
+ response = {"responseId":"2", "restartAgent": False}
loadsMock.return_value = response
def one_heartbeat(*args, **kwargs):
@@ -651,8 +652,9 @@ class TestController(unittest.TestCase):
self.controller.sendRequest = sendRequest
self.controller.responseId = 1
- response = {"responseId":"2", "restartAgent":"false", "restartEnabled":'true'}
- loadsMock.return_value = response
+ response1 = {"responseId": "2", "restartAgent": False, "restartEnabled": True}
+ response2 = {"responseId": "2", "restartAgent": False, "restartEnabled": False}
+ loadsMock.side_effect = [response1, response2, response1]
def one_heartbeat(*args, **kwargs):
self.controller.DEBUG_STOP_HEARTBEATING = True
@@ -665,9 +667,9 @@ class TestController(unittest.TestCase):
# one successful request, after stop
self.controller.actionQueue = actionQueue
- self.controller.componentActualState = State.INSTALLED
+ self.controller.componentActualState = State.FAILED
self.controller.componentExpectedState = State.STARTED
- self.assertTrue(self.controller.componentActualState, State.INSTALLED)
+ self.assertTrue(self.controller.componentActualState, State.FAILED)
self.controller.actionQueue.customServiceOrchestrator.stored_command = {
'commandType': 'EXECUTION_COMMAND',
'role': u'HBASE',
@@ -699,6 +701,23 @@ class TestController(unittest.TestCase):
'commandId': '8-1',
'auto_generated': True}])])
self.controller.config = original_value
+
+ # restartEnabled = False
+ self.controller.componentActualState = State.FAILED
+ self.controller.heartbeatWithServer()
+
+ self.assertTrue(sendRequest.called)
+ self.assertTrue(self.controller.componentActualState, State.FAILED)
+ self.assertTrue(self.controller.componentExpectedState, State.STARTED)
+
+ # restartEnabled = True
+ self.controller.componentActualState = State.INSTALLED
+ self.controller.componentExpectedState = State.INSTALLED
+ self.controller.heartbeatWithServer()
+
+ self.assertTrue(sendRequest.called)
+ self.assertTrue(self.controller.componentActualState, State.INSTALLED)
+ self.assertTrue(self.controller.componentExpectedState, State.INSTALLED)
pass
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/47822dff/slider-agent/src/test/python/agent/TestCustomServiceOrchestrator.py
----------------------------------------------------------------------
diff --git a/slider-agent/src/test/python/agent/TestCustomServiceOrchestrator.py b/slider-agent/src/test/python/agent/TestCustomServiceOrchestrator.py
index 6408809..e545afe 100644
--- a/slider-agent/src/test/python/agent/TestCustomServiceOrchestrator.py
+++ b/slider-agent/src/test/python/agent/TestCustomServiceOrchestrator.py
@@ -230,6 +230,8 @@ class TestCustomServiceOrchestrator(TestCase):
self.assertEqual(ret['allocated_ports'], {'a.a.port': '10233'})
self.assertTrue(run_file_mock.called)
self.assertEqual(run_file_mock.call_count, 1)
+ self.assertEqual(orchestrator.allocated_ports, {'a.a.port': '10233'})
+ self.assertEqual(orchestrator.stored_command, {})
@patch.object(socket, "close")
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/47822dff/slider-agent/src/test/python/agent/TestHeartbeat.py
----------------------------------------------------------------------
diff --git a/slider-agent/src/test/python/agent/TestHeartbeat.py b/slider-agent/src/test/python/agent/TestHeartbeat.py
index 70870ec..b012218 100644
--- a/slider-agent/src/test/python/agent/TestHeartbeat.py
+++ b/slider-agent/src/test/python/agent/TestHeartbeat.py
@@ -50,7 +50,7 @@ class TestHeartbeat(TestCase):
dummy_controller = MagicMock()
actionQueue = ActionQueue(config, dummy_controller)
heartbeat = Heartbeat(actionQueue, config)
- result = heartbeat.build({}, State.STARTED, 100)
+ result = heartbeat.build({}, 100)
print "Heartbeat: " + str(result)
self.assertEquals(result['hostname'] != '', True,
"hostname should not be empty")
@@ -63,7 +63,6 @@ class TestHeartbeat(TestCase):
self.assertEquals(len(result['nodeStatus']), 2)
self.assertEquals(result['nodeStatus']['cause'], "NONE")
self.assertEquals(result['nodeStatus']['status'], "HEALTHY")
- self.assertEquals(result['agentState'], State.STARTED)
# result may or may NOT have an agentEnv structure in it
self.assertEquals((len(result) is 6) or (len(result) is 7), True)
self.assertEquals(not heartbeat.reports, True,
@@ -147,7 +146,7 @@ class TestHeartbeat(TestCase):
}
heartbeat = Heartbeat(actionQueue, config)
# State.STARTED results in agentState to be set to 4 (enum order)
- hb = heartbeat.build({}, State.STARTED, 10)
+ hb = heartbeat.build({}, 10)
hb['hostname'] = 'hostname'
hb['timestamp'] = 'timestamp'
hb['fqdn'] = 'fqdn'
@@ -155,7 +154,7 @@ class TestHeartbeat(TestCase):
{'status': 'HEALTHY',
'cause': 'NONE'},
'timestamp': 'timestamp', 'hostname': 'hostname', 'fqdn': 'fqdn',
- 'agentState': 4, 'responseId': 10, 'reports': [
+ 'responseId': 10, 'reports': [
{'status': 'IN_PROGRESS', 'roleCommand': u'INSTALL',
'serviceName': u'HDFS', 'role': u'DATANODE', 'actionId': '1-1',
'stderr': 'Read from /tmp/errors-3.txt',
@@ -178,6 +177,81 @@ class TestHeartbeat(TestCase):
self.assertEqual.__self__.maxDiff = None
self.assertEquals(hb, expected)
+ @patch.object(ActionQueue, "result")
+ def test_build_result2(self, result_mock):
+ config = AgentConfig("", "")
+ config.set('agent', 'prefix', 'tmp')
+ dummy_controller = MagicMock()
+ actionQueue = ActionQueue(config, dummy_controller)
+ result_mock.return_value = {
+ 'reports': [{'status': 'IN_PROGRESS',
+ 'stderr': 'Read from /tmp/errors-3.txt',
+ 'stdout': 'Read from /tmp/output-3.txt',
+ 'clusterName': u'cc',
+ 'roleCommand': u'INSTALL',
+ 'serviceName': u'HDFS',
+ 'role': u'DATANODE',
+ 'actionId': '1-1',
+ 'taskId': 3,
+ 'exitcode': 777,
+ 'reportResult' : False}
+ ],
+ 'componentStatus': []
+ }
+ heartbeat = Heartbeat(actionQueue, config)
+
+ commandResult = {}
+ hb = heartbeat.build(commandResult, 10)
+ hb['hostname'] = 'hostname'
+ hb['timestamp'] = 'timestamp'
+ hb['fqdn'] = 'fqdn'
+ expected = {'nodeStatus':
+ {'status': 'HEALTHY',
+ 'cause': 'NONE'},
+ 'timestamp': 'timestamp', 'hostname': 'hostname', 'fqdn': 'fqdn',
+ 'responseId': 10, 'reports': []}
+ self.assertEqual.__self__.maxDiff = None
+ self.assertEquals(hb, expected)
+ self.assertEquals(commandResult, {'commandStatus': 'IN_PROGRESS'})
+
+ @patch.object(ActionQueue, "result")
+ def test_build_result3(self, result_mock):
+ config = AgentConfig("", "")
+ config.set('agent', 'prefix', 'tmp')
+ dummy_controller = MagicMock()
+ actionQueue = ActionQueue(config, dummy_controller)
+ result_mock.return_value = {
+ 'reports': [{'status': 'COMPLETED',
+ 'stderr': 'Read from /tmp/errors-3.txt',
+ 'stdout': 'Read from /tmp/output-3.txt',
+ 'clusterName': u'cc',
+ 'roleCommand': u'INSTALL',
+ 'serviceName': u'HDFS',
+ 'role': u'DATANODE',
+ 'actionId': '1-1',
+ 'taskId': 3,
+ 'exitcode': 777,
+ 'reportResult' : False}
+ ],
+ 'componentStatus': []
+ }
+ heartbeat = Heartbeat(actionQueue, config)
+
+ commandResult = {}
+ hb = heartbeat.build(commandResult, 10)
+ hb['hostname'] = 'hostname'
+ hb['timestamp'] = 'timestamp'
+ hb['fqdn'] = 'fqdn'
+ expected = {'nodeStatus':
+ {'status': 'HEALTHY',
+ 'cause': 'NONE'},
+ 'timestamp': 'timestamp', 'hostname': 'hostname', 'fqdn': 'fqdn',
+ 'responseId': 10, 'reports': []}
+ self.assertEqual.__self__.maxDiff = None
+ self.assertEquals(hb, expected)
+ self.assertEquals(commandResult, {'commandStatus': 'COMPLETED'})
+
+
if __name__ == "__main__":
logging.basicConfig(format='%(asctime)s %(message)s', level=logging.DEBUG)
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/47822dff/slider-agent/src/test/python/agent/TestRegistration.py
----------------------------------------------------------------------
diff --git a/slider-agent/src/test/python/agent/TestRegistration.py b/slider-agent/src/test/python/agent/TestRegistration.py
index 5245af9..f91fe29 100644
--- a/slider-agent/src/test/python/agent/TestRegistration.py
+++ b/slider-agent/src/test/python/agent/TestRegistration.py
@@ -26,6 +26,7 @@ import tempfile
from mock.mock import patch
from mock.mock import MagicMock
from Register import Register
+from Controller import State
from AgentConfig import AgentConfig
class TestRegistration(TestCase):
@@ -47,14 +48,17 @@ class TestRegistration(TestCase):
text_file.write("1.3.0")
register = Register(config)
- data = register.build(1)
+ data = register.build(State.INIT, State.INIT, {}, 1)
#print ("Register: " + pprint.pformat(data))
self.assertEquals(data['hostname'] != "", True, "hostname should not be empty")
self.assertEquals(data['publicHostname'] != "", True, "publicHostname should not be empty")
self.assertEquals(data['responseId'], 1)
self.assertEquals(data['timestamp'] > 1353678475465L, True, "timestamp should not be empty")
self.assertEquals(data['agentVersion'], '1.3.0', "agentVersion should not be empty")
- self.assertEquals(len(data), 5)
+ self.assertEquals(data['actualState'], State.INIT, "actualState should not be empty")
+ self.assertEquals(data['expectedState'], State.INIT, "expectedState should not be empty")
+ self.assertEquals(data['allocatedPorts'], {}, "allocatedPorts should be empty")
+ self.assertEquals(len(data), 8)
self.assertEquals(os.path.join(tmpdir, "app/definition"), config.getResolvedPath("app_pkg_dir"))
self.assertEquals(os.path.join(tmpdir, "app/install"), config.getResolvedPath("app_install_dir"))
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/47822dff/slider-core/src/main/java/org/apache/slider/providers/agent/AgentProviderService.java
----------------------------------------------------------------------
diff --git a/slider-core/src/main/java/org/apache/slider/providers/agent/AgentProviderService.java b/slider-core/src/main/java/org/apache/slider/providers/agent/AgentProviderService.java
index b22ef39..8ce1f18 100644
--- a/slider-core/src/main/java/org/apache/slider/providers/agent/AgentProviderService.java
+++ b/slider-core/src/main/java/org/apache/slider/providers/agent/AgentProviderService.java
@@ -398,9 +398,19 @@ public class AgentProviderService extends AbstractProviderService implements
log.info("Handling registration: " + registration);
RegistrationResponse response = new RegistrationResponse();
String label = registration.getHostname();
+ State agentState = registration.getActualState();
if (getComponentStatuses().containsKey(label)) {
response.setResponseStatus(RegistrationStatus.OK);
- getComponentStatuses().get(label).heartbeat(System.currentTimeMillis());
+ ComponentInstanceState componentStatus = getComponentStatuses().get(label);
+ componentStatus.heartbeat(System.currentTimeMillis());
+ updateComponentStatusWithAgentState(componentStatus, agentState);
+
+ Map<String, String> ports = registration.getAllocatedPorts();
+ if (ports != null && !ports.isEmpty()) {
+ String roleName = getRoleName(label);
+ String containerId = getContainerId(label);
+ processAllocatedPorts(registration.getPublicHostname(), roleName, containerId, ports);
+ }
} else {
response.setResponseStatus(RegistrationStatus.FAILED);
response.setLog("Label not recognized.");
@@ -424,9 +434,8 @@ public class AgentProviderService extends AbstractProviderService implements
String label = heartBeat.getHostname();
String roleName = getRoleName(label);
- State agentState = heartBeat.getAgentState();
-
String containerId = getContainerId(label);
+
StateAccessForProviders accessor = getAmState();
String scriptPath = getScriptPathFromMetainfo(roleName);
@@ -441,13 +450,22 @@ public class AgentProviderService extends AbstractProviderService implements
Boolean isMaster = isMaster(roleName);
ComponentInstanceState componentStatus = getComponentStatuses().get(label);
- updateComponentStatusWithAgentState(componentStatus, agentState);
componentStatus.heartbeat(System.currentTimeMillis());
// If no Master can explicitly publish then publish if its a master
// Otherwise, wait till the master that can publish is ready
if (isMaster &&
(canAnyMasterPublishConfig() == false || canPublishConfig(roleName))) {
publishConfigAndExportGroups(heartBeat, componentStatus);
+ } else {
+ // Ack that config has been reported
+ List<ComponentStatus> statuses = heartBeat.getComponentStatus();
+ if (statuses != null && !statuses.isEmpty()) {
+ ComponentStatus status = statuses.get(0);
+ if(status.getConfigs().size() > 0) {
+ log.info("Config got reported by {} but discarded as component {} cannot publish.", label, roleName);
+ componentStatus.setConfigReported(true);
+ }
+ }
}
List<CommandReport> reports = heartBeat.getReports();
@@ -455,14 +473,7 @@ public class AgentProviderService extends AbstractProviderService implements
CommandReport report = reports.get(0);
Map<String, String> ports = report.getAllocatedPorts();
if (ports != null && !ports.isEmpty()) {
- for (Map.Entry<String, String> port : ports.entrySet()) {
- log.info("Recording allocated port for {} as {}", port.getKey(), port.getValue());
- this.getAllocatedPorts().put(port.getKey(), port.getValue());
- this.getAllocatedPorts(containerId).put(port.getKey(), port.getValue());
- }
-
- // component specific publishes
- processAndPublishComponentSpecificData(ports, containerId, heartBeat.getFqdn(), roleName);
+ processAllocatedPorts(heartBeat.getFqdn(), roleName, containerId, ports);
}
CommandResult result = CommandResult.getCommandResult(report.getStatus());
Command command = Command.getCommand(report.getRoleCommand());
@@ -506,6 +517,7 @@ public class AgentProviderService extends AbstractProviderService implements
if (isMaster && componentStatus.getState() == State.STARTED
&& command == Command.NOP) {
if (!componentStatus.getConfigReported()) {
+ log.info("Requesting applied config for {} on {}.", roleName, containerId);
addGetConfigCommand(roleName, containerId, response);
}
}
@@ -525,6 +537,20 @@ public class AgentProviderService extends AbstractProviderService implements
return response;
}
+ protected void processAllocatedPorts(String fqdn,
+ String roleName,
+ String containerId,
+ Map<String, String> ports) {
+ for (Map.Entry<String, String> port : ports.entrySet()) {
+ log.info("Recording allocated port for {} as {}", port.getKey(), port.getValue());
+ this.getAllocatedPorts().put(port.getKey(), port.getValue());
+ this.getAllocatedPorts(containerId).put(port.getKey(), port.getValue());
+ }
+
+ // component specific publishes
+ processAndPublishComponentSpecificData(ports, containerId, fqdn, roleName);
+ }
+
private void updateComponentStatusWithAgentState(
ComponentInstanceState componentStatus, State agentState) {
if (agentState != null) {
@@ -803,6 +829,7 @@ public class AgentProviderService extends AbstractProviderService implements
}
}
}
+ log.info("Received and stored config for {}", heartBeat.getHostname());
componentStatus.setConfigReported(true);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/47822dff/slider-core/src/main/java/org/apache/slider/server/appmaster/web/rest/agent/HeartBeat.java
----------------------------------------------------------------------
diff --git a/slider-core/src/main/java/org/apache/slider/server/appmaster/web/rest/agent/HeartBeat.java b/slider-core/src/main/java/org/apache/slider/server/appmaster/web/rest/agent/HeartBeat.java
index 62df18d..a08d46d 100644
--- a/slider-core/src/main/java/org/apache/slider/server/appmaster/web/rest/agent/HeartBeat.java
+++ b/slider-core/src/main/java/org/apache/slider/server/appmaster/web/rest/agent/HeartBeat.java
@@ -45,7 +45,6 @@ public class HeartBeat {
HostStatus nodeStatus;
private AgentEnv agentEnv = null;
private String fqdn;
- private State agentState;
public long getResponseId() {
return responseId;
@@ -125,14 +124,6 @@ public class HeartBeat {
this.mounts = mounts;
}
- public State getAgentState() {
- return agentState;
- }
-
- public void setAgentState(State agentState) {
- this.agentState = agentState;
- }
-
@Override
public String toString() {
return "HeartBeat{" +
@@ -142,7 +133,6 @@ public class HeartBeat {
", reports=" + reports +
", componentStatus=" + componentStatus +
", nodeStatus=" + nodeStatus +
- ", agentState=" + agentState +
'}';
}
}
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/47822dff/slider-core/src/main/java/org/apache/slider/server/appmaster/web/rest/agent/Register.java
----------------------------------------------------------------------
diff --git a/slider-core/src/main/java/org/apache/slider/server/appmaster/web/rest/agent/Register.java b/slider-core/src/main/java/org/apache/slider/server/appmaster/web/rest/agent/Register.java
index 9299a16..a44c3a4 100644
--- a/slider-core/src/main/java/org/apache/slider/server/appmaster/web/rest/agent/Register.java
+++ b/slider-core/src/main/java/org/apache/slider/server/appmaster/web/rest/agent/Register.java
@@ -16,15 +16,14 @@
*/
package org.apache.slider.server.appmaster.web.rest.agent;
+import org.apache.slider.providers.agent.State;
import org.codehaus.jackson.annotate.JsonIgnoreProperties;
import org.codehaus.jackson.annotate.JsonProperty;
import org.codehaus.jackson.map.annotate.JsonSerialize;
-/**
- *
- * Data model for agent to send heartbeat to ambari and/or app master.
- *
- */
+import java.util.Map;
+
+/** Data model for agent to send heartbeat to ambari and/or app master. */
@JsonIgnoreProperties(ignoreUnknown = true)
@JsonSerialize(include = JsonSerialize.Inclusion.NON_NULL)
public class Register {
@@ -36,6 +35,9 @@ public class Register {
private String publicHostname;
private AgentEnv agentEnv;
private String agentVersion;
+ private State actualState;
+ private State expectedState;
+ private Map<String, String> allocatedPorts;
@JsonProperty("responseId")
public int getResponseId() {
@@ -44,13 +46,17 @@ public class Register {
@JsonProperty("responseId")
public void setResponseId(int responseId) {
- this.responseId=responseId;
+ this.responseId = responseId;
}
public long getTimestamp() {
return timestamp;
}
+ public void setTimestamp(long timestamp) {
+ this.timestamp = timestamp;
+ }
+
public String getHostname() {
return hostname;
}
@@ -67,10 +73,6 @@ public class Register {
this.hardwareProfile = hardwareProfile;
}
- public void setTimestamp(long timestamp) {
- this.timestamp = timestamp;
- }
-
public String getPublicHostname() {
return publicHostname;
}
@@ -103,15 +105,45 @@ public class Register {
this.currentPingPort = currentPingPort;
}
+ public State getActualState() {
+ return actualState;
+ }
+
+ public void setActualState(State actualState) {
+ this.actualState = actualState;
+ }
+
+ public State getExpectedState() {
+ return expectedState;
+ }
+
+ public void setExpectedState(State expectedState) {
+ this.expectedState = expectedState;
+ }
+
+ /** @return the allocated ports, or <code>null</code> if none are present */
+ @JsonProperty("allocatedPorts")
+ public Map<String, String> getAllocatedPorts() {
+ return allocatedPorts;
+ }
+
+ /** @param ports allocated ports */
+ @JsonProperty("allocatedPorts")
+ public void setAllocatedPorts(Map<String, String> ports) {
+ this.allocatedPorts = ports;
+ }
+
@Override
public String toString() {
String ret = "responseId=" + responseId + "\n" +
"timestamp=" + timestamp + "\n" +
- "hostname=" + hostname + "\n" +
- "currentPingPort=" + currentPingPort + "\n";
+ "hostname=" + hostname + "\n" +
+ "expectedState=" + expectedState + "\n" +
+ "actualState=" + actualState + "\n";
- if (hardwareProfile != null)
+ if (hardwareProfile != null) {
ret = ret + "hardwareprofile=" + this.hardwareProfile.toString();
+ }
return ret;
}
}
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/47822dff/slider-core/src/test/java/org/apache/slider/providers/agent/TestAgentProviderService.java
----------------------------------------------------------------------
diff --git a/slider-core/src/test/java/org/apache/slider/providers/agent/TestAgentProviderService.java b/slider-core/src/test/java/org/apache/slider/providers/agent/TestAgentProviderService.java
index 1aeedbc..20c16ff 100644
--- a/slider-core/src/test/java/org/apache/slider/providers/agent/TestAgentProviderService.java
+++ b/slider-core/src/test/java/org/apache/slider/providers/agent/TestAgentProviderService.java
@@ -87,6 +87,7 @@ import static org.easymock.EasyMock.replay;
import static org.junit.Assert.assertEquals;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anyCollection;
+import static org.mockito.Matchers.anyMap;
import static org.mockito.Matchers.anyString;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.doReturn;
@@ -271,6 +272,12 @@ public class TestAgentProviderService {
} catch (SliderException e) {
}
+ doNothing().when(mockAps).processAllocatedPorts(
+ anyString(),
+ anyString(),
+ anyString(),
+ anyMap()
+ );
expect(access.isApplicationLive()).andReturn(true).anyTimes();
ClusterDescription desc = new ClusterDescription();
desc.setOption(OptionKeys.ZOOKEEPER_QUORUM, "host1:2181");
@@ -305,10 +312,20 @@ public class TestAgentProviderService {
Register reg = new Register();
reg.setResponseId(0);
reg.setHostname("mockcontainer_1___HBASE_MASTER");
+ Map<String,String> ports = new HashMap();
+ ports.put("a","100");
+ reg.setAllocatedPorts(ports);
RegistrationResponse resp = mockAps.handleRegistration(reg);
Assert.assertEquals(0, resp.getResponseId());
Assert.assertEquals(RegistrationStatus.OK, resp.getResponseStatus());
+ Mockito.verify(mockAps, Mockito.times(1)).processAllocatedPorts(
+ anyString(),
+ anyString(),
+ anyString(),
+ anyMap()
+ );
+
HeartBeat hb = new HeartBeat();
hb.setResponseId(1);
hb.setHostname("mockcontainer_1___HBASE_MASTER");