Add Job and Action tracking for runner with GRPC notifications

.gitignore

@@ -114,3 +114,6 @@ tumlive.service
 
 # osx
 .DS_Store
+
+# downloaded protoc binary
+protoc-*.zip

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)

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(),
 	}
 }

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
+}

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
+}

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