tools/out-of-tree
1
0
Fork 0
Code Releases Activity

feat(daemon): timestamps

Browse Source
This commit is contained in:
dump_stack() 2024-02-27 02:00:07 +00:00
parent e633fd2e79
commit c909c2a352
Signed by: dump_stack
GPG Key ID: C9905BA72B5E02BB
7 changed files with 116 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
api/api.go
View File

@ -6,6 +6,7 @@ import (
"fmt"
"net"
"reflect"
"time"
"code.dumpstack.io/tools/out-of-tree/artifact"
"code.dumpstack.io/tools/out-of-tree/distro"
@ -57,6 +58,10 @@ type Job struct {
Artifact artifact.Artifact
Target distro.KernelInfo
Created time.Time
Started time.Time
Finished time.Time
Status Status
}
@ -77,6 +82,12 @@ type ListJobsParams struct {
// Status of the job
Status Status
// Time range (unix timestamps)
Time struct {
After int64
Before int64
}
}
type Repo struct {

11
cmd/daemon.go
View File

@ -7,6 +7,7 @@ package cmd
import (
"encoding/json"
"fmt"
"time"
"github.com/rs/zerolog/log"
@ -32,6 +33,8 @@ type DaemonJobsListCmd struct {
Repo string `help:"repo name"`
Commit string `help:"commit sha"`
Status string `help:"job status"`
After time.Time `help:"created after" format:"2006-01-02 15:04:05"`
Before time.Time `help:"created before" format:"2006-01-02 15:04:05"`
}
func (cmd *DaemonJobsListCmd) Run(dm *DaemonCmd, g *Globals) (err error) {
@ -44,6 +47,14 @@ func (cmd *DaemonJobsListCmd) Run(dm *DaemonCmd, g *Globals) (err error) {
Status: api.Status(cmd.Status),
}
if !cmd.After.IsZero() {
params.Time.After = cmd.After.Unix()
}
if !cmd.Before.IsZero() {
params.Time.Before = cmd.Before.Unix()
}
jobs, err := c.Jobs(params)
if err != nil {
log.Error().Err(err).Msg("")

15
daemon/commands.go
View File

@ -10,6 +10,7 @@ import (
"os/exec"
"path/filepath"
"sync"
"time"
"github.com/davecgh/go-spew/spew"
"github.com/google/uuid"
@ -122,6 +123,18 @@ func listJobs(req *api.Req, resp *api.Resp, e cmdenv) (err error) {
if params.Status != "" && j.Status != params.Status {
continue
}
if params.Time.After != 0 {
if time.Unix(params.Time.After, 0).
After(j.Created) {
continue
}
}
if params.Time.Before != 0 {
if time.Unix(params.Time.Before, 0).
Before(j.Created) {
continue
}
}
result = append(result, j)
}
@ -139,6 +152,8 @@ func addJob(req *api.Req, resp *api.Resp, e cmdenv) (err error) {
job.GenUUID()
job.Created = time.Now()
var repos []api.Repo
repos, err = db.Repos(e.DB)
if err != nil {

9
daemon/daemon.go
View File

@ -62,6 +62,8 @@ func (d *Daemon) Daemon() {
swg := sizedwaitgroup.New(d.Threads)
log.Info().Int("threads", d.Threads).Msg("start")
first := true
for !d.shutdown {
d.wg.Add(1)
@ -80,6 +82,11 @@ func (d *Daemon) Daemon() {
pj := newJobProcessor(job, d.db)
if first && job.Status == api.StatusRunning {
pj.SetStatus(api.StatusWaiting)
continue
}
if job.Status == api.StatusNew {
pj.SetStatus(api.StatusWaiting)
continue
@ -97,6 +104,8 @@ func (d *Daemon) Daemon() {
}(pj)
}
first = false
d.wg.Done()
time.Sleep(time.Second)
}

90
daemon/db/job.go
View File

@ -3,6 +3,7 @@ package db
import (
"database/sql"
"encoding/json"
"time"
"code.dumpstack.io/tools/out-of-tree/api"
)
@ -17,6 +18,9 @@ func createJobTable(db *sql.DB) (err error) {
"commit" TEXT,
config TEXT,
target TEXT,
created INT,
started INT,
finished INT,
status TEXT DEFAULT "new"
)`)
return
@ -24,8 +28,8 @@ func createJobTable(db *sql.DB) (err error) {
func AddJob(db *sql.DB, job *api.Job) (err error) {
stmt, err := db.Prepare(`INSERT INTO job (uuid, group_uuid, repo, "commit", ` +
`config, target) ` +
`VALUES ($1, $2, $3, $4, $5, $6);`)
`config, target, created, started, finished) ` +
`VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9);`)
if err != nil {
return
}
@ -36,8 +40,9 @@ func AddJob(db *sql.DB, job *api.Job) (err error) {
target := api.Marshal(job.Target)
res, err := stmt.Exec(job.UUID, job.Group,
job.RepoName, job.Commit,
config, target,
job.RepoName, job.Commit, config, target,
job.Created.Unix(), job.Started.Unix(),
job.Finished.Unix(),
)
if err != nil {
return
@ -48,8 +53,12 @@ func AddJob(db *sql.DB, job *api.Job) (err error) {
}
func UpdateJob(db *sql.DB, job *api.Job) (err error) {
stmt, err := db.Prepare(`UPDATE job SET uuid=$1, group_uuid=$2, repo=$3, ` +
`"commit"=$4, config=$5, target=$6, status=$7 WHERE id=$8`)
stmt, err := db.Prepare(`UPDATE job ` +
`SET uuid=$1, group_uuid=$2, repo=$3, ` +
`"commit"=$4, config=$5, target=$6, ` +
`created=$7, started=$8, finished=$9, ` +
`status=$10 ` +
`WHERE id=$11`)
if err != nil {
return
}
@ -61,31 +70,17 @@ func UpdateJob(db *sql.DB, job *api.Job) (err error) {
_, err = stmt.Exec(job.UUID, job.Group,
job.RepoName, job.Commit,
config, target,
job.Status, job.ID)
job.Created.Unix(), job.Started.Unix(),
job.Finished.Unix(), job.Status, job.ID)
return
}
func Jobs(db *sql.DB) (jobs []api.Job, err error) {
stmt, err := db.Prepare(`SELECT id, uuid, group_uuid, repo, "commit", ` +
`config, target, status FROM job`)
if err != nil {
return
}
defer stmt.Close()
rows, err := stmt.Query()
if err != nil {
return
}
defer rows.Close()
for rows.Next() {
var job api.Job
func scanJob(scan func(dest ...any) error) (job api.Job, err error) {
var config, target []byte
err = rows.Scan(&job.ID, &job.UUID, &job.Group,
&job.RepoName, &job.Commit,
&config, &target, &job.Status)
var created, started, finished int64
err = scan(&job.ID, &job.UUID, &job.Group,
&job.RepoName, &job.Commit, &config, &target,
&created, &started, &finished, &job.Status)
if err != nil {
return
}
@ -100,6 +95,34 @@ func Jobs(db *sql.DB) (jobs []api.Job, err error) {
return
}
job.Created = time.Unix(created, 0)
job.Started = time.Unix(started, 0)
job.Finished = time.Unix(finished, 0)
return
}
func Jobs(db *sql.DB) (jobs []api.Job, err error) {
stmt, err := db.Prepare(`SELECT id, uuid, group_uuid, ` +
`repo, "commit", config, target, created, ` +
`started, finished, status FROM job`)
if err != nil {
return
}
defer stmt.Close()
rows, err := stmt.Query()
if err != nil {
return
}
defer rows.Close()
for rows.Next() {
var job api.Job
job, err = scanJob(rows.Scan)
if err != nil {
return
}
jobs = append(jobs, job)
}
@ -108,22 +131,15 @@ func Jobs(db *sql.DB) (jobs []api.Job, err error) {
func Job(db *sql.DB, uuid string) (job api.Job, err error) {
stmt, err := db.Prepare(`SELECT id, uuid, group_uuid, ` +
`repo, "commit", ` +
`config, target, status ` +
`repo, "commit", config, target, ` +
`created, started, finished, status ` +
`FROM job WHERE uuid=$1`)
if err != nil {
return
}
defer stmt.Close()
err = stmt.QueryRow(uuid).Scan(&job.ID, &job.UUID,
&job.Group, &job.RepoName, &job.Commit,
&job.Artifact, &job.Target, &job.Status)
if err != nil {
return
}
return
return scanJob(stmt.QueryRow(uuid).Scan)
}
func JobStatus(db *sql.DB, uuid string) (st api.Status, err error) {

12
daemon/db/job_test.go
View File

@ -26,6 +26,8 @@ func TestJobTable(t *testing.T) {
job.Group = uuid.New().String()
job.Status = api.StatusSuccess
err = UpdateJob(db, &job)
assert.Nil(t, err)
@ -35,4 +37,14 @@ func TestJobTable(t *testing.T) {
assert.Equal(t, 1, len(jobs))
assert.Equal(t, job.Group, jobs[0].Group)
job, err = Job(db, job.UUID)
assert.Nil(t, err)
assert.Equal(t, api.StatusSuccess, job.Status)
st, err := JobStatus(db, job.UUID)
assert.Nil(t, err)
assert.Equal(t, job.Status, st)
}

4
daemon/process.go
View File

@ -7,6 +7,7 @@ import (
"os"
"os/exec"
"path/filepath"
"time"
"github.com/rs/zerolog"
"github.com/rs/zerolog/log"
@ -76,7 +77,10 @@ func (pj *jobProcessor) Process(res *Resources) (err error) {
log.Info().Msgf("process job %v", pj.job.UUID)
pj.SetStatus(api.StatusRunning)
pj.job.Started = time.Now()
defer func() {
pj.job.Finished = time.Now()
if err != nil {
pj.SetStatus(api.StatusFailure)
} else {