line log generator

Author: TheSafo

Cargo.lock

@@ -30,6 +30,7 @@ use tokio::{
     fs,
     io::{AsyncWriteExt, BufWriter},
     task::{JoinError, JoinSet},
+    time::{Duration, Instant},
 };
 use tracing::{error, info};
 
@@ -120,6 +121,13 @@ pub struct Config {
     rotate: bool,
     /// The load throttle configuration
     pub throttle: Option<BytesThrottleConfig>,
+    /// Optional fixed interval between blocks. When set, the generator waits
+    /// this duration before emitting the next block, regardless of byte size.
+    pub block_interval_millis: Option<u64>,
+    /// Flush after each block. Useful when block intervals are large and the
+    /// buffered writer would otherwise delay writes to disk.
+    #[serde(default)]
+    pub flush_each_block: bool,
 }
 
 #[derive(Debug)]
@@ -195,6 +203,10 @@ impl Server {
                 file_index: Arc::clone(&file_index),
                 rotate: config.rotate,
                 shutdown: shutdown.clone(),
+                block_interval: config
+                    .block_interval_millis
+                    .map(Duration::from_millis),
+                flush_each_block: config.flush_each_block,
             };
 
             handles.spawn(child.spin());
@@ -269,6 +281,8 @@ struct Child {
     rotate: bool,
     file_index: Arc<AtomicU32>,
     shutdown: lading_signal::Watcher,
+    block_interval: Option<Duration>,
+    flush_each_block: bool,
 }
 
 impl Child {
@@ -303,13 +317,33 @@ impl Child {
 
         let shutdown_wait = self.shutdown.recv();
         tokio::pin!(shutdown_wait);
+        let mut next_tick = self
+            .block_interval
+            .as_ref()
+            .map(|dur| Instant::now() + *dur);
         loop {
             let total_bytes = self.block_cache.peek_next_size(&handle);
 
             tokio::select! {
                 result = self.throttle.wait_for(total_bytes) => {
                     match result {
                         Ok(()) => {
+                            if let Some(dur) = self.block_interval {
+                                if let Some(deadline) = next_tick {
+                                    tokio::select! {
+                                        _ = tokio::time::sleep_until(deadline) => {},
+                                        () = &mut shutdown_wait => {
+                                            fp.flush().await?;
+                                            info!("shutdown signal received");
+                                            return Ok(());
+                                        },
+                                    }
+                                    next_tick = Some(deadline + dur);
+                                } else {
+                                    next_tick = Some(Instant::now() + dur);
+                                }
+                            }
+
                             let block = self.block_cache.advance(&mut handle);
                             let total_bytes = u64::from(total_bytes.get());
 
@@ -318,6 +352,9 @@ impl Child {
                                 counter!("bytes_written").increment(total_bytes);
                                 total_bytes_written += total_bytes;
                             }
+                            if self.flush_each_block {
+                                fp.flush().await?;
+                            }
 
                             if total_bytes_written > maximum_bytes_per_file {
                                 fp.flush().await?;

lading/src/generator/file_gen/traditional.rs

@@ -28,6 +28,7 @@ rand = { workspace = true, features = [
   "std",
   "std_rng",
 ] }
+chrono = { version = "0.4", default-features = true, features = ["std"] }
 rmp-serde = { version = "1.1", default-features = false }
 serde = { workspace = true }
 serde_json = { workspace = true }

lading_payload/Cargo.toml

@@ -23,6 +23,12 @@ pub enum SpinError {
     /// Static payload creation error
     #[error(transparent)]
     Static(#[from] crate::statik::Error),
+    /// Static line-rate payload creation error
+    #[error(transparent)]
+    StaticLinesPerSecond(#[from] crate::statik_line_rate::Error),
+    /// Static second-grouped payload creation error
+    #[error(transparent)]
+    StaticSecond(#[from] crate::statik_second::Error),
     /// rng slice is Empty
     #[error("RNG slice is empty")]
     EmptyRng,
@@ -55,6 +61,12 @@ pub enum Error {
     /// Static payload creation error
     #[error(transparent)]
     Static(#[from] crate::statik::Error),
+    /// Static line-rate payload creation error
+    #[error(transparent)]
+    StaticLinesPerSecond(#[from] crate::statik_line_rate::Error),
+    /// Static second-grouped payload creation error
+    #[error(transparent)]
+    StaticSecond(#[from] crate::statik_second::Error),
     /// Error for crate deserialization
     #[error("Deserialization error: {0}")]
     Deserialize(#[from] crate::Error),
@@ -337,6 +349,37 @@ impl Cache {
                     total_bytes.get(),
                 )?
             }
+            crate::Config::StaticLinesPerSecond {
+                static_path,
+                lines_per_second,
+            } => {
+                let span = span!(Level::INFO, "fixed", payload = "static-lines-per-second");
+                let _guard = span.enter();
+                let mut serializer =
+                    crate::StaticLinesPerSecond::new(static_path, *lines_per_second)?;
+                construct_block_cache_inner(
+                    &mut rng,
+                    &mut serializer,
+                    maximum_block_bytes,
+                    total_bytes.get(),
+                )?
+            }
+            crate::Config::StaticSecond {
+                static_path,
+                timestamp_format,
+                emit_placeholder,
+            } => {
+                let span = span!(Level::INFO, "fixed", payload = "static-second");
+                let _guard = span.enter();
+                let mut serializer =
+                    crate::StaticSecond::new(static_path, &timestamp_format, *emit_placeholder)?;
+                construct_block_cache_inner(
+                    &mut rng,
+                    &mut serializer,
+                    maximum_block_bytes,
+                    total_bytes.get(),
+                )?
+            }
             crate::Config::OpentelemetryTraces => {
                 let mut pyld = crate::OpentelemetryTraces::new(&mut rng);
                 let span = span!(Level::INFO, "fixed", payload = "otel-traces");

lading_payload/src/block.rs

@@ -27,6 +27,8 @@ pub use opentelemetry::metric::OpentelemetryMetrics;
 pub use opentelemetry::trace::OpentelemetryTraces;
 pub use splunk_hec::SplunkHec;
 pub use statik::Static;
+pub use statik_second::StaticSecond;
+pub use statik_line_rate::StaticLinesPerSecond;
 pub use syslog::Syslog5424;
 
 pub mod apache_common;
@@ -40,6 +42,8 @@ pub mod opentelemetry;
 pub mod procfs;
 pub mod splunk_hec;
 pub mod statik;
+pub mod statik_second;
+pub mod statik_line_rate;
 pub mod syslog;
 pub mod trace_agent;
 
@@ -129,6 +133,28 @@ pub enum Config {
         /// assumed to be line-oriented but no other claim is made on the file.
         static_path: PathBuf,
     },
+    /// Generates static data but limits the number of lines emitted per block
+    StaticLinesPerSecond {
+        /// Defines the file path to read static variant data from. Content is
+        /// assumed to be line-oriented but no other claim is made on the file.
+        static_path: PathBuf,
+        /// Number of lines to emit in each generated block
+        lines_per_second: u32,
+    },
+    /// Generates static data grouped by second; each block contains one
+    /// second's worth of logs as determined by a parsed timestamp prefix.
+    StaticSecond {
+        /// Defines the file path to read static variant data from. Content is
+        /// assumed to be line-oriented.
+        static_path: PathBuf,
+        /// Chrono-compatible timestamp format string used to parse the leading
+        /// timestamp in each line.
+        timestamp_format: String,
+        /// Emit a minimal placeholder block (single newline) for seconds with
+        /// no lines. When false, empty seconds are skipped.
+        #[serde(default)]
+        emit_placeholder: bool,
+    },
     /// Generates a line of printable ascii characters
     Ascii,
     /// Generates a json encoded line
@@ -167,6 +193,10 @@ pub enum Payload {
     SplunkHec(splunk_hec::SplunkHec),
     /// Static file content
     Static(Static),
+    /// Static file content with a fixed number of lines emitted per block
+    StaticLinesPerSecond(StaticLinesPerSecond),
+    /// Static file content grouped into one-second blocks based on timestamps
+    StaticSecond(StaticSecond),
     /// Syslog RFC 5424 format
     Syslog(Syslog5424),
     /// OpenTelemetry traces
@@ -195,6 +225,8 @@ impl Serialize for Payload {
             Payload::Json(ser) => ser.to_bytes(rng, max_bytes, writer),
             Payload::SplunkHec(ser) => ser.to_bytes(rng, max_bytes, writer),
             Payload::Static(ser) => ser.to_bytes(rng, max_bytes, writer),
+            Payload::StaticLinesPerSecond(ser) => ser.to_bytes(rng, max_bytes, writer),
+            Payload::StaticSecond(ser) => ser.to_bytes(rng, max_bytes, writer),
             Payload::Syslog(ser) => ser.to_bytes(rng, max_bytes, writer),
             Payload::OtelTraces(ser) => ser.to_bytes(rng, max_bytes, writer),
             Payload::OtelLogs(ser) => ser.to_bytes(rng, max_bytes, writer),
@@ -207,6 +239,8 @@ impl Serialize for Payload {
     fn data_points_generated(&self) -> Option<u64> {
         match self {
             Payload::OtelMetrics(ser) => ser.data_points_generated(),
+            Payload::StaticLinesPerSecond(ser) => ser.data_points_generated(),
+            Payload::StaticSecond(ser) => ser.data_points_generated(),
             // Other implementations use the default None
             _ => None,
         }