vieter-v
/
vieter
2
1
Fork 5
Code Issues 84 Pull Requests 4 Packages Projects Releases 12 Wiki Activity

chore: remove old cron daemon code

Jef Roosens 2023-01-13 19:03:58 +01:00
parent 2d50889e8e
commit 84e7e14a19
6 changed files with 0 additions and 492 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
src/cron/cli.v
View File

@ -1,32 +0,0 @@
module cron
import cli
import conf as vconf
struct Config {
pub:
log_level string = 'WARN'
api_key string
address string
data_dir string
base_image string = 'archlinux:base-devel'
max_concurrent_builds int = 1
api_update_frequency int = 15
image_rebuild_frequency int = 1440
// Replicates the behavior of the original cron system
global_schedule string = '0 3'
}
// cmd returns the cli module that handles the cron daemon.
pub fn cmd() cli.Command {
return cli.Command{
name: 'cron'
description: 'Start the cron service that periodically runs builds.'
execute: fn (cmd cli.Command) ! {
config_file := cmd.flags.get_string('config-file')!
conf := vconf.load<Config>(prefix: 'VIETER_', default_path: config_file)!
cron(conf)!
}
}
}

33
src/cron/cron.v
View File

@ -1,33 +0,0 @@
module cron
import log
import cron.daemon
import cron.expression
import os
const log_file_name = 'vieter.cron.log'
// cron starts a cron daemon & starts periodically scheduling builds.
pub fn cron(conf Config) ! {
// Configure logger
log_level := log.level_from_tag(conf.log_level) or {
return error('Invalid log level. The allowed values are FATAL, ERROR, WARN, INFO & DEBUG.')
}
mut logger := log.Log{
level: log_level
}
log_file := os.join_path_single(conf.data_dir, cron.log_file_name)
logger.set_full_logpath(log_file)
logger.log_to_console_too()
ce := expression.parse_expression(conf.global_schedule) or {
return error('Error while parsing global cron expression: $err.msg()')
}
mut d := daemon.init_daemon(logger, conf.address, conf.api_key, conf.base_image, ce,
conf.max_concurrent_builds, conf.api_update_frequency, conf.image_rebuild_frequency)!
d.run()
}

115
src/cron/daemon/build.v
View File

@ -1,115 +0,0 @@
module daemon
import time
import sync.stdatomic
import build
import os
const (
build_empty = 0
build_running = 1
build_done = 2
)
// clean_finished_builds removes finished builds from the build slots & returns
// them.
fn (mut d Daemon) clean_finished_builds() []ScheduledBuild {
mut out := []ScheduledBuild{}
for i in 0 .. d.atomics.len {
if stdatomic.load_u64(&d.atomics[i]) == daemon.build_done {
stdatomic.store_u64(&d.atomics[i], daemon.build_empty)
out << d.builds[i]
}
}
return out
}
// update_builds starts as many builds as possible.
fn (mut d Daemon) start_new_builds() {
now := time.now()
for d.queue.len() > 0 {
elem := d.queue.peek() or {
d.lerror("queue.peek() unexpectedly returned an error. This shouldn't happen.")
break
}
if elem.timestamp < now {
sb := d.queue.pop() or {
d.lerror("queue.pop() unexpectedly returned an error. This shouldn't happen.")
break
}
// If this build couldn't be scheduled, no more will be possible.
if !d.start_build(sb) {
d.queue.insert(sb)
break
}
} else {
break
}
}
}
// start_build starts a build for the given ScheduledBuild object.
fn (mut d Daemon) start_build(sb ScheduledBuild) bool {
for i in 0 .. d.atomics.len {
if stdatomic.load_u64(&d.atomics[i]) == daemon.build_empty {
stdatomic.store_u64(&d.atomics[i], daemon.build_running)
d.builds[i] = sb
go d.run_build(i, sb)
return true
}
}
return false
}
// run_build actually starts the build process for a given target.
fn (mut d Daemon) run_build(build_index int, sb ScheduledBuild) {
d.linfo('started build: $sb.target.url -> $sb.target.repo')
// 0 means success, 1 means failure
mut status := 0
res := build.build_target(d.client.address, d.client.api_key, d.builder_images.last(),
&sb.target, false) or {
d.ldebug('build_target error: $err.msg()')
status = 1
build.BuildResult{}
}
if status == 0 {
d.linfo('finished build: $sb.target.url -> $sb.target.repo; uploading logs...')
build_arch := os.uname().machine
d.client.add_build_log(sb.target.id, res.start_time, res.end_time, build_arch,
res.exit_code, res.logs) or {
d.lerror('Failed to upload logs for build: $sb.target.url -> $sb.target.repo')
}
} else {
d.linfo('an error occured during build: $sb.target.url -> $sb.target.repo')
}
stdatomic.store_u64(&d.atomics[build_index], daemon.build_done)
}
// current_build_count returns how many builds are currently running.
fn (mut d Daemon) current_build_count() int {
mut res := 0
for i in 0 .. d.atomics.len {
if stdatomic.load_u64(&d.atomics[i]) == daemon.build_running {
res += 1
}
}
return res
}

