feat: add reliable batched HEC delivery

.gitignore

@@ -45,6 +45,7 @@ htmlcov/
 .coverage
 .coverage.*
 .cache
+.local/
 nosetests.xml
 coverage.xml
 *.cover

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

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)