mirror of
https://github.com/avatao-content/baseimage-tutorial-framework
synced 2025-06-28 23:55:13 +00:00
Separate event handlers from independent components
This commit is contained in:
R. Richard
Kristóf Tóth
@ -1,18 +1,9 @@
|
||||
# Copyright (C) 2018 Avatao.com Innovative Learning Kft.
|
||||
# All Rights Reserved. See LICENSE file for details.
|
||||
|
||||
from .directory_monitoring_event_handler import DirectoryMonitoringEventHandler
|
||||
from .process_managing_event_handler import ProcessManagingEventHandler
|
||||
from .terminal_event_handler import TerminalEventHandler
|
||||
from .ide_event_handler import IdeEventHandler
|
||||
from .history_monitor import HistoryMonitor, BashMonitor, GDBMonitor
|
||||
from .terminal_commands import TerminalCommands
|
||||
from .log_monitoring_event_handler import LogMonitoringEventHandler
|
||||
from .fsm_managing_event_handler import FSMManagingEventHandler
|
||||
from .snapshot_provider import SnapshotProvider
|
||||
from .pipe_io_event_handler import PipeIOEventHandlerBase, PipeIOEventHandler, PipeIOServer
|
||||
from .pipe_io_event_handler import TransformerPipeIOEventHandler, CommandEventHandler
|
||||
from .directory_snapshotting_event_handler import DirectorySnapshottingEventHandler
|
||||
from .commands_equal import CommandsEqual
|
||||
from .frontend_event_handler import FrontendEventHandler
|
||||
from .file_manager import FileManager
|
||||
from .history_monitor import HistoryMonitor, BashMonitor, GDBMonitor
|
||||
from .message_sender import MessageSender
|
||||
from .snapshot_provider import SnapshotProvider
|
||||
from .terminal_commands import TerminalCommands
|
||||
|
||||
@ -1,70 +0,0 @@
|
||||
# Copyright (C) 2018 Avatao.com Innovative Learning Kft.
|
||||
# All Rights Reserved. See LICENSE file for details.
|
||||
|
||||
import logging
|
||||
from os.path import isdir, exists
|
||||
|
||||
from tfw.event_handlers import FrontendEventHandlerBase
|
||||
from tfw.mixins.monitor_manager_mixin import MonitorManagerMixin
|
||||
from tfw.components.directory_monitor import DirectoryMonitor
|
||||
|
||||
LOG = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class DirectoryMonitoringEventHandler(FrontendEventHandlerBase, MonitorManagerMixin):
|
||||
def __init__(self, key, directory):
|
||||
super().__init__(key)
|
||||
self._directory = directory
|
||||
MonitorManagerMixin.__init__(
|
||||
self,
|
||||
DirectoryMonitor,
|
||||
key,
|
||||
self._directory
|
||||
)
|
||||
|
||||
self.commands = {
|
||||
'pause': self.pause,
|
||||
'resume': self.resume,
|
||||
'ignore': self.ignore,
|
||||
'selectdir': self.selectdir
|
||||
}
|
||||
|
||||
@property
|
||||
def directory(self):
|
||||
return self._directory
|
||||
|
||||
@directory.setter
|
||||
def directory(self, directory):
|
||||
if not exists(directory) or not isdir(directory):
|
||||
raise EnvironmentError('No such directory!')
|
||||
self._directory = directory
|
||||
|
||||
def handle_event(self, message):
|
||||
try:
|
||||
message['data'] = self.commands[message['data']['command']](message['data'])
|
||||
return message
|
||||
except KeyError:
|
||||
LOG.error('IGNORING MESSAGE: Invalid message received: %s', message)
|
||||
|
||||
def pause(self, data):
|
||||
self.monitor.pause()
|
||||
return data
|
||||
|
||||
def resume(self, data):
|
||||
self.monitor.resume()
|
||||
return data
|
||||
|
||||
def ignore(self, data):
|
||||
self.monitor.ignore += data['ignore']
|
||||
return data
|
||||
|
||||
def selectdir(self, data):
|
||||
try:
|
||||
self.directory = data['directory']
|
||||
self.reload_monitor()
|
||||
return data
|
||||
except EnvironmentError:
|
||||
LOG.error('Failed to switch directory!')
|
||||
|
||||
def cleanup(self):
|
||||
self.monitor.stop()
|
||||
@ -1,87 +0,0 @@
|
||||
# Copyright (C) 2018 Avatao.com Innovative Learning Kft.
|
||||
# All Rights Reserved. See LICENSE file for details.
|
||||
|
||||
import logging
|
||||
from os.path import join as joinpath
|
||||
from os.path import basename
|
||||
from os import makedirs
|
||||
from datetime import datetime
|
||||
|
||||
from dateutil import parser as dateparser
|
||||
|
||||
from tfw.event_handlers import FrontendEventHandlerBase
|
||||
from tfw.components.snapshot_provider import SnapshotProvider
|
||||
from tfw.config import TFWENV
|
||||
|
||||
LOG = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class DirectorySnapshottingEventHandler(FrontendEventHandlerBase):
|
||||
def __init__(self, key, directories, exclude_unix_patterns=None):
|
||||
super().__init__(key)
|
||||
self.snapshot_providers = {}
|
||||
self._exclude_unix_patterns = exclude_unix_patterns
|
||||
self.init_snapshot_providers(directories)
|
||||
|
||||
self.command_handlers = {
|
||||
'take_snapshot': self.handle_take_snapshot,
|
||||
'restore_snapshot': self.handle_restore_snapshot,
|
||||
'exclude': self.handle_exclude
|
||||
}
|
||||
|
||||
def init_snapshot_providers(self, directories):
|
||||
for index, directory in enumerate(directories):
|
||||
git_dir = self.init_git_dir(index, directory)
|
||||
self.snapshot_providers[directory] = SnapshotProvider(
|
||||
directory,
|
||||
git_dir,
|
||||
self._exclude_unix_patterns
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def init_git_dir(index, directory):
|
||||
git_dir = joinpath(
|
||||
TFWENV.SNAPSHOTS_DIR,
|
||||
f'{basename(directory)}-{index}'
|
||||
)
|
||||
makedirs(git_dir, exist_ok=True)
|
||||
return git_dir
|
||||
|
||||
def handle_event(self, message):
|
||||
try:
|
||||
data = message['data']
|
||||
message['data'] = self.command_handlers[data['command']](data)
|
||||
return message
|
||||
except KeyError:
|
||||
LOG.error('IGNORING MESSAGE: Invalid message received: %s', message)
|
||||
|
||||
def handle_take_snapshot(self, data):
|
||||
LOG.debug('Taking snapshots of directories %s', self.snapshot_providers.keys())
|
||||
for provider in self.snapshot_providers.values():
|
||||
provider.take_snapshot()
|
||||
return data
|
||||
|
||||
def handle_restore_snapshot(self, data):
|
||||
date = dateparser.parse(
|
||||
data.get(
|
||||
'value',
|
||||
datetime.now().isoformat()
|
||||
)
|
||||
)
|
||||
LOG.debug(
|
||||
'Restoring snapshots (@ %s) of directories %s',
|
||||
date,
|
||||
self.snapshot_providers.keys()
|
||||
)
|
||||
for provider in self.snapshot_providers.values():
|
||||
provider.restore_snapshot(date)
|
||||
return data
|
||||
|
||||
def handle_exclude(self, data):
|
||||
exclude_unix_patterns = data['value']
|
||||
if not isinstance(exclude_unix_patterns, list):
|
||||
raise KeyError
|
||||
|
||||
for provider in self.snapshot_providers.values():
|
||||
provider.exclude = exclude_unix_patterns
|
||||
return data
|
||||
1
lib/tfw/components/file_manager/__init__.py
Normal file
1
lib/tfw/components/file_manager/__init__.py
Normal file
@ -0,0 +1 @@
|
||||
from .file_manager import FileManager
|
||||
@ -1,63 +0,0 @@
|
||||
from abc import ABC, abstractmethod
|
||||
from contextlib import suppress
|
||||
|
||||
from tfw.event_handlers import FrontendEventHandlerBase
|
||||
|
||||
from .message_sender import MessageSender
|
||||
|
||||
|
||||
class FrontendEventHandler(FrontendEventHandlerBase):
|
||||
def __init__(self):
|
||||
frontend_keys = ('message', 'queueMessages', 'dashboard', 'console')
|
||||
self._frontend_message_storage = FrontendMessageStorage(frontend_keys)
|
||||
super().__init__((*frontend_keys, 'recover'))
|
||||
|
||||
def handle_event(self, message):
|
||||
self._frontend_message_storage.save_message(message)
|
||||
if message['key'] == 'recover':
|
||||
self.recover_frontend()
|
||||
return message
|
||||
|
||||
def recover_frontend(self):
|
||||
for message in self._frontend_message_storage.messages:
|
||||
self.send_message(message)
|
||||
|
||||
|
||||
class MessageStorage(ABC):
|
||||
def __init__(self):
|
||||
self._messages = []
|
||||
|
||||
def save_message(self, message):
|
||||
with suppress(KeyError, AttributeError):
|
||||
if self._filter_message(message):
|
||||
self._messages.extend(self._transform_message(message))
|
||||
|
||||
@abstractmethod
|
||||
def _filter_message(self, message):
|
||||
raise NotImplementedError
|
||||
|
||||
def _transform_message(self, message): # pylint: disable=no-self-use
|
||||
yield message
|
||||
|
||||
def clear(self):
|
||||
self._messages.clear()
|
||||
|
||||
@property
|
||||
def messages(self):
|
||||
yield from self._messages
|
||||
|
||||
|
||||
class FrontendMessageStorage(MessageStorage):
|
||||
def __init__(self, keys):
|
||||
self._keys = keys
|
||||
super().__init__()
|
||||
|
||||
def _filter_message(self, message):
|
||||
key = message['key']
|
||||
return key in self._keys
|
||||
|
||||
def _transform_message(self, message):
|
||||
if message['key'] == 'queueMessages':
|
||||
yield from MessageSender.generate_messages_from_queue(message)
|
||||
else:
|
||||
yield message
|
||||
@ -1,100 +0,0 @@
|
||||
# Copyright (C) 2018 Avatao.com Innovative Learning Kft.
|
||||
# All Rights Reserved. See LICENSE file for details.
|
||||
|
||||
import logging
|
||||
|
||||
from tfw.event_handlers import FrontendEventHandlerBase
|
||||
from tfw.crypto import KeyManager, sign_message, verify_message
|
||||
from tfw.networking import Scope
|
||||
|
||||
LOG = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class FSMManagingEventHandler(FrontendEventHandlerBase):
|
||||
"""
|
||||
EventHandler responsible for managing the state machine of
|
||||
the framework (TFW FSM).
|
||||
|
||||
tfw.networking.TFWServer instances automatically send 'trigger'
|
||||
commands to the event handler listening on the 'fsm' key,
|
||||
which should be an instance of this event handler.
|
||||
|
||||
This event handler accepts messages that have a
|
||||
data['command'] key specifying a command to be executed.
|
||||
|
||||
An 'fsm_update' message is broadcasted after every successful
|
||||
command.
|
||||
"""
|
||||
def __init__(self, key, fsm_type, require_signature=False):
|
||||
super().__init__(key)
|
||||
self.fsm = fsm_type()
|
||||
self._fsm_updater = FSMUpdater(self.fsm)
|
||||
self.auth_key = KeyManager().auth_key
|
||||
self._require_signature = require_signature
|
||||
|
||||
self.command_handlers = {
|
||||
'trigger': self.handle_trigger,
|
||||
'update': self.handle_update
|
||||
}
|
||||
|
||||
def handle_event(self, message):
|
||||
try:
|
||||
message = self.command_handlers[message['data']['command']](message)
|
||||
if message:
|
||||
fsm_update_message = self._fsm_updater.fsm_update
|
||||
sign_message(self.auth_key, message)
|
||||
sign_message(self.auth_key, fsm_update_message)
|
||||
self.server_connector.send_message(fsm_update_message, Scope.BROADCAST)
|
||||
return message
|
||||
except KeyError:
|
||||
LOG.error('IGNORING MESSAGE: Invalid message received: %s', message)
|
||||
|
||||
def handle_trigger(self, message):
|
||||
"""
|
||||
Attempts to step the FSM with the supplied trigger.
|
||||
|
||||
:param message: TFW message with a data field containing
|
||||
the action to try triggering in data['value']
|
||||
"""
|
||||
trigger = message['data']['value']
|
||||
if self._require_signature:
|
||||
if not verify_message(self.auth_key, message):
|
||||
LOG.error('Ignoring unsigned trigger command: %s', message)
|
||||
return None
|
||||
if self.fsm.step(trigger):
|
||||
return message
|
||||
return None
|
||||
|
||||
def handle_update(self, message):
|
||||
"""
|
||||
Does nothing, but triggers an 'fsm_update' message.
|
||||
"""
|
||||
# pylint: disable=no-self-use
|
||||
return message
|
||||
|
||||
|
||||
class FSMUpdater:
|
||||
def __init__(self, fsm):
|
||||
self.fsm = fsm
|
||||
|
||||
@property
|
||||
def fsm_update(self):
|
||||
return {
|
||||
'key': 'fsm_update',
|
||||
'data': self.fsm_update_data
|
||||
}
|
||||
|
||||
@property
|
||||
def fsm_update_data(self):
|
||||
valid_transitions = [
|
||||
{'trigger': trigger}
|
||||
for trigger in self.fsm.get_triggers(self.fsm.state)
|
||||
]
|
||||
last_fsm_event = self.fsm.event_log[-1]
|
||||
last_fsm_event['timestamp'] = last_fsm_event['timestamp'].isoformat()
|
||||
return {
|
||||
'current_state': self.fsm.state,
|
||||
'valid_transitions': valid_transitions,
|
||||
'in_accepted_state': self.fsm.in_accepted_state,
|
||||
'last_event': last_fsm_event
|
||||
}
|
||||
@ -1 +0,0 @@
|
||||
from .ide_event_handler import IdeEventHandler
|
||||
@ -1,169 +0,0 @@
|
||||
# Copyright (C) 2018 Avatao.com Innovative Learning Kft.
|
||||
# All Rights Reserved. See LICENSE file for details.
|
||||
|
||||
import logging
|
||||
from os.path import isfile, join, relpath, exists, isdir, realpath
|
||||
from glob import glob
|
||||
from fnmatch import fnmatchcase
|
||||
from typing import Iterable
|
||||
|
||||
from tfw.event_handlers import FrontendEventHandlerBase
|
||||
from tfw.mixins.monitor_manager_mixin import MonitorManagerMixin
|
||||
from tfw.components.directory_monitor import DirectoryMonitor
|
||||
|
||||
from .file_manager import FileManager
|
||||
|
||||
LOG = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class IdeEventHandler(FrontendEventHandlerBase, MonitorManagerMixin):
|
||||
# pylint: disable=too-many-arguments,anomalous-backslash-in-string
|
||||
"""
|
||||
Event handler implementing the backend of our browser based IDE.
|
||||
By default all files in the directory specified in __init__ are displayed
|
||||
on the fontend. Note that this is a stateful component.
|
||||
|
||||
When any file in the selected directory changes they are automatically refreshed
|
||||
on the frontend (this is done by listening to inotify events).
|
||||
|
||||
This EventHandler accepts messages that have a data['command'] key specifying
|
||||
a command to be executed.
|
||||
|
||||
The API of each command is documented in their respective handler.
|
||||
"""
|
||||
def __init__(self, key, directory, allowed_directories, selected_file=None, exclude=None):
|
||||
"""
|
||||
:param key: the key this instance should listen to
|
||||
:param directory: working directory which the EventHandler should serve files from
|
||||
:param allowed_directories: list of directories that can be switched to using selectdir
|
||||
:param selected_file: file that is selected by default
|
||||
:param exclude: list of filenames that should not appear between files (for .o, .pyc, etc.)
|
||||
"""
|
||||
super().__init__(key)
|
||||
try:
|
||||
self.filemanager = FileManager(
|
||||
allowed_directories=allowed_directories,
|
||||
working_directory=directory,
|
||||
selected_file=selected_file,
|
||||
exclude=exclude
|
||||
)
|
||||
except IndexError:
|
||||
raise EnvironmentError(
|
||||
f'No file(s) in IdeEventHandler working_directory "{directory}"!'
|
||||
)
|
||||
|
||||
MonitorManagerMixin.__init__(
|
||||
self,
|
||||
DirectoryMonitor,
|
||||
self.key,
|
||||
self.filemanager.allowed_directories
|
||||
)
|
||||
|
||||
self.commands = {
|
||||
'read': self.read,
|
||||
'write': self.write,
|
||||
'select': self.select,
|
||||
'selectdir': self.select_dir,
|
||||
'exclude': self.exclude
|
||||
}
|
||||
|
||||
def read(self, data):
|
||||
"""
|
||||
Read the currently selected file.
|
||||
|
||||
:return dict: TFW message data containing key 'content'
|
||||
(contents of the selected file)
|
||||
"""
|
||||
try:
|
||||
data['content'] = self.filemanager.file_contents
|
||||
except PermissionError:
|
||||
data['content'] = 'You have no permission to open that file :('
|
||||
except FileNotFoundError:
|
||||
data['content'] = 'This file was removed :('
|
||||
except Exception: # pylint: disable=broad-except
|
||||
data['content'] = 'Failed to read file :('
|
||||
return data
|
||||
|
||||
def write(self, data):
|
||||
"""
|
||||
Overwrites a file with the desired string.
|
||||
|
||||
:param data: TFW message data containing key 'content'
|
||||
(new file content)
|
||||
|
||||
"""
|
||||
self.monitor.ignore = self.monitor.ignore + 1
|
||||
try:
|
||||
self.filemanager.file_contents = data['content']
|
||||
except Exception: # pylint: disable=broad-except
|
||||
LOG.exception('Error writing file!')
|
||||
del data['content']
|
||||
return data
|
||||
|
||||
def select(self, data):
|
||||
"""
|
||||
Selects a file from the current directory.
|
||||
|
||||
:param data: TFW message data containing 'filename'
|
||||
(name of file to select relative to the current directory)
|
||||
"""
|
||||
try:
|
||||
self.filemanager.filename = data['filename']
|
||||
except EnvironmentError:
|
||||
LOG.exception('Failed to select file "%s"', data['filename'])
|
||||
return data
|
||||
|
||||
def select_dir(self, data):
|
||||
"""
|
||||
Select a new working directory to display files from.
|
||||
|
||||
:param data: TFW message data containing 'directory'
|
||||
(absolute path of diretory to select.
|
||||
must be a path whitelisted in
|
||||
self.allowed_directories)
|
||||
"""
|
||||
try:
|
||||
self.filemanager.workdir = data['directory']
|
||||
self.reload_monitor()
|
||||
try:
|
||||
self.filemanager.filename = self.filemanager.files[0]
|
||||
self.read(data)
|
||||
except IndexError:
|
||||
data['content'] = 'No files in this directory :('
|
||||
except EnvironmentError as err:
|
||||
LOG.error('Failed to select directory "%s". Reason: %s', data['directory'], str(err))
|
||||
return data
|
||||
|
||||
def exclude(self, data):
|
||||
"""
|
||||
Overwrite list of excluded files
|
||||
|
||||
:param data: TFW message data containing 'exclude'
|
||||
(list of unix-style filename patterns to be excluded,
|
||||
e.g.: ["\*.pyc", "\*.o")
|
||||
"""
|
||||
try:
|
||||
self.filemanager.exclude = list(data['exclude'])
|
||||
except TypeError:
|
||||
LOG.error('Exclude must be Iterable!')
|
||||
return data
|
||||
|
||||
def attach_fileinfo(self, data):
|
||||
"""
|
||||
Basic information included in every response to the frontend.
|
||||
"""
|
||||
data['filename'] = self.filemanager.filename
|
||||
data['files'] = self.filemanager.files
|
||||
data['directory'] = self.filemanager.workdir
|
||||
|
||||
def handle_event(self, message):
|
||||
try:
|
||||
data = message['data']
|
||||
message['data'] = self.commands[data['command']](data)
|
||||
self.attach_fileinfo(data)
|
||||
return message
|
||||
except KeyError:
|
||||
LOG.error('IGNORING MESSAGE: Invalid message received: %s', message)
|
||||
|
||||
def cleanup(self):
|
||||
self.monitor.stop()
|
||||
@ -1,68 +0,0 @@
|
||||
# Copyright (C) 2018 Avatao.com Innovative Learning Kft.
|
||||
# All Rights Reserved. See LICENSE file for details.
|
||||
|
||||
import logging
|
||||
|
||||
from tfw.event_handlers import FrontendEventHandlerBase
|
||||
from tfw.mixins.monitor_manager_mixin import MonitorManagerMixin
|
||||
from tfw.components.log_monitor import LogMonitor
|
||||
|
||||
LOG = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class LogMonitoringEventHandler(FrontendEventHandlerBase, MonitorManagerMixin):
|
||||
"""
|
||||
Monitors the output of a supervisor process (stdout, stderr) and
|
||||
sends the results to the frontend.
|
||||
|
||||
Accepts messages that have a data['command'] key specifying
|
||||
a command to be executed.
|
||||
|
||||
The API of each command is documented in their respective handler.
|
||||
"""
|
||||
def __init__(self, key, process_name, log_tail=0):
|
||||
super().__init__(key)
|
||||
self.process_name = process_name
|
||||
self.log_tail = log_tail
|
||||
MonitorManagerMixin.__init__(
|
||||
self,
|
||||
LogMonitor,
|
||||
self.process_name,
|
||||
self.log_tail
|
||||
)
|
||||
|
||||
self.command_handlers = {
|
||||
'process_name': self.handle_process_name,
|
||||
'log_tail': self.handle_log_tail
|
||||
}
|
||||
|
||||
def handle_event(self, message):
|
||||
try:
|
||||
data = message['data']
|
||||
self.command_handlers[data['command']](data)
|
||||
self.reload_monitor()
|
||||
except KeyError:
|
||||
LOG.error('IGNORING MESSAGE: Invalid message received: %s', message)
|
||||
|
||||
def handle_process_name(self, data):
|
||||
"""
|
||||
Changes the monitored process.
|
||||
|
||||
:param data: TFW message data containing 'value'
|
||||
(name of the process to monitor)
|
||||
"""
|
||||
self.set_monitor_args(data['value'], self.log_tail)
|
||||
|
||||
def handle_log_tail(self, data):
|
||||
"""
|
||||
Sets tail length of the log the monitor will send
|
||||
to the frontend (the monitor will send back the last
|
||||
'value' characters of the log).
|
||||
|
||||
:param data: TFW message data containing 'value'
|
||||
(new tail length)
|
||||
"""
|
||||
self.set_monitor_args(self.process_name, data['value'])
|
||||
|
||||
def cleanup(self):
|
||||
self.monitor.stop()
|
||||
@ -1,143 +0,0 @@
|
||||
import logging
|
||||
from abc import abstractmethod
|
||||
from json import loads, dumps
|
||||
from subprocess import run, PIPE, Popen
|
||||
from functools import partial
|
||||
from os import getpgid, killpg
|
||||
from os.path import join
|
||||
from signal import SIGTERM
|
||||
from secrets import token_urlsafe
|
||||
from threading import Thread
|
||||
from contextlib import suppress
|
||||
|
||||
from tfw.event_handlers import EventHandlerBase
|
||||
|
||||
from .pipe_io_server import PipeIOServer, terminate_process_on_failure
|
||||
|
||||
LOG = logging.getLogger(__name__)
|
||||
DEFAULT_PERMISSIONS = 0o600
|
||||
|
||||
|
||||
class PipeIOEventHandlerBase(EventHandlerBase):
|
||||
def __init__(self, key, in_pipe_path, out_pipe_path, permissions=DEFAULT_PERMISSIONS):
|
||||
super().__init__(key)
|
||||
self.pipe_io = CallbackPipeIOServer(
|
||||
in_pipe_path,
|
||||
out_pipe_path,
|
||||
self.handle_pipe_event,
|
||||
permissions
|
||||
)
|
||||
self.pipe_io.start()
|
||||
|
||||
@abstractmethod
|
||||
def handle_pipe_event(self, message_bytes):
|
||||
raise NotImplementedError()
|
||||
|
||||
def cleanup(self):
|
||||
self.pipe_io.stop()
|
||||
|
||||
|
||||
class CallbackPipeIOServer(PipeIOServer):
|
||||
def __init__(self, in_pipe_path, out_pipe_path, callback, permissions):
|
||||
super().__init__(in_pipe_path, out_pipe_path, permissions)
|
||||
self.callback = callback
|
||||
|
||||
def handle_message(self, message):
|
||||
try:
|
||||
self.callback(message)
|
||||
except: # pylint: disable=bare-except
|
||||
LOG.exception('Failed to handle message %s from pipe %s!', message, self.in_pipe)
|
||||
|
||||
|
||||
class PipeIOEventHandler(PipeIOEventHandlerBase):
|
||||
def handle_event(self, message):
|
||||
json_bytes = dumps(message).encode()
|
||||
self.pipe_io.send_message(json_bytes)
|
||||
|
||||
def handle_pipe_event(self, message_bytes):
|
||||
json = loads(message_bytes)
|
||||
self.send_message(json)
|
||||
|
||||
|
||||
class TransformerPipeIOEventHandler(PipeIOEventHandlerBase):
|
||||
# pylint: disable=too-many-arguments
|
||||
def __init__(
|
||||
self, key, in_pipe_path, out_pipe_path,
|
||||
transform_in_cmd, transform_out_cmd,
|
||||
permissions=DEFAULT_PERMISSIONS
|
||||
):
|
||||
self._transform_in = partial(self._transform_message, transform_in_cmd)
|
||||
self._transform_out = partial(self._transform_message, transform_out_cmd)
|
||||
super().__init__(key, in_pipe_path, out_pipe_path, permissions)
|
||||
|
||||
@staticmethod
|
||||
def _transform_message(transform_cmd, message):
|
||||
proc = run(
|
||||
transform_cmd,
|
||||
input=message,
|
||||
stdout=PIPE,
|
||||
stderr=PIPE,
|
||||
shell=True
|
||||
)
|
||||
if proc.returncode == 0:
|
||||
return proc.stdout
|
||||
raise ValueError(f'Transforming message {message} failed!')
|
||||
|
||||
def handle_event(self, message):
|
||||
json_bytes = dumps(message).encode()
|
||||
transformed_bytes = self._transform_out(json_bytes)
|
||||
if transformed_bytes:
|
||||
self.pipe_io.send_message(transformed_bytes)
|
||||
|
||||
def handle_pipe_event(self, message_bytes):
|
||||
transformed_bytes = self._transform_in(message_bytes)
|
||||
if transformed_bytes:
|
||||
json_message = loads(transformed_bytes)
|
||||
self.send_message(json_message)
|
||||
|
||||
|
||||
class CommandEventHandler(PipeIOEventHandler):
|
||||
def __init__(self, key, command, permissions=DEFAULT_PERMISSIONS):
|
||||
super().__init__(
|
||||
key,
|
||||
self._generate_tempfilename(),
|
||||
self._generate_tempfilename(),
|
||||
permissions
|
||||
)
|
||||
|
||||
self._proc_stdin = open(self.pipe_io.out_pipe, 'rb')
|
||||
self._proc_stdout = open(self.pipe_io.in_pipe, 'wb')
|
||||
self._proc = Popen(
|
||||
command, shell=True, executable='/bin/bash',
|
||||
stdin=self._proc_stdin, stdout=self._proc_stdout, stderr=PIPE,
|
||||
start_new_session=True
|
||||
)
|
||||
self._monitor_proc_thread = self._start_monitor_proc()
|
||||
|
||||
def _generate_tempfilename(self):
|
||||
# pylint: disable=no-self-use
|
||||
random_filename = partial(token_urlsafe, 10)
|
||||
return join('/tmp', f'{type(self).__name__}.{random_filename()}')
|
||||
|
||||
def _start_monitor_proc(self):
|
||||
thread = Thread(target=self._monitor_proc, daemon=True)
|
||||
thread.start()
|
||||
return thread
|
||||
|
||||
@terminate_process_on_failure
|
||||
def _monitor_proc(self):
|
||||
return_code = self._proc.wait()
|
||||
if return_code == -int(SIGTERM):
|
||||
# supervisord asked the program to terminate, this is fine
|
||||
return
|
||||
if return_code != 0:
|
||||
_, stderr = self._proc.communicate()
|
||||
raise RuntimeError(f'Subprocess failed ({return_code})! Stderr:\n{stderr.decode()}')
|
||||
|
||||
def cleanup(self):
|
||||
with suppress(ProcessLookupError):
|
||||
process_group_id = getpgid(self._proc.pid)
|
||||
killpg(process_group_id, SIGTERM)
|
||||
self._proc_stdin.close()
|
||||
self._proc_stdout.close()
|
||||
super().cleanup()
|
||||
@ -1,64 +0,0 @@
|
||||
# Copyright (C) 2018 Avatao.com Innovative Learning Kft.
|
||||
# All Rights Reserved. See LICENSE file for details.
|
||||
|
||||
import logging
|
||||
from xmlrpc.client import Fault as SupervisorFault
|
||||
|
||||
from tfw.event_handlers import FrontendEventHandlerBase
|
||||
from tfw.mixins.supervisor_mixin import SupervisorMixin, SupervisorLogMixin
|
||||
from tfw.components.directory_monitor import with_monitor_paused
|
||||
|
||||
LOG = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class ProcessManager(SupervisorMixin, SupervisorLogMixin):
|
||||
def __init__(self):
|
||||
self.commands = {
|
||||
'start': self.start_process,
|
||||
'stop': self.stop_process,
|
||||
'restart': self.restart_process
|
||||
}
|
||||
|
||||
def __call__(self, command, process_name):
|
||||
return self.commands[command](process_name)
|
||||
|
||||
|
||||
class ProcessManagingEventHandler(FrontendEventHandlerBase):
|
||||
"""
|
||||
Event handler that can manage processes managed by supervisor.
|
||||
|
||||
This EventHandler accepts messages that have a data['command'] key specifying
|
||||
a command to be executed.
|
||||
Every message must contain a data['process_name'] field with the name of the
|
||||
process to manage. This is the name specified in supervisor config files like so:
|
||||
[program:someprogram]
|
||||
|
||||
Commands available: start, stop, restart, readlog
|
||||
(the names are as self-documenting as it gets)
|
||||
"""
|
||||
def __init__(self, key, dirmonitor=None, log_tail=0):
|
||||
super().__init__(key)
|
||||
self.monitor = dirmonitor
|
||||
self.processmanager = ProcessManager()
|
||||
self.log_tail = log_tail
|
||||
|
||||
@with_monitor_paused
|
||||
def handle_event(self, message):
|
||||
try:
|
||||
data = message['data']
|
||||
try:
|
||||
self.processmanager(data['command'], data['process_name'])
|
||||
except SupervisorFault as fault:
|
||||
message['data']['error'] = fault.faultString
|
||||
finally:
|
||||
message['data']['stdout'] = self.processmanager.read_stdout(
|
||||
data['process_name'],
|
||||
self.log_tail
|
||||
)
|
||||
message['data']['stderr'] = self.processmanager.read_stderr(
|
||||
data['process_name'],
|
||||
self.log_tail
|
||||
)
|
||||
return message
|
||||
except KeyError:
|
||||
LOG.error('IGNORING MESSAGE: Invalid message received: %s', message)
|
||||
@ -1,88 +0,0 @@
|
||||
# Copyright (C) 2018 Avatao.com Innovative Learning Kft.
|
||||
# All Rights Reserved. See LICENSE file for details.
|
||||
|
||||
import logging
|
||||
|
||||
from tfw.event_handlers import FrontendEventHandlerBase
|
||||
from tfw.components.terminado_mini_server import TerminadoMiniServer
|
||||
from tfw.config import TFWENV
|
||||
from tao.config import TAOENV
|
||||
|
||||
LOG = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class TerminalEventHandler(FrontendEventHandlerBase):
|
||||
"""
|
||||
Event handler responsible for managing terminal sessions for frontend xterm
|
||||
sessions to connect to. You need to instanciate this in order for frontend
|
||||
terminals to work.
|
||||
|
||||
This EventHandler accepts messages that have a data['command'] key specifying
|
||||
a command to be executed.
|
||||
The API of each command is documented in their respective handler.
|
||||
"""
|
||||
def __init__(self, key, monitor):
|
||||
"""
|
||||
:param key: key this EventHandler listens to
|
||||
:param monitor: tfw.components.HistoryMonitor instance to read command history from
|
||||
"""
|
||||
super().__init__(key)
|
||||
self._historymonitor = monitor
|
||||
bash_as_user_cmd = ['sudo', '-u', TAOENV.USER, 'bash']
|
||||
|
||||
self.terminado_server = TerminadoMiniServer(
|
||||
'/terminal',
|
||||
TFWENV.TERMINADO_PORT,
|
||||
TFWENV.TERMINADO_WD,
|
||||
bash_as_user_cmd
|
||||
)
|
||||
|
||||
self.commands = {
|
||||
'write': self.write,
|
||||
'read': self.read
|
||||
}
|
||||
|
||||
if self._historymonitor:
|
||||
self._historymonitor.start()
|
||||
self.terminado_server.listen()
|
||||
|
||||
@property
|
||||
def historymonitor(self):
|
||||
return self._historymonitor
|
||||
|
||||
def handle_event(self, message):
|
||||
try:
|
||||
data = message['data']
|
||||
message['data'] = self.commands[data['command']](data)
|
||||
return message
|
||||
except KeyError:
|
||||
LOG.error('IGNORING MESSAGE: Invalid message received: %s', message)
|
||||
|
||||
def write(self, data):
|
||||
"""
|
||||
Writes a string to the terminal session (on the pty level).
|
||||
Useful for pre-typing and executing commands for the user.
|
||||
|
||||
:param data: TFW message data containing 'value'
|
||||
(command to be written to the pty)
|
||||
"""
|
||||
self.terminado_server.pty.write(data['value'])
|
||||
return data
|
||||
|
||||
def read(self, data):
|
||||
"""
|
||||
Reads the history of commands executed.
|
||||
|
||||
:param data: TFW message data containing 'count'
|
||||
(the number of history elements to return)
|
||||
:return dict: message with list of commands in data['history']
|
||||
"""
|
||||
data['count'] = int(data.get('count', 1))
|
||||
if self.historymonitor:
|
||||
data['history'] = self.historymonitor.history[-data['count']:]
|
||||
return data
|
||||
|
||||
def cleanup(self):
|
||||
self.terminado_server.stop()
|
||||
if self.historymonitor:
|
||||
self.historymonitor.stop()
|
||||
Reference in New Issue
Block a user