oneagent-sdk

import os from oneagent_sdk import SDKConfig, OneAgentSDK, SDKState, ChannelType, DatabaseVendor, MessageBusVendor def initialize_sdk(): """Initializes the Dynatrace OneAgent SDK.""" # SDKConfig can be customized, e.g., to enable/disable specific features. # For most cases, the default configuration is sufficient as it connects # to the local OneAgent automatically. config = SDKConfig() sdk = OneAgentSDK(config) if sdk.get_current_state() != SDKState.ACTIVE: print("Dynatrace OneAgent SDK is not active. Is OneAgent running and configured?") print(f"Current state: {sdk.get_current_state().name}") # In a production environment, you might log this as a warning or error # and ensure your application can run gracefully without SDK full functionality. else: print("Dynatrace OneAgent SDK is active.") return sdk def example_tracing_and_database_call(sdk: OneAgentSDK): """Demonstrates basic tracing for a custom service, database, and message bus.""" print("\n--- Starting example tracing ---") # Trace a custom service method with sdk.trace_custom_service("MyPythonApplication", "process_order"): print("Processing an order...") # Simulate a database call db_info = sdk.create_database_info( name="CustomerDB", vendor=DatabaseVendor.POSTGRESQL, channel_type=ChannelType.TCP_IP, channel_endpoint="db.example.com:5432", # You can also specify user, port, etc., but avoid sensitive data ) with sdk.trace_database_request(db_info, "SELECT * FROM orders WHERE status = 'pending';"): print("Executing database query...") import time time.sleep(0.1) # Simulate network/DB latency print("Database operation complete.") # Simulate a message bus interaction (e.g., receiving a message) msg_bus_info = sdk.create_message_bus_info( vendor=MessageBusVendor.RABBITMQ, destination_name="order_queue", # You might specify topic, broker, etc., depending on the vendor ) with sdk.trace_message_bus_receive_activity(msg_bus_info, "order_msg_001"): print("Receiving message from message bus...") time.sleep(0.05) # Simulate message processing print("Message bus interaction complete.") print("Order processing finished. OneAgent should have captured this trace.") print("--- Example tracing finished ---\n") if __name__ == "__main__": sdk_instance = initialize_sdk() if sdk_instance.get_current_state() == SDKState.ACTIVE: example_tracing_and_database_call(sdk_instance) else: print("SDK not active, skipping tracing examples. Please ensure Dynatrace OneAgent is running.")