try: ffi for metrics

Author: anunaym14

livekit-ffi/generate_proto.sh

@@ -31,4 +31,5 @@ protoc \
     $PROTOCOL/e2ee.proto \
     $PROTOCOL/stats.proto \
     $PROTOCOL/rpc.proto \
-    $PROTOCOL/data_stream.proto
+    $PROTOCOL/data_stream.proto \
+    $PROTOCOL/metrics.proto

livekit-ffi/protocol/ffi.proto

@@ -26,6 +26,7 @@ import "video_frame.proto";
 import "audio_frame.proto";
 import "rpc.proto";
 import "data_stream.proto";
+import "metrics.proto";
 
 // **How is the livekit-ffi working:
 // We refer as the ffi server the Rust server that is running the LiveKit client implementation, and we
@@ -148,7 +149,9 @@ message FfiRequest {
     TextStreamWriterWriteRequest text_stream_write = 65;
     TextStreamWriterCloseRequest text_stream_close = 66;
 
-    // NEXT_ID: 67
+    PublishMetricsRequest publish_metrics = 67;
+
+    // NEXT_ID: 68
   }
 }
 
@@ -243,7 +246,9 @@ message FfiResponse {
     TextStreamWriterWriteResponse text_stream_write = 64;
     TextStreamWriterCloseResponse text_stream_close = 65;
 
-    // NEXT_ID: 66
+    PublishMetricsResponse publish_metrics = 66;
+
+    // NEXT_ID: 67
   }
 }
 
@@ -301,6 +306,18 @@ message FfiEvent {
   }
 }
 
+message PublishMetricsRequest {
+  required uint64 local_participant_handle = 1;
+  repeated string str_data = 2;
+  repeated TimeSeriesMetric time_series = 3;
+  repeated EventMetric events = 4;
+  optional uint64 async_id = 5;
+}
+
+message PublishMetricsResponse {
+  required uint64 async_id = 1;
+}
+
 // Stop all rooms synchronously (Do we need async here?).
 // e.g: This is used for the Unity Editor after each assemblies reload.
 // TODO(theomonnom): Implement a debug mode where we can find all leaked handles?

livekit-ffi/protocol/metrics.proto

