Attempt at writing renew_queue function; seems to just stop in the

middle
cron
Jef Roosens 2022-04-13 16:12:22 +02:00
parent ff57d73998
commit f7e1aba30b
Signed by untrusted user: Jef Roosens
GPG Key ID: B75D4F293C7052DB
10 changed files with 89 additions and 30 deletions

View File

@ -5,12 +5,13 @@ import time
import log import log
import util import util
import cron.daemon import cron.daemon
import cron.expression
// cron starts a cron daemon & starts periodically scheduling builds. // cron starts a cron daemon & starts periodically scheduling builds.
pub fn cron(conf Config) ? { pub fn cron(conf Config) ? {
// Configure logger // Configure logger
log_level := log.level_from_tag(conf.log_level) or { log_level := log.level_from_tag(conf.log_level) or {
util.exit_with_message(1, 'Invalid log level. The allowed values are FATAL, ERROR, WARN, INFO & DEBUG.') return error('Invalid log level. The allowed values are FATAL, ERROR, WARN, INFO & DEBUG.')
} }
mut logger := log.Log{ mut logger := log.Log{
@ -20,5 +21,12 @@ pub fn cron(conf Config) ? {
logger.set_full_logpath(conf.log_file) logger.set_full_logpath(conf.log_file)
logger.log_to_console_too() logger.log_to_console_too()
d := daemon.init(conf) 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) ?
d.run() ?
} }

View File

@ -3,9 +3,12 @@ module daemon
import git import git
import time import time
import log import log
import datatypes import datatypes { MinHeap }
import cron.expression { CronExpression, parse_expression }
struct ScheduledBuild { struct ScheduledBuild {
pub:
repo_id string
repo git.GitRepo repo git.GitRepo
timestamp time.Time timestamp time.Time
} }
@ -16,12 +19,16 @@ fn (r1 ScheduledBuild) < (r2 ScheduledBuild) bool {
pub struct Daemon { pub struct Daemon {
mut: mut:
conf Config address string
api_key string
base_image string
global_schedule CronExpression
api_update_frequency int
// Repos currently loaded from API. // Repos currently loaded from API.
repos_map map[string]git.GitRepo repos_map map[string]git.GitRepo
// At what point to update the list of repositories. // At what point to update the list of repositories.
api_update_timestamp time.Time api_update_timestamp time.Time
queue datatypes.MinHeap<ScheduledBuild> queue MinHeap<ScheduledBuild>
// Which builds are currently running // Which builds are currently running
builds []git.GitRepo builds []git.GitRepo
// Atomic variables used to detect when a build has finished; length is the // Atomic variables used to detect when a build has finished; length is the
@ -30,25 +37,66 @@ mut:
logger shared log.Log logger shared log.Log
} }
// init 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) ?Daemon {
pub fn init(conf Config) Daemon { mut d := Daemon{
return Daemon{ address: address
conf: conf api_key: api_key
atomics: [conf.max_concurrent_builds]u64{} base_image: base_image
} global_schedule: global_schedule
api_update_frequency: api_update_frequency
atomics: []u64{len: max_concurrent_builds}
builds: []git.GitRepo{len: max_concurrent_builds}
logger: logger
} }
fn (mut d Daemon) run() ? { // Initialize the repos & queue
d.renew_repos() ? d.renew_repos() ?
d.renew_queue() ? d.renew_queue() ?
return d
}
pub fn (mut d Daemon) run() ? {
println(d.queue)
} }
fn (mut d Daemon) renew_repos() ? { fn (mut d Daemon) renew_repos() ? {
mut new_repos := git.get_repos(d.conf.address, d.conf.api_key) ? mut new_repos := git.get_repos(d.address, d.api_key) ?
d.repos_map = new_repos.move() d.repos_map = new_repos.move()
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 repos_map.
fn (mut d Daemon) renew_queue() ? { fn (mut d Daemon) renew_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 d.queue.len() > 0 && d.queue.peek() ?.timestamp < now {
new_queue.insert(d.queue.pop() ?)
}
println('hey')
println(d.repos_map)
// For each repository in repos_map, parse their cron expression (or use
// the default one if not present) & add them to the queue
for id, repo in d.repos_map {
println('hey')
ce := parse_expression(repo.schedule) or { d.global_schedule }
// A repo that can't be scheduled will just be skipped for now
timestamp := ce.next(now) or { continue }
new_queue.insert(ScheduledBuild{
repo_id: id
repo: repo
timestamp: timestamp
})
}
d.queue = new_queue
} }

View File

@ -1,8 +1,8 @@
module cron module expression
import time import time
struct CronExpression { pub struct CronExpression {
minutes []int minutes []int
hours []int hours []int
days []int days []int
@ -219,7 +219,7 @@ fn parse_part(s string, min int, max int) ?[]int {
// parse_expression parses an entire cron expression string into a // parse_expression parses an entire cron expression string into a
// CronExpression object, if possible. // CronExpression object, if possible.
fn parse_expression(exp string) ?CronExpression { pub fn parse_expression(exp string) ?CronExpression {
// The filter allows for multiple spaces between parts // The filter allows for multiple spaces between parts
mut parts := exp.split(' ').filter(it != '') mut parts := exp.split(' ').filter(it != '')

View File

@ -1,4 +1,4 @@
module cron module expression
// parse_range_error returns the returned error message. If the result is '', // parse_range_error returns the returned error message. If the result is '',
// that means the function didn't error. // that means the function didn't error.

View File

@ -1,4 +1,4 @@
module cron module expression
import time { parse } import time { parse }

View File

@ -15,7 +15,7 @@ pub mut:
// Which repo the builder should publish packages to // Which repo the builder should publish packages to
repo string repo string
// Cron schedule describing how frequently to build the repo. // Cron schedule describing how frequently to build the repo.
schedule string schedule string [optional]
} }
// patch_from_params patches a GitRepo from a map[string]string, usually // patch_from_params patches a GitRepo from a map[string]string, usually
@ -74,7 +74,7 @@ pub fn repo_from_params(params map[string]string) ?GitRepo {
// If we're creating a new GitRepo, we want all fields to be present before // If we're creating a new GitRepo, we want all fields to be present before
// "patching". // "patching".
$for field in GitRepo.fields { $for field in GitRepo.fields {
if field.name !in params { if field.name !in params && !field.attrs.contains('optional') {
return error('Missing parameter: ${field.name}.') return error('Missing parameter: ${field.name}.')
} }
} }

0
src/v.mod 100644
View File

View File

@ -8,3 +8,6 @@ repos_file = "data/repos.json"
default_arch = "x86_64" default_arch = "x86_64"
address = "http://localhost:8000" address = "http://localhost:8000"
global_schedule = '0 3'