5780689709

pending completion

Build # 5780689709

Build Type

push

github

Committed by

web-flow

Commit Message

Remove s390x (#1827)

Run Details

45527 of 59015 relevant lines covered (77.14%)

50795.91 hits per line

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

24.36

/dozer-cli/src/lib.rs

pub mod cli;
pub mod errors;
pub mod pipeline;
pub mod shutdown;
pub mod simple;
mod ui_helper;
use dozer_core::{app::AppPipeline, errors::ExecutionError};
use dozer_sql::pipeline::{builder::statement_to_pipeline, errors::PipelineError};
use dozer_types::log::debug;
use errors::OrchestrationError;
use shutdown::ShutdownSender;
use std::{
    backtrace::{Backtrace, BacktraceStatus},
    panic, process,
    thread::current,
};
use tokio::task::JoinHandle;
#[cfg(feature = "cloud")]
pub mod cloud_app_context;
#[cfg(feature = "cloud")]
mod cloud_helper;
pub mod config_helper;
pub mod console_helper;
#[cfg(feature = "cloud")]
mod progress_printer;
#[cfg(test)]
mod tests;
mod utils;

#[cfg(feature = "cloud")]
pub trait CloudOrchestrator {
    fn deploy(
        &mut self,
        cloud: Cloud,
        deploy: DeployCommandArgs,
        config_paths: Vec<String>,
    ) -> Result<(), OrchestrationError>;
    fn delete(&mut self, cloud: Cloud) -> Result<(), OrchestrationError>;
    fn list(&mut self, cloud: Cloud, list: ListCommandArgs) -> Result<(), OrchestrationError>;
    fn status(&mut self, cloud: Cloud) -> Result<(), OrchestrationError>;
    fn monitor(&mut self, cloud: Cloud) -> Result<(), OrchestrationError>;
    fn trace_logs(&mut self, cloud: Cloud, logs: LogCommandArgs) -> Result<(), OrchestrationError>;
    fn login(
        &mut self,
        cloud: Cloud,
        organisation_slug: Option<String>,
        profile: Option<String>,
        client_id: Option<String>,
        client_secret: Option<String>,
    ) -> Result<(), OrchestrationError>;
    fn execute_secrets_command(
        &mut self,
        cloud: Cloud,
        command: SecretsCommand,
    ) -> Result<(), OrchestrationError>;
}

// Re-exports
pub use dozer_ingestion::{
    connectors::{get_connector, TableInfo},
    errors::ConnectorError,
};
pub use dozer_sql::pipeline::builder::QueryContext;
pub use ui_helper::config_to_ui_dag;
pub fn wrapped_statement_to_pipeline(sql: &str) -> Result<QueryContext, PipelineError> {
    let mut pipeline = AppPipeline::new();
    statement_to_pipeline(sql, &mut pipeline, None)
}

#[cfg(feature = "cloud")]
use crate::cli::cloud::{
    Cloud, DeployCommandArgs, ListCommandArgs, LogCommandArgs, SecretsCommand,
};
pub use dozer_types::models::connection::Connection;
use dozer_types::tracing::error;

async fn flatten_join_handle(
    handle: JoinHandle<Result<(), OrchestrationError>>,
) -> Result<(), OrchestrationError> {
    match handle.await {
        Ok(Ok(_)) => Ok(()),
        Ok(Err(err)) => Err(err),
        Err(err) => Err(OrchestrationError::JoinError(err)),
    }
}

fn join_handle_map_err<E: Send + 'static>(
    handle: JoinHandle<Result<(), E>>,
    f: impl FnOnce(E) -> OrchestrationError + Send + 'static,
) -> JoinHandle<Result<(), OrchestrationError>> {
    tokio::spawn(async move {
        match handle.await {
            Ok(Ok(_)) => Ok(()),
            Ok(Err(err)) => Err(f(err)),
            Err(err) => Err(OrchestrationError::JoinError(err)),
        }
    })
}

