Merge pull request #121 from Azure-Samples/tracing

Tracing

.devcontainer/devcontainer.json

@@ -34,8 +34,6 @@
     },
     "forwardPorts": [
         5173,
-	8000, 
-	8080
-
+	8000
 	]
 }

src/api/main.py

@@ -3,10 +3,11 @@
 from fastapi import FastAPI
 from dotenv import load_dotenv
 from prompty.tracer import trace
-from prompty.core import PromptyStream
+from prompty.core import PromptyStream, AsyncPromptyStream
 from fastapi.responses import StreamingResponse
 from fastapi.middleware.cors import CORSMiddleware
 from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
+from azure.monitor.opentelemetry import configure_azure_monitor
 
 from tracing import init_tracing
 from orchestrator import Task, create
@@ -20,11 +21,14 @@
 app = FastAPI()
 
 code_space = os.getenv("CODESPACE_NAME")
+app_insights = os.getenv("APPINSIGHTS_CONNECTIONSTRING")
 
 if code_space: 
     origin_8000= f"https://{code_space}-8000.app.github.dev"
     origin_5173 = f"https://{code_space}-5173.app.github.dev"
-    origins = [origin_8000, origin_5173, os.getenv("API_SERVICE_ACA_URI"), os.getenv("WEB_SERVICE_ACA_URI")]
+    ingestion_endpoint = app_insights.split(';')[1].split('=')[1]
+
+    origins = [origin_8000, origin_5173, os.getenv("API_SERVICE_ACA_URI"), os.getenv("WEB_SERVICE_ACA_URI"), ingestion_endpoint]
 else:
     origins = [
         o.strip()
@@ -58,4 +62,4 @@ async def create_article(task: Task):
 
 
 # TODO: fix open telemetry so it doesn't slow app so much
-# FastAPIInstrumentor.instrument_app(app)
+FastAPIInstrumentor.instrument_app(app)

src/api/tracing.py

@@ -3,10 +3,13 @@
 import logging
 import contextlib
 from typing import AsyncIterator, List
-from prompty.tracer import Tracer, PromptyTracer, console_tracer, to_dict
+from prompty.tracer import Tracer, PromptyTracer
 from opentelemetry import trace as oteltrace
+from opentelemetry import trace
 from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace.export import BatchSpanProcessor
 from opentelemetry.sdk.trace.export import SimpleSpanProcessor
+from opentelemetry.sdk.trace.sampling import ParentBasedTraceIdRatio
 from azure.monitor.opentelemetry.exporter import AzureMonitorTraceExporter
 
 _tracer = "prompty"
@@ -17,7 +20,7 @@ def trace_span(name: str):
     tracer = oteltrace.get_tracer(_tracer)
     with tracer.start_as_current_span(name) as span:
         yield lambda key, value: span.set_attribute(
-            key, json.dumps(value).replace("\n", "")
+            key, json.dumps(value).replace("\n", "") 
         )
 
 
@@ -33,16 +36,18 @@ def init_tracing(local_tracing: bool = False):
     azmon_logger = logging.getLogger("azure")
     azmon_logger.setLevel(logging.INFO)
 
-    oteltrace.set_tracer_provider(TracerProvider())
+    # oteltrace.set_tracer_provider(TracerProvider())
 
     # Configure Azure Monitor as the Exporter
     app_insights = os.getenv("APPINSIGHTS_CONNECTIONSTRING")
-    trace_exporter = AzureMonitorTraceExporter(connection_string=app_insights)
 
     # Add the Azure exporter to the tracer provider
-    oteltrace.get_tracer_provider().add_span_processor(
-        SimpleSpanProcessor(trace_exporter)
-    )
+
+    oteltrace.set_tracer_provider(TracerProvider(sampler=ParentBasedTraceIdRatio(1.0)))
+    oteltrace.get_tracer_provider().add_span_processor(BatchSpanProcessor(AzureMonitorTraceExporter(connection_string=app_insights)))
+    # oteltrace.get_tracer_provider().add_span_processor(
+    #     SimpleSpanProcessor(trace_exporter)
+    # )
 
     return oteltrace.get_tracer(_tracer)
 

src/web/src/store/index.ts

@@ -30,25 +30,25 @@ export const startWritingTask = (
   addToArticle: { (text: string): void }
 ) => {
   // internal function to read chunks from a stream
-  function readChunks(reader: ReadableStreamDefaultReader<Uint8Array>) {
-    return {
-      async *[Symbol.asyncIterator]() {
-        let readResult = await reader.read();
-        while (!readResult.done) {
-          yield readResult.value;
-          readResult = await reader.read();
-        }
-      },
-    };
+  const readChunks = async (reader: ReadableStreamDefaultReader<Uint8Array>) => {
+      return {
+        async *[Symbol.asyncIterator]() {
+          let readResult = await reader.read();
+          while (!readResult.done) {
+            yield readResult.value;
+            readResult = await reader.read();
+          }
+        },
+    }  
   }
 
-
   const configuration = {
     method: "POST",
     headers: {
       "Content-Type": "application/json",
       "Connection": "keep-alive",
     },
+    signal:AbortSignal.timeout(100), 
     body: JSON.stringify({
       research: research,
       products: products,
@@ -62,11 +62,11 @@ export const startWritingTask = (
 
   const callApi = async () => {
     try {
-      const response = await fetch(url, configuration);
-      const reader = response.body?.getReader();
+      const response = await fetch(url, configuration).catch();
+      const reader = await response.body?.getReader();
       if (!reader) return;
 
-      const chunks = readChunks(reader);
+      const chunks = await readChunks(reader);
       for await (const chunk of chunks) {
         const text = new TextDecoder().decode(chunk);
         const parts = text.split("\n");
@@ -91,7 +91,7 @@ export const startWritingTask = (
         }
       }
     } catch (e) {
-      console.log('complete');
+      console.log(e);
     }
   };