fix(sdk): reuse a stream or project created concurrently (SAO-18644)

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

src/splunk_ao/logger/logger.py

@@ -90,6 +90,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
@@ -809,15 +834,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:
@@ -826,8 +859,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
 

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