telemetry

package
v0.1.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 30, 2026 License: Apache-2.0 Imports: 42 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	DefaultCloudTraceQueueSize = 2048
	DefaultCloudTraceBatchSize = 512
	DefaultCloudTraceBatchWait = 5 * time.Second
	DefaultCloudExportTimeout  = 10 * time.Second
)
View Source 
const (
	DefaultCloudLogCapacity      = 1024
	DefaultCloudLogBatchSize     = 128
	DefaultCloudLogFlushInterval = 200 * time.Millisecond
)
View Source 
const (
	AttrSpeechID                   = "lk.speech_id"
	AttrAgentLabel                 = "lk.agent_label"
	AttrStartTime                  = "lk.start_time"
	AttrEndTime                    = "lk.end_time"
	AttrRetryCount                 = "lk.retry_count"
	AttrProviderRequestIDs         = "lk.provider_request_ids"
	AttrParticipantID              = "lk.participant_id"
	AttrParticipantIdentity        = "lk.pii.participant_identity"
	AttrParticipantKind            = "lk.participant_kind"
	AttrJobID                      = "lk.job_id"
	AttrAgentName                  = "lk.agent_name"
	AttrCloudAgentID               = "lk.cloud_agent_id"
	AttrDeploymentID               = "lk.deployment_id"
	AttrRoomName                   = "lk.pii.room_name"
	AttrSessionOptions             = "lk.session_options"
	AttrAgentTurnID                = "lk.generation_id"
	AttrAgentParentTurnID          = "lk.parent_generation_id"
	AttrUserInput                  = "lk.pii.user_input"
	AttrInstructions               = "lk.pii.instructions"
	AttrSpeechInterrupted          = "lk.interrupted"
	AttrChatContext                = "lk.pii.chat_ctx"
	AttrFunctionTools              = "lk.function_tools"
	AttrProviderTools              = "lk.provider_tools"
	AttrToolSets                   = "lk.tool_sets"
	AttrResponseText               = "lk.pii.response.text"
	AttrResponseFunctionCalls      = "lk.pii.response.function_calls"
	AttrResponseTTFT               = "lk.response.ttft"
	AttrFunctionToolID             = "lk.function_tool.id"
	AttrFunctionToolName           = "lk.function_tool.name"
	AttrFunctionToolArguments      = "lk.pii.function_tool.arguments"
	AttrFunctionToolIsError        = "lk.function_tool.is_error"
	AttrFunctionToolOutput         = "lk.pii.function_tool.output"
	AttrTTSInputText               = "lk.pii.input_text"
	AttrTTSStreaming               = "lk.tts.streaming"
	AttrTTSLabel                   = "lk.tts.label"
	AttrResponseTTFB               = "lk.response.ttfb"
	AttrEOUProbability             = "lk.eou.probability"
	AttrEOUUnlikelyThreshold       = "lk.eou.unlikely_threshold"
	AttrEOUDelay                   = "lk.eou.endpointing_delay"
	AttrEOULanguage                = "lk.eou.language"
	AttrEOUSource                  = "lk.eou.source"
	AttrEOUFromCache               = "lk.eou.from_cache"
	AttrEOUDetectionDelay          = "lk.eou.detection_delay"
	AttrUserTranscript             = "lk.pii.user_transcript"
	AttrTranscriptConfidence       = "lk.transcript_confidence"
	AttrTranscriptionDelay         = "lk.transcription_delay"
	AttrEndOfTurnDelay             = "lk.end_of_turn_delay"
	AttrAMDCategory                = "lk.amd.category"
	AttrAMDReason                  = "lk.amd.reason"
	AttrAMDIsMachine               = "lk.amd.is_machine"
	AttrAMDInterruptOnMachine      = "lk.amd.interrupt_on_machine"
	AttrAMDSpeechDuration          = "lk.amd.speech_duration"
	AttrAMDDelay                   = "lk.amd.delay"
	AttrAMDTranscript              = "lk.pii.amd.transcript"
	AttrIsInterruption             = "lk.is_interruption"
	AttrInterruptionProbability    = "lk.interruption.probability"
	AttrInterruptionTotalDuration  = "lk.interruption.total_duration"
	AttrInterruptionPredictionTime = "lk.interruption.prediction_duration"
	AttrInterruptionDetectionDelay = "lk.interruption.detection_delay"
	AttrLLMMetrics                 = "lk.llm_metrics"
	AttrTTSMetrics                 = "lk.tts_metrics"
	AttrRealtimeModelMetrics       = "lk.realtime_model_metrics"
	AttrEndToEndLatency            = "lk.e2e_latency"

	AttrGenAIOperationName         = "gen_ai.operation.name"
	AttrGenAIRequestModel          = "gen_ai.request.model"
	AttrGenAIProviderName          = "gen_ai.provider.name"
	AttrGenAIUsageInputTokens      = "gen_ai.usage.input_tokens"
	AttrGenAIUsageOutputTokens     = "gen_ai.usage.output_tokens"
	AttrGenAIUsageInputTextTokens  = "gen_ai.usage.input_text_tokens"
	AttrGenAIUsageInputAudioTokens = "gen_ai.usage.input_audio_tokens"
	AttrGenAIUsageInputCached      = "gen_ai.usage.input_cached_tokens"
	AttrGenAIUsageOutputTextTokens = "gen_ai.usage.output_text_tokens"
	AttrGenAIUsageOutputAudio      = "gen_ai.usage.output_audio_tokens"

	EventGenAISystemMessage    = "gen_ai.system.message"
	EventGenAIUserMessage      = "gen_ai.user.message"
	EventGenAIAssistantMessage = "gen_ai.assistant.message"
	EventGenAIToolMessage      = "gen_ai.tool.message"
	EventGenAIChoice           = "gen_ai.choice"

	AttrExceptionTrace   = "exception.stacktrace"
	AttrExceptionType    = "exception.type"
	AttrExceptionMessage = "exception.message"

	AttrLangfuseCompletionStartTime = "langfuse.observation.completion_start_time"
)

