From 7d93ef178bf3479dc0b83b6c93af077c2d0c9a88 Mon Sep 17 00:00:00 2001 From: David Schramm Date: Tue, 19 Mar 2024 17:23:30 +0100 Subject: [PATCH] Revert "add logo" This reverts commit c05e6f0e7640930bfe6b6072fa97078074416478. --- .github/workflows/rust.yml | 20 ++ .gitignore | 14 + Cargo.toml | 32 +++ Dockerfile | 24 ++ LICENSE.md | 21 ++ README.md | 24 ++ build.rs | 6 + build_plugin.sh | 13 + conf.ini | 16 ++ images/LogSQlite Main Logo.svg | 34 --- logsqlite.service | 17 ++ plugin/config.json | 27 ++ src/cleaner.rs | 159 +++++++++++ src/config.rs | 231 ++++++++++++++++ src/docker.rs | 198 ++++++++++++++ src/entry.proto | 17 ++ src/logger.rs | 465 +++++++++++++++++++++++++++++++++ src/main.rs | 122 +++++++++ src/statehandler.rs | 220 ++++++++++++++++ 19 files changed, 1626 insertions(+), 34 deletions(-) create mode 100644 .github/workflows/rust.yml create mode 100644 .gitignore create mode 100644 Cargo.toml create mode 100644 Dockerfile create mode 100644 LICENSE.md create mode 100644 README.md create mode 100644 build.rs create mode 100755 build_plugin.sh create mode 100644 conf.ini delete mode 100644 images/LogSQlite Main Logo.svg create mode 100644 logsqlite.service create mode 100644 plugin/config.json create mode 100644 src/cleaner.rs create mode 100644 src/config.rs create mode 100644 src/docker.rs create mode 100644 src/entry.proto create mode 100644 src/logger.rs create mode 100644 src/main.rs create mode 100644 src/statehandler.rs diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml new file mode 100644 index 0000000..af0cdb7 --- /dev/null +++ b/.github/workflows/rust.yml @@ -0,0 +1,20 @@ +name: Rust + +on: + push: + branches: [ "main" ] + pull_request: + branches: [ "main" ] + +env: + CARGO_TERM_COLOR: always + +jobs: + build: + + runs-on: ubuntu-latest + + steps: + - uses: actions/checkout@v3 + - name: Build + run: cargo build --verbose diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..6985cf1 --- /dev/null +++ b/.gitignore @@ -0,0 +1,14 @@ +# Generated by Cargo +# will have compiled files and executables +debug/ +target/ + +# Remove Cargo.lock from gitignore if creating an executable, leave it for libraries +# More information here https://doc.rust-lang.org/cargo/guide/cargo-toml-vs-cargo-lock.html +Cargo.lock + +# These are backup files generated by rustfmt +**/*.rs.bk + +# MSVC Windows builds of rustc generate these, which store debugging information +*.pdb diff --git a/Cargo.toml b/Cargo.toml new file mode 100644 index 0000000..2f0d783 --- /dev/null +++ b/Cargo.toml @@ -0,0 +1,32 @@ +[package] +name = "logsqlite" +version = "0.1.0" +edition = "2021" + +# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html + +[dependencies] +axum = { version= "0.6.20", features= ["macros"]} +futures = "0.3.28" +serde_json = "1.0.105" +serde = {version = "1.0.183", features = ["derive"]} +hyperlocal = "0.8" +hyper = "0.14.27" +tokio = {version = "1.29.1", features = ["fs", "io-util", "rt-multi-thread", "macros", "time", "sync"] } +chrono = "0.4.26" +bytes = "1" +futures-util = { version = "0.3", default-features = false, features = ["alloc"] } +prost = { version= "0.12" } +http = "0.2" +log = "0.4" +simple_logger = "4.2" +configparser = "3.0" +sqlx = { version = "0.7.3", features = ["runtime-tokio", "macros", "sqlite"]} +bincode = "1.3.3" + +[build-dependencies] +prost-build = "0.5" + +[profile.release] +lto=true +strip=true diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..45c1082 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,24 @@ +# syntax=docker/dockerfile:1.7-labs + +FROM rust:bookworm as prefetch +WORKDIR /src/logsqlite +RUN apt update && apt install -y build-essential libsqlite3-0 libsqlite3-dev protobuf-compiler && apt clean +COPY Cargo.toml /src/logsqlite/ +RUN mkdir /src/logsqlite/src +RUN echo 'fn main() {}' >/src/logsqlite/src/main.rs +RUN cargo b --release + +FROM prefetch as build +WORKDIR /src/logsqlite +COPY Cargo.toml build.rs /src/logsqlite/ +COPY src /src/logsqlite/src +RUN cargo b --release && cp target/*/logsqlite . + +FROM debian:bookworm as logsqlite_rootfs +RUN apt update && apt install -y libsqlite3-0 && apt clean +RUN mkdir -p /var/spool/logsqlite /etc/logsqlite /run/docker/plugins +COPY --from=build /src/logsqlite/logsqlite /bin/logsqlite +COPY conf.ini /etc/logsqlite/ +VOLUME /var/spool/logsqlite +WORKDIR /bin/ +ENTRYPOINT [ "/bin/logsqlite", "/etc/logsqlite/conf.ini" ] diff --git a/LICENSE.md b/LICENSE.md new file mode 100644 index 0000000..79839f7 --- /dev/null +++ b/LICENSE.md @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2023 David Schramm + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/README.md b/README.md new file mode 100644 index 0000000..536ce22 --- /dev/null +++ b/README.md @@ -0,0 +1,24 @@ +# logsqlite +Moby/Docker logging driver plugin which uses sqlite3 databases. + +Allows **faster** querying of logs (f.e using `docker logs --since/--until`) than the default JSON File logging driver + +# Building +* `cargo b --release` + +# Installation (systemd) +1. `cp logsqlite /usr/local/bin/` +2. `mkdir /etc/logsqlite && cp conf.ini /etc/logsqlite/` +3. `cp logsqlite.service /etc/systemd/system/ && systemctl daemon-reload` +4. `mkdir /var/spool/logsqlite/` +5. `systemctl enable logsqlite && systemctl start logsqlite` + +# Configuration +See `conf.ini` + +# Using the driver +- as the default logging driver: + - add `"log-driver": "logsqlite"` in the `daemon.json` +- per container + - `docker run --log-driver logsqlite` + - or in the docker-compose: https://docs.docker.com/compose/compose-file/compose-file-v3/#logging diff --git a/build.rs b/build.rs new file mode 100644 index 0000000..e9f551a --- /dev/null +++ b/build.rs @@ -0,0 +1,6 @@ +extern crate prost_build; + +fn main() { + println!("cargo:rerun-if-changed=src/entry.proto"); + prost_build::compile_protos(&["src/entry.proto"], &["src/"]).unwrap(); +} diff --git a/build_plugin.sh b/build_plugin.sh new file mode 100755 index 0000000..8a59c81 --- /dev/null +++ b/build_plugin.sh @@ -0,0 +1,13 @@ +#!/bin/bash +set -e +docker build -t logsqlite_rootfs . + +id=$(docker create logsqlite_rootfs true) + +mkdir -p plugin/rootfs + +docker export "$id" | sudo tar -x -C plugin/rootfs + +docker rm -vf "$id" + +docker plugin create logsqlite ./plugin/ diff --git a/conf.ini b/conf.ini new file mode 100644 index 0000000..190196d --- /dev/null +++ b/conf.ini @@ -0,0 +1,16 @@ +[general] +# the directory where docker plugins sockets are placed in +# (usually you can leave this to the default path) +# don't forget to also adapt your systemd-service if you change this one +plugins_dir=/run/docker/plugins/ + +# the directory where logsqlite puts the databases into +databases_dir=/var/spool/logsqlite/ + +# this database is used for keeping the state of to-be-logged containers +# across restarts of logsqlite +state_database=/etc/logsqlite/state.db + +[cleanup] +# in which interval the log cleaner runs +interval=600 # in seconds diff --git a/images/LogSQlite Main Logo.svg b/images/LogSQlite Main Logo.svg deleted file mode 100644 index a573901..0000000 --- a/images/LogSQlite Main Logo.svg +++ /dev/null @@ -1,34 +0,0 @@ - - - - - - - - - - diff --git a/logsqlite.service b/logsqlite.service new file mode 100644 index 0000000..722ceb0 --- /dev/null +++ b/logsqlite.service @@ -0,0 +1,17 @@ +[Unit] +Description=Moby/Docker logdriver for logging into sqlite databases + +[Service] +Type=simple +User=root +Group=root +Restart=always +#RemainAfterExit=no +#Environment="DEBUG=true" +WorkingDirectory=/var/spool/logsqlite/ +ExecStartPre=-rm /var/run/docker/plugins/logsqlite.sock +ExecStart=/usr/local/bin/logsqlite /etc/logsqlite/conf.ini +#StandardOutput=null + +[Install] +WantedBy=multi-user.target diff --git a/plugin/config.json b/plugin/config.json new file mode 100644 index 0000000..4964946 --- /dev/null +++ b/plugin/config.json @@ -0,0 +1,27 @@ +{ + "description": "Sqlite Logging Driver", + "documentation": "https://github.com/daschr/logsqlite", + "entrypoint": [ + "/bin/logsqlite", + "/etc/logsqlite/conf.ini" + ], + "network": { + "type": "host" + }, + "interface": { + "types": [ + "docker.logdriver/1.0" + ], + "socket": "logsqlite.sock" + }, + "env": [ + { + "name": "DEBUG", + "description": "Log debug messages", + "value": "false", + "settable": [ + "value" + ] + } + ] +} diff --git a/src/cleaner.rs b/src/cleaner.rs new file mode 100644 index 0000000..68e0708 --- /dev/null +++ b/src/cleaner.rs @@ -0,0 +1,159 @@ +use log::{debug, error, info}; +use sqlx::{Connection, SqliteConnection}; +use std::collections::HashMap; +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::time::Duration; +use tokio::{sync::RwLock, time}; + +use crate::config::LogConfig; + +#[derive(Clone)] +pub struct LogCleaner { + fifos: Arc>>, + containers: Arc>>, + dbs_path: String, +} + +impl LogCleaner { + pub fn new(dbs_path: String) -> Self { + LogCleaner { + fifos: Arc::new(RwLock::new(HashMap::new())), + containers: Arc::new(RwLock::new(HashMap::new())), + dbs_path, + } + } + + pub async fn add(&self, container_id: &str, fifo: PathBuf, log_conf: LogConfig) { + self.fifos + .write() + .await + .insert(fifo, container_id.to_string()); + + let mut map = self.containers.write().await; + map.insert(container_id.to_string(), log_conf); + } + + pub async fn remove(&self, fifo: &Path) -> Option { + let container_id: String = match self.fifos.write().await.remove(fifo) { + Some(v) => v, + None => return None, + }; + + self.containers.write().await.remove(&container_id) + } + + async fn get_first_tail_rowid( + con: &mut SqliteConnection, + tail: u64, + ) -> Result { + let count = match sqlx::query_as::("SELECT count(*) FROM logs") + .fetch_one(&mut *con) + .await + { + Ok(v) => v.0 as u64, + Err(sqlx::Error::RowNotFound) => { + return Ok(0); + } + Err(e) => { + return Err(e); + } + }; + + let rowid = + sqlx::query_as::("SELECT ROWID FROM logs LIMIT 1 OFFSET ?1") + .bind(if count > tail { + (count - tail) as i64 + } else { + 0 + }) + .fetch_one(&mut *con) + .await? + .0 as u64; + + Ok(rowid) + } + + async fn cleanup_db( + &self, + log_conf: &LogConfig, + con: &mut SqliteConnection, + ) -> Result<(), sqlx::Error> { + match (log_conf.cleanup_age, log_conf.cleanup_max_lines) { + (Some(cleanup_age), Some(max_lines)) => { + let rowid = Self::get_first_tail_rowid(con, max_lines as u64).await?; + + let mut max_time = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap(); + + max_time -= cleanup_age; + + debug!( + "cleanup: DELETE FROM logs WHERE ts < {} OR ROWID < {}", + max_time.as_nanos(), + rowid + ); + sqlx::query("DELETE FROM logs WHERE ts < ?1 OR ROWID < ?2") + .bind(max_time.as_nanos() as i64) + .bind(rowid as i64) + .execute(con) + .await?; + } + (Some(cleanup_age), None) => { + let mut max_time = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap(); + + max_time -= cleanup_age; + + debug!( + "cleanup: DELETE FROM logs WHERE ts < {}", + max_time.as_nanos() + ); + sqlx::query("DELETE FROM logs WHERE ts < ?1") + .bind(max_time.as_nanos() as i64) + .execute(con) + .await?; + } + (None, Some(max_lines)) => { + let rowid = Self::get_first_tail_rowid(con, max_lines as u64).await?; + + debug!("cleanup: DELETE FROM logs WHERE ROWID < {}", rowid); + sqlx::query("DELETE FROM logs WHERE ROWID < ?1") + .bind(rowid as i64) + .execute(con) + .await?; + } + (None, None) => (), // never happens + } + + Ok(()) + } + + pub async fn run(&self, cleanup_interval: Duration) -> Result<(), sqlx::Error> { + loop { + info!("starting cleanup"); + for (container, log_conf) in self.containers.read().await.iter() { + debug!( + "[cleanup] cleaning up container: {}, max_age: {:?} max_lines: {:?}", + container, log_conf.cleanup_age, log_conf.cleanup_max_lines + ); + let db_url = format!("sqlite://{}/{}", self.dbs_path, container); + match SqliteConnection::connect(&db_url).await { + Err(e) => { + error!("[cleanup] failed to open connection: {:?}", e); + } + Ok(mut con) => { + if let Err(e) = self.cleanup_db(&log_conf, &mut con).await { + error!("[cleanup] could not cleanup {}: {:?}", container, e); + } + } + } + } + info!("cleanup done"); + + time::sleep(cleanup_interval).await; + } + } +} diff --git a/src/config.rs b/src/config.rs new file mode 100644 index 0000000..fb033ec --- /dev/null +++ b/src/config.rs @@ -0,0 +1,231 @@ +use configparser::ini::Ini; +use serde::{Deserialize, Serialize}; +use std::collections::HashMap; +use std::num::ParseIntError; +use std::path::PathBuf; +use std::time::Duration; + +#[derive(Debug)] +pub struct Config { + pub unix_socket_path: PathBuf, + pub databases_dir: PathBuf, + pub state_database: PathBuf, + pub cleanup_interval: Duration, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct LogConfig { + pub max_lines_per_tx: u64, + pub max_size_per_tx: usize, // bytes + pub message_read_timeout: Duration, + pub cleanup_age: Option, + pub cleanup_max_lines: Option, + pub cleanup_interval: Duration, + pub delete_when_stopped: bool, +} + +#[allow(unused)] +pub enum ConfigSource { + File(T), + Text(T), +} + +#[derive(Debug, Serialize)] +pub enum ParsingError { + IniError(String), + ConfError(String), +} + +impl From for ParsingError { + fn from(value: String) -> Self { + ParsingError::IniError(value) + } +} + +impl From for ParsingError { + fn from(value: ParseIntError) -> Self { + ParsingError::ConfError(value.to_string()) + } +} + +fn get_dir>( + conf: &Ini, + section: &str, + name: &str, + default: T, +) -> Result { + match conf.get(section, name) { + Some(s) => { + let p = PathBuf::from(s.as_str()); + if !p.exists() { + return Err(ParsingError::ConfError(format!("{s} does not exist"))); + } + + if !p.is_dir() { + return Err(ParsingError::ConfError(format!("{s} is not a directory"))); + } + + Ok(p) + } + None => Ok(default.into()), + } +} + +fn parse_as_duration(v: &str) -> Result { + let pos = { + let mut r = 0; + for c in v.chars() { + if !c.is_ascii_digit() { + break; + } + r += 1; + } + r + }; + + if pos == 0 { + return Err(String::from("Cannot parse time: no number")); + } + + let mut num: u64 = v[0..pos].parse::().unwrap(); + match &v[pos..] { + "w" | "W" => num *= 7 * 24 * 60 * 60, + "d" | "D" => num *= 24 * 60 * 60, + "h" | "H" => num *= 60 * 60, + "m" | "M" => num *= 60, + "s" | "S" => (), + s => { + return Err(format!("Unknown time specifier \"{}\"", s)); + } + } + + Ok(Duration::from_secs(num)) +} + +fn parse_si_prefixed_size(v: &str) -> Result { + let pos = { + let mut r = 0; + for c in v.chars() { + if !c.is_ascii_digit() { + break; + } + r += 1; + } + r + }; + + if pos == 0 { + return Err(String::from("Cannot parse size: no number")); + } + + let mut num: u64 = v[0..pos].parse::().unwrap(); + match &v[pos..] { + "g" | "G" => num *= 1024 * 1024 * 1024, + "m" | "M" => num *= 1024 * 1024, + "k" | "K" => num *= 1024, + "b" | "B" => (), + s => { + return Err(format!("Unknown time specifier \"{}\"", s)); + } + } + + Ok(num as usize) +} + +impl TryFrom> for Config { + type Error = ParsingError; + + fn try_from(v: ConfigSource) -> Result { + let mut config = Ini::new(); + match v { + ConfigSource::File(f) => config.load(f)?, + ConfigSource::Text(t) => config.read(t)?, + }; + + let c = Config { + unix_socket_path: get_dir( + &config, + "general", + "plugins_dir", + "/var/run/docker/plugins/", + )? + .join("logsqlite.sock"), + databases_dir: get_dir(&config, "general", "databases_dir", "/var/spool/logsqlite/")?, + state_database: match config.get("general", "state_database") { + Some(p) => PathBuf::from(p), + None => { + return Err(ParsingError::ConfError(String::from( + "Missing \"state_database\"", + ))) + } + }, + cleanup_interval: config + .getuint("cleanup", "interval")? + .map(Duration::from_secs) + .unwrap_or(Duration::from_secs(30 * 60)), + }; + + Ok(c) + } +} + +impl Default for LogConfig { + fn default() -> Self { + LogConfig { + max_lines_per_tx: 10_000, + max_size_per_tx: 10 * 1024 * 1024, + message_read_timeout: Duration::from_millis(100), + cleanup_age: None, + cleanup_max_lines: None, + cleanup_interval: Duration::from_secs(10 * 60), + delete_when_stopped: true, + } + } +} + +impl TryFrom<&Option>> for LogConfig { + type Error = ParsingError; + + fn try_from(cust_conf: &Option>) -> Result { + let mut conf = LogConfig::default(); + + if cust_conf.is_none() { + return Ok(conf); + } + + let cust_conf = cust_conf.as_ref().unwrap(); + for (opt, val) in cust_conf.iter() { + match opt.as_str() { + "message_read_timeout" => { + conf.message_read_timeout = Duration::from_millis(val.parse::()?); + } + "max_lines_per_tx" => { + conf.max_lines_per_tx = val.parse::()?; + } + "max_size_per_tx" => { + conf.max_size_per_tx = parse_si_prefixed_size(val)?; + } + "cleanup_age" => { + conf.cleanup_age = Some(parse_as_duration(val)?); + } + "cleanup_max_lines" => { + conf.cleanup_max_lines = Some(val.parse::()?); + } + "delete_when_stopped" => { + conf.delete_when_stopped = match val.to_lowercase().as_str() { + "true" => true, + "false" => false, + _ => { + return Err(ParsingError::ConfError(String::from( + "delete_when_stopped is neither \"true\" or \"false\"", + ))); + } + } + } + _ => (), + } + } + + Ok(conf) + } +} diff --git a/src/docker.rs b/src/docker.rs new file mode 100644 index 0000000..0f60b6d --- /dev/null +++ b/src/docker.rs @@ -0,0 +1,198 @@ +use crate::config::{Config, LogConfig}; +use crate::logger::SqliteLogStream; +use crate::statehandler::StateHandlerMessage; + +use axum::{ + body::StreamBody, + debug_handler, + extract::{Json, State}, + http::Uri, + response::IntoResponse, +}; +use log::{debug, error, info}; +use serde::{Deserialize, Serialize}; +use serde_json::{json, Value}; +use std::{collections::HashMap, path::PathBuf, sync::Arc}; +use tokio::sync::mpsc::Sender; + +pub struct ApiState { + pub state_handler_tx: Sender, + pub config: Arc, +} + +impl ApiState { + pub async fn new( + state_handler_tx: Sender, + config: Arc, + ) -> Result { + Ok(ApiState { + state_handler_tx, + config, + }) + } +} + +#[allow(non_snake_case)] +#[derive(Debug, Serialize, Deserialize)] +pub struct Info { + Config: Option>, + ContainerID: String, + ContainerName: Option, + ContainerEntrypoint: Option, + ContainerArgs: Option>, + ContainerImageID: Option, + ContainerImageName: Option, + ContainerCreated: Option, + ContainerEnv: Option>, + ContainerLabels: Option>, + LogPath: Option, + DaemonName: Option, +} + +#[allow(non_snake_case)] +#[derive(Debug, Serialize, Deserialize)] +pub struct StartLoggingConf { + File: String, + Info: Info, +} + +pub async fn start_logging( + State(state): State>, + Json(conf): Json, +) -> Json { + info!("[start_logging] conf: {:?}", conf); + + state + .state_handler_tx + .send(StateHandlerMessage::StartLogging { + container_id: conf.Info.ContainerID.clone(), + fifo: PathBuf::from(conf.File.as_str()), + log_conf: match LogConfig::try_from(&conf.Info.Config) { + Ok(c) => c, + Err(e) => { + error!("Failed to parse config: {:?}: {:?}", conf.Info.Config, e); + return json!({"Err": e}).into(); + } + }, + }) + .await + .expect("failed to enqueue StateHandlerMessage"); + + info!("[docker::start_logging] send StartLogging message to StateHandler"); + + json!({"Err": ""}).into() +} + +#[allow(non_snake_case)] +#[derive(Debug, Serialize, Deserialize)] +pub struct StopLoggingConf { + File: String, +} + +#[debug_handler] +pub async fn stop_logging( + State(state): State>, + Json(conf): Json, +) -> Json { + info!("[stop_logging] conf: {:?}", conf); + + state + .state_handler_tx + .send(StateHandlerMessage::StopLogging { + fifo: PathBuf::from(conf.File), + }) + .await + .expect("Failed to enqueue StateHandlerMessage"); + info!("[docker::stop_logging] send StopLogging message to StateHandler"); + + json!({"Err": ""}).into() +} + +pub async fn capabilities() -> Json { + debug!("[capabilities] called"); + json!({"Cap": {"ReadLogs": true}}).into() +} + +#[allow(non_snake_case)] +#[derive(Debug, Serialize, Deserialize)] +pub struct ReadConfig { + Since: Option, + Until: Option, + Tail: Option, + Follow: Option, +} + +#[allow(non_snake_case)] +#[derive(Debug, Serialize, Deserialize)] +pub struct ReadLogsConf { + Config: ReadConfig, + Info: Info, +} + +/* +Object {"Config": Object {"Follow": Bool(false), "Since": String("0001-01-01T00:00:00Z"), +"Tail": Number(-1), "Until": String("0001-01-01T00:00:00Z")}, +"Info": Object {"Config": Object {}, "ContainerArgs": Array [], "ContainerCreated": String("2023-08-17T14:15:45.983205858Z"), "ContainerEntrypoint": String("bash"), "ContainerEnv": Array [String("PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin")], "ContainerID": String("71a33f78bc9b362c91de1746d54d8fea51a17bfddde1745c2f9fbfdb1e4d2b90"), "ContainerImageID": String("sha256:01f29b872827fa6f9aed0ea0b2ede53aea4ad9d66c7920e81a8db6d1fd9ab7f9"), "ContainerImageName": String("ubuntu"), "ContainerLabels": Object {"org.opencontainers.image.ref.name": String("ubuntu"), "org.opencontainers.image.version": String("22.04")}, "ContainerName": String("/wonderful_raman"), "DaemonName": String("docker"), "LogPath": String("")}} +*/ + +pub async fn read_logs( + State(state): State>, + Json(mut conf): Json, +) -> Result> { + info!("[read_logs] conf: {:?}", conf); + + let tail = match conf.Config.Tail { + Some(v) if v < 1 => None, + Some(v) => Some(v as u64), + None => None, + }; + + if conf + .Config + .Since + .as_mut() + .is_some_and(|x| x == "0001-01-01T00:00:00Z") + { + conf.Config.Since = None; + } + + if conf + .Config + .Until + .as_mut() + .is_some_and(|x| x == "0001-01-01T00:00:00Z") + { + conf.Config.Until = None; + } + + let logstream = match SqliteLogStream::new( + &state.config.databases_dir.as_path(), + &conf.Info.ContainerID, + conf.Config.Since, + conf.Config.Until, + tail, + conf.Config.Follow.unwrap_or(false), + ) + .await + { + Ok(l) => l, + Err(e) => { + error!("Error creating logstream: {:?}", &e); + return Err( + json!({ "Err": format!("[logsqlite] Could not read logs: {:?}", e) }).into(), + ); + } + }; + + Ok(StreamBody::new(logstream)) +} + +pub async fn activate() -> Json { + debug!("activate called"); + json!({"Implements": ["LogDriver"]}).into() +} + +pub async fn fallback(uri: Uri) -> &'static str { + debug!("[fallback] uri: {:?}", uri); + "not found" +} diff --git a/src/entry.proto b/src/entry.proto new file mode 100644 index 0000000..a74787f --- /dev/null +++ b/src/entry.proto @@ -0,0 +1,17 @@ +syntax = "proto3"; + +package docker.logentry; + +message LogEntry { + string source = 1; + int64 time_nano = 2; + bytes line = 3; + bool partial = 4; + PartialLogEntryMetadata partial_log_metadata = 5; +} + +message PartialLogEntryMetadata { + bool last = 1; + string id = 2; + int32 ordinal = 3; +} diff --git a/src/logger.rs b/src/logger.rs new file mode 100644 index 0000000..ed574ac --- /dev/null +++ b/src/logger.rs @@ -0,0 +1,465 @@ +use crate::config::{Config, LogConfig}; +use chrono::DateTime; +use log::{debug, error, info, warn}; + +use crate::statehandler::StateHandlerMessage; +use tokio::sync::mpsc::Sender; + +use core::pin::Pin; +use futures::{ + executor::block_on, + stream::Stream, + task::{Context, Poll}, +}; +use sqlx::{sqlite::SqliteConnectOptions, ConnectOptions, SqliteConnection}; +use std::path::Path; +use std::{cell::UnsafeCell, collections::HashMap, path::PathBuf, str::FromStr, sync::Arc}; + +use prost::Message; +use tokio::{ + self, + fs::File, + io::{AsyncRead, AsyncReadExt}, + sync::RwLock, + task::JoinHandle, + time::{sleep, timeout, Duration, Instant}, +}; + +// Include the `items` module, which is generated from items.proto. +pub mod logentry { + include!(concat!(env!("OUT_DIR"), "/docker.logentry.rs")); +} + +pub struct Logger { + exit: RwLock, +} + +#[derive(Debug)] +pub enum LoggerError { + IoError(std::io::Error), + SqlError(sqlx::Error), + JoinError(tokio::task::JoinError), + DecodeError(prost::DecodeError), + EncodeError(prost::EncodeError), + Exited, +} + +impl From for LoggerError { + fn from(e: std::io::Error) -> Self { + LoggerError::IoError(e) + } +} + +impl From for LoggerError { + fn from(e: sqlx::Error) -> Self { + LoggerError::SqlError(e) + } +} + +impl From for LoggerError { + fn from(e: tokio::task::JoinError) -> Self { + LoggerError::JoinError(e) + } +} + +impl From for LoggerError { + fn from(e: prost::DecodeError) -> Self { + LoggerError::DecodeError(e) + } +} + +impl From for LoggerError { + fn from(e: prost::EncodeError) -> Self { + LoggerError::EncodeError(e) + } +} + +impl Logger { + fn new() -> Self { + Logger { + exit: RwLock::new(false), + } + } + + async fn read_protobuf( + &self, + reader: &mut R, + msg: &mut Vec, + read_timeout: Duration, + ) -> Result, LoggerError> { + #[allow(unused_assignments)] + let mut msg_size = 0usize; + + match timeout(read_timeout, reader.read_u32()).await { + Ok(Ok(v)) => { + msg_size = v as usize; + } + Ok(Err(e)) => { + // this is some serious error and we cannot continue, the fifo may be closed + return Err(e.into()); + } + Err(_) => { + return Ok(None); + } + } + + let mut read = 0; + let mut bf = [0u8; 1024]; + msg.clear(); + + while read < msg_size { + let tbr = if msg_size - read >= bf.len() { + bf.len() + } else { + msg_size - read + }; + let read_bytes = reader.read(&mut bf[0..tbr]).await?; + + read += read_bytes; + msg.extend_from_slice(&bf[0..read_bytes]); + } + + let mut dec_msg = logentry::LogEntry::decode(msg.as_slice())?; + dec_msg.line.push(b'\n'); + + msg.clear(); + msg.extend_from_slice(&(dec_msg.encoded_len() as u32).to_be_bytes()); + msg.reserve(dec_msg.encoded_len()); + dec_msg.encode(msg)?; + + Ok(Some(dec_msg.time_nano as u64)) + } + + async fn log( + &self, + fifo: PathBuf, + db_path: PathBuf, + log_conf: LogConfig, + ) -> Result<(), LoggerError> { + let mut dbcon = + SqliteConnectOptions::from_str(&format!("sqlite://{}", &db_path.display()))? + .create_if_missing(true) + .connect() + .await?; + + sqlx::query( + "CREATE TABLE IF NOT EXISTS logs (ts NUMBER, message BLOB); + CREATE INDEX IF NOT EXISTS idx_ts ON logs(ts);", + ) + .execute(&mut dbcon) + .await?; + + let mut fd = File::open(&fifo).await?; + let mut message: Vec = Vec::new(); + + sqlx::query("BEGIN TRANSACTION;") + .execute(&mut dbcon) + .await?; + + let mut no_entries_ts = Instant::now(); + let mut nb_entries = 0; + let mut acc_entries_size = 0; + + while !*self.exit.read().await { + debug!("read_protobuf from {}...", fifo.display()); + match self + .read_protobuf(&mut fd, &mut message, log_conf.message_read_timeout) + .await + { + Ok(Some(ts)) => { + sqlx::query("INSERT INTO logs(ts, message) VALUES(?1, ?2)") + .bind(ts as i64) + .bind(&message) + .persistent(true) + .execute(&mut dbcon) + .await?; + + nb_entries += 1; + acc_entries_size += message.len(); + + if nb_entries >= log_conf.max_lines_per_tx + || acc_entries_size >= log_conf.max_size_per_tx + { + sqlx::query("END TRANSACTION;BEGIN TRANSACTION;") + .execute(&mut dbcon) + .await?; + + info!( + "reached {} entries with {} bytes size, ending current transacion ({:.2} lines/s) for {}", + nb_entries, + acc_entries_size, + nb_entries as f64 + * (1_000_000f64 + / Instant::now().duration_since(no_entries_ts).as_micros() + as f64), + fifo.display() + ); + + no_entries_ts = Instant::now(); + nb_entries = 0; + acc_entries_size = 0; + } + } + Ok(None) => { + sqlx::query("END TRANSACTION;BEGIN TRANSACTION;") + .execute(&mut dbcon) + .await?; + + no_entries_ts = Instant::now(); + nb_entries = 0; + acc_entries_size = 0; + } + Err(e) => { + warn!("Failed to read a protobuf message: {:?}", e); + return Err(e); + } + } + } + + sqlx::query("END TRANSACTION;").execute(&mut dbcon).await?; + + Err(LoggerError::Exited) + } + + async fn exit(&self) { + *self.exit.write().await = true; + } +} + +pub struct LoggerPool { + workers: RwLock, tokio::task::JoinHandle<()>)>>, + config: Arc, +} + +impl LoggerPool { + pub fn new(config: Arc) -> Self { + LoggerPool { + workers: RwLock::new(HashMap::new()), + config, + } + } + + pub async fn start_logging( + &self, + container_id: &str, + fifo_path: PathBuf, + log_conf: LogConfig, + tx: Sender, + ) { + let logger = Arc::new(Logger::new()); + let mut db_path = self.config.databases_dir.clone(); + db_path.push(container_id); + + let f_path = fifo_path.clone(); + let c_l = logger.clone(); + + let c_container_id = container_id.to_string(); + + let handle = tokio::spawn(async move { + tx.send(StateHandlerMessage::LoggingStopped { + container_id: c_container_id, + fifo: f_path.clone(), + result: c_l.log(f_path, db_path, log_conf).await, + }) + .await + .expect("Could not enqueue StateHandlerMessage"); + }); + + self.workers + .write() + .await + .insert(fifo_path, (logger.clone(), handle)); + } + + pub async fn stop_logging(&self, fifo_path: &Path) { + let res = self.workers.write().await.remove(fifo_path); + + if let Some((logger, handle)) = res { + logger.exit().await; + handle.await.ok(); + } + } +} + +// DIRTY WORKAROUND, since we cannot know if the client is disconnected +// so that we can retun Poll::Ready(None) instant of Poll::Pending + +const FOLLOW_WAKETIME: u64 = 1; +const FOLLOW_COUNTER_MAX: usize = 60 * 60; + +#[derive(Debug)] +pub struct SqliteLogStream { + stmt_s: String, + parameters: Vec, + next_rowid: u64, + counter: u64, + tail: Option, + follow: bool, + follow_counter: usize, + con: UnsafeCell, + waker: Option>, +} + +impl SqliteLogStream { + pub async fn new( + dbs_path: &Path, + container_id: &str, + since: Option, + until: Option, + tail: Option, + follow: bool, + ) -> Result { + let db_path = format!("{}/{}", dbs_path.display(), container_id); + debug!("[SqliteLogStream] db_path: {}", &db_path); + let mut con = SqliteConnectOptions::from_str(&format!("sqlite://{}", &db_path))? + .immutable(true) + .read_only(true) + .serialized(false) + .connect() + .await?; + + let mut cond = String::from("WHERE ROWID >= ?1"); + let mut parameters: Vec = vec![0]; + + if since.is_some() { + if let Ok(time) = DateTime::parse_from_str(since.as_ref().unwrap().as_str(), "%+") { + let since = time + .timestamp_nanos_opt() + .expect("timestamp in nanosecs does not fit in i64 anymore"); + + cond.push_str(&format!(" AND ts>=?{}", parameters.len() + 1)); + parameters.push(since as u64); + } + }; + + if until.is_some() { + if let Ok(time) = DateTime::parse_from_str(until.as_ref().unwrap().as_str(), "%+") { + let until = time + .timestamp_nanos_opt() + .expect("timestamp in nanosecs does not fit in i64 anymore"); + + cond.push_str(&format!(" AND ts<=?{}", parameters.len() + 1)); + parameters.push(until as u64); + } + }; + + let mut first_rowid = 1u64; + if let Some(tail) = tail { + let stmt_s = format!("SELECT count(*) FROM logs {}", cond); + debug!("stmt_s: {} params {:?}", stmt_s, ¶meters); + + let mut stmt = sqlx::query_as::<_, (i64,)>(&stmt_s); + for param in ¶meters { + stmt = stmt.bind(*param as i64); + } + + let nrows: u64 = stmt.fetch_one(&mut con).await?.0 as u64; + + let stmt_s = format!( + "SELECT ROWID FROM logs {} LIMIT 1 OFFSET ?{}", + cond, + parameters.len() + 1 + ); + + debug!("stmt_s: {}", &stmt_s); + + stmt = sqlx::query_as::<_, (i64,)>(&stmt_s); + + parameters.push(if nrows > tail { nrows - tail } else { 0 }); + + for param in ¶meters { + stmt = stmt.bind(*param as i64); + } + + first_rowid = stmt.fetch_one(&mut con).await?.0 as u64; + + parameters.pop(); + debug!("first_rowid: {}", first_rowid); + } + + let stmt_s = format!("SELECT ROWID,message FROM logs {} LIMIT 1", cond); + + Ok(SqliteLogStream { + stmt_s, + parameters, + next_rowid: first_rowid, + counter: 0, + tail: if follow { None } else { tail }, + follow, + follow_counter: 0, + con: UnsafeCell::new(con), + waker: None, + }) + } +} + +impl Stream for SqliteLogStream { + type Item = Result, &'static str>; + + fn poll_next(mut self: Pin<&mut Self>, ctx: &mut Context<'_>) -> Poll> { + if self.tail.is_some() && self.counter >= self.tail.unwrap() { + return Poll::Ready(None); + } + + self.parameters[0] = self.next_rowid; + let res = { + let stmt_s = self.stmt_s.as_str(); + let mut stmt = sqlx::query_as::<_, (i64, Vec)>(stmt_s); + for param in &self.parameters { + stmt = stmt.bind(*param as i64); + } + + let res: Option<(i64, Vec)> = + match block_on(stmt.fetch_one(unsafe { &mut *self.con.get() })) { + Ok(s) => Some(s), + Err(sqlx::Error::RowNotFound) => None, + Err(e) => { + error!("[SqliteLogStream] Got SQL error: {:?}", e); + return Poll::Ready(None); + } + }; + + debug!( + "[stream] {:?} [{}]", + res, + if res.is_some() { + res.as_ref().unwrap().1.len() + } else { + 0 + } + ); + + res + }; + + if res.is_some() { + self.counter += 1; + self.follow_counter = 0; + self.next_rowid = res.as_ref().unwrap().0 as u64 + 1; + } + + match res { + Some((_, r)) => Poll::Ready(Some(Ok(r))), + None if self.follow && self.follow_counter < FOLLOW_COUNTER_MAX => { + self.follow_counter += 1; + let waker = ctx.waker().clone(); + self.waker = Some(tokio::spawn(async move { + sleep(Duration::from_secs(FOLLOW_WAKETIME)).await; + waker.wake(); + })); + + Poll::Pending + } + None => Poll::Ready(None), + } + } +} + +impl Drop for SqliteLogStream { + fn drop(&mut self) { + if let Some(waker) = self.waker.as_mut() { + debug!("aborting new_entry_waker"); + waker.abort(); + } + } +} diff --git a/src/main.rs b/src/main.rs new file mode 100644 index 0000000..407bc5e --- /dev/null +++ b/src/main.rs @@ -0,0 +1,122 @@ +mod cleaner; +mod config; +mod docker; +mod logger; +mod statehandler; + +use axum::http::Request; +use axum::middleware::map_request; +use axum::{routing::post, Router, Server}; +use docker::ApiState; +use hyperlocal::UnixServerExt; +use log::{self, debug, error}; +use statehandler::{State, StateHandler}; +use std::{env, process::exit, sync::Arc}; +use tokio::task; + +async fn normalize_dockerjson(mut req: Request) -> Request { + let headers = req.headers_mut(); + match headers.get("content-type") { + Some(ct) => { + debug!("[normalize_dockerjson] {:?}", ct); + } + None => { + headers.insert("content-type", "application/json".parse().unwrap()); + } + } + + req +} + +#[tokio::main] +async fn main() -> Result<(), config::ParsingError> { + simple_logger::init_with_level(env::var("DEBUG").map_or_else( + |_| log::Level::Info, + |s| { + if matches!(s.to_lowercase().as_str(), "true" | "1") { + log::Level::Debug + } else { + log::Level::Info + } + }, + )) + .expect("could not set loglevel"); + + let args: Vec = env::args().collect(); + if args.len() < 2 { + eprintln!("Usage: {} [path to config]", args[0]); + return Ok(()); + } + + let conf: Arc = Arc::new(config::Config::try_from( + config::ConfigSource::File(args[1].clone()), + )?); + + debug!("config: {:?}", &conf); + + let state = Arc::new(State::new(conf.clone()).await); + + let (mut state_handler, state_handler_tx) = match StateHandler::new(state.clone()).await { + Ok(s) => s, + Err(e) => { + eprintln!("failed to initialize StateHandler: {:?}", e); + exit(1); + } + }; + + let api_state = Arc::new( + ApiState::new(state_handler_tx, conf.clone()) + .await + .expect("Failed to create ApiState"), + ); + + let cleaner = state.cleaner.clone(); + let cleanup_interval = conf.cleanup_interval.clone(); + task::spawn(async move { + match cleaner.run(cleanup_interval).await { + Ok(()) => { + error!("Cleaner exited!"); + } + Err(e) => { + error!("Error running cleaner: {:?}", e); + } + } + }); + + if let Err(e) = state_handler.replay_state().await { + eprintln!("Failed to replay state: {:?}", e); + exit(1); + } + + let state_handler_handle = tokio::spawn(async move { + if let Err(e) = state_handler.handle().await { + error!("Error at StateHandler: {:?}", e); + } + }); + + let router = Router::new() + .route("/LogDriver.StartLogging", post(docker::start_logging)) + .route("/LogDriver.StopLogging", post(docker::stop_logging)) + .route("/LogDriver.Capabilities", post(docker::capabilities)) + .route("/LogDriver.ReadLogs", post(docker::read_logs)) + .route("/Plugin.Activate", post(docker::activate)) + .layer(map_request(normalize_dockerjson)) + .fallback(docker::fallback) + .with_state(api_state); + let builder = Server::bind_unix(conf.unix_socket_path.as_path()) + .expect("could not listen on unix socket"); + + let api_server_handle = + tokio::spawn(async move { builder.serve(router.into_make_service()).await.unwrap() }); + + tokio::select! { + _ = state_handler_handle => { + error!("StateHandler exited!"); + } + _ = api_server_handle => { + error!("DockerAPI server exited!"); + } + }; + + Ok(()) +} diff --git a/src/statehandler.rs b/src/statehandler.rs new file mode 100644 index 0000000..de3e13a --- /dev/null +++ b/src/statehandler.rs @@ -0,0 +1,220 @@ +use bincode::{deserialize, serialize}; +use futures_util::StreamExt; +use sqlx::Sqlite; +use sqlx::{sqlite::SqliteConnectOptions, ConnectOptions, Error as SqlxError, SqliteConnection}; +use std::fs; +use std::sync::Arc; +use std::{ + path::{Path, PathBuf}, + str::FromStr, +}; +use tokio::sync::mpsc::{channel, Receiver, Sender}; + +use crate::cleaner::LogCleaner; +use crate::config::{Config, LogConfig}; +use crate::log::{info, warn}; +use crate::logger::{LoggerError, LoggerPool}; + +pub struct State { + logger_pool: LoggerPool, + config: Arc, + pub cleaner: LogCleaner, +} + +impl State { + pub async fn new(config: Arc) -> Self { + State { + logger_pool: LoggerPool::new(config.clone()), + cleaner: LogCleaner::new(config.databases_dir.display().to_string()), + config, + } + } + + async fn start_logging( + &self, + container_id: &str, + fifo: &Path, + log_conf: LogConfig, + tx: Sender, + ) { + self.cleaner + .add(&container_id, fifo.to_path_buf(), log_conf.clone()) + .await; + + self.logger_pool + .start_logging(&container_id, fifo.to_path_buf(), log_conf, tx) + .await; + } +} + +pub enum StateHandlerMessage { + StartLogging { + container_id: String, + fifo: PathBuf, + log_conf: LogConfig, + }, + StopLogging { + fifo: PathBuf, + }, + LoggingStopped { + container_id: String, + fifo: PathBuf, + result: Result<(), LoggerError>, + }, +} + +pub struct StateHandler { + dbcon: SqliteConnection, + rx: Receiver, + tx: Sender, + state: Arc, +} + +impl StateHandler { + pub async fn new(state: Arc) -> Result<(Self, Sender), SqlxError> { + let mut dbcon = SqliteConnectOptions::from_str(&format!( + "sqlite://{}", + state.config.state_database.display() + ))? + .create_if_missing(true) + .connect() + .await?; + + sqlx::query( + "CREATE TABLE IF NOT EXISTS active_fetches(container_id text PRIMARY KEY, fifo text, log_conf blob)", + ) + .execute(&mut dbcon) + .await?; + + let (tx, rx) = channel(1024); + + Ok(( + StateHandler { + dbcon, + rx, + tx: tx.clone(), + state, + }, + tx, + )) + } + + pub async fn handle(&mut self) -> Result<(), SqlxError> { + while let Some(msg) = self.rx.recv().await { + match msg { + StateHandlerMessage::StartLogging { + container_id, + fifo, + log_conf, + } => { + info!("[StateHandler] starting logging of {}", container_id); + + let blob_log_conf: Vec = serialize(&log_conf).unwrap(); + sqlx::query( + "INSERT OR REPLACE INTO active_fetches(container_id, fifo, log_conf) VALUES (?1, ?2, ?3)", + ) + .bind(&container_id) + .bind(fifo.display().to_string()) + .bind(blob_log_conf) + .execute(&mut self.dbcon) + .await?; + + self.state + .start_logging(&container_id, &fifo, log_conf, self.tx.clone()) + .await; + } + StateHandlerMessage::StopLogging { fifo } => { + info!("[StateHandler] stopping logging of fifo {}", fifo.display()); + + sqlx::query("DELETE FROM active_fetches where fifo = ?1") + .bind(fifo.as_path().display().to_string()) + .execute(&mut self.dbcon) + .await?; + + self.state.logger_pool.stop_logging(&fifo).await; + } + StateHandlerMessage::LoggingStopped { + container_id, + fifo, + result, + } => { + if let Err(e) = result { + warn!( + "[StateHandler] logger of container {} returned: {:?}", + container_id, e + ); + match e { + LoggerError::DecodeError(_) => { + info!( + "[StateHandler] restarting logging of {} using fifo {}, since last error was a decode error", + &container_id, + fifo.display() + ); + + if let Some(log_conf) = self.state.cleaner.remove(&fifo).await { + self.state + .start_logging( + &container_id, + &fifo, + log_conf, + self.tx.clone(), + ) + .await; + } else { + warn!("Container {} stopped but it had no LogConfig in the cleaner!", container_id); + } + } + _ => { + sqlx::query("DELETE FROM active_fetches where fifo = ?1") + .bind(fifo.as_path().display().to_string()) + .execute(&mut self.dbcon) + .await?; + + if let Some(log_conf) = self.state.cleaner.remove(&fifo).await { + info!("log_conf: '{:?}'", &log_conf); + if log_conf.delete_when_stopped { + let mut dbpath = self.state.config.databases_dir.clone(); + dbpath.push(container_id); + + info!("removing {}", dbpath.display()); + fs::remove_file(dbpath).ok(); + } + } + } + } + } + } + } + } + + Ok(()) + } + + pub async fn replay_state(&mut self) -> Result<(), sqlx::Error> { + let mut stream = sqlx::query_as::)>( + "SELECT container_id, fifo FROM active_fetches", + ) + .fetch(&mut (self.dbcon)); + + while let Some(r) = stream.next().await { + let (container_id, fifo, blob_log_conf) = r?; + + let log_conf: LogConfig = deserialize(&blob_log_conf).unwrap(); + info!( + "[StateHandler] replaying logging of {} using fifo {}", + container_id, fifo + ); + + self.state + .start_logging( + &container_id, + PathBuf::from(fifo).as_path(), + log_conf, + self.tx.clone(), + ) + .await; + } + + Ok(()) + } +}