Why: a crash after a tool was about to run left the call unpaired, and workspace writes could not be rewound. Evidence: GOWORK=off go test ./runtime/ ./runtime/llm/ ./runtime/checkpoint/ ./runtime/telemetry/ ./runtime/tools/ ./runtime/session/ ./adapters/native/ Impact: pending calls rerun only when both the stored and current replay declarations are safe; otherwise the result is an error. Empty fallback lists and a nil checkpoint log keep the previous behavior. Refs: feature/durable-rewind Co-authored-by: Cursor <cursoragent@cursor.com>
113 lines
2.8 KiB
Go
113 lines
2.8 KiB
Go
// Package telemetry 记录 GenAI span。默认不采集提示词正文。
|
|
package telemetry
|
|
|
|
import (
|
|
"bytes"
|
|
"crypto/rand"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"net/http"
|
|
"os"
|
|
"strings"
|
|
"time"
|
|
|
|
"logging-system/logsys"
|
|
)
|
|
|
|
// Span 是一次 invoke_agent、chat 或 execute_tool。
|
|
type Span struct {
|
|
Name string
|
|
TraceID string
|
|
SpanID string
|
|
Start time.Time
|
|
End time.Time
|
|
Attrs map[string]string
|
|
}
|
|
|
|
// StartFunc 返回循环使用的追踪函数。设置了 OTEL_EXPORTER_OTLP_ENDPOINT 时,
|
|
// 结束 span 后以 OTLP/HTTP JSON 上报,不引入新的依赖。
|
|
func StartFunc(log *logsys.Logger) func(name string, attrs map[string]string) func() {
|
|
endpoint := strings.TrimRight(os.Getenv("OTEL_EXPORTER_OTLP_ENDPOINT"), "/")
|
|
return func(name string, attrs map[string]string) func() {
|
|
sp := Span{
|
|
Name: name, TraceID: newID(16), SpanID: newID(8),
|
|
Start: time.Now(), Attrs: publicAttrs(attrs),
|
|
}
|
|
return func() {
|
|
sp.End = time.Now()
|
|
if log != nil {
|
|
kvs := []any{"span", sp.Name, "trace_id", sp.TraceID, "span_id", sp.SpanID}
|
|
for k, v := range sp.Attrs {
|
|
kvs = append(kvs, k, v)
|
|
}
|
|
log.Info("genai span", kvs...)
|
|
}
|
|
if endpoint != "" {
|
|
_ = ExportOTLP(endpoint, sp)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// publicAttrs 丢掉提示词正文。调用方即使传入也不会进入 span。
|
|
func publicAttrs(in map[string]string) map[string]string {
|
|
out := map[string]string{}
|
|
for k, v := range in {
|
|
if k == "prompt" || k == "messages" || k == "content" {
|
|
continue
|
|
}
|
|
out[k] = v
|
|
}
|
|
return out
|
|
}
|
|
|
|
// ExportOTLP 把一个 span POST 到端点的 /v1/traces。
|
|
func ExportOTLP(endpoint string, sp Span) error {
|
|
endpoint = strings.TrimRight(endpoint, "/")
|
|
if !strings.HasSuffix(endpoint, "/v1/traces") {
|
|
endpoint += "/v1/traces"
|
|
}
|
|
body, err := json.Marshal(otlpBody(sp))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
req, err := http.NewRequest(http.MethodPost, endpoint, bytes.NewReader(body))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
req.Header.Set("Content-Type", "application/json")
|
|
resp, err := http.DefaultClient.Do(req)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer resp.Body.Close()
|
|
return nil
|
|
}
|
|
|
|
func otlpBody(sp Span) map[string]any {
|
|
var attrs []map[string]any
|
|
for k, v := range publicAttrs(sp.Attrs) {
|
|
attrs = append(attrs, map[string]any{"key": k, "value": map[string]any{"stringValue": v}})
|
|
}
|
|
return map[string]any{
|
|
"resourceSpans": []any{map[string]any{
|
|
"scopeSpans": []any{map[string]any{
|
|
"spans": []any{map[string]any{
|
|
"traceId": sp.TraceID,
|
|
"spanId": sp.SpanID,
|
|
"name": sp.Name,
|
|
"startTimeUnixNano": sp.Start.UnixNano(),
|
|
"endTimeUnixNano": sp.End.UnixNano(),
|
|
"attributes": attrs,
|
|
}},
|
|
}},
|
|
}},
|
|
}
|
|
}
|
|
|
|
func newID(n int) string {
|
|
b := make([]byte, n)
|
|
_, _ = rand.Read(b)
|
|
return hex.EncodeToString(b)
|
|
}
|