LiveKit attributes containing conversational/user data deliberately include a dot-delimited "pii" segment. LiveKit Cloud's collector uses that marker for recursive redaction. Never place such content in span or event names.

View Source 
const (
	InstrumentationName      = "livekit-agents"
	RedactedExceptionMessage = "exception details redacted"
)

Variables

View Source 
var DefaultUploadGate = NewUploadGate(nil)

Functions

func RecordException 

func RecordException(span trace.Span, err error, options RecordExceptionOptions)

func RecordRealtimeMetrics 

func RecordRealtimeMetrics(ctx context.Context, span trace.Span, value metrics.Realtime) error

func SetTracerProvider 

func SetTracerProvider(provider trace.TracerProvider) error

SetTracerProvider updates only the Agents framework tracer. Applications can separately call otel.SetTracerProvider when they want a process-wide change.

func SetTracerProviderWithOptions 

func SetTracerProviderWithOptions(provider trace.TracerProvider, options SetTracerProviderOptions) error

SetTracerProviderWithOptions updates the framework tracer and records the processor attachment seam used by SetupCloudTracer. It never changes the process-global OTel provider.

func StartSpan 

func StartSpan(ctx context.Context, options StartSpanOptions) (context.Context, trace.Span, error)

func Tracer 

func Tracer() trace.Tracer

Tracer returns the current framework tracer. The OpenTelemetry global provider is resolved lazily so importing telemetry has no SDK startup cost.

func TracerProvider 

func TracerProvider() trace.TracerProvider

func WithSpan 

func WithSpan(ctx context.Context, options StartSpanOptions, callback func(context.Context, trace.Span) error) error

WithSpan creates an active span in the supplied context, runs callback, and always ends the span. Callback errors are recorded without leaking them into span names or event names.

Types

type CloudLogHandler 

type CloudLogHandler struct {
	// contains filtered or unexported fields
}

CloudLogHandler is a bounded slog bridge to the standalone LiveKit Cloud OTLP log exporter. Low-severity logs drop rather than stalling audio when the queue is saturated; errors apply caller-context backpressure. Flush and Shutdown deterministically drain accepted records.

func NewCloudLogHandler 

func NewCloudLogHandler(options CloudLogHandlerOptions) (*CloudLogHandler, error)

func (*CloudLogHandler) Dropped 

func (h *CloudLogHandler) Dropped() uint64

func (*CloudLogHandler) Enabled 

func (h *CloudLogHandler) Enabled(ctx context.Context, level slog.Level) bool

func (*CloudLogHandler) Flush 

func (h *CloudLogHandler) Flush(ctx context.Context) error

func (*CloudLogHandler) Handle 

func (h *CloudLogHandler) Handle(ctx context.Context, record slog.Record) error

func (*CloudLogHandler) Shutdown 

func (h *CloudLogHandler) Shutdown(ctx context.Context) error

func (*CloudLogHandler) WithAttrs 