@@ -0,0 +1,89 @@
+syntax = "proto3";
+
+package livekit.proto;
+option go_package = "github.com/livekit/protocol/livekit";
+option csharp_namespace = "LiveKit.Proto";
+option ruby_package = "LiveKit::Proto";
+
+import "google/protobuf/timestamp.proto";
+
+
+/*
+  Protocol used to record metrics for a specific session.
+
+  Clients send their timestamp in their own monotonically increasing time (e.g `performance.now` on JS).
+  These timestamps are then augmented by the SFU to its time base.
+
+  A metric can be linked to a specific track by setting `track_sid`.
+*/
+
+
+// index from [0: MAX_LABEL_PREDEFINED_MAX_VALUE) are for predefined labels (`MetricLabel`)
+enum MetricLabel {
+  AGENTS_LLM_TTFT = 0; // time to first token from LLM
+  AGENTS_STT_TTFT = 1; // time to final transcription
+  AGENTS_TTS_TTFB = 2; // time to first byte
+
+  CLIENT_VIDEO_SUBSCRIBER_FREEZE_COUNT = 3;                           // Number of video freezes
+  CLIENT_VIDEO_SUBSCRIBER_TOTAL_FREEZE_DURATION = 4;                  // total duration of freezes
+  CLIENT_VIDEO_SUBSCRIBER_PAUSE_COUNT = 5;                            // number of video pauses
+  CLIENT_VIDEO_SUBSCRIBER_TOTAL_PAUSES_DURATION = 6;                  // total duration of pauses
+  CLIENT_AUDIO_SUBSCRIBER_CONCEALED_SAMPLES = 7;                      // number of concealed (synthesized) audio samples
+  CLIENT_AUDIO_SUBSCRIBER_SILENT_CONCEALED_SAMPLES = 8;               // number of silent concealed samples
+  CLIENT_AUDIO_SUBSCRIBER_CONCEALMENT_EVENTS = 9;                     // number of concealment events
+  CLIENT_AUDIO_SUBSCRIBER_INTERRUPTION_COUNT = 10;                    // number of interruptions
+  CLIENT_AUDIO_SUBSCRIBER_TOTAL_INTERRUPTION_DURATION = 11;           // total duration of interruptions
+  CLIENT_SUBSCRIBER_JITTER_BUFFER_DELAY = 12;                         // total time spent in jitter buffer
+  CLIENT_SUBSCRIBER_JITTER_BUFFER_EMITTED_COUNT = 13;                 // total time spent in jitter buffer
+  CLIENT_VIDEO_PUBLISHER_QUALITY_LIMITATION_DURATION_BANDWIDTH = 14;  // total duration spent in bandwidth quality limitation
+  CLIENT_VIDEO_PUBLISHER_QUALITY_LIMITATION_DURATION_CPU = 15;        // total duration spent in cpu quality limitation
+  CLIENT_VIDEO_PUBLISHER_QUALITY_LIMITATION_DURATION_OTHER = 16;      // total duration spent in other quality limitation
+
+  PUBLISHER_RTT = 17;   // Publisher RTT (participant -> server)
+  SERVER_MESH_RTT = 18; // RTT between publisher node and subscriber node (could involve intermedia node(s))
+  SUBSCRIBER_RTT = 19;  // Subscribe RTT (server -> participant)
+
+  METRIC_LABEL_PREDEFINED_MAX_VALUE = 4096;
+}
+
+message MetricsBatch {
+  int64 timestamp_ms = 1; // time at which this batch is sent based on a monotonic clock (millisecond resolution)
+  google.protobuf.Timestamp normalized_timestamp = 2;
+  // To avoid repeating string values, we store them in a separate list and reference them by index
+  // This is useful for storing participant identities, track names, etc.
+  // There is also a predefined list of labels that can be used to reference common metrics.
+  // They have reserved indices from 0 to (METRIC_LABEL_PREDEFINED_MAX_VALUE - 1).
+  // Indexes pointing at str_data should start from METRIC_LABEL_PREDEFINED_MAX_VALUE, 
+  // such that str_data[0] == index of METRIC_LABEL_PREDEFINED_MAX_VALUE.
+  repeated string str_data = 3;
+  repeated TimeSeriesMetric time_series = 4;
+  repeated EventMetric events = 5;
+}
+
+message TimeSeriesMetric {
+  // Metric name e.g "speech_probablity". The string value is not directly stored in the message, but referenced by index
+  // in the `str_data` field of `MetricsBatch`
+  uint32 label = 1;
+  uint32 participant_identity = 2; // index into `str_data`
+  uint32 track_sid = 3; // index into `str_data`
+  repeated MetricSample samples = 4;
+  uint32 rid = 5; // index into 'str_data'
+}
+
+message MetricSample {
+  int64 timestamp_ms = 1; // time of metric based on a monotonic clock (in milliseconds)
+  google.protobuf.Timestamp normalized_timestamp = 2;
+  float value = 3;
+}
+
+message EventMetric {
+  uint32 label = 1;
+  uint32 participant_identity = 2; // index into `str_data`
+  uint32 track_sid = 3; // index into `str_data`
+  int64 start_timestamp_ms = 4; // start time of event based on a monotonic clock (in milliseconds)
+  optional int64 end_timestamp_ms = 5; // end time of event based on a monotonic clock (in milliseconds), if needed
+  google.protobuf.Timestamp normalized_start_timestamp = 6;
+  optional google.protobuf.Timestamp normalized_end_timestamp = 7;
+  string metadata = 8;
+  uint32 rid = 9; // index into 'str_data'
+}

livekit-ffi/src/conversion/metrics.rs

