Distributed Tracing

1
import os
2
import requests
3
from flask import Flask, request, jsonify
4
from opentelemetry import trace
5
from opentelemetry.sdk.trace import TracerProvider
6
from opentelemetry.sdk.trace.export import BatchSpanProcessor
7
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
8
from opentelemetry.propagate import inject
9
10
app = Flask(__name__)
11
12
# OpenTelemetry setup
13
OTLP_ENDPOINT = os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT")
14
CONFIDENT_API_KEY = os.getenv("CONFIDENT_API_KEY")
15
16
trace_provider = TracerProvider()
17
exporter = OTLPSpanExporter(
18
    endpoint=f"{OTLP_ENDPOINT}/v1/traces",
19
    headers={"x-confident-api-key": CONFIDENT_API_KEY},
20
)
21
trace_provider.add_span_processor(BatchSpanProcessor(exporter))
22
trace.set_tracer_provider(trace_provider)
23
tracer = trace.get_tracer("api-gateway")
24
25
26
@app.route("/chat", methods=["POST"])
27
def chat():
28
    user_query = request.json["query"]
29
    user_id = request.json.get("user_id", "anonymous")
30
31
    with tracer.start_as_current_span("api-gateway") as span:
32
        # Set trace-level attributes (apply to entire trace)
33
        span.set_attribute("confident.trace.name", "rag-pipeline")
34
        span.set_attribute("confident.trace.input", user_query)
35
        span.set_attribute("confident.trace.user_id", user_id)
36
        span.set_attribute("confident.trace.tags", ["rag", "production", "multi-language"])
37
38
        # Set span-level attributes
39
        span.set_attribute("confident.span.type", "agent")
40
        span.set_attribute("confident.span.input", user_query)
41
42
        # Inject trace context into headers for downstream service
43
        headers = {"Content-Type": "application/json"}
44
        inject(headers)
45
46
        # Call Query Processor (TypeScript service)
47
        response = requests.post(
48
            "http://query-processor:3000/process",
49
            json={"query": user_query},
50
            headers=headers
51
        )
52
        result = response.json()
53
54
        span.set_attribute("confident.span.output", result["answer"])
55
        span.set_attribute("confident.trace.output", result["answer"])
56
57
        return jsonify(result)
58
59
60
if __name__ == "__main__":
61
    app.run(host="0.0.0.0", port=8000)

