broker stats

Author: aep

Cargo.lock

@@ -19,6 +19,7 @@ futurize            = "0.5.0"
 futurize-derive     = "0.5.0"
 gcmap               = "0.1.4"
 lazy_static         = "1.2.0"
+dotenv              = "0.13.0"
 
 [dependencies.env_logger]
 version = "0.5.12"

broker/Cargo.toml

@@ -18,6 +18,7 @@ use tokio;
 use tokio::net::UdpSocket;
 use carrier::transport;
 use xlog;
+use stats;
 
 pub struct Listener {
     ep:      endpoint::Endpoint,
@@ -45,7 +46,7 @@ pub fn listen(secret: identity::Secret) -> Result<(Listener, shadow::broker::Han
     let mut work = ep.work.clone();
     work.try_send(endpoint::EndpointWorkerCmd::InsertChannel(
         0,
-        endpoint::ChannelBus::User { inc: tx },
+        endpoint::ChannelBus::User { inc: tx, tc: stats::PacketCounter::default() },
     ))?;
 
     Ok((
@@ -115,7 +116,11 @@ impl ChannelHandshake {
 
     pub fn accept(self, secret: identity::Secret) -> impl Future<Item = channel::Channel, Error = Error> {
         let (tx, rx) = mpsc::channel(100);
-        let bus = endpoint::ChannelBus::User { inc: tx };
+        let tc = stats::PacketCounter{
+            initiator: Some(self.identity.clone()),
+            ..stats::PacketCounter::default()
+        };
+        let bus = endpoint::ChannelBus::User { inc: tx, tc };
 
         let (route_tx, route_rx) = oneshot::channel();
 

broker/src/listener.rs

@@ -12,33 +12,46 @@ extern crate futurize_derive;
 extern crate gcmap;
 #[macro_use]
 extern crate lazy_static;
+extern crate dotenv;
 
 use carrier::*;
 use failure::Error;
 use futures::{Future, Stream};
 use std::env;
+use dotenv::dotenv;
+use std::collections::HashSet;
 
 mod ptrmap;
 mod shadow;
 mod listener;
 mod xlog;
 
 pub fn main() {
+    dotenv().ok();
+
     if let Err(_) = env::var("RUST_LOG") {
         env::set_var("RUST_LOG", "carrier=info,carrier-broker");
     }
     env_logger::init();
 
+
+    let coordinators : HashSet<identity::Identity> = env::var("COORDINATOR_IDENTITIES")
+        .expect("must set env COORDINATOR_IDENTITIES")
+        .split(":")
+        .map(|v|v.parse().expect("parsing COORDINATOR_IDENTITIES"))
+        .collect();
+
     let secrets = keystore::Secrets::load().unwrap();
     tokio::run(futures::lazy(move || {
-        broker(secrets.identity).map_err(|e| error!("{}", e))
+        broker(secrets.identity, coordinators).map_err(|e| error!("{}", e))
     }));
 }
 
-pub fn broker(secret: identity::Secret) -> impl Future<Item = (), Error = Error> {
+pub fn broker(secret: identity::Secret, coordinators: HashSet<identity::Identity>) -> impl Future<Item = (), Error = Error> {
     let (lst, sb) = listener::listen(secret.clone()).unwrap();
     let ep = lst.handle();
     lst.for_each(move |ch| {
+        let coordinators = coordinators.clone();
         info!("incomming channel {} {}", ch.identity(), ch.addr());
         let addr = ch.addr().clone();
         let sb = sb.clone();
@@ -47,7 +60,7 @@ pub fn broker(secret: identity::Secret) -> impl Future<Item = (), Error = Error>
             .accept(secret.clone())
             .and_then(move |ch| {
                 info!("accepted channel {} for route {}", ch.identity(), ch.route());
-                sb.dispatch(ep, ch, addr)
+                sb.dispatch(ep, ch, addr, coordinators)
             }).and_then(|_| {
                 info!("dispatch ended");
                 Ok(())

broker/src/main.rs

@@ -2,6 +2,7 @@ use carrier::channel;
 use carrier::endpoint;
 use failure::Error;
 use futures::sync::mpsc;
+use futures::sync::oneshot;
 use futures::{self, Future, Sink, Stream};
 use futurize;
 use gcmap::{HashMap, MarkOnDrop};
@@ -12,6 +13,8 @@ use ptrmap;
 use std::net::SocketAddr;
 use tokio;
 use xlog;
+use stats;
+use std::collections::HashSet;
 
 macro_rules! wrk_try {
     ($self:ident, $x:expr) => {
@@ -288,6 +291,7 @@ impl Broker {
 }
 
 impl broker::Worker for Broker {
+
     fn subscribe(
         mut self,
         identity: identity::Identity,
@@ -397,11 +401,13 @@ impl peer::Worker for Peer {
 }
 
 struct Srv {
-    endpoint: endpoint::Endpoint,
-    broker:   broker::Handle,
-    identity: identity::Identity,
-    worker:   peer::Handle,
-    ipaddr:   SocketAddr,
+    endpoint:       endpoint::Endpoint,
+    broker:         broker::Handle,
+    identity:       identity::Identity,
+    worker:         peer::Handle,
+    ipaddr:         SocketAddr,
+    coordinators:   HashSet<identity::Identity>,
+    epoch:          u64,
 }
 
 impl broker::Handle {
@@ -410,6 +416,7 @@ impl broker::Handle {
         endpoint: endpoint::Endpoint,
         mut channel: channel::Channel,
         ipaddr: SocketAddr,
+        coordinators: HashSet<identity::Identity>,
     ) -> impl Future<Item = (), Error = Error> {
         let lst = channel.listener().unwrap();
         let identity = channel.identity().clone();
@@ -429,12 +436,56 @@ impl broker::Handle {
             worker: handle,
             endpoint,
             ipaddr,
+            coordinators,
+            epoch: 0,
         };
         proto::Broker::dispatch(lst, srv)
     }
 }
 
 impl proto::Broker::Service for Srv {
+    fn epochsync(
+        &mut self,
+        _headers: Headers,
+        msg: proto::EpochSyncRequest,
+    ) -> Result<Box<Future<Item = proto::EpochSyncResponse, Error = Error> + Sync + Send + 'static>, Error> {
+
+        // check if the sender is a trusted coordinator
+        if !self.coordinators.contains(&self.identity) {
+            return Ok(Box::new(futures::future::ok(proto::EpochSyncResponse::default())));
+        }
+
+
+        info!("epoch sync from {} to {} by coordinator {}",
+              self.epoch, msg.epoch, self.identity);
+
+        let clear = {
+            if self.epoch != msg.epoch {
+                self.epoch = msg.epoch;
+                true
+            } else {
+                false
+            }
+        };
+
+
+        let (r_tx, r_rx) = oneshot::channel();
+        let ft = self
+            .endpoint.work.clone().send(endpoint::EndpointWorkerCmd::DumpStats(r_tx, clear))
+            .map_err(Error::from)
+            .and_then(move |_| {
+                r_rx
+                    .map_err(Error::from)
+                    .and_then(|dump|{
+                        Ok(proto::EpochSyncResponse{
+                            dump: Some(dump),
+                        })
+                    })
+            });
+
+        Ok(Box::new(ft))
+    }
+
     fn subscribe(
         &mut self,
         _headers: Headers,
@@ -500,15 +551,22 @@ impl proto::Broker::Service for Srv {
 
         let selfipaddr = self.ipaddr.clone();
         let selfidentity = self.identity.as_bytes().to_vec();
+        let selfidentity_ = self.identity.clone();
         let mut endpoint = self.endpoint.clone();
 
         let mut broker = self.broker.clone();
         let ft = futures::future::result(identity::Identity::from_bytes(msgidentity))
             .and_then(move |target| {
+                let tc = stats::PacketCounter{
+                    initiator: Some(selfidentity_),
+                    responder: Some(target.clone()),
+                    ..stats::PacketCounter::default()
+                };
                 broker.get_peer(target).and_then(move |maybe| {
                     if let Some((mut peer, ipaddr)) = maybe {
+
                         let ft = endpoint
-                            .proxy(selfipaddr, ipaddr)
+                            .proxy(selfipaddr, ipaddr, tc)
                             .and_then(move |proxy| {
                                 peer.connect(proto::PeerConnectRequest {
                                     identity:  selfidentity,

broker/src/shadow.rs

@@ -93,6 +93,11 @@ pub fn main_() -> Result<(), Error> {
 
         .subcommand(SubCommand::with_name("mkshadow").about("create a shadow address"))
         .subcommand(
+            SubCommand::with_name("sync")
+                .about("coordinate a broker epoch")
+                .arg(Arg::with_name("broker").takes_value(true).required(true).index(1))
+                .arg(Arg::with_name("epoch").takes_value(true).required(true).index(2))
+        ).subcommand(
             SubCommand::with_name("dns")
                 .about("create dns record")
                 .arg(Arg::with_name("priority").takes_value(true).required(true).index(2))
@@ -449,6 +454,16 @@ pub fn main_() -> Result<(), Error> {
             }
             Ok(())
         }
+        ("sync", Some(submatches)) => {
+            let secrets = keystore::Secrets::load()?;
+            let broker: std::net::IpAddr = submatches.value_of("broker").unwrap().to_string().parse().expect("broker ip");
+            let epoch: u64 = submatches.value_of("epoch").unwrap().to_string().parse().expect("epoch");
+
+            tokio::run(futures::lazy(move || {
+                sync(secrets.identity, broker, epoch).map_err(|e| error!("{}", e))
+            }));
+            Ok(())
+        }
         _ => unreachable!(),
     }
 }
@@ -667,4 +682,28 @@ pub fn update(
 }
 
 
+pub fn sync(
+    secret: identity::Secret,
+    broker: std::net::IpAddr,
+    epoch:  u64,
+) -> impl Future<Item = (), Error = Error> {
+    let domain = env::var("CARRIER_BROKER_DOMAIN").unwrap_or("2.carrier.devguard.io".to_string());
+    connect::connect_to_ip(domain, broker, secret.clone()).and_then(move |(_ep, mut brk, _sock, _addr)| {
+        brk.message("/carrier.broker.v1/broker/epochsync")
+            .unwrap()
+            .send(proto::EpochSyncRequest{
+                epoch,
+            }).flatten_stream()
+        .for_each(move |m: proto::EpochSyncResponse| {
+            if let Some(dump) = m.dump {
+                println!("{:#?}", dump);
+            }
+            Ok(())
+        })
+        .and_then(|_| {
+            drop(brk);
+            Ok(())
+        })
+    })
+}
 

cli/src/main.rs

@@ -71,11 +71,36 @@ message ConnectResponse {
     repeated Path paths     = 4;
 }
 
+
+message EpochSyncRequest {
+    uint64 epoch = 1;
+}
+
+
+
+message EpochDumpPacketCounter {
+    bytes   identity    = 1;
+    uint64  dcs         = 2;
+    uint64  bx          = 3;
+    uint64  tx          = 4;
+    uint64  rx          = 5;
+}
+
+message EpochDump {
+    repeated EpochDumpPacketCounter packets = 1;
+}
+
+
+message EpochSyncResponse {
+    EpochDump dump = 1;
+}
+
 service Broker {
     rpc subscribe   (SubscribeRequest)  returns (stream SubscribeChange) {}
     rpc publish     (PublishRequest)    returns (stream PublishChange)   {}
-
     rpc connect     (ConnectRequest)    returns (stream ConnectResponse) {}
+
+    rpc epochsync   (EpochSyncRequest)   returns (EpochSyncResponse)     {}
 }
 
 message PeerConnectRequest {