fix: update telemetry schema and fix telemtry calls (#2708)

This commit is contained in:
Gabriel Luiz Freitas Almeida 2024-07-15 16:16:23 -03:00 • committed by GitHub
commit 6e006354c3
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 11 additions and 8 deletions

View file

@ -173,6 +173,7 @@ async def build_vertex(
next_runnable_vertices = [] next_runnable_vertices = []
top_level_vertices = [] top_level_vertices = []
start_time = time.perf_counter() start_time = time.perf_counter()
error_message = None
try: try:
cache = await chat_service.get_cache(flow_id_str) cache = await chat_service.get_cache(flow_id_str)
if not cache: if not cache:
@ -215,6 +216,7 @@ async def build_vertex(
params = format_exception_message(exc) params = format_exception_message(exc)
message = {"errorMessage": params, "stackTrace": tb} message = {"errorMessage": params, "stackTrace": tb}
valid = False valid = False
error_message = params
output_label = vertex.outputs[0]["name"] if vertex.outputs else "output" output_label = vertex.outputs[0]["name"] if vertex.outputs else "output"
outputs = {output_label: OutputLog(message=message, type="error")} outputs = {output_label: OutputLog(message=message, type="error")}
result_data_response = ResultDataResponse(results={}, outputs=outputs) result_data_response = ResultDataResponse(results={}, outputs=outputs)
@ -231,7 +233,7 @@ async def build_vertex(
background_tasks.add_task( background_tasks.add_task(
log_vertex_build, log_vertex_build,
flow_id=flow_id_str, flow_id=flow_id_str,
vertex_id=vertex_id, vertex_id=vertex_id.split("-")[0],
valid=valid, valid=valid,
params=params, params=params,
data=result_data_response, data=result_data_response,
@ -256,7 +258,7 @@ async def build_vertex(
if graph.stop_vertex and graph.stop_vertex in next_runnable_vertices: if graph.stop_vertex and graph.stop_vertex in next_runnable_vertices:
next_runnable_vertices = [graph.stop_vertex] next_runnable_vertices = [graph.stop_vertex]
if graph.run_manager.all_predecessors_are_fulfilled() and not next_runnable_vertices: if not graph.run_manager.vertices_to_run and not next_runnable_vertices:
background_tasks.add_task(graph.end_all_traces) background_tasks.add_task(graph.end_all_traces)
build_response = VertexBuildResponse( build_response = VertexBuildResponse(
@ -271,10 +273,10 @@ async def build_vertex(
background_tasks.add_task( background_tasks.add_task(
telemetry_service.log_package_component, telemetry_service.log_package_component,
ComponentPayload( ComponentPayload(
componentName=vertex_id, componentName=vertex_id.split("-")[0],
componentSeconds=int(time.perf_counter() - start_time), componentSeconds=int(time.perf_counter() - start_time),
componentSuccess=valid, componentSuccess=valid,
componentErrorMessage=params, componentErrorMessage=error_message,
), ),
) )
return build_response return build_response
@ -282,7 +284,7 @@ async def build_vertex(
background_tasks.add_task( background_tasks.add_task(
telemetry_service.log_package_component, telemetry_service.log_package_component,
ComponentPayload( ComponentPayload(
componentName=vertex_id, componentName=vertex_id.split("-")[0],
componentSeconds=int(time.perf_counter() - start_time), componentSeconds=int(time.perf_counter() - start_time),
componentSuccess=False, componentSuccess=False,
componentErrorMessage=str(exc), componentErrorMessage=str(exc),

View file

@ -13,6 +13,7 @@ class ShutdownPayload(BaseModel):
class VersionPayload(BaseModel): class VersionPayload(BaseModel):
package: str
version: str version: str
platform: str platform: str
python: str python: str

View file

@ -8,9 +8,9 @@ from typing import TYPE_CHECKING
import httpx import httpx
from loguru import logger from loguru import logger
from pydantic import BaseModel from pydantic import BaseModel
from langflow.services.telemetry.opentelemetry import OpenTelemetry
from langflow.services.base import Service from langflow.services.base import Service
from langflow.services.telemetry.opentelemetry import OpenTelemetry
from langflow.services.telemetry.schema import ( from langflow.services.telemetry.schema import (
ComponentPayload, ComponentPayload,
PlaygroundPayload, PlaygroundPayload,
@ -34,7 +34,6 @@ class TelemetryService(Service):
self.telemetry_queue: asyncio.Queue = asyncio.Queue() self.telemetry_queue: asyncio.Queue = asyncio.Queue()
self.client = httpx.AsyncClient(timeout=10.0) # Set a reasonable timeout self.client = httpx.AsyncClient(timeout=10.0) # Set a reasonable timeout
self.running = False self.running = False
self.package = get_version_info()["package"]
self.ot = OpenTelemetry(prometheus_enabled=settings_service.settings.prometheus_enabled) self.ot = OpenTelemetry(prometheus_enabled=settings_service.settings.prometheus_enabled)
@ -58,7 +57,7 @@ class TelemetryService(Service):
logger.debug("Telemetry tracking is disabled.") logger.debug("Telemetry tracking is disabled.")
return return
url = f"{self.base_url}/{self.package.lower()}" url = f"{self.base_url}"
if path: if path:
url = f"{url}/{path}" url = f"{url}/{path}"
try: try:
@ -86,6 +85,7 @@ class TelemetryService(Service):
version_info = get_version_info() version_info = get_version_info()
architecture = platform.architecture()[0] architecture = platform.architecture()[0]
payload = VersionPayload( payload = VersionPayload(
package=version_info["package"].lower(),
version=version_info["version"], version=version_info["version"],
platform=platform.platform(), platform=platform.platform(),
python=python_version, python=python_version,