fix(log): 精简请求完成日志字段
This commit is contained in:
5 files changed
+169
-39
No files matched your search
@@ -0,0 +1,83 @@
|
||||
package utils
|
||||
|
||||
import "strconv"
|
||||
|
||||
const nanoUSDPerUSD int64 = 1_000_000_000
|
||||
|
||||
// FormatDurationMS formats a millisecond duration for human-readable runtime logs.
|
||||
func FormatDurationMS(milliseconds int64) string {
|
||||
if milliseconds < 0 {
|
||||
return "?"
|
||||
}
|
||||
if milliseconds < 1_000 {
|
||||
return strconv.FormatInt(milliseconds, 10) + "ms"
|
||||
}
|
||||
if milliseconds < 60_000 {
|
||||
return formatSubminuteDuration(milliseconds)
|
||||
}
|
||||
|
||||
totalSeconds := milliseconds / 1_000
|
||||
if milliseconds%1_000 >= 500 {
|
||||
totalSeconds++
|
||||
}
|
||||
hours := totalSeconds / 3_600
|
||||
minutes := (totalSeconds % 3_600) / 60
|
||||
seconds := totalSeconds % 60
|
||||
if hours > 0 {
|
||||
return strconv.FormatInt(hours, 10) + "h" + twoDigits(minutes) + "m" + twoDigits(seconds) + "s"
|
||||
}
|
||||
return strconv.FormatInt(minutes, 10) + "m" + twoDigits(seconds) + "s"
|
||||
}
|
||||
|
||||
func formatSubminuteDuration(milliseconds int64) string {
|
||||
if milliseconds < 10_000 {
|
||||
hundredths := (milliseconds + 5) / 10
|
||||
whole := hundredths / 100
|
||||
fraction := hundredths % 100
|
||||
return formatDecimalSeconds(whole, fraction, 2)
|
||||
}
|
||||
|
||||
tenths := (milliseconds + 50) / 100
|
||||
whole := tenths / 10
|
||||
fraction := tenths % 10
|
||||
return formatDecimalSeconds(whole, fraction, 1)
|
||||
}
|
||||
|
||||
func formatDecimalSeconds(whole, fraction int64, fractionDigits int) string {
|
||||
if fraction == 0 {
|
||||
return strconv.FormatInt(whole, 10) + "s"
|
||||
}
|
||||
fractionText := strconv.FormatInt(fraction, 10)
|
||||
for len(fractionText) < fractionDigits {
|
||||
fractionText = "0" + fractionText
|
||||
}
|
||||
return strconv.FormatInt(whole, 10) + "." + fractionText + "s"
|
||||
}
|
||||
|
||||
func twoDigits(value int64) string {
|
||||
if value < 10 {
|
||||
return "0" + strconv.FormatInt(value, 10)
|
||||
}
|
||||
return strconv.FormatInt(value, 10)
|
||||
}
|
||||
|
||||
// FormatNanoUSD converts an integer nano-USD amount to a compact decimal string.
|
||||
func FormatNanoUSD(nanoUSD int64) string {
|
||||
if nanoUSD < 0 {
|
||||
return "?"
|
||||
}
|
||||
dollars := nanoUSD / nanoUSDPerUSD
|
||||
nanos := nanoUSD % nanoUSDPerUSD
|
||||
if nanos == 0 {
|
||||
return strconv.FormatInt(dollars, 10)
|
||||
}
|
||||
|
||||
fraction := strconv.FormatInt(nanos, 10)
|
||||
for len(fraction) < 9 {
|
||||
fraction = "0" + fraction
|
||||
}
|
||||
for len(fraction) > 0 && fraction[len(fraction)-1] == '0' {
|
||||
fraction = fraction[:len(fraction)-1]
|
||||
}
|
||||
return strconv.FormatInt(dollars, 10) + "." + fraction
|
||||
}
|
||||
@@ -0,0 +1,46 @@
|
||||
package utils
|
||||
|
||||
import "testing"
|
||||
|
||||
func TestFormatDurationMS(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
ms int64
|
||||
want string
|
||||
}{
|
||||
{name: "zero", ms: 0, want: "0ms"},
|
||||
{name: "milliseconds", ms: 999, want: "999ms"},
|
||||
{name: "short seconds", ms: 1_234, want: "1.23s"},
|
||||
{name: "long seconds", ms: 14_109, want: "14.1s"},
|
||||
{name: "minute", ms: 61_000, want: "1m01s"},
|
||||
{name: "hour", ms: 3_723_000, want: "1h02m03s"},
|
||||
{name: "negative", ms: -1, want: "?"},
|
||||
}
|
||||
for _, test := range tests {
|
||||
t.Run(test.name, func(t *testing.T) {
|
||||
if got := FormatDurationMS(test.ms); got != test.want {
|
||||
t.Fatalf("FormatDurationMS(%d) = %q, want %q", test.ms, got, test.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestFormatNanoUSD(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
nanos int64
|
||||
want string
|
||||
}{
|
||||
{name: "zero", nanos: 0, want: "0"},
|
||||
{name: "fraction", nanos: 55_000, want: "0.000055"},
|
||||
{name: "one dollar", nanos: 1_000_000_000, want: "1"},
|
||||
{name: "precise", nanos: 1_234_567_890, want: "1.23456789"},
|
||||
}
|
||||
for _, test := range tests {
|
||||
t.Run(test.name, func(t *testing.T) {
|
||||
if got := FormatNanoUSD(test.nanos); got != test.want {
|
||||
t.Fatalf("FormatNanoUSD(%d) = %q, want %q", test.nanos, got, test.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -36,41 +36,41 @@ func projectProcessLog(
|
||||
}
|
||||
|
||||
fields := logrus.Fields{
|
||||
"event": "data_plane_request_completed",
|
||||
"request_id": row.ID,
|
||||
"status": row.Status,
|
||||
"protocol": row.Protocol,
|
||||
"access_key_id": row.AccessKeyID,
|
||||
"duration_ms": row.DurationMs,
|
||||
"event": "data_plane_request_completed",
|
||||
"req_id": row.ID,
|
||||
"status": row.Status,
|
||||
"proto": row.Protocol,
|
||||
"ak_id": row.AccessKeyID,
|
||||
"duration": utils.FormatDurationMS(row.DurationMs),
|
||||
}
|
||||
if row.StatusCode != http.StatusOK {
|
||||
fields["status_code"] = row.StatusCode
|
||||
fields["http"] = row.StatusCode
|
||||
}
|
||||
if row.ClientModel != "" {
|
||||
fields["client_model"] = row.ClientModel
|
||||
fields["model"] = row.ClientModel
|
||||
}
|
||||
if row.UpstreamModel != "" && row.UpstreamModel != row.ClientModel {
|
||||
fields["upstream_model"] = row.UpstreamModel
|
||||
fields["up_model"] = row.UpstreamModel
|
||||
}
|
||||
if groupID > 0 {
|
||||
fields["group_id"] = groupID
|
||||
fields["group"] = groupID
|
||||
}
|
||||
if keyID > 0 {
|
||||
fields["key_id"] = keyID
|
||||
fields["key"] = keyID
|
||||
}
|
||||
if attemptCount := len(event.Attempts); attemptCount != 1 {
|
||||
fields["attempt_count"] = attemptCount
|
||||
fields["attempts"] = attemptCount
|
||||
}
|
||||
if inputTokens, ok := processInputTokens(row); ok && inputTokens > 0 {
|
||||
fields["input_tokens"] = inputTokens
|
||||
fields["in_tokens"] = inputTokens
|
||||
}
|
||||
if row.OutputTokens > 0 {
|
||||
fields["output_tokens"] = row.OutputTokens
|
||||
fields["out_tokens"] = row.OutputTokens
|
||||
}
|
||||
if row.UsageState != "" &&
|
||||
row.UsageState != string(usage.StateComplete) &&
|
||||
row.UsageState != string(usage.StateNotApplicable) {
|
||||
fields["usage_state"] = row.UsageState
|
||||
fields["usage"] = row.UsageState
|
||||
}
|
||||
if row.CostState != "" &&
|
||||
row.CostState != string(pricing.CostStatePriced) &&
|
||||
@@ -78,18 +78,18 @@ func projectProcessLog(
|
||||
fields["cost_state"] = row.CostState
|
||||
}
|
||||
if row.CostState == string(pricing.CostStatePriced) {
|
||||
fields["estimated_cost_nano_usd"] = row.EstimatedCostNanoUSD
|
||||
fields["cost_usd"] = utils.FormatNanoUSD(row.EstimatedCostNanoUSD)
|
||||
}
|
||||
if event.Status != telemetry.RequestStatusSuccess {
|
||||
if row.ErrorCode != "" {
|
||||
fields["error_code"] = row.ErrorCode
|
||||
fields["err"] = row.ErrorCode
|
||||
}
|
||||
if summary := sanitizeSummaryLimit(
|
||||
redactor,
|
||||
event.ErrorSummary,
|
||||
maxProcessSummaryBytes,
|
||||
); summary != "" {
|
||||
fields["error_summary"] = summary
|
||||
fields["err_msg"] = summary
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -52,6 +52,7 @@ func TestProjectProcessLogUsesFrozenPricingAndInputOutputTotals(t *testing.T) {
|
||||
event.StatusCode = http.StatusBadGateway
|
||||
event.ErrorCode = "upstream_error"
|
||||
event.ErrorSummary = "provider failed"
|
||||
event.DurationMs = 58_773
|
||||
event.Usage.AttemptSequence = 1
|
||||
event.Usage.KeyID = 8
|
||||
event.Usage.Result = usage.Result{
|
||||
@@ -72,14 +73,14 @@ func TestProjectProcessLogUsesFrozenPricingAndInputOutputTotals(t *testing.T) {
|
||||
t.Fatalf("projection = %t/%s", ok, level)
|
||||
}
|
||||
want := logrus.Fields{
|
||||
"event": "data_plane_request_completed", "request_id": event.RequestID,
|
||||
"status": "error", "status_code": http.StatusBadGateway,
|
||||
"protocol": string(protocol.OpenAICompletions), "access_key_id": uint(42),
|
||||
"client_model": "client-model", "upstream_model": "upstream-model",
|
||||
"group_id": uint(7), "key_id": uint(8), "duration_ms": int64(25),
|
||||
"input_tokens": int64(15), "output_tokens": int64(6),
|
||||
"usage_state": "partial", "estimated_cost_nano_usd": int64(55_000),
|
||||
"error_code": "upstream_error", "error_summary": "provider failed",
|
||||
"event": "data_plane_request_completed", "req_id": event.RequestID,
|
||||
"status": "error", "http": http.StatusBadGateway,
|
||||
"proto": string(protocol.OpenAICompletions), "ak_id": uint(42),
|
||||
"model": "client-model", "up_model": "upstream-model",
|
||||
"group": uint(7), "key": uint(8), "duration": "58.8s",
|
||||
"in_tokens": int64(15), "out_tokens": int64(6),
|
||||
"usage": "partial", "cost_usd": "0.000055",
|
||||
"err": "upstream_error", "err_msg": "provider failed",
|
||||
}
|
||||
assertProcessFields(t, fields, want)
|
||||
}
|
||||
@@ -105,7 +106,7 @@ func TestProjectProcessLogUsesStatusLevelAndConditionalDiagnostics(t *testing.T)
|
||||
if !ok || level != test.level {
|
||||
t.Fatalf("status %q projection = %t/%s", test.status, ok, level)
|
||||
}
|
||||
_, hasCode := fields["error_code"]
|
||||
_, hasCode := fields["err"]
|
||||
if hasCode != (test.status != telemetry.RequestStatusSuccess) {
|
||||
t.Fatalf("status %q error_code presence = %t", test.status, hasCode)
|
||||
}
|
||||
@@ -123,10 +124,10 @@ func TestProjectProcessLogOmitsDefaultAndZeroValueFields(t *testing.T) {
|
||||
t.Fatalf("projection = %t/%s", ok, level)
|
||||
}
|
||||
want := logrus.Fields{
|
||||
"event": "data_plane_request_completed", "request_id": event.RequestID,
|
||||
"status": "success", "protocol": "anthropic", "access_key_id": uint(42),
|
||||
"client_model": "client-model", "group_id": uint(7), "key_id": uint(8),
|
||||
"duration_ms": int64(25),
|
||||
"event": "data_plane_request_completed", "req_id": event.RequestID,
|
||||
"status": "success", "proto": "anthropic", "ak_id": uint(42),
|
||||
"model": "client-model", "group": uint(7), "key": uint(8),
|
||||
"duration": "25ms",
|
||||
}
|
||||
assertProcessFields(t, fields, want)
|
||||
}
|
||||
@@ -143,8 +144,8 @@ func TestProjectProcessLogPreservesNoCandidateDiagnostics(t *testing.T) {
|
||||
},
|
||||
}
|
||||
level, fields, ok := projectProcessLog(redact.New(), event)
|
||||
if !ok || level != logrus.WarnLevel || fields["attempt_count"] != 0 ||
|
||||
fields["error_code"] != "no_available_candidate" {
|
||||
if !ok || level != logrus.WarnLevel || fields["attempts"] != 0 ||
|
||||
fields["err"] != "no_available_candidate" {
|
||||
t.Fatalf("no-candidate fields = %#v", fields)
|
||||
}
|
||||
}
|
||||
@@ -174,7 +175,7 @@ func TestProjectProcessLogOmitsMissingAttribution(t *testing.T) {
|
||||
if !ok {
|
||||
t.Fatal("projection skipped")
|
||||
}
|
||||
for _, name := range []string{"group_id", "key_id", "group_name", "upstream_model"} {
|
||||
for _, name := range []string{"group", "key", "group_name", "up_model"} {
|
||||
if value, exists := fields[name]; exists {
|
||||
t.Fatalf("%s = %#v, want omitted", name, value)
|
||||
}
|
||||
|
||||
@@ -42,7 +42,7 @@ func TestServiceEmitLogsAcceptedCompletionExactlyOnce(t *testing.T) {
|
||||
}
|
||||
entries := processLogEntries(t, output.Bytes())
|
||||
if len(entries) != 1 ||
|
||||
entries[0]["request_id"] != "accepted" ||
|
||||
entries[0]["req_id"] != "accepted" ||
|
||||
entries[0]["msg"] != "[DATA] Request completed" {
|
||||
t.Fatalf("completion entries = %#v, want accepted event", entries)
|
||||
}
|
||||
@@ -155,7 +155,7 @@ func TestServiceEmitCompletionSurvivesLaterWriteFailure(t *testing.T) {
|
||||
t.Fatalf("WriteFailureTotal = %d, want 1", got)
|
||||
}
|
||||
entries := processLogEntries(t, output.Bytes())
|
||||
if len(entries) != 1 || entries[0]["request_id"] != "persist-failed" {
|
||||
if len(entries) != 1 || entries[0]["req_id"] != "persist-failed" {
|
||||
t.Fatalf("completion entries = %#v, want persisted-failed event", entries)
|
||||
}
|
||||
}
|
||||
@@ -306,7 +306,7 @@ func TestServiceDropsNewEventAtExactQueueCapacity(t *testing.T) {
|
||||
t.Fatalf("stats at capacity = %+v", stats)
|
||||
}
|
||||
entries := processLogEntries(t, output.Bytes())
|
||||
if len(entries) != 1 || entries[0]["request_id"] != "must-drop" {
|
||||
if len(entries) != 1 || entries[0]["req_id"] != "must-drop" {
|
||||
t.Fatalf("queue-full completion entries = %#v", entries)
|
||||
}
|
||||
|
||||
@@ -488,8 +488,8 @@ func TestEmitPersistsGatewayFrozenQuoteWithoutRepricing(t *testing.T) {
|
||||
timer := receiveValue(t, timers.created)
|
||||
entries := processLogEntries(t, output.Bytes())
|
||||
if len(entries) != 1 ||
|
||||
entries[0]["request_id"] != "snapshot-a" ||
|
||||
entries[0]["estimated_cost_nano_usd"] != float64(1_000_000_000) {
|
||||
entries[0]["req_id"] != "snapshot-a" ||
|
||||
entries[0]["cost_usd"] != "1" {
|
||||
t.Fatalf("snapshot-a completion entries = %#v", entries)
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user