274
src/cron/daemon/daemon.v
View File

@ -1,274 +0,0 @@
module daemon
import time
import log
import datatypes { MinHeap }
import cron.expression { CronExpression, parse_expression }
import math
import build
import docker
import os
import client
import models { Target }
const (
// How many seconds to wait before retrying to update API if failed
api_update_retry_timeout = 5
// How many seconds to wait before retrying to rebuild image if failed
rebuild_base_image_retry_timout = 30
)
struct ScheduledBuild {
pub:
target Target
timestamp time.Time
}
// Overloaded operator for comparing ScheduledBuild objects
fn (r1 ScheduledBuild) < (r2 ScheduledBuild) bool {
return r1.timestamp < r2.timestamp
}
pub struct Daemon {
mut:
client client.Client
base_image string
builder_images []string
global_schedule CronExpression
api_update_frequency int
image_rebuild_frequency int
// Targets currently loaded from API.
targets []Target
// At what point to update the list of targets.
api_update_timestamp time.Time
image_build_timestamp time.Time
queue MinHeap<ScheduledBuild>
// Which builds are currently running
builds []ScheduledBuild
// Atomic variables used to detect when a build has finished; length is the
// same as builds
atomics []u64
logger shared log.Log
}
// init_daemon initializes a new Daemon object. It renews the targets &
// populates the build queue for the first time.
pub fn init_daemon(logger log.Log, address string, api_key string, base_image string, global_schedule CronExpression, max_concurrent_builds int, api_update_frequency int, image_rebuild_frequency int) !Daemon {
mut d := Daemon{
client: client.new(address, api_key)
base_image: base_image
global_schedule: global_schedule
api_update_frequency: api_update_frequency
image_rebuild_frequency: image_rebuild_frequency
atomics: []u64{len: max_concurrent_builds}
builds: []ScheduledBuild{len: max_concurrent_builds}
logger: logger
}
// Initialize the targets & queue
d.renew_targets()
d.renew_queue()
if !d.rebuild_base_image() {
return error('The base image failed to build. The Vieter cron daemon cannot run without an initial builder image.')
}
return d
}
// run starts the actual daemon process. It runs builds when possible &
// periodically refreshes the list of targets to ensure we stay in sync.
pub fn (mut d Daemon) run() {
for {
finished_builds := d.clean_finished_builds()
// Update the API's contents if needed & renew the queue
if time.now() >= d.api_update_timestamp {
d.renew_targets()
d.renew_queue()
}
// The finished builds should only be rescheduled if the API contents
// haven't been renewed.
else {
for sb in finished_builds {
d.schedule_build(sb.target)
}
}
// TODO remove old builder images.
// This issue is less trivial than it sounds, because a build could
// still be running when the image has to be rebuilt. That would
// prevent the image from being removed. Therefore, we will need to
// keep track of a list or something & remove an image once we have
// made sure it isn't being used anymore.
if time.now() >= d.image_build_timestamp {
d.rebuild_base_image()
// In theory, executing this function here allows an old builder
// image to exist for at most image_rebuild_frequency minutes.
d.clean_old_base_images()
}
// Schedules new builds when possible
d.start_new_builds()
// If there are builds currently running, the daemon should refresh
// every second to clean up any finished builds & start new ones.
mut delay := time.Duration(1 * time.second)
// Sleep either until we have to refresh the targets or when the next
// build has to start, with a minimum of 1 second.
if d.current_build_count() == 0 {
now := time.now()
delay = d.api_update_timestamp - now
if d.queue.len() > 0 {
elem := d.queue.peek() or {
d.lerror("queue.peek() unexpectedly returned an error. This shouldn't happen.")
// This is just a fallback option. In theory, queue.peek()
// should *never* return an error or none, because we check
// its len beforehand.
time.sleep(1)
continue
}
time_until_next_job := elem.timestamp - now
delay = math.min(delay, time_until_next_job)
}
}
// We sleep for at least one second. This is to prevent the program
// from looping agressively when a cronjob can be scheduled, but
// there's no spots free for it to be started.
delay = math.max(delay, 1 * time.second)
d.ldebug('Sleeping for ${delay}...')
time.sleep(delay)
}
}
// schedule_build adds the next occurence of the given targets build to the
// queue.
fn (mut d Daemon) schedule_build(target Target) {
ce := if target.schedule != '' {
parse_expression(target.schedule) or {
// TODO This shouldn't return an error if the expression is empty.
d.lerror("Error while parsing cron expression '$target.schedule' (id $target.id): $err.msg()")
d.global_schedule
}
} else {
d.global_schedule
}
// A target that can't be scheduled will just be skipped for now
timestamp := ce.next_from_now() or {
d.lerror("Couldn't calculate next timestamp from '$target.schedule'; skipping")
return
}
d.queue.insert(ScheduledBuild{
target: target
timestamp: timestamp
})
}
// renew_targets requests the newest list of targets from the server & replaces
// the old one.
fn (mut d Daemon) renew_targets() {
d.linfo('Renewing targets...')
mut new_targets := d.client.get_all_targets() or {
d.lerror('Failed to renew targets. Retrying in ${daemon.api_update_retry_timeout}s...')
d.api_update_timestamp = time.now().add_seconds(daemon.api_update_retry_timeout)
return
}
// Filter out any targets that shouldn't run on this architecture
cur_arch := os.uname().machine
new_targets = new_targets.filter(it.arch.any(it.value == cur_arch))
d.targets = new_targets
d.api_update_timestamp = time.now().add_seconds(60 * d.api_update_frequency)
}
// renew_queue replaces the old queue with a new one that reflects the newest
// values in targets.
fn (mut d Daemon) renew_queue() {
d.linfo('Renewing queue...')
mut new_queue := MinHeap<ScheduledBuild>{}
// Move any jobs that should have already started from the old queue onto
// the new one
now := time.now()
// For some reason, using
// ```v
// for d.queue.len() > 0 && d.queue.peek() !.timestamp < now {
//```
// here causes the function to prematurely just exit, without any errors or anything, very weird
// https://github.com/vlang/v/issues/14042
for d.queue.len() > 0 {
elem := d.queue.pop() or {
d.lerror("queue.pop() returned an error. This shouldn't happen.")
continue
}
if elem.timestamp < now {
new_queue.insert(elem)
} else {
break
}
}
d.queue = new_queue
// For each target in targets, parse their cron expression (or use the
// default one if not present) & add them to the queue
for target in d.targets {
d.schedule_build(target)
}
}
// rebuild_base_image recreates the builder image.
fn (mut d Daemon) rebuild_base_image() bool {
d.linfo('Rebuilding builder image....')
d.builder_images << build.create_build_image(d.base_image) or {
d.lerror('Failed to rebuild base image. Retrying in ${daemon.rebuild_base_image_retry_timout}s...')
d.image_build_timestamp = time.now().add_seconds(daemon.rebuild_base_image_retry_timout)
return false
}
d.image_build_timestamp = time.now().add_seconds(60 * d.image_rebuild_frequency)
return true
}
// clean_old_base_images tries to remove any old but still present builder
// images.
fn (mut d Daemon) clean_old_base_images() {
mut i := 0
mut dd := docker.new_conn() or {
d.lerror('Failed to connect to Docker socket.')
return
}
defer {
dd.close() or {}
}
for i < d.builder_images.len - 1 {
// For each builder image, we try to remove it by calling the Docker
// API. If the function returns an error or false, that means the image
// wasn't deleted. Therefore, we move the index over. If the function
// returns true, the array's length has decreased by one so we don't
// move the index.
dd.image_remove(d.builder_images[i]) or { i += 1 }
}
}

