filip/Stalwart
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Database schema optimization - part 4

Browse Source
This commit is contained in:
mdecimus
2025-11-02 16:01:05 +01:00
parent 8e1316895b
commit 3d20a82354
13 changed files with 794 additions and 419 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
crates/common/src/config/telemetry.rs
View File

@@ -99,6 +99,7 @@ pub struct WebhookTracer {
#[cfg(feature = "enterprise")]
pub struct StoreTracer {
pub store: store::Store,
pub indexed: bool,
}
// SPDX-SnippetEnd
@@ -538,6 +539,9 @@ impl Tracers {
lossy: false,
typ: TelemetrySubscriberType::StoreTracer(StoreTracer {
store: store.clone(),
indexed: config
.property_or_default("tracing.history.indexed", "false")
.unwrap_or(false),
}),
};

178
crates/common/src/telemetry/tracers/index.rs
View File

@@ -1,178 +0,0 @@
fn build_index() {
let mut queue_ids = AHashSet::new();
let mut values = AHashSet::new();
for event in events.iter().chain([span, &event]) {
for (key, value) in &event.keys {
match (key, value) {
(Key::QueueId, Value::UInt(queue_id)) => {
queue_ids.insert(*queue_id);
}
(Key::From | Key::To | Key::Domain | Key::Hostname, Value::String(address)) => {
values.insert(address.clone());
}
(Key::To, Value::Array(value)) => {
for value in value {
if let Value::String(address) = value {
values.insert(address.clone());
}
}
}
(Key::RemoteIp, Value::Ipv4(ip)) => {
values.insert(ip.to_string().into());
}
(Key::RemoteIp, Value::Ipv6(ip)) => {
values.insert(ip.to_string().into());
}
_ => {}
}
}
}
// Build index
batch.set(
ValueClass::Telemetry(TelemetryClass::Index {
span_id,
value: (span.inner.typ.code() as u16).to_be_bytes().to_vec(),
}),
vec![],
);
for queue_id in queue_ids {
batch.set(
ValueClass::Telemetry(TelemetryClass::Index {
span_id,
value: queue_id.to_be_bytes().to_vec(),
}),
vec![],
);
}
for value in values {
batch.set(
ValueClass::Telemetry(TelemetryClass::Index {
span_id,
value: value.as_bytes().to_vec(),
}),
vec![],
);
}
}
/*
enum SpanCollector {
Vec(Vec<u64>),
HashSet(AHashSet<u64>),
Empty,
}
impl SpanCollector {
fn new(num_params: usize) -> Self {
if num_params == 1 {
Self::Vec(Vec::new())
} else {
Self::HashSet(AHashSet::new())
}
}
fn insert(&mut self, span_id: u64) {
match self {
Self::Vec(vec) => vec.push(span_id),
Self::HashSet(set) => {
set.insert(span_id);
}
_ => unreachable!(),
}
}
fn into_vec(self) -> Vec<u64> {
match self {
Self::Vec(mut vec) => {
vec.sort_unstable_by(|a, b| b.cmp(a));
vec
}
Self::HashSet(set) => {
let mut vec: Vec<u64> = set.into_iter().collect();
vec.sort_unstable_by(|a, b| b.cmp(a));
vec
}
Self::Empty => Vec::new(),
}
}
fn intersect(&mut self, other_span: Self) -> bool {
match (self, other_span) {
(Self::HashSet(set), Self::HashSet(other_set)) => {
set.retain(|span_id| other_set.contains(span_id));
set.is_empty()
}
_ => unreachable!(),
}
}
}
let mut spans = SpanCollector::Empty;
let num_params = params.len();
let todo = "use FTS";
for (param_num, param) in params.iter().enumerate() {
let (value, exact_len) = match param {
TracingQuery::EventType(event) => (
(event.code() as u16).to_be_bytes().to_vec(),
std::mem::size_of::<u16>() + U64_LEN,
),
TracingQuery::QueueId(id) => (
id.to_be_bytes().to_vec(),
std::mem::size_of::<u64>() + U64_LEN,
),
TracingQuery::Keywords(value) => {
if let Some(value) = value.strip_prefix('"').and_then(|v| v.strip_suffix('"')) {
(value.as_bytes().to_vec(), value.len() + U64_LEN)
} else {
(value.as_bytes().to_vec(), 0)
}
}
};
let mut param_spans = SpanCollector::new(num_params);
self.iterate(
IterateParams::new(
ValueKey::from(ValueClass::Telemetry(TelemetryClass::Index {
span_id: 0,
value: value.clone(),
})),
ValueKey::from(ValueClass::Telemetry(TelemetryClass::Index {
span_id: u64::MAX,
value,
})),
)
.no_values(),
|key, _| {
if exact_len == 0 || key.len() == exact_len {
let span_id = key
.deserialize_be_u64(key.len() - U64_LEN)
.caused_by(trc::location!())?;
if (from_span_id == 0 || span_id >= from_span_id)
&& (to_span_id == 0 || span_id <= to_span_id)
{
param_spans.insert(span_id);
}
}
Ok(true)
},
)
.await
.caused_by(trc::location!())?;
if param_num == 0 {
spans = param_spans;
} else if spans.intersect(param_spans) {
return Ok(Vec::new());
}
}
Ok(spans.into_vec())
*/

158
crates/common/src/telemetry/tracers/store.rs
View File

@@ -10,9 +10,11 @@
use crate::config::telemetry::StoreTracer;
use ahash::{AHashMap, AHashSet};
use nlp::language::Language;
use std::{future::Future, time::Duration};
use store::{
Deserialize, Store, ValueKey,
Deserialize, SearchStore, Store, ValueKey,
search::{IndexDocument, SearchField, SearchFilter, SearchQuery, TracingSearchField},
write::{BatchBuilder, SearchIndex, TaskQueueClass, TelemetryClass, ValueClass, now},
};
use trc::{
@@ -51,27 +53,30 @@ pub(crate) fn spawn_store_tracer(builder: SubscriberBuilder, settings: StoreTrac
.any(|(k, v)| matches!((k, v), (Key::QueueId, Value::UInt(_))))
{
// Serialize events
batch
.set(
ValueClass::Telemetry(TelemetryClass::Span { span_id }),
serialize_events(
[span.as_ref()]
.into_iter()
.chain(events.iter().map(|event| event.as_ref()))
.chain([event.as_ref()].into_iter()),
events.len() + 2,
),
)
.with_account_id((span_id >> 32) as u32) // TODO: This is hacky, improve
.with_document(span_id as u32)
.set(
ValueClass::TaskQueue(TaskQueueClass::UpdateIndex {
due: now,
index: SearchIndex::TracingSpan,
is_insert: true,
}),
vec![],
);
batch.set(
ValueClass::Telemetry(TelemetryClass::Span { span_id }),
serialize_events(
[span.as_ref()]
.into_iter()
.chain(events.iter().map(|event| event.as_ref()))
.chain([event.as_ref()].into_iter()),
events.len() + 2,
),
);
if settings.indexed {
batch
.with_account_id((span_id >> 32) as u32) // TODO: This is hacky, improve
.with_document(span_id as u32)
.set(
ValueClass::TaskQueue(TaskQueueClass::UpdateIndex {
due: now,
index: SearchIndex::TracingSpan,
is_insert: true,
}),
vec![],
);
}
}
}
}
@@ -86,12 +91,6 @@ pub(crate) fn spawn_store_tracer(builder: SubscriberBuilder, settings: StoreTrac
});
}
pub enum TracingQuery {
EventType(EventType),
QueueId(u64),
Keywords(String),
}
pub trait TracingStore: Sync + Send {
fn get_span(
&self,
@@ -101,13 +100,11 @@ pub trait TracingStore: Sync + Send {
&self,
span_id: u64,
) -> impl Future<Output = trc::Result<Option<Vec<u8>>>> + Send;
fn query_spans(
fn purge_spans(
&self,
params: &[TracingQuery],
from_span_id: u64,
to_span_id: u64,
) -> impl Future<Output = trc::Result<Vec<u64>>> + Send;
fn purge_spans(&self, period: Duration) -> impl Future<Output = trc::Result<()>> + Send;
period: Duration,
search_store: Option<&SearchStore>,
) -> impl Future<Output = trc::Result<()>> + Send;
}
impl TracingStore for Store {
@@ -129,16 +126,11 @@ impl TracingStore for Store {
.map(|span| span.map(|span| span.0))
}
async fn query_spans(
async fn purge_spans(
&self,
params: &[TracingQuery],
from_span_id: u64,
to_span_id: u64,
) -> trc::Result<Vec<u64>> {
todo!()
}
async fn purge_spans(&self, period: Duration) -> trc::Result<()> {
period: Duration,
search_store: Option<&SearchStore>,
) -> trc::Result<()> {
let until_span_id = SnowflakeIdGenerator::from_duration(period).ok_or_else(|| {
trc::StoreEvent::UnexpectedError
.caused_by(trc::location!())
@@ -154,7 +146,15 @@ impl TracingStore for Store {
.await
.caused_by(trc::location!())?;
let todo = "delete from index";
if let Some(search_store) = search_store {
search_store
.unindex(
SearchQuery::new(SearchIndex::TracingSpan)
.with_filter(SearchFilter::lt(SearchField::Id, until_span_id)),
)
.await
.caused_by(trc::location!())?;
}
Ok(())
}
@@ -240,3 +240,73 @@ impl Deserialize for RawSpan {
Ok(Self(bytes.to_vec()))
}
}
pub fn build_span_document(
span_id: u64,
events: Vec<Event<EventDetails>>,
index_fields: &AHashSet<SearchField>,
) -> IndexDocument {
let mut document = IndexDocument::with_default_language(Language::None);
document.index_unsigned(SearchField::Id, span_id);
for event in events {
for (idx, (key, value)) in event.keys.into_iter().enumerate() {
if idx == 0
&& (index_fields.is_empty()
|| index_fields.contains(&TracingSearchField::EventType.into()))
{
document.index_unsigned(TracingSearchField::EventType, event.inner.typ.code());
}
match (key, value) {
(Key::QueueId, Value::UInt(queue_id)) => {
if index_fields.is_empty()
|| index_fields.contains(&TracingSearchField::QueueId.into())
{
document.insert_keyword(TracingSearchField::QueueId, queue_id);
}
}
(Key::From | Key::To | Key::Domain | Key::Hostname, Value::String(address)) => {
if index_fields.is_empty()
|| index_fields.contains(&TracingSearchField::Address.into())
{
document.insert_keyword(TracingSearchField::Address, address.into_string());
}
}
(Key::To, Value::Array(value)) => {
if index_fields.is_empty()
|| index_fields.contains(&TracingSearchField::Address.into())
{
for value in value {
if let Value::String(address) = value {
document.insert_keyword(
TracingSearchField::Address,
address.into_string(),
);
}
}
}
}
(Key::RemoteIp, Value::Ipv4(ip)) => {
if index_fields.is_empty()
|| index_fields.contains(&TracingSearchField::RemoteIp.into())
{
document.insert_keyword(TracingSearchField::RemoteIp, ip.to_string());
}
}
(Key::RemoteIp, Value::Ipv6(ip)) => {
if index_fields.is_empty()
|| index_fields.contains(&TracingSearchField::RemoteIp.into())
{
document.insert_keyword(TracingSearchField::RemoteIp, ip.to_string());
}
}
_ => {}
}
}
}
document
}