func (h *CloudLogHandler) WithAttrs(attrs []slog.Attr) slog.Handler

func (*CloudLogHandler) WithGroup 

func (h *CloudLogHandler) WithGroup(name string) slog.Handler

type CloudLogHandlerOptions 

type CloudLogHandlerOptions struct {
	Exporter      *SimpleOTLPHTTPLogExporter
	Next          slog.Handler
	Level         slog.Leveler
	Capacity      int
	BatchSize     int
	FlushInterval time.Duration
	StaticAttrs   map[string]any
}

type CloudSpanProcessorOptions 

type CloudSpanProcessorOptions struct {
	URL      string
	Headers  map[string]string
	Exporter sdktrace.SpanExporter
}

CloudSpanProcessorOptions is passed to a custom cloud processor factory. Exporter already enforces the recording-disabled upload gate.

type CloudTelemetry 

type CloudTelemetry struct {
	// contains filtered or unexported fields
}

CloudTelemetry owns the processors/exporters created for one setup call. Shutdown is idempotent and must be called at job/session teardown.

func SetupCloudTracer 

func SetupCloudTracer(ctx context.Context, options SetupCloudTracerOptions) (*CloudTelemetry, error)

SetupCloudTracer installs LiveKit Cloud OTLP/HTTP protobuf trace export and constructs the matching raw-OTLP log exporter. No work occurs at package import time; credentials, DNS, and goroutines are touched only here.

func (*CloudTelemetry) ForceFlush 

func (c *CloudTelemetry) ForceFlush(ctx context.Context) error

func (*CloudTelemetry) LogExporter 

func (c *CloudTelemetry) LogExporter() *SimpleOTLPHTTPLogExporter

func (*CloudTelemetry) Shutdown 

func (c *CloudTelemetry) Shutdown(ctx context.Context) error

type FanoutSpanProcessor 

type FanoutSpanProcessor struct {
	// contains filtered or unexported fields
}

FanoutSpanProcessor lets a provider attach processors after construction. Add and callbacks are concurrency-safe; callbacks run over a snapshot so a processor can add another processor without deadlocking.

func (*FanoutSpanProcessor) Add 

func (p *FanoutSpanProcessor) Add(processor sdktrace.SpanProcessor) error

func (*FanoutSpanProcessor) ForceFlush 

func (p *FanoutSpanProcessor) ForceFlush(ctx context.Context) error

func (*FanoutSpanProcessor) OnEnd 

func (p *FanoutSpanProcessor) OnEnd(span sdktrace.ReadOnlySpan)

func (*FanoutSpanProcessor) OnStart 

func (p *FanoutSpanProcessor) OnStart(ctx context.Context, span sdktrace.ReadWriteSpan)

func (*FanoutSpanProcessor) Shutdown 

func (p *FanoutSpanProcessor) Shutdown(ctx context.Context) error

type LogExportError 

type LogExportError struct {
	StatusCode int
	Status     string
	Body       string
}

func (*LogExportError) Error 

func (e *LogExportError) Error() string

type MetadataSpanProcessor 

type MetadataSpanProcessor struct {
	// contains filtered or unexported fields
}

MetadataSpanProcessor adds immutable session metadata synchronously when a span starts. The hot path is only one SetAttributes call.

func NewMetadataSpanProcessor 

func NewMetadataSpanProcessor(metadata map[string]any) (*MetadataSpanProcessor, error)

func (*MetadataSpanProcessor) ForceFlush 

func (*MetadataSpanProcessor) ForceFlush(context.Context) error

func (*MetadataSpanProcessor) OnEnd 

func (*MetadataSpanProcessor) OnEnd(sdktrace.ReadOnlySpan)

func (*MetadataSpanProcessor) OnStart 

func (p *MetadataSpanProcessor) OnStart(_ context.Context, span sdktrace.ReadWriteSpan)

func (*MetadataSpanProcessor) Shutdown 

func (*MetadataSpanProcessor) Shutdown(context.Context) error

type RecordExceptionOptions 

type RecordExceptionOptions struct {
	Redacted bool
}

type SessionReportUploader 

type SessionReportUploader struct {
	// contains filtered or unexported fields
}

func NewSessionReportUploader 

func NewSessionReportUploader(config SessionReportUploaderConfig) (*SessionReportUploader, error)

func (*SessionReportUploader) Upload 

func (u *SessionReportUploader) Upload(ctx context.Context, report *voice.SessionReport) error

type SessionReportUploaderConfig 

type SessionReportUploaderConfig struct {
	AgentName         string
	CloudHostname     string
	RecordingEndpoint string
	LogEndpoint       string
	APIKey            agents.SecretString
	APISecret         agents.SecretString
	HTTPClient        *http.Client
	UploadGate        *UploadGate
	Metadata          map[string]any
	Logger            *slog.Logger
	MaxAudioBytes     int64
	MaxErrorBodyBytes int64
	// Zero selects three retries; a negative value disables retries.
	MaxRetries int
	// Sleep is injectable for deterministic tests. Nil uses a context-aware timer.
	Sleep func(context.Context, time.Duration) error
}

type SessionUploadError 

type SessionUploadError struct {
	StatusCode int
	Status     string
	Body       string
	Attempts   int
}

func (*SessionUploadError) Error 

func (e *SessionUploadError) Error() string

type SetTracerProviderOptions 

type SetTracerProviderOptions struct {
	Metadata                 map[string]any
	RegisterSpanProcessor    func(sdktrace.SpanProcessor)
	UnregisterSpanProcessor  func(sdktrace.SpanProcessor)
	CreateCloudSpanProcessor func(CloudSpanProcessorOptions) sdktrace.SpanProcessor
}

SetTracerProviderOptions describes how an application-owned provider can accept processors that are attached after construction. Go's SDK provider already supports this through RegisterSpanProcessor and UnregisterSpanProcessor; custom providers can expose equivalent callbacks.

type SetupCloudTracerOptions 

type SetupCloudTracerOptions struct {
	RoomID        string
	JobID         string
	CloudHostname string
	AgentName     string
	EnableTraces  *bool
	EnableLogs    *bool
	Metadata      map[string]any

	APIKey     agents.SecretString
	APISecret  agents.SecretString
	HTTPClient *http.Client
	UploadGate *UploadGate

	TraceQueueSize int
	TraceBatchSize int
	TraceBatchWait time.Duration
	ExportTimeout  time.Duration
	// SetGlobalProvider is opt-in because libraries should not overwrite an
	// application's process-global provider by surprise.
	SetGlobalProvider bool
}

type SimpleLogRecord 

type SimpleLogRecord struct {
	Body           string
	Timestamp      time.Time
	TimestampMS    float64 // TypeScript compatibility; used when Timestamp is zero.
	Attributes     map[string]any
	SeverityNumber int32
	SeverityText   string
	TraceID        string
	SpanID         string
}

type SimpleOTLPHTTPLogExporter 

type SimpleOTLPHTTPLogExporter struct {
	// contains filtered or unexported fields
}

func NewSimpleOTLPHTTPLogExporter 

func NewSimpleOTLPHTTPLogExporter(config SimpleOTLPHTTPLogExporterConfig) (*SimpleOTLPHTTPLogExporter, error)

func (*SimpleOTLPHTTPLogExporter) Export 

func (e *SimpleOTLPHTTPLogExporter) Export(ctx context.Context, records []SimpleLogRecord) error

type SimpleOTLPHTTPLogExporterConfig 

type SimpleOTLPHTTPLogExporterConfig struct {
	CloudHostname      string
	Endpoint           string
	ResourceAttributes map[string]any
	ScopeName          string
	ScopeAttributes    map[string]any
	APIKey             agents.SecretString
	APISecret          agents.SecretString
	HTTPClient         *http.Client
	UploadGate         *UploadGate
	MaxRequestBytes    int64
	MaxErrorBodyBytes  int64
}

type StartSpanOptions 

type StartSpanOptions struct {
	Name       string
	Attributes map[string]any
	StartTime  time.Time
}

type UploadGate 

type UploadGate struct {
	// contains filtered or unexported fields
}

UploadGate is a session-generation latch that suppresses all subsequent Cloud telemetry after the project reports that data recording is disabled. Generations prevent a delayed response from a previous session disabling a newly started one.

func NewUploadGate 

func NewUploadGate(logger *slog.Logger) *UploadGate

func (*UploadGate) Disable 

func (g *UploadGate) Disable(generation uint64) bool

Disable returns true only for the first matching disable transition.

func (*UploadGate) Disabled 

func (g *UploadGate) Disabled() bool

func (*UploadGate) Generation 

func (g *UploadGate) Generation() uint64

func (*UploadGate) IsDisabledResponse 

func (g *UploadGate) IsDisabledResponse(statusCode int, body []byte) bool

func (*UploadGate) Reset 

func (g *UploadGate) Reset() uint64

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.