pub fn set_panic_hook() {
    panic::set_hook(Box::new(move |panic_info| {
        // All the orchestrator errors are captured here
        if let Some(e) = panic_info.payload().downcast_ref::<OrchestrationError>() {
            error!("{}", e);
            debug!("{:?}", e);
        // All the connector errors are captured here
        } else if let Some(e) = panic_info.payload().downcast_ref::<ConnectorError>() {
            error!("{}", e);
            debug!("{:?}", e);
        // All the pipeline errors are captured here
        } else if let Some(e) = panic_info.payload().downcast_ref::<ExecutionError>() {
            error!("{}", e);
            debug!("{:?}", e);
        // If any errors are sent as strings.
        } else if let Some(s) = panic_info.payload().downcast_ref::<&str>() {
            error!("{s:?}");
        } else {
            error!("{}", panic_info);
        }

        let backtrace = Backtrace::capture();
        if backtrace.status() == BacktraceStatus::Captured {
            error!(
                "thread '{}' panicked at '{}'\n stack backtrace:\n{}",
                current()
                    .name()
                    .map(ToString::to_string)
                    .unwrap_or_default(),
                panic_info
                    .location()
                    .map(ToString::to_string)
                    .unwrap_or_default(),
                backtrace
            );
        }

        process::exit(1);
    }));
}

pub fn set_ctrl_handler(shutdown_sender: ShutdownSender) {
    let mut shutdown = Some(shutdown_sender);
    ctrlc::set_handler(move || {
        if let Some(shutdown) = shutdown.take() {
            shutdown.shutdown()
        }
    })
    .expect("Error setting Ctrl-C handler");
}

