From aa72eecad1a737a51fda850fba3fa7f64ac1463c Mon Sep 17 00:00:00 2001 From: etserend Date: Fri, 23 Oct 2026 11:32:42 -0500 Subject: [PATCH 1/2] fix(sdk): reuse the agent stream or project another process created (SAO-18644) When several processes start together, each can find no agent stream or project and try to create it. The server keeps one and rejects the rest (HTTP 400 for a stream, HTTP 422 for a project, HTTP 500 on older servers). The rejected processes raised from SplunkAOLogger and failed at startup. After a failed create, look the stream or project up again and use what the winner made. For a project, wait briefly (5 look-ups, 0.2 s apart) because the winner may not have granted access yet. If it never appears, re-raise the original create error, so other failures still show. Co-Authored-By: Claude Code --- src/splunk_ao/logger/logger.py | 66 ++++++++-- tests/test_logger_lost_create_race.py | 177 ++++++++++++++++++++++++++ 2 files changed, 232 insertions(+), 11 deletions(-) create mode 100644 tests/test_logger_lost_create_race.py diff --git a/src/splunk_ao/logger/logger.py b/src/splunk_ao/logger/logger.py index 1c4241eb..72b0b17f 100644 --- a/src/splunk_ao/logger/logger.py +++ b/src/splunk_ao/logger/logger.py @@ -89,6 +89,31 @@ # Type alias for metadata values that can be auto-converted to strings MetadataValue = str | bool | int | float | None StepT = TypeVar("StepT", bound=BaseStep) +LookupT = TypeVar("LookupT") + +# Look-ups after a lost create race. The process that won may still be granting access to a new project. +LOST_RACE_LOOKUP_ATTEMPTS = 5 +LOST_RACE_LOOKUP_DELAY_SECONDS = 0.2 + + +def _find_after_lost_create_race(lookup: Callable[[], LookupT | None], create_error: Exception) -> LookupT: + """Return what another process just created, or re-raise the error from our own create. + + Processes that start together can all find nothing and all try to create. The server keeps one and + rejects the rest, for example with HTTP 400 for an agent stream and HTTP 422 for a project. Look again, + and use what the winner made. If it is still not there, the create failed for another reason. + """ + for attempt in range(LOST_RACE_LOOKUP_ATTEMPTS): + try: + found = lookup() + except Exception as lookup_error: + raise create_error from lookup_error + if found is not None: + return found + if attempt < LOST_RACE_LOOKUP_ATTEMPTS - 1: + time.sleep(LOST_RACE_LOOKUP_DELAY_SECONDS) + raise create_error + DEFAULT_TERMINATE_TIMEOUT_SECONDS = 90 # Absolute threshold above which a slow SplunkAOLogger.terminate() shutdown is @@ -722,15 +747,23 @@ def _init_project(self) -> None: project_obj = projects_client.get(name=self.project_name) if project_obj is None: # Create project if it doesn't exist - project = projects_client.create(name=self.project_name) - if project is None: - raise SplunkAOLoggerException(f"Failed to create project {self.project_name}.") - self.project_id = project.id - self._logger.info(f"🚀 Creating new project... project {self.project_name} created!") - else: - if project_obj.type != "gen_ai": - raise Exception(f"Project {self.project_name} is not a Splunk AO project") - self.project_id = project_obj.id + try: + project = projects_client.create(name=self.project_name) + except Exception as create_error: + # Another process created it first. Use that one. + project_obj = _find_after_lost_create_race( + lambda: projects_client.get(name=self.project_name), create_error + ) + self._logger.info(f"Project {self.project_name} was created by another process; using it.") + else: + if project is None: + raise SplunkAOLoggerException(f"Failed to create project {self.project_name}.") + self.project_id = project.id + self._logger.info(f"🚀 Creating new project... project {self.project_name} created!") + return + if project_obj.type != "gen_ai": + raise Exception(f"Project {self.project_name} is not a Splunk AO project") + self.project_id = project_obj.id @nop_sync def _init_agent_stream(self) -> None: @@ -739,8 +772,19 @@ def _init_agent_stream(self) -> None: log_stream_obj = log_streams_client.get(name=self.agent_stream_name, project_id=self.project_id) if log_stream_obj is None: # Create log stream if it doesn't exist - self.agent_stream_id = log_streams_client.create(name=self.agent_stream_name, project_id=self.project_id).id - self._logger.info(f"🚀 Creating new agent stream... agent stream {self.agent_stream_name} created!") + try: + created = log_streams_client.create(name=self.agent_stream_name, project_id=self.project_id) + except Exception as create_error: + # Another process created it first. Use that one. + log_stream_obj = _find_after_lost_create_race( + lambda: log_streams_client.get(name=self.agent_stream_name, project_id=self.project_id), + create_error, + ) + self.agent_stream_id = log_stream_obj.id + self._logger.info(f"Agent stream {self.agent_stream_name} was created by another process; using it.") + else: + self.agent_stream_id = created.id + self._logger.info(f"🚀 Creating new agent stream... agent stream {self.agent_stream_name} created!") else: self.agent_stream_id = log_stream_obj.id diff --git a/tests/test_logger_lost_create_race.py b/tests/test_logger_lost_create_race.py new file mode 100644 index 00000000..a52e841a --- /dev/null +++ b/tests/test_logger_lost_create_race.py @@ -0,0 +1,177 @@ +"""SplunkAOLogger must use what another process created when it loses a create race (SAO-18644).""" + +import atexit +from collections.abc import Generator +from unittest.mock import MagicMock, patch + +import pytest + +from splunk_ao.exceptions import BadRequestError +from splunk_ao.logger import SplunkAOLogger +from splunk_ao.logger.logger import LOST_RACE_LOOKUP_ATTEMPTS +from splunk_ao.projects import ProjectsAPIException + +STREAM_EXISTS = b'{"detail": "Log stream demo-stream already exists for project demo-project."}' +PROJECT_EXISTS = '{"detail": "A project with this (name, type) already exists."}' + + +def make_logger() -> SplunkAOLogger: + """A logger with no backend set-up. An ingestion hook skips project and stream look-ups in the constructor.""" + logger = SplunkAOLogger(project="demo-project", agent_stream="demo-stream", ingestion_hook=lambda request: None) + atexit.unregister(logger.terminate) + logger.project_id = "project-1" + return logger + + +@pytest.fixture +def sleep() -> Generator[MagicMock, None, None]: + with patch("splunk_ao.logger.logger.time.sleep") as mock_sleep: + yield mock_sleep + + +@pytest.fixture +def streams() -> Generator[MagicMock, None, None]: + with patch("splunk_ao.logger.logger.AgentStreams") as client_class: + yield client_class.return_value + + +@pytest.fixture +def projects() -> Generator[MagicMock, None, None]: + with patch("splunk_ao.logger.logger.Projects") as client_class: + yield client_class.return_value + + +class TestInitAgentStream: + def test_existing_stream_is_used_without_creating(self, streams: MagicMock, sleep: MagicMock) -> None: + streams.get.return_value = MagicMock(id="stream-1") + logger = make_logger() + + logger._init_agent_stream() + + assert logger.agent_stream_id == "stream-1" + streams.create.assert_not_called() + + def test_new_stream_is_created(self, streams: MagicMock, sleep: MagicMock) -> None: + streams.get.return_value = None + streams.create.return_value = MagicMock(id="stream-2") + logger = make_logger() + + logger._init_agent_stream() + + assert logger.agent_stream_id == "stream-2" + streams.get.assert_called_once() + sleep.assert_not_called() + + def test_lost_race_uses_the_winners_stream(self, streams: MagicMock, sleep: MagicMock) -> None: + # Given: the first look-up finds nothing, our create is rejected, and the winner's stream is there next + streams.get.side_effect = [None, MagicMock(id="winner-stream")] + streams.create.side_effect = BadRequestError(400, STREAM_EXISTS) + logger = make_logger() + + # When: the logger initializes + logger._init_agent_stream() + + # Then: it uses the winner's stream and does not raise + assert logger.agent_stream_id == "winner-stream" + assert streams.get.call_count == 2 + streams.create.assert_called_once() + sleep.assert_not_called() + + def test_create_error_is_raised_when_the_stream_never_appears(self, streams: MagicMock, sleep: MagicMock) -> None: + # Given: a create that fails for a reason other than a lost race, so no stream ever shows up + error = BadRequestError(400, b'{"detail": "Log stream is not allowed in this project type."}') + streams.get.return_value = None + streams.create.side_effect = error + logger = make_logger() + + # When / Then: the original error comes back, after a bounded number of look-ups + with pytest.raises(BadRequestError) as raised: + logger._init_agent_stream() + assert raised.value is error + assert streams.get.call_count == 1 + LOST_RACE_LOOKUP_ATTEMPTS + assert sleep.call_count == LOST_RACE_LOOKUP_ATTEMPTS - 1 + assert logger.agent_stream_id is None + + def test_create_error_is_raised_when_the_second_look_up_fails(self, streams: MagicMock, sleep: MagicMock) -> None: + # Given: a rejected create and a look-up that then fails + error = BadRequestError(400, STREAM_EXISTS) + lookup_failure = RuntimeError("connection reset") + streams.get.side_effect = [None, lookup_failure] + streams.create.side_effect = error + logger = make_logger() + + # When / Then: the create error is raised, with the look-up failure as its cause + with pytest.raises(BadRequestError) as raised: + logger._init_agent_stream() + assert raised.value is error + assert raised.value.__cause__ is lookup_failure + + +class TestInitProject: + def test_existing_project_is_used_without_creating(self, projects: MagicMock, sleep: MagicMock) -> None: + projects.get.return_value = MagicMock(id="project-1", type="gen_ai") + logger = make_logger() + + logger._init_project() + + assert logger.project_id == "project-1" + projects.create.assert_not_called() + + def test_new_project_is_created(self, projects: MagicMock, sleep: MagicMock) -> None: + projects.get.return_value = None + projects.create.return_value = MagicMock(id="project-2") + logger = make_logger() + + logger._init_project() + + assert logger.project_id == "project-2" + projects.get.assert_called_once() + sleep.assert_not_called() + + def test_lost_race_uses_the_winners_project(self, projects: MagicMock, sleep: MagicMock) -> None: + # Given: the first look-up finds nothing, our create is rejected, and the winner's project is there next + projects.get.side_effect = [None, MagicMock(id="winner-project", type="gen_ai")] + projects.create.side_effect = ProjectsAPIException(PROJECT_EXISTS) + logger = make_logger() + + # When: the logger initializes + logger._init_project() + + # Then: it uses the winner's project and does not raise + assert logger.project_id == "winner-project" + projects.create.assert_called_once() + + def test_lost_race_waits_until_the_winner_has_granted_access(self, projects: MagicMock, sleep: MagicMock) -> None: + # Given: the winner's project is not visible to us for the first two look-ups + projects.get.side_effect = [None, None, None, MagicMock(id="winner-project", type="gen_ai")] + projects.create.side_effect = ProjectsAPIException(PROJECT_EXISTS) + logger = make_logger() + + # When: the logger initializes + logger._init_project() + + # Then: it waited between look-ups and then used the project + assert logger.project_id == "winner-project" + assert sleep.call_count == 2 + + def test_winner_project_of_another_type_is_rejected(self, projects: MagicMock, sleep: MagicMock) -> None: + projects.get.side_effect = [None, MagicMock(id="other", type="protect")] + projects.create.side_effect = ProjectsAPIException(PROJECT_EXISTS) + logger = make_logger() + + with pytest.raises(Exception, match="is not a Splunk AO project"): + logger._init_project() + + def test_create_error_is_raised_when_the_project_never_appears(self, projects: MagicMock, sleep: MagicMock) -> None: + # Given: a create that fails and a project that never shows up + error = ProjectsAPIException('{"detail": "Not allowed."}') + projects.get.return_value = None + projects.create.side_effect = error + logger = make_logger() + + # When / Then: the original error comes back, after a bounded number of look-ups + with pytest.raises(ProjectsAPIException) as raised: + logger._init_project() + assert raised.value is error + assert projects.get.call_count == 1 + LOST_RACE_LOOKUP_ATTEMPTS + assert sleep.call_count == LOST_RACE_LOOKUP_ATTEMPTS - 1 From ef9479030801605ff111df856315fc1c3f969ba2 Mon Sep 17 00:00:00 2001 From: etserend Date: Tue, 27 Oct 2026 10:16:42 -0500 Subject: [PATCH 2/2] docs(changelog): note reuse of an agent stream or project created concurrently (SAO-18644) Co-Authored-By: Claude Code --- CHANGELOG.md | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index d4e2a41e..1ecebdde 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed + +- `SplunkAOLogger` no longer fails at startup when several processes start together and race to create the same + agent stream or project. After a failed create, the SDK looks it up again and uses the one the winner made. + A project is retried up to 5 times, 0.2 s apart, while access is granted. If it never appears, the original + create error is raised. + ## [0.6.1] ### Fixed