5960771901

Committed 24 Aug 2023 07:28AM UTC coverage: 75.706% (-0.07%) from 75.778%

Build # 5960771901

Build Type

push

github

Committed by

web-flow

Commit Message

chore: Remove `LogOperation::Terminate` (#1907)

Run Details

46995 of 62076 relevant lines covered (75.71%)

89223.8 hits per line

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

68.14

/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::dozer_log::replication::LogOperation;
use dozer_cache::{
    cache::{CacheRecord, CacheWriteOptions, RwCache, RwCacheManager, UpsertResult},
    errors::CacheError,
};
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<(LogOperation, 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<(LogOperation, 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 {
            LogOperation::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,
                        );
                    }
                }
            },
            LogOperation::Commit { decision_instant } => {
                cache.set_metadata(pos)?;
                cache.commit()?;
                if let Ok(duration) = decision_instant.elapsed() {
                    histogram!(
                        DATA_LATENCY_HISTOGRAM_NAME,
                        duration,
                        cache.labels().clone()
                    );
                }
            }
            LogOperation::SnapshottingDone { connection_name } => {
                cache.set_metadata(pos)?;
                cache.set_connection_snapshotting_done(&connection_name)?;
                cache.commit()?;
                snapshotting = !cache.is_snapshotting_done()?;
            }
        }
    }

    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::dozer_log::replication::LogOperation;
