-
Notifications
You must be signed in to change notification settings - Fork 7
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
LIU-66: Download logs from all running NM as a tar archive. #46
Changes from 1 commit
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -24,23 +24,29 @@ | |
Data Managers (DROPManager and DataIslandManager) to the outside world. | ||
""" | ||
|
||
import io | ||
import os | ||
import cgi | ||
import functools | ||
import json | ||
import logging | ||
import threading | ||
import tarfile | ||
|
||
import bottle | ||
import pkg_resources | ||
|
||
from bottle import static_file | ||
|
||
from . import constants | ||
from .client import NodeManagerClient | ||
from .client import NodeManagerClient, DataIslandManagerClient | ||
from .. import utils | ||
from ..exceptions import InvalidGraphException, InvalidSessionState, \ | ||
DaliugeException, NoSessionException, SessionAlreadyExistsException, \ | ||
InvalidDropException, InvalidRelationshipException, SubManagerException | ||
from ..restserver import RestServer | ||
from ..restutils import RestClient, RestClientException | ||
|
||
from .session import generateLogFileName | ||
|
||
logger = logging.getLogger(__name__) | ||
|
||
|
@@ -54,6 +60,13 @@ def daliuge_aware(func): | |
def fwrapper(*args, **kwargs): | ||
try: | ||
res = func(*args, **kwargs) | ||
|
||
if isinstance(res, bytes): | ||
return res | ||
|
||
if isinstance(res, bottle.HTTPResponse): | ||
return res | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
|
||
|
||
if res is not None: | ||
bottle.response.content_type = 'application/json' | ||
return json.dumps(res) | ||
|
@@ -120,6 +133,7 @@ def __init__(self, dm, maxreqsize=10): | |
app.get( '/api/sessions', callback=self.getSessions) | ||
app.get( '/api/sessions/<sessionId>', callback=self.getSessionInformation) | ||
app.delete('/api/sessions/<sessionId>', callback=self.destroySession) | ||
app.get( '/api/sessions/<sessionId>/logs', callback=self.getLogFile) | ||
app.get( '/api/sessions/<sessionId>/status', callback=self.getSessionStatus) | ||
app.post( '/api/sessions/<sessionId>/deploy', callback=self.deploySession) | ||
app.post( '/api/sessions/<sessionId>/cancel', callback=self.cancelSession) | ||
|
@@ -269,6 +283,14 @@ def getNMStatus(self): | |
# future | ||
return {'sessions': self.sessions()} | ||
|
||
@daliuge_aware | ||
def getLogFile(self, sessionId): | ||
logdir = self.dm.getLogDir() | ||
logfile = generateLogFileName(logdir, sessionId) | ||
if not os.path.isfile(logfile): | ||
raise NoSessionException(sessionId, 'Log file not found.') | ||
return static_file(os.path.basename(logfile), root=logdir, download=os.path.basename(logfile)) | ||
|
||
@daliuge_aware | ||
def linkGraphParts(self, sessionId): | ||
params = bottle.request.params | ||
|
@@ -333,6 +355,9 @@ def getCMStatus(self): | |
def getCMNodes(self): | ||
return self.dm.nodes | ||
|
||
def getAllCMNodes(self): | ||
return self.dm.nodes | ||
|
||
@daliuge_aware | ||
def addCMNode(self, node): | ||
self.dm.add_node(node) | ||
|
@@ -348,6 +373,44 @@ def getNodeSessions(self, node): | |
with NodeManagerClient(host=node) as dm: | ||
return dm.sessions() | ||
|
||
def _tarfile_write(self, tar, headers, stream): | ||
file_header = headers.getheader('Content-Disposition') | ||
length = headers.getheader('Content-Length') | ||
_, params = cgi.parse_header(file_header) | ||
filename = params['filename'] | ||
info = tarfile.TarInfo(filename) | ||
info.size = int(length) | ||
|
||
content = [] | ||
while True: | ||
buffer = stream.read() | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Wouldn't this There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Because its a tcp stream there is no guarantee that a single read will get all the data unless there is code underneath this library that is accumulating? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I think ill leave it for now and we can review again. |
||
if not buffer: | ||
break | ||
content.append(buffer) | ||
|
||
tar.addfile(info, io.BytesIO(initial_bytes=''.join(content).encode())) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Mmmm.... now that I read this: wouldn't it be possible to simply pass |
||
|
||
@daliuge_aware | ||
def getLogFile(self, sessionId): | ||
fh = io.BytesIO() | ||
with tarfile.open(fileobj=fh, mode='w:gz') as tar: | ||
for node in self.getAllCMNodes(): | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Maybe add a TODO here to point out that in the future we might want to fetch in this loop concurrently instead of sequentially. |
||
with NodeManagerClient(host=node) as dm: | ||
try: | ||
stream, resp = dm.get_log_file(sessionId) | ||
self._tarfile_write(tar, resp, stream) | ||
except NoSessionException: | ||
pass | ||
|
||
|
||
data = fh.getvalue() | ||
size = len(data) | ||
bottle.response.set_header('Content-type', 'application/x-tar') | ||
bottle.response['Content-Disposition'] = f'attachment; ' \ | ||
f'filename=dlg_{sessionId}.tar' | ||
bottle.response['Content-Length'] = size | ||
return data | ||
|
||
@daliuge_aware | ||
def getNodeSessionInformation(self, node, sessionId): | ||
if node not in self.dm.nodes: | ||
|
@@ -404,4 +467,11 @@ def initializeSpecifics(self, app): | |
def createDataIsland(self, host): | ||
with RestClient(host=host, port=constants.DAEMON_DEFAULT_REST_PORT, timeout=10) as c: | ||
c._post_json('/managers/dataisland', bottle.request.body.read()) | ||
self.dm.addDmHost(host) | ||
self.dm.addDmHost(host) | ||
|
||
def getAllCMNodes(self): | ||
nodes = [] | ||
for node in self.dm.dmHosts: | ||
with DataIslandManagerClient(host=node) as dm: | ||
nodes += dm.nodes() | ||
return nodes |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This is more of a personal preference, but could we have imports alphasorted in general? It would be nicer to have this checked automatically by a tool instead of me telling it, we could do that at some point.