diff --git a/examples/chatbot/sample-project-chatbot/anthropic/app.py b/examples/chatbot/sample-project-chatbot/anthropic/app.py index 0b5b7128..5488e595 100644 --- a/examples/chatbot/sample-project-chatbot/anthropic/app.py +++ b/examples/chatbot/sample-project-chatbot/anthropic/app.py @@ -11,7 +11,7 @@ as a workflow span - The call to the LLM is logged manually as an LLM span. - After the response is received, the trace is concluded with the response - and flushed to ensure it is sent to Splunk AO. + and flushed so it is exported immediately. To run this, you will need to have the following environment variables set: - `SPLUNK_AO_API_KEY`: Your Splunk AO API key. @@ -25,14 +25,13 @@ """ -from datetime import datetime import os +from datetime import datetime from anthropic import Anthropic - from dotenv import load_dotenv -from splunk_ao import splunk_ao_context, log +from splunk_ao import log, splunk_ao_context # Load the environment variables from the .env file # This will override any existing environment variables with the same name @@ -102,10 +101,7 @@ def send_chat_to_anthropic() -> str: # Send the chat history to the Anthropic API and get the response response = client.messages.create( - max_tokens=1024, - messages=chat_history_anthropic, - system=system_prompt, - model=MODEL_NAME, + max_tokens=1024, messages=chat_history_anthropic, system=system_prompt, model=MODEL_NAME ) # Print the response to the console @@ -182,8 +178,8 @@ def main() -> None: # Call the chat_with_llm function to get a response from the LLM response = chat_with_llm(user_input) - # Conclude and flush the logger after each interaction - # so that a new trace is started each time + # conclude() ends the trace so the next interaction starts a new one; + # flush() exports it immediately instead of waiting for the batch timer logger.conclude(output=response) logger.flush() diff --git a/examples/chatbot/sample-project-chatbot/azure-inference/app.py b/examples/chatbot/sample-project-chatbot/azure-inference/app.py index dcf9f903..e33ad8ef 100644 --- a/examples/chatbot/sample-project-chatbot/azure-inference/app.py +++ b/examples/chatbot/sample-project-chatbot/azure-inference/app.py @@ -11,7 +11,7 @@ as a workflow span - The call to the LLM is logged manually as an LLM span. - After the response is received, the trace is concluded with the response - and flushed to ensure it is sent to Splunk AO. + and flushed so it is exported immediately. To run this, you will need to have the following environment variables set: - `SPLUNK_AO_API_KEY`: Your Splunk AO API key. @@ -26,16 +26,15 @@ """ -from datetime import datetime import os +from datetime import datetime from azure.ai.inference import ChatCompletionsClient -from azure.ai.inference.models import SystemMessage, UserMessage, AssistantMessage +from azure.ai.inference.models import AssistantMessage, SystemMessage, UserMessage from azure.core.credentials import AzureKeyCredential - from dotenv import load_dotenv -from splunk_ao import splunk_ao_context, log +from splunk_ao import log, splunk_ao_context # Load the environment variables from the .env file # This will override any existing environment variables with the same name @@ -183,8 +182,8 @@ def main() -> None: # Call the chat_with_llm function to get a response from the LLM response = chat_with_llm(user_input) - # Conclude and flush the logger after each interaction - # so that a new trace is started each time + # conclude() ends the trace so the next interaction starts a new one; + # flush() exports it immediately instead of waiting for the batch timer logger.conclude(output=response) logger.flush() diff --git a/examples/chatbot/sample-project-chatbot/openai-ollama/app.py b/examples/chatbot/sample-project-chatbot/openai-ollama/app.py index 9b6fa658..9429abc8 100644 --- a/examples/chatbot/sample-project-chatbot/openai-ollama/app.py +++ b/examples/chatbot/sample-project-chatbot/openai-ollama/app.py @@ -12,7 +12,7 @@ - The call to the LLM is logged as an LLM span using the Splunk AO OpenAI integration which logs the span automatically. - After the response is received, the trace is concluded with the response - and flushed to ensure it is sent to Splunk AO. + and flushed so it is exported immediately. To run this, you will need to have the following environment variables set: - `SPLUNK_AO_API_KEY`: Your Splunk AO API key. @@ -28,12 +28,12 @@ """ -from datetime import datetime import os +from datetime import datetime from dotenv import load_dotenv -from splunk_ao import splunk_ao_context, log +from splunk_ao import log, splunk_ao_context from splunk_ao.openai import OpenAI # Load the environment variables from the .env file @@ -165,8 +165,8 @@ def main() -> None: # Call the chat_with_llm function to get a response from the LLM response = chat_with_llm(user_input) - # Conclude and flush the logger after each interaction - # so that a new trace is started each time + # conclude() ends the trace so the next interaction starts a new one; + # flush() exports it immediately instead of waiting for the batch timer logger.conclude(output=response) logger.flush() diff --git a/examples/chatbot/sample-project-chatbot/openai-ollama/create_sample_logs.py b/examples/chatbot/sample-project-chatbot/openai-ollama/create_sample_logs.py index 384afe8c..29ad774c 100644 --- a/examples/chatbot/sample-project-chatbot/openai-ollama/create_sample_logs.py +++ b/examples/chatbot/sample-project-chatbot/openai-ollama/create_sample_logs.py @@ -1,20 +1,20 @@ # A script to generate log streams # Load the dataset.json -from datetime import datetime import json import uuid +from datetime import datetime -from splunk_ao import splunk_ao_context - -from app import chat_with_llm, chat_history +from app import chat_history, chat_with_llm # Load environment variables from .env file from dotenv import load_dotenv +from splunk_ao import splunk_ao_context + load_dotenv(override=True) -with open("../dataset.json", "r", encoding="utf-8") as f: +with open("../dataset.json", encoding="utf-8") as f: dataset_content = json.load(f) print(f"Starting to log {len(dataset_content)} interactions...") @@ -44,8 +44,8 @@ # Print the response from the LLM print(f"LLM Response: {response}") - # Conclude and flush the logger after each interaction - # so that a new trace is started each time + # conclude() ends the trace so the next interaction starts a new one; + # flush() exports it immediately instead of waiting for the batch timer logger.conclude(output=response) logger.flush() diff --git a/splunk-ao-migration-tool/README.md b/splunk-ao-migration-tool/README.md index c2397fd5..e8739b58 100644 --- a/splunk-ao-migration-tool/README.md +++ b/splunk-ao-migration-tool/README.md @@ -25,7 +25,7 @@ Additionally there are a handful of **removed features** (Protect, `GalileoScore ### 1.1 Package Availability -> **`splunk-ao` is not yet published to PyPI.** +> **`splunk-ao` is not yet published to PyPI.** > Use one of the two installation methods below until a public release is available. **Option A — Install directly from GitHub (recommended for most users)** @@ -81,7 +81,7 @@ splunk-ao = { path = "../splunk-ao-python", develop = true } ### 1.2 Optional Extra Groups -The extras keys are unchanged (`langchain`, `openai`, `crewai`, `middleware`, `otel`, `all`). +The extras keys are unchanged (`langchain`, `openai`, `crewai`, `middleware`, `otel`, `all`). One new dependency was added to the `otel` and `all` extras: | Extra | Change | @@ -421,6 +421,17 @@ from `galileo-python`, it can be deleted at leisure or simply ignored. The directory can be overridden via `SPLUNK_AO_HOME_DIR` (previously `GALILEO_HOME_DIR`). +### 5.4 `flush()` No Longer Required + +In `galileo`, `logger.flush()` uploaded the accumulated traces and was required +before the process exited. In `splunk-ao`, `conclude()` enqueues the span with the +batch processor and an `atexit` hook exports at interpreter exit, so an explicit +call is no longer needed. + +`flush()` still exists and still exports — call it to push spans out immediately +rather than waiting for the batch timer. For deterministic shutdown, call +`terminate()`, which drains and then shuts down the exporter. + --- ## 6. HTTP Tracing Headers @@ -462,6 +473,7 @@ logger = GalileoLogger(project="my-project", log_stream="production") logger.start_session(name="my-session") logger.add_llm_span(input="Hello", output="Hi", model="gpt-4") logger.conclude() # closes current span; no flush kwarg +logger.flush() # uploads traces ``` ### After (splunk-ao) @@ -488,6 +500,7 @@ logger = SplunkAOLogger(project="my-project", agent_stream="production") logger.start_session(name="my-session") logger.add_llm_span(input="Hello", output="Hi", model="gpt-4") logger.conclude() # closes current span; no flush kwarg +# no explicit flush() required — see 5.4 ``` --- diff --git a/splunk-ao-migration-tool/examples/before_galileo.py b/splunk-ao-migration-tool/examples/before_galileo.py index 3c70b763..7f7c1433 100644 --- a/splunk-ao-migration-tool/examples/before_galileo.py +++ b/splunk-ao-migration-tool/examples/before_galileo.py @@ -21,3 +21,4 @@ def call_llm(prompt: str) -> str: logger.start_session(name="my-session") logger.add_llm_span(input="Hello", output="Hi", model="gpt-4") logger.conclude() # closes current span; no flush kwarg +logger.flush() # uploads traces diff --git a/src/splunk_ao/decorator.py b/src/splunk_ao/decorator.py index fb02472b..911297bf 100644 --- a/src/splunk_ao/decorator.py +++ b/src/splunk_ao/decorator.py @@ -1309,9 +1309,12 @@ def flush( on_error: Callable[[Exception], None] | None = None, ) -> None: """ - Upload all captured traces under a project and agent stream context to Splunk AO. + Drain completed spans for the resolved project and agent stream context; does not conclude open spans. - If no project or agent stream is provided, then the currently initialized context is used. + Falls back to the currently initialized context when no project or agent stream is provided. + + Draining is not a shutdown: exporters stay open and the logger stays cached. Use + ``reset()`` to terminate the current context's loggers and clear the context. Parameters ---------- @@ -1349,17 +1352,26 @@ def _on_flush_error(exc: Exception) -> None: def flush_all(self) -> None: """ - Upload all captured traces under all contexts to Splunk AO. + Drain completed spans for every logger across all contexts. + + Open spans are left unconcluded, except for hook-backed loggers, which conclude any + open spans on the active trace before handing it off. - This method flushes all traces regardless of project or log stream. + Draining is not a shutdown: exporters stay open and every cached logger is retained. + Loggers terminate via their ``atexit`` hooks at interpreter exit. """ SplunkAOLoggerSingleton().flush_all() def reset(self) -> None: """ - Reset the entire context, which also deletes all traces that haven't been flushed. + Reset the entire context and terminate the loggers for the current context. + + Terminating drains completed spans before shutting the exporter down, so work that + has already concluded is still exported. Spans left open at that point are discarded + rather than exported. - This method clears all context variables and resets the logger singleton. + This method clears all context variables and stacks, and evicts the terminated + loggers from the singleton cache. """ SplunkAOLoggerSingleton().reset( project=_project_context.get(), diff --git a/src/splunk_ao/logger/logger.py b/src/splunk_ao/logger/logger.py index f7213c3b..5ee40aab 100644 --- a/src/splunk_ao/logger/logger.py +++ b/src/splunk_ao/logger/logger.py @@ -367,7 +367,9 @@ def __init__( "User must provide project_name or project_id to SplunkAOLogger, or set it as an environment variable." ) if self.experiment_id is None and self.agent_stream_name is None and self.agent_stream_id is None: - raise SplunkAOLoggerException("agent_stream or agent_stream_id is required to initialize SplunkAOLogger.") + raise SplunkAOLoggerException( + "agent_stream or agent_stream_id is required to initialize SplunkAOLogger." + ) if local_metrics: self.local_metrics = local_metrics @@ -2216,7 +2218,18 @@ def conclude( @nop_sync def flush(self, on_error: Callable[[Exception], None] | None = None) -> None: """ - Drain completed spans waiting in the batch processor. + Drain telemetry that is ready to export. + + Behavior depends on the egress path: + + - OTLP export (default): drains completed spans waiting in the batch processor. + Open spans are left alone; unconcluded steps are not converted or emitted. + - Ingestion hook: concludes any open spans on the active trace, computes local + metrics if configured, hands the accumulated traces to the hook, and clears + them. Distributed stub traces are left unconcluded. + + Neither path shuts down owned resources; call ``terminate()`` during application + teardown. Parameters ---------- @@ -2225,9 +2238,6 @@ def flush(self, on_error: Callable[[Exception], None] | None = None) -> None: is passed to the callback instead of being logged as a warning. The callback itself is protected: if it raises, the exception is logged as a warning. Defaults to None (swallow and log warning). - - Unconcluded steps are not converted or emitted. This method does not - shut down the processor; call ``terminate()`` during application teardown. """ try: if self._ingestion_hook: @@ -2254,7 +2264,12 @@ def flush(self, on_error: Callable[[Exception], None] | None = None) -> None: @nop_async @async_warn_catch_exception(exceptions=(Exception,)) async def async_flush(self) -> None: - """Drain completed spans without blocking the caller's event loop.""" + """ + Drain telemetry that is ready to export without blocking the caller's event loop. + + Path-dependent behavior matches ``flush()``, including concluding open spans and + clearing accumulated traces when an ingestion hook is configured. + """ if self._ingestion_hook: await self._flush_batch() return diff --git a/src/splunk_ao/utils/singleton.py b/src/splunk_ao/utils/singleton.py index 99c00017..86311a29 100644 --- a/src/splunk_ao/utils/singleton.py +++ b/src/splunk_ao/utils/singleton.py @@ -20,8 +20,12 @@ class SplunkAOLoggerSingleton: provides a thread-safe way to retrieve or create SplunkAOLogger clients based on the given 'project' and 'agent_stream' parameters. If the parameters are not provided, the class attempts to read the values from the environment variables - SPLUNK_AO_PROJECT and SPLUNK_AO_AGENT_STREAM. The loggers are stored in a dictionary - using a tuple (project, agent_stream) as the key. + SPLUNK_AO_PROJECT and SPLUNK_AO_AGENT_STREAM, falling back to the standalone defaults. + + Loggers are cached under a tuple key built from the calling thread's name, the logger + mode, the deployment, and the resolved project and agent stream (or experiment) + identity, plus the distributed trace and span IDs when present. Because the thread name + is part of the key, instances are never shared across threads. """ _instance = None # Class-level attribute to hold the singleton instance. @@ -244,7 +248,14 @@ def reset( agent_stream_id: str | None = None, ) -> None: """ - Reset (terminate and remove) one or all SplunkAOLogger instances. + Reset (terminate and remove) the SplunkAOLogger instances matching the given key. + + Matching is by key prefix, so a logger's per-trace and hook-backed variants are + included. With no arguments this covers the current thread's loggers at the default + mode and resolved routing, not every cached instance; use ``reset_all()`` for that. + + Terminating drains completed spans before shutting the exporter down. Spans still + open at that point are discarded rather than exported. Parameters ---------- @@ -295,11 +306,18 @@ def flush( agent_stream_id: str | None = None, ) -> None: """ - Flush (upload and clear) a SplunkAOLogger instance. + Drain completed spans for the matching cached SplunkAOLogger instances. + + With no arguments, drains the loggers registered for the current thread whose mode and + resolved project/agent stream match the active defaults — not every cached logger. + Passing a project or agent stream narrows this to the loggers matching that key. + + Open spans are left unconcluded, except for hook-backed loggers, which conclude any + open spans on the active trace before handing it off. - If both project and agent_stream are None, then all cached loggers are flushed - and cleared. Otherwise, only the specific logger corresponding to the provided - key (project, agent_stream) is flushed and removed. + Draining is not a shutdown: exporters stay open and the loggers remain cached. Use + ``reset()`` or ``reset_all()`` to terminate and evict them; otherwise each logger + terminates via its ``atexit`` hook at interpreter exit. Parameters ---------- @@ -331,9 +349,17 @@ def flush( self._splunk_ao_loggers[key].flush() def flush_all(self) -> None: - """Flush (upload and clear) all SplunkAOLogger instances.""" + """ + Drain completed spans for every cached SplunkAOLogger instance. + + Open spans are left unconcluded, except for hook-backed loggers, which conclude any + open spans on the active trace before handing it off. + + Draining is not a shutdown: exporters stay open and the loggers remain cached. Use + ``reset_all()`` to terminate and evict them; otherwise each logger terminates via its + ``atexit`` hook at interpreter exit. + """ with self._lock: - # Terminate and clear all logger instances. for logger in self._splunk_ao_loggers.values(): logger.flush()