5609791039

pending completion

Build # 5609791039

Build Type

push

github

Committed by

web-flow

Commit Message

refactor: Make config helper public (#1775)

Run Details

42723 of 55349 relevant lines covered (77.19%)

30697.34 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

57.8

/dozer-api/src/cache_builder/mod.rs

use std::collections::HashSet;
use std::time::Duration;

use crate::grpc::types_helper;
use dozer_cache::dozer_log::reader::{LogReader, LogReaderBuilder};
use dozer_cache::{
    cache::{CacheRecord, CacheWriteOptions, RwCache, RwCacheManager, UpsertResult},
    errors::CacheError,
};
use dozer_types::epoch::ExecutorOperation;
use dozer_types::indicatif::MultiProgress;
use dozer_types::labels::Labels;
use dozer_types::log::debug;
use dozer_types::types::SchemaWithIndex;
use dozer_types::{
    grpc_types::types::Operation as GrpcOperation,
    log::error,
    types::{Field, Operation, Record, Schema},
};
use futures_util::stream::FuturesUnordered;
use futures_util::{
    future::{select, Either},
    Future,
};
use metrics::{describe_counter, describe_histogram, histogram, increment_counter};
use tokio::sync::broadcast::Sender;
use tokio::sync::mpsc;
use tokio_stream::StreamExt;

pub async fn build_cache(
    cache: Box<dyn RwCache>,
    cancel: impl Future<Output = ()> + Unpin + Send + 'static,
    log_reader_builder: LogReaderBuilder,
    operations_sender: Option<(String, Sender<GrpcOperation>)>,
    multi_pb: Option<MultiProgress>,
) -> Result<(), CacheError> {
    // Create log reader.
    let pos = cache.get_metadata()?.unwrap_or(0);
    debug!(
        "Starting log reader {} from position {pos}",
        log_reader_builder.options.endpoint
    );
    let log_reader = log_reader_builder.build(pos, multi_pb);

    // Spawn tasks
    let mut futures = FuturesUnordered::new();
    let (sender, receiver) = mpsc::channel(1);
    futures.push(tokio::spawn(async move {
        read_log_task(cancel, log_reader, sender).await;
        Ok(())
    }));
    futures.push({
        tokio::task::spawn_blocking(|| build_cache_task(cache, receiver, operations_sender))
    });

    while let Some(result) = futures.next().await {
        match result {
            Ok(Ok(())) => (),
            Ok(Err(e)) => return Err(e),
            Err(e) => return Err(CacheError::InternalThreadPanic(e)),
        }
    }

    Ok(())
}

pub fn open_or_create_cache(
    cache_manager: &dyn RwCacheManager,
    labels: Labels,
    schema: SchemaWithIndex,
    connections: &HashSet<String>,
    write_options: CacheWriteOptions,
) -> Result<Box<dyn RwCache>, CacheError> {
    match cache_manager.open_rw_cache(labels.clone(), write_options)? {
        Some(cache) => {
            debug_assert!(cache.get_schema() == &schema);
            Ok(cache)
        }
        None => {
            let cache = cache_manager.create_cache(
                labels,
                schema.0,
                schema.1,
                connections,
                write_options,
            )?;
            Ok(cache)
        }
    }
}

const READ_LOG_RETRY_INTERVAL: Duration = Duration::from_secs(1);

async fn read_log_task(
    mut cancel: impl Future<Output = ()> + Unpin + Send + 'static,
    mut log_reader: LogReader,
    sender: mpsc::Sender<(ExecutorOperation, u64)>,
) {
    loop {
        let next_op = std::pin::pin!(log_reader.next_op());
        match select(cancel, next_op).await {
            Either::Left(_) => break,
            Either::Right((op, c)) => {
                let op = match op {
                    Ok(op) => op,
                    Err(e) => {
                        error!(
                            "Failed to read log: {e}, retrying after {READ_LOG_RETRY_INTERVAL:?}"
                        );
                        tokio::time::sleep(READ_LOG_RETRY_INTERVAL).await;
                        cancel = c;
                        continue;
                    }
                };

                cancel = c;
                if sender.send(op).await.is_err() {
                    debug!("Stop reading log because receiver is dropped");
                    break;
                }
            }
        }
    }
}

fn build_cache_task(
    mut cache: Box<dyn RwCache>,
    mut receiver: mpsc::Receiver<(ExecutorOperation, u64)>,
    operations_sender: Option<(String, Sender<GrpcOperation>)>,
) -> Result<(), CacheError> {
    let schema = cache.get_schema().0.clone();

    const CACHE_OPERATION_COUNTER_NAME: &str = "cache_operation";
    describe_counter!(
        CACHE_OPERATION_COUNTER_NAME,
        "Number of message processed by cache builder"
    );

    const DATA_LATENCY_HISTOGRAM_NAME: &str = "data_latency";
    describe_histogram!(
        DATA_LATENCY_HISTOGRAM_NAME,
        "End-to-end data latency in seconds"
    );

    const OPERATION_TYPE_LABEL: &str = "operation_type";
    const SNAPSHOTTING_LABEL: &str = "snapshotting";

    let mut snapshotting = !cache.is_snapshotting_done()?;

    while let Some((op, pos)) = receiver.blocking_recv() {
        match op {
            ExecutorOperation::Op { op } => match op {
                Operation::Delete { mut old } => {
                    old.schema_id = schema.identifier;
                    if let Some(meta) = cache.delete(&old)? {
                        if let Some((endpoint_name, operations_sender)) = operations_sender.as_ref()
                        {
                            let operation = types_helper::map_delete_operation(
                                endpoint_name.clone(),
                                CacheRecord::new(meta.id, meta.version, old),
                            );
                            send_and_log_error(operations_sender, operation);
                        }
                    }
                    let mut labels = cache.labels().clone();
                    labels.push(OPERATION_TYPE_LABEL, "delete");
                    labels.push(SNAPSHOTTING_LABEL, snapshotting_str(snapshotting));
                    increment_counter!(CACHE_OPERATION_COUNTER_NAME, labels);
                }
                Operation::Insert { mut new } => {
                    new.schema_id = schema.identifier;
                    let result = cache.insert(&new)?;
                    let mut labels = cache.labels().clone();
                    labels.push(OPERATION_TYPE_LABEL, "insert");
                    labels.push(SNAPSHOTTING_LABEL, snapshotting_str(snapshotting));
                    increment_counter!(CACHE_OPERATION_COUNTER_NAME, labels);

                    if let Some((endpoint_name, operations_sender)) = operations_sender.as_ref() {
                        send_upsert_result(
                            endpoint_name,
                            operations_sender,
                            result,
                            &schema,
                            None,
                            new,
                        );
                    }
                }
                Operation::Update { mut old, mut new } => {
                    old.schema_id = schema.identifier;
                    new.schema_id = schema.identifier;
                    let upsert_result = cache.update(&old, &new)?;
                    let mut labels = cache.labels().clone();
                    labels.push(OPERATION_TYPE_LABEL, "update");
                    labels.push(SNAPSHOTTING_LABEL, snapshotting_str(snapshotting));
                    increment_counter!(CACHE_OPERATION_COUNTER_NAME, labels);

                    if let Some((endpoint_name, operations_sender)) = operations_sender.as_ref() {
                        send_upsert_result(
                            endpoint_name,
                            operations_sender,
                            upsert_result,
                            &schema,
                            Some(old),
                            new,
                        );
                    }
                }
            },
            ExecutorOperation::Commit { epoch } => {
                cache.set_metadata(pos)?;
                cache.commit()?;
                if let Ok(duration) = epoch.decision_instant.elapsed() {
                    histogram!(
                        DATA_LATENCY_HISTOGRAM_NAME,
                        duration,
                        cache.labels().clone()
                    );
                }
            }
            ExecutorOperation::SnapshottingDone { connection_name } => {
                cache.set_metadata(pos)?;
                cache.set_connection_snapshotting_done(&connection_name)?;
                cache.commit()?;
                snapshotting = !cache.is_snapshotting_done()?;
            }
            ExecutorOperation::Terminate => {
                break;
            }
        }
    }

    Ok(())
}

fn send_upsert_result(
    endpoint_name: &str,
    operations_sender: &Sender<GrpcOperation>,
    upsert_result: UpsertResult,
    schema: &Schema,
    old: Option<Record>,
    new: Record,
) {
    match upsert_result {
        UpsertResult::Inserted { meta } => {
            let op = types_helper::map_insert_operation(
                endpoint_name.to_string(),
                CacheRecord::new(meta.id, meta.version, new),
            );
            send_and_log_error(operations_sender, op);
        }
        UpsertResult::Updated { old_meta, new_meta } => {
            // If `old` is `None`, it means `Updated` comes from `Insert` operation.
            // In this case, we can't get the full old record, but the fields in the primary index must be the same with the new record.
            // So we create the old record with only the fields in the primary index, cloned from `new`.
            let old = old.unwrap_or_else(|| {
                let mut record = Record::new(new.schema_id, vec![Field::Null; new.values.len()]);
                for index in schema.primary_index.iter() {
                    record.values[*index] = new.values[*index].clone();
                }
                record
            });
            let op = types_helper::map_update_operation(
                endpoint_name.to_string(),
                CacheRecord::new(old_meta.id, old_meta.version, old),
                CacheRecord::new(new_meta.id, new_meta.version, new),
            );
            send_and_log_error(operations_sender, op);
        }
        UpsertResult::Ignored => {}
    }
}

fn send_and_log_error<T: Send + Sync + 'static>(sender: &Sender<T>, msg: T) {
    if let Err(e) = sender.send(msg) {
        error!("Failed to send broadcast message: {}", e);
    }
}

