-
Notifications
You must be signed in to change notification settings - Fork 336
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: support skywalking integration
- Loading branch information
Showing
7 changed files
with
195 additions
and
1 deletion.
There are no files selected for viewing
Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,12 @@ | ||
| [package] | ||
| name = "integration_skywalking" | ||
| version = "0.1.0" | ||
| edition = "2021" | ||
|
|
||
| [dependencies] | ||
| hyper = { version = "0.14", features = ["full"] } | ||
| bytes = "1.0" | ||
| tokio = { version = "1.20.1", features = ["full"] } | ||
| public = { path = "../../crates/public" } | ||
| prost = "0.11" | ||
| log = "0.4" |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,93 @@ | ||
| use bytes::{BufMut, BytesMut}; | ||
| use hyper::{ | ||
| header::{HeaderValue, CONTENT_TYPE}, | ||
| Body, HeaderMap, Request, Response, StatusCode, Version, | ||
| }; | ||
| use log::warn; | ||
| use prost::{EncodeError, Message}; | ||
| use public::{ | ||
| proto::metric, | ||
| queue::DebugSender, | ||
| sender::{SendMessageType, Sendable}, | ||
| }; | ||
| use std::{ | ||
| future::Future, | ||
| net::{IpAddr, SocketAddr}, | ||
| }; | ||
|
|
||
| const CONTENT_TYPE_GRPC: &str = "application/grpc"; | ||
|
|
||
| #[derive(Debug, PartialEq)] | ||
| pub struct SkyWalkingExtra(pub metric::SkyWalkingExtra); | ||
|
|
||
| impl Sendable for SkyWalkingExtra { | ||
| fn encode(self, buf: &mut Vec<u8>) -> Result<usize, EncodeError> { | ||
| self.0.encode(buf).map(|_| self.0.encoded_len()) | ||
| } | ||
|
|
||
| fn message_type(&self) -> SendMessageType { | ||
| SendMessageType::SkyWalking | ||
| } | ||
| } | ||
|
|
||
| pub async fn handle_skywalking_request<F, Fut>( | ||
| peer_addr: SocketAddr, | ||
| req: Request<Body>, | ||
| skywalking_sender: DebugSender<SkyWalkingExtra>, | ||
| decode_data: F, | ||
| ) -> Response<Body> | ||
| where | ||
| F: FnOnce(Body, HeaderMap) -> Fut, | ||
| Fut: Future<Output = Result<Vec<u8>, Response<Body>>>, | ||
| { | ||
| let is_http2_request = req | ||
| .headers() | ||
| .get(CONTENT_TYPE.as_str()) | ||
| .map(|v| v.as_bytes().starts_with(CONTENT_TYPE_GRPC.as_bytes())) | ||
| .unwrap_or(false); | ||
|
|
||
| // uniform sender send to remote server | ||
| let (parts, body) = req.into_parts(); | ||
| let skywalking_data = decode_data(body, parts.headers.clone()).await.unwrap(); | ||
|
|
||
| let mut skywalking_extra = metric::SkyWalkingExtra::default(); | ||
| skywalking_extra.data = skywalking_data; | ||
| skywalking_extra.peer_ip = match peer_addr.ip() { | ||
| IpAddr::V4(ip4) => ip4.octets().to_vec(), | ||
| IpAddr::V6(ip6) => ip6.octets().to_vec(), | ||
| }; | ||
|
|
||
| if let Err(e) = skywalking_sender.send(SkyWalkingExtra(skywalking_extra)) { | ||
| warn!("skywalking_sender failed to send data, because {:?}", e); | ||
| } | ||
| if is_http2_request { | ||
| empty_grpc_response() | ||
| } else { | ||
| Response::builder() | ||
| .status(StatusCode::OK) | ||
| .body(Body::empty()) | ||
| .unwrap() | ||
| } | ||
| } | ||
|
|
||
| pub fn empty_grpc_response() -> Response<Body> { | ||
| let mut response_buf = BytesMut::new(); | ||
| response_buf.put_u8(0); // grpc not compression | ||
| response_buf.put_u32(0); // grpc data, return length = 0 | ||
|
|
||
| let mut trailers = HeaderMap::new(); | ||
| trailers.insert("grpc-status", HeaderValue::from_static("0")); | ||
|
|
||
| let (mut sender, body) = Body::channel(); | ||
| tokio::spawn(async move { | ||
| sender.send_data(response_buf.freeze()).await.unwrap(); | ||
| sender.send_trailers(trailers).await.unwrap(); | ||
| }); | ||
|
|
||
| Response::builder() | ||
| .version(Version::HTTP_2) | ||
| .status(StatusCode::OK) | ||
| .header(CONTENT_TYPE, HeaderValue::from_static(&CONTENT_TYPE_GRPC)) | ||
| .body(body) | ||
| .unwrap() | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters