py-zipkin

import requests import os from py_zipkin.zipkin import zipkin_span from py_zipkin.transport import BaseTransportHandler # Configure your Zipkin collector endpoint (e.g., http://localhost:9411/api/v2/spans) ZIPKIN_COLLECTOR_ENDPOINT = os.environ.get('ZIPKIN_COLLECTOR_ENDPOINT', 'http://localhost:9411/api/v2/spans') # A simple HTTP transport handler for demonstration class HTTPTransportHandler(BaseTransportHandler): def send(self, encoded_span): # The collector expects a thrift-encoded list of spans. try: requests.post( ZIPKIN_COLLECTOR_ENDPOINT, data=encoded_span, headers={'Content-Type': 'application/x-thrift'}, timeout=5 ) except requests.exceptions.RequestException as e: print(f"Failed to send span to Zipkin: {e}") def get_max_payload_bytes(self): return None # No explicit limit for this example def do_stuff(a, b): print(f"Doing stuff with {a} and {b}") return a + b # Instantiate the transport handler my_transport_handler = HTTPTransportHandler() # Usage as a context manager def my_function(x, y): with zipkin_span( service_name='my_python_service', span_name='my_function_span', transport_handler=my_transport_handler, port=8000, # Optional: Port of the service sample_rate=100.0 # Sample all traces for demonstration (0.0 - 100.0) ) as zipkin_context: result = do_stuff(x, y) zipkin_context.update_binary_annotations({'input_x': x, 'input_y': y, 'result': result}) print(f"Trace for my_function completed with result: {result}") return result # Usage as a decorator @zipkin_span( service_name='my_python_service', span_name='decorated_function', transport_handler=my_transport_handler, sample_rate=50.0 ) def another_function(data): print(f"Processing data: {data}") return data.upper() if __name__ == "__main__": print("--- Running context manager example ---") my_function(10, 20) print("\n--- Running decorator example ---") another_function("hello world") print("Check your Zipkin UI (e.g., http://localhost:9411/zipkin) for traces.")