diff --git a/journalpump/senders/aws_cloudwatch.py b/journalpump/senders/aws_cloudwatch.py index 8b83596..f46f280 100644 --- a/journalpump/senders/aws_cloudwatch.py +++ b/journalpump/senders/aws_cloudwatch.py @@ -1,4 +1,5 @@ -from .base import LogSender, SenderInitializationError +from .base import ThreadedLogSender, SenderInitializationError + import boto3 import botocore @@ -8,7 +9,7 @@ MAX_INIT_TRIES = 3 -class AWSCloudWatchSender(LogSender): +class AWSCloudWatchSender(ThreadedLogSender): def __init__(self, *, config, aws_cloudwatch_logs=None, **kwargs): super().__init__(config=config, max_send_interval=config.get("max_send_interval", 0.3), **kwargs) self._logs = aws_cloudwatch_logs diff --git a/journalpump/senders/base.py b/journalpump/senders/base.py index 02142d4..0e452d0 100644 --- a/journalpump/senders/base.py +++ b/journalpump/senders/base.py @@ -67,7 +67,7 @@ def add_item(self, *, item, cursor): self.total_size += len(item) -class LogSender(Thread, Tagged): +class LogSender(Tagged): def __init__( self, *, @@ -81,9 +81,9 @@ def __init__( tags=None, msg_buffer_max_length=50000 ): - Thread.__init__(self) Tagged.__init__(self, tags, sender=name) self.log = logging.getLogger("LogSender:{}".format(reader.name)) + self._wait_for = 0.1 self.name = name self.stats = stats self.config = config @@ -106,12 +106,14 @@ def __init__( self._backoff_attempt = 0 self.log.info("Initialized %s", self.__class__.__name__) - def _backoff(self, *, base=0.5, cap=1800.0): + def _get_backoff_secs(self, *, base=0.5, cap=1000.0): self._backoff_attempt += 1 t = min(cap, base * 2 ** self._backoff_attempt) / 2 t = random.random() * t + t - self.log.info("Sleeping for %.0f seconds", t) - time.sleep(t) + return t + + def _backoff(self, *, base=0.5, cap=1800.0): + raise NotImplementedError() def refresh_stats(self): tags = self.make_tags() @@ -192,48 +194,74 @@ def maintenance_operations(self): # This can be overridden in the classes that inherit this pass + def handle_maintenance_operations(self): + try: + # Don't run maintenance operations again immediately if it just failed + if self.last_maintenance_fail or time.monotonic() - self.last_maintenance_fail > 60: + self.maintenance_operations() + except Exception as ex: # pylint: disable=broad-except + self.log.error("Maintenance operation failed: %r", ex) + self.stats.unexpected_exception(ex=ex, where="maintenance_operation") + self.last_maintenance_fail = time.monotonic() + + def should_try_sending_messages(self): + return len(self.msg_buffer) > 1000 or \ + time.monotonic() - self.last_send_time > self.max_send_interval + + def get_message_bodies_and_cursor(self): + messages = self.msg_buffer.get_items() + ret = [] + while messages: + batch_size = len(messages[0][0]) + self.batch_message_overhead + index = 1 + while index < len(messages): + item_size = len(messages[index][0]) + self.batch_message_overhead + if batch_size + item_size >= self.max_batch_size: + break + batch_size += item_size + index += 1 + + batch = messages[:index] + message_bodies = [m[0] for m in batch] + ret.append((message_bodies, batch[-1][1])) + del messages[:index] + return ret + + +class ThreadedLogSender(Thread, LogSender): + def __init__(self, **kw): + Thread.__init__(self) + LogSender.__init__(self, **kw) + + def _backoff(self, *, base=0.5, cap=1800.0): + t = self._get_backoff_secs(base=base, cap=cap) + self.log.info("Sleeping for %.0f seconds", t) + time.sleep(t) + def run(self): while self.running: - try: - # Don't run maintenance operations again immediately if it just failed - if not self.last_maintenance_fail or time.monotonic() - self.last_maintenance_fail > 60: - self.maintenance_operations() - except Exception as ex: # pylint: disable=broad-except - self.log.error("Maintenance operation failed: %r", ex) - self.stats.unexpected_exception(ex=ex, where="maintenance_operation") - self.last_maintenance_fail = time.monotonic() - if len(self.msg_buffer) > 1000 or \ - time.monotonic() - self.last_send_time > self.max_send_interval: + self.handle_maintenance_operations() + if self.should_try_sending_messages(): self.get_and_send_messages() else: - time.sleep(0.1) + time.sleep(self._wait_for) self.log.info("Stopping") def get_and_send_messages(self): + batches = self.get_message_bodies_and_cursor() + msg_count = sum(len(batch[0]) for batch in batches) + self.log.debug("Got %d items from msg_buffer", msg_count) start_time = time.monotonic() - msg_count = None try: - messages = self.msg_buffer.get_items() - msg_count = len(messages) - self.log.debug("Got %d items from msg_buffer", msg_count) - - while self.running and messages: - batch_size = len(messages[0][0]) + self.batch_message_overhead - index = 1 - while index < len(messages): - item_size = len(messages[index][0]) + self.batch_message_overhead - if batch_size + item_size >= self.max_batch_size: - break - batch_size += item_size - index += 1 - - messages_batch = messages[:index] - message_bodies = [m[0] for m in messages_batch] - if self.send_messages(messages=message_bodies, cursor=messages_batch[-1][1]): - messages = messages[index:] - + # pop to get free up memory as soon as the send was successful + while batches: + batch = batches.pop(0) + # die retrying, backoff is part of sending mechanism + while self.running and not self.send_messages(messages=batch[0], cursor=batch[1]): + pass self.log.debug("Sending %d msgs, took %.4fs", msg_count, time.monotonic() - start_time) self.last_send_time = time.monotonic() except Exception: # pylint: disable=broad-except + # there is already a broad except handler in send_messages, so why this ? self.log.exception("Problem sending %r messages", msg_count) self._backoff() diff --git a/journalpump/senders/elasticsearch.py b/journalpump/senders/elasticsearch.py index 03f3f79..be89c17 100644 --- a/journalpump/senders/elasticsearch.py +++ b/journalpump/senders/elasticsearch.py @@ -1,4 +1,4 @@ -from .base import LogSender +from .base import ThreadedLogSender from io import BytesIO from journalpump.util import default_json_serialization, get_requests_session @@ -7,7 +7,7 @@ import time -class ElasticsearchSender(LogSender): +class ElasticsearchSender(ThreadedLogSender): def __init__(self, *, config, **kwargs): super().__init__(config=config, max_send_interval=config.get("max_send_interval", 10.0), **kwargs) self.session_url = self.config.get("elasticsearch_url") diff --git a/journalpump/senders/file.py b/journalpump/senders/file.py index a864b7b..a695ac1 100644 --- a/journalpump/senders/file.py +++ b/journalpump/senders/file.py @@ -1,7 +1,7 @@ -from .base import LogSender +from .base import ThreadedLogSender -class FileSender(LogSender): +class FileSender(ThreadedLogSender): def __init__(self, *, config, **kwargs): super().__init__(config=config, max_send_interval=config.get("max_send_interval", 0.3), **kwargs) self.mark_disconnected() diff --git a/journalpump/senders/google_cloud_logging.py b/journalpump/senders/google_cloud_logging.py index 9eb9cbc..2069f81 100644 --- a/journalpump/senders/google_cloud_logging.py +++ b/journalpump/senders/google_cloud_logging.py @@ -1,4 +1,4 @@ -from .base import LogSender +from .base import ThreadedLogSender from googleapiclient.discovery import build from googleapiclient.errors import Error as GoogleApiClientError from oauth2client.service_account import ServiceAccountCredentials @@ -10,7 +10,7 @@ logging.getLogger("googleapiclient.discovery").setLevel(logging.WARNING) -class GoogleCloudLoggingSender(LogSender): +class GoogleCloudLoggingSender(ThreadedLogSender): _SEVERITY_MAPPING = { # mapping from journald priority to cloud logging severity 7: "DEBUG", 6: "INFO", diff --git a/journalpump/senders/kafka.py b/journalpump/senders/kafka.py index 4875da1..f4a951e 100644 --- a/journalpump/senders/kafka.py +++ b/journalpump/senders/kafka.py @@ -1,4 +1,4 @@ -from .base import LogSender +from .base import ThreadedLogSender from kafka import errors, KafkaAdminClient, KafkaProducer from kafka.admin import NewTopic @@ -23,7 +23,7 @@ logging.getLogger("kafka").setLevel(logging.CRITICAL) # remove client-internal tracebacks from logging output -class KafkaSender(LogSender): +class KafkaSender(ThreadedLogSender): def __init__(self, *, config, **kwargs): super().__init__(config=config, max_send_interval=config.get("max_send_interval", 0.3), **kwargs) self.kafka_producer = None diff --git a/journalpump/senders/logplex.py b/journalpump/senders/logplex.py index 1e37c96..5c11b84 100644 --- a/journalpump/senders/logplex.py +++ b/journalpump/senders/logplex.py @@ -1,11 +1,11 @@ -from .base import LogSender +from .base import ThreadedLogSender from journalpump.util import get_requests_session import datetime import json -class LogplexSender(LogSender): +class LogplexSender(ThreadedLogSender): def __init__(self, *, config, **kwargs): super().__init__(config=config, max_send_interval=config.get("max_send_interval", 5.0), **kwargs) self.logplex_input_url = config["logplex_log_input_url"] diff --git a/journalpump/senders/rsyslog.py b/journalpump/senders/rsyslog.py index 4af7620..f7c2640 100644 --- a/journalpump/senders/rsyslog.py +++ b/journalpump/senders/rsyslog.py @@ -1,4 +1,4 @@ -from .base import LogSender +from .base import ThreadedLogSender from journalpump.rsyslog import SyslogTcpClient import json @@ -8,7 +8,7 @@ RSYSLOG_CONN_ERRORS = (socket.timeout, ConnectionRefusedError) -class RsyslogSender(LogSender): +class RsyslogSender(ThreadedLogSender): def __init__(self, *, config, **kwargs): super().__init__(config=config, max_send_interval=config.get("max_send_interval", 0.3), **kwargs) self.rsyslog_client = None diff --git a/journalpump/senders/websocket.py b/journalpump/senders/websocket.py index 1120335..eb298bf 100644 --- a/journalpump/senders/websocket.py +++ b/journalpump/senders/websocket.py @@ -1,4 +1,4 @@ -from .base import LogSender, MAX_KAFKA_MESSAGE_SIZE +from .base import ThreadedLogSender, MAX_KAFKA_MESSAGE_SIZE from aiohttp_socks import ProxyConnectionError, ProxyError, ProxyTimeoutError from aiohttp_socks.utils import Proxy from concurrent.futures import CancelledError, TimeoutError as ConnectionTimeoutError @@ -51,9 +51,9 @@ def __init__( self.stop_event = asyncio.Event() self.stopped_event = asyncio.Event() self.running = True - # Send messages as batches, LogSender base class takes care of batch size. + # Send messages as batches, ThreadedLogSender base class takes care of batch size. # If batching is disabled, do not adjust the max_batch_size member variable, - # so that LogSender will still give us multiple messages in a single send_messages() call. + # so that ThreadedLogSender will still give us multiple messages in a single send_messages() call. self.batching_enabled = False if max_batch_size > 0: self.max_batch_size = max_batch_size @@ -81,7 +81,7 @@ async def async_send(self, messages): return False if self.batching_enabled: - # LogSender has limited the batch size already + # ThreadedLogSender has limited the batch size already batch = b"\x00".join(messages) messages = [batch] @@ -263,7 +263,7 @@ async def comms_channel_loop(self): self.log.info("Websocket closed") -class WebsocketSender(LogSender): +class WebsocketSender(ThreadedLogSender): def __init__(self, *, config, **kwargs): super().__init__(config=config, max_send_interval=config.get("max_send_interval", 1.0), **kwargs) self.runner = None