Merge pull request #40 from Overmuse/SR/remove_reporting

factor out reporting layer

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "order-manager"
-version = "6.1.7"
+version = "7.0.0"
 authors = ["Sebastian Rollen <seb@overmu.se>"]
 edition = "2018"
 
@@ -14,9 +14,7 @@ config = "0.10.1"
 dotenv = "0.15.0"
 futures = "0.3.15"
 kafka-settings = {git = "ssh://git@github.com/Overmuse/kafka-settings.git", tag = "v0.3.1" }
-lazy_static = "1.4.0"
 num-traits = "0.2.14"
-prometheus = "0.12.0"
 rdkafka = { version = "0.26.0", features = ["ssl-vendored"] }
 refinery = { version = "0.5.0", features = ["tokio-postgres"] }
 rust_decimal = { version = "1.14.1", features = ["tokio-pg"] }

src/lib.rs

@@ -8,7 +8,6 @@ use tracing::error;
 mod db;
 mod intent_scheduler;
 pub mod manager;
-mod metrics;
 mod order_sender;
 mod settings;
 pub mod types;

src/manager/order_updates.rs

@@ -1,7 +1,6 @@
 use super::OrderManager;
 use crate::db;
-use crate::metrics::{GROSS_TRADE_AMOUNT, NET_INVESTMENT_AMOUNT, NUM_TRADES};
-use crate::types::{split_lot, Allocation, Lot, Owner};
+use crate::types::{split_lot, Allocation, Lot};
 use alpaca::{Event, OrderEvent, Side};
 use anyhow::{anyhow, Context, Result};
 use chrono::{DateTime, Utc};
@@ -33,10 +32,6 @@ impl OrderManager {
                 price, timestamp, ..
             } => {
                 debug!("Order filled");
-                NUM_TRADES.with_label_values(&[&ticker]).inc();
-                GROSS_TRADE_AMOUNT
-                    .with_label_values(&[&ticker])
-                    .inc_by((qty * price).abs().to_f64().unwrap());
                 let new_lot = self
                     .make_lot(&id, ticker, timestamp, price, qty)
                     .await
@@ -141,13 +136,6 @@ impl OrderManager {
             .context("Failed to get claim")?;
         let allocations = split_lot(&claims, &lot);
         for allocation in allocations {
-            let owner = match allocation.owner {
-                Owner::House => "house",
-                Owner::Strategy(ref strat, _) => &strat,
-            };
-            NET_INVESTMENT_AMOUNT
-                .with_label_values(&[owner, &allocation.ticker])
-                .add(allocation.basis.to_f64().unwrap_or(0.0));
             self.adjust_claim(&allocation)
                 .await
                 .context("Failed to adjust claim")?;

src/webserver.rs

@@ -1,11 +1,9 @@
 use crate::db;
-use crate::metrics::{register_custom_metrics, REGISTRY};
 use crate::types::Owner;
 use std::convert::Infallible;
 use std::net::{Ipv4Addr, SocketAddrV4};
 use std::sync::Arc;
 use tokio_postgres::Client;
-use tracing::warn;
 use uuid::Uuid;
 use warp::reply::{json, Reply};
 use warp::{any, body, get, path, put, reject, serve, Filter, Rejection};
@@ -48,43 +46,8 @@ async fn get_pending_orders(db: Db) -> Result<impl Reply, Rejection> {
     Ok(json(&pending_orders))
 }
 
-async fn metrics_handler() -> Result<impl Reply, Rejection> {
-    use prometheus::Encoder;
-    let encoder = prometheus::TextEncoder::new();
-
-    let mut buffer = Vec::new();
-    if let Err(e) = encoder.encode(&REGISTRY.gather(), &mut buffer) {
-        warn!("could not encode custom metrics: {}", e);
-    };
-    let mut res = match String::from_utf8(buffer.clone()) {
-        Ok(v) => v,
-        Err(e) => {
-            warn!("custom metrics could not be from_utf8'd: {}", e);
-            String::default()
-        }
-    };
-    buffer.clear();
-
-    let mut buffer = Vec::new();
-    if let Err(e) = encoder.encode(&prometheus::gather(), &mut buffer) {
-        warn!("could not encode prometheus metrics: {}", e);
-    };
-    let res_custom = match String::from_utf8(buffer.clone()) {
-        Ok(v) => v,
-        Err(e) => {
-            warn!("prometheus metrics could not be from_utf8'd: {}", e);
-            String::default()
-        }
-    };
-    buffer.clear();
-
-    res.push_str(&res_custom);
-    Ok(res)
-}
-
 #[tracing::instrument(skip(db))]
 pub async fn run(port: u16, db: Db) {
-    register_custom_metrics();
     let health = path!("health").map(|| "");
     let get_allocations = path("allocations")
         .and(get())
@@ -107,15 +70,13 @@ pub async fn run(port: u16, db: Db) {
         .and(get())
         .and(with_db(db.clone()))
         .and_then(get_pending_orders);
-    let metrics = path("metrics").and_then(metrics_handler);
     let routes = get()
         .and(health)
         .or(get_allocations)
         .or(set_allocation_owner)
         .or(lots)
         .or(claims)
-        .or(pending_orders)
-        .or(metrics);
+        .or(pending_orders);
     let address = SocketAddrV4::new(Ipv4Addr::new(0, 0, 0, 0), port);
     serve(routes).run(address).await
 }