diff --git a/.gitignore b/.gitignore index d58fe10f3..c50d269cb 100644 --- a/.gitignore +++ b/.gitignore @@ -114,3 +114,6 @@ tumlive.service # osx .DS_Store + +# downloaded protoc binary +protoc-*.zip diff --git a/cmd/tumlive/main.go b/cmd/tumlive/main.go index 17678fa7e..ef7a18113 100644 --- a/cmd/tumlive/main.go +++ b/cmd/tumlive/main.go @@ -113,6 +113,8 @@ func run(ctx context.Context) error { &model.TranscodingFailure{}, &model.Email{}, &model.Runner{}, + &model.Action{}, + &model.Job{}, ) if err != nil { return fmt.Errorf("migration: %w", err) diff --git a/dao/dao_base.go b/dao/dao_base.go index 6f1888853..4bf6752d0 100644 --- a/dao/dao_base.go +++ b/dao/dao_base.go @@ -37,6 +37,7 @@ type DaoWrapper struct { TranscodingFailureDao EmailDao RunnerDao RunnerDao + JobDao JobDao } func NewDaoWrapper() DaoWrapper { @@ -65,5 +66,6 @@ func NewDaoWrapper() DaoWrapper { TranscodingFailureDao: NewTranscodingFailureDao(), EmailDao: NewEmailDao(), RunnerDao: NewRunnerDao(), + JobDao: NewJobDao(), } } diff --git a/dao/job.go b/dao/job.go new file mode 100644 index 000000000..ece02d02b --- /dev/null +++ b/dao/job.go @@ -0,0 +1,135 @@ +package dao + +import ( + "context" + + "gorm.io/gorm" + + "github.com/TUM-Dev/gocast/model" +) + +//go:generate go tool mockgen -source=job.go -destination ../mock_dao/job.go + +// JobDao interface defines methods for job data access +type JobDao interface { + // Create creates a new job in the database + Create(context.Context, *model.Job) error + + // Update updates an existing job in the database + Update(context.Context, *model.Job) error + + // GetByJobID retrieves a job by its unique job ID + GetByJobID(context.Context, string) (model.Job, error) + + // GetByStreamID retrieves all jobs for a given stream ID + GetByStreamID(context.Context, uint) ([]model.Job, error) + + // GetByRunnerHostname retrieves all active jobs for a given runner + GetByRunnerHostname(context.Context, string) ([]model.Job, error) + + // GetActiveJobs retrieves all active jobs (created or running) + GetActiveJobs(context.Context) ([]model.Job, error) + + // Delete deletes a job by its job ID + Delete(context.Context, string) error + + // DeleteByStreamID deletes all jobs for a given stream ID + DeleteByStreamID(context.Context, uint) error + + // CreateAction creates a new action for a job + CreateAction(context.Context, *model.Action) error + + // UpdateAction updates an existing action + UpdateAction(context.Context, *model.Action) error + + // GetActionByJobIDAndType retrieves an action by job ID and action type + GetActionByJobIDAndType(context.Context, uint, string) (model.Action, error) + + // GetActionsByJobID retrieves all actions for a given job ID + GetActionsByJobID(context.Context, uint) ([]model.Action, error) +} + +type jobDao struct { + db *gorm.DB +} + +// NewJobDao creates a new JobDao instance +func NewJobDao() JobDao { + return jobDao{db: DB} +} + +// Create creates a new job in the database +func (d jobDao) Create(ctx context.Context, job *model.Job) error { + return d.db.WithContext(ctx).Create(job).Error +} + +// Update updates an existing job in the database +func (d jobDao) Update(ctx context.Context, job *model.Job) error { + return d.db.WithContext(ctx).Save(job).Error +} + +// GetByJobID retrieves a job by its unique job ID +func (d jobDao) GetByJobID(ctx context.Context, jobID string) (model.Job, error) { + var job model.Job + err := d.db.WithContext(ctx).Where("job_id = ?", jobID).First(&job).Error + return job, err +} + +// GetByStreamID retrieves all jobs for a given stream ID +func (d jobDao) GetByStreamID(ctx context.Context, streamID uint) ([]model.Job, error) { + var jobs []model.Job + err := d.db.WithContext(ctx).Where("stream_id = ?", streamID).Find(&jobs).Error + return jobs, err +} + +// GetByRunnerHostname retrieves all active jobs for a given runner +func (d jobDao) GetByRunnerHostname(ctx context.Context, hostname string) ([]model.Job, error) { + var jobs []model.Job + err := d.db.WithContext(ctx). + Where("runner_hostname = ? AND status IN ?", hostname, []model.WorkState{model.WorkStateCreated, model.WorkStateRunning}). + Find(&jobs).Error + return jobs, err +} + +// GetActiveJobs retrieves all active jobs (created or running) +func (d jobDao) GetActiveJobs(ctx context.Context) ([]model.Job, error) { + var jobs []model.Job + err := d.db.WithContext(ctx). + Where("status IN ?", []model.WorkState{model.WorkStateCreated, model.WorkStateRunning}). + Find(&jobs).Error + return jobs, err +} + +// Delete deletes a job by its job ID +func (d jobDao) Delete(ctx context.Context, jobID string) error { + return d.db.WithContext(ctx).Where("job_id = ?", jobID).Delete(&model.Job{}).Error +} + +// DeleteByStreamID deletes all jobs for a given stream ID +func (d jobDao) DeleteByStreamID(ctx context.Context, streamID uint) error { + return d.db.WithContext(ctx).Where("stream_id = ?", streamID).Delete(&model.Job{}).Error +} + +// CreateAction creates a new action for a job +func (d jobDao) CreateAction(ctx context.Context, action *model.Action) error { + return d.db.WithContext(ctx).Create(action).Error +} + +// UpdateAction updates an existing action +func (d jobDao) UpdateAction(ctx context.Context, action *model.Action) error { + return d.db.WithContext(ctx).Save(action).Error +} + +// GetActionByJobIDAndType retrieves an action by job ID and action type +func (d jobDao) GetActionByJobIDAndType(ctx context.Context, jobID uint, actionType string) (model.Action, error) { + var action model.Action + err := d.db.WithContext(ctx).Where("job_id = ? AND action_type = ?", jobID, actionType).First(&action).Error + return action, err +} + +// GetActionsByJobID retrieves all actions for a given job ID +func (d jobDao) GetActionsByJobID(ctx context.Context, jobID uint) ([]model.Action, error) { + var actions []model.Action + err := d.db.WithContext(ctx).Where("job_id = ?", jobID).Find(&actions).Error + return actions, err +} diff --git a/dao/runner.go b/dao/runner.go index 93af304ad..c9755bef5 100644 --- a/dao/runner.go +++ b/dao/runner.go @@ -27,6 +27,9 @@ type RunnerDao interface { // GetAll gets a list of all Runners. GetAll(context.Context) ([]model.Runner, error) + // GetAllWithJobs gets a list of all Runners with their active jobs preloaded. + GetAllWithJobs(context.Context) ([]model.Runner, error) + // ReserveRunner returns the runner that currently runs the least jobs and is not draining. // It also increments the number of jobs assigned to the runner. ReserveRunner(context.Context) (model.Runner, error) @@ -100,3 +103,12 @@ func (d runnerDao) GetAll(c context.Context) ([]model.Runner, error) { err := d.db.WithContext(c).Find(&runners).Error return runners, err } + +// GetAllWithJobs returns all Runners with their active jobs preloaded +func (d runnerDao) GetAllWithJobs(c context.Context) ([]model.Runner, error) { + var runners []model.Runner + err := d.db.WithContext(c). + Preload("Jobs", "status IN ?", []model.WorkState{model.WorkStateCreated, model.WorkStateRunning}). + Find(&runners).Error + return runners, err +} diff --git a/go.mod b/go.mod index 692d02807..7cae1b9e5 100644 --- a/go.mod +++ b/go.mod @@ -19,7 +19,6 @@ require ( github.com/gorilla/websocket v1.5.3 // indirect github.com/jinzhu/now v1.1.5 github.com/microcosm-cc/bluemonday v1.0.27 - github.com/pkg/profile v1.7.0 github.com/robfig/cron/v3 v3.0.1 github.com/russross/blackfriday/v2 v2.1.0 github.com/satori/go.uuid v1.2.0 @@ -70,12 +69,10 @@ require ( github.com/bytedance/sonic/loader v0.2.2 // indirect github.com/cloudwego/base64x v0.1.4 // indirect github.com/dgraph-io/ristretto v0.1.0 // indirect - github.com/felixge/fgprof v0.9.5 // indirect github.com/gabriel-vasile/mimetype v1.4.8 // indirect github.com/go-viper/mapstructure/v2 v2.4.0 // indirect github.com/golang-jwt/jwt/v4 v4.5.1 // indirect github.com/golang/glog v1.2.4 // indirect - github.com/google/pprof v0.0.0-20241210010833-40e02aabc2ad // indirect github.com/josharian/intern v1.0.0 // indirect github.com/klauspost/cpuid/v2 v2.2.9 // indirect github.com/mailru/easyjson v0.9.0 // indirect diff --git a/go.sum b/go.sum index 635134c5d..acf0912ed 100644 --- a/go.sum +++ b/go.sum @@ -10,8 +10,6 @@ github.com/Masterminds/semver/v3 v3.3.1 h1:QtNSWtVZ3nBfk8mAOu/B6v7FMJ+NHTIgUPi7r github.com/Masterminds/semver/v3 v3.3.1/go.mod h1:4V+yj/TJE1HU9XfppCwVMZq3I84lprf4nC11bSS5beM= github.com/Masterminds/sprig/v3 v3.3.0 h1:mQh0Yrg1XPo6vjYXgtf5OtijNAKJRNcTdOOGZe3tPhs= github.com/Masterminds/sprig/v3 v3.3.0/go.mod h1:Zy1iXRYNqNLUolqCpL4uhk6SHUMAOSCzdgBfDb35Lz0= -github.com/RBG-TUM/CAMPUSOnline v0.0.0-20230412070523-8db58ed5c0b4 h1:t2Po3FzeYWa8fO2GdacQtiLDUvGYlJ+ukGqe7gp6/ro= -github.com/RBG-TUM/CAMPUSOnline v0.0.0-20230412070523-8db58ed5c0b4/go.mod h1:pcWaw3hQOKmvRfldxtPlYBGAFbWQXUU44+KcDYwqduc= github.com/RBG-TUM/CAMPUSOnline v0.0.0-20251116171131-fe44da0a604a h1:d699gVqCIpup0WFK3wKB4uG0qm+dZwHViBW7VI5nPAw= github.com/RBG-TUM/CAMPUSOnline v0.0.0-20251116171131-fe44da0a604a/go.mod h1:pcWaw3hQOKmvRfldxtPlYBGAFbWQXUU44+KcDYwqduc= github.com/RBG-TUM/commons v0.0.0-20220406105618-030c095f6a1b h1:87IltMZV7hN5jGCHbqDIEf0p8+X0f4Ep40PRmz1Mexg= @@ -52,15 +50,6 @@ github.com/bytedance/sonic/loader v0.2.2/go.mod h1:N8A3vUdtUebEY2/VQC0MyhYeKUFos github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= -github.com/chromedp/cdproto v0.0.0-20230802225258-3cf4e6d46a89/go.mod h1:GKljq0VrfU4D5yc+2qA6OVr8pmO/MBbPEWqWQ/oqGEs= -github.com/chromedp/chromedp v0.9.2/go.mod h1:LkSXJKONWTCHAfQasKFUZI+mxqS4tZqhmtGzzhLsnLs= -github.com/chromedp/sysutil v1.0.0/go.mod h1:kgWmDdq8fTzXYcKIBqIYvRRTnYb9aNS9moAV0xufSww= -github.com/chzyer/logex v1.1.10/go.mod h1:+Ywpsq7O8HXn0nuIou7OrIPyXbp3wmkHB+jjWRnGsAI= -github.com/chzyer/logex v1.2.1/go.mod h1:JLbx6lG2kDbNRFnfkgvh4eRJRPX1QCoOIWomwysCBrQ= -github.com/chzyer/readline v0.0.0-20180603132655-2972be24d48e/go.mod h1:nSuG5e5PlCu98SY8svDHJxuZscDgtXS6KTTbou5AhLI= -github.com/chzyer/readline v1.5.1/go.mod h1:Eh+b79XXUwfKfcPLepksvw2tcLE/Ct21YObkaSkeBlk= -github.com/chzyer/test v0.0.0-20180213035817-a1ea475d72b1/go.mod h1:Q3SI9o4m/ZMnBNeIyt5eFwwo7qiLfzFZmjNmxjkiQlU= -github.com/chzyer/test v1.0.0/go.mod h1:2JlltgoNkt4TW/z9V/IzDdFaMTM2JPIi26O1pF38GC8= github.com/cloudwego/base64x v0.1.4 h1:jwCgWpFanWmN8xoIUHa2rtzmkd5J2plF/dnLS6Xd/0Y= github.com/cloudwego/base64x v0.1.4/go.mod h1:0zlkT4Wn5C6NdauXdJRhSKRlJvmclQ1hhJgA0rcu/8w= github.com/cloudwego/iasm v0.2.0/go.mod h1:8rXZaNYT2n95jn+zTI1sDr+IgcD2GVs0nlbbQPiEFhY= @@ -83,9 +72,6 @@ github.com/dgryski/go-farm v0.0.0-20200201041132-a6ae2369ad13/go.mod h1:SqUrOPUn github.com/dustin/go-humanize v1.0.0/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk= github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY= github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto= -github.com/felixge/fgprof v0.9.3/go.mod h1:RdbpDgzqYVh/T9fPELJyV7EYJuHB55UTEULNun8eiPw= -github.com/felixge/fgprof v0.9.5 h1:8+vR6yu2vvSKn08urWyEuxx75NWPEvybbkBirEpsbVY= -github.com/felixge/fgprof v0.9.5/go.mod h1:yKl+ERSa++RYOs32d8K6WEXCB4uXdLls4ZaZPpayhMM= github.com/frankban/quicktest v1.14.6 h1:7Xjx+VpznH+oBnejlPUj8oUpdxnVs4f8XU8WnHkI4W8= github.com/frankban/quicktest v1.14.6/go.mod h1:4ptaffx2x8+WTWXmUCuVU6aPUX1/Mz7zb5vbUoiM6w0= github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo= @@ -131,9 +117,6 @@ github.com/go-sql-driver/mysql v1.9.1 h1:FrjNGn/BsJQjVRuSa8CBrM5BWA9BWoXXat3KrtS github.com/go-sql-driver/mysql v1.9.1/go.mod h1:qn46aNg1333BRMNU69Lq93t8du/dwxI64Gl8i5p1WMU= github.com/go-viper/mapstructure/v2 v2.4.0 h1:EBsztssimR/CONLSZZ04E8qAkxNYq4Qp9LvH92wZUgs= github.com/go-viper/mapstructure/v2 v2.4.0/go.mod h1:oJDH3BJKyqBA2TXFhDsKDGDTlndYOZ6rGS0BRZIxGhM= -github.com/gobwas/httphead v0.1.0/go.mod h1:O/RXo79gxV8G+RqlR/otEwx4Q36zl9rqC5u12GKvMCM= -github.com/gobwas/pool v0.2.1/go.mod h1:q8bcK0KcYlCgd9e7WYLm9LpyS+YeLd8JVDW6WezmKEw= -github.com/gobwas/ws v1.2.1/go.mod h1:hRKAFb8wOxFROYNsT1bqfWnhX+b5MFeJM9r2ZSwg/KY= github.com/goccy/go-json v0.10.4 h1:JSwxQzIqKfmFX1swYPpUThQZp/Ka4wzJdK0LWVytLPM= github.com/goccy/go-json v0.10.4/go.mod h1:oq7eo15ShAhp70Anwd5lgX2pLfOS3QCiwU/PULtXL6M= github.com/gogo/protobuf v1.3.1/go.mod h1:SlYgWuQ5SjCEi6WLHjHCa1yvBfUnHcTbrrZtXPKa29o= @@ -153,10 +136,6 @@ github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeN github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg= -github.com/google/pprof v0.0.0-20211214055906-6f57359322fd/go.mod h1:KgnwoLYCZ8IQu3XUZ8Nc/bM9CCZFOyjUNOSygVozoDg= -github.com/google/pprof v0.0.0-20240227163752-401108e1b7e7/go.mod h1:czg5+yv1E0ZGTi6S6vVK1mke0fV+FaUhNGcd6VRS9Ik= -github.com/google/pprof v0.0.0-20241210010833-40e02aabc2ad h1:a6HEuzUHeKH6hwfN/ZoQgRgVIWFJljSWa/zetS2WTvg= -github.com/google/pprof v0.0.0-20241210010833-40e02aabc2ad/go.mod h1:vavhavw2zAxS5dIdcRluK6cSGGPlZynqzFM8NdvU144= github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= @@ -175,8 +154,6 @@ github.com/hashicorp/go-uuid v1.0.3/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/b github.com/hashicorp/golang-lru v0.5.4/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4= github.com/huandu/xstrings v1.5.0 h1:2ag3IFq9ZDANvthTwTiqSSZLjDc+BedvHPAp5tJy2TI= github.com/huandu/xstrings v1.5.0/go.mod h1:y5/lhBue+AyNmUVz9RLU9xbLR0o4KIIExikq4ovT0aE= -github.com/ianlancetaylor/demangle v0.0.0-20210905161508-09a460cdf81d/go.mod h1:aYm2/VgdVmcIU8iMfdMvDMsRAQjcfZSKFby6HOFvi/w= -github.com/ianlancetaylor/demangle v0.0.0-20230524184225-eabc099b10ab/go.mod h1:gx7rwoVhcfuVKG5uya9Hs3Sxj7EIvldVofAWIUtGouw= github.com/icholy/digest v1.1.0 h1:HfGg9Irj7i+IX1o1QAmPfIBNu/Q5A5Tu3n/MED9k9H4= github.com/icholy/digest v1.1.0/go.mod h1:QNrsSGQ5v7v9cReDI0+eyjsXGUoRSUZQHeQ5C4XLa0Y= github.com/jcmturner/aescts/v2 v2.0.0 h1:9YKLH6ey7H4eDBXW8khjYslgyqG2xZikXP0EQFKrle8= @@ -218,10 +195,8 @@ github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= -github.com/ledongthuc/pdf v0.0.0-20220302134840-0c2507a12d80/go.mod h1:imJHygn/1yfhB7XSJJKlFZKl/J+dCPAknuiaGOshXAs= github.com/leodido/go-urn v1.4.0 h1:WT9HwE9SGECu3lg4d/dIA+jxlljEa1/ffXKmRjqdmIQ= github.com/leodido/go-urn v1.4.0/go.mod h1:bvxc+MVxLKB4z00jd1z+Dvzr47oO32F/QSNjSBOlFxI= -github.com/mailru/easyjson v0.7.7/go.mod h1:xzfreul335JAWq5oZzymOObrkdz5UnU4kGfJJLY9Nlc= github.com/mailru/easyjson v0.9.0 h1:PrnmzHw7262yW8sTBwxi1PdJA3Iw/EKBa8psRf7d9a4= github.com/mailru/easyjson v0.9.0/go.mod h1:1+xMtQp2MRNVL/V1bOzuP3aP8VNwRW55fQUto+XFtTU= github.com/mattermost/xml-roundtrip-validator v0.1.0 h1:RXbVD2UAl7A7nOTR4u7E3ILa4IbtvKBHw64LDsmu9hU= @@ -248,7 +223,6 @@ github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjY github.com/olahol/melody v0.0.0-20180227134253-7bd65910e5ab/go.mod h1:3lo03f1jM3KFUG/rsujuLB1rBmlvIzVM3SCqbuHqsBU= github.com/orandin/slog-gorm v1.4.0 h1:FgA8hJufF9/jeNSYoEXmHPPBwET2gwlF3B85JdpsTUU= github.com/orandin/slog-gorm v1.4.0/go.mod h1:MoZ51+b7xE9lwGNPYEhxcUtRNrYzjdcKvA8QXQQGEPA= -github.com/orisano/pixelmatch v0.0.0-20220722002657-fb0b55479cde/go.mod h1:nZgzbfBr3hhjoZnS66nKrHmduYNpc34ny7RK4z5/HM0= github.com/panjf2000/ants/v2 v2.4.2/go.mod h1:f6F0NZVFsGCp5A7QW/Zj/m92atWwOkY0OIhFxRNFr4A= github.com/pelletier/go-toml/v2 v2.2.3 h1:YmeHyLY8mFWbdkNWwpr+qIL2bEqT0o95WSdkNHvL12M= github.com/pelletier/go-toml/v2 v2.2.3/go.mod h1:MfCQTFTvCcUyyvvwm1+G6H/jORL20Xlb6rzQu9GuUkc= @@ -259,8 +233,6 @@ github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINE github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/pkg/profile v1.4.0/go.mod h1:NWz/XGvpEW1FyYQ7fCx4dqYBLlfTcE+A9FLAkNKqjFE= -github.com/pkg/profile v1.7.0 h1:hnbDkaNWPCLMO9wGLdBFTIZvzDrDfBM2072E1S9gJkA= -github.com/pkg/profile v1.7.0/go.mod h1:8Uer0jas47ZQMJ7VD+OHknK4YDY07LPUC6dEvqDjvNo= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U= github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= @@ -353,8 +325,6 @@ golang.org/x/crypto v0.13.0/go.mod h1:y6Z2r+Rw4iayiXXAIxJIDAJ1zMW4yaTpebo8fPOliY golang.org/x/crypto v0.19.0/go.mod h1:Iy9bg/ha4yyC70EfRS8jz+B6ybOBKMaSxLj6P6oBDfU= golang.org/x/crypto v0.23.0/go.mod h1:CKFgDieR+mRhux2Lsu27y0fO304Db0wZe70UKqHu0v8= golang.org/x/crypto v0.31.0/go.mod h1:kDsLvtWBEx7MV9tJOj9bnXsPbxwJQ6csT/x4KIN4Ssk= -golang.org/x/crypto v0.43.0 h1:dduJYIi3A3KOfdGOHX8AVZ/jGiyPa3IbBozJ5kNuE04= -golang.org/x/crypto v0.43.0/go.mod h1:BFbav4mRNlXJL4wNeejLpWxB7wMbc79PdRGhWKncxR0= golang.org/x/crypto v0.45.0 h1:jMBrvKuj23MTlT0bQEOBcAE0mjg8mK9RXFhRH6nyF3Q= golang.org/x/crypto v0.45.0/go.mod h1:XTGrrkGJve7CYK7J8PEww4aY7gM3qMCElcJQ8n8JdX4= golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4= @@ -362,8 +332,7 @@ golang.org/x/mod v0.8.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs= golang.org/x/mod v0.12.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs= golang.org/x/mod v0.15.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c= golang.org/x/mod v0.17.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c= -golang.org/x/mod v0.28.0 h1:gQBtGhjxykdjY9YhZpSlZIsbnaE2+PgjfLWUQTnoZ1U= -golang.org/x/mod v0.28.0/go.mod h1:yfB/L0NOf/kmEbXjzCPOx1iK1fRutOydrCMsqRhEBxI= +golang.org/x/mod v0.29.0 h1:HV8lRxZC4l2cr3Zq1LvtOsi/ThTgWnUk/y64QSs8GwA= golang.org/x/mod v0.29.0/go.mod h1:NyhrlYXJ2H4eJiRy/WDBO6HMqZQ6q9nk4JzS3NuCK+w= golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= @@ -379,8 +348,6 @@ golang.org/x/net v0.15.0/go.mod h1:idbUs1IY1+zTqbi8yxTbhexhEEk5ur9LInksu6HrEpk= golang.org/x/net v0.21.0/go.mod h1:bIjVDfnllIU7BJ2DNgfnXvpSvtn8VRwhlsaeUTyUS44= golang.org/x/net v0.25.0/go.mod h1:JkAGAh7GEvH74S6FOH42FLoXpXbE/aqXSrIQjXgsiwM= golang.org/x/net v0.33.0/go.mod h1:HXLR5J+9DxmrqMwG9qjGCxZ+zKXxBru04zlTvWlWuN4= -golang.org/x/net v0.46.0 h1:giFlY12I07fugqwPuWJi68oOnpfqFnJIJzaIIm2JVV4= -golang.org/x/net v0.46.0/go.mod h1:Q9BGdFy1y4nkUwiLvT5qtyhAnEHgnQ/zd8PfU6nc210= golang.org/x/net v0.47.0 h1:Mx+4dIFzqraBXUugkia1OOvlD6LemFo1ALMHjrXDOhY= golang.org/x/net v0.47.0/go.mod h1:/jNxtkgq5yWUGYkaZGqo27cfGZ1c5Nen03aYrrKpVRU= golang.org/x/oauth2 v0.27.0 h1:da9Vo7/tDv5RH/7nZDz1eMGS/q1Vv1N/7FCrBhI9I3M= @@ -392,8 +359,6 @@ golang.org/x/sync v0.3.0/go.mod h1:FU7BRWz2tNW+3quACPkgCx/L+uEAv1htQ0V83Z9Rj+Y= golang.org/x/sync v0.6.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= golang.org/x/sync v0.7.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= golang.org/x/sync v0.10.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= -golang.org/x/sync v0.17.0 h1:l60nONMj9l5drqw6jlhIELNv9I0A4OFgRsG9k2oT9Ug= -golang.org/x/sync v0.17.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= golang.org/x/sync v0.18.0 h1:kr88TuHDroi+UVf+0hZnirlk8o8T+4MrK6mr60WkH/I= golang.org/x/sync v0.18.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= @@ -403,8 +368,6 @@ golang.org/x/sys v0.0.0-20200602225109-6fdc65e7d980/go.mod h1:h1NjWce9XRLGQEsW7w golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.0.0-20211007075335-d3039528d8ac/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.0.0-20220310020820-b874c991c1a5/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= @@ -415,8 +378,6 @@ golang.org/x/sys v0.12.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.17.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= golang.org/x/sys v0.20.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= golang.org/x/sys v0.28.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= -golang.org/x/sys v0.37.0 h1:fdNQudmxPjkdUTPnLn5mdQv7Zwvbvpaxqs831goi9kQ= -golang.org/x/sys v0.37.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= golang.org/x/sys v0.38.0 h1:3yZWxaJjBmCWXqhN1qh02AkOnCQ1poK6oF+a7xWL6Gc= golang.org/x/sys v0.38.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= golang.org/x/telemetry v0.0.0-20240228155512-f48c80bd79b2/go.mod h1:TeRTkGYfJXctD9OcfyVLyj2J3IxLnKwHJR8f4D8a3YE= @@ -438,8 +399,6 @@ golang.org/x/text v0.13.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE= golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU= golang.org/x/text v0.15.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU= golang.org/x/text v0.21.0/go.mod h1:4IBbMaMmOPCJ8SecivzSH54+73PCFmPWxNTLm+vZkEQ= -golang.org/x/text v0.30.0 h1:yznKA/E9zq54KzlzBEAWn1NXSQ8DIp/NYMy88xJjl4k= -golang.org/x/text v0.30.0/go.mod h1:yDdHFIX9t+tORqspjENWgzaCVXgk0yYnYuSZ8UzzBVM= golang.org/x/text v0.31.0 h1:aC8ghyu4JhP8VojJ2lEHBnochRno1sgL6nEi9WGFGMM= golang.org/x/text v0.31.0/go.mod h1:tKRAlv61yKIjGGHX/4tP1LTbc13YSec1pxVEWXzfoeM= golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ= @@ -450,22 +409,15 @@ golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc golang.org/x/tools v0.6.0/go.mod h1:Xwgl3UAJ/d3gWutnCtw505GrjyAbvKui8lOU390QaIU= golang.org/x/tools v0.13.0/go.mod h1:HvlwmtVNQAhOuCjW7xxvovg8wbNq7LwfXh/k7wXUl58= golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d/go.mod h1:aiJjzUbINMkxbQROHiO6hDPo2LHcIPhhQsa9DLh0yGk= -golang.org/x/tools v0.37.0 h1:DVSRzp7FwePZW356yEAChSdNcQo6Nsp+fex1SUW09lE= -golang.org/x/tools v0.37.0/go.mod h1:MBN5QPQtLMHVdvsbtarmTNukZDdgwdwlO5qGacAzF0w= +golang.org/x/tools v0.38.0 h1:Hx2Xv8hISq8Lm16jvBZ2VQf+RLmbd7wVUsALibYI/IQ= golang.org/x/tools v0.38.0/go.mod h1:yEsQ/d/YK8cjh0L6rZlY8tgtlKiBNTL14pGDJPJpYQs= golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= -google.golang.org/genproto/googleapis/api v0.0.0-20250324211829-b45e905df463 h1:hE3bRWtU6uceqlh4fhrSnUyjKHMKB9KrTLLG+bc0ddM= -google.golang.org/genproto/googleapis/api v0.0.0-20250324211829-b45e905df463/go.mod h1:U90ffi8eUL9MwPcrJylN5+Mk2v3vuPDptd5yyNUiRR8= google.golang.org/genproto/googleapis/api v0.0.0-20251202230838-ff82c1b0f217 h1:fCvbg86sFXwdrl5LgVcTEvNC+2txB5mgROGmRL5mrls= google.golang.org/genproto/googleapis/api v0.0.0-20251202230838-ff82c1b0f217/go.mod h1:+rXWjjaukWZun3mLfjmVnQi18E1AsFbDN9QdJ5YXLto= -google.golang.org/genproto/googleapis/rpc v0.0.0-20250313205543-e70fdf4c4cb4 h1:iK2jbkWL86DXjEx0qiHcRE9dE4/Ahua5k6V8OWFb//c= -google.golang.org/genproto/googleapis/rpc v0.0.0-20250313205543-e70fdf4c4cb4/go.mod h1:LuRYeWDFV6WOn90g357N17oMCaxpgCnbi/44qJvDn2I= google.golang.org/genproto/googleapis/rpc v0.0.0-20251124214823-79d6a2a48846 h1:Wgl1rcDNThT+Zn47YyCXOXyX/COgMTIdhJ717F0l4xk= google.golang.org/genproto/googleapis/rpc v0.0.0-20251124214823-79d6a2a48846/go.mod h1:7i2o+ce6H/6BluujYR+kqX3GKH+dChPTQU19wjRPiGk= google.golang.org/grpc v1.71.0 h1:kF77BGdPTQ4/JZWMlb9VpJ5pa25aqvVqogsxNHHdeBg= google.golang.org/grpc v1.71.0/go.mod h1:H0GRtasmQOh9LkFoCPDu3ZrwUtD1YGE+b2vYBYd/8Ec= -google.golang.org/protobuf v1.36.6 h1:z1NpPI8ku2WgiWnf+t9wTPsn6eP1L7ksHUlkfLvd9xY= -google.golang.org/protobuf v1.36.6/go.mod h1:jduwjTPXsFjZGTmRluh+L6NjiWu7pchiJ2/5YcXBHnY= google.golang.org/protobuf v1.36.10 h1:AYd7cD/uASjIL6Q9LiTjz8JLcrh/88q5UObnmY3aOOE= google.golang.org/protobuf v1.36.10/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= diff --git a/model/action.go b/model/action.go new file mode 100644 index 000000000..a7868d525 --- /dev/null +++ b/model/action.go @@ -0,0 +1,29 @@ +package model + +import ( + "time" + + "gorm.io/gorm" +) + +// Action represents a single action within a job +type Action struct { + gorm.Model + // JobID is the ID of the job this action belongs to + JobID uint `gorm:"column:job_id;not null;index"` + // ActionType is the type of action (e.g., "stream", "stream_end", "mk_vod", "check_vod", "mk_thumb") + ActionType string `gorm:"column:action_type;not null"` + // Status is the current status of the action + Status WorkState `gorm:"column:status;not null;default:'running'"` + // StartedAt is the timestamp when the action started + StartedAt *time.Time `gorm:"column:started_at"` + // CompletedAt is the timestamp when the action completed + CompletedAt *time.Time `gorm:"column:completed_at"` + // LastError is the error message if the action failed + LastError string `gorm:"column:last_error;type:text"` +} + +// TableName returns the name of the table for the Action model in the database. +func (*Action) TableName() string { + return "actions" +} diff --git a/model/job.go b/model/job.go new file mode 100644 index 000000000..22d30e62d --- /dev/null +++ b/model/job.go @@ -0,0 +1,56 @@ +package model + +import ( + "time" + + "gorm.io/gorm" +) + +// WorkState represents the current state of a job or action +type WorkState string + +const ( + // WorkStateCreated means the job/action was just created + WorkStateCreated WorkState = "created" + // WorkStateRunning means the job/action is currently running + WorkStateRunning WorkState = "running" + // WorkStateCompleted means the job/action completed successfully + WorkStateCompleted WorkState = "completed" + // WorkStateFailed means the job/action failed + WorkStateFailed WorkState = "failed" + // WorkStateCancelled means the job/action was cancelled + WorkStateCancelled WorkState = "cancelled" +) + +// Job represents a runner job that processes stream-related tasks. +// Jobs are created when a stream request is made to a runner, +// and track the progress through multiple actions. +type Job struct { + gorm.Model + // JobID is the unique identifier for the job (UUID from runner) + JobID string `gorm:"column:job_id;uniqueIndex;not null"` + // RunnerHostname is the hostname of the runner executing the job + RunnerHostname string `gorm:"column:runner_hostname;not null;index"` + // StreamID is the ID of the stream this job is associated with + StreamID uint `gorm:"column:stream_id;not null;index"` + // StreamVersion is the version of the stream (COMB, CAM, PRES) + StreamVersion StreamVersion `gorm:"column:stream_version;not null"` + // Status is the current status of the job + Status WorkState `gorm:"column:status;not null;default:'created';index"` + // StartedAt is the timestamp when the job started + StartedAt *time.Time `gorm:"column:started_at"` + // CompletedAt is the timestamp when the job completed (or failed) + CompletedAt *time.Time `gorm:"column:completed_at"` + // Actions is the list of actions executed as part of this job + Actions []Action `gorm:"foreignKey:JobID;references:ID"` +} + +// TableName returns the name of the table for the Job model in the database. +func (*Job) TableName() string { + return "jobs" +} + +// IsActive returns true if the job is still active (not completed, failed, or cancelled) +func (j *Job) IsActive() bool { + return j.Status == WorkStateCreated || j.Status == WorkStateRunning +} diff --git a/model/runner.go b/model/runner.go index 433635cb6..26176b749 100644 --- a/model/runner.go +++ b/model/runner.go @@ -28,6 +28,9 @@ type Runner struct { JobCount uint64 `gorm:"column:job_count;not null;default:0"` // Version is the version of the runner. Version string `gorm:"column:version;not null;default:'dev'"` + // Jobs is the list of jobs currently associated with this runner. + // This is populated when explicitly requested. + Jobs []Job `gorm:"foreignKey:RunnerHostname;references:Hostname"` } // TableName returns the name of the table for the Runner model in the database. diff --git a/model/stream.go b/model/stream.go index 8f55883c8..51bcdb1c7 100755 --- a/model/stream.go +++ b/model/stream.go @@ -55,6 +55,7 @@ type Stream struct { VideoSections []VideoSection TranscodingProgresses []TranscodingProgress `gorm:"foreignKey:StreamID"` Private bool `gorm:"not null;default:false"` + Jobs []Job `gorm:"foreignKey:StreamID"` // Jobs associated with this stream Watched bool `gorm:"-"` // Used to determine if stream is watched when loaded for a specific user. } diff --git a/pkg/runner_manager/manager.go b/pkg/runner_manager/manager.go index c0e92a32e..2d6e27643 100644 --- a/pkg/runner_manager/manager.go +++ b/pkg/runner_manager/manager.go @@ -226,6 +226,10 @@ func (m *Manager) Notify(ctx context.Context, notification *protobuf.Notificatio return m.handleVODReady(ctx, notification.GetVodReady()) case *protobuf.Notification_ThumbnailReady: return &protobuf.NotificationResponse{}, m.saveThumbnail(ctx, notification.GetThumbnailReady()) + case *protobuf.Notification_JobUpdate: + return &protobuf.NotificationResponse{}, m.handleJobUpdate(ctx, notification.GetJobUpdate()) + case *protobuf.Notification_ActionUpdate: + return &protobuf.NotificationResponse{}, m.handleActionUpdate(ctx, notification.GetActionUpdate()) default: return nil, status.Error(codes.Unimplemented, "unsupported notification type") } @@ -598,6 +602,115 @@ func (m *Manager) streamEnded(ctx context.Context, notification *protobuf.Stream return nil } +// handleJobUpdate handles job update notifications from runners +func (m *Manager) handleJobUpdate(ctx context.Context, notification *protobuf.JobUpdateNotification) error { + m.logger.Debug("jobUpdate", "payload", notification) + + // Convert protobuf types to model types + workState := protoWorkStateToModel(notification.GetStatus()) + streamVersion := protoStreamVersionToModel(notification.GetStreamVersion()) + + // Try to get existing job or create new one + job, err := m.dao.JobDao.GetByJobID(ctx, notification.GetJobId()) + if err != nil { + // Job doesn't exist, create new one + now := time.Now() + job = model.Job{ + JobID: notification.GetJobId(), + RunnerHostname: notification.GetRunnerHostname(), + StreamID: uint(notification.GetStream().GetId()), + StreamVersion: streamVersion, + Status: workState, + StartedAt: &now, + } + return m.dao.JobDao.Create(ctx, &job) + } + + // Update existing job + job.Status = workState + + // Set completion time if job is done + if workState == model.WorkStateCompleted || workState == model.WorkStateFailed || workState == model.WorkStateCancelled { + now := time.Now() + job.CompletedAt = &now + } + + return m.dao.JobDao.Update(ctx, &job) +} + +// handleActionUpdate handles action update notifications from runners +func (m *Manager) handleActionUpdate(ctx context.Context, notification *protobuf.ActionUpdateNotification) error { + m.logger.Debug("actionUpdate", "payload", notification) + + // Get the job by its string JobID + job, err := m.dao.JobDao.GetByJobID(ctx, notification.GetJobId()) + if err != nil { + return fmt.Errorf("job not found: %w", err) + } + + // Convert protobuf status to model status + workState := protoWorkStateToModel(notification.GetStatus()) + + // Try to get existing action or create new one + action, err := m.dao.JobDao.GetActionByJobIDAndType(ctx, job.ID, notification.GetActionType()) + if err != nil { + // Action doesn't exist, create new one + now := time.Now() + action = model.Action{ + JobID: job.ID, + ActionType: notification.GetActionType(), + Status: workState, + StartedAt: &now, + LastError: notification.GetLastError(), + } + return m.dao.JobDao.CreateAction(ctx, &action) + } + + // Update existing action + action.Status = workState + action.LastError = notification.GetLastError() + + // Set completion time if action is done + if workState == model.WorkStateCompleted || workState == model.WorkStateFailed { + now := time.Now() + action.CompletedAt = &now + } + + return m.dao.JobDao.UpdateAction(ctx, &action) +} + +// protoWorkStateToModel converts protobuf WorkState to model WorkState +func protoWorkStateToModel(status protobuf.WorkState) model.WorkState { + switch status { + case protobuf.WorkState_WORK_STATE_CREATED: + return model.WorkStateCreated + case protobuf.WorkState_WORK_STATE_RUNNING: + return model.WorkStateRunning + case protobuf.WorkState_WORK_STATE_COMPLETED: + return model.WorkStateCompleted + case protobuf.WorkState_WORK_STATE_FAILED: + return model.WorkStateFailed + case protobuf.WorkState_WORK_STATE_CANCELLED: + return model.WorkStateCancelled + default: + return model.WorkStateCreated + } +} + +// protoStreamVersionToModel converts protobuf StreamVersion to model StreamVersion +func protoStreamVersionToModel(version protobuf.StreamVersion) model.StreamVersion { + switch version { + case protobuf.StreamVersion_STREAM_VERSION_COMBINED: + return model.COMB + case protobuf.StreamVersion_STREAM_VERSION_CAMERA: + return model.CAM + case protobuf.StreamVersion_STREAM_VERSION_PRESENTATION: + return model.PRES + default: + return model.COMB + } +} + func dialRunner(runner model.Runner) (*grpc.ClientConn, error) { return grpc.NewClient(fmt.Sprintf("%s:%d", runner.Hostname, runner.Port), grpc.WithTransportCredentials(insecure.NewCredentials())) } diff --git a/runner/go.mod b/runner/go.mod index dea0e06be..f4967ca0f 100644 --- a/runner/go.mod +++ b/runner/go.mod @@ -8,6 +8,7 @@ require ( github.com/caarlos0/env v3.5.0+incompatible github.com/google/uuid v1.6.0 github.com/icza/gox v0.2.0 + github.com/joschahenningsen/thumbgen v0.1.2 github.com/otiai10/copy v1.14.1 github.com/prometheus/client_golang v1.21.1 github.com/sethvargo/go-retry v0.3.0 @@ -24,7 +25,6 @@ require ( github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect github.com/ebitengine/purego v0.9.0 // indirect github.com/go-ole/go-ole v1.3.0 // indirect - github.com/joschahenningsen/thumbgen v0.1.2 // indirect github.com/klauspost/compress v1.17.11 // indirect github.com/lufia/plan9stats v0.0.0-20231016141302-07b5767bb0ed // indirect github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect diff --git a/runner/notifications.proto b/runner/notifications.proto index 309ab1783..b44438930 100644 --- a/runner/notifications.proto +++ b/runner/notifications.proto @@ -12,6 +12,8 @@ message Notification { HeartbeatNotification heartbeat = 3; VODReadyNotification vod_ready = 4; ThumbnailReadyNotification thumbnail_ready = 5; + JobUpdateNotification job_update = 6; + ActionUpdateNotification action_update = 7; } } @@ -47,6 +49,33 @@ message ThumbnailReadyNotification { bytes thumbnail = 3; } +// WorkState represents the state of a job or action +enum WorkState { + WORK_STATE_UNSPECIFIED = 0; + WORK_STATE_CREATED = 1; + WORK_STATE_RUNNING = 2; + WORK_STATE_COMPLETED = 3; + WORK_STATE_FAILED = 4; + WORK_STATE_CANCELLED = 5; +} + +// JobUpdateNotification is sent when a job's status changes +message JobUpdateNotification { + string job_id = 1; + string runner_hostname = 2; + StreamInfo stream = 3; + StreamVersion stream_version = 4; + WorkState status = 5; +} + +// ActionUpdateNotification is sent when an action's status changes +message ActionUpdateNotification { + string job_id = 1; + string action_type = 2; + WorkState status = 3; + string last_error = 4; +} + message NotificationResponse { } diff --git a/runner/protobuf/commons.pb.go b/runner/protobuf/commons.pb.go index 334b2fe15..56142ffa9 100644 --- a/runner/protobuf/commons.pb.go +++ b/runner/protobuf/commons.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.6 -// protoc v6.30.2 +// protoc-gen-go v1.36.10 +// protoc v5.28.0 // source: commons.proto package protobuf diff --git a/runner/protobuf/notifications.pb.go b/runner/protobuf/notifications.pb.go index 4cd1a4f97..843488ca0 100644 --- a/runner/protobuf/notifications.pb.go +++ b/runner/protobuf/notifications.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.6 -// protoc v6.30.2 +// protoc-gen-go v1.36.10 +// protoc v5.28.0 // source: notifications.proto package protobuf @@ -21,6 +21,65 @@ const ( _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) ) +// WorkState represents the state of a job or action +type WorkState int32 + +const ( + WorkState_WORK_STATE_UNSPECIFIED WorkState = 0 + WorkState_WORK_STATE_CREATED WorkState = 1 + WorkState_WORK_STATE_RUNNING WorkState = 2 + WorkState_WORK_STATE_COMPLETED WorkState = 3 + WorkState_WORK_STATE_FAILED WorkState = 4 + WorkState_WORK_STATE_CANCELLED WorkState = 5 +) + +// Enum value maps for WorkState. +var ( + WorkState_name = map[int32]string{ + 0: "WORK_STATE_UNSPECIFIED", + 1: "WORK_STATE_CREATED", + 2: "WORK_STATE_RUNNING", + 3: "WORK_STATE_COMPLETED", + 4: "WORK_STATE_FAILED", + 5: "WORK_STATE_CANCELLED", + } + WorkState_value = map[string]int32{ + "WORK_STATE_UNSPECIFIED": 0, + "WORK_STATE_CREATED": 1, + "WORK_STATE_RUNNING": 2, + "WORK_STATE_COMPLETED": 3, + "WORK_STATE_FAILED": 4, + "WORK_STATE_CANCELLED": 5, + } +) + +func (x WorkState) Enum() *WorkState { + p := new(WorkState) + *p = x + return p +} + +func (x WorkState) String() string { + return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x)) +} + +func (WorkState) Descriptor() protoreflect.EnumDescriptor { + return file_notifications_proto_enumTypes[0].Descriptor() +} + +func (WorkState) Type() protoreflect.EnumType { + return &file_notifications_proto_enumTypes[0] +} + +func (x WorkState) Number() protoreflect.EnumNumber { + return protoreflect.EnumNumber(x) +} + +// Deprecated: Use WorkState.Descriptor instead. +func (WorkState) EnumDescriptor() ([]byte, []int) { + return file_notifications_proto_rawDescGZIP(), []int{0} +} + type Notification struct { state protoimpl.MessageState `protogen:"open.v1"` // Types that are valid to be assigned to Data: @@ -30,6 +89,8 @@ type Notification struct { // *Notification_Heartbeat // *Notification_VodReady // *Notification_ThumbnailReady + // *Notification_JobUpdate + // *Notification_ActionUpdate Data isNotification_Data `protobuf_oneof:"data"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache @@ -117,6 +178,24 @@ func (x *Notification) GetThumbnailReady() *ThumbnailReadyNotification { return nil } +func (x *Notification) GetJobUpdate() *JobUpdateNotification { + if x != nil { + if x, ok := x.Data.(*Notification_JobUpdate); ok { + return x.JobUpdate + } + } + return nil +} + +func (x *Notification) GetActionUpdate() *ActionUpdateNotification { + if x != nil { + if x, ok := x.Data.(*Notification_ActionUpdate); ok { + return x.ActionUpdate + } + } + return nil +} + type isNotification_Data interface { isNotification_Data() } @@ -141,6 +220,14 @@ type Notification_ThumbnailReady struct { ThumbnailReady *ThumbnailReadyNotification `protobuf:"bytes,5,opt,name=thumbnail_ready,json=thumbnailReady,oneof"` } +type Notification_JobUpdate struct { + JobUpdate *JobUpdateNotification `protobuf:"bytes,6,opt,name=job_update,json=jobUpdate,oneof"` +} + +type Notification_ActionUpdate struct { + ActionUpdate *ActionUpdateNotification `protobuf:"bytes,7,opt,name=action_update,json=actionUpdate,oneof"` +} + func (*Notification_StreamStart) isNotification_Data() {} func (*Notification_StreamEnd) isNotification_Data() {} @@ -151,6 +238,10 @@ func (*Notification_VodReady) isNotification_Data() {} func (*Notification_ThumbnailReady) isNotification_Data() {} +func (*Notification_JobUpdate) isNotification_Data() {} + +func (*Notification_ActionUpdate) isNotification_Data() {} + type StreamInfo struct { state protoimpl.MessageState `protogen:"open.v1"` Id *uint64 `protobuf:"varint,1,opt,name=id" json:"id,omitempty"` @@ -479,6 +570,152 @@ func (x *ThumbnailReadyNotification) GetThumbnail() []byte { return nil } +// JobUpdateNotification is sent when a job's status changes +type JobUpdateNotification struct { + state protoimpl.MessageState `protogen:"open.v1"` + JobId *string `protobuf:"bytes,1,opt,name=job_id,json=jobId" json:"job_id,omitempty"` + RunnerHostname *string `protobuf:"bytes,2,opt,name=runner_hostname,json=runnerHostname" json:"runner_hostname,omitempty"` + Stream *StreamInfo `protobuf:"bytes,3,opt,name=stream" json:"stream,omitempty"` + StreamVersion *StreamVersion `protobuf:"varint,4,opt,name=stream_version,json=streamVersion,enum=protobuf.StreamVersion" json:"stream_version,omitempty"` + Status *WorkState `protobuf:"varint,5,opt,name=status,enum=protobuf.WorkState" json:"status,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *JobUpdateNotification) Reset() { + *x = JobUpdateNotification{} + mi := &file_notifications_proto_msgTypes[7] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *JobUpdateNotification) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*JobUpdateNotification) ProtoMessage() {} + +func (x *JobUpdateNotification) ProtoReflect() protoreflect.Message { + mi := &file_notifications_proto_msgTypes[7] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use JobUpdateNotification.ProtoReflect.Descriptor instead. +func (*JobUpdateNotification) Descriptor() ([]byte, []int) { + return file_notifications_proto_rawDescGZIP(), []int{7} +} + +func (x *JobUpdateNotification) GetJobId() string { + if x != nil && x.JobId != nil { + return *x.JobId + } + return "" +} + +func (x *JobUpdateNotification) GetRunnerHostname() string { + if x != nil && x.RunnerHostname != nil { + return *x.RunnerHostname + } + return "" +} + +func (x *JobUpdateNotification) GetStream() *StreamInfo { + if x != nil { + return x.Stream + } + return nil +} + +func (x *JobUpdateNotification) GetStreamVersion() StreamVersion { + if x != nil && x.StreamVersion != nil { + return *x.StreamVersion + } + return StreamVersion_STREAM_VERSION_UNSPECIFIED +} + +func (x *JobUpdateNotification) GetStatus() WorkState { + if x != nil && x.Status != nil { + return *x.Status + } + return WorkState_WORK_STATE_UNSPECIFIED +} + +// ActionUpdateNotification is sent when an action's status changes +type ActionUpdateNotification struct { + state protoimpl.MessageState `protogen:"open.v1"` + JobId *string `protobuf:"bytes,1,opt,name=job_id,json=jobId" json:"job_id,omitempty"` + ActionType *string `protobuf:"bytes,2,opt,name=action_type,json=actionType" json:"action_type,omitempty"` + Status *WorkState `protobuf:"varint,3,opt,name=status,enum=protobuf.WorkState" json:"status,omitempty"` + LastError *string `protobuf:"bytes,4,opt,name=last_error,json=lastError" json:"last_error,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *ActionUpdateNotification) Reset() { + *x = ActionUpdateNotification{} + mi := &file_notifications_proto_msgTypes[8] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *ActionUpdateNotification) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ActionUpdateNotification) ProtoMessage() {} + +func (x *ActionUpdateNotification) ProtoReflect() protoreflect.Message { + mi := &file_notifications_proto_msgTypes[8] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ActionUpdateNotification.ProtoReflect.Descriptor instead. +func (*ActionUpdateNotification) Descriptor() ([]byte, []int) { + return file_notifications_proto_rawDescGZIP(), []int{8} +} + +func (x *ActionUpdateNotification) GetJobId() string { + if x != nil && x.JobId != nil { + return *x.JobId + } + return "" +} + +func (x *ActionUpdateNotification) GetActionType() string { + if x != nil && x.ActionType != nil { + return *x.ActionType + } + return "" +} + +func (x *ActionUpdateNotification) GetStatus() WorkState { + if x != nil && x.Status != nil { + return *x.Status + } + return WorkState_WORK_STATE_UNSPECIFIED +} + +func (x *ActionUpdateNotification) GetLastError() string { + if x != nil && x.LastError != nil { + return *x.LastError + } + return "" +} + type NotificationResponse struct { state protoimpl.MessageState `protogen:"open.v1"` unknownFields protoimpl.UnknownFields @@ -487,7 +724,7 @@ type NotificationResponse struct { func (x *NotificationResponse) Reset() { *x = NotificationResponse{} - mi := &file_notifications_proto_msgTypes[7] + mi := &file_notifications_proto_msgTypes[9] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -499,7 +736,7 @@ func (x *NotificationResponse) String() string { func (*NotificationResponse) ProtoMessage() {} func (x *NotificationResponse) ProtoReflect() protoreflect.Message { - mi := &file_notifications_proto_msgTypes[7] + mi := &file_notifications_proto_msgTypes[9] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -512,21 +749,24 @@ func (x *NotificationResponse) ProtoReflect() protoreflect.Message { // Deprecated: Use NotificationResponse.ProtoReflect.Descriptor instead. func (*NotificationResponse) Descriptor() ([]byte, []int) { - return file_notifications_proto_rawDescGZIP(), []int{7} + return file_notifications_proto_rawDescGZIP(), []int{9} } var File_notifications_proto protoreflect.FileDescriptor const file_notifications_proto_rawDesc = "" + "\n" + - "\x13notifications.proto\x12\bprotobuf\x1a\rcommons.proto\"\xf1\x02\n" + + "\x13notifications.proto\x12\bprotobuf\x1a\rcommons.proto\"\xfe\x03\n" + "\fNotification\x12F\n" + "\fstream_start\x18\x01 \x01(\v2!.protobuf.StreamStartNotificationH\x00R\vstreamStart\x12@\n" + "\n" + "stream_end\x18\x02 \x01(\v2\x1f.protobuf.StreamEndNotificationH\x00R\tstreamEnd\x12?\n" + "\theartbeat\x18\x03 \x01(\v2\x1f.protobuf.HeartbeatNotificationH\x00R\theartbeat\x12=\n" + "\tvod_ready\x18\x04 \x01(\v2\x1e.protobuf.VODReadyNotificationH\x00R\bvodReady\x12O\n" + - "\x0fthumbnail_ready\x18\x05 \x01(\v2$.protobuf.ThumbnailReadyNotificationH\x00R\x0ethumbnailReadyB\x06\n" + + "\x0fthumbnail_ready\x18\x05 \x01(\v2$.protobuf.ThumbnailReadyNotificationH\x00R\x0ethumbnailReady\x12@\n" + + "\n" + + "job_update\x18\x06 \x01(\v2\x1f.protobuf.JobUpdateNotificationH\x00R\tjobUpdate\x12I\n" + + "\raction_update\x18\a \x01(\v2\".protobuf.ActionUpdateNotificationH\x00R\factionUpdateB\x06\n" + "\x04data\"\x1c\n" + "\n" + "StreamInfo\x12\x0e\n" + @@ -548,8 +788,28 @@ const file_notifications_proto_rawDesc = "" + "\x1aThumbnailReadyNotification\x12,\n" + "\x06stream\x18\x01 \x01(\v2\x14.protobuf.StreamInfoR\x06stream\x12>\n" + "\x0estream_version\x18\x02 \x01(\x0e2\x17.protobuf.StreamVersionR\rstreamVersion\x12\x1c\n" + - "\tthumbnail\x18\x03 \x01(\fR\tthumbnail\"\x16\n" + - "\x14NotificationResponseB\x11Z\x0frunner/protobufb\beditionsp\xe8\a" + "\tthumbnail\x18\x03 \x01(\fR\tthumbnail\"\xf2\x01\n" + + "\x15JobUpdateNotification\x12\x15\n" + + "\x06job_id\x18\x01 \x01(\tR\x05jobId\x12'\n" + + "\x0frunner_hostname\x18\x02 \x01(\tR\x0erunnerHostname\x12,\n" + + "\x06stream\x18\x03 \x01(\v2\x14.protobuf.StreamInfoR\x06stream\x12>\n" + + "\x0estream_version\x18\x04 \x01(\x0e2\x17.protobuf.StreamVersionR\rstreamVersion\x12+\n" + + "\x06status\x18\x05 \x01(\x0e2\x13.protobuf.WorkStateR\x06status\"\x9e\x01\n" + + "\x18ActionUpdateNotification\x12\x15\n" + + "\x06job_id\x18\x01 \x01(\tR\x05jobId\x12\x1f\n" + + "\vaction_type\x18\x02 \x01(\tR\n" + + "actionType\x12+\n" + + "\x06status\x18\x03 \x01(\x0e2\x13.protobuf.WorkStateR\x06status\x12\x1d\n" + + "\n" + + "last_error\x18\x04 \x01(\tR\tlastError\"\x16\n" + + "\x14NotificationResponse*\xa2\x01\n" + + "\tWorkState\x12\x1a\n" + + "\x16WORK_STATE_UNSPECIFIED\x10\x00\x12\x16\n" + + "\x12WORK_STATE_CREATED\x10\x01\x12\x16\n" + + "\x12WORK_STATE_RUNNING\x10\x02\x12\x18\n" + + "\x14WORK_STATE_COMPLETED\x10\x03\x12\x15\n" + + "\x11WORK_STATE_FAILED\x10\x04\x12\x18\n" + + "\x14WORK_STATE_CANCELLED\x10\x05B\x11Z\x0frunner/protobufb\beditionsp\xe8\a" var ( file_notifications_proto_rawDescOnce sync.Once @@ -563,36 +823,46 @@ func file_notifications_proto_rawDescGZIP() []byte { return file_notifications_proto_rawDescData } -var file_notifications_proto_msgTypes = make([]protoimpl.MessageInfo, 8) +var file_notifications_proto_enumTypes = make([]protoimpl.EnumInfo, 1) +var file_notifications_proto_msgTypes = make([]protoimpl.MessageInfo, 10) var file_notifications_proto_goTypes = []any{ - (*Notification)(nil), // 0: protobuf.Notification - (*StreamInfo)(nil), // 1: protobuf.StreamInfo - (*StreamStartNotification)(nil), // 2: protobuf.StreamStartNotification - (*StreamEndNotification)(nil), // 3: protobuf.StreamEndNotification - (*HeartbeatNotification)(nil), // 4: protobuf.HeartbeatNotification - (*VODReadyNotification)(nil), // 5: protobuf.VODReadyNotification - (*ThumbnailReadyNotification)(nil), // 6: protobuf.ThumbnailReadyNotification - (*NotificationResponse)(nil), // 7: protobuf.NotificationResponse - (StreamVersion)(0), // 8: protobuf.StreamVersion + (WorkState)(0), // 0: protobuf.WorkState + (*Notification)(nil), // 1: protobuf.Notification + (*StreamInfo)(nil), // 2: protobuf.StreamInfo + (*StreamStartNotification)(nil), // 3: protobuf.StreamStartNotification + (*StreamEndNotification)(nil), // 4: protobuf.StreamEndNotification + (*HeartbeatNotification)(nil), // 5: protobuf.HeartbeatNotification + (*VODReadyNotification)(nil), // 6: protobuf.VODReadyNotification + (*ThumbnailReadyNotification)(nil), // 7: protobuf.ThumbnailReadyNotification + (*JobUpdateNotification)(nil), // 8: protobuf.JobUpdateNotification + (*ActionUpdateNotification)(nil), // 9: protobuf.ActionUpdateNotification + (*NotificationResponse)(nil), // 10: protobuf.NotificationResponse + (StreamVersion)(0), // 11: protobuf.StreamVersion } var file_notifications_proto_depIdxs = []int32{ - 2, // 0: protobuf.Notification.stream_start:type_name -> protobuf.StreamStartNotification - 3, // 1: protobuf.Notification.stream_end:type_name -> protobuf.StreamEndNotification - 4, // 2: protobuf.Notification.heartbeat:type_name -> protobuf.HeartbeatNotification - 5, // 3: protobuf.Notification.vod_ready:type_name -> protobuf.VODReadyNotification - 6, // 4: protobuf.Notification.thumbnail_ready:type_name -> protobuf.ThumbnailReadyNotification - 1, // 5: protobuf.StreamStartNotification.stream:type_name -> protobuf.StreamInfo - 8, // 6: protobuf.StreamStartNotification.stream_version:type_name -> protobuf.StreamVersion - 1, // 7: protobuf.StreamEndNotification.stream:type_name -> protobuf.StreamInfo - 1, // 8: protobuf.VODReadyNotification.stream:type_name -> protobuf.StreamInfo - 8, // 9: protobuf.VODReadyNotification.stream_version:type_name -> protobuf.StreamVersion - 1, // 10: protobuf.ThumbnailReadyNotification.stream:type_name -> protobuf.StreamInfo - 8, // 11: protobuf.ThumbnailReadyNotification.stream_version:type_name -> protobuf.StreamVersion - 12, // [12:12] is the sub-list for method output_type - 12, // [12:12] is the sub-list for method input_type - 12, // [12:12] is the sub-list for extension type_name - 12, // [12:12] is the sub-list for extension extendee - 0, // [0:12] is the sub-list for field type_name + 3, // 0: protobuf.Notification.stream_start:type_name -> protobuf.StreamStartNotification + 4, // 1: protobuf.Notification.stream_end:type_name -> protobuf.StreamEndNotification + 5, // 2: protobuf.Notification.heartbeat:type_name -> protobuf.HeartbeatNotification + 6, // 3: protobuf.Notification.vod_ready:type_name -> protobuf.VODReadyNotification + 7, // 4: protobuf.Notification.thumbnail_ready:type_name -> protobuf.ThumbnailReadyNotification + 8, // 5: protobuf.Notification.job_update:type_name -> protobuf.JobUpdateNotification + 9, // 6: protobuf.Notification.action_update:type_name -> protobuf.ActionUpdateNotification + 2, // 7: protobuf.StreamStartNotification.stream:type_name -> protobuf.StreamInfo + 11, // 8: protobuf.StreamStartNotification.stream_version:type_name -> protobuf.StreamVersion + 2, // 9: protobuf.StreamEndNotification.stream:type_name -> protobuf.StreamInfo + 2, // 10: protobuf.VODReadyNotification.stream:type_name -> protobuf.StreamInfo + 11, // 11: protobuf.VODReadyNotification.stream_version:type_name -> protobuf.StreamVersion + 2, // 12: protobuf.ThumbnailReadyNotification.stream:type_name -> protobuf.StreamInfo + 11, // 13: protobuf.ThumbnailReadyNotification.stream_version:type_name -> protobuf.StreamVersion + 2, // 14: protobuf.JobUpdateNotification.stream:type_name -> protobuf.StreamInfo + 11, // 15: protobuf.JobUpdateNotification.stream_version:type_name -> protobuf.StreamVersion + 0, // 16: protobuf.JobUpdateNotification.status:type_name -> protobuf.WorkState + 0, // 17: protobuf.ActionUpdateNotification.status:type_name -> protobuf.WorkState + 18, // [18:18] is the sub-list for method output_type + 18, // [18:18] is the sub-list for method input_type + 18, // [18:18] is the sub-list for extension type_name + 18, // [18:18] is the sub-list for extension extendee + 0, // [0:18] is the sub-list for field type_name } func init() { file_notifications_proto_init() } @@ -607,19 +877,22 @@ func file_notifications_proto_init() { (*Notification_Heartbeat)(nil), (*Notification_VodReady)(nil), (*Notification_ThumbnailReady)(nil), + (*Notification_JobUpdate)(nil), + (*Notification_ActionUpdate)(nil), } type x struct{} out := protoimpl.TypeBuilder{ File: protoimpl.DescBuilder{ GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_notifications_proto_rawDesc), len(file_notifications_proto_rawDesc)), - NumEnums: 0, - NumMessages: 8, + NumEnums: 1, + NumMessages: 10, NumExtensions: 0, NumServices: 0, }, GoTypes: file_notifications_proto_goTypes, DependencyIndexes: file_notifications_proto_depIdxs, + EnumInfos: file_notifications_proto_enumTypes, MessageInfos: file_notifications_proto_msgTypes, }.Build() File_notifications_proto = out.File diff --git a/runner/protobuf/runner.pb.go b/runner/protobuf/runner.pb.go index 563d89871..3d7f342c4 100644 --- a/runner/protobuf/runner.pb.go +++ b/runner/protobuf/runner.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.6 -// protoc v6.30.2 +// protoc-gen-go v1.36.10 +// protoc v5.28.0 // source: runner.proto package protobuf diff --git a/runner/protobuf/runner_grpc.pb.go b/runner/protobuf/runner_grpc.pb.go index 518ca9ecf..423f04ea9 100644 --- a/runner/protobuf/runner_grpc.pb.go +++ b/runner/protobuf/runner_grpc.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go-grpc. DO NOT EDIT. // versions: -// - protoc-gen-go-grpc v1.5.1 -// - protoc v6.30.2 +// - protoc-gen-go-grpc v1.6.0 +// - protoc v5.28.0 // source: runner.proto package protobuf @@ -78,10 +78,10 @@ type RunnerServiceServer interface { type UnimplementedRunnerServiceServer struct{} func (UnimplementedRunnerServiceServer) RequestStream(context.Context, *StreamRequest) (*StreamResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method RequestStream not implemented") + return nil, status.Error(codes.Unimplemented, "method RequestStream not implemented") } func (UnimplementedRunnerServiceServer) RequestStreamEnd(context.Context, *StreamEndRequest) (*StreamEndResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method RequestStreamEnd not implemented") + return nil, status.Error(codes.Unimplemented, "method RequestStreamEnd not implemented") } func (UnimplementedRunnerServiceServer) mustEmbedUnimplementedRunnerServiceServer() {} func (UnimplementedRunnerServiceServer) testEmbeddedByValue() {} @@ -94,7 +94,7 @@ type UnsafeRunnerServiceServer interface { } func RegisterRunnerServiceServer(s grpc.ServiceRegistrar, srv RunnerServiceServer) { - // If the following call pancis, it indicates UnimplementedRunnerServiceServer was + // If the following call panics, it indicates UnimplementedRunnerServiceServer was // embedded by pointer and is nil. This will cause panics if an // unimplemented method is ever invoked, so we test this at initialization // time to prevent it from happening at runtime later due to I/O. @@ -226,10 +226,10 @@ type RunnerManagerServiceServer interface { type UnimplementedRunnerManagerServiceServer struct{} func (UnimplementedRunnerManagerServiceServer) Register(context.Context, *RegisterRequest) (*RegisterResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method Register not implemented") + return nil, status.Error(codes.Unimplemented, "method Register not implemented") } func (UnimplementedRunnerManagerServiceServer) Notify(context.Context, *Notification) (*NotificationResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method Notify not implemented") + return nil, status.Error(codes.Unimplemented, "method Notify not implemented") } func (UnimplementedRunnerManagerServiceServer) mustEmbedUnimplementedRunnerManagerServiceServer() {} func (UnimplementedRunnerManagerServiceServer) testEmbeddedByValue() {} @@ -242,7 +242,7 @@ type UnsafeRunnerManagerServiceServer interface { } func RegisterRunnerManagerServiceServer(s grpc.ServiceRegistrar, srv RunnerManagerServiceServer) { - // If the following call pancis, it indicates UnimplementedRunnerManagerServiceServer was + // If the following call panics, it indicates UnimplementedRunnerManagerServiceServer was // embedded by pointer and is nil. This will cause panics if an // unimplemented method is ever invoked, so we test this at initialization // time to prevent it from happening at runtime later due to I/O. diff --git a/runner/runner.go b/runner/runner.go index dbc305bb5..01620adaf 100644 --- a/runner/runner.go +++ b/runner/runner.go @@ -8,6 +8,7 @@ import ( "os" "reflect" "runtime" + "strings" "time" "github.com/google/uuid" @@ -170,13 +171,32 @@ func (r *Runner) RunAction(a []actions.Action, data map[string]any, logger *slog job := uuid.New().String() r.JobCount <- 1 r.jobs[job] = cancel + + // Extract stream info for job notifications + streamID, _ := data["streamID"].(uint64) + streamVersion, _ := data["streamVersion"].(string) + + // Send job created notification + r.sendJobUpdate(job, streamID, streamVersion, protobuf.WorkState_WORK_STATE_CREATED) + go func() { defer func() { cancel() delete(r.jobs, job) r.JobCount <- -1 }() + + jobFailed := false + for _, action := range a { + actionType := getActionTypeString(getFunctionName(action)) + + // Send action running notification + r.sendActionUpdate(job, actionType, protobuf.WorkState_WORK_STATE_RUNNING, "") + + // Update job status to running + r.sendJobUpdate(job, streamID, streamVersion, protobuf.WorkState_WORK_STATE_RUNNING) + for { log := logger.With("action", getFunctionName(action)).With("job", job) log.Info("running action") @@ -187,17 +207,90 @@ func (r *Runner) RunAction(a []actions.Action, data map[string]any, logger *slog log.Error("action error", "error", err) // use action specific logger if actions.IsAbortingError(err) { log.Info("action can't continue") + jobFailed = true + // Send action failed notification with error + r.sendActionUpdate(job, actionType, protobuf.WorkState_WORK_STATE_FAILED, err.Error()) break // escape retry loop on unrecoverable error } } else { + // Send action completed notification + r.sendActionUpdate(job, actionType, protobuf.WorkState_WORK_STATE_COMPLETED, "") break // escape retry loop on no error } } + if jobFailed { + break + } + } + + // Send final job status notification + if c.Err() != nil { + // Context was cancelled + r.sendJobUpdate(job, streamID, streamVersion, protobuf.WorkState_WORK_STATE_CANCELLED) + } else if jobFailed { + r.sendJobUpdate(job, streamID, streamVersion, protobuf.WorkState_WORK_STATE_FAILED) + } else { + r.sendJobUpdate(job, streamID, streamVersion, protobuf.WorkState_WORK_STATE_COMPLETED) } }() return job } +// sendJobUpdate sends a job update notification +func (r *Runner) sendJobUpdate(jobID string, streamID uint64, streamVersion string, status protobuf.WorkState) { + // Safely convert stream version string to protobuf enum + var version protobuf.StreamVersion + if val, ok := protobuf.StreamVersion_value[streamVersion]; ok { + version = protobuf.StreamVersion(val) + } else { + version = protobuf.StreamVersion_STREAM_VERSION_UNSPECIFIED + } + + r.notifications <- &protobuf.Notification{ + Data: &protobuf.Notification_JobUpdate{ + JobUpdate: &protobuf.JobUpdateNotification{ + JobId: ptr.Take(jobID), + RunnerHostname: ptr.Take(config.Config.Hostname), + Stream: &protobuf.StreamInfo{Id: ptr.Take(streamID)}, + StreamVersion: ptr.Take(version), + Status: ptr.Take(status), + }, + }, + } +} + +// sendActionUpdate sends an action update notification +func (r *Runner) sendActionUpdate(jobID string, actionType string, status protobuf.WorkState, lastError string) { + r.notifications <- &protobuf.Notification{ + Data: &protobuf.Notification_ActionUpdate{ + ActionUpdate: &protobuf.ActionUpdateNotification{ + JobId: ptr.Take(jobID), + ActionType: ptr.Take(actionType), + Status: ptr.Take(status), + LastError: ptr.Take(lastError), + }, + }, + } +} + +// getActionTypeString converts an action function name to a string action type +func getActionTypeString(funcName string) string { + switch { + case strings.HasSuffix(funcName, "Stream"): + return "stream" + case strings.HasSuffix(funcName, "StreamEnd"): + return "stream_end" + case strings.HasSuffix(funcName, "MkVOD"): + return "mk_vod" + case strings.HasSuffix(funcName, "CheckVoD"): + return "check_vod" + case strings.HasSuffix(funcName, "MkThumb"): + return "mk_thumb" + default: + return funcName + } +} + func (r *Runner) handleNotifications(ctx context.Context) { b := retry.NewFibonacci(1 * time.Second) b = retry.WithJitter(500*time.Millisecond, b)