7	use dozer_cache::{
8	cache::{CacheRecord, CacheWriteOptions, RwCache, RwCacheManager, UpsertResult},
9	errors::CacheError,
10	};
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(	18✔
31	cache: Box<dyn RwCache>,	18✔
32	cancel: impl Future<Output = ()> + Unpin + Send + 'static,	18✔
33	log_reader_builder: LogReaderBuilder,	18✔
34	operations_sender: Option<(String, Sender<GrpcOperation>)>,	18✔
35	multi_pb: Option<MultiProgress>,	18✔
36	) -> Result<(), CacheError> {	18✔
37	// Create log reader.
38	let pos = cache.get_metadata()?.unwrap_or(0);	18✔
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);	18✔
44		18✔
45	// Spawn tasks	18✔
46	let mut futures = FuturesUnordered::new();	18✔
47	let (sender, receiver) = mpsc::channel(1);	18✔
48	futures.push(tokio::spawn(async move {	18✔
49	read_log_task(cancel, log_reader, sender).await;	90✔
50	Ok(())	18✔
51	}));	18✔
52	futures.push({	18✔
53	tokio::task::spawn_blocking(\|\| build_cache_task(cache, receiver, operations_sender))	18✔
54	});	18✔
55
56	while let Some(result) = futures.next().await {	54✔
57	match result {	36✔
58	Ok(Ok(())) => (),	36✔
59	Ok(Err(e)) => return Err(e),	×
60	Err(e) => return Err(CacheError::InternalThreadPanic(e)),	×
61	}
62	}
63
64	Ok(())	18✔
65	}	18✔
66
67	pub fn open_or_create_cache(	96✔
68	cache_manager: &dyn RwCacheManager,	96✔
69	labels: Labels,	96✔
70	schema: SchemaWithIndex,	96✔
71	connections: &HashSet<String>,	96✔
72	write_options: CacheWriteOptions,	96✔
73	) -> Result<Box<dyn RwCache>, CacheError> {	96✔
74	match cache_manager.open_rw_cache(labels.clone(), write_options)? {	96✔
75	Some(cache) => {	×
76	debug_assert!(cache.get_schema() == &schema);	×
77	Ok(cache)	×
78	}
79	None => {
80	let cache = cache_manager.create_cache(	96✔
81	labels,	96✔
82	schema.0,	96✔
83	schema.1,	96✔
84	connections,	96✔
85	write_options,	96✔
86	)?;	96✔
87	Ok(cache)	96✔
88	}
89	}
90	}	96✔
91
92	const READ_LOG_RETRY_INTERVAL: Duration = Duration::from_secs(1);
93
94	async fn read_log_task(	18✔
95	mut cancel: impl Future<Output = ()> + Unpin + Send + 'static,	18✔
96	mut log_reader: LogReader,	18✔
97	sender: mpsc::Sender<(LogOperation, u64)>,	18✔
98	) {	18✔
99	loop {	93✔
100	let next_op = std::pin::pin!(log_reader.next_op());	93✔
101	match select(cancel, next_op).await {	93✔
102	Either::Left(_) => break,	18✔
103	Either::Right((op, c)) => {	75✔
104	let op = match op {	75✔
105	Ok(op) => op,	75✔
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;	75✔
117	if sender.send(op).await.is_err() {	75✔
118	debug!("Stop reading log because receiver is dropped");	×
119	break;	×
120	}	75✔
121	}
122	}
123	}
124	}	18✔
125
126	fn build_cache_task(	24✔
127	mut cache: Box<dyn RwCache>,	24✔
128	mut receiver: mpsc::Receiver<(LogOperation, u64)>,	24✔
129	operations_sender: Option<(String, Sender<GrpcOperation>)>,	24✔
130	) -> Result<(), CacheError> {	24✔
131	let schema = cache.get_schema().0.clone();	24✔
132		24✔
133	const CACHE_OPERATION_COUNTER_NAME: &str = "cache_operation";	24✔
134	describe_counter!(	24✔
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!(	24✔
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()?;	24✔
149
150	while let Some((op, pos)) = receiver.blocking_recv() {	124✔
151	match op {	100✔
152	LogOperation::Op { op } => match op {	44✔
153	Operation::Delete { old } => {	4✔
154	if let Some(meta) = cache.delete(&old)? {	4✔
155	if let Some((endpoint_name, operations_sender)) = operations_sender.as_ref()	4✔
156	{	4✔
157	let operation = types_helper::map_delete_operation(	4✔
158	endpoint_name.clone(),	4✔
159	CacheRecord::new(meta.id, meta.version, old),	4✔
160	);	4✔
161	send_and_log_error(operations_sender, operation);	4✔
162	}	4✔
163	}	×
164	let mut labels = cache.labels().clone();	4✔
165	labels.push(OPERATION_TYPE_LABEL, "delete");	4✔
166	labels.push(SNAPSHOTTING_LABEL, snapshotting_str(snapshotting));	4✔
167	increment_counter!(CACHE_OPERATION_COUNTER_NAME, labels);	4✔
168	}
169	Operation::Insert { new } => {	40✔
170	let result = cache.insert(&new)?;	40✔
171	let mut labels = cache.labels().clone();	40✔
172	labels.push(OPERATION_TYPE_LABEL, "insert");	40✔
173	labels.push(SNAPSHOTTING_LABEL, snapshotting_str(snapshotting));	40✔
174	increment_counter!(CACHE_OPERATION_COUNTER_NAME, labels);	40✔
175
176	if let Some((endpoint_name, operations_sender)) = operations_sender.as_ref() {	40✔
177	send_upsert_result(	40✔
178	endpoint_name,	40✔
179	operations_sender,	40✔
180	result,	40✔
181	&schema,	40✔
182	None,	40✔
183	new,	40✔
184	);	40✔
185	}	40✔
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	LogOperation::Commit { decision_instant } => {	32✔
207	cache.set_metadata(pos)?;	32✔
208	cache.commit()?;	32✔
209	if let Ok(duration) = decision_instant.elapsed() {	32✔
210	histogram!(	32✔
211	DATA_LATENCY_HISTOGRAM_NAME,
212	duration,	×
213	cache.labels().clone()	×
214	);
215	}	×
216	}
217	LogOperation::SnapshottingDone { connection_name } => {	24✔
218	cache.set_metadata(pos)?;	24✔
219	cache.set_connection_snapshotting_done(&connection_name)?;	24✔
220	cache.commit()?;	24✔
221	snapshotting = !cache.is_snapshotting_done()?;	24✔
222	}
223	}
224	}	×
225
226	Ok(())	24✔
227	}	24✔
228
229	fn send_upsert_result(	40✔
230	endpoint_name: &str,	40✔
231	operations_sender: &Sender<GrpcOperation>,	40✔
232	upsert_result: UpsertResult,	40✔
233	schema: &Schema,	40✔
234	old: Option<Record>,	40✔
235	new: Record,	40✔
236	) {	40✔
237	match upsert_result {	40✔
238	UpsertResult::Inserted { meta } => {	40✔
239	let op = types_helper::map_insert_operation(	40✔
240	endpoint_name.to_string(),	40✔
241	CacheRecord::new(meta.id, meta.version, new),	40✔
242	);	40✔
243	send_and_log_error(operations_sender, op);	40✔
244	}	40✔
245	UpsertResult::Updated { old_meta, new_meta } => {	×
246	// If `old` is `None`, it means `Updated` comes from `Insert` operation.	×
247	// 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.	×
248	// So we create the old record with only the fields in the primary index, cloned from `new`.	×
249	let old = old.unwrap_or_else(\|\| {	×
250	let mut record = Record::new(vec![Field::Null; new.values.len()]);	×
251	for index in schema.primary_index.iter() {	×
252	record.values[index] = new.values[index].clone();	×
253	}	×
254	record	×
255	});	×
256	let op = types_helper::map_update_operation(	×
257	endpoint_name.to_string(),	×
258	CacheRecord::new(old_meta.id, old_meta.version, old),	×
259	CacheRecord::new(new_meta.id, new_meta.version, new),	×
260	);	×
261	send_and_log_error(operations_sender, op);	×
262	}	×
263	UpsertResult::Ignored => {}	×
264	}	×
265	}	40✔
266		×
267	fn send_and_log_error<T: Send + Sync + 'static>(sender: &Sender<T>, msg: T) {
268	if let Err(e) = sender.send(msg) {	44✔
269	error!("Failed to send broadcast message: {}", e);	×
270	}	44✔
271	}	44✔
272		×
273	fn snapshotting_str(snapshotting: bool) -> &'static str {	44✔
274	if snapshotting {	44✔
275	"true"	36✔
276	} else {	×
277	"false"	8✔
278	}	×
279	}	44✔

getdozer / dozer / 5960771901

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