fn snapshotting_str(snapshotting: bool) -> &'static str {
    if snapshotting {
        "true"
    } else {
        "false"
    }
}

1	use std::collections::HashSet;
2	use std::time::Duration;
3
4	use crate::grpc::types_helper;
5	use dozer_cache::dozer_log::reader::{LogReader, LogReaderBuilder};
6	use dozer_cache::{
7	cache::{CacheRecord, CacheWriteOptions, RwCache, RwCacheManager, UpsertResult},
8	errors::CacheError,
9	};
10	use dozer_types::epoch::ExecutorOperation;
11	use dozer_types::indicatif::MultiProgress;
12	use dozer_types::labels::Labels;
13	use dozer_types::log::debug;
14	use dozer_types::types::SchemaWithIndex;
15	use dozer_types::{
16	grpc_types::types::Operation as GrpcOperation,
17	log::error,
18	types::{Field, Operation, Record, Schema},
19	};
20	use futures_util::stream::FuturesUnordered;
21	use futures_util::{
22	future::{select, Either},
23	Future,
24	};
25	use metrics::{describe_counter, describe_histogram, histogram, increment_counter};
26	use tokio::sync::broadcast::Sender;
27	use tokio::sync::mpsc;
28	use tokio_stream::StreamExt;
29
30	pub async fn build_cache(	6✔
31	cache: Box<dyn RwCache>,	6✔
32	cancel: impl Future<Output = ()> + Unpin + Send + 'static,	6✔
33	log_reader_builder: LogReaderBuilder,	6✔
34	operations_sender: Option<(String, Sender<GrpcOperation>)>,	6✔
35	multi_pb: Option<MultiProgress>,	6✔
36	) -> Result<(), CacheError> {	6✔
37	// Create log reader.
38	let pos = cache.get_metadata()?.unwrap_or(0);	6✔
39	debug!(
40	"Starting log reader {} from position {pos}",	×
41	log_reader_builder.options.endpoint
42	);
43	let log_reader = log_reader_builder.build(pos, multi_pb);	6✔
44		6✔
45	// Spawn tasks	6✔
46	let mut futures = FuturesUnordered::new();	6✔
47	let (sender, receiver) = mpsc::channel(1);	6✔
48	futures.push(tokio::spawn(async move {	6✔
49	read_log_task(cancel, log_reader, sender).await;	36✔
50	Ok(())	6✔
51	}));	6✔
52	futures.push({	6✔
53	tokio::task::spawn_blocking(\|\| build_cache_task(cache, receiver, operations_sender))	6✔
54	});	6✔
55
56	while let Some(result) = futures.next().await {	18✔
57	match result {	12✔
58	Ok(Ok(())) => (),	12✔
59	Ok(Err(e)) => return Err(e),	×
60	Err(e) => return Err(CacheError::InternalThreadPanic(e)),	×
61	}
62	}
63
64	Ok(())	6✔
65	}	6✔
66
67	pub fn open_or_create_cache(	18✔
68	cache_manager: &dyn RwCacheManager,	18✔
69	labels: Labels,	18✔
70	schema: SchemaWithIndex,	18✔
71	connections: &HashSet<String>,	18✔
72	write_options: CacheWriteOptions,	18✔
73	) -> Result<Box<dyn RwCache>, CacheError> {	18✔
74	match cache_manager.open_rw_cache(labels.clone(), write_options)? {	18✔
75	Some(cache) => {	×
76	debug_assert!(cache.get_schema() == &schema);	×
77	Ok(cache)	×
78	}
79	None => {
80	let cache = cache_manager.create_cache(	18✔
81	labels,	18✔
82	schema.0,	18✔
83	schema.1,	18✔
84	connections,	18✔
85	write_options,	18✔
86	)?;	18✔
87	Ok(cache)	18✔
88	}
89	}
90	}	18✔
91
92	const READ_LOG_RETRY_INTERVAL: Duration = Duration::from_secs(1);
93
94	async fn read_log_task(	6✔
95	mut cancel: impl Future<Output = ()> + Unpin + Send + 'static,	6✔
96	mut log_reader: LogReader,	6✔
97	sender: mpsc::Sender<(ExecutorOperation, u64)>,	6✔
98	) {	6✔
99	loop {	40✔
100	let next_op = std::pin::pin!(log_reader.next_op());	40✔
101	match select(cancel, next_op).await {	40✔
102	Either::Left(_) => break,	6✔
103	Either::Right((op, c)) => {	34✔
104	let op = match op {	34✔
105	Ok(op) => op,	34✔
106	Err(e) => {	×
107	error!(
108	"Failed to read log: {e}, retrying after {READ_LOG_RETRY_INTERVAL:?}"	×
109	);
110	tokio::time::sleep(READ_LOG_RETRY_INTERVAL).await;	×
111	cancel = c;	×
112	continue;	×
113	}
114	};
115
116	cancel = c;	34✔
117	if sender.send(op).await.is_err() {	34✔
118	debug!("Stop reading log because receiver is dropped");	×
119	break;	×
120	}	34✔
121	}
122	}
123	}
124	}	6✔
125
126	fn build_cache_task(	9✔
127	mut cache: Box<dyn RwCache>,	9✔
128	mut receiver: mpsc::Receiver<(ExecutorOperation, u64)>,	9✔
129	operations_sender: Option<(String, Sender<GrpcOperation>)>,	9✔
130	) -> Result<(), CacheError> {	9✔
131	let schema = cache.get_schema().0.clone();	9✔
132		9✔
133	const CACHE_OPERATION_COUNTER_NAME: &str = "cache_operation";	9✔
134	describe_counter!(	9✔
135	CACHE_OPERATION_COUNTER_NAME,	×
136	"Number of message processed by cache builder"	×
137	);
138
139	const DATA_LATENCY_HISTOGRAM_NAME: &str = "data_latency";
140	describe_histogram!(	9✔
141	DATA_LATENCY_HISTOGRAM_NAME,	×
142	"End-to-end data latency in seconds"	×
143	);
144
145	const OPERATION_TYPE_LABEL: &str = "operation_type";	×
146	const SNAPSHOTTING_LABEL: &str = "snapshotting";	×
147		×
148	let mut snapshotting = !cache.is_snapshotting_done()?;	9✔
149		×
150	while let Some((op, pos)) = receiver.blocking_recv() {	60✔
151	match op {	51✔
152	ExecutorOperation::Op { op } => match op {	18✔
153	Operation::Delete { mut old } => {	×
154	old.schema_id = schema.identifier;	×
155	if let Some(meta) = cache.delete(&old)? {	×
156	if let Some((endpoint_name, operations_sender)) = operations_sender.as_ref()	×
157	{	×
158	let operation = types_helper::map_delete_operation(	×
159	endpoint_name.clone(),	×
160	CacheRecord::new(meta.id, meta.version, old),	×
161	);	×
162	send_and_log_error(operations_sender, operation);	×
163	}	×
164	}	×
165	let mut labels = cache.labels().clone();	×
166	labels.push(OPERATION_TYPE_LABEL, "delete");	×
167	labels.push(SNAPSHOTTING_LABEL, snapshotting_str(snapshotting));	×
168	increment_counter!(CACHE_OPERATION_COUNTER_NAME, labels);	×
169	}	×
170	Operation::Insert { mut new } => {	18✔
171	new.schema_id = schema.identifier;	18✔
172	let result = cache.insert(&new)?;	18✔
173	let mut labels = cache.labels().clone();	18✔
174	labels.push(OPERATION_TYPE_LABEL, "insert");	18✔
175	labels.push(SNAPSHOTTING_LABEL, snapshotting_str(snapshotting));	18✔
176	increment_counter!(CACHE_OPERATION_COUNTER_NAME, labels);	18✔
177		×
178	if let Some((endpoint_name, operations_sender)) = operations_sender.as_ref() {	18✔
179	send_upsert_result(	18✔
180	endpoint_name,	18✔
181	operations_sender,	18✔
182	result,	18✔
183	&schema,	18✔
184	None,	18✔
185	new,	18✔
186	);	18✔
187	}	18✔
188	}	×
189	Operation::Update { mut old, mut new } => {	×
190	old.schema_id = schema.identifier;	×
191	new.schema_id = schema.identifier;	×
192	let upsert_result = cache.update(&old, &new)?;	×
193	let mut labels = cache.labels().clone();	×
194	labels.push(OPERATION_TYPE_LABEL, "update");	×
195	labels.push(SNAPSHOTTING_LABEL, snapshotting_str(snapshotting));	×
196	increment_counter!(CACHE_OPERATION_COUNTER_NAME, labels);	×
197		×
198	if let Some((endpoint_name, operations_sender)) = operations_sender.as_ref() {	×
199	send_upsert_result(	×
200	endpoint_name,	×
201	operations_sender,	×
202	upsert_result,	×
203	&schema,	×
204	Some(old),	×
205	new,	×
206	);	×
207	}	×
208	}	×
209	},	×
210	ExecutorOperation::Commit { epoch } => {	15✔
211	cache.set_metadata(pos)?;	15✔
212	cache.commit()?;	15✔
213	if let Ok(duration) = epoch.decision_instant.elapsed() {	15✔
214	histogram!(	15✔
215	DATA_LATENCY_HISTOGRAM_NAME,	×
216	duration,	×
217	cache.labels().clone()	×
218	);
219	}	×
220	}
221	ExecutorOperation::SnapshottingDone { connection_name } => {	18✔
222	cache.set_metadata(pos)?;	18✔
223	cache.set_connection_snapshotting_done(&connection_name)?;	18✔
224	cache.commit()?;	18✔
225	snapshotting = !cache.is_snapshotting_done()?;	18✔
226	}
227	ExecutorOperation::Terminate => {	×
228	break;	×
229	}	×
230	}	×
231	}	×
232		×
233	Ok(())	9✔
234	}	9✔
235		×
236	fn send_upsert_result(	18✔
237	endpoint_name: &str,	18✔
238	operations_sender: &Sender<GrpcOperation>,	18✔
239	upsert_result: UpsertResult,	18✔
240	schema: &Schema,	18✔
241	old: Option<Record>,	18✔
242	new: Record,	18✔
243	) {	18✔
244	match upsert_result {	18✔
245	UpsertResult::Inserted { meta } => {	18✔
246	let op = types_helper::map_insert_operation(	18✔
247	endpoint_name.to_string(),	18✔
248	CacheRecord::new(meta.id, meta.version, new),	18✔
249	);	18✔
250	send_and_log_error(operations_sender, op);	18✔
251	}	18✔
252	UpsertResult::Updated { old_meta, new_meta } => {	×
253	// If `old` is `None`, it means `Updated` comes from `Insert` operation.	×
254	// In this case, we can't get the full old record, but the fields in the primary index must be the same with the new record.	×
255	// So we create the old record with only the fields in the primary index, cloned from `new`.	×
256	let old = old.unwrap_or_else(\|\| {	×
257	let mut record = Record::new(new.schema_id, vec![Field::Null; new.values.len()]);	×
258	for index in schema.primary_index.iter() {	×
259	record.values[index] = new.values[index].clone();	×
260	}	×
261	record	×
262	});	×
263	let op = types_helper::map_update_operation(	×
264	endpoint_name.to_string(),	×
265	CacheRecord::new(old_meta.id, old_meta.version, old),	×
266	CacheRecord::new(new_meta.id, new_meta.version, new),	×
267	);	×
268	send_and_log_error(operations_sender, op);	×
269	}	×
270	UpsertResult::Ignored => {}	×
271	}
272	}	18✔
273
274	fn send_and_log_error<T: Send + Sync + 'static>(sender: &Sender<T>, msg: T) {
275	if let Err(e) = sender.send(msg) {	18✔
276	error!("Failed to send broadcast message: {}", e);	×
277	}	18✔
278	}	18✔
279
280	fn snapshotting_str(snapshotting: bool) -> &'static str {	18✔
281	if snapshotting {	18✔
282	"true"	12✔
283	} else {
284	"false"	6✔
285	}
286	}	18✔

getdozer / dozer / 5609791039

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous