feat(new sink): initial websocket_server sink

Cargo.toml

@@ -730,6 +730,7 @@ sinks-logs = [
   "sinks-vector",
   "sinks-webhdfs",
   "sinks-websocket",
+  "sinks-websocket-server",
 ]
 sinks-metrics = [
   "sinks-appsignal",
@@ -798,6 +799,8 @@ sinks-statsd = ["sinks-utils-udp", "tokio-util/net"]
 sinks-utils-udp = []
 sinks-vector = ["sinks-utils-udp", "dep:tonic", "protobuf-build", "dep:prost"]
 sinks-websocket = ["dep:tokio-tungstenite"]
+sinks-websocket-server = ["dep:tokio-tungstenite", "sources-utils-http-auth",
+"sources-utils-http-error", "sources-utils-http-prelude"]
 sinks-webhdfs = ["dep:opendal"]
 
 # Identifies that the build is a nightly build

changelog.d/22213_websocket_server_sink.feature.md

@@ -0,0 +1,3 @@
+Add `websocket_server` sink that acts as a websocket server and broadcasts events to all clients.
+
+authors: esensar

src/internal_events/mod.rs

@@ -127,6 +127,8 @@ mod udp;
 mod unix;
 #[cfg(feature = "sinks-websocket")]
 mod websocket;
+#[cfg(feature = "sinks-websocket-server")]
+mod websocket_server;
 
 #[cfg(any(
     feature = "sources-file",
@@ -261,6 +263,8 @@ pub(crate) use self::throttle::*;
 pub(crate) use self::unix::*;
 #[cfg(feature = "sinks-websocket")]
 pub(crate) use self::websocket::*;
+#[cfg(feature = "sinks-websocket-server")]
+pub(crate) use self::websocket_server::*;
 #[cfg(windows)]
 pub(crate) use self::windows::*;
 pub use self::{

src/internal_events/websocket_server.rs

@@ -0,0 +1,101 @@
+use std::error::Error;
+use std::fmt::Debug;
+
+use futures::channel::mpsc::TrySendError;
+use metrics::{counter, gauge};
+use tokio_tungstenite::tungstenite::Message;
+use vector_lib::internal_event::InternalEvent;
+
+use vector_lib::internal_event::{error_stage, error_type};
+
+#[derive(Debug)]
+pub struct WsListenerConnectionEstablished {
+    pub client_count: usize,
+}
+
+impl InternalEvent for WsListenerConnectionEstablished {
+    fn emit(self) {
+        debug!(message = "Websocket client connected.");
+        counter!("connection_established_total").increment(1);
+        gauge!("active_clients").set(self.client_count as f64);
+    }
+
+    fn name(&self) -> Option<&'static str> {
+        Some("WsListenerConnectionEstablished")
+    }
+}
+
+#[derive(Debug)]
+pub struct WsListenerConnectionFailedError {
+    pub error: Box<dyn Error>,
+}
+
+impl InternalEvent for WsListenerConnectionFailedError {
+    fn emit(self) {
+        error!(
+            message = "WebSocket connection failed.",
+            error = %self.error,
+            error_code = "ws_connection_error",
+            error_type = error_type::CONNECTION_FAILED,
+            stage = error_stage::SENDING,
+            internal_log_rate_limit = true,
+        );
+        counter!(
+            "component_errors_total",
+            "error_code" => "ws_connection_failed",
+            "error_type" => error_type::CONNECTION_FAILED,
+            "stage" => error_stage::SENDING,
+        )
+        .increment(1);
+    }
+
+    fn name(&self) -> Option<&'static str> {
+        Some("WsListenerConnectionFailed")
+    }
+}
+
+#[derive(Debug)]
+pub struct WsListenerConnectionShutdown {
+    pub client_count: usize,
+}
+
+impl InternalEvent for WsListenerConnectionShutdown {
+    fn emit(self) {
+        warn!(message = "Client connection closed.");
+        counter!("connection_shutdown_total").increment(1);
+        gauge!("active_clients").set(self.client_count as f64);
+    }
+
+    fn name(&self) -> Option<&'static str> {
+        Some("WsListenerConnectionShutdown")
+    }
+}
+
+#[derive(Debug)]
+pub struct WsListenerSendError {
+    pub error: TrySendError<Message>,
+}
+
+impl InternalEvent for WsListenerSendError {
+    fn emit(self) {
+        error!(
+            message = "WebSocket message send error.",
+            error = %self.error,
+            error_code = "ws_connection_error",
+            error_type = error_type::WRITER_FAILED,
+            stage = error_stage::SENDING,
+            internal_log_rate_limit = true,
+        );
+        counter!(
+            "component_errors_total",
+            "error_code" => "ws_connection_error",
+            "error_type" => error_type::WRITER_FAILED,
+            "stage" => error_stage::SENDING,
+        )
+        .increment(1);
+    }
+
+    fn name(&self) -> Option<&'static str> {
+        Some("WsListenerConnectionError")
+    }
+}

