Update Dockerfile to use local wheel file and enable observability in myapp_2.py

Author: aanshu-ss

dist/singlestore_pulse-0.3-py3-none-any.whl

@@ -5,7 +5,9 @@ WORKDIR /app
 
 COPY requirements.txt .
 RUN pip install --no-cache-dir -r requirements.txt
-RUN pip3 install git+https://github.com/singlestore-labs/singlestore-pulse.git@master
+COPY singlestore_pulse-0.3-py3-none-any.whl .
+RUN pip install --no-cache-dir singlestore_pulse-0.3-py3-none-any.whl
+# RUN pip3 install git+https://github.com/singlestore-labs/singlestore-pulse.git@master
 
 COPY myapp_2.py .
 COPY .env .

dist/singlestore_pulse-0.3.tar.gz

@@ -326,9 +326,9 @@ def cftocf_endpoint(request: Request, body: Item):
 
 def main():
     # write to otel collector 
-    _ = Pulse(
-        otel_collector_endpoint="http://otel-collector:4317",
-    )
+    # _ = Pulse(
+    #     otel_collector_endpoint="http://otel-collector:4317",
+    # )
 
     # Create a FastAPI app
     uvicorn.run(app, host="0.0.0.0", port=8000)

examples/otel-collector/Dockerfile.myapp_2

@@ -13,7 +13,7 @@
 
 from opentelemetry.sdk._logs import LoggingHandler
 
-from pulse_otel import Pulse, pulse_agent, pulse_tool
+from pulse_otel import Pulse, pulse_agent, pulse_tool, observe
 
 import logging
 from tenacity import retry, stop_after_attempt, wait_fixed
@@ -179,7 +179,7 @@ def health_check():
 def root():
     return {"message": "Welcome to the Pulse OTel FastAPI agent!"}
 
-# @pulse_agent("getdata")
+@pulse_agent("getdata")
 @app.post("/getdata")
 def cftocf_endpoint(request: Request, body: Item):
     """
@@ -255,8 +255,9 @@ def http_req(body: Item):
         logger.error(f"Request error occurred when calling myapp_2: {e}")
         raise HTTPException(status_code=500, detail="Failed to make request to myapp_2 service")
 
-# @pulse_agent("getdata")
+
 @app.post("/go_py_py")
+@observe("cftocf_endpoint")
 def cftocf_endpoint(request: Request, body: Item):
     """
     This is the target endpoint that myapp will call.
@@ -284,9 +285,9 @@ def cftocf_endpoint(request: Request, body: Item):
 
 def main():
     # write to otel collector 
-    # _ = Pulse(
-    #     otel_collector_endpoint="http://otel-collector:4317",
-    # )
+    _ = Pulse(
+        otel_collector_endpoint="http://otel-collector:4317",
+    )
 
     # Create a FastAPI app
     uvicorn.run(app, host="0.0.0.0", port=8000)

examples/otel-collector/myapp.py

@@ -319,58 +319,34 @@ def wrapper(*args, **kwargs):
 
 
 def observe(name):
-	"""
-	A decorator factory that instruments a function for observability using opentelemetry tracing.
-	Args:
-		name (str): The name of the span to be created for tracing.
-	Returns:
-		Callable: A decorator that wraps the target function, extracting opentelemetry tracing context
-		from the incoming request (if available), and starts a new tracing span using
-		the provided name. If no context is found, a new span is started without context.
-	Behavior:
-		- Adds session ID to span attributes if available in kwargs.
-		- Attempts to extract a tracing context from the 'request' argument or from positional arguments.
-		- Starts a tracing span with the extracted context (if present) or as a new trace.
-		- Logs debug information about the tracing context and span creation.
-		- Supports usage both within and outside of HTTP request contexts.
-	Example:
-		@observe("my_function_span")
-		def my_function(request: Request, ...):
-			...
-	"""
 	def decorator(func):
-		decorated_func = agent(name)(func)
-		logger.debug("Decorating function with observe:", name)
-
+		print("Decorating:", func.__name__)
 		@functools.wraps(func)
 		def wrapper(*args, **kwargs):
-			add_session_id_to_span_attributes(**kwargs)
-			request: Request = kwargs.get("request")
+			print(f"[observe] wrapper called for {func.__name__}")
+			request = kwargs.get("request")
+			print(f"request: {request}")
+			print(f"kwargs: {kwargs}")
 			if request is None:
 				for arg in args:
 					if isinstance(arg, Request):
 						request = arg
 						break
 
-			# Extract context from request if available
-			ctx = extract(request.headers) if request else None
-
-			if ctx:
-				logger.debug(f"Starting span with context: {ctx}")
-				# Start span with context
-				with tracer.start_as_current_span(name, context=ctx, kind=SpanKind.SERVER):
-					return decorated_func(*args, **kwargs)
+			if request:
+				print(f"[observe] Request headers: {request.headers}")
+				ctx = extract(request.headers) if request else None
+				if ctx:
+					print(f"[observe] Extracted context: {ctx}")
+					with tracer.start_as_current_span(name, context=ctx, kind=SpanKind.SERVER):
+						return func(*args, **kwargs)
+				else:
+					print("[observe] No context found in request headers.")
 			else:
-				logger.debug("No context found, starting span without context.")
-				
-				# Start span without context
-				# This is useful for cases where we want to start a span without any specific context
-				# e.g., when the function is called outside of an HTTP request context
-				# or when we want to create a fresh new trace or context is not properly propagated.
-				return decorated_func(*args, **kwargs)
+				print("[observe] No request found.")
 
+			return func(*args, **kwargs)
 		return wrapper
-
 	return decorator
 
 class CustomFileSpanExporter(SpanExporter):