feat: storing tasks runs in Sqlite DB

This commit is contained in:
Matthieu Bessat 2024-07-21 17:03:37 +02:00
parent 41443e5a7d
commit 0d322ee70c
12 changed files with 274 additions and 112 deletions

35
Cargo.lock generated
View file

@ -148,7 +148,9 @@ dependencies = [
"axum",
"chrono",
"clap",
"env_logger",
"fully_pub",
"log",
"minijinja",
"serde",
"serde_json",
@ -499,6 +501,29 @@ dependencies = [
"serde",
]
[[package]]
name = "env_filter"
version = "0.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a009aa4810eb158359dda09d0c87378e4bbb89b5a801f016885a4707ba24f7ea"
dependencies = [
"log",
"regex",
]
[[package]]
name = "env_logger"
version = "0.11.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "38b35839ba51819680ba087cd351788c9a3c476841207e0b8cee0b04722343b9"
dependencies = [
"anstream",
"anstyle",
"env_filter",
"humantime",
"log",
]
[[package]]
name = "equivalent"
version = "1.0.1"
@ -825,6 +850,12 @@ dependencies = [
"libm",
]
[[package]]
name = "humantime"
version = "2.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9a3a5bfb195931eeb336b2a7b4d761daec841b97f947d34394601737a7bba5e4"
[[package]]
name = "hyper"
version = "1.3.1"
@ -993,9 +1024,9 @@ dependencies = [
[[package]]
name = "log"
version = "0.4.21"
version = "0.4.22"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "90ed8c1e510134f979dbc4f070f87d4313098b704861a105fe34231c70a3901c"
checksum = "a7a70ba024b9dc04c27ea2f0c0548feb474ec5c54bba33a7f72f873a39d07b24"
[[package]]
name = "matchit"

View file

@ -17,4 +17,6 @@ axum = { version = "0.7.5", features = ["json"] }
minijinja = { version = "1.0.20", features = ["builtins"] }
uuid = { version = "1.8.0", features = ["serde", "v4"] }
fully_pub = "0.1.4"
log = "0.4.22"
env_logger = "0.11.3"

17
config.example.yaml Normal file
View file

@ -0,0 +1,17 @@
tasks:
- id: do_magic_stuff
name: Do magic incantation
env:
PYTHONUNBUFFERED: "1"
SIMULATION_SPEED: 0.2
command:
- /usr/bin/python3
- /home/mbess/workspace/autotasker/examples/do_something_1.py
store_logs: true
- id: reindex_db
name: Reindex the whole database
env: {}
command:
- ls
- /etc/fstab

View file

@ -1,11 +1,12 @@
CREATE TABLE IF NOT EXISTS task_runs (
DROP TABLE IF EXISTS task_runs;
CREATE TABLE task_runs (
id TEXT PRIMARY KEY,
task_id TEXT NOT NULL,
status TEXT CHECK(status IN ('pending','running','failed','success')) NOT NULL DEFAULT 'pending',
trigger_mode TEXT CHECK(status IN ('manual','webhook','schedule')) NOT NULL,
trigger_mode TEXT CHECK(trigger_mode IN ('manual','webhook','schedule')) NOT NULL DEFAULT 'manual',
exit_code INT,
logs TEXT,
submitted_at DATETIME,
started_at DATETIME,
end_at DATETIME
)
ended_at DATETIME
);

View file

@ -1,21 +1,9 @@
use crate::models::{Config, Task};
use anyhow::{anyhow, Context, Result};
use crate::models::{Task, TaskRun, TaskRunSummary};
use axum::extract::{Path as ExtractPath, State};
use axum::http::{Response, StatusCode};
use axum::http::StatusCode;
use axum::Json;
use axum::{response::Html, routing::get, Router};
use axum::response::Html;
use minijinja::render;
use serde::{Deserialize, Serialize};
use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};
use sqlx::{ConnectOptions, Connection, Pool, Sqlite};
use std::collections::HashMap;
use std::fs;
use std::process::Stdio;
use std::str::FromStr;
use std::sync::Arc;
use tokio::io::{AsyncBufReadExt, BufReader};
use tokio::process::{Child, Command};
use tokio::sync::mpsc::{self, Receiver, Sender};
use uuid::Uuid;
use crate::models::ExecutorOrder;
@ -24,7 +12,6 @@ use crate::AppState;
pub async fn home() -> Html<String> {
Html(render!(
include_str!("./templates/home.html"),
first_name => "John Doe"
))
}
@ -40,7 +27,6 @@ pub async fn trigger_task(
State(app_state): State<AppState>,
ExtractPath(task_id): ExtractPath<String>,
) -> (StatusCode, Html<String>) {
println!("Run task {}", task_id);
let tasks: Vec<Task> = app_state.config.tasks;
let task = match tasks.iter().find(|t| t.id == task_id) {
Some(t) => t,
@ -70,7 +56,7 @@ pub async fn trigger_task(
}
pub async fn handle_webhook(
State(app_state): State<AppState>,
State(_app_state): State<AppState>,
ExtractPath(token): ExtractPath<String>,
) -> (StatusCode, Json<String>) {
println!("Webhook token {}", token);
@ -82,16 +68,41 @@ pub async fn list_task_runs(
State(app_state): State<AppState>,
ExtractPath(task_id): ExtractPath<String>,
) -> Html<String> {
Html(render!(include_str!("./templates/list_task_runs.html")))
let runs = sqlx::query_as::<_, TaskRunSummary>("SELECT id,status,trigger_mode,submitted_at,started_at,ended_at FROM task_runs WHERE task_id = $1 LIMIT 100")
.bind(&task_id)
.fetch_all(&app_state.db)
.await
.unwrap();
let task = match app_state.config.tasks.iter().find(|t| t.id == task_id) {
Some(v) => v,
None => {
return Html("<h1>Task not found</h1>".to_string());
}
};
Html(render!(
include_str!("./templates/list_task_runs.html"),
task => task,
runs => runs,
))
}
pub async fn get_task_run(
State(app_state): State<AppState>,
ExtractPath(task_id): ExtractPath<String>,
ExtractPath(run_id): ExtractPath<String>,
ExtractPath((_task_id, run_id)): ExtractPath<(String, String)>,
) -> Html<String> {
let run_details = match sqlx::query_as::<_, TaskRun>("SELECT * FROM task_runs WHERE id = $1")
.bind(run_id)
.fetch_one(&app_state.db)
.await {
Ok(v) => v,
Err(e) => {
dbg!(e);
return Html("<h1>Task run not found</h1>".to_string());
}
};
Html(render!(
include_str!("./templates/task_run_details.html"),
run => false
run => run_details
))
}

85
src/executor.rs Normal file
View file

@ -0,0 +1,85 @@
use anyhow::{anyhow, Result};
use chrono::{SecondsFormat, Utc};
use log::{debug, info, trace, error};
use std::process::Stdio;
use tokio::io::{AsyncBufReadExt, BufReader};
use tokio::process::Command;
use tokio::sync::mpsc::Receiver;
use crate::models::ExecutorOrder;
use crate::AppState;
async fn run_task(state: AppState, order: ExecutorOrder) -> Result<()> {
let executable = match order.task.command.first() {
Some(v) => v,
None => return Err(anyhow!("Could not find command to execute")),
};
debug!("Start processing of order {}", order.id);
// save in DB
let _result = sqlx::query("INSERT INTO task_runs (id, task_id, trigger_mode, status, submitted_at, started_at) VALUES ($1, $2, $3, $4, $5, $5)")
.bind(order.id.to_string())
.bind(order.task.id.clone())
.bind("manual")
.bind("running")
.bind(Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true))
.execute(&state.db)
.await.unwrap();
let mut cmd = Command::new(executable);
cmd.args(order.task.command.iter().skip(1).collect::<Vec<&String>>())
.stdout(Stdio::piped());
for (key, val) in order.task.env.iter() {
cmd.env(key, val);
}
let mut child = cmd.spawn().expect("Failed to execute process");
let stdout = child.stdout.take().unwrap();
let process_handle = tokio::spawn(async move {
let status = child
.wait()
.await
.expect("child process encountered an error");
debug!("Child process done, exit status is {}", status);
status
});
let mut collected_logs = String::new();
let mut lines = BufReader::with_capacity(16, stdout).lines();
while let Some(line) = lines.next_line().await.unwrap() {
trace!("{}: {}", order.id, line);
collected_logs += &format!("{}\n", line);
}
let status = process_handle.await?;
if !status.success() {
error!("Non successful exit code found {}", status);
}
let _result = sqlx::query("UPDATE task_runs SET status = $2, ended_at = $3, exit_code = $4, logs = $5 WHERE id = $1")
.bind(order.id.to_string())
.bind(match status.success() {
true => "success",
false => "failed"
})
.bind(Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true))
.bind(status.code())
.bind(collected_logs)
.execute(&state.db)
.await.unwrap();
debug!("End of task, details saved");
Ok(())
}
pub async fn run_executor(state: AppState, mut rx: Receiver<ExecutorOrder>) {
debug!("Executor started");
while let Some(order) = rx.recv().await {
println!("Got Order: {:?}", order);
let local_state = state.clone();
tokio::spawn(async {
run_task(local_state, order).await
});
}
info!("Executor stopped");
}

View file

@ -1,25 +1,19 @@
use anyhow::{anyhow, Context, Result};
use axum::extract::{Path as ExtractPath, State};
use axum::http::{Response, StatusCode};
use axum::routing::get;
use axum::{Json, Router};
use minijinja::render;
use models::{Config, Task};
use serde::{Deserialize, Serialize};
use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};
use sqlx::{ConnectOptions, Connection, Pool, Sqlite};
use std::collections::HashMap;
use std::fs;
use std::process::Stdio;
use std::str::FromStr;
use std::sync::Arc;
use tokio::io::{AsyncBufReadExt, BufReader};
use tokio::process::{Child, Command};
use tokio::sync::mpsc::{self, Receiver, Sender};
use uuid::Uuid;
mod controllers;
mod models;
mod executor;
use log::info;
use anyhow::{anyhow, Context, Result};
use axum::routing::get;
use axum::Router;
use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};
use sqlx::{ConnectOptions, Pool, Sqlite};
use std::fs;
use std::str::FromStr;
use std::sync::Arc;
use tokio::sync::mpsc::{self, Sender};
use executor::run_executor;
use models::{Config, ExecutorOrder};
#[derive(Debug, Clone)]
pub struct AppState {
@ -32,64 +26,21 @@ fn get_config() -> Result<Config> {
let inp_def_yaml = fs::read_to_string("./config.yaml")
.expect("Should have been able to read the the config file");
return serde_yaml::from_str(&inp_def_yaml)
.map_err(|e| anyhow!("Failed to parse config, {:?}", e));
serde_yaml::from_str(&inp_def_yaml)
.map_err(|e| anyhow!("Failed to parse config, {:?}", e))
}
use crate::models::ExecutorOrder;
async fn run_task(order: ExecutorOrder) {
let executable = match order.task.command.iter().next() {
Some(v) => v,
None => return,
};
println!("Start processing of order {}", order.id);
let mut cmd = Command::new(executable);
cmd.args(order.task.command.iter().skip(1).collect::<Vec<&String>>())
.stdout(Stdio::piped());
for (key, val) in order.task.env.iter() {
cmd.env(key, val);
}
let mut child = cmd.spawn().expect("failed to execute process");
let stdout = child.stdout.take().unwrap();
tokio::spawn(async move {
let status = child
.wait()
.await
.expect("child process encountered an error");
println!("child status was: {}", status);
});
let mut lines = BufReader::with_capacity(16, stdout).lines();
while let Some(line) = lines.next_line().await.unwrap() {
println!("{}: {}", order.id, line);
}
println!("End of task")
}
async fn run_executor(mut rx: Receiver<ExecutorOrder>) {
println!("Executor started");
while let Some(order) = rx.recv().await {
println!("Got Order: {:?}", order);
tokio::spawn(async { run_task(order).await });
}
println!("Executor stopped");
}
#[tokio::main]
async fn main() -> Result<()> {
env_logger::init();
info!("Starting autotasker");
let pool = prepare_database().await.context("Prepare db")?;
// start executor daemon
// start channel to talk to executor daemon
let (tx, rx) = mpsc::channel::<ExecutorOrder>(32);
let executor_handle = tokio::spawn(async { run_executor(rx).await });
let config: Config = get_config().expect("Cannot get config");
let state = AppState {
config,
@ -97,6 +48,12 @@ async fn main() -> Result<()> {
executor_tx: Arc::new(tx),
};
// start executor daemon
let executor_app_state = state.clone();
let executor_handle = tokio::spawn(async {
run_executor(executor_app_state, rx).await
});
// build our application with a single route
let app = Router::new()
.route("/", get(controllers::home))
@ -110,27 +67,33 @@ async fn main() -> Result<()> {
.route("/webhooks/:token", get(controllers::handle_webhook))
.with_state(state);
let listener = tokio::net::TcpListener::bind("0.0.0.0:8085").await.unwrap();
let listen_addr = "0.0.0.0:8085";
info!("Starting web server on http://{}", &listen_addr);
let listener = tokio::net::TcpListener::bind(listen_addr).await.unwrap();
axum::serve(listener, app).await?;
executor_handle.await?;
Ok(())
}
async fn prepare_database() -> Result<Pool<Sqlite>> {
// create database if it does not exist
let conn = SqliteConnectOptions::from_str("sqlite:autotasker.db")?
.create_if_missing(true)
.connect()
.await?;
let _ = conn.close().await;
let conn_str = "sqlite:./tmp/dbs/autotasker.db";
// // create database if it does not exist
// let conn = SqliteConnectOptions::from_str(&conn_str)?
// .create_if_missing(true)
// .connect()
// .await?;
// let _ = conn.close().await;
let pool = SqlitePoolOptions::new()
.max_connections(50)
.connect("sqlite:autotasker.db")
.connect_with(
SqliteConnectOptions::from_str(conn_str)?
.log_statements(log::LevelFilter::Trace)
)
.await
.context("could not connect to database_url")?;
sqlx::migrate!().run(&pool).await?;
Ok(pool)
}

View file

@ -24,6 +24,17 @@ enum TaskStatus {
Success,
}
#[derive(sqlx::FromRow, Deserialize, Serialize)]
#[fully_pub]
struct TaskRunSummary {
id: String,
status: TaskStatus,
trigger_mode: TriggerMode,
submitted_at: DateTime<Utc>,
started_at: Option<DateTime<Utc>>,
ended_at: Option<DateTime<Utc>>,
}
#[derive(sqlx::FromRow, Deserialize, Serialize)]
#[fully_pub]
struct TaskRun {
@ -31,11 +42,11 @@ struct TaskRun {
task_id: String,
status: TaskStatus,
trigger_mode: TriggerMode,
exit_code: u32,
exit_code: Option<u32>,
logs: String,
submitted_at: DateTime<Utc>,
started_at: DateTime<Utc>,
end_at: DateTime<Utc>,
started_at: Option<DateTime<Utc>>,
ended_at: Option<DateTime<Utc>>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]

View file

@ -1 +1,12 @@
<h1>Hello, {{ first_name }}</h1>
<h1>Hello welcome on autotasker</h1>
You main want to start in here.
<ul>
<li><a href="/tasks">List tasks</a></li>
</ul>
<p>
Autotasker is free software under <a href="https://www.gnu.org/licenses/gpl-3.0.txt">GPLv3</a> licence.
You can find source code on a <a href="https://forge.lefuturiste.fr/mbess/autotasker">self-hosted forge repository</a>.
</p>

View file

@ -0,0 +1,6 @@
<h1>List of task runs for {{ task.name }}</h1>
<ul>
{% for task_run in runs %}
<li><a href="/tasks/{{ task.id }}/runs/{{ task_run.id }}">{{ task_run.id }}</a> {{ task_run.status }}</li>
{% endfor %}
</ul>

View file

@ -1,5 +1,13 @@
<h1>Tasks</h1>
{% if tasks | length == 0 %}
No tasks were configured.
{% endif %}
<ul>
{% for task in tasks %}
<li>{{ task.name }} <a href="/tasks/{{ task.id }}/run">Run task</a></li>
<li>
{{ task.name }}
<a href="/tasks/{{ task.id }}/trigger">Trigger task</a>
<a href="/tasks/{{ task.id }}/runs">See runs</a>
</li>
{% endfor %}
</ul>

View file

@ -0,0 +1,16 @@
<h1>Task run {{ run.id }}</h1>
<ul>
<li>Id: {{ run.id }}</li>
<li>State: {{ run.status }}</li>
<li>Submitted at: {{ run.submitted_at }}</li>
<li>Started at: {{ run.started_at }}</li>
<li>Ended at: {{ run.ended_at }}</li>
<li>Trigger Mode: {{ run.trigger_mode }}</li>
</ul>
<h3>Logs</h3>
<p>Exit code is {{ run.exit_code }}</p>
<pre>
{{ run.logs }}
</pre>