RSPEED-2326: feat(observability): add async Splunk HEC client

Implement send_splunk_event() async function for sending events to Splunk:
- Uses aiohttp for async HTTP requests
- Reads HEC token from file path
- Graceful degradation when config missing/disabled
- Logs warnings on send failures (no exceptions raised)

Add splunk and deployment_environment properties to AppConfig.

Signed-off-by: Major Hayden <major@redhat.com>

Author: major

src/configuration.py

@@ -23,6 +23,7 @@
     DatabaseConfiguration,
     ConversationHistoryConfiguration,
     QuotaHandlersConfiguration,
+    SplunkConfiguration,
 )
 
 from cache.cache import Cache
@@ -39,7 +40,7 @@ class LogicError(Exception):
     """Error in application logic."""
 
 
-class AppConfig:
+class AppConfig:  # pylint: disable=too-many-public-methods
     """Singleton class to load and store the configuration."""
 
     _instance = None
@@ -348,5 +349,19 @@ def azure_entra_id(self) -> Optional[AzureEntraIdConfiguration]:
             raise LogicError("logic error: configuration is not loaded")
         return self._configuration.azure_entra_id
 
+    @property
+    def splunk(self) -> Optional[SplunkConfiguration]:
+        """Return Splunk configuration, or None if not provided."""
+        if self._configuration is None:
+            raise LogicError("logic error: configuration is not loaded")
+        return self._configuration.splunk
+
+    @property
+    def deployment_environment(self) -> str:
+        """Return deployment environment name."""
+        if self._configuration is None:
+            raise LogicError("logic error: configuration is not loaded")
+        return self._configuration.deployment_environment
+
 
 configuration: AppConfig = AppConfig()

src/observability/splunk.py

@@ -0,0 +1,90 @@
+"""Async Splunk HEC client for sending telemetry events."""
+
+import logging
+import platform
+import time
+from typing import Any
+
+import aiohttp
+
+from configuration import configuration
+from version import __version__
+
+logger = logging.getLogger(__name__)
+
+
+def _get_hostname() -> str:
+    """Get the hostname for Splunk event metadata."""
+    return platform.node() or "unknown"
+
+
+def _read_token_from_file(token_path: str) -> str | None:
+    """Read HEC token from file path."""
+    try:
+        with open(token_path, encoding="utf-8") as f:
+            return f.read().strip()
+    except OSError as e:
+        logger.warning("Failed to read Splunk HEC token from %s: %s", token_path, e)
+        return None
+
+
+async def send_splunk_event(event: dict[str, Any], sourcetype: str) -> None:
+    """Send an event to Splunk HEC.
+
+    This function sends events asynchronously and handles failures gracefully
+    by logging warnings instead of raising exceptions. This ensures that
+    Splunk connectivity issues don't affect the main application flow.
+
+    Args:
+        event: The event payload to send.
+        sourcetype: The Splunk sourcetype (e.g., "infer_with_llm", "infer_error").
+    """
+    splunk_config = configuration.splunk
+    if splunk_config is None or not splunk_config.enabled:
+        logger.debug("Splunk integration disabled, skipping event")
+        return
+
+    if not splunk_config.url or not splunk_config.token_path or not splunk_config.index:
+        logger.warning("Splunk configuration incomplete, skipping event")
+        return
+
+    # Read token on each request to support rotation without restart
+    token = _read_token_from_file(str(splunk_config.token_path))
+    if not token:
+        return
+
+    payload = {
+        "time": int(time.time()),
+        "host": _get_hostname(),
+        "source": f"{splunk_config.source} (v{__version__})",
+        "sourcetype": sourcetype,
+        "index": splunk_config.index,
+        "event": event,
+    }
+
+    headers = {
+        "Authorization": f"Splunk {token}",
+        "Content-Type": "application/json",
+    }
+
+    timeout = aiohttp.ClientTimeout(total=splunk_config.timeout)
+    connector = aiohttp.TCPConnector(ssl=splunk_config.verify_ssl)
+
+    try:
+        async with aiohttp.ClientSession(
+            timeout=timeout, connector=connector
+        ) as session:
+            async with session.post(
+                splunk_config.url, json=payload, headers=headers
+            ) as response:
+                if response.status >= 400:
+                    body = await response.text()
+                    logger.warning(
+                        "Splunk HEC request failed with status %d: %s",
+                        response.status,
+                        body[:200],
+                    )
+    except aiohttp.ClientError as e:
+        logger.warning("Splunk HEC request failed: %s", e)
+    except TimeoutError:
+        logger.warning("Splunk HEC request timed out after %ds", splunk_config.timeout)

tests/unit/observability/__init__.py

@@ -0,0 +1 @@
+"""Unit tests for observability module."""

tests/unit/observability/test_splunk.py

@@ -0,0 +1,155 @@
+"""Unit tests for Splunk HEC client."""
+
+from pathlib import Path
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import aiohttp
+import pytest
+
+from observability.splunk import send_splunk_event, _read_token_from_file
+
+
+@pytest.fixture(name="mock_splunk_config")
+def mock_splunk_config_fixture(tmp_path: Path) -> MagicMock:
+    """Create a mock SplunkConfiguration."""
+    token_file = tmp_path / "token"
+    token_file.write_text("test-hec-token")
+
+    config = MagicMock()
+    config.enabled = True
+    config.url = "https://splunk.example.com:8088/services/collector"
+    config.token_path = token_file
+    config.index = "test_index"
+    config.source = "test-source"
+    config.timeout = 5
+    config.verify_ssl = True
+    return config
+
+
+@pytest.fixture(name="mock_session")
+def mock_session_fixture() -> AsyncMock:
+    """Create a mock aiohttp session with successful response."""
+    mock_response = AsyncMock()
+    mock_response.status = 200
+    session = AsyncMock(spec=aiohttp.ClientSession)
+    session.post.return_value.__aenter__.return_value = mock_response
+    return session
+
+
+@pytest.mark.parametrize(
+    ("token_content", "expected"),
+    [
+        ("  my-secret-token  \n", "my-secret-token"),
+        ("token-no-whitespace", "token-no-whitespace"),
+    ],
+    ids=["strips_whitespace", "no_whitespace"],
+)
+def test_read_token_from_file(
+    tmp_path: Path, token_content: str, expected: str
+) -> None:
+    """Test reading and stripping token from file."""
+    token_file = tmp_path / "token"
+    token_file.write_text(token_content)
+    assert _read_token_from_file(str(token_file)) == expected
+
+
+def test_read_token_returns_none_for_missing_file(tmp_path: Path) -> None:
+    """Test returns None when file doesn't exist."""
+    assert _read_token_from_file(str(tmp_path / "nonexistent")) is None
+
+
+def _make_config(
+    enabled: bool = True,
+    url: str | None = "https://splunk:8088",
+    token_path: Path | None = None,
+    index: str | None = "idx",
+) -> MagicMock:
+    """Helper to create mock config with specific fields."""
+    config = MagicMock()
+    config.enabled = enabled
+    config.url = url
+    config.token_path = token_path
+    config.index = index
+    return config
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("splunk_config",),
+    [
+        (None,),
+        (_make_config(enabled=False),),
+        (_make_config(url=None, index=None),),
+    ],
+    ids=["config_none", "disabled", "incomplete"],
+)
+async def test_skips_event_when_not_configured(splunk_config: Any) -> None:
+    """Test event is skipped when Splunk is not properly configured."""
+    with patch("observability.splunk.configuration") as mock_config:
+        mock_config.splunk = splunk_config
+        # Should not raise, just skip silently
+        await send_splunk_event({"test": "event"}, "test_sourcetype")
+
+
+@pytest.mark.asyncio
+async def test_sends_event_successfully(
+    mock_splunk_config: MagicMock, mock_session: AsyncMock
+) -> None:
+    """Test event is sent successfully to Splunk HEC."""
+    with (
+        patch("observability.splunk.configuration") as mock_config,
+        patch("observability.splunk.aiohttp.ClientSession") as mock_client,
+    ):
+        mock_config.splunk = mock_splunk_config
+        mock_client.return_value.__aenter__.return_value = mock_session
+
+        await send_splunk_event({"question": "test"}, "infer_with_llm")
+
+        mock_session.post.assert_called_once()
+        call_args = mock_session.post.call_args
+        assert call_args[0][0] == mock_splunk_config.url
+        assert "Authorization" in call_args[1]["headers"]
+        assert call_args[1]["json"]["sourcetype"] == "infer_with_llm"
+        assert call_args[1]["json"]["event"] == {"question": "test"}
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("error_setup",),
+    [
+        (
+            lambda s: setattr(
+                s.post.return_value.__aenter__.return_value, "status", 503
+            ),
+        ),
+        (
+            lambda s: setattr(
+                s.return_value.__aenter__, "side_effect", aiohttp.ClientError()
+            ),
+        ),
+    ],
+    ids=["http_error", "client_error"],
+)
+async def test_logs_warning_on_error(
+    mock_splunk_config: MagicMock, error_setup: Any
+) -> None:
+    """Test warning is logged on HTTP or client errors."""
+    mock_session = AsyncMock(spec=aiohttp.ClientSession)
+    mock_response = AsyncMock()
+    mock_response.status = 503
+    mock_response.text.return_value = "error"
+    mock_session.post.return_value.__aenter__.return_value = mock_response
+
+    with (
+        patch("observability.splunk.configuration") as mock_config,
+        patch("observability.splunk.aiohttp.ClientSession") as mock_client,
+        patch("observability.splunk.logger") as mock_logger,
+    ):
+        mock_config.splunk = mock_splunk_config
+        error_setup(mock_client)
+        mock_client.return_value.__aenter__.return_value = mock_session
+
+        await send_splunk_event({"test": "event"}, "test_sourcetype")
+
+        mock_logger.warning.assert_called()