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

@@ -590,7 +590,7 @@ impl Api {
       .iter()
       .map(|(k, v)| {
         (
-          k.to_string(),
+          k.clone(),
           ProtoData {
             data_type: Some(Data_type::StringData(v.clone())),
             ..Default::default()

api

@@ -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;

bd-api/src/api.rs

@@ -590,11 +590,7 @@ impl Uploader {
 
         #[cfg(test)]
         if let Some(hooks) = &self.test_hooks {
-          hooks
-            .entry_received_tx
-            .send(uuid.to_string())
-            .await
-            .unwrap();
+          hooks.entry_received_tx.send(uuid.clone()).await.unwrap();
         }
         return;
       },
@@ -606,11 +602,7 @@ impl Uploader {
 
       #[cfg(test)]
       if let Some(hooks) = &self.test_hooks {
-        hooks
-          .entry_received_tx
-          .send(uuid.to_string())
-          .await
-          .unwrap();
+        hooks.entry_received_tx.send(uuid.clone()).await.unwrap();
       }
       return;
     }
@@ -668,11 +660,7 @@ impl Uploader {
 
     #[cfg(test)]
     if let Some(hooks) = &self.test_hooks {
-      hooks
-        .entry_received_tx
-        .send(uuid.to_string())
-        .await
-        .unwrap();
+      hooks.entry_received_tx.send(uuid.clone()).await.unwrap();
     }
   }
 
@@ -766,8 +754,8 @@ impl Uploader {
         upload_uuid.clone(),
         UploadArtifactIntentRequest {
           type_id: "client_report".to_string(),
-          artifact_id: id.to_string(),
-          intent_uuid: upload_uuid.to_string(),
+          artifact_id: id.clone(),
+          intent_uuid: upload_uuid.clone(),
           time: timestamp.into_proto(),
           // TODO(snowp): Figure out how to send relevant metadata about the artifact here.
           metadata: vec![],

bd-api/src/api_test.rs

@@ -1053,7 +1053,7 @@ fn test_value_none_removed_partial_errors() {
 
 #[test]
 fn test_all_variants() {
-  let all_value_types = vec![
+  let all_value_types = [
     Value::Null,
     Value::Bool(true),
     Value::Float(std::f64::consts::PI),

bd-artifact-upload/src/uploader.rs

@@ -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-bonjson/src/decoder_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),