Frontend: Retrieve and Display Logs of Services

This commit adds following features:

- Add an REST resource to retrieve the log messages of a service of an review
  app with pagination.
- Utilize this resource to display the log messages in the frontend.

api/res/config.toml

@@ -1,4 +1,2 @@
 [containers]
 memory_limit = '1g'
-
-[[companions]]

api/res/openapi.yml

@@ -103,13 +103,7 @@ paths:
     delete:
       summary: Shutdown a review app
       parameters:
-        - in: path
-          name: appName
-          allowEmptyValue: false
-          schema:
-            type: string
-          required: true
-          description: Name of review app to delete
+        - $ref: '#/components/parameters/appName'
       responses:
         '200':
           description: 'List of deleted containers'
@@ -131,6 +125,43 @@ paths:
             application/problem+json:
               schema:
                 $ref: '#/components/schemas/ProblemDetails'
+  /apps/{appName}/logs/{serviceName}/:
+    get:
+      summary: Retrieves the logs from stdout/stderr of the specified container.
+      parameters:
+        - $ref: '#/components/parameters/appName'
+        - in: path
+          name: serviceName
+          description: Name of the service
+          required: true
+          schema:
+            type: string
+        - in: query
+          name: since
+          description: >-
+            Date and time since when the logs have to retrieved. By default the logs from the beginning are crawled.
+          schema:
+            type: string
+            format: date-time
+          example: '2019-07-22T08:42:47-00:00'
+        - in: query
+          name: limit
+          description: The number of log lines to retrieve. If not present, 1000 lines will be retrieved.
+          schema:
+            type: integer
+      responses:
+        '200':
+          description: The available log statements
+          headers:
+            Link:
+              schema:
+                type: string
+              description: The links for pagination
+              example: </apps/master/logs/service-a/?limit=1000&since=2019-07-22T08:42:47-00:00>;rel=next
+          content:
+            text/plain:
+              schema:
+                type: string
   /webhooks/:
     post:
       summary: Cleans up apps when webhook triggers this resource.
@@ -169,6 +200,15 @@ paths:
               schema:
                 $ref: '#/components/schemas/ProblemDetails'
 components:
+  parameters:
+    appName:
+      in: path
+      name: appName
+      allowEmptyValue: false
+      schema:
+        type: string
+      required: true
+      description: Name of the application
   schemas:
     Service:
       type: object

api/src/apps.rs

@@ -26,13 +26,15 @@
 
 use crate::models::request_info::RequestInfo;
 use crate::models::service::{Service, ServiceConfig};
-use crate::models::{AppName, AppNameError};
+use crate::models::{AppName, AppNameError, LogChunk};
 use crate::services::apps_service::AppsService;
+use chrono::DateTime;
 use http_api_problem::HttpApiProblem;
 use multimap::MultiMap;
 use rocket::data::{self, FromDataSimple};
 use rocket::http::Status;
 use rocket::request::{Form, Request};
+use rocket::response::{Responder, Response};
 use rocket::Outcome::{Failure, Success};
 use rocket::{Data, State};
 use rocket_contrib::json::Json;
@@ -99,6 +101,50 @@ pub fn create_app(
     Ok(Json(services))
 }
 
+#[get(
+    "/apps/<app_name>/logs/<service_name>?<since>&<limit>",
+    format = "text/plain"
+)]
+pub fn logs(
+    app_name: Result<AppName, AppNameError>,
+    service_name: String,
+    since: Option<String>,
+    limit: Option<usize>,
+    apps_service: State<AppsService>,
+) -> Result<LogsResponse, HttpApiProblem> {
+    let app_name = app_name?;
+
+    let since = match since {
+        None => None,
+        Some(since) => match DateTime::parse_from_rfc3339(&since) {
+            Ok(since) => Some(since),
+            Err(err) => {
+                return Err(HttpApiProblem::with_title_and_type_from_status(
+                    http_api_problem::StatusCode::BAD_REQUEST,
+                )
+                .set_detail(format!("{}", err)));
+            }
+        },
+    };
+    let limit = limit.unwrap_or(20_000);
+
+    let log_chunk = apps_service.get_logs(&app_name, &service_name, &since, limit)?;
+
+    Ok(LogsResponse {
+        log_chunk,
+        app_name,
+        service_name,
+        limit,
+    })
+}
+
+pub struct LogsResponse {
+    log_chunk: Option<LogChunk>,
+    app_name: AppName,
+    service_name: String,
+    limit: usize,
+}
+
 #[derive(FromForm)]
 pub struct CreateAppOptions {
     #[form(field = "replicateFrom")]
@@ -137,3 +183,31 @@ impl FromDataSimple for ServiceConfigsData {
         Success(ServiceConfigsData { service_configs })
     }
 }
