mosuka · mosuka · Nov 22, 2019 · Nov 22, 2019 · Nov 22, 2019
diff --git a/Cargo.toml b/Cargo.toml
@@ -27,6 +27,7 @@ env_logger = "0.7.0"
 futures = "0.1.29"
 grpcio = { version = "0.4.5", features = [ "secure" ] }
 log = "0.4.8"
+prometheus = "0.7.0"
 protobuf = "2.8.0"
 raft = "0.4.3"
 rand = "0.7.2"

diff --git a/proto/indexpb.proto b/proto/indexpb.proto
@@ -11,6 +11,8 @@ service Index {
     }
     rpc Peers (indexrpcpb.IndexReq) returns (indexrpcpb.PeersResp) {
     }
+    rpc Metrics (indexrpcpb.IndexReq) returns (indexrpcpb.MetricsResp) {
+    }
     rpc Get (indexrpcpb.IndexReq) returns (indexrpcpb.GetResp) {
     }
     rpc Put (indexrpcpb.IndexReq) returns (indexrpcpb.PutResp) {

diff --git a/proto/indexrpcpb.proto b/proto/indexrpcpb.proto
@@ -19,6 +19,7 @@ enum ReqType {
     Join = 6;
     Leave = 7;
     Peers = 8;
+    Metrics = 9;
 }
 
 message IndexReq {
@@ -70,6 +71,11 @@ message PeersResp {
     RespErr err = 2;
 }
 
+message MetricsResp {
+    string value = 1;
+    RespErr err = 2;
+}
+
 message RaftDone {
     RespErr err = 1;
 }
diff --git a/src/client/client.rs b/src/client/client.rs
@@ -8,8 +8,8 @@ use raft::eraftpb::{ConfChange, ConfChangeType};
 
 use crate::proto::indexpb_grpc::IndexClient;
 use crate::proto::indexrpcpb::{
-    CommitResp, ConfChangeReq, DeleteResp, GetResp, IndexReq, PeersResp, PutResp, RaftDone,
-    ReqType, RespErr, SchemaResp, SearchResp,
+    CommitResp, ConfChangeReq, DeleteResp, GetResp, IndexReq, MetricsResp, PeersResp, PutResp,
+    RaftDone, ReqType, RespErr, SchemaResp, SearchResp,
 };
 
 pub fn create_client(addr: &str) -> IndexClient {
@@ -117,6 +117,31 @@ impl Clerk {
         }
     }
 
+    pub fn metrics(&mut self) -> String {
+        let mut req = IndexReq::new();
+        req.set_client_id(self.client_id);
+        req.set_req_type(ReqType::Metrics);
+        req.set_seq(self.request_seq);
+        self.request_seq += 1;
+
+        loop {
+            let reply = self.servers[self.leader_id]
+                .metrics(&req)
+                .unwrap_or_else(|_e| {
+                    let mut resp = MetricsResp::new();
+                    resp.set_err(RespErr::ErrWrongLeader);
+                    resp
+                });
+            match reply.err {
+                RespErr::OK => return reply.value,
+                RespErr::ErrWrongLeader => (),
+                RespErr::ErrNoKey => return String::from(""),
+            }
+            self.leader_id = (self.leader_id + 1) % self.servers.len();
+            thread::sleep(Duration::from_millis(100));
+        }
+    }
+
     pub fn get(&mut self, key: &str) -> String {
         let mut req = IndexReq::new();
         req.set_client_id(self.client_id);

diff --git a/src/cmd.rs b/src/cmd.rs
@@ -2,6 +2,7 @@ pub mod commit;
 pub mod delete;
 pub mod get;
 pub mod leave;
+pub mod metrics;
 pub mod peers;
 pub mod schema;
 pub mod search;

diff --git a/src/cmd/metrics.rs b/src/cmd/metrics.rs
@@ -0,0 +1,22 @@
+use clap::ArgMatches;
+
+use crate::client::client::{create_client, Clerk};
+use crate::util::log::set_logger;
+
+pub fn run_metrics_cli(matches: &ArgMatches) -> Result<(), String> {
+    set_logger();
+
+    let servers: Vec<_> = matches
+        .values_of("SERVERS")
+        .unwrap()
+        .map(|addr| create_client(addr))
+        .collect();
+
+    let client_id = rand::random();
+
+    let mut client = Clerk::new(&servers, client_id);
+    let value = client.metrics();
+    print!("{}", value);
+
+    Ok(())
+}
diff --git a/src/main.rs b/src/main.rs
@@ -10,6 +10,7 @@ use bayard::cmd::commit::run_commit_cli;
 use bayard::cmd::delete::run_delete_cli;
 use bayard::cmd::get::run_get_cli;
 use bayard::cmd::leave::run_leave_cli;
+use bayard::cmd::metrics::run_metrics_cli;
 use bayard::cmd::peers::run_peers_cli;
 use bayard::cmd::schema::run_schema_cli;
 use bayard::cmd::search::run_search_cli;
@@ -118,6 +119,27 @@ fn main() {
                         .takes_value(true),
                 )
         )
+        .subcommand(
+            SubCommand::with_name("metrics")
+                .name("metrics")
+                .setting(AppSettings::DeriveDisplayOrder)
+                .version(crate_version!())
+                .author(crate_authors!())
+                .about("Get metrics")
+                .arg(
+                    Arg::with_name("SERVERS")
+                        .help("The server addresses. Use `,` to separate address. Example: `127.0.0.1:5000,127.0.0.1:5001`")
+                        .short("s")
+                        .long("servers")
+                        .value_name("IP:PORT")
+                        .default_value("127.0.0.1:5000")
+                        .multiple(true)
+                        .use_delimiter(true)
+                        .require_delimiter(true)
+                        .value_delimiter(",")
+                        .takes_value(true),
+                )
+        )
         .subcommand(
             SubCommand::with_name("leave")
                 .name("leave")
@@ -340,6 +362,7 @@ fn main() {
     let run_cli = match subcommand {
         "serve" => run_serve_cli,
         "peers" => run_peers_cli,
+        "metrics" => run_metrics_cli,
         "leave" => run_leave_cli,
         "set" => run_set_cli,
         "get" => run_get_cli,

diff --git a/src/proto/indexpb.rs b/src/proto/indexpb.rs
@@ -28,44 +28,48 @@ const _PROTOBUF_VERSION_CHECK: () = ::protobuf::VERSION_2_8_0;
 
 static file_descriptor_proto_data: &'static [u8] = b"\
     \n\rindexpb.proto\x12\x07indexpb\x1a\x10indexrpcpb.proto\x1a\reraftpb.pr\
-    oto2\x86\x04\n\x05Index\x120\n\x04Raft\x12\x10.eraftpb.Message\x1a\x14.i\
+    oto2\xc2\x04\n\x05Index\x120\n\x04Raft\x12\x10.eraftpb.Message\x1a\x14.i\
     ndexrpcpb.RaftDone\"\0\x12C\n\x0eRaftConfChange\x12\x19.indexrpcpb.ConfC\
     hangeReq\x1a\x14.indexrpcpb.RaftDone\"\0\x126\n\x05Peers\x12\x14.indexrp\
-    cpb.IndexReq\x1a\x15.indexrpcpb.PeersResp\"\0\x122\n\x03Get\x12\x14.inde\
-    xrpcpb.IndexReq\x1a\x13.indexrpcpb.GetResp\"\0\x122\n\x03Put\x12\x14.ind\
-    exrpcpb.IndexReq\x1a\x13.indexrpcpb.PutResp\"\0\x128\n\x06Delete\x12\x14\
-    .indexrpcpb.IndexReq\x1a\x16.indexrpcpb.DeleteResp\"\0\x128\n\x06Commit\
-    \x12\x14.indexrpcpb.IndexReq\x1a\x16.indexrpcpb.CommitResp\"\0\x128\n\
-    \x06Search\x12\x14.indexrpcpb.IndexReq\x1a\x16.indexrpcpb.SearchResp\"\0\
-    \x128\n\x06Schema\x12\x14.indexrpcpb.IndexReq\x1a\x16.indexrpcpb.SchemaR\
-    esp\"\0J\xc2\x04\n\x06\x12\x04\0\0\x19\x01\n\x08\n\x01\x0c\x12\x03\0\0\
-    \x12\n\x08\n\x01\x02\x12\x03\x01\0\x10\n\t\n\x02\x03\0\x12\x03\x03\0\x1a\
-    \n\t\n\x02\x03\x01\x12\x03\x04\0\x17\n\n\n\x02\x06\0\x12\x04\x06\0\x19\
-    \x01\n\n\n\x03\x06\0\x01\x12\x03\x06\x08\r\n\x0c\n\x04\x06\0\x02\0\x12\
-    \x04\x07\x04\x08\x05\n\x0c\n\x05\x06\0\x02\0\x01\x12\x03\x07\x08\x0c\n\
-    \x0c\n\x05\x06\0\x02\0\x02\x12\x03\x07\x0e\x1d\n\x0c\n\x05\x06\0\x02\0\
-    \x03\x12\x03\x07(;\n\x0c\n\x04\x06\0\x02\x01\x12\x04\t\x04\n\x05\n\x0c\n\
-    \x05\x06\0\x02\x01\x01\x12\x03\t\x08\x16\n\x0c\n\x05\x06\0\x02\x01\x02\
-    \x12\x03\t\x180\n\x0c\n\x05\x06\0\x02\x01\x03\x12\x03\t;N\n\x0c\n\x04\
-    \x06\0\x02\x02\x12\x04\x0b\x04\x0c\x05\n\x0c\n\x05\x06\0\x02\x02\x01\x12\
-    \x03\x0b\x08\r\n\x0c\n\x05\x06\0\x02\x02\x02\x12\x03\x0b\x0f\"\n\x0c\n\
-    \x05\x06\0\x02\x02\x03\x12\x03\x0b-A\n\x0c\n\x04\x06\0\x02\x03\x12\x04\r\
-    \x04\x0e\x05\n\x0c\n\x05\x06\0\x02\x03\x01\x12\x03\r\x08\x0b\n\x0c\n\x05\
-    \x06\0\x02\x03\x02\x12\x03\r\r\x20\n\x0c\n\x05\x06\0\x02\x03\x03\x12\x03\
-    \r+=\n\x0c\n\x04\x06\0\x02\x04\x12\x04\x0f\x04\x10\x05\n\x0c\n\x05\x06\0\
-    \x02\x04\x01\x12\x03\x0f\x08\x0b\n\x0c\n\x05\x06\0\x02\x04\x02\x12\x03\
-    \x0f\r\x20\n\x0c\n\x05\x06\0\x02\x04\x03\x12\x03\x0f+=\n\x0c\n\x04\x06\0\
-    \x02\x05\x12\x04\x11\x04\x12\x05\n\x0c\n\x05\x06\0\x02\x05\x01\x12\x03\
-    \x11\x08\x0e\n\x0c\n\x05\x06\0\x02\x05\x02\x12\x03\x11\x10#\n\x0c\n\x05\
-    \x06\0\x02\x05\x03\x12\x03\x11.C\n\x0c\n\x04\x06\0\x02\x06\x12\x04\x13\
-    \x04\x14\x05\n\x0c\n\x05\x06\0\x02\x06\x01\x12\x03\x13\x08\x0e\n\x0c\n\
-    \x05\x06\0\x02\x06\x02\x12\x03\x13\x10#\n\x0c\n\x05\x06\0\x02\x06\x03\
-    \x12\x03\x13.C\n\x0c\n\x04\x06\0\x02\x07\x12\x04\x15\x04\x16\x05\n\x0c\n\
-    \x05\x06\0\x02\x07\x01\x12\x03\x15\x08\x0e\n\x0c\n\x05\x06\0\x02\x07\x02\
-    \x12\x03\x15\x10#\n\x0c\n\x05\x06\0\x02\x07\x03\x12\x03\x15.C\n\x0c\n\
-    \x04\x06\0\x02\x08\x12\x04\x17\x04\x18\x05\n\x0c\n\x05\x06\0\x02\x08\x01\
-    \x12\x03\x17\x08\x0e\n\x0c\n\x05\x06\0\x02\x08\x02\x12\x03\x17\x10#\n\
-    \x0c\n\x05\x06\0\x02\x08\x03\x12\x03\x17.Cb\x06proto3\
+    cpb.IndexReq\x1a\x15.indexrpcpb.PeersResp\"\0\x12:\n\x07Metrics\x12\x14.\
+    indexrpcpb.IndexReq\x1a\x17.indexrpcpb.MetricsResp\"\0\x122\n\x03Get\x12\
+    \x14.indexrpcpb.IndexReq\x1a\x13.indexrpcpb.GetResp\"\0\x122\n\x03Put\
+    \x12\x14.indexrpcpb.IndexReq\x1a\x13.indexrpcpb.PutResp\"\0\x128\n\x06De\
+    lete\x12\x14.indexrpcpb.IndexReq\x1a\x16.indexrpcpb.DeleteResp\"\0\x128\
+    \n\x06Commit\x12\x14.indexrpcpb.IndexReq\x1a\x16.indexrpcpb.CommitResp\"\
+    \0\x128\n\x06Search\x12\x14.indexrpcpb.IndexReq\x1a\x16.indexrpcpb.Searc\
+    hResp\"\0\x128\n\x06Schema\x12\x14.indexrpcpb.IndexReq\x1a\x16.indexrpcp\
+    b.SchemaResp\"\0J\xfa\x04\n\x06\x12\x04\0\0\x1b\x01\n\x08\n\x01\x0c\x12\
+    \x03\0\0\x12\n\x08\n\x01\x02\x12\x03\x01\0\x10\n\t\n\x02\x03\0\x12\x03\
+    \x03\0\x1a\n\t\n\x02\x03\x01\x12\x03\x04\0\x17\n\n\n\x02\x06\0\x12\x04\
+    \x06\0\x1b\x01\n\n\n\x03\x06\0\x01\x12\x03\x06\x08\r\n\x0c\n\x04\x06\0\
+    \x02\0\x12\x04\x07\x04\x08\x05\n\x0c\n\x05\x06\0\x02\0\x01\x12\x03\x07\
+    \x08\x0c\n\x0c\n\x05\x06\0\x02\0\x02\x12\x03\x07\x0e\x1d\n\x0c\n\x05\x06\
+    \0\x02\0\x03\x12\x03\x07(;\n\x0c\n\x04\x06\0\x02\x01\x12\x04\t\x04\n\x05\
+    \n\x0c\n\x05\x06\0\x02\x01\x01\x12\x03\t\x08\x16\n\x0c\n\x05\x06\0\x02\
+    \x01\x02\x12\x03\t\x180\n\x0c\n\x05\x06\0\x02\x01\x03\x12\x03\t;N\n\x0c\
+    \n\x04\x06\0\x02\x02\x12\x04\x0b\x04\x0c\x05\n\x0c\n\x05\x06\0\x02\x02\
+    \x01\x12\x03\x0b\x08\r\n\x0c\n\x05\x06\0\x02\x02\x02\x12\x03\x0b\x0f\"\n\
+    \x0c\n\x05\x06\0\x02\x02\x03\x12\x03\x0b-A\n\x0c\n\x04\x06\0\x02\x03\x12\
+    \x04\r\x04\x0e\x05\n\x0c\n\x05\x06\0\x02\x03\x01\x12\x03\r\x08\x0f\n\x0c\
+    \n\x05\x06\0\x02\x03\x02\x12\x03\r\x11$\n\x0c\n\x05\x06\0\x02\x03\x03\
+    \x12\x03\r/E\n\x0c\n\x04\x06\0\x02\x04\x12\x04\x0f\x04\x10\x05\n\x0c\n\
+    \x05\x06\0\x02\x04\x01\x12\x03\x0f\x08\x0b\n\x0c\n\x05\x06\0\x02\x04\x02\
+    \x12\x03\x0f\r\x20\n\x0c\n\x05\x06\0\x02\x04\x03\x12\x03\x0f+=\n\x0c\n\
+    \x04\x06\0\x02\x05\x12\x04\x11\x04\x12\x05\n\x0c\n\x05\x06\0\x02\x05\x01\
+    \x12\x03\x11\x08\x0b\n\x0c\n\x05\x06\0\x02\x05\x02\x12\x03\x11\r\x20\n\
+    \x0c\n\x05\x06\0\x02\x05\x03\x12\x03\x11+=\n\x0c\n\x04\x06\0\x02\x06\x12\
+    \x04\x13\x04\x14\x05\n\x0c\n\x05\x06\0\x02\x06\x01\x12\x03\x13\x08\x0e\n\
+    \x0c\n\x05\x06\0\x02\x06\x02\x12\x03\x13\x10#\n\x0c\n\x05\x06\0\x02\x06\
+    \x03\x12\x03\x13.C\n\x0c\n\x04\x06\0\x02\x07\x12\x04\x15\x04\x16\x05\n\
+    \x0c\n\x05\x06\0\x02\x07\x01\x12\x03\x15\x08\x0e\n\x0c\n\x05\x06\0\x02\
+    \x07\x02\x12\x03\x15\x10#\n\x0c\n\x05\x06\0\x02\x07\x03\x12\x03\x15.C\n\
+    \x0c\n\x04\x06\0\x02\x08\x12\x04\x17\x04\x18\x05\n\x0c\n\x05\x06\0\x02\
+    \x08\x01\x12\x03\x17\x08\x0e\n\x0c\n\x05\x06\0\x02\x08\x02\x12\x03\x17\
+    \x10#\n\x0c\n\x05\x06\0\x02\x08\x03\x12\x03\x17.C\n\x0c\n\x04\x06\0\x02\
+    \t\x12\x04\x19\x04\x1a\x05\n\x0c\n\x05\x06\0\x02\t\x01\x12\x03\x19\x08\
+    \x0e\n\x0c\n\x05\x06\0\x02\t\x02\x12\x03\x19\x10#\n\x0c\n\x05\x06\0\x02\
+    \t\x03\x12\x03\x19.Cb\x06proto3\
 ";
 
 static mut file_descriptor_proto_lazy: ::protobuf::lazy::Lazy<::protobuf::descriptor::FileDescriptorProto> = ::protobuf::lazy::Lazy {

diff --git a/src/proto/indexpb_grpc.rs b/src/proto/indexpb_grpc.rs
@@ -39,6 +39,13 @@ const METHOD_INDEX_PEERS: ::grpcio::Method<super::indexrpcpb::IndexReq, super::i
     resp_mar: ::grpcio::Marshaller { ser: ::grpcio::pb_ser, de: ::grpcio::pb_de },
 };
 
+const METHOD_INDEX_METRICS: ::grpcio::Method<super::indexrpcpb::IndexReq, super::indexrpcpb::MetricsResp> = ::grpcio::Method {
+    ty: ::grpcio::MethodType::Unary,
+    name: "/indexpb.Index/Metrics",
+    req_mar: ::grpcio::Marshaller { ser: ::grpcio::pb_ser, de: ::grpcio::pb_de },
+    resp_mar: ::grpcio::Marshaller { ser: ::grpcio::pb_ser, de: ::grpcio::pb_de },
+};
+
 const METHOD_INDEX_GET: ::grpcio::Method<super::indexrpcpb::IndexReq, super::indexrpcpb::GetResp> = ::grpcio::Method {
     ty: ::grpcio::MethodType::Unary,
     name: "/indexpb.Index/Get",
@@ -141,6 +148,22 @@ impl IndexClient {
         self.peers_async_opt(req, ::grpcio::CallOption::default())
     }
 
+    pub fn metrics_opt(&self, req: &super::indexrpcpb::IndexReq, opt: ::grpcio::CallOption) -> ::grpcio::Result<super::indexrpcpb::MetricsResp> {
+        self.client.unary_call(&METHOD_INDEX_METRICS, req, opt)
+    }
+
+    pub fn metrics(&self, req: &super::indexrpcpb::IndexReq) -> ::grpcio::Result<super::indexrpcpb::MetricsResp> {
+        self.metrics_opt(req, ::grpcio::CallOption::default())
+    }
+
+    pub fn metrics_async_opt(&self, req: &super::indexrpcpb::IndexReq, opt: ::grpcio::CallOption) -> ::grpcio::Result<::grpcio::ClientUnaryReceiver<super::indexrpcpb::MetricsResp>> {
+        self.client.unary_call_async(&METHOD_INDEX_METRICS, req, opt)
+    }
+
+    pub fn metrics_async(&self, req: &super::indexrpcpb::IndexReq) -> ::grpcio::Result<::grpcio::ClientUnaryReceiver<super::indexrpcpb::MetricsResp>> {
+        self.metrics_async_opt(req, ::grpcio::CallOption::default())
+    }
+
     pub fn get_opt(&self, req: &super::indexrpcpb::IndexReq, opt: ::grpcio::CallOption) -> ::grpcio::Result<super::indexrpcpb::GetResp> {
         self.client.unary_call(&METHOD_INDEX_GET, req, opt)
     }
@@ -245,6 +268,7 @@ pub trait Index {
     fn raft(&mut self, ctx: ::grpcio::RpcContext, req: super::eraftpb::Message, sink: ::grpcio::UnarySink<super::indexrpcpb::RaftDone>);
     fn raft_conf_change(&mut self, ctx: ::grpcio::RpcContext, req: super::indexrpcpb::ConfChangeReq, sink: ::grpcio::UnarySink<super::indexrpcpb::RaftDone>);
     fn peers(&mut self, ctx: ::grpcio::RpcContext, req: super::indexrpcpb::IndexReq, sink: ::grpcio::UnarySink<super::indexrpcpb::PeersResp>);
+    fn metrics(&mut self, ctx: ::grpcio::RpcContext, req: super::indexrpcpb::IndexReq, sink: ::grpcio::UnarySink<super::indexrpcpb::MetricsResp>);
     fn get(&mut self, ctx: ::grpcio::RpcContext, req: super::indexrpcpb::IndexReq, sink: ::grpcio::UnarySink<super::indexrpcpb::GetResp>);
     fn put(&mut self, ctx: ::grpcio::RpcContext, req: super::indexrpcpb::IndexReq, sink: ::grpcio::UnarySink<super::indexrpcpb::PutResp>);
     fn delete(&mut self, ctx: ::grpcio::RpcContext, req: super::indexrpcpb::IndexReq, sink: ::grpcio::UnarySink<super::indexrpcpb::DeleteResp>);
@@ -268,6 +292,10 @@ pub fn create_index<S: Index + Send + Clone + 'static>(s: S) -> ::grpcio::Servic
         instance.peers(ctx, req, resp)
     });
     let mut instance = s.clone();
+    builder = builder.add_unary_handler(&METHOD_INDEX_METRICS, move |ctx, req, resp| {
+        instance.metrics(ctx, req, resp)
+    });
+    let mut instance = s.clone();
     builder = builder.add_unary_handler(&METHOD_INDEX_GET, move |ctx, req, resp| {
         instance.get(ctx, req, resp)
     });