mod collector;
mod destination;
mod error;
mod manifest;
mod pipeline;
mod single_source;
mod uri;
#[cfg(test)]
mod tests;
use std::path::PathBuf;
pub use collector::{
CollectLimits, Collected, CollectedFile, DEFAULT_MAX_FILE_BYTES, DEFAULT_MAX_WIRE_BYTES, Level,
LogSource, OversizeFile, collect,
};
pub use destination::{LIST_LIMIT, LogDestination, ObjectMeta, ObjectStoreDestination, PutMeta};
pub use error::DrainError;
pub use manifest::{
DrainManifest, MANIFEST_FILENAME, MANIFEST_VERSION, ManifestEntry, ManifestOrigin, SkipReason,
SkipRecord, StatDecision,
};
pub use single_source::{
DEFAULT_INTERVAL_SECS, DrainOutcome, LogDrainSetting, ResolvedLogDrain, SingleSourceError,
SingleSourceSection, resolve_single_source, run_tick,
};
pub use uri::{DestinationScheme, DestinationUri};
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DrainTarget {
pub owner: String,
pub project: String,
}
impl DrainTarget {
pub fn validate(&self) -> Result<(), DrainError> {
if self.owner.trim().is_empty() {
return Err(DrainError::MissingIdentity { field: "owner" });
}
if self.project.trim().is_empty() {
return Err(DrainError::MissingIdentity { field: "project" });
}
Ok(())
}
pub fn key_prefix(&self) -> String {
format!("{}/{}", self.owner, self.project)
}
pub fn object_key(&self, relative_key: &str) -> String {
format!("{}/{}", self.key_prefix(), relative_key)
}
pub fn manifest_key(&self) -> String {
format!("{}/{}", self.key_prefix(), MANIFEST_FILENAME)
}
fn cache_key(&self) -> String {
self.key_prefix()
}
}
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct DrainConfig {
pub state_dir: PathBuf,
pub secrets: Vec<String>,
pub max_file_bytes: u64,
pub max_wire_bytes: u64,
}
impl DrainConfig {
pub fn new(state_dir: impl Into<PathBuf>) -> Self {
let limits = CollectLimits::default();
Self {
state_dir: state_dir.into(),
secrets: Vec::new(),
max_file_bytes: limits.max_file_bytes,
max_wire_bytes: limits.max_wire_bytes,
}
}
#[must_use]
pub fn with_secrets(mut self, secrets: Vec<String>) -> Self {
self.secrets = secrets;
self
}
#[must_use]
pub fn with_max_file_bytes(mut self, max: u64) -> Self {
self.max_file_bytes = max;
self
}
#[must_use]
pub fn with_max_wire_bytes(mut self, max: u64) -> Self {
self.max_wire_bytes = max;
self
}
pub fn limits(&self) -> CollectLimits {
CollectLimits::new(self.max_file_bytes, self.max_wire_bytes)
}
}
#[derive(Debug, Clone, Default, PartialEq, Eq)]
#[non_exhaustive]
pub struct DrainReport {
pub uploaded: usize,
pub skipped_unchanged: usize,
pub skipped_too_large: usize,
pub skips_recorded: usize,
pub bytes_plain: u64,
pub bytes_wire: u64,
pub errors: Vec<(String, String)>,
pub manifest_spot_check_missing: usize,
}
pub async fn run_once(
cfg: &DrainConfig,
dest: &dyn LogDestination,
target: &DrainTarget,
sources: &[LogSource],
) -> Result<DrainReport, DrainError> {
target.validate()?;
let manifest_key = target.manifest_key();
let cache_key = target.cache_key();
let (mut manifest, origin) =
DrainManifest::load_with_origin(dest, &cfg.state_dir, &manifest_key, &cache_key).await?;
let collected = collect(sources, &cfg.secrets, cfg.limits())?;
let mut report = DrainReport {
skipped_too_large: collected.oversize.len(),
..DrainReport::default()
};
for (path, message) in collected.errors {
report.errors.push((path.display().to_string(), message));
}
if origin == ManifestOrigin::Remote
&& let Some(missing) = manifest.spot_check(dest, &target.key_prefix()).await
{
report.manifest_spot_check_missing += 1;
tracing::warn!(
key = %missing,
manifest = %manifest_key,
"log-drain manifest lists an object this destination does not have; \
delete the manifest object to force a full re-upload (see #6548)"
);
}
let now = chrono::Utc::now().to_rfc3339();
let mut manifest_dirty = false;
for over in &collected.oversize {
if manifest.skip_recorded(&over.relative_key, over.size, over.mtime_unix) {
tracing::debug!(
path = %over.path.display(),
size = over.size,
"log-drain skip already recorded for this size and mtime"
);
continue;
}
tracing::warn!(
path = %over.path.display(),
size = over.size,
limit = over.reason.limit_name(),
"log-drain is not uploading this file; the decision is recorded in the \
manifest and is not logged again until the file's size or mtime changes"
);
manifest.record_skip(SkipRecord {
relative_file: over.relative_key.clone(),
size: over.size,
mtime_unix: over.mtime_unix,
reason: over.reason,
decided_at: now.clone(),
});
report.skips_recorded += 1;
manifest_dirty = true;
}
for file in collected.files {
if manifest.forget_skip(&file.relative_key) {
manifest_dirty = true;
}
if manifest.decide(&file.relative_key, file.plaintext_len, file.mtime_unix)
== StatDecision::SkipUnchanged
{
report.skipped_unchanged += 1;
continue;
}
if manifest.digest_matches(&file.relative_key, &file.sha256_plaintext) {
report.skipped_unchanged += 1;
manifest.record(entry_for(&file, &now));
manifest_dirty = true;
continue;
}
let key = target.object_key(&file.relative_key);
let wire_len = file.body.len() as u64;
match dest
.put(&key, file.body.clone(), PutMeta::gzipped_text())
.await
{
Ok(()) => {
report.uploaded += 1;
report.bytes_plain += file.plaintext_len;
report.bytes_wire += wire_len;
manifest.record(entry_for(&file, &now));
manifest_dirty = true;
}
Err(e) => report.errors.push((key, e.to_string())),
}
}
if manifest_dirty {
manifest
.save(dest, &cfg.state_dir, &manifest_key, &cache_key)
.await?;
}
Ok(report)
}
#[derive(Debug, Clone, Default, PartialEq, Eq)]
#[non_exhaustive]
pub struct PruneOutcome {
pub pruned: usize,
pub errors: Vec<(String, String)>,
pub refused: Option<PruneRefusal>,
}
pub const PRUNE_BATCH_COUNT_CEILING: usize = 500;
pub const PRUNE_BATCH_FRACTION_CEILING: f64 = 0.5;
pub const PRUNE_FRACTION_FLOOR: usize = 20;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum PruneRefusalReason {
CountCeiling,
FractionCeiling,
}
impl std::fmt::Display for PruneRefusalReason {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::CountCeiling => {
write!(f, "batch exceeds the {PRUNE_BATCH_COUNT_CEILING}-entry cap")
}
Self::FractionCeiling => write!(
f,
"batch exceeds {:.0}% of the manifest",
PRUNE_BATCH_FRACTION_CEILING * 100.0
),
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub struct PruneRefusal {
pub batch_len: usize,
pub manifest_len: usize,
pub reason: PruneRefusalReason,
}
fn prune_batch_refusal(batch_len: usize, manifest_len: usize) -> Option<PruneRefusal> {
if batch_len > PRUNE_BATCH_COUNT_CEILING {
return Some(PruneRefusal {
batch_len,
manifest_len,
reason: PruneRefusalReason::CountCeiling,
});
}
if manifest_len > PRUNE_FRACTION_FLOOR
&& (batch_len as f64) > (manifest_len as f64) * PRUNE_BATCH_FRACTION_CEILING
{
return Some(PruneRefusal {
batch_len,
manifest_len,
reason: PruneRefusalReason::FractionCeiling,
});
}
None
}
pub async fn prunable_candidates(
dest: &dyn LogDestination,
target: &DrainTarget,
state_dir: &std::path::Path,
retention: std::time::Duration,
) -> Result<Vec<String>, DrainError> {
let manifest =
DrainManifest::load(dest, state_dir, &target.manifest_key(), &target.cache_key()).await?;
Ok(manifest.prunable(chrono::Utc::now(), retention))
}
pub async fn prune_confirmed(
dest: &dyn LogDestination,
target: &DrainTarget,
state_dir: &std::path::Path,
confirm: &[String],
) -> Result<PruneOutcome, DrainError> {
let manifest_key = target.manifest_key();
let cache_key = target.cache_key();
let mut manifest = DrainManifest::load(dest, state_dir, &manifest_key, &cache_key).await?;
if let Some(refusal) = prune_batch_refusal(confirm.len(), manifest.entries.len()) {
return Ok(PruneOutcome {
refused: Some(refusal),
..PruneOutcome::default()
});
}
let mut outcome = PruneOutcome::default();
let mut dirty = false;
for relative_file in confirm {
if !manifest
.entries
.iter()
.any(|e| &e.relative_file == relative_file)
{
continue;
}
let key = target.object_key(relative_file);
match dest.delete(&key).await {
Ok(()) => {
manifest.remove_entry(relative_file);
outcome.pruned += 1;
dirty = true;
}
Err(e) => outcome.errors.push((key, e.to_string())),
}
}
if dirty {
manifest
.save(dest, state_dir, &manifest_key, &cache_key)
.await?;
}
Ok(outcome)
}
fn entry_for(file: &CollectedFile, uploaded_at: &str) -> ManifestEntry {
ManifestEntry {
relative_file: file.relative_key.clone(),
size: file.plaintext_len,
mtime_unix: file.mtime_unix,
sha256: file.sha256_plaintext.clone(),
uploaded_at: uploaded_at.to_string(),
}
}