@@ -0,0 +1,95 @@
+// Copyright 2023 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::proto;
+use livekit_protocol as protocol;
+
+impl From<protocol::MetricLabel> for proto::MetricLabel {
+    fn from(value: protocol::MetricLabel) -> Self {
+        match value {
+            protocol::MetricLabel::AgentsLlmTtft => Self::AgentsLlmTtft,
+            protocol::MetricLabel::AgentsSttTtft => Self::AgentsSttTtft,
+            protocol::MetricLabel::AgentsTtsTtfb => Self::AgentsTtsTtfb,
+            protocol::MetricLabel::ClientVideoSubscriberFreezeCount => Self::ClientVideoSubscriberFreezeCount,
+            protocol::MetricLabel::ClientVideoSubscriberTotalFreezeDuration => Self::ClientVideoSubscriberTotalFreezeDuration,
+            protocol::MetricLabel::ClientVideoSubscriberPauseCount => Self::ClientVideoSubscriberPauseCount,
+            protocol::MetricLabel::ClientVideoSubscriberTotalPausesDuration => Self::ClientVideoSubscriberTotalPausesDuration,
+            protocol::MetricLabel::ClientAudioSubscriberConcealedSamples => Self::ClientAudioSubscriberConcealedSamples,
+            protocol::MetricLabel::ClientAudioSubscriberSilentConcealedSamples => Self::ClientAudioSubscriberSilentConcealedSamples,
+            protocol::MetricLabel::ClientAudioSubscriberConcealmentEvents => Self::ClientAudioSubscriberConcealmentEvents,
+            protocol::MetricLabel::ClientAudioSubscriberInterruptionCount => Self::ClientAudioSubscriberInterruptionCount,
+            protocol::MetricLabel::ClientAudioSubscriberTotalInterruptionDuration => Self::ClientAudioSubscriberTotalInterruptionDuration,
+            protocol::MetricLabel::ClientSubscriberJitterBufferDelay => Self::ClientSubscriberJitterBufferDelay,
+            protocol::MetricLabel::ClientSubscriberJitterBufferEmittedCount => Self::ClientSubscriberJitterBufferEmittedCount,
+            protocol::MetricLabel::ClientVideoPublisherQualityLimitationDurationBandwidth => Self::ClientVideoPublisherQualityLimitationDurationBandwidth,
+            protocol::MetricLabel::ClientVideoPublisherQualityLimitationDurationCpu => Self::ClientVideoPublisherQualityLimitationDurationCpu,
+            protocol::MetricLabel::ClientVideoPublisherQualityLimitationDurationOther => Self::ClientVideoPublisherQualityLimitationDurationOther,
+            protocol::MetricLabel::PublisherRtt => Self::PublisherRtt,
+            protocol::MetricLabel::ServerMeshRtt => Self::ServerMeshRtt,
+            protocol::MetricLabel::SubscriberRtt => Self::SubscriberRtt,
+            protocol::MetricLabel::PredefinedMaxValue => Self::MetricLabelPredefinedMaxValue,
+        }
+    }
+}
+
+impl From<protocol::MetricSample> for proto::MetricSample {
+    fn from(value: protocol::MetricSample) -> Self {
+        Self {
+            timestamp_ms: value.timestamp_ms,
+            normalized_timestamp: value.normalized_timestamp.map(Into::into),
+            value: value.value,
+        }
+    }
+}
+
+impl From<protocol::TimeSeriesMetric> for proto::TimeSeriesMetric {
+    fn from(value: protocol::TimeSeriesMetric) -> Self {
+        Self {
+            label: value.label,
+            participant_identity: value.participant_identity,
+            track_sid: value.track_sid,
+            samples: value.samples.into_iter().map(Into::into).collect(),
+            rid: value.rid,
+        }
+    }
+}
+
+impl From<protocol::EventMetric> for proto::EventMetric {
+    fn from(value: protocol::EventMetric) -> Self {
+        Self {
+            label: value.label,
+            participant_identity: value.participant_identity,
+            track_sid: value.track_sid,
+            start_timestamp_ms: value.start_timestamp_ms,
+            end_timestamp_ms: value.end_timestamp_ms,
+            normalized_start_timestamp: value.normalized_start_timestamp.map(Into::into),
+            normalized_end_timestamp: value.normalized_end_timestamp.map(Into::into),
+            metadata: value.metadata,
+            rid: value.rid,
+        }
+    }
+}
+
+impl From<protocol::MetricsBatch> for proto::MetricsBatch {
+    fn from(value: protocol::MetricsBatch) -> Self {
+        Self {
+            timestamp_ms: value.timestamp_ms,
+            normalized_timestamp: value.normalized_timestamp.map(Into::into),
+            str_data: value.str_data,
+            time_series: value.time_series.into_iter().map(Into::into).collect(),
+            events: value.events.into_iter().map(Into::into).collect(),
+        }
+    }
+}
+

livekit-ffi/src/conversion/mod.rs

@@ -20,3 +20,4 @@ pub mod room;
 pub mod stats;
 pub mod track;
 pub mod video_frame;
+pub mod metrics;