opentelemetry-instrumentation-weaviate

import os import weaviate from opentelemetry import trace from opentelemetry.sdk.resources import Resource from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import ConsoleSpanExporter, SimpleSpanProcessor from opentelemetry.instrumentation.weaviate import WeaviateInstrumentor # 1. Setup OpenTelemetry TracerProvider and Exporter resource = Resource.create({"service.name": "weaviate-app"}) provider = TracerProvider(resource=resource) processor = SimpleSpanProcessor(ConsoleSpanExporter()) provider.add_span_processor(processor) trace.set_tracer_provider(provider) # 2. Instrument Weaviate WeaviateInstrumentor().instrument() # 3. Use Weaviate client (ensure a Weaviate instance is running or mocked) # Replace with your actual Weaviate URL and API key if needed weaviate_url = os.environ.get("WEAVIATE_URL", "http://localhost:8080") weaviate_api_key = os.environ.get("WEAVIATE_API_KEY", "YOUR_API_KEY_HERE") try: client = weaviate.Client( url=weaviate_url # For Weaviate Cloud Services (WCS), uncomment and provide API key: # auth_client_secret=weaviate.AuthApiKey(api_key=weaviate_api_key) ) if client.is_connected(): print("Successfully connected to Weaviate. Spans will be generated for operations.") # Example: Perform a simple query (uncomment and adapt to your schema) # collection = client.collections.get("YourCollectionName") # response = collection.query.fetch_objects(limit=1) # print(f"Query executed. Response: {response}") else: print(f"Could not connect to Weaviate at {weaviate_url}. " "Please ensure Weaviate is running and accessible.") except Exception as e: print(f"Error connecting to Weaviate or during operation: {e}") print("Please ensure your Weaviate instance is running and accessible.") # Optional: Flush spans to ensure they are exported before program exit provider.force_flush()