ring buffer: store logs as proto instead of fb

For logs with many fields, the fact that fb is padded and doesn't
use varint encoding leads to substantial overhead. This removes
that overhead. As part of this change I went a bit over the top
and also implemented a custom encoder for the Log proto which
allows for zero copy into the ring buffer. This is better than the
previous fb implementation which encoded and then did a memcpy into
the buffer.

Signed-off-by: Matt Klein <[email protected]>

Author: mattklein123

Cargo.lock

@@ -20,7 +20,7 @@ use bd_client_stats_store::Collector;
 use bd_client_stats_store::test::StatsHelper;
 use bd_grpc_codec::code::Code;
 use bd_grpc_codec::{Decompression, Encoder, OptimizeFor};
-use bd_internal_logging::{LogFields, LogLevel, LogType};
+use bd_internal_logging::{LogFields, LogLevel};
 use bd_key_value::Store;
 use bd_metadata::{Metadata, Platform};
 use bd_network_quality::{NetworkQuality, NetworkQualityResolver as _};
@@ -37,6 +37,7 @@ use bd_proto::protos::client::api::{
   RuntimeUpdate,
   StatsUploadRequest,
 };
+use bd_proto::protos::logging::payload::LogType;
 use bd_runtime::runtime::{ConfigLoader, FeatureFlag};
 use bd_stats_common::labels;
 use bd_test_helpers::make_mut;

api

@@ -15,6 +15,7 @@ bd-client-common.path           = "../bd-client-common"
 bd-client-stats-store.path      = "../bd-client-stats-store"
 bd-completion.path              = "../bd-completion"
 bd-error-reporter.path          = "../bd-error-reporter"
+bd-log-primitives.path          = "../bd-log-primitives"
 bd-proto.path                   = "../bd-proto"
 bd-runtime.path                 = "../bd-runtime"
 bd-stats-common.path            = "../bd-stats-common"

bd-api/src/api_test.rs

@@ -24,11 +24,11 @@ use super::{
   RingBufferProducer,
   RingBufferStats,
   VolatileRingBuffer,
-  to_u32,
 };
 #[cfg(test)]
 use crate::buffer::test::thread_synchronizer::ThreadSynchronizer;
 use crate::{AbslCode, Error, Result};
+use bd_log_primitives::LossyIntToU32;
 use parking_lot::Mutex;
 #[cfg(test)]
 use std::any::Any;
