Skip to content

fix(weave): ignore unfinished calls in latency stats #7719

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
gtarpenning wants to merge 1 commit into
base: master
Choose a base branch
from
+77 −12
Open

fix(weave): ignore unfinished calls in latency stats #7719

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 7 additions & 7 deletions tests/trace_server/query_builder/test_call_metrics_query_builder.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,7 @@ def test_latency_basic():
count() AS count
FROM
(SELECT toStartOfInterval(sortable_datetime, INTERVAL 3600 SECOND, {pb_3:String}) AS bucket,
dateDiff('millisecond', started_at, ended_at) AS m_latency_ms
if(ended_at >= started_at, dateDiff('millisecond', started_at, ended_at), NULL) AS m_latency_ms
FROM
(SELECT anyIf(cm.sortable_datetime, cm.sortable_datetime IS NOT NULL) AS sortable_datetime,
anyIf(cm.started_at, cm.started_at IS NOT NULL) AS started_at,
Expand Down Expand Up @@ -192,7 +192,7 @@ def test_all_aggregation_types():
count() AS count
FROM
(SELECT toStartOfInterval(sortable_datetime, INTERVAL 300 SECOND, {pb_3:String}) AS bucket,
dateDiff('millisecond', started_at, ended_at) AS m_latency_ms
if(ended_at >= started_at, dateDiff('millisecond', started_at, ended_at), NULL) AS m_latency_ms
FROM
(SELECT anyIf(cm.sortable_datetime, cm.sortable_datetime IS NOT NULL) AS sortable_datetime,
anyIf(cm.started_at, cm.started_at IS NOT NULL) AS started_at,
Expand Down Expand Up @@ -273,7 +273,7 @@ def test_latency_percentiles():
count() AS count
FROM
(SELECT toStartOfInterval(sortable_datetime, INTERVAL 300 SECOND, {pb_3:String}) AS bucket,
dateDiff('millisecond', started_at, ended_at) AS m_latency_ms
if(ended_at >= started_at, dateDiff('millisecond', started_at, ended_at), NULL) AS m_latency_ms
FROM
(SELECT anyIf(cm.sortable_datetime, cm.sortable_datetime IS NOT NULL) AS sortable_datetime,
anyIf(cm.started_at, cm.started_at IS NOT NULL) AS started_at,
Expand Down Expand Up @@ -407,7 +407,7 @@ def test_trace_roots_only_filter():
count() AS count
FROM
(SELECT toStartOfInterval(sortable_datetime, INTERVAL 3600 SECOND, {pb_3:String}) AS bucket,
dateDiff('millisecond', started_at, ended_at) AS m_latency_ms
if(ended_at >= started_at, dateDiff('millisecond', started_at, ended_at), NULL) AS m_latency_ms
FROM
(SELECT anyIf(cm.sortable_datetime, cm.sortable_datetime IS NOT NULL) AS sortable_datetime,
anyIf(cm.started_at, cm.started_at IS NOT NULL) AS started_at,
Expand Down Expand Up @@ -544,7 +544,7 @@ def test_combined_metrics():
count() AS count
FROM
(SELECT toStartOfInterval(sortable_datetime, INTERVAL 3600 SECOND, {pb_3:String}) AS bucket,
dateDiff('millisecond', started_at, ended_at) AS m_latency_ms,
if(ended_at >= started_at, dateDiff('millisecond', started_at, ended_at), NULL) AS m_latency_ms,
1 AS m_call_count,
if(
exception IS NOT NULL, 1, 0) AS m_error_count
Expand Down Expand Up @@ -752,7 +752,7 @@ def test_calls_complete_latency_basic():
count() AS count
FROM
(SELECT toStartOfInterval(started_at, INTERVAL 3600 SECOND, {pb_3:String}) AS bucket,
dateDiff('millisecond', started_at, ended_at) AS m_latency_ms
if(ended_at >= started_at, dateDiff('millisecond', started_at, ended_at), NULL) AS m_latency_ms
FROM
(SELECT cm.started_at AS started_at,
cm.started_at AS started_at,
Expand Down Expand Up @@ -887,7 +887,7 @@ def test_calls_complete_trace_roots_only_filter():
count() AS count
FROM
(SELECT toStartOfInterval(started_at, INTERVAL 3600 SECOND, {pb_3:String}) AS bucket,
dateDiff('millisecond', started_at, ended_at) AS m_latency_ms
if(ended_at >= started_at, dateDiff('millisecond', started_at, ended_at), NULL) AS m_latency_ms
FROM
(SELECT cm.started_at AS started_at,
cm.started_at AS started_at,
Expand Down
70 changes: 66 additions & 4 deletions tests/trace_server/test_calls_complete.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2448,7 +2448,9 @@ def test_call_stats_with_calls_complete(trace_server, clickhouse_trace_server):
model_name = "gpt-4o-complete-test"

now = datetime.datetime.now(datetime.timezone.utc)
start_time = now - datetime.timedelta(minutes=30)
start_time = (now - datetime.timedelta(hours=2)).replace(
minute=10, second=0, microsecond=0
)

# Insert calls directly into calls_complete with usage data
call_id_1 = str(uuid.uuid4())
Expand Down Expand Up @@ -2477,7 +2479,7 @@ def test_call_stats_with_calls_complete(trace_server, clickhouse_trace_server):
'{call_id_1}',
'test_op',
'{start_time.strftime("%Y-%m-%d %H:%M:%S")}',
'{(start_time + datetime.timedelta(milliseconds=100)).strftime("%Y-%m-%d %H:%M:%S")}',
'{(start_time + datetime.timedelta(seconds=1)).strftime("%Y-%m-%d %H:%M:%S")}',
'{trace_id_1}',
'',
'{{}}',
Expand Down Expand Up @@ -2515,7 +2517,7 @@ def test_call_stats_with_calls_complete(trace_server, clickhouse_trace_server):
'{call_id_2}',
'test_op',
'{(start_time + datetime.timedelta(minutes=1)).strftime("%Y-%m-%d %H:%M:%S")}',
'{(start_time + datetime.timedelta(minutes=1, milliseconds=200)).strftime("%Y-%m-%d %H:%M:%S")}',
'{(start_time + datetime.timedelta(minutes=1, seconds=3)).strftime("%Y-%m-%d %H:%M:%S")}',
'{trace_id_2}',
'',
'{{}}',
Expand All @@ -2527,6 +2529,51 @@ def test_call_stats_with_calls_complete(trace_server, clickhouse_trace_server):
"""
)

# Exercise the epoch sentinel and a valid zero-duration call in one bucket.
clickhouse_trace_server.ch_client.command(
f"""
INSERT INTO calls_complete (
project_id,
id,
op_name,
started_at,
ended_at,
trace_id,
parent_id,
attributes_dump,
inputs_dump,
output_dump,
summary_dump
)
VALUES (
'{internal_project_id}',
'{uuid.uuid4()}',
'test_op',
'{(start_time + datetime.timedelta(minutes=2)).strftime("%Y-%m-%d %H:%M:%S")}',
'{SENTINEL_EPOCH.strftime("%Y-%m-%d %H:%M:%S")}',
'{uuid.uuid4()}',
'',
'{{}}',
'{{}}',
'null',
'{{}}'
),
(
'{internal_project_id}',
'{uuid.uuid4()}',
'test_op',
'{(start_time + datetime.timedelta(minutes=3)).strftime("%Y-%m-%d %H:%M:%S")}',
'{(start_time + datetime.timedelta(minutes=3)).strftime("%Y-%m-%d %H:%M:%S")}',
'{uuid.uuid4()}',
'',
'{{}}',
'{{}}',
'null',
'{{}}'
)
"""
)

# Verify we're reading from calls_complete
read_table = clickhouse_trace_server.table_routing_resolver.resolve_read_table(
internal_project_id,
Expand Down Expand Up @@ -2556,6 +2603,14 @@ def test_call_stats_with_calls_complete(trace_server, clickhouse_trace_server):
metric="error_count",
aggregations=[tsi.AggregationType.SUM],
),
tsi.CallMetricSpec(
metric="latency_ms",
aggregations=[
tsi.AggregationType.AVG,
tsi.AggregationType.MAX,
],
percentiles=[50, 95, 99],
),
],
)
)
Expand All @@ -2572,9 +2627,16 @@ def test_call_stats_with_calls_complete(trace_server, clickhouse_trace_server):
assert len(result.call_buckets) > 0, "Expected call buckets"
total_call_count = sum(b.get("sum_call_count", 0) for b in result.call_buckets)
total_error_count = sum(b.get("sum_error_count", 0) for b in result.call_buckets)
assert total_call_count == 2, f"Expected 2 calls, got {total_call_count}"
assert total_call_count == 4, f"Expected 4 calls, got {total_call_count}"
assert total_error_count == 1, f"Expected 1 error, got {total_error_count}"

latency_bucket = next(b for b in result.call_buckets if b.get("count") == 4)
assert latency_bucket["avg_latency_ms"] == pytest.approx(4000 / 3)
assert latency_bucket["max_latency_ms"] == 3000
assert latency_bucket["p50_latency_ms"] >= 0
assert latency_bucket["p95_latency_ms"] >= 0
assert latency_bucket["p99_latency_ms"] >= 0


def test_calls_complete_converts_wb_metadata(trace_server, clickhouse_trace_server):
"""Test that calls_complete converts wb_run_id and wb_user_id through the external adapter.
Expand Down
5 changes: 4 additions & 1 deletion weave/trace_server/calls_query_builder/call_metrics_query_builder.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,7 +59,10 @@ def _get_call_metric_extraction_sql(
SQL expression that extracts the metric value.
"""
if metric == "latency_ms":
return "dateDiff('millisecond', started_at, ended_at)"
return (
"if(ended_at >= started_at, "
"dateDiff('millisecond', started_at, ended_at), NULL)"
)
elif metric == "call_count":
return "1"
elif metric == "error_count":
Expand Down
Loading