use std::io::{Cursor, Read, Write};
use crate::{
context::{BuildEventContext, read_build_event_context},
error::MuninError,
field_flags::BuildEventArgsFieldFlags,
header::{BinlogHeader, open_binlog},
jsonlog::schema::{JsonlogEventBody, JsonlogFile, MUNIN_JSONLOG_VERSION},
nvl_table::{NameValueListTable, NameValuePair},
primitives::{read_7bit_count, read_7bit_int},
reader::{ArchiveEntry, BinlogEvent, dispatch_event},
record_kind::BinaryLogRecordKind,
string_table::StringTable,
writers::{WriteContext, write_7bit_int as w_7bit},
};
#[derive(Debug, Clone)]
pub struct EventMeta {
pub record_kind: BinaryLogRecordKind,
pub byte_offset: u64,
pub payload_len: usize,
pub context: Option<BuildEventContext>,
}
#[derive(Debug, Clone)]
struct IndexEntry {
meta: EventMeta,
payload: Vec<u8>,
}
#[derive(Debug)]
pub struct BinlogIndex {
header: BinlogHeader,
strings: StringTable,
nvl_table: NameValueListTable,
archives: Vec<Vec<u8>>,
entries: Vec<IndexEntry>,
}
impl BinlogIndex {
pub fn open(reader: impl Read) -> Result<Self, MuninError> {
let (header, mut gz_reader) = open_binlog(reader)?;
let version = header.file_format_version;
let mut strings = StringTable::new();
let mut nvl_table = NameValueListTable::new();
let mut archives = Vec::new();
let mut entries = Vec::new();
let mut offset: u64 = 8;
loop {
let kind_start = offset;
let (kind_raw, kind_bytes) = read_7bit_int_counted(&mut gz_reader)?;
offset += kind_bytes;
if kind_raw == BinaryLogRecordKind::EndOfFile as i32 {
break;
}
let (record_length, len_bytes) = read_7bit_int_counted(&mut gz_reader)?;
if record_length < 0 {
return Err(MuninError::InvalidFormat(format!(
"negative record length: {record_length}"
)));
}
let record_length = record_length as usize;
if record_length > crate::primitives::MAX_BINLOG_FIELD_LEN {
return Err(MuninError::InvalidFormat(format!(
"record length too large: {record_length} (max {})",
crate::primitives::MAX_BINLOG_FIELD_LEN
)));
}
offset += len_bytes;
let mut payload = vec![0u8; record_length];
gz_reader.read_exact(&mut payload)?;
offset += record_length as u64;
let kind = BinaryLogRecordKind::from_raw(kind_raw);
match kind {
Some(BinaryLogRecordKind::String) => {
let s = String::from_utf8(payload).map_err(|_| MuninError::InvalidUtf8)?;
strings.add(s);
}
Some(BinaryLogRecordKind::NameValueList) => {
let mut cursor = Cursor::new(&payload);
let count = read_7bit_count(&mut cursor, "name-value list count")?;
let mut pairs = Vec::with_capacity(count);
for _ in 0..count {
let key_index = read_7bit_int(&mut cursor)?;
let value_index = read_7bit_int(&mut cursor)?;
pairs.push(NameValuePair {
key_index,
value_index,
});
}
nvl_table.add(pairs);
}
Some(BinaryLogRecordKind::ProjectImportArchive) => {
archives.push(payload);
}
Some(record_kind) if !record_kind.is_auxiliary() => {
let context = extract_context(&payload, version);
let meta = EventMeta {
record_kind,
byte_offset: kind_start,
payload_len: record_length,
context,
};
entries.push(IndexEntry { meta, payload });
}
_ => {}
}
}
Ok(Self {
header,
strings,
nvl_table,
archives,
entries,
})
}
pub fn open_json(reader: impl Read) -> Result<Self, MuninError> {
let file: JsonlogFile = serde_json::from_reader(reader)
.map_err(|e| MuninError::InvalidFormat(format!("jsonlog parse: {e}")))?;
Self::from_jsonlog(file)
}
pub fn from_jsonlog(file: JsonlogFile) -> Result<Self, MuninError> {
if file.munin_jsonlog_version != MUNIN_JSONLOG_VERSION {
return Err(MuninError::InvalidFormat(format!(
"unsupported jsonlog version {} (expected {})",
file.munin_jsonlog_version, MUNIN_JSONLOG_VERSION
)));
}
let header: BinlogHeader = file.header.into();
let version = header.file_format_version;
let mut strings = StringTable::new();
for s in file.strings {
strings.add(s);
}
let mut nvl_table = NameValueListTable::new();
for entry in file.name_value_lists {
let pairs = entry
.into_iter()
.map(|pair| NameValuePair {
key_index: pair[0] as i32,
value_index: pair[1] as i32,
})
.collect();
nvl_table.add(pairs);
}
let archives: Vec<Vec<u8>> = file
.archives
.into_iter()
.map(|a| {
use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64};
BASE64
.decode(a.data_b64.as_bytes())
.map_err(|e| MuninError::InvalidFormat(format!("archive base64: {e}")))
})
.collect::<Result<Vec<_>, _>>()?;
let mut entries: Vec<IndexEntry> = Vec::with_capacity(file.events.len());
let mut ctx = WriteContext::with_tables(version, strings, nvl_table);
for ev in file.events {
let record_kind = BinaryLogRecordKind::from_name(&ev.kind).ok_or_else(|| {
MuninError::InvalidFormat(format!("unknown event kind: {}", ev.kind))
})?;
let payload = match ev.body {
JsonlogEventBody::PayloadB64(s) => {
use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64};
BASE64
.decode(s.as_bytes())
.map_err(|e| MuninError::InvalidFormat(format!("payload base64: {e}")))?
}
JsonlogEventBody::Decoded(value) => {
encode_decoded_event(record_kind, value, &mut ctx)?
}
};
let context = extract_context(&payload, version);
entries.push(IndexEntry {
meta: EventMeta {
record_kind,
byte_offset: ev.byte_offset,
payload_len: payload.len(),
context,
},
payload,
});
}
let strings = ctx.strings;
let nvl_table = ctx.nvl_table;
Ok(Self {
header,
strings,
nvl_table,
archives,
entries,
})
}
pub fn write_binlog<W: Write>(&self, writer: W) -> Result<(), MuninError> {
use flate2::{Compression, write::GzEncoder};
let mut gz = GzEncoder::new(writer, Compression::default());
gz.write_all(&self.header.file_format_version.to_le_bytes())?;
gz.write_all(&self.header.min_reader_version.to_le_bytes())?;
for s in self.strings.entries() {
let bytes = s.as_bytes();
w_7bit(&mut gz, BinaryLogRecordKind::String as i32)?;
w_7bit(&mut gz, bytes.len() as i32)?;
gz.write_all(bytes)?;
}
for list in self.nvl_table.entries() {
let mut buf = Vec::new();
w_7bit(&mut buf, list.len() as i32)?;
for pair in list {
w_7bit(&mut buf, pair.key_index)?;
w_7bit(&mut buf, pair.value_index)?;
}
w_7bit(&mut gz, BinaryLogRecordKind::NameValueList as i32)?;
w_7bit(&mut gz, buf.len() as i32)?;
gz.write_all(&buf)?;
}
for archive in &self.archives {
w_7bit(&mut gz, BinaryLogRecordKind::ProjectImportArchive as i32)?;
w_7bit(&mut gz, archive.len() as i32)?;
gz.write_all(archive)?;
}
for entry in &self.entries {
w_7bit(&mut gz, entry.meta.record_kind as i32)?;
w_7bit(&mut gz, entry.payload.len() as i32)?;
gz.write_all(&entry.payload)?;
}
w_7bit(&mut gz, BinaryLogRecordKind::EndOfFile as i32)?;
gz.finish()?;
Ok(())
}
pub fn header(&self) -> &BinlogHeader {
&self.header
}
pub fn strings(&self) -> &StringTable {
&self.strings
}
pub fn strings_mut(&mut self) -> &mut StringTable {
&mut self.strings
}
pub fn nvl_table(&self) -> &NameValueListTable {
&self.nvl_table
}
pub fn archives(&self) -> &[Vec<u8>] {
&self.archives
}
pub fn len(&self) -> usize {
self.entries.len()
}
pub fn is_empty(&self) -> bool {
self.entries.is_empty()
}
pub fn meta(&self, index: usize) -> Option<&EventMeta> {
self.entries.get(index).map(|e| &e.meta)
}
pub fn iter_meta(&self) -> impl Iterator<Item = (usize, &EventMeta)> {
self.entries.iter().enumerate().map(|(i, e)| (i, &e.meta))
}
pub fn payload_bytes(&self, index: usize) -> Option<&[u8]> {
self.entries.get(index).map(|e| e.payload.as_slice())
}
pub fn get(&self, index: usize) -> Result<Option<BinlogEvent>, MuninError> {
let entry = match self.entries.get(index) {
Some(e) => e,
None => return Ok(None),
};
let mut cursor = Cursor::new(&entry.payload);
let event = dispatch_event(
&mut cursor,
entry.meta.record_kind,
&self.strings,
&self.nvl_table,
self.header.file_format_version,
)?;
Ok(Some(event))
}
pub fn get_all(&self) -> Result<Vec<BinlogEvent>, MuninError> {
let mut events = Vec::with_capacity(self.entries.len());
for i in 0..self.entries.len() {
if let Some(event) = self.get(i)? {
events.push(event);
}
}
Ok(events)
}
pub fn indices_by_kind(&self, kind: BinaryLogRecordKind) -> Vec<usize> {
self.entries
.iter()
.enumerate()
.filter(|(_, e)| e.meta.record_kind == kind)
.map(|(i, _)| i)
.collect()
}
pub fn indices_by_project_context(&self, project_context_id: i32) -> Vec<usize> {
self.entries
.iter()
.enumerate()
.filter(|(_, e)| {
e.meta
.context
.is_some_and(|c| c.project_context_id == project_context_id)
})
.map(|(i, _)| i)
.collect()
}
pub fn indices_by_target_id(&self, target_id: i32) -> Vec<usize> {
self.entries
.iter()
.enumerate()
.filter(|(_, e)| e.meta.context.is_some_and(|c| c.target_id == target_id))
.map(|(i, _)| i)
.collect()
}
pub fn indices_by_task_id(&self, task_id: i32) -> Vec<usize> {
self.entries
.iter()
.enumerate()
.filter(|(_, e)| e.meta.context.is_some_and(|c| c.task_id == task_id))
.map(|(i, _)| i)
.collect()
}
pub fn query(
&self,
kind: Option<BinaryLogRecordKind>,
project_context_id: Option<i32>,
target_id: Option<i32>,
task_id: Option<i32>,
) -> Vec<usize> {
self.entries
.iter()
.enumerate()
.filter(|(_, e)| {
if let Some(k) = kind
&& e.meta.record_kind != k
{
return false;
}
if let Some(pid) = project_context_id
&& e.meta.context.is_none_or(|c| c.project_context_id != pid)
{
return false;
}
if let Some(tid) = target_id
&& e.meta.context.is_none_or(|c| c.target_id != tid)
{
return false;
}
if let Some(tsk) = task_id
&& e.meta.context.is_none_or(|c| c.task_id != tsk)
{
return false;
}
true
})
.map(|(i, _)| i)
.collect()
}
pub fn extract_archives(&self) -> Result<Vec<ArchiveEntry>, MuninError> {
let mut entries = Vec::new();
for archive_bytes in &self.archives {
let cursor = Cursor::new(archive_bytes);
let mut archive = zip::ZipArchive::new(cursor).map_err(|e| {
MuninError::InvalidFormat(format!("invalid ProjectImportArchive zip: {e}"))
})?;
for i in 0..archive.len() {
let mut file = archive.by_index(i).map_err(|e| {
MuninError::InvalidFormat(format!("cannot read zip entry {i}: {e}"))
})?;
if file.is_dir() {
continue;
}
let path = file.name().to_string();
let mut contents = String::new();
if file.read_to_string(&mut contents).is_ok() {
entries.push(ArchiveEntry { path, contents });
}
}
}
Ok(entries)
}
}
fn read_7bit_int_counted(reader: &mut impl Read) -> Result<(i32, u64), MuninError> {
let mut result: u32 = 0;
let mut shift: u32 = 0;
let mut buf = [0u8; 1];
let mut count: u64 = 0;
for _ in 0..5 {
reader.read_exact(&mut buf)?;
count += 1;
let byte = buf[0];
result |= ((byte & 0x7F) as u32) << shift;
if byte & 0x80 == 0 {
return Ok((result as i32, count));
}
shift += 7;
}
Err(MuninError::OverlongVarInt)
}
fn extract_context(payload: &[u8], file_format_version: i32) -> Option<BuildEventContext> {
let mut cursor = Cursor::new(payload);
let flags_raw = read_7bit_int(&mut cursor).ok()? as u32;
let flags = BuildEventArgsFieldFlags::from_raw(flags_raw);
if flags.contains(BuildEventArgsFieldFlags::MESSAGE) {
let _ = read_7bit_int(&mut cursor).ok()?;
}
if flags.contains(BuildEventArgsFieldFlags::BUILD_EVENT_CONTEXT) {
read_build_event_context(&mut cursor, file_format_version).ok()
} else {
None
}
}
#[cfg(test)]
mod tests;
fn encode_decoded_event(
kind: BinaryLogRecordKind,
value: serde_json::Value,
ctx: &mut WriteContext,
) -> Result<Vec<u8>, MuninError> {
use crate::events as ev;
let mut buf = Vec::new();
macro_rules! enc {
($Ty:ty, $write:path) => {{
let parsed: $Ty = serde_json::from_value(value)
.map_err(|e| MuninError::InvalidFormat(format!("decoded event: {e}")))?;
$write(&mut buf, ctx, &parsed)
.map_err(|e| MuninError::InvalidFormat(format!("encode event: {e}")))?;
}};
}
match kind {
BinaryLogRecordKind::BuildStarted => enc!(ev::BuildStartedEvent, ev::write_build_started),
BinaryLogRecordKind::BuildFinished => {
enc!(ev::BuildFinishedEvent, ev::write_build_finished)
}
BinaryLogRecordKind::ProjectStarted => {
enc!(ev::ProjectStartedEvent, ev::write_project_started)
}
BinaryLogRecordKind::ProjectFinished => {
enc!(ev::ProjectFinishedEvent, ev::write_project_finished)
}
BinaryLogRecordKind::TargetStarted => {
enc!(ev::TargetStartedEvent, ev::write_target_started)
}
BinaryLogRecordKind::TargetFinished => {
enc!(ev::TargetFinishedEvent, ev::write_target_finished)
}
BinaryLogRecordKind::TargetSkipped => {
enc!(ev::TargetSkippedEvent, ev::write_target_skipped)
}
BinaryLogRecordKind::TaskStarted => enc!(ev::TaskStartedEvent, ev::write_task_started),
BinaryLogRecordKind::TaskFinished => enc!(ev::TaskFinishedEvent, ev::write_task_finished),
BinaryLogRecordKind::TaskCommandLine => {
enc!(ev::TaskCommandLineEvent, ev::write_task_command_line)
}
BinaryLogRecordKind::TaskParameter => {
enc!(ev::TaskParameterEvent, ev::write_task_parameter)
}
BinaryLogRecordKind::Error => enc!(ev::BuildErrorEvent, ev::write_build_error),
BinaryLogRecordKind::Warning => enc!(ev::BuildWarningEvent, ev::write_build_warning),
BinaryLogRecordKind::Message => enc!(ev::BuildMessageEvent, ev::write_build_message),
BinaryLogRecordKind::CriticalBuildMessage => {
enc!(
ev::CriticalBuildMessageEvent,
ev::write_critical_build_message
)
}
BinaryLogRecordKind::ProjectEvaluationStarted => enc!(
ev::ProjectEvaluationStartedEvent,
ev::write_project_evaluation_started
),
BinaryLogRecordKind::ProjectEvaluationFinished => enc!(
ev::ProjectEvaluationFinishedEvent,
ev::write_project_evaluation_finished
),
BinaryLogRecordKind::PropertyReassignment => enc!(
ev::PropertyReassignmentEvent,
ev::write_property_reassignment
),
BinaryLogRecordKind::UninitializedPropertyRead => enc!(
ev::UninitializedPropertyReadEvent,
ev::write_uninitialized_property_read
),
BinaryLogRecordKind::PropertyInitialValueSet => enc!(
ev::PropertyInitialValueSetEvent,
ev::write_property_initial_value_set
),
BinaryLogRecordKind::EnvironmentVariableRead => enc!(
ev::EnvironmentVariableReadEvent,
ev::write_environment_variable_read
),
BinaryLogRecordKind::ResponseFileUsed => {
enc!(ev::ResponseFileUsedEvent, ev::write_response_file_used)
}
BinaryLogRecordKind::AssemblyLoad => enc!(ev::AssemblyLoadEvent, ev::write_assembly_load),
BinaryLogRecordKind::ProjectImported => {
enc!(ev::ProjectImportedEvent, ev::write_project_imported)
}
BinaryLogRecordKind::BuildCheckMessage => {
enc!(ev::BuildCheckMessageEvent, ev::write_build_message)
}
BinaryLogRecordKind::BuildCheckWarning => {
enc!(ev::BuildCheckWarningEvent, ev::write_build_warning)
}
BinaryLogRecordKind::BuildCheckError => {
enc!(ev::BuildCheckErrorEvent, ev::write_build_error)
}
BinaryLogRecordKind::BuildCheckTracing => {
enc!(ev::BuildCheckTracingEvent, ev::write_build_check_tracing)
}
BinaryLogRecordKind::BuildCheckAcquisition => enc!(
ev::BuildCheckAcquisitionEvent,
ev::write_build_check_acquisition
),
BinaryLogRecordKind::BuildSubmissionStarted => enc!(
ev::BuildSubmissionStartedEvent,
ev::write_build_submission_started
),
BinaryLogRecordKind::BuildCanceled => {
enc!(ev::BuildCanceledEvent, ev::write_build_canceled)
}
BinaryLogRecordKind::EndOfFile
| BinaryLogRecordKind::String
| BinaryLogRecordKind::NameValueList
| BinaryLogRecordKind::ProjectImportArchive => {
return Err(MuninError::InvalidFormat(format!(
"cannot encode auxiliary record kind: {:?}",
kind
)));
}
}
Ok(buf)
}