1
import express from "express";
2
import * as opentelemetry from "@opentelemetry/api";
3
import { NodeTracerProvider } from "@opentelemetry/sdk-trace-node";
4
import { BatchSpanProcessor } from "@opentelemetry/sdk-trace-base";
5
import { OTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-proto";
6
import { W3CTraceContextPropagator } from "@opentelemetry/core";
7
8
const app = express();
9
app.use(express.json());
10
11
// OpenTelemetry setup
12
const OTLP_ENDPOINT = process.env.OTEL_EXPORTER_OTLP_ENDPOINT;
13
const CONFIDENT_API_KEY = process.env.CONFIDENT_API_KEY;
14
15
const provider = new NodeTracerProvider({
16
  spanProcessors: [
17
    new BatchSpanProcessor(
18
      new OTLPTraceExporter({
19
        url: `${OTLP_ENDPOINT}/v1/traces`,
20
        headers: { "x-confident-api-key": CONFIDENT_API_KEY || "" },
21
      })
22
    ),
23
  ],
24
});
25
26
opentelemetry.propagation.setGlobalPropagator(new W3CTraceContextPropagator());
27
opentelemetry.trace.setGlobalTracerProvider(provider);
28
const tracer = opentelemetry.trace.getTracer("query-processor");
29
30
app.post("/process", async (req, res) => {
31
  // Extract trace context from incoming headers
32
  const parentContext = opentelemetry.propagation.extract(
33
    opentelemetry.context.active(),
34
    req.headers
35
  );
36
37
  // Run within the extracted context
38
  await opentelemetry.context.with(parentContext, async () => {
39
    await tracer.startActiveSpan("query-processor", async (span) => {
40
      const query = req.body.query;
41
42
      span.setAttributes({
43
        "confident.span.type": "tool",
44
        "confident.tool.name": "query-processor",
45
        "confident.tool.description": "Processes and validates user queries",
46
        "confident.span.input": query,
47
      });
48
49
      // Prepare headers with trace context for downstream calls
50
      const headers: Record<string, string> = {
51
        "Content-Type": "application/json",
52
      };
53
      opentelemetry.propagation.inject(opentelemetry.context.active(), headers);
54
55
      // Call Retrieval Service (Go)
56
      const retrievalResponse = await fetch(
57
        "http://retrieval-service:8080/retrieve",
58
        {
59
          method: "POST",
60
          headers,
61
          body: JSON.stringify({ query }),
62
        }
63
      );
64
      const { contexts } = await retrievalResponse.json();
65
66
      // Call LLM Service (Java) with retrieved context
67
      const llmResponse = await fetch("http://llm-service:8081/generate", {
68
        method: "POST",
69
        headers,
70
        body: JSON.stringify({ query, contexts }),
71
      });
72
      const { answer } = await llmResponse.json();
73
74
      span.setAttribute(
75
        "confident.span.output",
76
        JSON.stringify({ answer, contexts })
77
      );
78
      span.end();
79
80
      res.json({ answer, contexts });
81
    });
82
  });
83
});
84
85
app.listen(3000, () => console.log("Query Processor running on port 3000"));

1
package main
2
3
import (
4
    "context"
5
    "encoding/json"
6
    "net/http"
7
    "os"
8
9
    "go.opentelemetry.io/otel"
10
    "go.opentelemetry.io/otel/attribute"
11
    "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp"
12
    "go.opentelemetry.io/otel/propagation"
13
    sdktrace "go.opentelemetry.io/otel/sdk/trace"
14
)
15
16
var tracer = otel.Tracer("retrieval-service")
17
18
func initTracer() *sdktrace.TracerProvider {
19
    endpoint := os.Getenv("OTLP_ENDPOINT") // "otel.confident-ai.com"
20
    apiKey := os.Getenv("CONFIDENT_API_KEY")
21
22
    exporter, _ := otlptracehttp.New(context.Background(),
23
        otlptracehttp.WithEndpoint(endpoint),
24
        otlptracehttp.WithHeaders(map[string]string{"x-confident-api-key": apiKey}),
25
    )
26
27
    tp := sdktrace.NewTracerProvider(sdktrace.WithBatcher(exporter))
28
    otel.SetTracerProvider(tp)
29
    otel.SetTextMapPropagator(propagation.TraceContext{})
30
    return tp
31
}
32
33
type RetrievalRequest struct {
34
	Query string `json:"query"`
35
}
36
37
type RetrievalResponse struct {
38
	Contexts []string `json:"contexts"`
39
}
40
41
func retrieveHandler(w http.ResponseWriter, r *http.Request) {
42
	// Extract trace context from incoming headers
43
	ctx := otel.GetTextMapPropagator().Extract(r.Context(), propagation.HeaderCarrier(r.Header))
44
45
	_, span := tracer.Start(ctx, "vector-search")
46
	defer span.End()
47
48
	var req RetrievalRequest
49
	json.NewDecoder(r.Body).Decode(&req)
50
51
	// Set retriever span attributes
52
	span.SetAttributes(
53
		attribute.String("confident.span.type", "retriever"),
54
		attribute.String("confident.retriever.embedder", "text-embedding-3-small"),
55
		attribute.String("confident.span.input", req.Query),
56
		attribute.Int("confident.retriever.top_k", 3),
57
		attribute.Int("confident.retriever.chunk_size", 512),
58
	)
59
60
	// Simulate vector search results
61
	contexts := []string{
62
		"Paris is the capital and largest city of France, with a population of over 2 million.",
63
		"France is a country in Western Europe, known for its rich history and culture.",
64
		"The Eiffel Tower, built in 1889, is located in Paris and stands 330 meters tall.",
65
	}
66
67
	span.SetAttributes(
68
		attribute.StringSlice("confident.retriever.retrieval_context", contexts),
69
	)
70
71
	w.Header().Set("Content-Type", "application/json")
72
	json.NewEncoder(w).Encode(RetrievalResponse{Contexts: contexts})
73
}
74
75
func main() {
76
	tp := initTracer()
77
	defer tp.Shutdown(context.Background())
78
79
	http.HandleFunc("/retrieve", retrieveHandler)
80
	http.ListenAndServe(":8080", nil)
81
}

1
package com.example;
2
3
import io.opentelemetry.api.GlobalOpenTelemetry;
4
import io.opentelemetry.api.trace.Span;
5
import io.opentelemetry.api.trace.Tracer;
6
import io.opentelemetry.context.Context;
7
import io.opentelemetry.context.propagation.TextMapGetter;
8
import io.opentelemetry.exporter.otlp.http.trace.OtlpHttpSpanExporter;
9
import io.opentelemetry.sdk.OpenTelemetrySdk;
10
import io.opentelemetry.sdk.trace.SdkTracerProvider;
11
import io.opentelemetry.sdk.trace.export.BatchSpanProcessor;
12
import com.sun.net.httpserver.HttpServer;
13
import com.sun.net.httpserver.HttpExchange;
14
import com.google.gson.Gson;
15
16
import java.io.*;
17
import java.net.InetSocketAddress;
18
import java.util.List;
19
import java.util.Map;
20
21
public class LlmService {
22
    private static final Gson gson = new Gson();
23
    private static Tracer tracer;
24
25
    public static void main(String[] args) throws IOException {
26
        initTracer();
27
28
        HttpServer server = HttpServer.create(new InetSocketAddress(8081), 0);
29
        server.createContext("/generate", LlmService::handleGenerate);
30
        server.start();
31
        System.out.println("LLM Service running on port 8081");
32
    }
33
34
    private static void initTracer() {
35
        String endpoint = System.getenv("OTEL_EXPORTER_OTLP_ENDPOINT");
36
        String apiKey = System.getenv("CONFIDENT_API_KEY");
37
38
        OtlpHttpSpanExporter exporter = OtlpHttpSpanExporter.builder()
39
            .setEndpoint(endpoint + "/v1/traces")
40
            .addHeader("x-confident-api-key", apiKey)
41
            .build();
42
43
        SdkTracerProvider tracerProvider = SdkTracerProvider.builder()
44
            .addSpanProcessor(BatchSpanProcessor.builder(exporter).build())
45
            .build();
46
47
        OpenTelemetrySdk.builder()
48
            .setTracerProvider(tracerProvider)
49
            .buildAndRegisterGlobal();
50
51
        tracer = GlobalOpenTelemetry.getTracer("llm-service");
52
    }
53
54
    private static void handleGenerate(HttpExchange exchange) throws IOException {
55
        // Extract trace context from headers
56
        Context extractedContext = GlobalOpenTelemetry.getPropagators()
57
            .getTextMapPropagator()
58
            .extract(Context.current(), exchange.getRequestHeaders(), new TextMapGetter<>() {
59
                @Override
60
                public Iterable<String> keys(com.sun.net.httpserver.Headers carrier) {
61
                    return carrier.keySet();
62
                }
63
                @Override
64
                public String get(com.sun.net.httpserver.Headers carrier, String key) {
65
                    List<String> values = carrier.get(key);
66
                    return values != null && !values.isEmpty() ? values.get(0) : null;
67
                }
68
            });
69
70
        // Create span within extracted context
71
        Span span = tracer.spanBuilder("llm-generation")
72
            .setParent(extractedContext)
73
            .startSpan();
74
75
        try {
76
            // Parse request
77
            InputStreamReader reader = new InputStreamReader(exchange.getRequestBody());
78
            Map<String, Object> request = gson.fromJson(reader, Map.class);
79
            String query = (String) request.get("query");
80
            List<String> contexts = (List<String>) request.get("contexts");
81
82
            // Set LLM span attributes
83
            span.setAttribute("confident.span.type", "llm");
84
            span.setAttribute("confident.llm.model", "gpt-4o");
85
            span.setAttribute("confident.span.input", gson.toJson(Map.of(
86
                "messages", List.of(
87
                    Map.of("role", "system", "content", "Context: " + String.join(" ", contexts)),
88
                    Map.of("role", "user", "content", query)
89
                )
90
            )));
91
92
            // Simulate LLM response
93
            String answer = "Paris is the capital of France. It is the largest city in France " +
94
                           "with over 2 million residents, and is home to the iconic Eiffel Tower, " +
95
                           "which was built in 1889 and stands 330 meters tall.";
96
97
            span.setAttribute("confident.span.output", answer);
98
            span.setAttribute("confident.llm.input_token_count", 180);
99
            span.setAttribute("confident.llm.output_token_count", 52);
100
101
            // Send response
102
            String response = gson.toJson(Map.of("answer", answer));
103
            exchange.getResponseHeaders().set("Content-Type", "application/json");
104
            exchange.sendResponseHeaders(200, response.length());
105
            exchange.getResponseBody().write(response.getBytes());
106
107
        } finally {
108
            span.end();
109
            exchange.close();
110
        }
111
    }
112
}

1
import os
2
import json
3
import asyncio
4
from opentelemetry import trace
5
from opentelemetry.sdk.trace import TracerProvider
6
from opentelemetry.sdk.trace.export import BatchSpanProcessor
7
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
8
from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
9
from mcp import ClientSession, StdioServerParameters
10
from mcp.client.stdio import stdio_client
11
12
# OpenTelemetry setup
13
OTLP_ENDPOINT = os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT")
14
CONFIDENT_API_KEY = os.getenv("CONFIDENT_API_KEY")
15
16
trace_provider = TracerProvider()
17
exporter = OTLPSpanExporter(
18
    endpoint=f"{OTLP_ENDPOINT}/v1/traces",
19
    headers={"x-confident-api-key": CONFIDENT_API_KEY},
20
)
21
trace_provider.add_span_processor(BatchSpanProcessor(exporter))
22
trace.set_tracer_provider(trace_provider)
23
tracer = trace.get_tracer("mcp-host")
24
propagator = TraceContextTextMapPropagator()
25
26
27
def inject_trace_context() -> dict:
28
    """Inject current trace context into a dict for MCP metadata."""
29
    carrier = {}
30
    propagator.inject(carrier)
31
    return carrier
32
33
34
async def call_tool_with_tracing(session: ClientSession, tool_name: str, arguments: dict):
35
    """Call an MCP tool with trace context propagation."""
36
    with tracer.start_as_current_span(f"mcp-tool-{tool_name}") as span:
37
        span.set_attribute("confident.span.type", "tool")
38
        span.set_attribute("confident.tool.name", tool_name)
39
        span.set_attribute("confident.span.input", json.dumps(arguments))
40
41
        # Inject trace context into MCP request metadata
42
        trace_meta = inject_trace_context()
43
44
        # Call the MCP tool with trace context in _meta
45
        result = await session.call_tool(
46
            tool_name,
47
            arguments=arguments,
48
            _meta=trace_meta  # Pass trace context
49
        )
50
51
        span.set_attribute("confident.span.output", json.dumps(result.content))
52
        return result
53
54
55
async def process_query(query: str):
56
    """Process a user query using MCP tools."""
57
    with tracer.start_as_current_span("mcp-agent") as span:
58
        span.set_attribute("confident.trace.name", "mcp-tool-orchestration")
59
        span.set_attribute("confident.span.type", "agent")
60
        span.set_attribute("confident.span.input", query)
61
        span.set_attribute("confident.agent.name", "mcp-orchestrator")
62
        span.set_attribute("confident.agent.available_tools", [
63
            "read_file", "query_database", "write_file"
64
        ])
65
66
        # Connect to File Server (TypeScript)
67
        async with stdio_client(StdioServerParameters(
68
            command="npx",
69
            args=["ts-node", "file-server/index.ts"]
70
        )) as (read, write):
71
            async with ClientSession(read, write) as file_session:
72
                await file_session.initialize()
73
74
                # Call read_file tool with tracing
75
                file_result = await call_tool_with_tracing(
76
                    file_session,
77
                    "read_file",
78
                    {"path": "/data/sales.csv"}
79
                )
80
81
        # Connect to Database Server (Python)
82
        async with stdio_client(StdioServerParameters(
83
            command="python",
84
            args=["db-server/main.py"]
85
        )) as (read, write):
86
            async with ClientSession(read, write) as db_session:
87
                await db_session.initialize()
88
89
                # Call query_database tool with tracing
90
                db_result = await call_tool_with_tracing(
91
                    db_session,
92
                    "query_database",
93
                    {"sql": "SELECT * FROM sales WHERE year = 2024"}
94
                )
95
96
        # Generate summary (simulated LLM call)
97
        with tracer.start_as_current_span("llm-summarize") as llm_span:
98
            llm_span.set_attribute("confident.span.type", "llm")
99
            llm_span.set_attribute("confident.llm.model", "claude-3-5-sonnet")
100
101
            summary = "Sales increased 23% YoY with Q4 showing strongest growth."
102
103
            llm_span.set_attribute("confident.span.output", summary)
104
105
        span.set_attribute("confident.span.output", summary)
106
        return summary
107
108
109
async def main():
110
    result = await process_query("Summarize our 2024 sales data")
111
    print(f"Result: {result}")
112
    trace_provider.force_flush()
113
114
115
if __name__ == "__main__":
116
    asyncio.run(main())

1
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
2
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
3
import * as opentelemetry from "@opentelemetry/api";
4
import { NodeTracerProvider } from "@opentelemetry/sdk-trace-node";
5
import { BatchSpanProcessor } from "@opentelemetry/sdk-trace-base";
6
import { OTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-proto";
7
import { W3CTraceContextPropagator } from "@opentelemetry/core";
8
import * as fs from "fs/promises";
9
10
// OpenTelemetry setup
11
const OTLP_ENDPOINT = process.env.OTEL_EXPORTER_OTLP_ENDPOINT;
12
const CONFIDENT_API_KEY = process.env.CONFIDENT_API_KEY;
13
14
const provider = new NodeTracerProvider({
15
  spanProcessors: [
16
    new BatchSpanProcessor(
17
      new OTLPTraceExporter({
18
        url: `${OTLP_ENDPOINT}/v1/traces`,
19
        headers: { "x-confident-api-key": CONFIDENT_API_KEY || "" },
20
      })
21
    ),
22
  ],
23
});
24
25
const propagator = new W3CTraceContextPropagator();
26
opentelemetry.propagation.setGlobalPropagator(propagator);
27
opentelemetry.trace.setGlobalTracerProvider(provider);
28
const tracer = opentelemetry.trace.getTracer("mcp-file-server");
29
30
// Create MCP server
31
const server = new Server(
32
  { name: "file-server", version: "1.0.0" },
33
  { capabilities: { tools: {} } }
34
);
35
36
// Define tools
37
server.setRequestHandler("tools/list", async () => ({
38
  tools: [
39
    {
40
      name: "read_file",
41
      description: "Read contents of a file",
42
      inputSchema: {
43
        type: "object",
44
        properties: {
45
          path: { type: "string", description: "File path to read" },
46
        },
47
        required: ["path"],
48
      },
49
    },
50
  ],
51
}));
52
53
server.setRequestHandler("tools/call", async (request) => {
54
  const { name, arguments: args, _meta } = request.params;
55
56
  // Extract trace context from MCP metadata
57
  let parentContext = opentelemetry.context.active();
58
  if (_meta?.traceparent) {
59
    parentContext = opentelemetry.propagation.extract(
60
      opentelemetry.context.active(),
61
      _meta
62
    );
63
  }
64
65
  // Execute within parent context
66
  return opentelemetry.context.with(parentContext, async () => {
67
    return tracer.startActiveSpan(`tool-${name}`, async (span) => {
68
      span.setAttributes({
69
        "confident.span.type": "tool",
70
        "confident.tool.name": name,
71
        "confident.tool.description": "MCP file system tool",
72
        "confident.span.input": JSON.stringify(args),
73
      });
74
75
      try {
76
        if (name === "read_file") {
77
          const content = await fs.readFile(args.path, "utf-8");
78
          span.setAttribute("confident.span.output", content.slice(0, 1000));
79
          span.end();
80
          return { content: [{ type: "text", text: content }] };
81
        }
82
83
        throw new Error(`Unknown tool: ${name}`);
84
      } catch (error) {
85
        span.recordException(error as Error);
86
        span.end();
87
        throw error;
88
      }
89
    });
90
  });
91
});
92
93
// Start server
94
const transport = new StdioServerTransport();
95
server.connect(transport);

1
import os
2
import json
3
from opentelemetry import trace
4
from opentelemetry.sdk.trace import TracerProvider
5
from opentelemetry.sdk.trace.export import BatchSpanProcessor
6
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
7
from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
8
from mcp.server import Server
9
from mcp.server.stdio import stdio_server
10
11
# OpenTelemetry setup
12
OTLP_ENDPOINT = os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT")
13
CONFIDENT_API_KEY = os.getenv("CONFIDENT_API_KEY")
14
15
trace_provider = TracerProvider()
16
exporter = OTLPSpanExporter(
17
    endpoint=f"{OTLP_ENDPOINT}/v1/traces",
18
    headers={"x-confident-api-key": CONFIDENT_API_KEY},
19
)
20
trace_provider.add_span_processor(BatchSpanProcessor(exporter))
21
trace.set_tracer_provider(trace_provider)
22
tracer = trace.get_tracer("mcp-db-server")
23
propagator = TraceContextTextMapPropagator()
24
25
# Create MCP server
26
server = Server("db-server")
27
28
29
@server.list_tools()
30
async def list_tools():
31
    return [
32
        {
33
            "name": "query_database",
34
            "description": "Execute a SQL query",
35
            "inputSchema": {
36
                "type": "object",
37
                "properties": {
38
                    "sql": {"type": "string", "description": "SQL query to execute"},
39
                },
40
                "required": ["sql"],
41
            },
42
        }
43
    ]
44
45
46
@server.call_tool()
47
async def call_tool(name: str, arguments: dict, _meta: dict = None):
48
    # Extract trace context from MCP metadata
49
    parent_context = None
50
    if _meta and "traceparent" in _meta:
51
        parent_context = propagator.extract(carrier=_meta)
52
53
    with tracer.start_as_current_span(
54
        f"tool-{name}",
55
        context=parent_context
56
    ) as span:
57
        span.set_attribute("confident.span.type", "tool")
58
        span.set_attribute("confident.tool.name", name)
59
        span.set_attribute("confident.tool.description", "MCP database tool")
60
        span.set_attribute("confident.span.input", json.dumps(arguments))
61
62
        if name == "query_database":
63
            sql = arguments["sql"]
64
65
            # Simulate database query
66
            results = [
67
                {"month": "Jan", "revenue": 125000},
68
                {"month": "Feb", "revenue": 142000},
69
                {"month": "Mar", "revenue": 158000},
70
            ]
71
72
            output = json.dumps(results)
73
            span.set_attribute("confident.span.output", output)
74
75
            return {"content": [{"type": "text", "text": output}]}
76
77
        raise ValueError(f"Unknown tool: {name}")
78
79
80
async def main():
81
    async with stdio_server() as (read, write):
82
        await server.run(read, write, server.create_initialization_options())
83
84
85
if __name__ == "__main__":
86
    import asyncio
87
    asyncio.run(main())