+
+impl Responder<'static> for LogsResponse {
+    fn respond_to(self, _request: &Request) -> Result<Response<'static>, Status> {
+        let log_chunk = match self.log_chunk {
+            None => {
+                return Ok(
+                    HttpApiProblem::from(http_api_problem::StatusCode::NOT_FOUND)
+                        .to_rocket_response(),
+                )
+            }
+            Some(log_chunk) => log_chunk,
+        };
+
+        let from = log_chunk.until().clone() + chrono::Duration::milliseconds(1);
+
+        let next_logs_url = format!(
+            "/api/apps/{}/logs/{}/?limit={}&since={}",
+            self.app_name,
+            self.service_name,
+            self.limit,
+            rocket::http::uri::Uri::percent_encode(&from.to_rfc3339()),
+        );
+        Response::build()
+            .raw_header("Link", format!("<{}>;rel=next", next_logs_url))
+            .sized_body(std::io::Cursor::new(log_chunk.log_lines().clone()))
+            .ok()
+    }
+}

api/src/infrastructure/docker.rs

@@ -27,6 +27,7 @@
 use crate::config::ContainerConfig;
 use crate::infrastructure::Infrastructure;
 use crate::models::service::{ContainerType, Image, Service, ServiceConfig, ServiceError};
+use chrono::{DateTime, FixedOffset};
 use failure::Error;
 use futures::future::join_all;
 use futures::{Future, Stream};
@@ -36,7 +37,7 @@ use shiplift::builder::ContainerOptions;
 use shiplift::errors::Error as ShipLiftError;
 use shiplift::rep::{Container, ContainerCreateInfo, ContainerDetails};
 use shiplift::{
-    ContainerConnectionOptions, ContainerFilter, ContainerListOptions, Docker,
+    ContainerConnectionOptions, ContainerFilter, ContainerListOptions, Docker, LogsOptions,
     NetworkCreateOptions, PullOptions,
 };
 use std::collections::HashMap;
@@ -45,6 +46,7 @@ use std::net::{AddrParseError, IpAddr};
 use std::str::FromStr;
 use std::sync::mpsc;
 use tokio::runtime::Runtime;
+use tokio::util::StreamExt;
 
 static APP_NAME_LABEL: &str = "com.aixigo.preview.servant.app-name";
 static SERVICE_NAME_LABEL: &str = "com.aixigo.preview.servant.service-name";
@@ -586,6 +588,76 @@ impl Infrastructure for DockerInfrastructure {
 
         Ok(configs)
     }
+
+    fn get_logs(
+        &self,
+        app_name: &String,
+        service_name: &String,
+        from: &Option<DateTime<FixedOffset>>,
+        limit: usize,
+    ) -> Result<Option<Vec<(DateTime<FixedOffset>, String)>>, failure::Error> {
+        match self.get_app_container(app_name, service_name)? {
+            None => Ok(None),
+            Some(container) => {
+                let docker = Docker::new();
+                let mut runtime = Runtime::new()?;
+
+                trace!(
+                    "Acquiring logs of container {} since {:?}",
+                    container.id,
+                    from
+                );
+
+                let log_options = match from {
+                    Some(from) => LogsOptions::builder()
+                        .since(from.timestamp())
+                        .stdout(true)
+                        .stderr(true)
+                        .timestamps(true)
+                        .build(),
+                    None => LogsOptions::builder()
+                        .stdout(true)
+                        .stderr(true)
+                        .timestamps(true)
+                        .build(),
+                };
+
+                let cloned_from = from.clone();
+                let logs = runtime.block_on(
+                    docker
+                        .containers()
+                        .get(&container.id)
+                        .logs(&log_options)
+                        .enumerate()
+                        // Unfortunately, docker API does not support head (cf. https://github.com/moby/moby/issues/13096)
+                        // Until then we have to skip these log messages which is super slow…
+                        .filter(move |(index, _)| index < &limit)
+                        .map(|(_, chunk)| {
+                            let line = chunk.as_string_lossy();
+
+                            let mut iter = line.splitn(2, ' ').into_iter();
+                            let timestamp = iter.next()
+                                .expect("This should never happen: docker should return timestamps, separated by space");
+
+                            let datetime = DateTime::parse_from_rfc3339(&timestamp).expect("Expecting a valid timestamp");
+
+                            let log_line : String = iter
+                                .collect::<Vec<&str>>()
+                                .join(" ");
+                            (datetime, log_line)
+                        })
+                        .filter(move |(timestamp, _)| {
+                            // Due to the fact that docker's REST API supports only unix time (cf. since),
+                            // it is necessary to filter the timestamps as well.
+                            cloned_from.map(|from| timestamp >= &from).unwrap_or_else(|| true)
+                        })
+                        .collect()
+                )?;
+
+                Ok(Some(logs))
+            }
+        }
+    }
 }
 
 impl TryFrom<&ContainerDetails> for Service {

api/src/infrastructure/dummy_infrastructure.rs

@@ -28,6 +28,7 @@ use crate::config::ContainerConfig;
 use crate::infrastructure::Infrastructure;
 use crate::models::service::Service;
 use crate::models::service::ServiceConfig;
+use chrono::{DateTime, FixedOffset};
 use multimap::MultiMap;
 use std::collections::HashSet;
 use std::sync::Mutex;
@@ -106,4 +107,27 @@ impl Infrastructure for DummyInfrastructure {
             Some(configs) => Ok(configs.clone()),
         }
     }
+
+    fn get_logs(
+        &self,
+        app_name: &String,
+        service_name: &String,
+        _from: &Option<DateTime<FixedOffset>>,
+        _limit: usize,
+    ) -> Result<Option<Vec<(DateTime<FixedOffset>, String)>>, failure::Error> {
+        Ok(Some(vec![
+            (
+                DateTime::parse_from_rfc3339("2019-07-18T07:25:00.000000000Z").unwrap(),
+                format!("Log msg 1 of {} of app {}\n", service_name, app_name),
+            ),
+            (
+                DateTime::parse_from_rfc3339("2019-07-18T07:30:00.000000000Z").unwrap(),
+                format!("Log msg 2 of {} of app {}\n", service_name, app_name),
+            ),
+            (
+                DateTime::parse_from_rfc3339("2019-07-18T07:35:00.000000000Z").unwrap(),
+                format!("Log msg 3 of {} of app {}\n", service_name, app_name),
+            ),
+        ]))
+    }
 }

api/src/infrastructure/infrastructure.rs

@@ -26,6 +26,7 @@
 
 use crate::config::ContainerConfig;
 use crate::models::service::{Service, ServiceConfig};
+use chrono::{DateTime, FixedOffset};
 use failure::Error;
 use multimap::MultiMap;
 
@@ -53,4 +54,13 @@ pub trait Infrastructure: Send + Sync {
     /// Returns the configuration of all services running for the given application name.
     /// It is required that the configurations of the companions are excluded.
     fn get_configs_of_app(&self, app_name: &String) -> Result<Vec<ServiceConfig>, Error>;
+
+    /// Returns the log lines with a the corresponding timestamps in it.
+    fn get_logs(
+        &self,
+        app_name: &String,
+        service_name: &String,
+        from: &Option<DateTime<FixedOffset>>,
+        limit: usize,
+    ) -> Result<Option<Vec<(DateTime<FixedOffset>, String)>>, Error>;
 }

api/src/main.rs

@@ -24,7 +24,7 @@
  * =========================LICENSE_END==================================
  */
 
-#![feature(custom_attribute, proc_macro_hygiene, decl_macro)]
+#![feature(custom_attribute, proc_macro_hygiene, decl_macro, option_flattening)]
 
 #[macro_use]
 extern crate cached;
@@ -70,7 +70,7 @@ fn index() -> CacheResponse<Option<NamedFile>> {
     }
 }
 
-#[get("/<path..>")]
+#[get("/<path..>", rank = 100)]
 fn files(path: PathBuf) -> CacheResponse<Option<NamedFile>> {
     CacheResponse::Private {
         responder: NamedFile::open(Path::new("frontend/").join(path)).ok(),
@@ -135,6 +135,7 @@ fn main() {
         .mount("/", routes![index])
         .mount("/openapi.yaml", routes![openapi])
         .mount("/", routes![files])
+        .mount("/api", routes![apps::logs])
         .mount("/api", routes![apps::apps])
         .mount("/api", routes![apps::create_app])
         .mount("/api", routes![apps::delete_app])

api/src/models/app_name.rs

@@ -43,6 +43,12 @@ impl Deref for AppName {
     }
 }
 
+impl std::fmt::Display for AppName {
+    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
+        write!(f, "{}", self.0)
+    }
+}
+
 impl FromStr for AppName {
     type Err = AppNameError;