Change behaviour to use default credentials (#135)

nitisht · nitisht · commit a53c929b9779 · 2022-05-29T22:37:10.000+05:30
Also fix a bug in stream metadata management
diff --git a/server/src/banner.rs b/server/src/banner.rs
@@ -37,8 +37,12 @@ fn status_info(scheme: &str, opt: &option::Opt) {
     let url = format!("{}://{}", scheme, opt.address).underlined();
     eprintln!(
         "
+    {}
+    {}
     {}",
-        format!("Parseable server started at {}", url).bold(),
+        format!("Parseable server started at: {}", url).bold(),
+        format!("Username: {}", opt.username).bold(),
+        format!("Password: {}", opt.password).bold(),
     )
 }
 
@@ -72,13 +76,55 @@ fn storage_info(opt: &option::Opt) {
 }
 
 fn warning(opt: &option::Opt) {
+    if opt.s3_endpoint_url == option::DEFAULT_S3_URL
+        && opt.username == option::DEFAULT_USERNAME
+        && opt.password == option::DEFAULT_PASSWORD
+    {
+        warning_line();
+        cred_warning(opt);
+        s3_warning(opt);
+    } else if opt.s3_endpoint_url == option::DEFAULT_S3_URL {
+        warning_line();
+        s3_warning(opt);
+    } else if opt.username == option::DEFAULT_USERNAME && opt.password == option::DEFAULT_PASSWORD {
+        warning_line();
+        cred_warning(opt);
+    }
+}
+
+fn warning_line() {
+    eprintln!(
+        "
+    {}",
+        "Warning:".to_string().red().bold(),
+    );
+}
+
+fn cred_warning(opt: &option::Opt) {
+    if opt.username == option::DEFAULT_USERNAME && opt.password == option::DEFAULT_PASSWORD {
+        eprintln!(
+            "
+        {}
+        {}",
+            "Parseable server is using default credentials."
+                .to_string()
+                .red(),
+            format!(
+                "Setup your credentials with {} and {} before storing production logs.",
+                option::USERNAME_ENV,
+                option::PASSOWRD_ENV
+            )
+            .red()
+        )
+    }
+}
+
+fn s3_warning(opt: &option::Opt) {
     if opt.s3_endpoint_url == option::DEFAULT_S3_URL {
         eprintln!(
             "
-    {}
         {}
         {}",
-            "Warning:".to_string().red().bold(),
             "Parseable server is using default object storage backend with public access."
                 .to_string()
                 .red(),
diff --git a/server/src/error.rs b/server/src/error.rs
@@ -65,10 +65,6 @@ pub enum Error {
     StreamLock,
     #[error("Metadata not found for log stream: {0}")]
     StreamMetaNotFound(String),
-    #[error("Schema not found for log stream: {0}")]
-    SchemaNotFound(String),
-    #[error("Alert config not found for log stream: {0}")]
-    AlertConfigNotFound(String),
     #[error("Invalid alert config: {0}")]
     InvalidAlert(String),
 }
diff --git a/server/src/handler.rs b/server/src/handler.rs
@@ -85,6 +85,9 @@ pub async fn cache_query(_req: HttpRequest, query: web::Json<query::Query>) -> H
     }
 }
 
+//
+// **** Stream related handlers ****
+//
 pub async fn delete_stream(req: HttpRequest) -> HttpResponse {
     let stream_name: String = req.match_info().get("logstream").unwrap().parse().unwrap();
     if let Err(e) = validator::stream_name(&stream_name) {
@@ -137,6 +140,10 @@ pub async fn list_streams(_: HttpRequest) -> impl Responder {
     response::list_response(storage::list_streams().unwrap())
 }
 
+//
+// **** Steam metadata related handlers ****
+//
+// Get log stream schema
 pub async fn get_schema(req: HttpRequest) -> HttpResponse {
     let stream_name: String = req.match_info().get("logstream").unwrap().parse().unwrap();
     // fail to proceed if there is an error in log stream name validation
@@ -246,22 +253,24 @@ pub async fn put_stream(req: HttpRequest) -> HttpResponse {
             .to_http();
         }
 
-        return response::ServerResponse {
-            msg: format!("Created log stream {}", stream_name),
-            code: StatusCode::OK,
+        if let Err(e) = metadata::STREAM_INFO.add_stream(
+            stream_name.to_string(),
+            "".to_string(),
+            "".to_string(),
+        ) {
+            return response::ServerResponse {
+                msg: format!(
+                    "Failed to add log stream {} to metadata due to err: {}",
+                    stream_name, e
+                ),
+                code: StatusCode::INTERNAL_SERVER_ERROR,
+            }
+            .to_http();
         }
-        .to_http();
-    }
 
-    if let Err(e) =
-        metadata::STREAM_INFO.add_stream(stream_name.to_string(), "".to_string(), "".to_string())
-    {
         return response::ServerResponse {
-            msg: format!(
-                "Failed to add log stream {} to metadata due to err: {}",
-                stream_name, e
-            ),
-            code: StatusCode::INTERNAL_SERVER_ERROR,
+            msg: format!("Created log stream {}", stream_name),
+            code: StatusCode::OK,
         }
         .to_http();
     }
@@ -324,7 +333,7 @@ pub async fn put_alert(req: HttpRequest, body: web::Json<serde_json::Value>) ->
 }
 
 pub async fn post_event(req: HttpRequest, body: web::Json<serde_json::Value>) -> HttpResponse {
-    let stream_name: String = req.match_info().get("log stream").unwrap().parse().unwrap();
+    let stream_name: String = req.match_info().get("logstream").unwrap().parse().unwrap();
 
     let labels = collect_labels(&req);
 
diff --git a/server/src/main.rs b/server/src/main.rs
@@ -79,8 +79,8 @@ async fn validator(
     credentials: BasicAuth,
 ) -> Result<ServiceRequest, actix_web::Error> {
     let opt = option::get_opts();
-    if credentials.user_id().trim() == opt.username.unwrap()
-        && credentials.password().unwrap().trim() == opt.password.unwrap()
+    if credentials.user_id().trim() == opt.username
+        && credentials.password().unwrap().trim() == opt.password
     {
         return Ok(req);
     }
@@ -100,27 +100,15 @@ async fn run_http(opt: option::Opt) -> anyhow::Result<()> {
         (_, _) => None,
     };
 
-    if opt.username.is_some() && opt.password.is_some() {
-        let http_server =
-            HttpServer::new(move || create_app!().wrap(HttpAuthentication::basic(validator)));
-        if let Some(builder) = ssl_acceptor {
-            http_server
-                .bind_openssl(opt_clone.address, builder)?
-                .run()
-                .await?;
-        } else {
-            http_server.bind(opt_clone.address)?.run().await?;
-        }
+    let http_server =
+        HttpServer::new(move || create_app!().wrap(HttpAuthentication::basic(validator)));
+    if let Some(builder) = ssl_acceptor {
+        http_server
+            .bind_openssl(opt_clone.address, builder)?
+            .run()
+            .await?;
     } else {
-        let http_server = HttpServer::new(move || create_app!());
-        if let Some(builder) = ssl_acceptor {
-            http_server
-                .bind_openssl(opt_clone.address, builder)?
-                .run()
-                .await?;
-        } else {
-            http_server.bind(opt_clone.address)?.run().await?;
-        }
+        http_server.bind(opt_clone.address)?.run().await?;
     }
 
     Ok(())
diff --git a/server/src/metadata.rs b/server/src/metadata.rs
@@ -49,13 +49,7 @@ impl STREAM_INFO {
         drop(&map);
 
         match meta {
-            Some(meta) => {
-                if meta.schema.clone() != "" {
-                    Ok(meta.schema.clone())
-                } else {
-                    Err(Error::SchemaNotFound(stream_name))
-                }
-            }
+            Some(meta) => Ok(meta.schema.clone()),
             None => Err(Error::StreamMetaNotFound(stream_name)),
         }
     }
@@ -66,13 +60,7 @@ impl STREAM_INFO {
         drop(&map);
 
         match meta {
-            Some(meta) => {
-                if meta.alert.clone() != "" {
-                    Ok(meta.alert.clone())
-                } else {
-                    Err(Error::AlertConfigNotFound(stream_name))
-                }
-            }
+            Some(meta) => Ok(meta.alert.clone()),
             None => Err(Error::StreamMetaNotFound(stream_name)),
         }
     }
@@ -139,7 +127,11 @@ impl STREAM_INFO {
                                 alert_config = format!("{:?}", x);
                             }
                         }
-                        Err(e) => Err(Error::S3(e))?,
+                        Err(_) => {
+                            // Ignore S3 errors here, because
+                            // we are just trying to load the
+                            // stream metadata based on whatever is available
+                        }
                     };
                     match storage::stream_exists(&stream_name) {
                         Ok(x) => {
@@ -149,7 +141,11 @@ impl STREAM_INFO {
                                 schema = format!("{:?}", x);
                             }
                         }
-                        Err(e) => Err(Error::S3(e))?,
+                        Err(_) => {
+                            // Ignore S3 errors here, because
+                            // we are just trying to load the
+                            // stream metadata based on whatever is available
+                        }
                     };
                     let metadata = LogStreamMetadata {
                         schema: schema,
diff --git a/server/src/option.rs b/server/src/option.rs
@@ -21,6 +21,10 @@ use structopt::StructOpt;
 
 pub const DEFAULT_S3_URL: &str = "http://127.0.0.1:9000";
 pub const S3_URL_ENV_VAR: &str = "P_S3_URL";
+pub const USERNAME_ENV: &str = "P_USERNAME";
+pub const PASSOWRD_ENV: &str = "P_PASSWORD";
+pub const DEFAULT_USERNAME: &str = "parseable";
+pub const DEFAULT_PASSWORD: &str = "parseable";
 
 #[derive(Debug, Clone, StructOpt)]
 #[structopt(
@@ -72,12 +76,12 @@ pub struct Opt {
     pub s3_bucket_name: String,
 
     /// Optional username to enable basic auth on the server
-    #[structopt(long, env = "P_USERNAME")]
-    pub username: Option<String>,
+    #[structopt(long, env = USERNAME_ENV, default_value = DEFAULT_USERNAME)]
+    pub username: String,
 
     /// Optional password to enable basic auth on the server
-    #[structopt(long, env = "P_PASSWORD")]
-    pub password: Option<String>,
+    #[structopt(long, env = PASSOWRD_ENV, default_value = DEFAULT_PASSWORD)]
+    pub password: String,
 }
 
 pub fn get_opts() -> Opt {
