Commit 64cf4e19 authored by Valentin Reis's avatar Valentin Reis

Merge branch 'cmd-listen-app_uuid' into 'master'

[refactor] messaging code style + cmd_listen application_uuid relaxing

See merge request !48
parents bfbddc9b 1120d6d5
Pipeline #5000 passed with stages
in 6 minutes and 13 seconds
...@@ -3,5 +3,5 @@ let argotest = ...@@ -3,5 +3,5 @@ let argotest =
"https://xgitlab.cels.anl.gov/argo/argotest/-/archive/master/argotest-master.tar.gz"; "https://xgitlab.cels.anl.gov/argo/argotest/-/archive/master/argotest-master.tar.gz";
in import "${argotest}/test.nix" { in import "${argotest}/test.nix" {
nrm-override = ./..; nrm-override = ./..;
testName = "refactored"; testName = "cmd-uuid";
} }
...@@ -65,19 +65,29 @@ class CommandLineInterface(object): ...@@ -65,19 +65,29 @@ class CommandLineInterface(object):
def print_if_filter(): def print_if_filter():
if argv.filter: if argv.filter:
if argv.filter == msg.type: if argv.filter == msg.type:
if (msg.type == "performance" or
msg.type == "progress"):
print("%s, %s, %s" % (msg.type, time.time(), print("%s, %s, %s" % (msg.type, time.time(),
msg.payload)) msg.payload))
if msg.type == "power":
print("%s, %s, %s" % (msg.type, time.time(),
msg.total))
sys.stdout.flush()
if msg.type == "container_exit":
print("%s, %s, %s" % (msg.type, time.time(),
msg.profile_data))
sys.stdout.flush() sys.stdout.flush()
else: else:
print("%s, %s" % (msg.type, time.time())) print("%s, %s" % (msg.type, time.time()))
sys.stdout.flush() sys.stdout.flush()
if argv.uuid:
uuid = getattr(msg, 'container_uuid', None)
if argv.uuid == uuid or msg.type == "power":
print_if_filter()
else:
print_if_filter() print_if_filter()
# if argv.uuid:
# uuid = getattr(msg, 'container_uuid', None)
# if argv.uuid == uuid or msg.type == "power":
# print_if_filter()
# else:
# print_if_filter()
def do_run(self, argv): def do_run(self, argv):
""" Connect to the NRM and ask to spawn a container and run a command """ Connect to the NRM and ask to spawn a container and run a command
......
...@@ -41,7 +41,7 @@ class Daemon(object): ...@@ -41,7 +41,7 @@ class Daemon(object):
pub = {'api': 'up_pub', pub = {'api': 'up_pub',
'type': 'progress', 'type': 'progress',
'payload': msg.payload, 'payload': msg.payload,
'container_uuid': msg.container_uuid} 'application_uuid': msg.application_uuid}
self.upstream_pub_server.sendmsg( self.upstream_pub_server.sendmsg(
PUB_MSG['progress'](**pub)) PUB_MSG['progress'](**pub))
elif msg.type == 'performance': elif msg.type == 'performance':
......
...@@ -20,53 +20,98 @@ logger = logging.getLogger('nrm') ...@@ -20,53 +20,98 @@ logger = logging.getLogger('nrm')
# basic field type information. # basic field type information.
APIS = ['up_rpc_req', 'up_rpc_rep', 'up_pub', 'down_event'] APIS = ['up_rpc_req', 'up_rpc_rep', 'up_pub', 'down_event']
MSGFORMATS = {k: {} for k in APIS} MSGFORMATS = {k: {} for k in APIS}
MSGFORMATS['up_rpc_req'] = {'list': {},
'run': {'manifest': basestring, MSGFORMATS['up_rpc_req'] = {
'list': {},
'run': {
'manifest': basestring,
'path': basestring, 'path': basestring,
'args': list, 'args': list,
'container_uuid': basestring, 'container_uuid': basestring,
'environ': dict}, 'environ': dict
'kill': {'container_uuid': basestring}, },
'setpower': {'limit': basestring}, 'kill': {
'container_uuid': basestring
},
'setpower': {
'limit': basestring
}
}
MSGFORMATS['up_rpc_rep'] = {
'list': {
'payload': list
},
'stdout': {
'container_uuid': basestring,
'payload': basestring
},
'stderr': {
'container_uuid': basestring,
'payload': basestring
},
'process_start': {
'container_uuid': basestring,
'pid': int
},
'process_exit': {
'container_uuid': basestring,
'status': basestring
},
'getpower': {
'limit': basestring
} }
MSGFORMATS['up_rpc_rep'] = {'list': {'payload': list},
'stdout': {'container_uuid': basestring,
'payload': basestring},
'stderr': {'container_uuid': basestring,
'payload': basestring},
'process_start': {'container_uuid': basestring,
'pid': int},
'process_exit': {'container_uuid': basestring,
'status': basestring},
'getpower': {'limit': basestring},
} }
MSGFORMATS['up_pub'] = {'power': {'total': float, 'limit': float},
'container_start': {'container_uuid': basestring, MSGFORMATS['up_pub'] = {
'power': {
'total': float,
'limit': float
},
'container_start': {
'container_uuid': basestring,
'errno': int, 'errno': int,
'power': basestring}, 'power': dict
'container_exit': {'container_uuid': basestring, },
'profile_data': dict}, 'container_exit': {
'performance': {'container_uuid': basestring, 'container_uuid': basestring,
'payload': int}, 'profile_data': dict
'progress': {'application_uuid': basestring, },
'payload': int}, 'performance': {
'container_uuid': basestring,
'payload': int
},
'progress': {
'application_uuid': basestring,
'payload': int
} }
MSGFORMATS['down_event'] = {'application_start': }
{'container_uuid': basestring,
'application_uuid': basestring}, MSGFORMATS['down_event'] = {
'application_exit': 'application_start': {
{'application_uuid': basestring}, 'container_uuid': basestring,
'performance': {'payload': int, 'application_uuid': basestring
},
'application_exit': {
'application_uuid': basestring
},
'performance': {
'container_uuid': basestring,
'application_uuid': basestring, 'application_uuid': basestring,
'container_uuid': basestring}, 'payload': int,
'progress': {'payload': int, },
'progress': {
'application_uuid': basestring, 'application_uuid': basestring,
'container_uuid': basestring}, 'container_uuid': basestring,
'phase_context': {'cpu': int, 'payload': int
'aggregation': int, },
'computetime': int, 'phase_context': {
'totaltime': int, 'cpu': int,
'application_uuid': basestring}, 'startcompute': int,
'endcompute': int,
'startbarrier': int,
'endbarrier': int
}
} }
# Mirror of the message formats, using namedtuples as the actual transport # Mirror of the message formats, using namedtuples as the actual transport
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment