From 0771a068e015fac6257a75543048a0014dcbadc6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Krist=C3=B3f=20T=C3=B3th?= Date: Thu, 2 May 2019 14:27:48 +0200 Subject: [PATCH 01/21] Implement helper to get all EventHandler instances in a given stack frame --- lib/tfw/event_handler_base.py | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/lib/tfw/event_handler_base.py b/lib/tfw/event_handler_base.py index aaf408c..04bcaf4 100644 --- a/lib/tfw/event_handler_base.py +++ b/lib/tfw/event_handler_base.py @@ -2,6 +2,7 @@ # All Rights Reserved. See LICENSE file for details. from abc import ABC, abstractmethod +from inspect import currentframe from tfw.networking.event_handlers import ServerConnector from tfw.crypto import message_checksum, KeyManager, verify_message @@ -105,6 +106,18 @@ class EventHandlerBase(ABC): """ pass + @classmethod + def get_local_instances(cls): + frame = currentframe() + if frame is None: + raise EnvironmentError('inspect.currentframe() is not supported!') + + locals_values = frame.f_back.f_locals.values() + return { + instance for instance in locals_values + if isinstance(instance, cls) + } + class FSMAwareEventHandler(EventHandlerBase, ABC): # pylint: disable=abstract-method From 6ea0967a21618227e7cd5e9c26d03871cc61011d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Krist=C3=B3f=20T=C3=B3th?= Date: Thu, 2 May 2019 14:53:59 +0200 Subject: [PATCH 02/21] Fork PipeIOServer to TFW --- lib/tfw/components/__init__.py | 1 + lib/tfw/components/pipe_io_server/__init__.py | 2 + lib/tfw/components/pipe_io_server/deque.py | 27 +++++++ lib/tfw/components/pipe_io_server/pipe.py | 16 ++++ .../pipe_io_server/pipe_io_server.py | 73 +++++++++++++++++++ .../pipe_io_server/pipe_reader_thread.py | 44 +++++++++++ .../pipe_io_server/pipe_writer_thread.py | 50 +++++++++++++ .../terminate_process_on_failure.py | 15 ++++ 8 files changed, 228 insertions(+) create mode 100644 lib/tfw/components/pipe_io_server/__init__.py create mode 100644 lib/tfw/components/pipe_io_server/deque.py create mode 100644 lib/tfw/components/pipe_io_server/pipe.py create mode 100644 lib/tfw/components/pipe_io_server/pipe_io_server.py create mode 100644 lib/tfw/components/pipe_io_server/pipe_reader_thread.py create mode 100644 lib/tfw/components/pipe_io_server/pipe_writer_thread.py create mode 100644 lib/tfw/components/pipe_io_server/terminate_process_on_failure.py diff --git a/lib/tfw/components/__init__.py b/lib/tfw/components/__init__.py index 4c0475c..5251a4e 100644 --- a/lib/tfw/components/__init__.py +++ b/lib/tfw/components/__init__.py @@ -10,3 +10,4 @@ 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 PipeIOEventHandler diff --git a/lib/tfw/components/pipe_io_server/__init__.py b/lib/tfw/components/pipe_io_server/__init__.py new file mode 100644 index 0000000..cab334b --- /dev/null +++ b/lib/tfw/components/pipe_io_server/__init__.py @@ -0,0 +1,2 @@ +from .pipe_io_server import PipeIOServer +from .terminate_process_on_failure import terminate_process_on_failure diff --git a/lib/tfw/components/pipe_io_server/deque.py b/lib/tfw/components/pipe_io_server/deque.py new file mode 100644 index 0000000..b2f1ab4 --- /dev/null +++ b/lib/tfw/components/pipe_io_server/deque.py @@ -0,0 +1,27 @@ +from collections import deque +from threading import Lock, Condition + + +class Deque: + def __init__(self): + self._queue = deque() + + self._mutex = Lock() + self._not_empty = Condition(self._mutex) + + def pop(self): + with self._mutex: + while not self._queue: + self._not_empty.wait() + return self._queue.pop() + + def push(self, item): + self._push(item, self._queue.appendleft) + + def push_front(self, item): + self._push(item, self._queue.append) + + def _push(self, item, put_method): + with self._mutex: + put_method(item) + self._not_empty.notify() diff --git a/lib/tfw/components/pipe_io_server/pipe.py b/lib/tfw/components/pipe_io_server/pipe.py new file mode 100644 index 0000000..eb021ae --- /dev/null +++ b/lib/tfw/components/pipe_io_server/pipe.py @@ -0,0 +1,16 @@ +from os import mkfifo, remove, chmod +from os.path import exists + + +class Pipe: + def __init__(self, path): + self.path = path + + def recreate(self, permissions): + self.remove() + mkfifo(self.path) + chmod(self.path, permissions) # use chmod to ignore umask + + def remove(self): + if exists(self.path): + remove(self.path) diff --git a/lib/tfw/components/pipe_io_server/pipe_io_server.py b/lib/tfw/components/pipe_io_server/pipe_io_server.py new file mode 100644 index 0000000..2715f40 --- /dev/null +++ b/lib/tfw/components/pipe_io_server/pipe_io_server.py @@ -0,0 +1,73 @@ +from abc import ABC, abstractmethod +from threading import Thread, Event +from typing import Callable + +from .pipe_reader_thread import PipeReaderThread +from .pipe_writer_thread import PipeWriterThread +from .pipe import Pipe +from .terminate_process_on_failure import terminate_process_on_failure + + +class PipeIOServer(ABC, Thread): + def __init__(self, in_pipe=None, out_pipe=None, permissions=0o600): + super().__init__(daemon=True) + self._in_pipe, self._out_pipe = in_pipe, out_pipe + self._create_pipes(permissions) + self._stop_event = Event() + self._reader_thread, self._writer_thread = self._create_io_threads() + self._io_threads = (self._reader_thread, self._writer_thread) + self._on_stop = lambda: None + + def _create_pipes(self, permissions): + Pipe(self.in_pipe).recreate(permissions) + Pipe(self.out_pipe).recreate(permissions) + + @property + def in_pipe(self): + return self._in_pipe + + @property + def out_pipe(self): + return self._out_pipe + + def _create_io_threads(self): + reader_thread = PipeReaderThread(self.in_pipe, self._stop_event, self.handle_message) + writer_thread = PipeWriterThread(self.out_pipe, self._stop_event) + return reader_thread, writer_thread + + @abstractmethod + def handle_message(self, message): + raise NotImplementedError() + + def send_message(self, message): + self._writer_thread.write(message) + + @terminate_process_on_failure + def run(self): + for thread in self._io_threads: + thread.start() + self._stop_event.wait() + self._stop_threads() + + def stop(self): + self._stop_event.set() + if self.is_alive(): + self.join() + + def _stop_threads(self): + for thread in self._io_threads: + if thread.is_alive(): + thread.stop() + Pipe(self.in_pipe).remove() + Pipe(self.out_pipe).remove() + self._on_stop() + + def _set_on_stop(self, value): + if not isinstance(value, Callable): + raise ValueError("Supplied object is not callable!") + self._on_stop = value + + on_stop = property(fset=_set_on_stop) + + def wait(self): + self._stop_event.wait() diff --git a/lib/tfw/components/pipe_io_server/pipe_reader_thread.py b/lib/tfw/components/pipe_io_server/pipe_reader_thread.py new file mode 100644 index 0000000..4bce19d --- /dev/null +++ b/lib/tfw/components/pipe_io_server/pipe_reader_thread.py @@ -0,0 +1,44 @@ +from contextlib import suppress +from os import open as osopen +from os import write, close, O_WRONLY, O_NONBLOCK +from threading import Thread + +from .terminate_process_on_failure import terminate_process_on_failure + + +class PipeReaderThread(Thread): + eof = b'' + stop_sequence = b'stop_reading\n' + + def __init__(self, pipe_path, stop_event, message_handler): + super().__init__(daemon=True) + self._message_handler = message_handler + self._pipe_path = pipe_path + self._stop_event = stop_event + + @terminate_process_on_failure + def run(self): + with self._open() as pipe: + while True: + message = pipe.readline() + if message == self.stop_sequence: + self._stop_event.set() + break + if message == self.eof: + self._open().close() + continue + self._message_handler(message[:-1]) + + def _open(self): + return open(self._pipe_path, 'rb') + + def stop(self): + while self.is_alive(): + self._unblock() + self.join() + + def _unblock(self): + with suppress(OSError): + fd = osopen(self._pipe_path, O_WRONLY | O_NONBLOCK) + write(fd, self.stop_sequence) + close(fd) diff --git a/lib/tfw/components/pipe_io_server/pipe_writer_thread.py b/lib/tfw/components/pipe_io_server/pipe_writer_thread.py new file mode 100644 index 0000000..3bc3fe7 --- /dev/null +++ b/lib/tfw/components/pipe_io_server/pipe_writer_thread.py @@ -0,0 +1,50 @@ +from contextlib import suppress +from os import O_NONBLOCK, O_RDONLY, close +from os import open as osopen +from threading import Thread + +from .terminate_process_on_failure import terminate_process_on_failure +from .deque import Deque + + +class PipeWriterThread(Thread): + def __init__(self, pipe_path, stop_event): + super().__init__(daemon=True) + self._pipe_path = pipe_path + self._stop_event = stop_event + self._write_queue = Deque() + + def write(self, message): + self._write_queue.push(message) + + @terminate_process_on_failure + def run(self): + with self._open() as pipe: + while True: + message = self._write_queue.pop() + if message is None: + self._stop_event.set() + break + try: + pipe.write(message + b'\n') + pipe.flush() + except BrokenPipeError: + try: # pipe was reopened, close() flushed the message + pipe.close() + except BrokenPipeError: # close() discarded the message + self._write_queue.push_front(message) + pipe = self._open() + + def _open(self): + return open(self._pipe_path, 'wb') + + def stop(self): + while self.is_alive(): + self._unblock() + self.join() + + def _unblock(self): + with suppress(OSError): + fd = osopen(self._pipe_path, O_RDONLY | O_NONBLOCK) + self._write_queue.push_front(None) + close(fd) diff --git a/lib/tfw/components/pipe_io_server/terminate_process_on_failure.py b/lib/tfw/components/pipe_io_server/terminate_process_on_failure.py new file mode 100644 index 0000000..7a0804c --- /dev/null +++ b/lib/tfw/components/pipe_io_server/terminate_process_on_failure.py @@ -0,0 +1,15 @@ +from functools import wraps +from os import kill, getpid +from signal import SIGTERM +from traceback import print_exc + + +def terminate_process_on_failure(fun): + @wraps(fun) + def wrapper(*args, **kwargs): + try: + return fun(*args, **kwargs) + except: # pylint: disable=bare-except + print_exc() + kill(getpid(), SIGTERM) + return wrapper From 065aa561824b745b99b2ab0b65500dfcba6a2955 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Krist=C3=B3f=20T=C3=B3th?= Date: Thu, 2 May 2019 14:54:55 +0200 Subject: [PATCH 03/21] Implement EventHandler capable of proxying between TFW and named pipes --- lib/tfw/components/pipe_io_event_handler.py | 35 +++++++++++++++++++++ 1 file changed, 35 insertions(+) create mode 100644 lib/tfw/components/pipe_io_event_handler.py diff --git a/lib/tfw/components/pipe_io_event_handler.py b/lib/tfw/components/pipe_io_event_handler.py new file mode 100644 index 0000000..df3d012 --- /dev/null +++ b/lib/tfw/components/pipe_io_event_handler.py @@ -0,0 +1,35 @@ +from json import loads, dumps, JSONDecodeError + +from tfw import EventHandlerBase +from tfw.config.logs import logging + +from .pipe_io_server import PipeIOServer + +LOG = logging.getLogger(__name__) + + +class PipeIOEventHandler(EventHandlerBase): + def __init__(self, key, in_pipe_path, out_pipe_path): + super().__init__(key) + self._pipe_io_server = JSONProxyPipeIOServer(in_pipe_path, out_pipe_path, self.server_connector.send) + self._pipe_io_server.start() + + def cleanup(self): + self._pipe_io_server.stop() + + def handle_event(self, message): + json_bytes = dumps(message).encode() + self._pipe_io_server.send_message(json_bytes) + + +class JSONProxyPipeIOServer(PipeIOServer): + def __init__(self, in_pipe_path, out_pipe_path, proxy_method): + super().__init__(in_pipe_path, out_pipe_path) + self.proxy = proxy_method + + def handle_message(self, message): + try: + json = loads(message) + self.proxy(json) + except JSONDecodeError: + LOG.debug("Invalid JSON received on %s! Ignoring...", self._in_pipe) From f94d571d19c00ab9978eb0462f11bbac81548d1e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Krist=C3=B3f=20T=C3=B3th?= Date: Sat, 4 May 2019 21:10:05 +0200 Subject: [PATCH 04/21] Support initializing IO pipes with specific permissions --- lib/tfw/components/pipe_io_event_handler.py | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) diff --git a/lib/tfw/components/pipe_io_event_handler.py b/lib/tfw/components/pipe_io_event_handler.py index df3d012..d610f4f 100644 --- a/lib/tfw/components/pipe_io_event_handler.py +++ b/lib/tfw/components/pipe_io_event_handler.py @@ -9,9 +9,14 @@ LOG = logging.getLogger(__name__) class PipeIOEventHandler(EventHandlerBase): - def __init__(self, key, in_pipe_path, out_pipe_path): + def __init__(self, key, in_pipe_path, out_pipe_path, permissions=0o600): super().__init__(key) - self._pipe_io_server = JSONProxyPipeIOServer(in_pipe_path, out_pipe_path, self.server_connector.send) + self._pipe_io_server = JSONProxyPipeIOServer( + in_pipe_path, + out_pipe_path, + self.server_connector.send, + permissions + ) self._pipe_io_server.start() def cleanup(self): @@ -23,8 +28,8 @@ class PipeIOEventHandler(EventHandlerBase): class JSONProxyPipeIOServer(PipeIOServer): - def __init__(self, in_pipe_path, out_pipe_path, proxy_method): - super().__init__(in_pipe_path, out_pipe_path) + def __init__(self, in_pipe_path, out_pipe_path, proxy_method, permissions): + super().__init__(in_pipe_path, out_pipe_path, permissions) self.proxy = proxy_method def handle_message(self, message): @@ -32,4 +37,4 @@ class JSONProxyPipeIOServer(PipeIOServer): json = loads(message) self.proxy(json) except JSONDecodeError: - LOG.debug("Invalid JSON received on %s! Ignoring...", self._in_pipe) + LOG.error("Invalid JSON received on %s! Ignoring...", self._in_pipe) From bb8e0c74582dae669bd34bf4ab9915a29d407b23 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Krist=C3=B3f=20T=C3=B3th?= Date: Sat, 4 May 2019 21:13:58 +0200 Subject: [PATCH 05/21] Handle JSON serialization errors in PipeIOEventHandler --- lib/tfw/components/pipe_io_event_handler.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/lib/tfw/components/pipe_io_event_handler.py b/lib/tfw/components/pipe_io_event_handler.py index d610f4f..7604787 100644 --- a/lib/tfw/components/pipe_io_event_handler.py +++ b/lib/tfw/components/pipe_io_event_handler.py @@ -23,8 +23,12 @@ class PipeIOEventHandler(EventHandlerBase): self._pipe_io_server.stop() def handle_event(self, message): - json_bytes = dumps(message).encode() - self._pipe_io_server.send_message(json_bytes) + try: + json_bytes = dumps(message).encode() + self._pipe_io_server.send_message(json_bytes) + except TypeError: + LOG.error("Message %s not JSON serializable! Ignoring...", message) + class JSONProxyPipeIOServer(PipeIOServer): From ddc79c9717d5ef28ca9e140f5708fe9ad7eac28b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Krist=C3=B3f=20T=C3=B3th?= Date: Sun, 5 May 2019 20:57:22 +0200 Subject: [PATCH 06/21] Allow subscribing to all ZMQ topics in EventHandlerBase --- lib/tfw/event_handler_base.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/lib/tfw/event_handler_base.py b/lib/tfw/event_handler_base.py index 04bcaf4..6f9819b 100644 --- a/lib/tfw/event_handler_base.py +++ b/lib/tfw/event_handler_base.py @@ -55,6 +55,8 @@ class EventHandlerBase(ABC): subscribed to 'fsm' will receive 'fsm_update' messages as well. """ + if '' in self.keys: + return True return message['key'] in self.keys def dispatch_handling(self, message): From 90b780a5c0324712ce8fc762f1c1f73a944940ef Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Krist=C3=B3f=20T=C3=B3th?= Date: Mon, 6 May 2019 15:23:21 +0200 Subject: [PATCH 07/21] Allow subclassing pipe based event handlers (refactor PipeIOEventHandler to base class and impl) --- lib/tfw/components/__init__.py | 2 +- lib/tfw/components/pipe_io_event_handler.py | 41 +++++++++++++-------- 2 files changed, 26 insertions(+), 17 deletions(-) diff --git a/lib/tfw/components/__init__.py b/lib/tfw/components/__init__.py index 5251a4e..2c36b84 100644 --- a/lib/tfw/components/__init__.py +++ b/lib/tfw/components/__init__.py @@ -10,4 +10,4 @@ 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 PipeIOEventHandler +from .pipe_io_event_handler import PipeIOEventHandlerBase, PipeIOEventHandler diff --git a/lib/tfw/components/pipe_io_event_handler.py b/lib/tfw/components/pipe_io_event_handler.py index 7604787..c681567 100644 --- a/lib/tfw/components/pipe_io_event_handler.py +++ b/lib/tfw/components/pipe_io_event_handler.py @@ -1,3 +1,4 @@ +from abc import abstractmethod from json import loads, dumps, JSONDecodeError from tfw import EventHandlerBase @@ -8,37 +9,45 @@ from .pipe_io_server import PipeIOServer LOG = logging.getLogger(__name__) -class PipeIOEventHandler(EventHandlerBase): +class PipeIOEventHandlerBase(EventHandlerBase): def __init__(self, key, in_pipe_path, out_pipe_path, permissions=0o600): super().__init__(key) - self._pipe_io_server = JSONProxyPipeIOServer( + self.pipe_io = CallbackPipeIOServer( in_pipe_path, out_pipe_path, - self.server_connector.send, + self.handle_pipe_event, permissions ) - self._pipe_io_server.start() + self.pipe_io.start() + + @abstractmethod + def handle_pipe_event(self, message): + raise NotImplementedError() def cleanup(self): - self._pipe_io_server.stop() + 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): + self.callback(message) + + +class PipeIOEventHandler(PipeIOEventHandlerBase): def handle_event(self, message): try: json_bytes = dumps(message).encode() - self._pipe_io_server.send_message(json_bytes) + self.pipe_io.send_message(json_bytes) except TypeError: LOG.error("Message %s not JSON serializable! Ignoring...", message) - - -class JSONProxyPipeIOServer(PipeIOServer): - def __init__(self, in_pipe_path, out_pipe_path, proxy_method, permissions): - super().__init__(in_pipe_path, out_pipe_path, permissions) - self.proxy = proxy_method - - def handle_message(self, message): + def handle_pipe_event(self, message): try: json = loads(message) - self.proxy(json) + self.server_connector.send(json) except JSONDecodeError: - LOG.error("Invalid JSON received on %s! Ignoring...", self._in_pipe) + LOG.error("Invalid JSON received on %s! Ignoring...", self.pipe_io.in_pipe) From 9ad77eaed8653d7bcc9ebfd3d9e366a27d7e6d4a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Krist=C3=B3f=20T=C3=B3th?= Date: Mon, 6 May 2019 17:11:56 +0200 Subject: [PATCH 08/21] Allow passing an iterable with keys to subscribe to in EventHandlerBase.__init__() --- lib/tfw/event_handler_base.py | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/lib/tfw/event_handler_base.py b/lib/tfw/event_handler_base.py index 6f9819b..beb79a4 100644 --- a/lib/tfw/event_handler_base.py +++ b/lib/tfw/event_handler_base.py @@ -3,6 +3,7 @@ from abc import ABC, abstractmethod from inspect import currentframe +from typing import Iterable from tfw.networking.event_handlers import ServerConnector from tfw.crypto import message_checksum, KeyManager, verify_message @@ -20,7 +21,12 @@ class EventHandlerBase(ABC): """ def __init__(self, key): self.server_connector = ServerConnector() - self.keys = [key] + self.keys = [] + if isinstance(key, str): + self.keys.append(key) + elif isinstance(key, Iterable): + self.keys = list(key) + self.subscribe(*self.keys) self.server_connector.register_callback(self.event_handler_callback) From 8f1ae9e286b37b01b1a0b9dafe6bb2c03e478e42 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Krist=C3=B3f=20T=C3=B3th?= Date: Tue, 7 May 2019 12:58:58 +0200 Subject: [PATCH 09/21] Add jq as a dependency --- Dockerfile | 1 + 1 file changed, 1 insertion(+) diff --git a/Dockerfile b/Dockerfile index a264278..f9ccf5e 100644 --- a/Dockerfile +++ b/Dockerfile @@ -10,6 +10,7 @@ RUN curl -sL https://deb.nodesource.com/setup_8.x | sudo -E bash - supervisor \ libzmq5 \ nginx \ + jq \ gettext-base &&\ rm -rf /var/lib/apt/lists/* &&\ ln -sf /bin/bash /bin/sh From 9e36bde97416324fd4414d282a9d125b74dfd368 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Krist=C3=B3f=20T=C3=B3th?= Date: Tue, 7 May 2019 13:08:11 +0200 Subject: [PATCH 10/21] Consider PipeIOServer public API from now on --- lib/tfw/components/__init__.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lib/tfw/components/__init__.py b/lib/tfw/components/__init__.py index 2c36b84..fd93d94 100644 --- a/lib/tfw/components/__init__.py +++ b/lib/tfw/components/__init__.py @@ -10,4 +10,4 @@ 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 +from .pipe_io_event_handler import PipeIOEventHandlerBase, PipeIOEventHandler, PipeIOServer From 1bfaac0493b7c16080b6d90a442bf8dafff74119 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Krist=C3=B3f=20T=C3=B3th?= Date: Thu, 9 May 2019 15:12:42 +0200 Subject: [PATCH 11/21] Remove potentially harmful 'error handling' --- lib/tfw/components/pipe_io_event_handler.py | 14 ++++---------- 1 file changed, 4 insertions(+), 10 deletions(-) diff --git a/lib/tfw/components/pipe_io_event_handler.py b/lib/tfw/components/pipe_io_event_handler.py index c681567..04cb634 100644 --- a/lib/tfw/components/pipe_io_event_handler.py +++ b/lib/tfw/components/pipe_io_event_handler.py @@ -39,15 +39,9 @@ class CallbackPipeIOServer(PipeIOServer): class PipeIOEventHandler(PipeIOEventHandlerBase): def handle_event(self, message): - try: - json_bytes = dumps(message).encode() - self.pipe_io.send_message(json_bytes) - except TypeError: - LOG.error("Message %s not JSON serializable! Ignoring...", message) + json_bytes = dumps(message).encode() + self.pipe_io.send_message(json_bytes) def handle_pipe_event(self, message): - try: - json = loads(message) - self.server_connector.send(json) - except JSONDecodeError: - LOG.error("Invalid JSON received on %s! Ignoring...", self.pipe_io.in_pipe) + json = loads(message) + self.server_connector.send(json) From 078f8532cc40a006d3749ffc29728ff7b0e53efa Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Krist=C3=B3f=20T=C3=B3th?= Date: Thu, 9 May 2019 15:14:47 +0200 Subject: [PATCH 12/21] Rename parameter to reflect type in PipeIOEventHandler.handle_pipe_event --- lib/tfw/components/pipe_io_event_handler.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/lib/tfw/components/pipe_io_event_handler.py b/lib/tfw/components/pipe_io_event_handler.py index 04cb634..ef162c9 100644 --- a/lib/tfw/components/pipe_io_event_handler.py +++ b/lib/tfw/components/pipe_io_event_handler.py @@ -21,7 +21,7 @@ class PipeIOEventHandlerBase(EventHandlerBase): self.pipe_io.start() @abstractmethod - def handle_pipe_event(self, message): + def handle_pipe_event(self, message_bytes): raise NotImplementedError() def cleanup(self): @@ -42,6 +42,6 @@ class PipeIOEventHandler(PipeIOEventHandlerBase): json_bytes = dumps(message).encode() self.pipe_io.send_message(json_bytes) - def handle_pipe_event(self, message): - json = loads(message) + def handle_pipe_event(self, message_bytes): + json = loads(message_bytes) self.server_connector.send(json) From 69b3b17724e3f28020a2aeec27d62b2981fbbc6a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Krist=C3=B3f=20T=C3=B3th?= Date: Thu, 9 May 2019 16:56:16 +0200 Subject: [PATCH 13/21] Implement EventHandler for transforming and filtering pipe messages --- lib/tfw/components/__init__.py | 1 + lib/tfw/components/pipe_io_event_handler.py | 48 ++++++++++++++++++++- 2 files changed, 47 insertions(+), 2 deletions(-) diff --git a/lib/tfw/components/__init__.py b/lib/tfw/components/__init__.py index fd93d94..7aabc38 100644 --- a/lib/tfw/components/__init__.py +++ b/lib/tfw/components/__init__.py @@ -11,3 +11,4 @@ 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 diff --git a/lib/tfw/components/pipe_io_event_handler.py b/lib/tfw/components/pipe_io_event_handler.py index ef162c9..3bdb38a 100644 --- a/lib/tfw/components/pipe_io_event_handler.py +++ b/lib/tfw/components/pipe_io_event_handler.py @@ -1,5 +1,6 @@ from abc import abstractmethod -from json import loads, dumps, JSONDecodeError +from json import loads, dumps +from subprocess import run, PIPE from tfw import EventHandlerBase from tfw.config.logs import logging @@ -7,10 +8,11 @@ from tfw.config.logs import logging from .pipe_io_server import PipeIOServer LOG = logging.getLogger(__name__) +DEFAULT_PERMISSIONS = 0o600 class PipeIOEventHandlerBase(EventHandlerBase): - def __init__(self, key, in_pipe_path, out_pipe_path, permissions=0o600): + def __init__(self, key, in_pipe_path, out_pipe_path, permissions=DEFAULT_PERMISSIONS): super().__init__(key) self.pipe_io = CallbackPipeIOServer( in_pipe_path, @@ -45,3 +47,45 @@ class PipeIOEventHandler(PipeIOEventHandlerBase): def handle_pipe_event(self, message_bytes): json = loads(message_bytes) self.server_connector.send(json) + + +class TransformerPipeIOEventHandler(PipeIOEventHandlerBase): + # pylint: disable=too-many-arguments + def __init__( + self, key, in_pipe_path, out_pipe_path, + serialize_cmd, deserialize_cmd, + permissions=DEFAULT_PERMISSIONS + ): + self._serialize_cmd, self._deserialize_cmd = serialize_cmd, deserialize_cmd + super().__init__(key, in_pipe_path, out_pipe_path, permissions) + + def _serialize_message(self, message): + return self._transform_message(self._serialize_cmd, message) + + def _deserialize_message(self, message): + return self._transform_message(self._deserialize_cmd, message) + + @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._serialize_message(json_bytes) + if transformed_bytes: + self.pipe_io.send_message(transformed_bytes) + + def handle_pipe_event(self, message_bytes): + transformed_bytes = self._deserialize_message(message_bytes) + if transformed_bytes: + json_message = loads(transformed_bytes) + self.server_connector.send(json_message) From c4d3319ed963510fe4276c1e240eb5cd2dd2a369 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Krist=C3=B3f=20T=C3=B3th?= Date: Thu, 9 May 2019 18:22:33 +0200 Subject: [PATCH 14/21] Refactor TransformerPipeIOEventHandler --- lib/tfw/components/pipe_io_event_handler.py | 16 ++++++---------- 1 file changed, 6 insertions(+), 10 deletions(-) diff --git a/lib/tfw/components/pipe_io_event_handler.py b/lib/tfw/components/pipe_io_event_handler.py index 3bdb38a..451e9d1 100644 --- a/lib/tfw/components/pipe_io_event_handler.py +++ b/lib/tfw/components/pipe_io_event_handler.py @@ -1,6 +1,7 @@ from abc import abstractmethod from json import loads, dumps from subprocess import run, PIPE +from functools import partial from tfw import EventHandlerBase from tfw.config.logs import logging @@ -53,18 +54,13 @@ class TransformerPipeIOEventHandler(PipeIOEventHandlerBase): # pylint: disable=too-many-arguments def __init__( self, key, in_pipe_path, out_pipe_path, - serialize_cmd, deserialize_cmd, + transform_in_cmd, transform_out_cmd, permissions=DEFAULT_PERMISSIONS ): - self._serialize_cmd, self._deserialize_cmd = serialize_cmd, deserialize_cmd + 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) - def _serialize_message(self, message): - return self._transform_message(self._serialize_cmd, message) - - def _deserialize_message(self, message): - return self._transform_message(self._deserialize_cmd, message) - @staticmethod def _transform_message(transform_cmd, message): proc = run( @@ -80,12 +76,12 @@ class TransformerPipeIOEventHandler(PipeIOEventHandlerBase): def handle_event(self, message): json_bytes = dumps(message).encode() - transformed_bytes = self._serialize_message(json_bytes) + 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._deserialize_message(message_bytes) + transformed_bytes = self._transform_in(message_bytes) if transformed_bytes: json_message = loads(transformed_bytes) self.server_connector.send(json_message) From fc5124afb0686574abcf04a364bdd49a83d50e05 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Krist=C3=B3f=20T=C3=B3th?= Date: Thu, 9 May 2019 18:56:30 +0200 Subject: [PATCH 15/21] Make arbitrary processes capable of being EventHandlers (PipeIO+stdin/stdout) --- lib/tfw/components/__init__.py | 2 +- lib/tfw/components/pipe_io_event_handler.py | 36 ++++++++++++++++++++- 2 files changed, 36 insertions(+), 2 deletions(-) diff --git a/lib/tfw/components/__init__.py b/lib/tfw/components/__init__.py index 7aabc38..73c889d 100644 --- a/lib/tfw/components/__init__.py +++ b/lib/tfw/components/__init__.py @@ -11,4 +11,4 @@ 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 +from .pipe_io_event_handler import TransformerPipeIOEventHandler, CommandEventHandler diff --git a/lib/tfw/components/pipe_io_event_handler.py b/lib/tfw/components/pipe_io_event_handler.py index 451e9d1..63855b6 100644 --- a/lib/tfw/components/pipe_io_event_handler.py +++ b/lib/tfw/components/pipe_io_event_handler.py @@ -1,7 +1,11 @@ from abc import abstractmethod from json import loads, dumps -from subprocess import run, PIPE +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 tfw import EventHandlerBase from tfw.config.logs import logging @@ -85,3 +89,33 @@ class TransformerPipeIOEventHandler(PipeIOEventHandlerBase): if transformed_bytes: json_message = loads(transformed_bytes) self.server_connector.send(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, + start_new_session=True + ) + + 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 cleanup(self): + process_group_id = getpgid(self._proc.pid) + killpg(process_group_id, SIGTERM) + self._proc_stdin.close() + self._proc_stdout.close() + super().cleanup() From 78d70f2f8b7460d701474864fba6dd962e5f3c2b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Krist=C3=B3f=20T=C3=B3th?= Date: Sun, 12 May 2019 22:28:31 +0200 Subject: [PATCH 16/21] Make PipeIOEventHandlerBase handle all input errors (while screaming) --- lib/tfw/components/pipe_io_event_handler.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/lib/tfw/components/pipe_io_event_handler.py b/lib/tfw/components/pipe_io_event_handler.py index 63855b6..d60fedf 100644 --- a/lib/tfw/components/pipe_io_event_handler.py +++ b/lib/tfw/components/pipe_io_event_handler.py @@ -41,7 +41,10 @@ class CallbackPipeIOServer(PipeIOServer): self.callback = callback def handle_message(self, message): - self.callback(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): From ca5be9d848c002017a5b8de70ccd1c7bea7a69d1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Krist=C3=B3f=20T=C3=B3th?= Date: Sun, 12 May 2019 22:30:49 +0200 Subject: [PATCH 17/21] Detect errors in the subprocess of CommandEventHandler --- lib/tfw/components/pipe_io_event_handler.py | 33 +++++++++++++++++---- 1 file changed, 27 insertions(+), 6 deletions(-) diff --git a/lib/tfw/components/pipe_io_event_handler.py b/lib/tfw/components/pipe_io_event_handler.py index d60fedf..676a8e5 100644 --- a/lib/tfw/components/pipe_io_event_handler.py +++ b/lib/tfw/components/pipe_io_event_handler.py @@ -6,11 +6,14 @@ from os import getpgid, killpg from os.path import join from signal import SIGTERM from secrets import token_urlsafe +from time import sleep +from threading import Thread +from contextlib import suppress from tfw import EventHandlerBase from tfw.config.logs import logging -from .pipe_io_server import PipeIOServer +from .pipe_io_server import PipeIOServer, terminate_process_on_failure LOG = logging.getLogger(__name__) DEFAULT_PERMISSIONS = 0o600 @@ -107,18 +110,36 @@ class CommandEventHandler(PipeIOEventHandler): 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, + stdin=self._proc_stdin, stdout=self._proc_stdout, stderr=PIPE, start_new_session=True ) + self._poll_check_proc_thread = self._start_poll_check_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_poll_check_proc(self): + thread = Thread(target=self._poll_check_proc, daemon=True) + thread.start() + return thread + + @terminate_process_on_failure + def _poll_check_proc(self): + while True: + self.check_proc() + sleep(1) + + def check_proc(self): + if self._proc.poll() is not None: + _, stderr = self._proc.communicate() + raise RuntimeError(f'Subprocess failed: {stderr.decode()}') + def cleanup(self): - process_group_id = getpgid(self._proc.pid) - killpg(process_group_id, SIGTERM) - self._proc_stdin.close() - self._proc_stdout.close() + 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() From 16177611848c028d167efc1b807f7413cc6125b5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Krist=C3=B3f=20T=C3=B3th?= Date: Mon, 13 May 2019 11:17:30 +0200 Subject: [PATCH 18/21] Improve CommandEventHandler error detection (avoid polling) --- lib/tfw/components/pipe_io_event_handler.py | 17 ++++++----------- 1 file changed, 6 insertions(+), 11 deletions(-) diff --git a/lib/tfw/components/pipe_io_event_handler.py b/lib/tfw/components/pipe_io_event_handler.py index 676a8e5..c0f54aa 100644 --- a/lib/tfw/components/pipe_io_event_handler.py +++ b/lib/tfw/components/pipe_io_event_handler.py @@ -6,7 +6,6 @@ from os import getpgid, killpg from os.path import join from signal import SIGTERM from secrets import token_urlsafe -from time import sleep from threading import Thread from contextlib import suppress @@ -113,26 +112,22 @@ class CommandEventHandler(PipeIOEventHandler): stdin=self._proc_stdin, stdout=self._proc_stdout, stderr=PIPE, start_new_session=True ) - self._poll_check_proc_thread = self._start_poll_check_proc() + 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_poll_check_proc(self): - thread = Thread(target=self._poll_check_proc, daemon=True) + def _start_monitor_proc(self): + thread = Thread(target=self._monitor_proc, daemon=True) thread.start() return thread @terminate_process_on_failure - def _poll_check_proc(self): - while True: - self.check_proc() - sleep(1) - - def check_proc(self): - if self._proc.poll() is not None: + def _monitor_proc(self): + return_code = self._proc.wait() + if return_code != 0: _, stderr = self._proc.communicate() raise RuntimeError(f'Subprocess failed: {stderr.decode()}') From 94dee63a4181cf778bdedbde8327e3cd1f08d4b8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Krist=C3=B3f=20T=C3=B3th?= Date: Mon, 13 May 2019 14:52:17 +0200 Subject: [PATCH 19/21] Improve CommandEventHandler subprocess failure error message --- lib/tfw/components/pipe_io_event_handler.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lib/tfw/components/pipe_io_event_handler.py b/lib/tfw/components/pipe_io_event_handler.py index c0f54aa..d0eda41 100644 --- a/lib/tfw/components/pipe_io_event_handler.py +++ b/lib/tfw/components/pipe_io_event_handler.py @@ -129,7 +129,7 @@ class CommandEventHandler(PipeIOEventHandler): return_code = self._proc.wait() if return_code != 0: _, stderr = self._proc.communicate() - raise RuntimeError(f'Subprocess failed: {stderr.decode()}') + raise RuntimeError(f'Subprocess failed ({return_code})! Stderr:\n{stderr.decode()}') def cleanup(self): with suppress(ProcessLookupError): From 3bfe6db036b1bf0f0676f8b5fa2b05e0cad2aa6d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Krist=C3=B3f=20T=C3=B3th?= Date: Mon, 13 May 2019 14:53:31 +0200 Subject: [PATCH 20/21] Avoid treating supervisord's SIGTERM as an error in CommandEventHandler --- lib/tfw/components/pipe_io_event_handler.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/lib/tfw/components/pipe_io_event_handler.py b/lib/tfw/components/pipe_io_event_handler.py index d0eda41..f9cd567 100644 --- a/lib/tfw/components/pipe_io_event_handler.py +++ b/lib/tfw/components/pipe_io_event_handler.py @@ -127,6 +127,9 @@ class CommandEventHandler(PipeIOEventHandler): @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()}') From 51b55782709e6dd676f6267531eaaa95a094b29e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Krist=C3=B3f=20T=C3=B3th?= Date: Mon, 13 May 2019 15:07:19 +0200 Subject: [PATCH 21/21] Update builder image in .drone.yml --- .drone.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.drone.yml b/.drone.yml index 9dbc1f4..e49f310 100644 --- a/.drone.yml +++ b/.drone.yml @@ -1,6 +1,6 @@ pipeline: build: - image: eu.gcr.io/avatao-public/docker:el7 + image: eu.gcr.io/avatao-challengestore/challenge-builder volumes: - /etc/docker:/etc/docker:ro - /root/.docker:/root/.docker:ro