Feature/service stats (#1267)

* Send message from service provider to stats service * Put some actual data in stats * Put stats sender on its own thread and send response data too * Use SQLite for storing stats * Add the data interval and timestamp * Fix clippy * Set description at boot * Guard stats service functionality under a feature for now * Make stats service address data into consts * Add README to network requester * Retrieve sql data in interval * Expose sql data via rocket rest api * Add entry to changelog
2022-05-16 12:06:15 +03:00
parent 82abfa5c5c
commit 0c66cc7393
21 changed files with 736 additions and 30 deletions
@@ -9,6 +9,7 @@
 - mixnet-contract: Replace all naked `-` with `saturating_sub`.
 - validator-api: add Swagger to document the REST API ([#1249]).
 - all: added network compilation target to `--help` (or `--version`) commands ([#1256]).
+- network-requester: send traffic statistics from all network requesters and receive it in a special network-requester that aggregates the data and exposes it via a rest API ([#1267]).

 ### Fixed

@@ -22,6 +23,7 @@
 [#1257]: https://github.com/nymtech/nym/pull/1257
 [#1260]: https://github.com/nymtech/nym/pull/1260
 [#1265]: https://github.com/nymtech/nym/pull/1265
+[#1267]: https://github.com/nymtech/nym/pull/1267

 ## [nym-wallet-v1.0.4](https://github.com/nymtech/nym/tree/nym-wallet-v1.0.4) (2022-05-04)

@@ -3165,18 +3165,23 @@ dependencies = [
 name = "nym-network-requester"
 version = "1.0.1"
 dependencies = [
+ "bincode",
 "clap 2.34.0",
 "dirs",
 "futures",
 "ipnetwork",
 "log",
+ "network-defaults",
 "nymsphinx",
 "ordered-buffer",
 "pretty_env_logger",
 "proxy-helpers",
 "publicsuffix",
 "rand 0.7.3",
+ "serde",
 "socks5-requests",
+ "sqlx",
+ "thiserror",
 "tokio",
 "tokio-tungstenite",
 "websocket-requests",
@@ -5219,6 +5224,7 @@ dependencies = [
 "bitflags",
 "byteorder",
 "bytes",
+ "chrono",
 "crc",
 "crossbeam-queue",
 "either",
@@ -16,7 +16,7 @@ use proxy_helpers::connection_controller::{
 };
 use proxy_helpers::proxy_runner::ProxyRunner;
 use rand::RngCore;
-use socks5_requests::{ConnectionId, RemoteAddress, Request};
+use socks5_requests::{ConnectionId, Message, RemoteAddress, Request};
 use std::io;
 use std::net::SocketAddr;
 use std::pin::Pin;
@@ -224,8 +224,9 @@ impl SocksClient {

    async fn send_connect_to_mixnet(&mut self, remote_address: RemoteAddress) {
        let req = Request::new_connect(self.connection_id, remote_address, self.self_address);
+        let msg = Message::Request(req);

-        let input_message = InputMessage::new_fresh(self.service_provider, req.into_bytes(), false);
+        let input_message = InputMessage::new_fresh(self.service_provider, msg.into_bytes(), false);
        self.input_sender.unbounded_send(input_message).unwrap();
    }

@@ -252,7 +253,8 @@ impl SocksClient {
        )
        .run(move |conn_id, read_data, socket_closed| {
            let provider_request = Request::new_send(conn_id, read_data, socket_closed);
-            InputMessage::new_fresh(recipient, provider_request.into_bytes(), false)
+            let provider_message = Message::Request(provider_request);
+            InputMessage::new_fresh(recipient, provider_message.into_bytes(), false)
        })
        .await
        .into_inner();
@@ -1,5 +1,7 @@
+pub mod msg;
 pub mod request;
 pub mod response;

+pub use msg::*;
 pub use request::*;
 pub use response::*;
@@ -0,0 +1,63 @@
+// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::request::{Request, RequestError};
+use crate::response::{Response, ResponseError};
+
+#[derive(Debug)]
+pub enum MessageError {
+    Request(RequestError),
+    Response(ResponseError),
+    NoData,
+    UnknownMessageType,
+}
+
+impl std::fmt::Display for MessageError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            MessageError::Request(r) => write!(f, "{}", r),
+            MessageError::Response(r) => write!(f, "{:?}", r),
+            MessageError::NoData => write!(f, "no data provided"),
+            MessageError::UnknownMessageType => write!(f, "unknown message type received"),
+        }
+    }
+}
+
+pub enum Message {
+    Request(Request),
+    Response(Response),
+}
+
+impl Message {
+    const REQUEST_FLAG: u8 = 0;
+    const RESPONSE_FLAG: u8 = 1;
+
+    pub fn try_from_bytes(b: &[u8]) -> Result<Message, MessageError> {
+        if b.is_empty() {
+            return Err(MessageError::NoData);
+        }
+
+        if b[0] == Self::REQUEST_FLAG {
+            Request::try_from_bytes(&b[1..])
+                .map(Message::Request)
+                .map_err(MessageError::Request)
+        } else if b[0] == Self::RESPONSE_FLAG {
+            Response::try_from_bytes(&b[1..])
+                .map(Message::Response)
+                .map_err(MessageError::Response)
+        } else {
+            Err(MessageError::UnknownMessageType)
+        }
+    }
+
+    pub fn into_bytes(self) -> Vec<u8> {
+        match self {
+            Self::Request(r) => std::iter::once(Self::REQUEST_FLAG)
+                .chain(r.into_bytes().iter().cloned())
+                .collect(),
+            Self::Response(r) => std::iter::once(Self::RESPONSE_FLAG)
+                .chain(r.into_bytes().iter().cloned())
+                .collect(),
+        }
+    }
+}
@@ -10,23 +10,36 @@ edition = "2021"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html

 [dependencies]
+bincode = "1.3"
 clap = "2.33.0"
 dirs = "3.0"
 futures = "0.3"
+ipnetwork = "0.17"
 log = "0.4"
 pretty_env_logger = "0.4"
-tokio = { version = "1.4", features = [ "net", "rt-multi-thread", "macros" ] }
-tokio-tungstenite = "0.14"
 publicsuffix = "1.5"
-ipnetwork = "0.17"
+rocket = { version = "0.5.0-rc.1", features = ["json"], optional = true }
+serde = { version = "1.0", features = ["derive"] }
+sqlx = { version = "0.5", features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate", "chrono"]}
+thiserror = "1"
+tokio = { version = "1.4", features = [ "net", "rt-multi-thread", "macros", "time" ] }
+tokio-tungstenite = "0.14"


 # internal
+network-defaults = { path = "../../common/network-defaults" }
 nymsphinx = { path = "../../common/nymsphinx" }
 ordered-buffer = {path = "../../common/socks5/ordered-buffer"}
-socks5-requests = { path = "../../common/socks5/requests" }
 proxy-helpers = { path = "../../common/socks5/proxy-helpers" }
+socks5-requests = { path = "../../common/socks5/requests" }
 websocket-requests = { path = "../../clients/native/websocket-requests" }

 [dev-dependencies]
 rand = "0.7"
+
+[build-dependencies]
+sqlx = { version = "0.5", features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate"] }
+tokio = { version = "1.4", features = ["rt-multi-thread", "macros"] }
+
+[features]
+stats-service = ["rocket"]
@@ -0,0 +1,26 @@
+Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
+SPDX-License-Identifier: Apache-2.0
+-->
+
+## Network requester
+
+The network requester is used to interpret socks5 client messages that need to
+be proxied to a running service i.e. a host and a port.
+
+If you have a service that you want to expose to the mixnet, you'd need to
+first run the native client and provide the client address to your users that
+will use it in their socks5 configuration.
+
+After starting the native client, start the network requester and configure it,
+setting your service's endpoint in  
+`${HOME}/.nym/service-providers/network-requester/allowed.list`
+
+Running in `open-proxy` mode allows any traffic to be proxied by the network
+requester.
+
+### Statistics service
+The network requester can be build and ran as a gatherer of statistics from all
+the other network requesters on the mixnet. For that, build the binary with the
+`stats-service` feature enabled. The native client address that corresponds to
+this network requester would have to be built into the constants of all the
+other network requesters that are sending the data.
@@ -0,0 +1,28 @@
+// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use sqlx::{Connection, SqliteConnection};
+use std::env;
+
+#[tokio::main]
+async fn main() {
+    let out_dir = env::var("OUT_DIR").unwrap();
+    let database_path = format!("{}/network-requester-example.sqlite", out_dir);
+
+    let mut conn = SqliteConnection::connect(&*format!("sqlite://{}?mode=rwc", database_path))
+        .await
+        .expect("Failed to create SQLx database connection");
+
+    sqlx::migrate!("./migrations")
+        .run(&mut conn)
+        .await
+        .expect("Failed to perform SQLx migrations");
+
+    #[cfg(target_family = "unix")]
+    println!("cargo:rustc-env=DATABASE_URL=sqlite://{}", &database_path);
+
+    #[cfg(target_family = "windows")]
+    // for some strange reason we need to add a leading `/` to the windows path even though it's
+    // not a valid windows path... but hey, it works...
+    println!("cargo:rustc-env=DATABASE_URL=sqlite:///{}", &database_path);
+}
@@ -0,0 +1,14 @@
+/*
+ * Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+CREATE TABLE mixnet_statistics
+(
+    id                         INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
+    service_description        VARCHAR NOT NULL,
+    request_processed_bytes    INTEGER NOT NULL,
+    response_processed_bytes   INTEGER NOT NULL,
+    interval_seconds           INTEGER NOT NULL,
+    timestamp                  DATETIME NOT NULL
+);
@@ -3,6 +3,7 @@

 use crate::allowed_hosts::{HostsStore, OutboundRequestFilter};
 use crate::connection::Connection;
+use crate::statistics::{Statistics, StatsData, Timer};
 use crate::websocket;
 use crate::websocket::TSWebsocketStream;
 use futures::channel::mpsc;
@@ -12,9 +13,11 @@ use log::*;
 use nymsphinx::addressing::clients::Recipient;
 use nymsphinx::receiver::ReconstructedMessage;
 use proxy_helpers::connection_controller::{Controller, ControllerCommand, ControllerSender};
-use socks5_requests::{ConnectionId, Request, Response};
+use socks5_requests::{ConnectionId, Message as Socks5Message, Request, Response};
 use std::path::PathBuf;
 use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::Arc;
+use tokio::sync::RwLock;
 use tokio_tungstenite::tungstenite::protocol::Message;
 use websocket::WebsocketConnectionError;
 use websocket_requests::{requests::ClientRequest, responses::ServerResponse};
@@ -24,12 +27,19 @@ static ACTIVE_PROXIES: AtomicUsize = AtomicUsize::new(0);

 pub struct ServiceProvider {
    listening_address: String,
+    description: String,
+    #[cfg(feature = "stats-service")]
+    db_path: PathBuf,
    outbound_request_filter: OutboundRequestFilter,
    open_proxy: bool,
 }

 impl ServiceProvider {
-    pub fn new(listening_address: String, open_proxy: bool) -> ServiceProvider {
+    pub fn new(
+        listening_address: String,
+        description: String,
+        open_proxy: bool,
+    ) -> ServiceProvider {
        let allowed_hosts = HostsStore::new(
            HostsStore::default_base_dir(),
            PathBuf::from("allowed.list"),
@@ -39,9 +49,19 @@ impl ServiceProvider {
            HostsStore::default_base_dir(),
            PathBuf::from("unknown.list"),
        );
+
+        #[cfg(feature = "stats-service")]
+        let db_path = HostsStore::default_base_dir()
+            .join("service-providers")
+            .join("network-requester")
+            .join("db.sqlite");
+
        let outbound_request_filter = OutboundRequestFilter::new(allowed_hosts, unknown_hosts);
        ServiceProvider {
            listening_address,
+            description,
+            #[cfg(feature = "stats-service")]
+            db_path,
            outbound_request_filter,
            open_proxy,
        }
@@ -52,13 +72,18 @@ impl ServiceProvider {
    async fn mixnet_response_listener(
        mut websocket_writer: SplitSink<TSWebsocketStream, Message>,
        mut mix_reader: mpsc::UnboundedReceiver<(Response, Recipient)>,
+        response_stats_data: &Arc<RwLock<StatsData>>,
    ) {
        // TODO: wire SURBs in here once they're available
        while let Some((response, return_address)) = mix_reader.next().await {
+            response_stats_data
+                .write()
+                .await
+                .processed(response.data.len() as u32);
            // make 'request' to native-websocket client
            let response_message = ClientRequest::Send {
                recipient: return_address,
-                message: response.into_bytes(),
+                message: Socks5Message::Response(response).into_bytes(),
                with_reply_surb: false,
            };

@@ -194,31 +219,50 @@ impl ServiceProvider {
            .unwrap()
    }

-    fn handle_proxy_request(
+    async fn handle_proxy_message(
        &mut self,
+        #[cfg(feature = "stats-service")] storage: &crate::storage::NetworkRequesterStorage,
        raw_request: &[u8],
        controller_sender: &mut ControllerSender,
        mix_input_sender: &mpsc::UnboundedSender<(Response, Recipient)>,
+        request_stats_data: &Arc<RwLock<StatsData>>,
    ) {
-        // try to treat each received mix message as a service provider request
-        let deserialized_request = match Request::try_from_bytes(raw_request) {
-            Ok(request) => request,
+        let deserialized_msg = match Socks5Message::try_from_bytes(raw_request) {
+            Ok(msg) => msg,
            Err(err) => {
-                error!("Failed to deserialized received request! - {}", err);
+                error!("Failed to deserialized received message! - {}", err);
                return;
            }
        };
+        match deserialized_msg {
+            Socks5Message::Request(deserialized_request) => match deserialized_request {
+                Request::Connect(req) => self.handle_proxy_connect(
+                    controller_sender,
+                    mix_input_sender,
+                    req.conn_id,
+                    req.remote_addr,
+                    req.return_address,
+                ),

-        match deserialized_request {
-            Request::Connect(req) => self.handle_proxy_connect(
-                controller_sender,
-                mix_input_sender,
-                req.conn_id,
-                req.remote_addr,
-                req.return_address,
-            ),
-            Request::Send(conn_id, data, closed) => {
-                self.handle_proxy_send(controller_sender, conn_id, data, closed)
+                Request::Send(conn_id, data, closed) => {
+                    request_stats_data
+                        .write()
+                        .await
+                        .processed(data.len() as u32);
+                    self.handle_proxy_send(controller_sender, conn_id, data, closed)
+                }
+            },
+            Socks5Message::Response(_deserialized_response) =>
+            {
+                #[cfg(feature = "stats-service")]
+                match crate::statistics::StatsMessage::from_bytes(&_deserialized_response.data) {
+                    Ok(data) => {
+                        if let Err(e) = storage.insert_service_statistics(data).await {
+                            error!("Could not store received statistics: {}", e);
+                        }
+                    }
+                    Err(e) => error!("Malformed statistics received: {}", e),
+                }
            }
        }
    }
@@ -234,19 +278,56 @@ impl ServiceProvider {
        // going to be used by `mixnet_response_listener`
        let (mix_input_sender, mix_input_receiver) = mpsc::unbounded::<(Response, Recipient)>();

+        let (mut timer_sender, timer_receiver) = Timer::new();
+        let interval = timer_sender.interval();
+        tokio::spawn(async move {
+            timer_sender.run().await;
+        });
+
        // controller for managing all active connections
        let (mut active_connections_controller, mut controller_sender) = Controller::new();
        tokio::spawn(async move {
            active_connections_controller.run().await;
        });

+        let mut stats = Statistics::new(self.description.clone(), interval, timer_receiver);
+        let request_stats_data = Arc::clone(stats.request_data());
+        let response_stats_data = Arc::clone(stats.response_data());
+        let mix_input_sender_clone = mix_input_sender.clone();
+        tokio::spawn(async move {
+            stats.run(&mix_input_sender_clone).await;
+        });
+
+        #[cfg(feature = "stats-service")]
+        let storage = crate::storage::NetworkRequesterStorage::init(&self.db_path)
+            .await
+            .expect("Could not create network requester storage");
+
+        #[cfg(feature = "stats-service")]
+        tokio::spawn(
+            rocket::build()
+                .mount(
+                    "/v1",
+                    rocket::routes![crate::storage::post_mixnet_statistics],
+                )
+                .manage(storage.clone())
+                .ignite()
+                .await
+                .expect("Could not ignite stats api service")
+                .launch(),
+        );
+
        // start the listener for mix messages
        tokio::spawn(async move {
-            Self::mixnet_response_listener(websocket_writer, mix_input_receiver).await;
+            Self::mixnet_response_listener(
+                websocket_writer,
+                mix_input_receiver,
+                &response_stats_data,
+            )
+            .await;
        });

        println!("\nAll systems go. Press CTRL-C to stop the server.");
-
        // for each incoming message from the websocket... (which in 99.99% cases is going to be a mix message)
        loop {
            let received = match Self::read_websocket_message(&mut websocket_reader).await {
@@ -260,7 +341,15 @@ impl ServiceProvider {
            let raw_message = received.message;
            // TODO: here be potential SURB (i.e. received.reply_SURB)

-            self.handle_proxy_request(&raw_message, &mut controller_sender, &mix_input_sender)
+            self.handle_proxy_message(
+                #[cfg(feature = "stats-service")]
+                &storage,
+                &raw_message,
+                &mut controller_sender,
+                &mix_input_sender,
+                &request_stats_data,
+            )
+            .await;
        }
    }

@@ -3,12 +3,19 @@

 use clap::{App, Arg, ArgMatches};

+use network_defaults::DEFAULT_WEBSOCKET_LISTENING_PORT;
+
 mod allowed_hosts;
 mod connection;
 mod core;
+mod statistics;
+#[cfg(feature = "stats-service")]
+mod storage;
 mod websocket;

 const OPEN_PROXY_ARG: &str = "open-proxy";
+const WS_PORT: &str = "websocket-port";
+const DESCRIPTION: &str = "description";

 fn parse_args<'a>() -> ArgMatches<'a> {
    App::new("Nym Network Requester")
@@ -20,6 +27,21 @@ fn parse_args<'a>() -> ArgMatches<'a> {
                .long(OPEN_PROXY_ARG)
                .short("o"),
        )
+        .arg(
+            Arg::with_name(WS_PORT)
+                .help("websocket port to bind to")
+                .long(WS_PORT)
+                .short("p")
+                .takes_value(true),
+        )
+        .arg(
+            Arg::with_name(DESCRIPTION)
+                .help("service description")
+                .long(DESCRIPTION)
+                .short("d")
+                .takes_value(true)
+                .default_value("undefined"),
+        )
        .get_matches()
 }

@@ -27,14 +49,22 @@ fn parse_args<'a>() -> ArgMatches<'a> {
 async fn main() {
    setup_logging();
    let matches = parse_args();
+
    let open_proxy = matches.is_present(OPEN_PROXY_ARG);
    if open_proxy {
        println!("\n\nYOU HAVE STARTED IN 'OPEN PROXY' MODE. ANYONE WITH YOUR CLIENT ADDRESS CAN MAKE REQUESTS FROM YOUR MACHINE. PLEASE QUIT IF YOU DON'T UNDERSTAND WHAT YOU'RE DOING.\n\n");
    }

-    let uri = "ws://localhost:1977";
+    let uri = format!(
+        "ws://localhost:{}",
+        matches
+            .value_of(WS_PORT)
+            .unwrap_or(&DEFAULT_WEBSOCKET_LISTENING_PORT.to_string())
+    );
+
+    let description = matches.value_of(DESCRIPTION).unwrap().to_string();
    println!("Starting socks5 service provider:");
-    let mut server = core::ServiceProvider::new(uri.into(), open_proxy);
+    let mut server = core::ServiceProvider::new(uri, description, open_proxy);
    server.run().await;
 }

@@ -0,0 +1,139 @@
+// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use futures::channel::mpsc;
+use futures::StreamExt;
+use log::*;
+use serde::{Deserialize, Serialize};
+use sqlx::types::chrono::{DateTime, Utc};
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::sync::RwLock;
+
+use nymsphinx::addressing::clients::{ClientEncryptionKey, ClientIdentity, Recipient};
+use nymsphinx::addressing::nodes::NodeIdentity;
+use socks5_requests::Response;
+
+use super::error::StatsError;
+
+const STATS_PROVIDER_CLIENT_IDENTITY: &str = "HqYWvCcB4sswYiyMj5Q8H5oc71kLf96vfrLK3npM7stH";
+const STATS_PROVIDER_ENCRYPTION_KEY: &str = "CoeC5dcqurgdxr5zcgU77nZBSBCc8ntCiwUivQ9TX3KT";
+const STATS_PROVIDER_GATEWAY_IDENTITY: &str = "E3mvZTHQCdBvhfr178Swx9g4QG3kkRUun7YnToLMcMbM";
+
+#[derive(Clone, Debug, Deserialize, Serialize)]
+pub struct StatsMessage {
+    pub description: String,
+    pub request_data: StatsData,
+    pub response_data: StatsData,
+    pub interval_seconds: u32,
+    pub timestamp: String,
+}
+
+impl StatsMessage {
+    pub fn to_bytes(&self) -> Result<Vec<u8>, StatsError> {
+        Ok(bincode::serialize(self)?)
+    }
+
+    #[cfg(feature = "stats-service")]
+    pub fn from_bytes(b: &[u8]) -> Result<Self, StatsError> {
+        Ok(bincode::deserialize(b)?)
+    }
+}
+
+#[derive(Clone, Debug, Deserialize, Serialize)]
+pub struct StatsData {
+    total_processed_bytes: u32,
+}
+
+impl StatsData {
+    pub fn new(total_processed_bytes: u32) -> Self {
+        StatsData {
+            total_processed_bytes,
+        }
+    }
+    pub fn processed(&mut self, bytes: u32) {
+        self.total_processed_bytes += bytes;
+    }
+
+    #[cfg(feature = "stats-service")]
+    pub fn total_processed_bytes(&self) -> u32 {
+        self.total_processed_bytes
+    }
+}
+
+pub struct Statistics {
+    description: String,
+    request_data: Arc<RwLock<StatsData>>,
+    response_data: Arc<RwLock<StatsData>>,
+    interval_seconds: u32,
+    timestamp: DateTime<Utc>,
+    timer_receiver: mpsc::Receiver<()>,
+    stats_provider_addr: Recipient,
+}
+
+impl Statistics {
+    pub fn new(
+        description: String,
+        interval_seconds: Duration,
+        timer_receiver: mpsc::Receiver<()>,
+    ) -> Self {
+        // those unwraps are ok because we set the strings in the constants above
+        let stats_provider_addr = Recipient::new(
+            ClientIdentity::from_base58_string(STATS_PROVIDER_CLIENT_IDENTITY).unwrap(),
+            ClientEncryptionKey::from_base58_string(STATS_PROVIDER_ENCRYPTION_KEY).unwrap(),
+            NodeIdentity::from_base58_string(STATS_PROVIDER_GATEWAY_IDENTITY).unwrap(),
+        );
+        Statistics {
+            description,
+            request_data: Arc::new(RwLock::new(StatsData::new(0))),
+            response_data: Arc::new(RwLock::new(StatsData::new(0))),
+            timestamp: Utc::now(),
+            interval_seconds: interval_seconds.as_secs() as u32,
+            timer_receiver,
+            stats_provider_addr,
+        }
+    }
+
+    pub fn request_data(&self) -> &Arc<RwLock<StatsData>> {
+        &self.request_data
+    }
+
+    pub fn response_data(&self) -> &Arc<RwLock<StatsData>> {
+        &self.response_data
+    }
+
+    pub async fn run(&mut self, mix_input_sender: &mpsc::UnboundedSender<(Response, Recipient)>) {
+        loop {
+            if self.timer_receiver.next().await == None {
+                error!("Timer thread has died. No more statistics will be sent");
+            } else {
+                let stats_message = StatsMessage {
+                    description: self.description.clone(),
+                    request_data: self.request_data.read().await.clone(),
+                    response_data: self.response_data.read().await.clone(),
+                    interval_seconds: self.interval_seconds,
+                    timestamp: self.timestamp.to_rfc3339(),
+                };
+                match stats_message.to_bytes() {
+                    Ok(data) => {
+                        trace!("Sending data to statistics service");
+                        mix_input_sender
+                            .unbounded_send((
+                                Response::new(0, data, false),
+                                self.stats_provider_addr,
+                            ))
+                            .unwrap();
+                    }
+                    Err(e) => error!("Statistics not sent: {}", e),
+                }
+                self.reset_stats().await;
+            }
+        }
+    }
+
+    async fn reset_stats(&mut self) {
+        self.request_data.write().await.total_processed_bytes = 0;
+        self.response_data.write().await.total_processed_bytes = 0;
+        self.timestamp = Utc::now();
+    }
+}
@@ -0,0 +1,10 @@
+// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use thiserror::Error;
+
+#[derive(Debug, Error)]
+pub enum StatsError {
+    #[error("Bincode error: {0}")]
+    BincodeError(#[from] bincode::Error),
+}
@@ -0,0 +1,9 @@
+// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+mod comm;
+mod error;
+mod timer;
+
+pub use comm::{Statistics, StatsData, StatsMessage};
+pub use timer::Timer;
@@ -0,0 +1,40 @@
+// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use futures::channel::mpsc;
+use std::time::Duration;
+use tokio::time;
+
+const STATISTICS_INTERVAL: Duration = Duration::from_secs(60);
+
+pub type TimerReceiver = mpsc::Receiver<()>;
+
+pub struct Timer {
+    interval: Duration,
+    stats_sender: mpsc::Sender<()>,
+}
+
+impl Timer {
+    pub fn new() -> (Self, TimerReceiver) {
+        let (stats_sender, stats_receiver) = mpsc::channel::<()>(1);
+        (
+            Timer {
+                interval: STATISTICS_INTERVAL,
+                stats_sender,
+            },
+            stats_receiver,
+        )
+    }
+
+    pub fn interval(&self) -> Duration {
+        self.interval
+    }
+
+    pub async fn run(&mut self) {
+        let mut interval = time::interval(self.interval);
+        loop {
+            interval.tick().await;
+            let _ = self.stats_sender.try_send(());
+        }
+    }
+}
@@ -0,0 +1,14 @@
+// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+#[derive(Debug, thiserror::Error)]
+pub enum NetworkRequesterStorageError {
+    #[error("SQL error - {0}")]
+    InternalDatabaseError(#[from] sqlx::Error),
+
+    #[error("SQL migrate error - {0}")]
+    DatabaseMigrateError(#[from] sqlx::migrate::MigrateError),
+
+    #[error("Timestamp could not be parsed")]
+    TimestampParse,
+}
@@ -0,0 +1,62 @@
+// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use sqlx::types::chrono::{DateTime, Utc};
+
+use crate::storage::models::MixnetStatistics;
+
+#[derive(Clone)]
+pub(crate) struct StorageManager {
+    pub(crate) connection_pool: sqlx::SqlitePool,
+}
+
+// all SQL goes here
+impl StorageManager {
+    /// Adds an entry for some statistical data.
+    ///
+    /// # Arguments
+    ///
+    /// * `service_description`: Description of the service that gathered the data.
+    /// * `request_processed_bytes`: Number of bytes for socks5 requests.
+    /// * `response_processed_bytes`: Number of bytes for socks5 responses.
+    /// * `interval_seconds`: Duration in seconds in which the data was gathered.
+    pub(super) async fn insert_service_statistics(
+        &self,
+        service_description: String,
+        request_processed_bytes: u32,
+        response_processed_bytes: u32,
+        interval_seconds: u32,
+        timestamp: DateTime<Utc>,
+    ) -> Result<(), sqlx::Error> {
+        sqlx::query!(
+            "INSERT INTO mixnet_statistics(service_description, request_processed_bytes, response_processed_bytes, interval_seconds, timestamp) VALUES (?, ?, ?, ?, ?)",
+                service_description,
+                request_processed_bytes,
+                response_processed_bytes,
+                interval_seconds,
+                timestamp,
+            ).execute(&self.connection_pool).await?;
+        Ok(())
+    }
+
+    /// Returns data submitted within the provided time interval.
+    ///
+    /// # Arguments
+    ///
+    /// * `since`: indicates the lower bound timestamp for the data
+    /// * `until`: indicates the upper bound timestamp for the data
+    pub(super) async fn get_service_statistics_in_interval(
+        &self,
+        since: DateTime<Utc>,
+        until: DateTime<Utc>,
+    ) -> Result<Vec<MixnetStatistics>, sqlx::Error> {
+        sqlx::query_as!(
+            MixnetStatistics,
+            "SELECT * FROM mixnet_statistics WHERE timestamp BETWEEN ? AND ?",
+            since,
+            until
+        )
+        .fetch_all(&self.connection_pool)
+        .await
+    }
+}
@@ -0,0 +1,92 @@
+// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use log::*;
+use sqlx::types::chrono::{DateTime, Utc};
+use sqlx::ConnectOptions;
+use std::path::PathBuf;
+
+use crate::statistics::StatsMessage;
+use crate::storage::error::NetworkRequesterStorageError;
+use crate::storage::manager::StorageManager;
+use crate::storage::models::MixnetStatistics;
+pub(crate) use crate::storage::routes::post_mixnet_statistics;
+
+mod error;
+mod manager;
+mod models;
+mod routes;
+
+// note that clone here is fine as upon cloning the same underlying pool will be used
+#[derive(Clone)]
+pub(crate) struct NetworkRequesterStorage {
+    manager: StorageManager,
+}
+
+impl NetworkRequesterStorage {
+    pub async fn init(database_path: &PathBuf) -> Result<Self, NetworkRequesterStorageError> {
+        let mut opts = sqlx::sqlite::SqliteConnectOptions::new()
+            .filename(database_path)
+            .create_if_missing(true);
+
+        opts.disable_statement_logging();
+
+        let connection_pool = sqlx::SqlitePool::connect_with(opts).await?;
+
+        sqlx::migrate!("./migrations").run(&connection_pool).await?;
+        info!("Database migration finished!");
+
+        let storage = NetworkRequesterStorage {
+            manager: StorageManager { connection_pool },
+        };
+
+        Ok(storage)
+    }
+
+    /// Adds an entry for some statistical data.
+    ///
+    /// # Arguments
+    ///
+    /// * `msg`: Message containing the statistical data.
+    pub(super) async fn insert_service_statistics(
+        &self,
+        msg: StatsMessage,
+    ) -> Result<(), NetworkRequesterStorageError> {
+        let timestamp: DateTime<Utc> = DateTime::parse_from_rfc3339(&msg.timestamp)
+            .map_err(|_| NetworkRequesterStorageError::TimestampParse)?
+            .into();
+        Ok(self
+            .manager
+            .insert_service_statistics(
+                msg.description,
+                msg.request_data.total_processed_bytes(),
+                msg.response_data.total_processed_bytes(),
+                msg.interval_seconds,
+                timestamp,
+            )
+            .await?)
+    }
+
+    /// Returns data submitted within the provided time interval.
+    ///
+    /// # Arguments
+    ///
+    /// * `since`: indicates the lower bound timestamp for the data, RFC 3339 format
+    /// * `until`: indicates the upper bound timestamp for the data, RFC 3339 format
+    pub(super) async fn get_service_statistics_in_interval(
+        &self,
+        since: &str,
+        until: &str,
+    ) -> Result<Vec<MixnetStatistics>, NetworkRequesterStorageError> {
+        let since = DateTime::parse_from_rfc3339(since)
+            .map_err(|_| NetworkRequesterStorageError::TimestampParse)?
+            .into();
+        let until = DateTime::parse_from_rfc3339(until)
+            .map_err(|_| NetworkRequesterStorageError::TimestampParse)?
+            .into();
+        Ok(self
+            .manager
+            .get_service_statistics_in_interval(since, until)
+            .await?)
+    }
+}
@@ -0,0 +1,15 @@
+// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use sqlx::types::chrono::NaiveDateTime;
+
+// Internally used struct to catch results from the database to get mixnet statistics
+pub(crate) struct MixnetStatistics {
+    #[allow(dead_code)]
+    pub(crate) id: i64,
+    pub(crate) service_description: String,
+    pub(crate) request_processed_bytes: i64,
+    pub(crate) response_processed_bytes: i64,
+    pub(crate) interval_seconds: i64,
+    pub(crate) timestamp: NaiveDateTime,
+}
@@ -0,0 +1,49 @@
+// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use rocket::serde::json::Json;
+use rocket::State;
+use serde::{Deserialize, Serialize};
+
+use crate::statistics::{StatsData, StatsMessage};
+use crate::storage::NetworkRequesterStorage;
+
+#[derive(Clone, Serialize, Deserialize, Debug)]
+pub struct MixnetStatisticsRequest {
+    // date, RFC 3339 format
+    since: String,
+    // date, RFC 3339 format
+    until: String,
+}
+
+#[derive(Clone, Serialize, Deserialize, Debug)]
+pub struct MixnetStatisticsResponse {
+    data: Vec<StatsMessage>,
+}
+
+#[rocket::post("/mixnet-statistics", data = "<mixnet_statistics_request>")]
+pub(crate) async fn post_mixnet_statistics(
+    mixnet_statistics_request: Json<MixnetStatisticsRequest>,
+    storage: &State<NetworkRequesterStorage>,
+) -> Json<MixnetStatisticsResponse> {
+    let mixnet_statistics = storage
+        .get_service_statistics_in_interval(
+            &mixnet_statistics_request.since,
+            &mixnet_statistics_request.until,
+        )
+        .await
+        .unwrap()
+        .into_iter()
+        .map(|data| StatsMessage {
+            description: data.service_description,
+            request_data: StatsData::new(data.request_processed_bytes as u32),
+            response_data: StatsData::new(data.response_processed_bytes as u32),
+            interval_seconds: data.interval_seconds as u32,
+            timestamp: data.timestamp.to_string(),
+        })
+        .collect();
+
+    Json(MixnetStatisticsResponse {
+        data: mixnet_statistics,
+    })
+}