4381907514

Build # 4381907514

Build Type

push

github

Committed by GitHub

Commit Message

feat: implement tracing using open telemetry (#1176)

Run Details

510 of 510 new or added lines in 31 files covered. (100.0%)

27878 of 39615 relevant lines covered (70.37%)

47752.4 hits per line

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

0.0

/dozer-admin/src/server.rs

use std::sync::atomic::Ordering;

use crate::{
    cli::{utils::get_db_path, AdminCliConfig},
    db::pool::establish_connection,
    services::{application_service::AppService, connection_service::ConnectionService},
};
use dozer_types::{log::info, tracing::Level};
use tonic::{transport::Server, Request, Response, Status};
use tower_http::trace::{self, TraceLayer};
pub mod dozer_admin_grpc {
    #![allow(clippy::derive_partial_eq_without_eq, clippy::large_enum_variant)]
    tonic::include_proto!("dozer_admin_grpc");
    pub(crate) const FILE_DESCRIPTOR_SET: &[u8] =
        tonic::include_file_descriptor_set!("dozer_admin_grpc_descriptor");
}
use self::dozer_admin_grpc::{
    GenerateGraphRequest, GenerateGraphResponse, GenerateYamlRequest, GenerateYamlResponse,
    ListAppRequest, ListAppResponse, ParseRequest, ParseResponse, ParseYamlRequest,
    ParseYamlResponse, StartRequest, StartResponse, StopRequest, StopResponse, UpdateAppRequest,
    ValidateConnectionResponse,
};
use dozer_admin_grpc::{
    dozer_admin_server::{DozerAdmin, DozerAdminServer},
    AppResponse, ConnectionRequest, ConnectionResponse, CreateAppRequest, GetAllConnectionRequest,
    GetAllConnectionResponse, GetAppRequest, GetTablesRequest, GetTablesResponse,
    UpdateConnectionRequest,
};

pub struct GrpcService {
    app_service: AppService,
    connection_service: ConnectionService,
}

#[tonic::async_trait]
impl DozerAdmin for GrpcService {
    async fn parse_sql(
        &self,
        request: tonic::Request<ParseRequest>,
    ) -> Result<tonic::Response<ParseResponse>, tonic::Status> {
        let result = self.app_service.parse_sql(request.into_inner());
        match result {
            Ok(response) => Ok(Response::new(response)),
            Err(e) => Err(Status::new(tonic::Code::Internal, e.message)),
        }
    }

    async fn parse_yaml(
        &self,
        request: tonic::Request<ParseYamlRequest>,
    ) -> Result<tonic::Response<ParseYamlResponse>, tonic::Status> {
        let result = self.app_service.parse_yaml(request.into_inner());
        match result {
            Ok(response) => Ok(Response::new(response)),
            Err(e) => Err(Status::new(tonic::Code::Internal, e.message)),
        }
    }

    async fn generate_yaml(
        &self,
        request: tonic::Request<GenerateYamlRequest>,
    ) -> Result<tonic::Response<GenerateYamlResponse>, tonic::Status> {
        let result = self.app_service.generate_yaml(request.into_inner());
        match result {
            Ok(response) => Ok(Response::new(response)),
            Err(e) => Err(Status::new(tonic::Code::Internal, e.message)),
        }
    }

    async fn generate_graph(
        &self,
        request: tonic::Request<GenerateGraphRequest>,
    ) -> Result<tonic::Response<GenerateGraphResponse>, tonic::Status> {
        let result = self.app_service.generate(request.into_inner());
        match result {
            Ok(response) => Ok(Response::new(response)),
            Err(e) => Err(Status::new(tonic::Code::Internal, e.message)),
        }
    }

    async fn create_application(
        &self,
        request: tonic::Request<CreateAppRequest>,
    ) -> Result<tonic::Response<AppResponse>, tonic::Status> {
        let result = self.app_service.create(request.into_inner());
        match result {
            Ok(response) => Ok(Response::new(response)),
            Err(e) => Err(Status::new(tonic::Code::Internal, e.message)),
        }
    }
    async fn list_applications(
        &self,
        request: tonic::Request<ListAppRequest>,
    ) -> Result<tonic::Response<ListAppResponse>, tonic::Status> {
        let result = self.app_service.list(request.into_inner());
        match result {
            Ok(response) => Ok(Response::new(response)),
            Err(e) => Err(Status::new(tonic::Code::Internal, e.message)),
        }
    }
    async fn update_application(
        &self,
        request: tonic::Request<UpdateAppRequest>,
    ) -> Result<tonic::Response<AppResponse>, tonic::Status> {
        let result = self.app_service.update_app(request.into_inner());
        match result {
            Ok(response) => Ok(Response::new(response)),
            Err(e) => Err(Status::new(tonic::Code::Internal, e.message)),
        }
    }

    async fn get_application(
        &self,
        request: tonic::Request<GetAppRequest>,
    ) -> Result<tonic::Response<AppResponse>, tonic::Status> {
        let result = self.app_service.get_app(request.into_inner());
        match result {
            Ok(response) => Ok(Response::new(response)),
            Err(e) => Err(Status::new(tonic::Code::Internal, e.message)),
        }
    }

    async fn validate_connection(
        &self,
        request: tonic::Request<ConnectionRequest>,
    ) -> Result<tonic::Response<ValidateConnectionResponse>, tonic::Status> {
        let result = self
            .connection_service
            .validate_connection(request.into_inner());
        match result.await {
            Ok(response) => Ok(Response::new(response)),
            Err(e) => Err(Status::new(tonic::Code::Internal, e.message)),
        }
    }

    async fn create_connection(
        &self,
        request: Request<ConnectionRequest>,
    ) -> Result<Response<ConnectionResponse>, Status> {
        let result = self
            .connection_service
            .create_connection(request.into_inner());
        match result {
            Ok(response) => Ok(Response::new(response)),
            Err(e) => Err(Status::new(tonic::Code::Internal, e.message)),
        }
    }

    async fn list_connections(
        &self,
        request: Request<GetAllConnectionRequest>,
    ) -> Result<Response<GetAllConnectionResponse>, Status> {
        let result = self.connection_service.list(request.into_inner());
        match result {
            Ok(response) => Ok(Response::new(response)),
            Err(e) => Err(Status::new(tonic::Code::Internal, e.message)),
        }
    }

    async fn get_tables(
        &self,
        request: Request<GetTablesRequest>,
    ) -> Result<Response<GetTablesResponse>, Status> {
        let result = self
            .connection_service
            .get_tables(request.into_inner())
            .await;
        match result {
            Ok(response) => Ok(Response::new(response)),
            Err(e) => Err(Status::new(tonic::Code::Internal, e.message)),
        }
    }

    async fn update_connection(
        &self,
        request: Request<UpdateConnectionRequest>,
    ) -> Result<Response<ConnectionResponse>, Status> {
        let result = self.connection_service.update(request.into_inner());
        match result {
            Ok(response) => Ok(Response::new(response)),
            Err(e) => Err(Status::new(tonic::Code::Internal, e.message)),
        }
    }

    async fn start_dozer(
        &self,
        request: tonic::Request<StartRequest>,
    ) -> Result<tonic::Response<StartResponse>, tonic::Status> {
        let result = self.app_service.start_dozer(request.into_inner());
        match result {
            Ok(response) => Ok(Response::new(response)),
            Err(e) => Err(Status::new(tonic::Code::Internal, e.message)),
        }
    }

    async fn stop_dozer(
        &self,
        request: tonic::Request<StopRequest>,
    ) -> Result<tonic::Response<StopResponse>, tonic::Status> {
        let result = self.app_service.stop_dozer(request.into_inner());
        match result {
            Ok(response) => Ok(Response::new(response)),
            Err(e) => Err(Status::new(tonic::Code::Internal, e.message)),
        }
    }
}

pub async fn start_admin_server(config: AdminCliConfig) -> Result<(), tonic::transport::Error> {
    dozer_tracing::init_telemetry(None, None);

    let host = config.host;
    let port = config.port;
    let addr = format!("{host:}:{port:}").parse().unwrap();
    let database_url: String = get_db_path();

    let db_pool = establish_connection(database_url);
    let app_service = AppService::new(db_pool.to_owned());
    let grpc_service = GrpcService {
        connection_service: ConnectionService::new(db_pool.to_owned()),
        app_service: app_service.to_owned(),
    };
    let server = DozerAdminServer::new(grpc_service);
    let server = tonic_web::config().allow_all_origins().enable(server);
    let reflection_service = tonic_reflection::server::Builder::configure()
        .register_encoded_file_descriptor_set(dozer_admin_grpc::FILE_DESCRIPTOR_SET)
        .build()
        .unwrap();

    info!("Starting Dozer Admin server on http://{}:{} ", host, port,);

    Server::builder()
        .layer(
            TraceLayer::new_for_http()
                .make_span_with(trace::DefaultMakeSpan::new().level(Level::INFO))
                .on_response(trace::DefaultOnResponse::new().level(Level::INFO)),
        )
        .accept_http1(true)
        .add_service(reflection_service)
        .add_service(server)
        .serve(addr)
        .await?;

    for (_, r) in app_service.apps.write().iter_mut() {
        r.store(false, Ordering::Relaxed);
    }
    Ok(())
}

1	use std::sync::atomic::Ordering;
2
3	use crate::{
4	cli::{utils::get_db_path, AdminCliConfig},
5	db::pool::establish_connection,
6	services::{application_service::AppService, connection_service::ConnectionService},
7	};
8	use dozer_types::{log::info, tracing::Level};
9	use tonic::{transport::Server, Request, Response, Status};
10	use tower_http::trace::{self, TraceLayer};
11	pub mod dozer_admin_grpc {
12	#![allow(clippy::derive_partial_eq_without_eq, clippy::large_enum_variant)]
13	tonic::include_proto!("dozer_admin_grpc");
14	pub(crate) const FILE_DESCRIPTOR_SET: &[u8] =
15	tonic::include_file_descriptor_set!("dozer_admin_grpc_descriptor");
16	}
17	use self::dozer_admin_grpc::{
18	GenerateGraphRequest, GenerateGraphResponse, GenerateYamlRequest, GenerateYamlResponse,
19	ListAppRequest, ListAppResponse, ParseRequest, ParseResponse, ParseYamlRequest,
20	ParseYamlResponse, StartRequest, StartResponse, StopRequest, StopResponse, UpdateAppRequest,
21	ValidateConnectionResponse,
22	};
23	use dozer_admin_grpc::{
24	dozer_admin_server::{DozerAdmin, DozerAdminServer},
25	AppResponse, ConnectionRequest, ConnectionResponse, CreateAppRequest, GetAllConnectionRequest,
26	GetAllConnectionResponse, GetAppRequest, GetTablesRequest, GetTablesResponse,
27	UpdateConnectionRequest,
28	};
29
30	pub struct GrpcService {
31	app_service: AppService,
32	connection_service: ConnectionService,
33	}
34
35	#[tonic::async_trait]
36	impl DozerAdmin for GrpcService {
37	async fn parse_sql(	×
38	&self,	×
39	request: tonic::Request<ParseRequest>,	×
40	) -> Result<tonic::Response<ParseResponse>, tonic::Status> {	×
41	let result = self.app_service.parse_sql(request.into_inner());	×
42	match result {	×
43	Ok(response) => Ok(Response::new(response)),	×
44	Err(e) => Err(Status::new(tonic::Code::Internal, e.message)),	×
45	}
46	}	×
47
48	async fn parse_yaml(	×
49	&self,	×
50	request: tonic::Request<ParseYamlRequest>,	×
51	) -> Result<tonic::Response<ParseYamlResponse>, tonic::Status> {	×
52	let result = self.app_service.parse_yaml(request.into_inner());	×
53	match result {	×
54	Ok(response) => Ok(Response::new(response)),	×
55	Err(e) => Err(Status::new(tonic::Code::Internal, e.message)),	×
56	}
57	}	×
58
59	async fn generate_yaml(	×
60	&self,	×
61	request: tonic::Request<GenerateYamlRequest>,	×
62	) -> Result<tonic::Response<GenerateYamlResponse>, tonic::Status> {	×
63	let result = self.app_service.generate_yaml(request.into_inner());	×
64	match result {	×
65	Ok(response) => Ok(Response::new(response)),	×
66	Err(e) => Err(Status::new(tonic::Code::Internal, e.message)),	×
67	}
68	}	×
69
70	async fn generate_graph(	×
71	&self,	×
72	request: tonic::Request<GenerateGraphRequest>,	×
73	) -> Result<tonic::Response<GenerateGraphResponse>, tonic::Status> {	×
74	let result = self.app_service.generate(request.into_inner());	×
75	match result {	×
76	Ok(response) => Ok(Response::new(response)),	×
77	Err(e) => Err(Status::new(tonic::Code::Internal, e.message)),	×
78	}
79	}	×
80
81	async fn create_application(	×
82	&self,	×
83	request: tonic::Request<CreateAppRequest>,	×
84	) -> Result<tonic::Response<AppResponse>, tonic::Status> {	×
85	let result = self.app_service.create(request.into_inner());	×
86	match result {	×
87	Ok(response) => Ok(Response::new(response)),	×
88	Err(e) => Err(Status::new(tonic::Code::Internal, e.message)),	×
89	}
90	}	×
91	async fn list_applications(	×
92	&self,	×
93	request: tonic::Request<ListAppRequest>,	×
94	) -> Result<tonic::Response<ListAppResponse>, tonic::Status> {	×
95	let result = self.app_service.list(request.into_inner());	×
96	match result {	×
97	Ok(response) => Ok(Response::new(response)),	×
98	Err(e) => Err(Status::new(tonic::Code::Internal, e.message)),	×
99	}
100	}	×
101	async fn update_application(	×
102	&self,	×
103	request: tonic::Request<UpdateAppRequest>,	×
104	) -> Result<tonic::Response<AppResponse>, tonic::Status> {	×
105	let result = self.app_service.update_app(request.into_inner());	×
106	match result {	×
107	Ok(response) => Ok(Response::new(response)),	×
108	Err(e) => Err(Status::new(tonic::Code::Internal, e.message)),	×
109	}
110	}	×
111
112	async fn get_application(	×
113	&self,	×
114	request: tonic::Request<GetAppRequest>,	×
115	) -> Result<tonic::Response<AppResponse>, tonic::Status> {	×
116	let result = self.app_service.get_app(request.into_inner());	×
117	match result {	×
118	Ok(response) => Ok(Response::new(response)),	×
119	Err(e) => Err(Status::new(tonic::Code::Internal, e.message)),	×
120	}
121	}	×
122
123	async fn validate_connection(	×
124	&self,	×
125	request: tonic::Request<ConnectionRequest>,	×
126	) -> Result<tonic::Response<ValidateConnectionResponse>, tonic::Status> {	×
127	let result = self	×
128	.connection_service	×
129	.validate_connection(request.into_inner());	×
130	match result.await {	×
131	Ok(response) => Ok(Response::new(response)),	×
132	Err(e) => Err(Status::new(tonic::Code::Internal, e.message)),	×
133	}
134	}	×
135
136	async fn create_connection(	×
137	&self,	×
138	request: Request<ConnectionRequest>,	×
139	) -> Result<Response<ConnectionResponse>, Status> {	×
140	let result = self	×
141	.connection_service	×
142	.create_connection(request.into_inner());	×
143	match result {	×
144	Ok(response) => Ok(Response::new(response)),	×
145	Err(e) => Err(Status::new(tonic::Code::Internal, e.message)),	×
146	}
147	}	×
148
149	async fn list_connections(	×
150	&self,	×
151	request: Request<GetAllConnectionRequest>,	×
152	) -> Result<Response<GetAllConnectionResponse>, Status> {	×
153	let result = self.connection_service.list(request.into_inner());	×
154	match result {	×
155	Ok(response) => Ok(Response::new(response)),	×
156	Err(e) => Err(Status::new(tonic::Code::Internal, e.message)),	×
157	}
158	}	×
159
160	async fn get_tables(	×
161	&self,	×
162	request: Request<GetTablesRequest>,	×
163	) -> Result<Response<GetTablesResponse>, Status> {	×
164	let result = self	×
165	.connection_service	×
166	.get_tables(request.into_inner())	×
167	.await;	×
168	match result {	×
169	Ok(response) => Ok(Response::new(response)),	×
170	Err(e) => Err(Status::new(tonic::Code::Internal, e.message)),	×
171	}
172	}	×
173
174	async fn update_connection(	×
175	&self,	×
176	request: Request<UpdateConnectionRequest>,	×
177	) -> Result<Response<ConnectionResponse>, Status> {	×
178	let result = self.connection_service.update(request.into_inner());	×
179	match result {	×
180	Ok(response) => Ok(Response::new(response)),	×
181	Err(e) => Err(Status::new(tonic::Code::Internal, e.message)),	×
182	}
183	}	×
184
185	async fn start_dozer(	×
186	&self,	×
187	request: tonic::Request<StartRequest>,	×
188	) -> Result<tonic::Response<StartResponse>, tonic::Status> {	×
189	let result = self.app_service.start_dozer(request.into_inner());	×
190	match result {	×
191	Ok(response) => Ok(Response::new(response)),	×
192	Err(e) => Err(Status::new(tonic::Code::Internal, e.message)),	×
193	}
194	}	×
195
196	async fn stop_dozer(	×
197	&self,	×
198	request: tonic::Request<StopRequest>,	×
199	) -> Result<tonic::Response<StopResponse>, tonic::Status> {	×
200	let result = self.app_service.stop_dozer(request.into_inner());	×
201	match result {	×
202	Ok(response) => Ok(Response::new(response)),	×
203	Err(e) => Err(Status::new(tonic::Code::Internal, e.message)),	×
204	}
205	}	×
206	}
207
208	pub async fn start_admin_server(config: AdminCliConfig) -> Result<(), tonic::transport::Error> {	×
209	dozer_tracing::init_telemetry(None, None);	×
210		×
211	let host = config.host;	×
212	let port = config.port;	×
213	let addr = format!("{host:}:{port:}").parse().unwrap();	×
214	let database_url: String = get_db_path();	×
215		×
216	let db_pool = establish_connection(database_url);	×
217	let app_service = AppService::new(db_pool.to_owned());	×
218	let grpc_service = GrpcService {	×
219	connection_service: ConnectionService::new(db_pool.to_owned()),	×
220	app_service: app_service.to_owned(),	×
221	};	×
222	let server = DozerAdminServer::new(grpc_service);	×
223	let server = tonic_web::config().allow_all_origins().enable(server);	×
224	let reflection_service = tonic_reflection::server::Builder::configure()	×
225	.register_encoded_file_descriptor_set(dozer_admin_grpc::FILE_DESCRIPTOR_SET)	×
226	.build()	×
227	.unwrap();	×
228
229	info!("Starting Dozer Admin server on http://{}:{} ", host, port,);	×
230
231	Server::builder()	×
232	.layer(	×
233	TraceLayer::new_for_http()	×
234	.make_span_with(trace::DefaultMakeSpan::new().level(Level::INFO))	×
235	.on_response(trace::DefaultOnResponse::new().level(Level::INFO)),	×
236	)	×
237	.accept_http1(true)	×
238	.add_service(reflection_service)	×
239	.add_service(server)	×
240	.serve(addr)	×
241	.await?;	×
242
243	for (_, r) in app_service.apps.write().iter_mut() {	×
244	r.store(false, Ordering::Relaxed);	×
245	}	×
246	Ok(())	×
247	}	×

getdozer / dozer / 4381907514

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