Better logging

Cargo.toml

@@ -19,6 +19,8 @@ futures = "0.3.8"
 tokio = { version = "0.2", features = ["macros"] }
 tonic = "0.3.1"
 lazy-init = "0.4.0"
+log = "0.4.11"
+env_logger = "0.8.2"
 
 [dev-dependencies]
 clap = "2.33.3"

cas/BUILD

@@ -10,7 +10,9 @@ rust_binary(
         "//cas/grpc_service:execution_server",
         "//cas/grpc_service:bytestream_server",
         "//cas/store",
+        "//util:common",
         "//third_party:tokio",
         "//third_party:tonic",
+        "//third_party:env_logger",
     ],
 )

cas/cas_main.rs

@@ -11,6 +11,10 @@ use store::{StoreConfig, StoreType};
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    env_logger::Builder::from_env(env_logger::Env::default().default_filter_or("info"))
+        .format_timestamp_millis()
+        .init();
+
     let addr = "0.0.0.0:50051".parse()?;
 
     let ac_store = store::create_store(&StoreConfig {

cas/grpc_service/ac_server.rs

@@ -3,6 +3,7 @@
 use std::convert::TryInto;
 use std::io::Cursor;
 use std::sync::Arc;
+use std::time::Instant;
 
 use prost::Message;
 use tonic::{Request, Response, Status};
@@ -12,7 +13,7 @@ use proto::build::bazel::remote::execution::v2::{
     ActionResult, GetActionResultRequest, UpdateActionResultRequest,
 };
 
-use common::DigestInfo;
+use common::{log, DigestInfo};
 use error::{make_err, Code, Error, ResultExt};
 use store::Store;
 
@@ -104,25 +105,33 @@ impl ActionCache for AcServer {
         &self,
         grpc_request: Request<GetActionResultRequest>,
     ) -> Result<Response<ActionResult>, Status> {
-        println!(
+        let now = Instant::now();
+        log::info!(
             "\x1b[0;31mget_action_result Req\x1b[0m: {:?}",
             grpc_request.get_ref()
         );
         let resp = self.inner_get_action_result(grpc_request).await;
-        println!("\x1b[0;31mget_action_result Resp\x1b[0m: {:?}", resp);
+        let d = now.elapsed().as_secs_f32();
+        log::info!("\x1b[0;31mget_action_result Resp\x1b[0m: {} {:?}", d, resp);
         return resp.map_err(|e| e.into());
     }
 
     async fn update_action_result(
         &self,
         grpc_request: Request<UpdateActionResultRequest>,
     ) -> Result<Response<ActionResult>, Status> {
-        println!(
+        let now = Instant::now();
+        log::info!(
             "\x1b[0;31mupdate_action_result Req\x1b[0m: {:?}",
             grpc_request.get_ref()
         );
         let resp = self.inner_update_action_result(grpc_request).await;
-        println!("\x1b[0;31mupdate_action_result Resp\x1b[0m: {:?}", resp);
+        let d = now.elapsed().as_secs_f32();
+        log::info!(
+            "\x1b[0;31mupdate_action_result Resp\x1b[0m: {} {:?}",
+            d,
+            resp
+        );
         return resp.map_err(|e| e.into());
     }
 }

cas/grpc_service/bytestream_server.rs

@@ -2,6 +2,7 @@
 
 use std::pin::Pin;
 use std::sync::Arc;
+use std::time::Instant;
 
 use async_fixed_buffer::AsyncFixedBuf;
 use futures::Stream;
@@ -14,7 +15,7 @@ use proto::google::bytestream::{
     WriteResponse,
 };
 
-use common::DigestInfo;
+use common::{log, DigestInfo};
 use error::{error_if, Error, ResultExt};
 use store::Store;
 
@@ -195,29 +196,31 @@ impl ByteStream for ByteStreamServer {
         &self,
         _grpc_request: Request<ReadRequest>,
     ) -> Result<Response<Self::ReadStream>, Status> {
-        println!("\x1b[0;31mread\x1b[0m {:?}", _grpc_request.get_ref());
+        log::info!("\x1b[0;31mread\x1b[0m {:?}", _grpc_request.get_ref());
         Err(Status::unimplemented(""))
     }
 
     async fn write(
         &self,
         grpc_request: Request<Streaming<WriteRequest>>,
     ) -> Result<Response<WriteResponse>, Status> {
-        println!("\x1b[0;31mWrite Req\x1b[0m: {:?}", grpc_request.get_ref());
+        log::info!("\x1b[0;31mWrite Req\x1b[0m: {:?}", grpc_request.get_ref());
+        let now = Instant::now();
         let resp = self
             .inner_write(grpc_request)
             .await
             .err_tip(|| format!("Failed on write() command"))
             .map_err(|e| e.into());
-        println!("\x1b[0;31mWrite Resp\x1b[0m: {:?}", resp);
+        let d = now.elapsed().as_secs_f32();
+        log::info!("\x1b[0;31mWrite Resp\x1b[0m: {} {:?}", d, resp);
         resp
     }
 
     async fn query_write_status(
         &self,
         _grpc_request: Request<QueryWriteStatusRequest>,
     ) -> Result<Response<QueryWriteStatusResponse>, Status> {
-        println!("query_write_status {:?}", _grpc_request.get_ref());
+        log::info!("query_write_status {:?}", _grpc_request.get_ref());
         Err(Status::unimplemented(""))
     }
 }

cas/grpc_service/cas_server.rs

@@ -5,6 +5,7 @@ use std::convert::TryInto;
 use std::io::Cursor;
 use std::pin::Pin;
 use std::sync::Arc;
+use std::time::Instant;
 
 use futures::{stream::Stream, FutureExt, StreamExt};
 use tonic::{Request, Response, Status};
@@ -19,7 +20,7 @@ use proto::build::bazel::remote::execution::v2::{
 };
 use proto::google::rpc::Status as GrpcStatus;
 
-use common::DigestInfo;
+use common::{log, DigestInfo};
 use error::{error_if, Error, ResultExt};
 use store::Store;
 
@@ -158,30 +159,57 @@ impl ContentAddressableStorage for CasServer {
         &self,
         grpc_request: Request<FindMissingBlobsRequest>,
     ) -> Result<Response<FindMissingBlobsResponse>, Status> {
-        self.inner_find_missing_blobs(grpc_request)
+        log::info!(
+            "\x1b[0;31mfind_missing_blobs Req\x1b[0m: {:?}",
+            grpc_request.get_ref()
+        );
+        let now = Instant::now();
+        let resp = self
+            .inner_find_missing_blobs(grpc_request)
             .await
             .err_tip(|| format!("Failed on find_missing_blobs() command"))
-            .map_err(|e| e.into())
+            .map_err(|e| e.into());
+        let d = now.elapsed().as_secs_f32();
+        log::info!("\x1b[0;31mfind_missing_blobs Resp\x1b[0m: {} {:?}", d, resp);
+        resp
     }
 
     async fn batch_update_blobs(
         &self,
         grpc_request: Request<BatchUpdateBlobsRequest>,
     ) -> Result<Response<BatchUpdateBlobsResponse>, Status> {
-        self.inner_batch_update_blobs(grpc_request)
+        log::info!(
+            "\x1b[0;31mbatch_update_blobs Req\x1b[0m: {:?}",
+            grpc_request.get_ref()
+        );
+        let now = Instant::now();
+        let resp = self
+            .inner_batch_update_blobs(grpc_request)
             .await
             .err_tip(|| format!("Failed on batch_update_blobs() command"))
-            .map_err(|e| e.into())
+            .map_err(|e| e.into());
+        let d = now.elapsed().as_secs_f32();
+        log::info!("\x1b[0;31mbatch_update_blobs Resp\x1b[0m: {} {:?}", d, resp);
+        resp
     }
 
     async fn batch_read_blobs(
         &self,
         grpc_request: Request<BatchReadBlobsRequest>,
     ) -> Result<Response<BatchReadBlobsResponse>, Status> {
-        self.inner_batch_read_blobs(grpc_request)
+        log::info!(
+            "\x1b[0;31mbatch_read_blobs Req\x1b[0m: {:?}",
+            grpc_request.get_ref()
+        );
+        let now = Instant::now();
+        let resp = self
+            .inner_batch_read_blobs(grpc_request)
             .await
             .err_tip(|| format!("Failed on batch_read_blobs() command"))
-            .map_err(|e| e.into())
+            .map_err(|e| e.into());
+        let d = now.elapsed().as_secs_f32();
+        log::info!("\x1b[0;31mbatch_read_blobs Resp\x1b[0m: {} {:?}", d, resp);
+        resp
     }
 
     type GetTreeStream =
@@ -192,7 +220,7 @@ impl ContentAddressableStorage for CasServer {
     ) -> Result<Response<Self::GetTreeStream>, Status> {
         use stdext::function_name;
         let output = format!("{} not yet implemented", function_name!());
-        println!("{}", output);
+        log::info!("\x1b[0;31mget_tree\x1b[0m: {:?}", output);
         Err(Status::unimplemented(output))
     }
 }

third_party/BUILD.bazel

@@ -48,6 +48,15 @@ alias(
     ],
 )
 
+alias(
+    name = "env_logger",
+    actual = "@raze__env_logger__0_8_2//:env_logger",
+    tags = [
+        "cargo-raze",
+        "manual",
+    ],
+)
+
 alias(
     name = "fixed_buffer",
     actual = "@raze__fixed_buffer__0_2_2//:fixed_buffer",
@@ -84,6 +93,15 @@ alias(
     ],
 )
 
+alias(
+    name = "log",
+    actual = "@raze__log__0_4_11//:log",
+    tags = [
+        "cargo-raze",
+        "manual",
+    ],
+)
+
 alias(
     name = "pretty_assertions",
     actual = "@raze__pretty_assertions__0_6_1//:pretty_assertions",

third_party/crates.bzl

@@ -481,6 +481,16 @@ def raze_fetch_remote_crates():
         build_file = Label("//third_party/remote:BUILD.env_logger-0.6.2.bazel"),
     )
 
+    maybe(
+        http_archive,
+        name = "raze__env_logger__0_8_2",
+        url = "https://crates.io/api/v1/crates/env_logger/0.8.2/download",
+        type = "tar.gz",
+        sha256 = "f26ecb66b4bdca6c1409b40fb255eefc2bd4f6d135dab3c3124f80ffa2a9661e",
+        strip_prefix = "env_logger-0.8.2",
+        build_file = Label("//third_party/remote:BUILD.env_logger-0.8.2.bazel"),
+    )
+
     maybe(
         http_archive,
         name = "raze__event_listener__2_5_1",
@@ -781,6 +791,16 @@ def raze_fetch_remote_crates():
         build_file = Label("//third_party/remote:BUILD.humantime-1.3.0.bazel"),
     )
 
+    maybe(
+        http_archive,
+        name = "raze__humantime__2_0_1",
+        url = "https://crates.io/api/v1/crates/humantime/2.0.1/download",
+        type = "tar.gz",
+        sha256 = "3c1ad908cc71012b7bea4d0c53ba96a8cba9962f048fa68d143376143d863b7a",
+        strip_prefix = "humantime-2.0.1",
+        build_file = Label("//third_party/remote:BUILD.humantime-2.0.1.bazel"),
+    )
+
     maybe(
         http_archive,
         name = "raze__hyper__0_13_9",