src/sinks/mod.rs

@@ -108,6 +108,8 @@ pub mod vector;
 pub mod webhdfs;
 #[cfg(feature = "sinks-websocket")]
 pub mod websocket;
+#[cfg(feature = "sinks-websocket-server")]
+pub mod websocket_server;
 
 pub use vector_lib::{config::Input, sink::VectorSink};
 

src/sinks/websocket_server/config.rs

@@ -0,0 +1,91 @@
+use std::net::SocketAddr;
+
+use vector_lib::codecs::JsonSerializerConfig;
+use vector_lib::configurable::configurable_component;
+
+use crate::{
+    codecs::EncodingConfig,
+    config::{AcknowledgementsConfig, Input, SinkConfig, SinkContext},
+    sinks::{Healthcheck, VectorSink},
+    sources::util::HttpSourceAuthConfig,
+    tls::TlsEnableableConfig,
+};
+
+use super::sink::WebSocketListenerSink;
+
+/// Configuration for the `websocket_server` sink.
+#[configurable_component(sink(
+    "websocket_server",
+    "Deliver observability event data to websocket clients."
+))]
+#[derive(Clone, Debug)]
+pub struct WebSocketListenerSinkConfig {
+    /// The socket address to listen for connections on.
+    ///
+    /// It _must_ include a port.
+    #[configurable(metadata(docs::examples = "0.0.0.0:80"))]
+    #[configurable(metadata(docs::examples = "localhost:80"))]
+    pub address: SocketAddr,
+
+    #[configurable(derived)]
+    pub tls: Option<TlsEnableableConfig>,
+
+    #[configurable(derived)]
+    pub encoding: EncodingConfig,
+
+    #[configurable(derived)]
+    #[serde(
+        default,
+        deserialize_with = "crate::serde::bool_or_struct",
+        skip_serializing_if = "crate::serde::is_default"
+    )]
+    pub acknowledgements: AcknowledgementsConfig,
+
+    #[configurable(derived)]
+    pub auth: Option<HttpSourceAuthConfig>,
+}
+
+impl Default for WebSocketListenerSinkConfig {
+    fn default() -> Self {
+        Self {
+            address: "0.0.0.0:8080".parse().unwrap(),
+            encoding: JsonSerializerConfig::default().into(),
+            tls: None,
+            acknowledgements: Default::default(),
+            auth: None,
+        }
+    }
+}
+
+#[async_trait::async_trait]
+#[typetag::serde(name = "websocket_server")]
+impl SinkConfig for WebSocketListenerSinkConfig {
+    async fn build(&self, _cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> {
+        let ws_sink = WebSocketListenerSink::new(self.clone())?;
+
+        Ok((
+            VectorSink::from_event_streamsink(ws_sink),
+            Box::pin(async move { Ok(()) }),
+        ))
+    }
+
+    fn input(&self) -> Input {
+        Input::new(self.encoding.config().input_type())
+    }
+
+    fn acknowledgements(&self) -> &AcknowledgementsConfig {
+        &self.acknowledgements
+    }
+}
+
+impl_generate_config_from_default!(WebSocketListenerSinkConfig);
+
+#[cfg(test)]
+mod test {
+    use super::*;
+
+    #[test]
+    fn generate_config() {
+        crate::test_util::test_generate_config::<WebSocketListenerSinkConfig>();
+    }
+}

src/sinks/websocket_server/mod.rs

@@ -0,0 +1,4 @@
+mod config;
+mod sink;
+
+pub use config::WebSocketListenerSinkConfig;