forked from vectordotdev/vector
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
chore(statsd sink): refactor
statsd
sink to stream-based style (vec…
…tordotdev#16199) This PR completely refactors the `statsd` sink in the "new style", which uses stream-based combinators to build a `Stream` implementation that drives component behavior. At a high-level, the PR is indeed for refactoring the sink, but ultimately includes as much, if not more, refactoring work around establishing reusable `Service`-based primitives for building other sinks like `statsd` i.e. `socket` or `syslog`. ## Reviewer Notes I've mostly copied the existing shared socket sink types -- `TcpSinkConfig`, etc -- and existing socket services -- `UdpService` -- and created consistent versions of them for TCP, UDP, and Unix Domain sockets. This includes a configuration type that is `Configurable`-compatible for all of them, with socket-specific configurations[1] and then methods for generating both the `Service` implementation and a `Healthcheck` implementation. Ultimately, this should form the basis of other sink refactors that use sockets directly (`socket`, `syslog`, etc) but it may be desirable to do some more trait-ifying to avoid some of the necessary boilerplate introduced here. ## Remaining Work - [x] fix normalizer unit tests + add one for sketches
- Loading branch information
Showing
31 changed files
with
2,346 additions
and
786 deletions.
There are no files selected for viewing
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 |
---|---|---|
|
@@ -52,6 +52,7 @@ Enot | |
Evercoss | ||
Explay | ||
FAQs | ||
FQDNs | ||
Fabro | ||
Figma | ||
|
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
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
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,52 @@ | ||
//! Networking-related helper functions. | ||
|
||
use std::{io, time::Duration}; | ||
|
||
use socket2::{SockRef, TcpKeepalive}; | ||
use tokio::net::TcpStream; | ||
|
||
/// Sets the receive buffer size for a socket. | ||
/// | ||
/// This is the equivalent of setting the `SO_RCVBUF` socket setting directly. | ||
/// | ||
/// # Errors | ||
/// | ||
/// If there is an error setting the receive buffer size on the given socket, or if the value given | ||
/// as the socket is not a valid socket, an error variant will be returned explaining the underlying | ||
/// I/O error. | ||
pub fn set_receive_buffer_size<'s, S>(socket: &'s S, size: usize) -> io::Result<()> | ||
where | ||
SockRef<'s>: From<&'s S>, | ||
{ | ||
SockRef::from(socket).set_recv_buffer_size(size) | ||
} | ||
|
||
/// Sets the send buffer size for a socket. | ||
/// | ||
/// This is the equivalent of setting the `SO_SNDBUF` socket setting directly. | ||
/// | ||
/// # Errors | ||
/// | ||
/// If there is an error setting the send buffer size on the given socket, or if the value given | ||
/// as the socket is not a valid socket, an error variant will be returned explaining the underlying | ||
/// I/O error. | ||
pub fn set_send_buffer_size<'s, S>(socket: &'s S, size: usize) -> io::Result<()> | ||
where | ||
SockRef<'s>: From<&'s S>, | ||
{ | ||
SockRef::from(socket).set_send_buffer_size(size) | ||
} | ||
|
||
/// Sets the TCP keepalive behavior on a socket. | ||
/// | ||
/// This is the equivalent of setting the `SO_KEEPALIVE` and `TCP_KEEPALIVE` socket settings | ||
/// directly. | ||
/// | ||
/// # Errors | ||
/// | ||
/// If there is an error with either enabling keepalive probes or setting the TCP keepalive idle | ||
/// timeout on the given socket, an error variant will be returned explaining the underlying I/O | ||
/// error. | ||
pub fn set_keepalive(socket: &TcpStream, ttl: Duration) -> io::Result<()> { | ||
SockRef::from(socket).set_tcp_keepalive(&TcpKeepalive::new().with_time(ttl)) | ||
} |
Oops, something went wrong.