diff --git a/.gitignore b/.gitignore index c3800d8..e942e5a 100644 --- a/.gitignore +++ b/.gitignore @@ -45,6 +45,7 @@ htmlcov/ .coverage .coverage.* .cache +.local/ nosetests.xml coverage.xml *.cover diff --git a/README.md b/README.md index 39be9e4..0bfd506 100644 --- a/README.md +++ b/README.md @@ -9,6 +9,8 @@ splunk instance using the HTTP Event Collector (HEC). pip install python-splunk-logging ``` +Python 3.9 or newer is required. + ## Usage ### JsonFormatter @@ -101,7 +103,6 @@ Initialize the collector: ```python import os import socket -import time from splunk_logging.forwarders import HecForwarder hec = HecForwarder( @@ -130,6 +131,80 @@ hec.forward_event( ) ``` +`forward_events()` sends a sequence of events in one HEC request using Splunk's +[JSON event batching protocol](https://help.splunk.com/en/splunk-enterprise/get-started/get-data-in/9.2/get-data-with-http-event-collector/format-events-for-http-event-collector): + +```python +hec.forward_events(events) +``` + +Forwarders are context managers. A context manager or an explicit `close()` call should be used to release the +underlying HTTP client: + +```python +with HecForwarder(host="localhost", token=os.environ["HEC_TOKEN"]) as hec: + hec.forward_event(event) +``` + +Event delivery retries connection-establishment failures, HTTP 408 responses, and Splunk HEC 503 responses whose +body reports error code 9 (`Server is busy`). Ambiguous failures—including HTTP 429, 500, 502, generic 503, and 504 +responses and read/write transport errors—are raised without retrying because HEC might already have accepted the +events. + +#### Indexer acknowledgment + +Indexer acknowledgment is opt-in and remains blocking: each forwarding call waits until Splunk confirms that its +request has been indexed. The default 10-second polling interval and 5-minute timeout follow Splunk's +[HEC indexer acknowledgment guidance](https://help.splunk.com/en/data-management/get-data-in/get-data-into-splunk-enterprise/9.3/get-data-with-http-event-collector/about-http-event-collector-indexer-acknowledgment). + +```python +with HecForwarder( + host="localhost", + token=os.environ["HEC_ACK_TOKEN"], + indexer_ack=True, + ack_poll_interval=10, + ack_timeout=300, +) as hec: + hec.forward_events(events) +``` + +The HEC token must have indexer acknowledgment enabled. A UUID channel is generated per forwarder unless +`channel_id` is supplied. Missing, malformed, failed, or timed-out acknowledgments raise typed exceptions from +`splunk_logging.exceptions`. + +#### Background batching + +`BatchHecForwarder` adds an in-memory bounded queue and one background delivery worker while preserving the +`forward_event()` and `forward_events()` interfaces. It sends a batch when it reaches the event or byte limit, when +the flush interval expires, or when `flush()`/`close()` is called. + +```python +from splunk_logging.forwarders import BatchHecForwarder + +with BatchHecForwarder( + host="localhost", + token=os.environ["HEC_TOKEN"], + batch_size=100, + max_batch_bytes=1_048_576, + flush_interval=2, + max_queue_size=10_000, + max_queue_bytes=10_485_760, + enqueue_timeout=None, +) as hec: + hec.forward_events(events) + hec.flush() +``` + +`enqueue_timeout=None` applies backpressure until capacity is available. Set it to a number of seconds, including +zero for a non-blocking attempt, to raise `HecQueueFullError` instead. After a background delivery failure, the next +forwarding call, `flush()`, or `close()` raises `HecWorkerError`. With `indexer_ack=True`, the same worker waits for +the current batch acknowledgment before sending the next batch. + +> [!WARNING] +> The batch queue is memory-only. Events that have not been successfully flushed before process termination are +> lost. Always use the context manager or call `close()` (which performs a final flush) from the application's +> graceful shutdown path. Abrupt termination, including `SIGKILL`, cannot flush the queue. + ### HecHandler This is a logging handler that will forward a logging record directly to a HEC handler using python's built in logging library. @@ -182,7 +257,36 @@ logger.info( ) ``` -To prevent Splunk logging from slowing down the application, a queue can be used to buffer the log messages. +The handler can opt into the same background batching behavior without changing its default synchronous behavior: + +```python +hec_handler = HecHandler( + host="localhost", + token=os.environ["HEC_TOKEN"], + batch_enabled=True, + batch_size=100, + flush_interval=2, +) + +try: + root.addHandler(hec_handler) + logger.info({"a": 1, "b": 2}) +finally: + hec_handler.close() +``` + +`HecHandler.flush()` and `HecHandler.close()` delegate to the batch forwarder when batching is enabled. All +forwarder acknowledgment and queue options can also be passed to the handler. + +Python calls [`logging.shutdown()`](https://docs.python.org/3/library/logging.html#logging.shutdown) during normal +interpreter shutdown, which flushes and closes registered handlers. An unhandled `SIGINT` normally reaches that path +through `KeyboardInterrupt`; the default `SIGTERM` behavior does +[not run Python exit handlers](https://docs.python.org/3/library/atexit.html). Service applications should handle +`SIGTERM`, stop producing logs (and stop any `QueueListener`), then call `logging.shutdown()` from their graceful +shutdown path. Hard termination cannot be made lossless with an in-memory queue. + +An application-level logging queue can also be used when logging dispatch itself needs to be isolated from the +calling thread. See [Dealing with handlers that block](https://docs.python.org/3/howto/logging-cookbook.html#dealing-with-handlers-that-block) for details. ```python diff --git a/splunk_logging/exceptions.py b/splunk_logging/exceptions.py new file mode 100644 index 0000000..bfc1d96 --- /dev/null +++ b/splunk_logging/exceptions.py @@ -0,0 +1,69 @@ +# Copyright 2021 Splunk Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + + +class HecError(Exception): + """Base exception for HEC delivery errors.""" + + +class HecAckError(HecError): + """Raised when HEC indexer acknowledgment fails.""" + + +class HecAckTimeoutError(HecAckError): + """Raised when HEC does not acknowledge a request before its deadline.""" + + +class HecBatchError(HecError): + """Raised when HEC accepts only the events before an invalid batch event.""" + + def __init__(self, invalid_event_number: int, total_count: int, response): + self.invalid_event_number = invalid_event_number + self.accepted_count = invalid_event_number + self.total_count = total_count + self.response = response + super().__init__( + f"HEC rejected event {invalid_event_number} after accepting {self.accepted_count} " + f"of {total_count} batched events" + ) + + +class HecEventTooLargeError(HecError): + """Raised when one event exceeds the configured batch request limit.""" + + def __init__(self, event_size: int, max_batch_bytes: int): + self.event_size = event_size + self.max_batch_bytes = max_batch_bytes + super().__init__(f"Serialized HEC event is {event_size} bytes; limit is {max_batch_bytes} bytes") + + +class HecQueueFullError(HecError): + """Raised when a batch queue cannot accept more events before its deadline.""" + + def __init__(self, enqueued_count: int, next_event_index: int): + self.enqueued_count = enqueued_count + self.next_event_index = next_event_index + super().__init__( + f"HEC batch queue is full after accepting {enqueued_count} events; " + f"resume at event index {next_event_index}" + ) + + +class HecWorkerError(HecError): + """Raised after a background batch delivery fails.""" + + def __init__(self, message: str, cause: Exception, failed_events: tuple[dict, ...]): + self.cause = cause + self.failed_events = failed_events + super().__init__(message) diff --git a/splunk_logging/forwarders.py b/splunk_logging/forwarders.py index 5b1be6c..8396488 100644 --- a/splunk_logging/forwarders.py +++ b/splunk_logging/forwarders.py @@ -12,18 +12,58 @@ # See the License for the specific language governing permissions and # limitations under the License. +import json +import queue import random import socket +import threading import time import uuid +from dataclasses import dataclass from datetime import datetime from typing import Callable, Optional, Union import httpx from dateutil.parser import isoparse +from .exceptions import ( + HecAckError, + HecAckTimeoutError, + HecBatchError, + HecEventTooLargeError, + HecQueueFullError, + HecWorkerError, +) + + +@dataclass(frozen=True) +class _QueuedEvent: + payload: str + size: int + + +@dataclass +class _FlushRequest: + completed: bool = False + + +class _RequestDeadlineExceededError(Exception): + """Raised internally when an HTTP retry deadline expires.""" + class HecForwarder: + _RETRY_STATUS_CODES = frozenset( + [ + 408, # Request Timeout + ] + ) + _RETRY_HEC_ERROR_CODES = frozenset( + [ + 9, # Server is busy + ] + ) + _RETRY_CONNECTION_ERRORS = (httpx.ConnectError, httpx.ConnectTimeout) + def __init__( self, host: str = "localhost", @@ -35,6 +75,11 @@ def __init__( default_source: Optional[str] = None, default_sourcetype: Optional[str] = None, default_index: Optional[str] = None, + *, + indexer_ack: bool = False, + channel_id: Optional[str] = None, + ack_poll_interval: float = 10.0, + ack_timeout: float = 300.0, ): """ Initializes a HecForwarder instance for sending events to a Splunk HEC listener. @@ -58,6 +103,14 @@ def __init__( Default sourcetype to use when forwarding events. (default: "") default_index (str, optional): Default index to send events. (default: "") + indexer_ack (bool, optional): + Wait for Splunk indexer acknowledgment after sending events. (default: False) + channel_id (str, optional): + GUID used for indexer acknowledgment requests. (default: generated UUID) + ack_poll_interval (float, optional): + Seconds to wait between acknowledgment queries. (default: 10) + ack_timeout (float, optional): + Seconds to wait for an acknowledgment before raising. (default: 300) """ self._host = host self._port = port @@ -68,56 +121,125 @@ def __init__( self._default_source = default_source or "" self._default_sourcetype = default_sourcetype or "" self._default_index = default_index or "" + if channel_id is not None: + uuid.UUID(channel_id) + self._channel_id = channel_id or str(uuid.uuid4()) + self._indexer_ack = indexer_ack + self._ack_poll_interval = ack_poll_interval + self._ack_timeout = ack_timeout + self._backoff_factor = 1 + self._max_retries = 3 self._client = self._create_client() return - def __del__(self): - if self._client: - self._client.close() - return + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_value, traceback): + self.close() + + def close(self): + self._client.close() def _create_client(self) -> httpx.Client: scheme = "https" if self._use_ssl else "http" - client = httpx.Client( + return httpx.Client( base_url=httpx.URL(scheme=scheme, host=self._host, port=self._port), verify=self._verify_ssl, headers={"Authorization": f"Splunk {self._token}"}, - event_hooks={"response": [self._client_retry]}, ) - client.backoff_factor = 1 - client.max_retries = 3 - client.retry_attempt = 0 - return client + def _request( + self, + method: str, + path: str, + *, + deadline: Optional[float] = None, + **kwargs, + ) -> httpx.Response: + retry_attempt = 0 + while True: + request_kwargs = kwargs + if deadline is not None: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise _RequestDeadlineExceededError + request_kwargs = dict(kwargs) + request_kwargs.setdefault("timeout", self._timeout_capped_by(remaining)) - def _client_retry(self, response: httpx.Response): - """ - Handles retry logic for HTTP requests based on response status codes. + try: + response = self._client.request(method, path, **request_kwargs) + except self._RETRY_CONNECTION_ERRORS: + if retry_attempt >= self._max_retries: + raise + self._sleep_before_retry(retry_attempt, deadline=deadline) + retry_attempt += 1 + continue - Args: - response (httpx.Response): The HTTP response object. + if deadline is not None and time.monotonic() >= deadline: + raise _RequestDeadlineExceededError + if not self._should_retry_response(response): + response.raise_for_status() + return response - Retries the request if the response status code indicates a transient error, - otherwise raises for status. - """ - retry_codes = [ - httpx.codes.REQUEST_TIMEOUT, - httpx.codes.TOO_MANY_REQUESTS, - httpx.codes.INTERNAL_SERVER_ERROR, - httpx.codes.SERVICE_UNAVAILABLE, - ] - if response.status_code in retry_codes and self._client.retry_attempt < self._client.max_retries: - time.sleep( - self._client.backoff_factor * 2**self._client.retry_attempt - + random.uniform(0, self._client.retry_attempt) + if retry_attempt >= self._max_retries: + response.raise_for_status() + + self._sleep_before_retry( + retry_attempt, + deadline=deadline, + retry_after=response.headers.get("Retry-After"), ) - self._client.retry_attempt += 1 - self._client.send(response.request) - else: - self._client.retry_attempt = 0 - response.raise_for_status() - return + retry_attempt += 1 + + def _should_retry_response(self, response: httpx.Response) -> bool: + if response.status_code in self._RETRY_STATUS_CODES: + return True + if response.status_code != httpx.codes.SERVICE_UNAVAILABLE: + return False + try: + response_data = response.json() + return response_data.get("code") in self._RETRY_HEC_ERROR_CODES + except (AttributeError, TypeError, ValueError): + return False + + def _sleep_before_retry( + self, + retry_attempt: int, + *, + deadline: Optional[float], + retry_after: Optional[str] = None, + ): + try: + delay = float(retry_after) if retry_after is not None else None + except ValueError: + delay = None + + if delay is None: + delay = self._backoff_factor * 2**retry_attempt + random.uniform(0, retry_attempt) + + if deadline is not None: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise _RequestDeadlineExceededError + delay = min(delay, remaining) + + time.sleep(delay) + if deadline is not None and time.monotonic() >= deadline: + raise _RequestDeadlineExceededError + + def _timeout_capped_by(self, seconds: float) -> httpx.Timeout: + def cap(value: Optional[float]) -> float: + return seconds if value is None else min(value, seconds) + + timeout = self._client.timeout + return httpx.Timeout( + connect=cap(timeout.connect), + read=cap(timeout.read), + write=cap(timeout.write), + pool=cap(timeout.pool), + ) def _parse_timestamp(self, timeinfo: Union[str, int, float, datetime], timefmt: Optional[str] = None) -> float: """ @@ -180,6 +302,72 @@ def forward_event( index to send this event to. This will overwrite the default index configured for this event only (default: default_index) """ + hec_event = self._build_hec_event(event, eventtime=eventtime, timefmt=timefmt, **kwargs) + + headers = self._event_headers() + + response = self._request("POST", "/services/collector/event", headers=headers, json=hec_event) + self._wait_for_ack_if_enabled(response) + return + + def _event_headers(self) -> dict[str, str]: + return {"X-Splunk-Request-Channel": self._channel_id} + + def _wait_for_ack_if_enabled(self, response: httpx.Response): + """Wait for the response's indexer acknowledgment when configured.""" + if not self._indexer_ack: + return + + try: + response_data = response.json() + ack_id = response_data.get("ackId", response_data.get("ackID")) + except (AttributeError, TypeError, ValueError) as error: + raise HecAckError("HEC returned an invalid acknowledgment response") from error + if ack_id is None: + raise HecAckError("HEC response did not include an acknowledgment ID") + try: + normalized_ack_id = int(ack_id) + except (TypeError, ValueError) as error: + raise HecAckError(f"HEC returned an invalid acknowledgment ID: {ack_id!r}") from error + self._wait_for_acknowledgments([normalized_ack_id]) + + def _wait_for_acknowledgments(self, ack_ids: list[int]): + deadline = time.monotonic() + self._ack_timeout + headers = {"X-Splunk-Request-Channel": self._channel_id} + while True: + try: + response = self._request( + "POST", + "/services/collector/ack", + headers=headers, + json={"acks": ack_ids}, + deadline=deadline, + ) + except _RequestDeadlineExceededError as error: + raise HecAckTimeoutError(f"Timed out waiting for acknowledgments: {ack_ids}") from error + except httpx.HTTPError as error: + if time.monotonic() >= deadline: + raise HecAckTimeoutError(f"Timed out waiting for acknowledgments: {ack_ids}") from error + raise HecAckError(f"Failed to query HEC acknowledgments: {ack_ids}") from error + try: + ack_statuses = response.json()["acks"] + statuses = [ack_statuses[str(ack_id)] for ack_id in ack_ids] + except (KeyError, TypeError, ValueError) as error: + raise HecAckError("HEC returned an invalid acknowledgment response") from error + if all(statuses): + return + remaining = deadline - time.monotonic() + if remaining <= 0: + raise HecAckTimeoutError(f"Timed out waiting for acknowledgments: {ack_ids}") + time.sleep(min(self._ack_poll_interval, remaining)) + + def _build_hec_event( + self, + event: dict, + eventtime: Optional[Union[str, int, float, datetime]] = None, + timefmt: Optional[str] = None, + **kwargs, + ) -> dict: eventtime = eventtime or datetime.now() host = kwargs.get("host", self._default_host) source = kwargs.get("source", self._default_source) @@ -199,12 +387,7 @@ def forward_event( hec_event["index"] = index hec_event["time"] = str(self._parse_timestamp(eventtime, timefmt=timefmt)) - - headers = {"X-Splunk-Request-Channel": str(uuid.uuid1())} - - resp = self._client.post("/services/collector/event", headers=headers, json=hec_event) - resp.raise_for_status() - return + return hec_event def forward_events( self, @@ -240,6 +423,265 @@ def forward_events( index to send this event to. This will overwrite the default index configured for all events in this list. (default: default_index) """ - for event in events: - self.forward_event(event, eventtime=eventtime(event), timefmt=timefmt, **kwargs) + if not events: + return + + hec_events = [ + self._build_hec_event(event, eventtime=eventtime(event), timefmt=timefmt, **kwargs) for event in events + ] + self._send_hec_events(hec_events) + return + + def _send_hec_events(self, hec_events: list[dict]): + payload = "".join(self._serialize_hec_event(hec_event) for hec_event in hec_events) + self._send_hec_payload(payload, event_count=len(hec_events)) + + def _serialize_hec_event(self, hec_event: dict) -> str: + return json.dumps(hec_event) + + def _send_hec_payload(self, payload: str, *, event_count: int): + headers = { + "Content-Type": "application/json", + **self._event_headers(), + } + try: + response = self._request("POST", "/services/collector/event", headers=headers, content=payload) + except httpx.HTTPStatusError as error: + try: + invalid_event_number = int(error.response.json()["invalid-event-number"]) + except (KeyError, TypeError, ValueError): + raise error + if not 0 <= invalid_event_number < event_count: + raise error + raise HecBatchError(invalid_event_number, event_count, error.response) from error + self._wait_for_ack_if_enabled(response) + + +class BatchHecForwarder(HecForwarder): + def __init__( + self, + *args, + batch_size: int = 100, + max_batch_bytes: int = 1_048_576, + flush_interval: float = 2.0, + max_queue_size: int = 10_000, + max_queue_bytes: int = 10_485_760, + enqueue_timeout: Optional[float] = None, + **kwargs, + ): + limits = { + "batch_size": batch_size, + "max_batch_bytes": max_batch_bytes, + "flush_interval": flush_interval, + "max_queue_size": max_queue_size, + "max_queue_bytes": max_queue_bytes, + } + for name, value in limits.items(): + if value <= 0: + raise ValueError(f"{name} must be greater than zero") + if enqueue_timeout is not None and enqueue_timeout < 0: + raise ValueError("enqueue_timeout cannot be negative") + + super().__init__(*args, **kwargs) + self._batch_size = batch_size + self._max_batch_bytes = max_batch_bytes + self._flush_interval = flush_interval + self._max_queue_size = max_queue_size + self._max_queue_bytes = max_queue_bytes + self._enqueue_timeout = enqueue_timeout + self._queue = queue.Queue() + self._capacity = threading.Condition() + self._producer_lock = threading.RLock() + self._pending_count = 0 + self._pending_bytes = 0 + self._worker_error = None + self._failed_events = () + self._accepting = True + self._stop = object() + self._closed = False + self._worker_thread = threading.Thread(target=self._worker, name="splunk-hec-batch", daemon=True) + self._worker_thread.start() + + def forward_event( + self, + event: dict, + eventtime: Optional[Union[str, int, float, datetime]] = None, + timefmt: Optional[str] = None, + **kwargs, + ): + queued_event = self._prepare_queued_event(event, eventtime=eventtime, timefmt=timefmt, **kwargs) + deadline = None if self._enqueue_timeout is None else time.monotonic() + self._enqueue_timeout + with self._producer_lock: + self._enqueue(queued_event, deadline=deadline, enqueued_count=0, next_event_index=0) + + def forward_events( + self, + events: list[dict], + eventtime: Callable = lambda _: datetime.now(), + timefmt: Optional[str] = None, + **kwargs, + ): + queued_events = [ + self._prepare_queued_event(event, eventtime=eventtime(event), timefmt=timefmt, **kwargs) for event in events + ] + deadline = None if self._enqueue_timeout is None else time.monotonic() + self._enqueue_timeout + with self._producer_lock: + for index, queued_event in enumerate(queued_events): + self._enqueue( + queued_event, + deadline=deadline, + enqueued_count=index, + next_event_index=index, + ) + + def _enqueue( + self, + queued_event: _QueuedEvent, + *, + deadline: Optional[float], + enqueued_count: int, + next_event_index: int, + ): + with self._capacity: + self._raise_if_worker_failed() + if not self._accepting: + raise RuntimeError("BatchHecForwarder is closed") + while ( + self._pending_count + 1 > self._max_queue_size + or self._pending_bytes + queued_event.size > self._max_queue_bytes + ): + if deadline is not None: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise HecQueueFullError(enqueued_count, next_event_index) + else: + remaining = None + self._capacity.wait(timeout=remaining) + self._raise_if_worker_failed() + if not self._accepting: + raise RuntimeError("BatchHecForwarder is closed") + + self._pending_count += 1 + self._pending_bytes += queued_event.size + self._queue.put(queued_event) + + def _prepare_queued_event( + self, + event: dict, + eventtime: Optional[Union[str, int, float, datetime]] = None, + timefmt: Optional[str] = None, + **kwargs, + ) -> _QueuedEvent: + envelope = self._build_hec_event(event, eventtime=eventtime, timefmt=timefmt, **kwargs) + payload = self._serialize_hec_event(envelope) + event_size = len(payload.encode("utf-8")) + event_limit = min(self._max_batch_bytes, self._max_queue_bytes) + if event_size > event_limit: + raise HecEventTooLargeError(event_size, event_limit) + return _QueuedEvent(payload=payload, size=event_size) + + def flush(self): + with self._producer_lock: + with self._capacity: + self._raise_if_worker_failed() + if self._pending_count == 0: + return + request = _FlushRequest() + self._queue.put(request) + + with self._capacity: + while not request.completed: + self._raise_if_worker_failed() + self._capacity.wait() + self._raise_if_worker_failed() + + def close(self): + if self._closed: + self._raise_if_worker_failed() + return + + with self._producer_lock: + self._accepting = False + + delivery_error = None + try: + self.flush() + except HecWorkerError as error: + delivery_error = error + + if self._worker_thread.is_alive(): + self._queue.put(self._stop) + self._worker_thread.join() + super().close() + self._closed = True + if delivery_error is not None: + raise delivery_error + + def _raise_if_worker_failed(self): + if self._worker_error is not None: + raise HecWorkerError( + "Background HEC batch delivery failed", + self._worker_error, + self._failed_events, + ) from self._worker_error + + def _worker(self): + deferred_item = None + while True: + if deferred_item is None: + item = self._queue.get() + else: + item = deferred_item + deferred_item = None + if item is self._stop: + return + if isinstance(item, _FlushRequest): + with self._capacity: + item.completed = True + self._capacity.notify_all() + continue + + batch = [item] + batch_bytes = item.size + deadline = time.monotonic() + self._flush_interval + flush_request = None + while len(batch) < self._batch_size: + timeout = deadline - time.monotonic() + if timeout <= 0: + break + try: + next_item = self._queue.get(timeout=timeout) + except queue.Empty: + break + if isinstance(next_item, _FlushRequest): + flush_request = next_item + break + if batch_bytes + next_item.size > self._max_batch_bytes: + deferred_item = next_item + break + batch.append(next_item) + batch_bytes += next_item.size + + try: + self._send_hec_payload( + "".join(queued_event.payload for queued_event in batch), + event_count=len(batch), + ) + except Exception as error: + with self._capacity: + self._worker_error = error + if isinstance(error, HecBatchError): + failed_batch = batch[error.accepted_count :] + else: + failed_batch = batch + self._failed_events = tuple(json.loads(queued_event.payload) for queued_event in failed_batch) + self._capacity.notify_all() + return + + with self._capacity: + self._pending_count -= len(batch) + self._pending_bytes -= sum(queued_event.size for queued_event in batch) + if flush_request is not None: + flush_request.completed = True + self._capacity.notify_all() return diff --git a/splunk_logging/handlers.py b/splunk_logging/handlers.py index 21c5a3e..e14f637 100644 --- a/splunk_logging/handlers.py +++ b/splunk_logging/handlers.py @@ -19,7 +19,7 @@ from typing import Optional from .formatters import JsonFormatter -from .forwarders import HecForwarder +from .forwarders import BatchHecForwarder, HecForwarder class HecHandler(logging.Handler): @@ -35,6 +35,18 @@ def __init__( default_sourcetype: Optional[str] = None, default_index: Optional[str] = None, ignore_exceptions: bool = True, + *, + indexer_ack: bool = False, + channel_id: Optional[str] = None, + ack_poll_interval: float = 10.0, + ack_timeout: float = 300.0, + batch_enabled: bool = False, + batch_size: int = 100, + max_batch_bytes: int = 1_048_576, + flush_interval: float = 2.0, + max_queue_size: int = 10_000, + max_queue_bytes: int = 10_485_760, + enqueue_timeout: Optional[float] = None, **kwargs, ): """ @@ -60,25 +72,61 @@ def __init__( default_index (str, optional): Default index to send events to. (default: None) ignore_exceptions (bool, optional): - Ignore exceptions thrown when sending an events. (default: False) + Ignore exceptions thrown when sending an events. (default: True) + indexer_ack (bool, optional): + Wait for Splunk indexer acknowledgment after sending events. (default: False) + channel_id (str, optional): + GUID used for indexer acknowledgment requests. (default: generated UUID) + ack_poll_interval (float, optional): + Seconds to wait between acknowledgment queries. (default: 10) + ack_timeout (float, optional): + Seconds to wait for an acknowledgment before raising. (default: 300) + batch_enabled (bool, optional): + Queue log records and send them in background batches. (default: False) + batch_size (int, optional): + Maximum events per request when batching. (default: 100) + max_batch_bytes (int, optional): + Maximum serialized bytes per batch request. (default: 1048576) + flush_interval (float, optional): + Maximum seconds to hold a partial batch. (default: 2) + max_queue_size (int, optional): + Maximum queued and in-flight event count. (default: 10000) + max_queue_bytes (int, optional): + Maximum queued and in-flight serialized bytes. (default: 10485760) + enqueue_timeout (float, optional): + Maximum seconds to wait for queue capacity. (default: wait indefinitely) """ super().__init__(**kwargs) self.setFormatter(JsonFormatter()) - - self._hec = HecForwarder( - host=host, - port=port, - token=token or os.environ.get("HEC_TOKEN", ""), - use_ssl=use_ssl, - verify_ssl=verify_ssl, - default_host=default_host or socket.gethostname(), - default_source=default_source or "", - default_sourcetype=default_sourcetype or "", - default_index=default_index or "", - ) - self._ignore_exceptions = ignore_exceptions + forwarder_class = BatchHecForwarder if batch_enabled else HecForwarder + forwarder_kwargs = { + "host": host, + "port": port, + "token": token or os.environ.get("HEC_TOKEN", ""), + "use_ssl": use_ssl, + "verify_ssl": verify_ssl, + "default_host": default_host or socket.gethostname(), + "default_source": default_source or "", + "default_sourcetype": default_sourcetype or "", + "default_index": default_index or "", + "indexer_ack": indexer_ack, + "channel_id": channel_id, + "ack_poll_interval": ack_poll_interval, + "ack_timeout": ack_timeout, + } + if batch_enabled: + forwarder_kwargs.update( + batch_size=batch_size, + max_batch_bytes=max_batch_bytes, + flush_interval=flush_interval, + max_queue_size=max_queue_size, + max_queue_bytes=max_queue_bytes, + enqueue_timeout=enqueue_timeout, + ) + self._hec = forwarder_class(**forwarder_kwargs) + # urllib3/httpx message creates log recortds when an http connection is established. # This can lead to an infinite loop when sending logs to HEC, so we disable it # if using a hec logger. @@ -114,3 +162,31 @@ def emit(self, record: logging.LogRecord): else: raise e return + + def flush(self): + forwarder = getattr(self, "_hec", None) + flush = getattr(forwarder, "flush", None) + if flush is None: + return + try: + flush() + except Exception as error: + if getattr(self, "_ignore_exceptions", True): + print(f"Exception when flushing logs to Splunk: {error}") + else: + raise error + + def close(self): + delivery_error = None + forwarder = getattr(self, "_hec", None) + if forwarder is not None: + try: + forwarder.close() + except Exception as error: + delivery_error = error + super().close() + if delivery_error is not None: + if getattr(self, "_ignore_exceptions", True): + print(f"Exception when closing Splunk logging handler: {delivery_error}") + else: + raise delivery_error diff --git a/test/test_forwarder.py b/test/test_forwarder.py index b06948b..94dcd3b 100644 --- a/test/test_forwarder.py +++ b/test/test_forwarder.py @@ -12,15 +12,20 @@ # See the License for the specific language governing permissions and # limitations under the License. +import json import random +import threading import unittest from datetime import datetime +from unittest.mock import patch import httpx import respx from splunk_logging.forwarders import HecForwarder +TEST_CHANNEL_ID = "00000000-0000-4000-8000-000000000001" + class TestHecForwarder(unittest.TestCase): def create_forwarder(self) -> HecForwarder: @@ -44,20 +49,867 @@ def test_forward_event(self): return @respx.mock - def test_retry(self): + def test_forward_event_preserves_channel_header_without_ack_polling(self): + hec = HecForwarder( + host="localhost", + port=8088, + token="", + use_ssl=False, + channel_id=TEST_CHANNEL_ID, + ) + route = respx.post("http://localhost:8088/services/collector/event") + route.return_value = httpx.Response(200, json={"text": "Success", "code": 0, "ackID": 7}) + + hec.forward_event({"message": "test"}) + + self.assertEqual(route.calls.last.request.headers["X-Splunk-Request-Channel"], TEST_CHANNEL_ID) + + @respx.mock + def test_forward_events_sends_one_batched_request(self): + hec = self.create_forwarder() + route = respx.post("http://localhost:8088/services/collector/event") + route.return_value = httpx.Response(200, json={"text": "Success", "code": 0}) + events = [{"message": "first"}, {"message": "second"}] + event_times = { + "first": datetime(2026, 1, 1, 12, 0, 0), + "second": datetime(2026, 1, 1, 12, 0, 1), + } + + hec.forward_events(events, eventtime=lambda event: event_times[event["message"]]) + + self.assertEqual(route.call_count, 1) + payload = route.calls.last.request.content.decode() + decoder = json.JSONDecoder() + first, offset = decoder.raw_decode(payload) + second, offset = decoder.raw_decode(payload, offset) + self.assertEqual(offset, len(payload)) + self.assertEqual(first["event"], events[0]) + self.assertEqual(second["event"], events[1]) + self.assertEqual(first["time"], str(event_times["first"].timestamp())) + self.assertEqual(second["time"], str(event_times["second"].timestamp())) + + @respx.mock + def test_forward_events_with_no_events_sends_no_request(self): + hec = self.create_forwarder() + route = respx.post("http://localhost:8088/services/collector/event") + + hec.forward_events([]) + + self.assertFalse(route.called) + + @respx.mock + def test_forward_events_reports_partial_batch_acceptance(self): + from splunk_logging.exceptions import HecBatchError + + hec = self.create_forwarder() + route = respx.post("http://localhost:8088/services/collector/event") + route.return_value = httpx.Response( + 400, + json={"text": "Incorrect data format", "code": 5, "invalid-event-number": 1}, + ) + + with self.assertRaises(HecBatchError) as raised: + hec.forward_events([{"message": "first"}, {"invalid": True}, {"message": "third"}]) + + self.assertEqual(raised.exception.invalid_event_number, 1) + self.assertEqual(raised.exception.accepted_count, 1) + self.assertEqual(raised.exception.total_count, 3) + + @respx.mock + def test_forward_event_waits_for_indexer_acknowledgment(self): + hec = HecForwarder( + host="localhost", + port=8088, + token="", + use_ssl=False, + indexer_ack=True, + channel_id=TEST_CHANNEL_ID, + ack_poll_interval=0, + ) + event_route = respx.post("http://localhost:8088/services/collector/event") + event_route.return_value = httpx.Response(200, json={"text": "Success", "code": 0, "ackId": 7}) + ack_route = respx.post("http://localhost:8088/services/collector/ack") + ack_route.return_value = httpx.Response(200, json={"acks": {"7": True}}) + + hec.forward_event({"message": "test"}) + + self.assertEqual(event_route.calls.last.request.headers["X-Splunk-Request-Channel"], TEST_CHANNEL_ID) + self.assertEqual(ack_route.calls.last.request.headers["X-Splunk-Request-Channel"], TEST_CHANNEL_ID) + self.assertEqual(json.loads(ack_route.calls.last.request.content), {"acks": [7]}) + + @respx.mock + @patch("splunk_logging.forwarders.time.sleep") + def test_forward_event_polls_until_indexer_acknowledges(self, sleep): + hec = HecForwarder( + host="localhost", + port=8088, + token="", + use_ssl=False, + indexer_ack=True, + ) + event_route = respx.post("http://localhost:8088/services/collector/event") + event_route.return_value = httpx.Response(200, json={"text": "Success", "code": 0, "ackId": 7}) + ack_route = respx.post("http://localhost:8088/services/collector/ack") + ack_route.side_effect = [ + httpx.Response(200, json={"acks": {"7": False}}), + httpx.Response(200, json={"acks": {"7": True}}), + ] + + hec.forward_event({"message": "test"}) + + self.assertEqual(ack_route.call_count, 2) + sleep.assert_called_once_with(10.0) + + @respx.mock + @patch("splunk_logging.forwarders.time.monotonic", side_effect=[0.0, 0.0, 1.0]) + def test_forward_event_raises_typed_acknowledgment_timeout(self, _monotonic): + from splunk_logging.exceptions import HecAckTimeoutError + + hec = HecForwarder( + host="localhost", + port=8088, + token="", + use_ssl=False, + indexer_ack=True, + ack_timeout=1.0, + ) + event_route = respx.post("http://localhost:8088/services/collector/event") + event_route.return_value = httpx.Response(200, json={"text": "Success", "code": 0, "ackId": 7}) + ack_route = respx.post("http://localhost:8088/services/collector/ack") + ack_route.return_value = httpx.Response(200, json={"acks": {"7": False}}) + + with self.assertRaises(HecAckTimeoutError): + hec.forward_event({"message": "test"}) + + @respx.mock + def test_ack_timeout_caps_poll_interval(self): + from splunk_logging.exceptions import HecAckTimeoutError + + clock = [0.0] + hec = HecForwarder( + host="localhost", + port=8088, + token="", + use_ssl=False, + indexer_ack=True, + ack_poll_interval=10.0, + ack_timeout=1.0, + ) + event_route = respx.post("http://localhost:8088/services/collector/event") + event_route.return_value = httpx.Response(200, json={"text": "Success", "code": 0, "ackID": 7}) + ack_route = respx.post("http://localhost:8088/services/collector/ack") + ack_route.return_value = httpx.Response(200, json={"acks": {"7": False}}) + + with ( + patch("splunk_logging.forwarders.time.monotonic", side_effect=lambda: clock[0]), + patch( + "splunk_logging.forwarders.time.sleep", side_effect=lambda delay: clock.__setitem__(0, clock[0] + delay) + ) as sleep, + self.assertRaises(HecAckTimeoutError), + ): + hec.forward_event({"message": "test"}) + + self.assertEqual(ack_route.call_count, 1) + sleep.assert_called_once_with(1.0) + + @respx.mock + def test_ack_timeout_caps_retry_after(self): + from splunk_logging.exceptions import HecAckTimeoutError + + clock = [0.0] + hec = HecForwarder( + host="localhost", + port=8088, + token="", + use_ssl=False, + indexer_ack=True, + ack_timeout=1.0, + ) + event_route = respx.post("http://localhost:8088/services/collector/event") + event_route.return_value = httpx.Response(200, json={"text": "Success", "code": 0, "ackID": 7}) + ack_route = respx.post("http://localhost:8088/services/collector/ack") + ack_route.return_value = httpx.Response( + 503, + headers={"Retry-After": "10"}, + json={"text": "Server is busy", "code": 9}, + ) + + with ( + patch("splunk_logging.forwarders.time.monotonic", side_effect=lambda: clock[0]), + patch( + "splunk_logging.forwarders.time.sleep", side_effect=lambda delay: clock.__setitem__(0, clock[0] + delay) + ) as sleep, + self.assertRaises(HecAckTimeoutError), + ): + hec.forward_event({"message": "test"}) + + self.assertEqual(ack_route.call_count, 1) + sleep.assert_called_once_with(1.0) + + @respx.mock + def test_forward_event_raises_ack_error_when_response_has_no_ack_id(self): + from splunk_logging.exceptions import HecAckError + + hec = HecForwarder( + host="localhost", + port=8088, + token="", + use_ssl=False, + indexer_ack=True, + ) + event_route = respx.post("http://localhost:8088/services/collector/event") + event_route.return_value = httpx.Response(200, json={"text": "Success", "code": 0}) + + with self.assertRaises(HecAckError): + hec.forward_event({"message": "test"}) + + @respx.mock + def test_forward_event_raises_ack_error_when_response_is_not_json(self): + from splunk_logging.exceptions import HecAckError + + hec = HecForwarder( + host="localhost", + port=8088, + token="", + use_ssl=False, + indexer_ack=True, + ) + event_route = respx.post("http://localhost:8088/services/collector/event") + event_route.return_value = httpx.Response(200, text="not json") + + with self.assertRaises(HecAckError): + hec.forward_event({"message": "test"}) + + @respx.mock + def test_forward_event_raises_ack_error_when_status_is_missing(self): + from splunk_logging.exceptions import HecAckError + + hec = HecForwarder( + host="localhost", + port=8088, + token="", + use_ssl=False, + indexer_ack=True, + ) + event_route = respx.post("http://localhost:8088/services/collector/event") + event_route.return_value = httpx.Response(200, json={"text": "Success", "code": 0, "ackId": 7}) + ack_route = respx.post("http://localhost:8088/services/collector/ack") + ack_route.return_value = httpx.Response(200, json={"acks": {}}) + + with self.assertRaises(HecAckError): + hec.forward_event({"message": "test"}) + + @respx.mock + def test_forward_event_wraps_ack_request_http_failure(self): + from splunk_logging.exceptions import HecAckError + + hec = HecForwarder( + host="localhost", + port=8088, + token="", + use_ssl=False, + indexer_ack=True, + ) + event_route = respx.post("http://localhost:8088/services/collector/event") + event_route.return_value = httpx.Response(200, json={"text": "Success", "code": 0, "ackID": 7}) + ack_route = respx.post("http://localhost:8088/services/collector/ack") + ack_route.return_value = httpx.Response(400, json={"text": "ACK is disabled", "code": 14}) + + with self.assertRaises(HecAckError) as raised: + hec.forward_event({"message": "test"}) + + self.assertIsInstance(raised.exception.__cause__, httpx.HTTPStatusError) + + @respx.mock + @patch("splunk_logging.forwarders.time.sleep") + def test_forward_event_wraps_ack_request_transport_failure(self, _sleep): + from splunk_logging.exceptions import HecAckError + + hec = HecForwarder( + host="localhost", + port=8088, + token="", + use_ssl=False, + indexer_ack=True, + ) + event_route = respx.post("http://localhost:8088/services/collector/event") + event_route.return_value = httpx.Response(200, json={"text": "Success", "code": 0, "ackID": 7}) + ack_route = respx.post("http://localhost:8088/services/collector/ack") + ack_route.side_effect = httpx.ConnectError("connection failed") + + with self.assertRaises(HecAckError) as raised: + hec.forward_event({"message": "test"}) + + self.assertIsInstance(raised.exception.__cause__, httpx.ConnectError) + + def test_context_manager_closes_forwarder(self): + hec = self.create_forwarder() + + with hec as entered: + self.assertIs(entered, hec) + + with self.assertRaises(RuntimeError): + hec.forward_event({"message": "test"}) + + def test_batch_forwarder_rejects_invalid_limits(self): + from splunk_logging.forwarders import BatchHecForwarder + + invalid_options = [ + {"batch_size": 0}, + {"max_batch_bytes": 0}, + {"flush_interval": 0}, + {"max_queue_size": 0}, + {"max_queue_bytes": 0}, + {"enqueue_timeout": -1}, + ] + for options in invalid_options: + with self.subTest(options=options), self.assertRaises(ValueError): + BatchHecForwarder(host="localhost", token="", use_ssl=False, **options) + + @respx.mock + def test_batch_forwarder_flushes_one_request_at_count_threshold(self): + from splunk_logging.forwarders import BatchHecForwarder + + route = respx.post("http://localhost:8088/services/collector/event") + route.return_value = httpx.Response(200, json={"text": "Success", "code": 0}) + + with BatchHecForwarder( + host="localhost", + port=8088, + token="", + use_ssl=False, + batch_size=2, + flush_interval=60, + ) as hec: + hec.forward_event({"message": "first"}) + hec.forward_event({"message": "second"}) + hec.flush() + + self.assertEqual(route.call_count, 1) + payload = route.calls.last.request.content.decode() + decoder = json.JSONDecoder() + first, offset = decoder.raw_decode(payload) + second, offset = decoder.raw_decode(payload, offset) + self.assertEqual(offset, len(payload)) + self.assertEqual(first["event"], {"message": "first"}) + self.assertEqual(second["event"], {"message": "second"}) + + @respx.mock + def test_batch_forwarder_explicit_flush_does_not_wait_for_interval(self): + from splunk_logging.forwarders import BatchHecForwarder + + sent = threading.Event() + route = respx.post("http://localhost:8088/services/collector/event") + + def mark_sent(_request): + sent.set() + return httpx.Response(200, json={"text": "Success", "code": 0}) + + route.side_effect = mark_sent + hec = BatchHecForwarder( + host="localhost", + port=8088, + token="", + use_ssl=False, + batch_size=2, + flush_interval=60, + ) + hec.forward_event({"message": "first"}) + flush_thread = threading.Thread(target=hec.flush) + flush_thread.start() + + sent_before_interval = sent.wait(0.2) + if not sent_before_interval: + hec.forward_event({"message": "release worker"}) + flush_thread.join(1) + hec.close() + + self.assertTrue(sent_before_interval) + + @respx.mock + def test_batch_forwarder_flushes_after_interval(self): + from splunk_logging.forwarders import BatchHecForwarder + + sent = threading.Event() + route = respx.post("http://localhost:8088/services/collector/event") + + def mark_sent(_request): + sent.set() + return httpx.Response(200, json={"text": "Success", "code": 0}) + + route.side_effect = mark_sent + hec = BatchHecForwarder( + host="localhost", + port=8088, + token="", + use_ssl=False, + batch_size=100, + flush_interval=0.05, + ) + hec.forward_event({"message": "test"}) + + sent_after_interval = sent.wait(0.2) + if not sent_after_interval: + hec.flush() + hec.close() + + self.assertTrue(sent_after_interval) + + @respx.mock + def test_batch_forwarder_splits_requests_at_byte_limit(self): + from splunk_logging.forwarders import BatchHecForwarder + + route = respx.post("http://localhost:8088/services/collector/event") + route.return_value = httpx.Response(200, json={"text": "Success", "code": 0}) + hec = BatchHecForwarder( + host="localhost", + port=8088, + token="", + use_ssl=False, + batch_size=100, + max_batch_bytes=700, + flush_interval=60, + ) + + hec.forward_events( + [{"message": "x" * 500}, {"message": "y" * 500}], + eventtime=lambda _: datetime(2026, 1, 1), + ) + hec.flush() + hec.close() + + self.assertEqual(route.call_count, 2) + + @respx.mock + def test_batch_forwarder_rejects_event_larger_than_batch_limit(self): + from splunk_logging.exceptions import HecEventTooLargeError + from splunk_logging.forwarders import BatchHecForwarder + + route = respx.post("http://localhost:8088/services/collector/event") + route.return_value = httpx.Response(200, json={"text": "Success", "code": 0}) + hec = BatchHecForwarder( + host="localhost", + port=8088, + token="", + use_ssl=False, + max_batch_bytes=100, + ) + + try: + with self.assertRaises(HecEventTooLargeError) as raised: + hec.forward_event({"message": "x" * 500}) + finally: + hec.close() + + self.assertGreater(raised.exception.event_size, raised.exception.max_batch_bytes) + self.assertFalse(route.called) + + @respx.mock + def test_batch_forwarder_applies_backpressure_to_in_flight_events(self): + from splunk_logging.exceptions import HecQueueFullError + from splunk_logging.forwarders import BatchHecForwarder + + request_started = threading.Event() + release_request = threading.Event() + route = respx.post("http://localhost:8088/services/collector/event") + + def block_request(_request): + request_started.set() + release_request.wait(1) + return httpx.Response(200, json={"text": "Success", "code": 0}) + + route.side_effect = block_request + hec = BatchHecForwarder( + host="localhost", + port=8088, + token="", + use_ssl=False, + batch_size=1, + max_queue_size=1, + max_queue_bytes=1_024, + enqueue_timeout=0, + ) + + try: + hec.forward_event({"message": "first"}) + self.assertTrue(request_started.wait(1)) + with self.assertRaises(HecQueueFullError) as raised: + hec.forward_event({"message": "second"}) + finally: + release_request.set() + hec.flush() + hec.close() + + self.assertEqual(raised.exception.enqueued_count, 0) + self.assertEqual(raised.exception.next_event_index, 0) + + @respx.mock + def test_batch_forwarder_applies_byte_backpressure_to_in_flight_events(self): + from splunk_logging.exceptions import HecQueueFullError + from splunk_logging.forwarders import BatchHecForwarder + + request_started = threading.Event() + release_request = threading.Event() + route = respx.post("http://localhost:8088/services/collector/event") + + def block_request(_request): + request_started.set() + release_request.wait(1) + return httpx.Response(200, json={"text": "Success", "code": 0}) + + route.side_effect = block_request + hec = BatchHecForwarder( + host="localhost", + port=8088, + token="", + use_ssl=False, + batch_size=1, + max_batch_bytes=600, + max_queue_size=10, + max_queue_bytes=600, + enqueue_timeout=0, + ) + + try: + hec.forward_event({"message": "x" * 300}) + self.assertTrue(request_started.wait(1)) + with self.assertRaises(HecQueueFullError): + hec.forward_event({"message": "y" * 300}) + finally: + release_request.set() + hec.flush() + hec.close() + + @respx.mock + def test_batch_forwarder_reports_partial_bulk_admission(self): + from splunk_logging.exceptions import HecQueueFullError + from splunk_logging.forwarders import BatchHecForwarder + + release_request = threading.Event() + route = respx.post("http://localhost:8088/services/collector/event") + + def block_request(_request): + release_request.wait(1) + return httpx.Response(200, json={"text": "Success", "code": 0}) + + route.side_effect = block_request + hec = BatchHecForwarder( + host="localhost", + port=8088, + token="", + use_ssl=False, + batch_size=1, + max_queue_size=1, + enqueue_timeout=0, + ) + + try: + with self.assertRaises(HecQueueFullError) as raised: + hec.forward_events([{"message": "first"}, {"message": "second"}]) + finally: + release_request.set() + hec.flush() + hec.close() + + self.assertEqual(raised.exception.enqueued_count, 1) + self.assertEqual(raised.exception.next_event_index, 1) + + @respx.mock + def test_batch_forwarder_prepares_all_events_before_admission(self): + from splunk_logging.forwarders import BatchHecForwarder + + route = respx.post("http://localhost:8088/services/collector/event") + route.return_value = httpx.Response(200, json={"text": "Success", "code": 0}) + hec = BatchHecForwarder( + host="localhost", + port=8088, + token="", + use_ssl=False, + batch_size=100, + flush_interval=60, + ) + + with self.assertRaises(TypeError): + hec.forward_events([{"message": "valid"}, {"message": object()}]) + hec.close() + + self.assertFalse(route.called) + + @respx.mock + def test_batch_forwarder_waits_for_ack_before_sending_next_batch(self): + from splunk_logging.forwarders import BatchHecForwarder + + timeline = [] + ack_attempts = {1: 0} + event_route = respx.post("http://localhost:8088/services/collector/event") + ack_route = respx.post("http://localhost:8088/services/collector/ack") + + def accept_event(_request): + ack_id = event_route.call_count + 1 + timeline.append(f"event-{ack_id}") + return httpx.Response(200, json={"text": "Success", "code": 0, "ackID": ack_id}) + + def acknowledge_event(request): + ack_id = json.loads(request.content)["acks"][0] + timeline.append(f"ack-{ack_id}") + if ack_id == 1: + ack_attempts[1] += 1 + acknowledged = ack_attempts[1] > 1 + else: + acknowledged = True + return httpx.Response(200, json={"acks": {str(ack_id): acknowledged}}) + + event_route.side_effect = accept_event + ack_route.side_effect = acknowledge_event + hec = BatchHecForwarder( + host="localhost", + port=8088, + token="", + use_ssl=False, + batch_size=1, + indexer_ack=True, + ack_poll_interval=0, + ) + + hec.forward_events([{"message": "first"}, {"message": "second"}]) + hec.flush() + hec.close() + + self.assertEqual(timeline, ["event-1", "ack-1", "ack-1", "event-2", "ack-2"]) + + @respx.mock + def test_batch_forwarder_reports_only_unaccepted_events_after_partial_failure(self): + from splunk_logging.exceptions import HecBatchError, HecWorkerError + from splunk_logging.forwarders import BatchHecForwarder + + route = respx.post("http://localhost:8088/services/collector/event") + route.return_value = httpx.Response( + 400, + json={"text": "Invalid data format", "code": 6, "invalid-event-number": 1}, + ) + hec = BatchHecForwarder( + host="localhost", + port=8088, + token="", + use_ssl=False, + batch_size=3, + flush_interval=60, + ) + events = [{"message": "accepted"}, {"message": "invalid"}, {"message": "not processed"}] + + hec.forward_events(events) + with self.assertRaises(HecWorkerError) as raised: + hec.flush() + with self.assertRaises(HecWorkerError): + hec.close() + + self.assertIsInstance(raised.exception.cause, HecBatchError) + self.assertEqual( + [envelope["event"] for envelope in raised.exception.failed_events], + events[1:], + ) + + @respx.mock + def test_batch_forwarder_failed_events_match_attempted_payload(self): + from splunk_logging.exceptions import HecWorkerError + from splunk_logging.forwarders import BatchHecForwarder + + request_started = threading.Event() + release_request = threading.Event() + route = respx.post("http://localhost:8088/services/collector/event") + + def reject_request(_request): + request_started.set() + release_request.wait(1) + return httpx.Response(400, json={"text": "Invalid data format", "code": 6}) + + route.side_effect = reject_request + hec = BatchHecForwarder( + host="localhost", + port=8088, + token="", + use_ssl=False, + batch_size=1, + ) + event = {"message": {"value": "attempted"}} + + hec.forward_event(event) + self.assertTrue(request_started.wait(1)) + event["message"]["value"] = "mutated" + release_request.set() + with self.assertRaises(HecWorkerError) as raised: + hec.flush() + with self.assertRaises(HecWorkerError): + hec.close() + + self.assertEqual(raised.exception.failed_events[0]["event"]["message"]["value"], "attempted") + + @respx.mock + def test_batch_forwarder_surfaces_background_delivery_failure(self): + from splunk_logging.exceptions import HecWorkerError + from splunk_logging.forwarders import BatchHecForwarder + + request_finished = threading.Event() + route = respx.post("http://localhost:8088/services/collector/event") + + def reject_request(_request): + request_finished.set() + return httpx.Response(400, json={"text": "Incorrect data format", "code": 5}) + + route.side_effect = reject_request + hec = BatchHecForwarder( + host="localhost", + port=8088, + token="", + use_ssl=False, + batch_size=1, + ) + hec.forward_event({"message": "first"}) + self.assertTrue(request_finished.wait(1)) + + with self.assertRaises(HecWorkerError): + hec.flush() + with self.assertRaises(HecWorkerError): + hec.forward_event({"message": "second"}) + with self.assertRaises(HecWorkerError): + hec.close() + + @respx.mock + @patch("splunk_logging.forwarders.time.sleep") + def test_retry(self, sleep): hec = self.create_forwarder() route = respx.post("http://localhost:8088/services/collector/event") event = {"message": "test"} # test successful forward - route.return_value = httpx.Response(httpx.codes.SERVICE_UNAVAILABLE) + route.return_value = httpx.Response( + httpx.codes.SERVICE_UNAVAILABLE, + json={"text": "Server is busy", "code": 9}, + ) with self.assertRaises(httpx.HTTPStatusError): hec.forward_event(event) self.assertTrue(route.called) - self.assertGreaterEqual(route.call_count, 3) + self.assertEqual(route.call_count, 4) + self.assertEqual(sleep.call_count, 3) return + @respx.mock + @patch("splunk_logging.forwarders.time.sleep") + def test_retry_returns_successful_response(self, _sleep): + hec = self.create_forwarder() + route = respx.post("http://localhost:8088/services/collector/event") + route.side_effect = [ + httpx.Response( + httpx.codes.SERVICE_UNAVAILABLE, + json={"text": "Server is busy", "code": 9}, + ), + httpx.Response(httpx.codes.OK, json={"text": "Success", "code": 0}), + ] + + hec.forward_event({"message": "test"}) + + self.assertEqual(route.call_count, 2) + + @respx.mock + @patch("splunk_logging.forwarders.time.sleep") + def test_retry_honors_retry_after(self, sleep): + hec = self.create_forwarder() + route = respx.post("http://localhost:8088/services/collector/event") + route.side_effect = [ + httpx.Response( + httpx.codes.SERVICE_UNAVAILABLE, + headers={"Retry-After": "7"}, + json={"text": "Server is busy", "code": 9}, + ), + httpx.Response(httpx.codes.OK, json={"text": "Success", "code": 0}), + ] + + hec.forward_event({"message": "test"}) + + sleep.assert_called_once_with(7.0) + + @respx.mock + @patch("splunk_logging.forwarders.time.sleep") + def test_retry_retries_connection_establishment_failures(self, sleep): + route = respx.post("http://localhost:8088/services/collector/event") + + for error_type in (httpx.ConnectError, httpx.ConnectTimeout): + with self.subTest(error_type=error_type): + hec = self.create_forwarder() + route.side_effect = [ + error_type("connection failed"), + httpx.Response(httpx.codes.OK, json={"text": "Success", "code": 0}), + ] + previous_call_count = route.call_count + + hec.forward_event({"message": "test"}) + + self.assertEqual(route.call_count, previous_call_count + 2) + hec.close() + + self.assertEqual(sleep.call_count, 2) + + @respx.mock + @patch("splunk_logging.forwarders.time.sleep") + def test_retry_retries_request_timeout(self, sleep): + hec = self.create_forwarder() + route = respx.post("http://localhost:8088/services/collector/event") + route.side_effect = [ + httpx.Response(httpx.codes.REQUEST_TIMEOUT), + httpx.Response(httpx.codes.OK, json={"text": "Success", "code": 0}), + ] + + hec.forward_event({"message": "test"}) + + self.assertEqual(route.call_count, 2) + sleep.assert_called_once() + + @respx.mock + @patch("splunk_logging.forwarders.time.sleep") + def test_retry_does_not_retry_ambiguous_http_failures(self, sleep): + failures = [ + httpx.Response(httpx.codes.TOO_MANY_REQUESTS), + httpx.Response(httpx.codes.INTERNAL_SERVER_ERROR), + httpx.Response(httpx.codes.BAD_GATEWAY), + httpx.Response(httpx.codes.SERVICE_UNAVAILABLE, text="load balancer unavailable"), + httpx.Response(httpx.codes.GATEWAY_TIMEOUT), + ] + + for response in failures: + with self.subTest(status_code=response.status_code): + hec = self.create_forwarder() + route = respx.post("http://localhost:8088/services/collector/event") + route.side_effect = None + route.return_value = response + previous_call_count = route.call_count + + with self.assertRaises(httpx.HTTPStatusError): + hec.forward_event({"message": "test"}) + + self.assertEqual(route.call_count, previous_call_count + 1) + hec.close() + + sleep.assert_not_called() + + @respx.mock + @patch("splunk_logging.forwarders.time.sleep") + def test_retry_does_not_retry_ambiguous_transport_failures(self, sleep): + for error_type in (httpx.ReadError, httpx.ReadTimeout, httpx.WriteError, httpx.WriteTimeout): + with self.subTest(error_type=error_type): + hec = self.create_forwarder() + route = respx.post("http://localhost:8088/services/collector/event") + route.side_effect = error_type("ambiguous delivery failure") + previous_call_count = route.call_count + + with self.assertRaises(error_type): + hec.forward_event({"message": "test"}) + + self.assertEqual(route.call_count, previous_call_count + 1) + hec.close() + + sleep.assert_not_called() + def test_parse_timestamp(self): hec = self.create_forwarder() dt = datetime( diff --git a/test/test_handler.py b/test/test_handler.py index e69533d..6dbd98c 100644 --- a/test/test_handler.py +++ b/test/test_handler.py @@ -22,6 +22,8 @@ from splunk_logging.formatters import JsonFormatter from splunk_logging.handlers import HecHandler +TEST_CHANNEL_ID = "00000000-0000-4000-8000-000000000001" + class TestHecHandler(unittest.TestCase): def create_logger(self): @@ -48,3 +50,59 @@ def test_hec_logger(self): self.assertIn("time", sent_event) self.assertEqual(json.dumps(sent_event["event"]), json.dumps(event)) return + + @respx.mock + def test_hec_logger_supports_indexer_acknowledgment(self): + log = logging.Logger("ack") + handler = HecHandler( + host="localhost", + port=8088, + token="", + use_ssl=False, + indexer_ack=True, + channel_id=TEST_CHANNEL_ID, + ack_poll_interval=0, + ) + log.addHandler(handler) + event_route = respx.post("http://localhost:8088/services/collector/event") + event_route.return_value = httpx.Response(200, json={"text": "Success", "code": 0, "ackId": 7}) + ack_route = respx.post("http://localhost:8088/services/collector/ack") + ack_route.return_value = httpx.Response(200, json={"acks": {"7": True}}) + + log.warning({"message": "test"}) + + self.assertEqual(event_route.calls.last.request.headers["X-Splunk-Request-Channel"], TEST_CHANNEL_ID) + self.assertEqual(ack_route.call_count, 1) + + @respx.mock + def test_hec_logger_supports_opt_in_batching(self): + log = logging.Logger("batch") + handler = HecHandler( + host="localhost", + port=8088, + token="", + use_ssl=False, + batch_enabled=True, + batch_size=2, + flush_interval=60, + ignore_exceptions=False, + ) + log.addHandler(handler) + route = respx.post("http://localhost:8088/services/collector/event") + route.return_value = httpx.Response(200, json={"text": "Success", "code": 0}) + + try: + log.info({"message": "first"}) + log.info({"message": "second"}) + handler.flush() + finally: + handler.close() + + self.assertEqual(route.call_count, 1) + payload = route.calls.last.request.content.decode() + decoder = json.JSONDecoder() + first, offset = decoder.raw_decode(payload) + second, offset = decoder.raw_decode(payload, offset) + self.assertEqual(offset, len(payload)) + self.assertEqual(first["event"]["message"], "first") + self.assertEqual(second["event"]["message"], "second")