5640896332

pending completion

Build # 5640896332

Build Type

push

github

Committed by

web-flow

Commit Message

chore: Remove `Schema::identifier` (#1776)

Run Details

1574 of 1574 new or added lines in 102 files covered. (100.0%)

42145 of 54025 relevant lines covered (78.01%)

17469.9 hits per line

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

70.56

/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 { old } => {
                    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 { new } => {
                    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 { old, new } => {
                    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(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;	42✔
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(	9✔
68	cache_manager: &dyn RwCacheManager,	9✔
69	labels: Labels,	9✔
70	schema: SchemaWithIndex,	9✔
71	connections: &HashSet<String>,	9✔
72	write_options: CacheWriteOptions,	9✔
73	) -> Result<Box<dyn RwCache>, CacheError> {	9✔
74	match cache_manager.open_rw_cache(labels.clone(), write_options)? {	9✔
75	Some(cache) => {	×
76	debug_assert!(cache.get_schema() == &schema);	×
77	Ok(cache)	×
78	}
79	None => {
80	let cache = cache_manager.create_cache(	9✔
81	labels,	9✔
82	schema.0,	9✔
83	schema.1,	9✔
84	connections,	9✔
85	write_options,	9✔
86	)?;	9✔
87	Ok(cache)	9✔
88	}
89	}
90	}	9✔
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 {	44✔
100	let next_op = std::pin::pin!(log_reader.next_op());	44✔
101	match select(cancel, next_op).await {	44✔
102	Either::Left(_) => break,	6✔
103	Either::Right((op, c)) => {	38✔
104	let op = match op {	38✔
105	Ok(op) => op,	38✔
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;	38✔
117	if sender.send(op).await.is_err() {	38✔
118	debug!("Stop reading log because receiver is dropped");	×
119	break;	×
120	}	38✔
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() {	66✔
151	match op {	57✔
152	ExecutorOperation::Op { op } => match op {	24✔
153	Operation::Delete { old } => {	3✔
154	if let Some(meta) = cache.delete(&old)? {	3✔
155	if let Some((endpoint_name, operations_sender)) = operations_sender.as_ref()	3✔
156	{	3✔
157	let operation = types_helper::map_delete_operation(	3✔
158	endpoint_name.clone(),	3✔
159	CacheRecord::new(meta.id, meta.version, old),	3✔
160	);	3✔
161	send_and_log_error(operations_sender, operation);	3✔
162	}	3✔
163	}	×
164	let mut labels = cache.labels().clone();	3✔
165	labels.push(OPERATION_TYPE_LABEL, "delete");	3✔
166	labels.push(SNAPSHOTTING_LABEL, snapshotting_str(snapshotting));	3✔
167	increment_counter!(CACHE_OPERATION_COUNTER_NAME, labels);	3✔
168	}
169	Operation::Insert { new } => {	21✔
170	let result = cache.insert(&new)?;	21✔
171	let mut labels = cache.labels().clone();	21✔
172	labels.push(OPERATION_TYPE_LABEL, "insert");	21✔
173	labels.push(SNAPSHOTTING_LABEL, snapshotting_str(snapshotting));	21✔
174	increment_counter!(CACHE_OPERATION_COUNTER_NAME, labels);	21✔
175
176	if let Some((endpoint_name, operations_sender)) = operations_sender.as_ref() {	21✔
177	send_upsert_result(	21✔
178	endpoint_name,	21✔
179	operations_sender,	21✔
180	result,	21✔
181	&schema,	21✔
182	None,	21✔
183	new,	21✔
184	);	21✔
185	}	21✔
186	}
187	Operation::Update { old, new } => {	×
188	let upsert_result = cache.update(&old, &new)?;	×
189	let mut labels = cache.labels().clone();	×
190	labels.push(OPERATION_TYPE_LABEL, "update");	×
191	labels.push(SNAPSHOTTING_LABEL, snapshotting_str(snapshotting));	×
192	increment_counter!(CACHE_OPERATION_COUNTER_NAME, labels);	×
193
194	if let Some((endpoint_name, operations_sender)) = operations_sender.as_ref() {	×
195	send_upsert_result(	×
196	endpoint_name,	×
197	operations_sender,	×
198	upsert_result,	×
199	&schema,	×
200	Some(old),	×
201	new,	×
202	);	×
203	}	×
204	}
205	},
206	ExecutorOperation::Commit { epoch } => {	15✔
207	cache.set_metadata(pos)?;	15✔
208	cache.commit()?;	15✔
209	if let Ok(duration) = epoch.decision_instant.elapsed() {	15✔
210	histogram!(	15✔
211	DATA_LATENCY_HISTOGRAM_NAME,
212	duration,	×
213	cache.labels().clone()	×
214	);
215	}	×
216	}
217	ExecutorOperation::SnapshottingDone { connection_name } => {	18✔
218	cache.set_metadata(pos)?;	18✔
219	cache.set_connection_snapshotting_done(&connection_name)?;	18✔
220	cache.commit()?;	18✔
221	snapshotting = !cache.is_snapshotting_done()?;	18✔
222	}
223	ExecutorOperation::Terminate => {
224	break;	×
225	}
226	}
227	}
228
229	Ok(())	9✔
230	}	9✔
231
232	fn send_upsert_result(	21✔
233	endpoint_name: &str,	21✔
234	operations_sender: &Sender<GrpcOperation>,	21✔
235	upsert_result: UpsertResult,	21✔
236	schema: &Schema,	21✔
237	old: Option<Record>,	21✔
238	new: Record,	21✔
239	) {	21✔
240	match upsert_result {	21✔
241	UpsertResult::Inserted { meta } => {	21✔
242	let op = types_helper::map_insert_operation(	21✔
243	endpoint_name.to_string(),	21✔
244	CacheRecord::new(meta.id, meta.version, new),	21✔
245	);	21✔
246	send_and_log_error(operations_sender, op);	21✔
247	}	21✔
248	UpsertResult::Updated { old_meta, new_meta } => {	×
249	// If `old` is `None`, it means `Updated` comes from `Insert` operation.	×
250	// 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.	×
251	// So we create the old record with only the fields in the primary index, cloned from `new`.	×
252	let old = old.unwrap_or_else(\|\| {	×
253	let mut record = Record::new(vec![Field::Null; new.values.len()]);	×
254	for index in schema.primary_index.iter() {	×
255	record.values[index] = new.values[index].clone();	×
256	}	×
257	record	×
258	});	×
259	let op = types_helper::map_update_operation(	×
260	endpoint_name.to_string(),	×
261	CacheRecord::new(old_meta.id, old_meta.version, old),	×
262	CacheRecord::new(new_meta.id, new_meta.version, new),	×
263	);	×
264	send_and_log_error(operations_sender, op);	×
265	}	×
266	UpsertResult::Ignored => {}	×
267	}
268	}	21✔
269
270	fn send_and_log_error<T: Send + Sync + 'static>(sender: &Sender<T>, msg: T) {
271	if let Err(e) = sender.send(msg) {	24✔
272	error!("Failed to send broadcast message: {}", e);	×
273	}	24✔
274	}	24✔
275
276	fn snapshotting_str(snapshotting: bool) -> &'static str {	24✔
277	if snapshotting {	24✔
278	"true"	18✔
279	} else {
280	"false"	6✔
281	}
282	}	24✔

getdozer / dozer / 5640896332

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