36
src/cron/daemon/log.v
View File

@ -1,36 +0,0 @@
module daemon
// lfatal create a log message with the fatal level
pub fn (mut d Daemon) lfatal(msg string) {
lock d.logger {
d.logger.fatal(msg)
}
}
// lerror create a log message with the error level
pub fn (mut d Daemon) lerror(msg string) {
lock d.logger {
d.logger.error(msg)
}
}
// lwarn create a log message with the warn level
pub fn (mut d Daemon) lwarn(msg string) {
lock d.logger {
d.logger.warn(msg)
}
}
// linfo create a log message with the info level
pub fn (mut d Daemon) linfo(msg string) {
lock d.logger {
d.logger.info(msg)
}
}
// ldebug create a log message with the debug level
pub fn (mut d Daemon) ldebug(msg string) {
lock d.logger {
d.logger.debug(msg)
}
}

2
src/main.v
View File

@ -9,7 +9,6 @@ import console.schedule
import console.man import console.man
import console.aur import console.aur
import console.repos import console.repos
import cron
import agent import agent
fn main() { fn main() {
@ -43,7 +42,6 @@ fn main() {
commands: [ commands: [
server.cmd(), server.cmd(),
targets.cmd(), targets.cmd(),
cron.cmd(),
logs.cmd(), logs.cmd(),
schedule.cmd(), schedule.cmd(),
man.cmd(), man.cmd(),