@@ -76,7 +76,7 @@ impl SharedData {
       // the volatile buffer.
       match producer
         .as_mut()
-        .reserve(to_u32(read_reservation.len()), true)
+        .reserve(read_reservation.len().to_u32(), true)
       {
         Ok(write_reservation) => {
           write_reservation.copy_from_slice(read_reservation);
@@ -138,8 +138,8 @@ impl RingBufferImpl {
     // For aggregate buffers, the size of the file (after subtracting header space) must be >= the
     // size of RAM. This is to avoid situations in which we accept a record into RAM but cannot ever
     // write it to disk.
-    if non_volatile_size < to_u32(std::mem::size_of::<FileHeader>())
-      || volatile_size > (non_volatile_size - to_u32(std::mem::size_of::<FileHeader>()))
+    if non_volatile_size < std::mem::size_of::<FileHeader>().to_u32()
+      || volatile_size > (non_volatile_size - std::mem::size_of::<FileHeader>().to_u32())
     {
       log::error!(
         "file size '{}' not big enough for header size '{}' or file size (minus header) not \

bd-buffer/Cargo.toml

@@ -18,8 +18,9 @@ use crate::buffer::test::{
   read_and_verify,
   start_read_and_verify,
 };
-use crate::buffer::{RingBuffer, RingBufferStats, StatsTestHelper, to_u32};
+use crate::buffer::{RingBuffer, RingBufferStats, StatsTestHelper};
 use bd_client_stats_store::Collector;
+use bd_log_primitives::LossyIntToU32;
 use futures::poll;
 use std::sync::Arc;
 use tempfile::TempDir;
@@ -48,7 +49,7 @@ impl Helper {
       "test",
       volatile_size,
       temp_dir.path().join("buffer"),
-      non_volatile_size + to_u32(std::mem::size_of::<FileHeader>()),
+      non_volatile_size + std::mem::size_of::<FileHeader>().to_u32(),
       PerRecordCrc32Check::Yes,
       allow_overwrite,
       Arc::new(RingBufferStats::default()),
@@ -80,7 +81,7 @@ impl Helper {
         "test",
         self.volatile_size,
         self.temp_dir.path().join("buffer"),
-        self.non_volatile_size + to_u32(std::mem::size_of::<FileHeader>()),
+        self.non_volatile_size + std::mem::size_of::<FileHeader>().to_u32(),
         PerRecordCrc32Check::Yes,
         self.allow_overwrite,
         Arc::new(RingBufferStats::default()),

bd-buffer/src/buffer/aggregate_ring_buffer.rs

@@ -9,11 +9,12 @@
 #[path = "./common_ring_buffer_test.rs"]
 mod common_ring_buffer_test;
 
+use super::RingBufferStats;
 #[cfg(test)]
 use super::test::thread_synchronizer::ThreadSynchronizer;
-use super::{RingBufferStats, to_u32};
 use crate::{AbslCode, Error, Result};
 use bd_client_common::error::InvariantError;
+use bd_log_primitives::LossyIntToU32;
 use parking_lot::{Condvar, Mutex, MutexGuard};
 use std::fmt::Display;
 use std::ptr::NonNull;
@@ -229,14 +230,14 @@ impl<ExtraLockedData> LockedData<ExtraLockedData> {
   // part of the record. The extra space is always fixed and is the value stored in
   // extra_bytes_per_record_.
   fn record_size_offset(&self, start: u32) -> u32 {
-    debug_assert!(self.extra_bytes_per_record >= to_u32(std::mem::size_of::<u32>()));
-    start + (self.extra_bytes_per_record - to_u32(std::mem::size_of::<u32>()))
+    debug_assert!(self.extra_bytes_per_record >= std::mem::size_of::<u32>().to_u32());
+    start + (self.extra_bytes_per_record - std::mem::size_of::<u32>().to_u32())
   }
 
   // Returns any extra space at the beginning of the reservation. See recordSizeOffset() for more
   // information on the record layout.
   fn extra_data(&mut self, start: u32) -> &mut [u8] {
-    debug_assert!(self.extra_bytes_per_record >= to_u32(std::mem::size_of::<u32>()));
+    debug_assert!(self.extra_bytes_per_record >= std::mem::size_of::<u32>().to_u32());
     let start = start as usize;
     let extra_bytes_per_record = self.extra_bytes_per_record as usize;
     &mut self.memory()[start .. start + extra_bytes_per_record - std::mem::size_of::<u32>()]
@@ -299,7 +300,7 @@ impl<ExtraLockedData> LockedData<ExtraLockedData> {
     // space. This can overflow but as long as we are within the memory space crc checks will catch
     // further corruption.
     let size_index = self.record_size_offset(next_read_start_to_use);
-    if size_index + to_u32(std::mem::size_of::<u32>()) > to_u32(self.memory.0.len()) {
+    if size_index + std::mem::size_of::<u32>().to_u32() > self.memory.0.len().to_u32() {
       return Err(Error::AbslStatus(
         AbslCode::DataLoss,
         "corrupted record size index".to_string(),
@@ -318,7 +319,7 @@ impl<ExtraLockedData> LockedData<ExtraLockedData> {
     // space crc checks will catch further corruption.
     if size == 0
       || Self::overflow_add(&[next_read_start_to_use, self.extra_bytes_per_record, size])
-        .is_none_or(|next_read_start_index| next_read_start_index > to_u32(self.memory.0.len()))
+        .is_none_or(|next_read_start_index| next_read_start_index > self.memory.0.len().to_u32())
     {
       return Err(Error::AbslStatus(
         AbslCode::DataLoss,
@@ -343,7 +344,7 @@ impl<ExtraLockedData> LockedData<ExtraLockedData> {
         .last_write_end_before_wrap
         .ok_or(InvariantError::Invariant)?
         + 1,
-      size: to_u32(self.memory.0.len())
+      size: self.memory.0.len().to_u32()
         - (reservation_data
           .last_write_end_before_wrap
           .ok_or(InvariantError::Invariant)?
@@ -469,7 +470,7 @@ impl<ExtraLockedData> LockedData<ExtraLockedData> {
     let mut reservation_data = TempReservationData::default();
     let next_write_start = *self.next_write_start();
 
-    if next_write_start + write_size <= to_u32(self.memory.0.len()) {
+    if next_write_start + write_size <= self.memory.0.len().to_u32() {
       // It fits in the remainder without wrapping.
       reservation_data.range.start = next_write_start;
       reservation_data.next_write_start = next_write_start + write_size;
@@ -529,7 +530,7 @@ impl<ExtraLockedData> LockedData<ExtraLockedData> {
     };
 
     // If size is 0 or > then the ring buffer size we can't do anything.
-    if size == 0 || actual_size > to_u32(self.memory.0.len()) {
+    if size == 0 || actual_size > self.memory.0.len().to_u32() {
       log::trace!("({}) invalid reservation size: {}", self.name, size);
       // Note that we don't record the bytes lost in this case, since a very large number is likely
       // to cause overflow.
@@ -996,7 +997,7 @@ impl<ExtraLockedData> CommonRingBuffer<ExtraLockedData> {
         lock_count: AtomicU32::default().into(),
         shutdown_lock: None,
         stats,
-        extra_bytes_per_record: extra_bytes_per_record + to_u32(std::mem::size_of::<u32>()),
+        extra_bytes_per_record: extra_bytes_per_record + std::mem::size_of::<u32>().to_u32(),
         extra_locked_data,
         allow_overwrite,
         on_total_data_loss_cb: Box::new(on_total_data_loss_cb),

bd-buffer/src/buffer/aggregate_ring_buffer_test.rs

@@ -22,11 +22,11 @@ use crate::buffer::{
   RingBufferStats,
   StatsTestHelper,
   VolatileRingBuffer,
-  to_u32,
 };
 use crate::{AbslCode, Error};
 use assert_matches::assert_matches;
 use bd_client_stats_store::Collector;
+use bd_log_primitives::LossyIntToU32;
 use parameterized::parameterized;
 use std::any::Any;
 use std::sync::Arc;
@@ -78,7 +78,7 @@ impl Helper {
       TestType::NonVolatile => NonVolatileRingBuffer::new(
         "test".to_string(),
         temp_dir.path().join("buffer"),
-        size + to_u32(std::mem::size_of::<FileHeader>()),
+        size + std::mem::size_of::<FileHeader>().to_u32(),
         AllowOverwrite::Yes,
         BlockWhenReservingIntoConcurrentRead::No,
         PerRecordCrc32Check::No,
@@ -89,7 +89,7 @@ impl Helper {
         "test",
         size,
         temp_dir.path().join("buffer"),
-        size + to_u32(std::mem::size_of::<FileHeader>()),
+        size + std::mem::size_of::<FileHeader>().to_u32(),
         PerRecordCrc32Check::No,
         AllowOverwrite::Yes,
         Arc::new(RingBufferStats::default()),

bd-buffer/src/buffer/common_ring_buffer.rs

@@ -16,20 +16,13 @@ mod volatile_ring_buffer;
 mod test;
 
 use bd_client_stats_store::Counter;
+use bd_log_primitives::LossyIntToU32;
 #[cfg(test)]
 use std::any::Any;
 use std::sync::Arc;
 #[cfg(test)]
 use test::thread_synchronizer::ThreadSynchronizer;
 
-// Wrapper that allows converting a usize to a u32. This avoids verbosity in places we know this is
-// safe.
-#[allow(clippy::cast_possible_truncation)]
-#[must_use]
-pub const fn to_u32(value: usize) -> u32 {
-  value as u32
-}
-
 //
 // RingBufferStats
 //
@@ -101,7 +94,7 @@ pub trait RingBufferProducer: Send {
 
   // Writes a single record to the buffer by copying the provided data into the record reservation.
   fn write(&mut self, data: &[u8]) -> Result<()> {
-    let reserved = self.reserve(to_u32(data.len()), true)?;
+    let reserved = self.reserve(data.len().to_u32(), true)?;
     reserved.copy_from_slice(data);
     self.commit()
   }

bd-buffer/src/buffer/common_ring_buffer_test.rs

@@ -29,10 +29,10 @@ use super::{
   RingBufferCursorConsumer,
   RingBufferProducer,
   RingBufferStats,
-  to_u32,
 };
 use crate::{AbslCode, Error, Result};
 use bd_client_common::error::InvariantError;
+use bd_log_primitives::LossyIntToU32;
 use crc32fast::Hasher;
 use fs2::FileExt;
 use intrusive_collections::offset_of;
@@ -632,7 +632,9 @@ pub struct FileHeader {
   crc32: u32,
 }
 
-const FILE_HEADER_CURRENT_VERSION: u32 = 1;
+// Version 1: Original version.
+// Version 2: Switched to protobuf encoding for records.
+const FILE_HEADER_CURRENT_VERSION: u32 = 2;
 
 //
 // ConsumerType
@@ -738,7 +740,7 @@ impl RingBufferImpl {
     const_assert_eq!(offset_of!(FileHeader, next_read_start), 28);
     const_assert_eq!(offset_of!(FileHeader, crc32), 36);
 
-    if size < to_u32(std::mem::size_of::<FileHeader>()) {
+    if size < std::mem::size_of::<FileHeader>().to_u32() {
       log::error!(
         "({name}) file size '{}' not big enough for header size '{}'",
         size,