1	pub mod cli;
2	pub mod errors;
3	pub mod pipeline;
4	pub mod shutdown;
5	pub mod simple;
6	mod ui_helper;
7	use dozer_core::{app::AppPipeline, errors::ExecutionError};
8	use dozer_sql::pipeline::{builder::statement_to_pipeline, errors::PipelineError};
9	use dozer_types::log::debug;
10	use errors::OrchestrationError;
11	use shutdown::ShutdownSender;
12	use std::{
13	backtrace::{Backtrace, BacktraceStatus},
14	panic, process,
15	thread::current,
16	};
17	use tokio::task::JoinHandle;
18	#[cfg(feature = "cloud")]
19	pub mod cloud_app_context;
20	#[cfg(feature = "cloud")]
21	mod cloud_helper;
22	pub mod config_helper;
23	pub mod console_helper;
24	#[cfg(feature = "cloud")]
25	mod progress_printer;
26	#[cfg(test)]
27	mod tests;
28	mod utils;
29
30	#[cfg(feature = "cloud")]
31	pub trait CloudOrchestrator {
32	fn deploy(
33	&mut self,
34	cloud: Cloud,
35	deploy: DeployCommandArgs,
36	config_paths: Vec<String>,
37	) -> Result<(), OrchestrationError>;
38	fn delete(&mut self, cloud: Cloud) -> Result<(), OrchestrationError>;
39	fn list(&mut self, cloud: Cloud, list: ListCommandArgs) -> Result<(), OrchestrationError>;
40	fn status(&mut self, cloud: Cloud) -> Result<(), OrchestrationError>;
41	fn monitor(&mut self, cloud: Cloud) -> Result<(), OrchestrationError>;
42	fn trace_logs(&mut self, cloud: Cloud, logs: LogCommandArgs) -> Result<(), OrchestrationError>;
43	fn login(
44	&mut self,
45	cloud: Cloud,
46	organisation_slug: Option<String>,
47	profile: Option<String>,
48	client_id: Option<String>,
49	client_secret: Option<String>,
50	) -> Result<(), OrchestrationError>;
51	fn execute_secrets_command(
52	&mut self,
53	cloud: Cloud,
54	command: SecretsCommand,
55	) -> Result<(), OrchestrationError>;
56	}
57
58	// Re-exports
59	pub use dozer_ingestion::{
60	connectors::{get_connector, TableInfo},
61	errors::ConnectorError,
62	};
63	pub use dozer_sql::pipeline::builder::QueryContext;
64	pub use ui_helper::config_to_ui_dag;
65	pub fn wrapped_statement_to_pipeline(sql: &str) -> Result<QueryContext, PipelineError> {	×
66	let mut pipeline = AppPipeline::new();	×
67	statement_to_pipeline(sql, &mut pipeline, None)	×
68	}	×
69		×
70	#[cfg(feature = "cloud")]
71	use crate::cli::cloud::{
72	Cloud, DeployCommandArgs, ListCommandArgs, LogCommandArgs, SecretsCommand,
73	};
74	pub use dozer_types::models::connection::Connection;
75	use dozer_types::tracing::error;
76
77	async fn flatten_join_handle(	24✔
78	handle: JoinHandle<Result<(), OrchestrationError>>,	24✔
79	) -> Result<(), OrchestrationError> {	24✔
80	match handle.await {	24✔
81	Ok(Ok(_)) => Ok(()),	24✔
82	Ok(Err(err)) => Err(err),	×
83	Err(err) => Err(OrchestrationError::JoinError(err)),	×
84	}	×
85	}	24✔
86		×
87	fn join_handle_map_err<E: Send + 'static>(	6✔
88	handle: JoinHandle<Result<(), E>>,	6✔
89	f: impl FnOnce(E) -> OrchestrationError + Send + 'static,	6✔
90	) -> JoinHandle<Result<(), OrchestrationError>> {	6✔
91	tokio::spawn(async move {	6✔
92	match handle.await {	6✔
93	Ok(Ok(_)) => Ok(()),	6✔
94	Ok(Err(err)) => Err(f(err)),	×
95	Err(err) => Err(OrchestrationError::JoinError(err)),	×
96	}	×
97	})	6✔
98	}	6✔
99		×
100	pub fn set_panic_hook() {	2✔
101	panic::set_hook(Box::new(move \|panic_info\| {	2✔
102	// All the orchestrator errors are captured here	×
103	if let Some(e) = panic_info.payload().downcast_ref::<OrchestrationError>() {	×
104	error!("{}", e);	×
105	debug!("{:?}", e);	×
106	// All the connector errors are captured here	×
107	} else if let Some(e) = panic_info.payload().downcast_ref::<ConnectorError>() {	×
108	error!("{}", e);	×
109	debug!("{:?}", e);	×
110	// All the pipeline errors are captured here	×
111	} else if let Some(e) = panic_info.payload().downcast_ref::<ExecutionError>() {	×
112	error!("{}", e);	×
113	debug!("{:?}", e);	×
114	// If any errors are sent as strings.	×
115	} else if let Some(s) = panic_info.payload().downcast_ref::<&str>() {	×
116	error!("{s:?}");	×
117	} else {	×
118	error!("{}", panic_info);	×
119	}	×
120
121	let backtrace = Backtrace::capture();	×
122	if backtrace.status() == BacktraceStatus::Captured {	×
123	error!(	×
124	"thread '{}' panicked at '{}'\n stack backtrace:\n{}",	×
125	current()	×
126	.name()	×
127	.map(ToString::to_string)	×
128	.unwrap_or_default(),	×
129	panic_info	×
130	.location()	×
131	.map(ToString::to_string)	×
132	.unwrap_or_default(),	×
133	backtrace	×
134	);	×
135	}	×
136		×
137	process::exit(1);	×
138	}));	2✔
139	}	2✔
140		×
141	pub fn set_ctrl_handler(shutdown_sender: ShutdownSender) {	×
142	let mut shutdown = Some(shutdown_sender);	×
143	ctrlc::set_handler(move \|\| {	×
144	if let Some(shutdown) = shutdown.take() {	×
145	shutdown.shutdown()	×
146	}	×
147	})	×
148	.expect("Error setting Ctrl-C handler");	×
149	}	×

getdozer / dozer / 5780689709

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