diff --git a/apiclient/Valkyrie/Execute a script.bru b/apiclient/Valkyrie/Execute a script.bru index 976ce84b..419db4ce 100644 --- a/apiclient/Valkyrie/Execute a script.bru +++ b/apiclient/Valkyrie/Execute a script.bru @@ -13,8 +13,7 @@ post { body:json { { - "code": "print('2nd minute')", - "language": "python", - "cron_expression": "*/2 * * * *" + "code": "from time import sleep\n\nfor i in range(30):\n sleep(1)\n print(\"Hello\")", + "language": "python" } } diff --git a/build/package/nix/odin.nix b/build/package/nix/odin.nix index 4e9cbe1b..b7ecbfed 100644 --- a/build/package/nix/odin.nix +++ b/build/package/nix/odin.nix @@ -12,7 +12,7 @@ buildGoModule rec { pname = "odin"; version = "0.0.1"; - vendorHash = "sha256-g+YA2d4tuAtGazjtNiIyyaWbJfnZXMeHk7e8EDr+uUw="; + vendorHash = "sha256-JE1JuOGw2mV0WRESdHw+TnsA4UzHVrtNadnoWmPwvOg="; src = ../../..; diff --git a/go.mod b/go.mod index 5b92b741..6ce80031 100644 --- a/go.mod +++ b/go.mod @@ -3,7 +3,6 @@ module github.com/deepakdinesh1123/valkyrie go 1.22.0 require ( - github.com/adhocore/gronx v1.19.0 github.com/containers/podman/v5 v5.2.0 github.com/docker/docker v27.1.1+incompatible github.com/exaring/otelpgx v0.6.2 diff --git a/go.sum b/go.sum index a63f1a4d..ef12d716 100644 --- a/go.sum +++ b/go.sum @@ -16,8 +16,6 @@ github.com/VividCortex/ewma v1.2.0 h1:f58SaIzcDXrSy3kWaHNvuJgJ3Nmz59Zji6XoJR/q1o github.com/VividCortex/ewma v1.2.0/go.mod h1:nz4BbCtbLyFDeC9SUHbtcT5644juEuWfUAUnGx7j5l4= github.com/acarl005/stripansi v0.0.0-20180116102854-5a71ef0e047d h1:licZJFw2RwpHMqeKTCYkitsPqHNxTmd4SNR5r94FGM8= github.com/acarl005/stripansi v0.0.0-20180116102854-5a71ef0e047d/go.mod h1:asat636LX7Bqt5lYEZ27JNDcqxfjdBQuJ/MM4CN/Lzo= -github.com/adhocore/gronx v1.19.0 h1:GrEvNMPDwXND+YFadCyFVQPC+/xxoGJaQzu+duNf6aU= -github.com/adhocore/gronx v1.19.0/go.mod h1:7oUY1WAU8rEJWmAxXR2DN0JaO4gi9khSgKjiRypqteg= github.com/asaskevich/govalidator v0.0.0-20230301143203-a9d515a09cc2 h1:DklsrG3dyBCFEj5IhUbnKptjxatkF07cF2ak3yi77so= github.com/asaskevich/govalidator v0.0.0-20230301143203-a9d515a09cc2/go.mod h1:WaHUgvxTVq04UNunO+XhnAqY/wQc+bxr74GqbsZ/Jqw= github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= diff --git a/internal/odin/db/job.sql.go b/internal/odin/db/job.sql.go index 869d579d..88773f64 100644 --- a/internal/odin/db/job.sql.go +++ b/internal/odin/db/job.sql.go @@ -35,13 +35,13 @@ where id = ( select id from jobs where status = 'pending' - and next_run_at <= now() + and retries < max_retries order by id asc for update skip locked limit 1 ) -returning id, cron_expression, last_scheduled_at, next_run_at, created_at, updated_at, exec_request_id, status, retries, max_retries +returning id, created_at, updated_at, exec_request_id, status, retries, max_retries ` func (q *Queries) FetchJob(ctx context.Context) (Job, error) { @@ -49,9 +49,6 @@ func (q *Queries) FetchJob(ctx context.Context) (Job, error) { var i Job err := row.Scan( &i.ID, - &i.CronExpression, - &i.LastScheduledAt, - &i.NextRunAt, &i.CreatedAt, &i.UpdatedAt, &i.ExecRequestID, @@ -127,7 +124,7 @@ func (q *Queries) GetAllExecutionResults(ctx context.Context, arg GetAllExecutio } const getAllJobs = `-- name: GetAllJobs :many -select jobs.id, cron_expression, last_scheduled_at, next_run_at, created_at, updated_at, exec_request_id, status, retries, max_retries, exec_request.id, hash, code, path, flake, args, programming_language from jobs +select jobs.id, created_at, updated_at, exec_request_id, status, retries, max_retries, exec_request.id, hash, code, path, flake, args, programming_language from jobs inner join exec_request on jobs.exec_request_id = exec_request.id order by jobs.id limit $1 offset $2 @@ -140,9 +137,6 @@ type GetAllJobsParams struct { type GetAllJobsRow struct { ID int64 `db:"id" json:"id"` - CronExpression pgtype.Text `db:"cron_expression" json:"cron_expression"` - LastScheduledAt pgtype.Timestamptz `db:"last_scheduled_at" json:"last_scheduled_at"` - NextRunAt pgtype.Timestamptz `db:"next_run_at" json:"next_run_at"` CreatedAt pgtype.Timestamptz `db:"created_at" json:"created_at"` UpdatedAt pgtype.Timestamptz `db:"updated_at" json:"updated_at"` ExecRequestID pgtype.Int4 `db:"exec_request_id" json:"exec_request_id"` @@ -169,9 +163,6 @@ func (q *Queries) GetAllJobs(ctx context.Context, arg GetAllJobsParams) ([]GetAl var i GetAllJobsRow if err := rows.Scan( &i.ID, - &i.CronExpression, - &i.LastScheduledAt, - &i.NextRunAt, &i.CreatedAt, &i.UpdatedAt, &i.ExecRequestID, @@ -262,14 +253,11 @@ func (q *Queries) GetExecutionResultsByID(ctx context.Context, arg GetExecutionR } const getJob = `-- name: GetJob :one -select jobs.id, cron_expression, last_scheduled_at, next_run_at, created_at, updated_at, exec_request_id, status, retries, max_retries, exec_request.id, hash, code, path, flake, args, programming_language from jobs inner join exec_request on jobs.exec_request_id = exec_request.id where jobs.id = $1 +select jobs.id, created_at, updated_at, exec_request_id, status, retries, max_retries, exec_request.id, hash, code, path, flake, args, programming_language from jobs inner join exec_request on jobs.exec_request_id = exec_request.id where jobs.id = $1 ` type GetJobRow struct { ID int64 `db:"id" json:"id"` - CronExpression pgtype.Text `db:"cron_expression" json:"cron_expression"` - LastScheduledAt pgtype.Timestamptz `db:"last_scheduled_at" json:"last_scheduled_at"` - NextRunAt pgtype.Timestamptz `db:"next_run_at" json:"next_run_at"` CreatedAt pgtype.Timestamptz `db:"created_at" json:"created_at"` UpdatedAt pgtype.Timestamptz `db:"updated_at" json:"updated_at"` ExecRequestID pgtype.Int4 `db:"exec_request_id" json:"exec_request_id"` @@ -290,9 +278,6 @@ func (q *Queries) GetJob(ctx context.Context, id int64) (GetJobRow, error) { var i GetJobRow err := row.Scan( &i.ID, - &i.CronExpression, - &i.LastScheduledAt, - &i.NextRunAt, &i.CreatedAt, &i.UpdatedAt, &i.ExecRequestID, @@ -345,34 +330,22 @@ func (q *Queries) GetTotalJobs(ctx context.Context) (int64, error) { const insertJob = `-- name: InsertJob :one insert into jobs - (cron_expression, exec_request_id, last_scheduled_at, next_run_at, max_retries) + (exec_request_id, max_retries) values - ($1, $2, $3, $4, $5) -returning id, cron_expression, last_scheduled_at, next_run_at, created_at, updated_at, exec_request_id, status, retries, max_retries + ($1, $2) +returning id, created_at, updated_at, exec_request_id, status, retries, max_retries ` type InsertJobParams struct { - CronExpression pgtype.Text `db:"cron_expression" json:"cron_expression"` - ExecRequestID pgtype.Int4 `db:"exec_request_id" json:"exec_request_id"` - LastScheduledAt pgtype.Timestamptz `db:"last_scheduled_at" json:"last_scheduled_at"` - NextRunAt pgtype.Timestamptz `db:"next_run_at" json:"next_run_at"` - MaxRetries pgtype.Int4 `db:"max_retries" json:"max_retries"` + ExecRequestID pgtype.Int4 `db:"exec_request_id" json:"exec_request_id"` + MaxRetries pgtype.Int4 `db:"max_retries" json:"max_retries"` } func (q *Queries) InsertJob(ctx context.Context, arg InsertJobParams) (Job, error) { - row := q.db.QueryRow(ctx, insertJob, - arg.CronExpression, - arg.ExecRequestID, - arg.LastScheduledAt, - arg.NextRunAt, - arg.MaxRetries, - ) + row := q.db.QueryRow(ctx, insertJob, arg.ExecRequestID, arg.MaxRetries) var i Job err := row.Scan( &i.ID, - &i.CronExpression, - &i.LastScheduledAt, - &i.NextRunAt, &i.CreatedAt, &i.UpdatedAt, &i.ExecRequestID, @@ -467,27 +440,6 @@ func (q *Queries) UpdateJobCompleted(ctx context.Context, id int64) error { return err } -const updateJobSchedule = `-- name: UpdateJobSchedule :exec -update jobs -set - status = 'pending', - last_scheduled_at = $2, - next_run_at = $3, - updated_at = now() -where id = $1 AND status = 'scheduled' -` - -type UpdateJobScheduleParams struct { - ID int64 `db:"id" json:"id"` - LastScheduledAt pgtype.Timestamptz `db:"last_scheduled_at" json:"last_scheduled_at"` - NextRunAt pgtype.Timestamptz `db:"next_run_at" json:"next_run_at"` -} - -func (q *Queries) UpdateJobSchedule(ctx context.Context, arg UpdateJobScheduleParams) error { - _, err := q.db.Exec(ctx, updateJobSchedule, arg.ID, arg.LastScheduledAt, arg.NextRunAt) - return err -} - const updateJobFailed = `-- name: updateJobFailed :exec update jobs set diff --git a/internal/odin/db/migrations/000001_init.up.sql b/internal/odin/db/migrations/000001_init.up.sql index b63ab583..0e675c54 100644 --- a/internal/odin/db/migrations/000001_init.up.sql +++ b/internal/odin/db/migrations/000001_init.up.sql @@ -34,15 +34,12 @@ create table exec_request ( create table jobs ( id bigint primary key default nextval('jobs_id_seq'), - cron_expression text, - last_scheduled_at timestamptz default null, - next_run_at timestamptz default null, created_at timestamptz not null default now(), updated_at timestamptz, exec_request_id int references exec_request on delete set null, status TEXT NOT NULL CHECK (status IN ('pending', 'scheduled', 'completed', 'failed', 'cancelled')) DEFAULT 'pending', retries int default 0, - max_retries int default 0 + max_retries int default 5 ); create sequence job_runs_id_seq as bigint; diff --git a/internal/odin/db/models.go b/internal/odin/db/models.go index 24bf73a1..9ad49b0e 100644 --- a/internal/odin/db/models.go +++ b/internal/odin/db/models.go @@ -19,16 +19,13 @@ type ExecRequest struct { } type Job struct { - ID int64 `db:"id" json:"id"` - CronExpression pgtype.Text `db:"cron_expression" json:"cron_expression"` - LastScheduledAt pgtype.Timestamptz `db:"last_scheduled_at" json:"last_scheduled_at"` - NextRunAt pgtype.Timestamptz `db:"next_run_at" json:"next_run_at"` - CreatedAt pgtype.Timestamptz `db:"created_at" json:"created_at"` - UpdatedAt pgtype.Timestamptz `db:"updated_at" json:"updated_at"` - ExecRequestID pgtype.Int4 `db:"exec_request_id" json:"exec_request_id"` - Status string `db:"status" json:"status"` - Retries pgtype.Int4 `db:"retries" json:"retries"` - MaxRetries pgtype.Int4 `db:"max_retries" json:"max_retries"` + ID int64 `db:"id" json:"id"` + CreatedAt pgtype.Timestamptz `db:"created_at" json:"created_at"` + UpdatedAt pgtype.Timestamptz `db:"updated_at" json:"updated_at"` + ExecRequestID pgtype.Int4 `db:"exec_request_id" json:"exec_request_id"` + Status string `db:"status" json:"status"` + Retries pgtype.Int4 `db:"retries" json:"retries"` + MaxRetries pgtype.Int4 `db:"max_retries" json:"max_retries"` } type JobGroup struct { diff --git a/internal/odin/db/querier.go b/internal/odin/db/querier.go index ef0694c3..3f4781af 100644 --- a/internal/odin/db/querier.go +++ b/internal/odin/db/querier.go @@ -34,7 +34,6 @@ type Querier interface { RetryJob(ctx context.Context, id int64) error StopJob(ctx context.Context, id int64) error UpdateJobCompleted(ctx context.Context, id int64) error - UpdateJobSchedule(ctx context.Context, arg UpdateJobScheduleParams) error updateJobFailed(ctx context.Context, id int64) error } diff --git a/internal/odin/db/queries/job.sql b/internal/odin/db/queries/job.sql index 18ac382d..2c0e792a 100644 --- a/internal/odin/db/queries/job.sql +++ b/internal/odin/db/queries/job.sql @@ -1,15 +1,11 @@ create table jobs ( id bigint primary key default nextval('jobs_id_seq'), - inserted_at timestamptz not null default now(), - cron_expression text, - last_scheduled_at timestamptz default null, - next_run_at timestamptz default null, created_at timestamptz not null default now(), updated_at timestamptz, exec_request_id int references exec_request on delete set null, status TEXT NOT NULL CHECK (status IN ('pending', 'scheduled', 'completed', 'failed', 'cancelled')) DEFAULT 'pending', retries int default 0, - max_retries int default 0 + max_retries int default 5 ); create table job_runs ( @@ -28,7 +24,7 @@ where id = ( select id from jobs where status = 'pending' - and next_run_at <= now() + and retries < max_retries order by id asc for update skip locked @@ -38,9 +34,9 @@ returning *; -- name: InsertJob :one insert into jobs - (cron_expression, exec_request_id, last_scheduled_at, next_run_at, max_retries) + (exec_request_id, max_retries) values - ($1, $2, $3, $4, $5) + ($1, $2) returning *; -- name: UpdateJobCompleted :exec @@ -50,15 +46,6 @@ set updated_at = now() where id = $1 AND status = 'scheduled'; --- name: UpdateJobSchedule :exec -update jobs -set - status = 'pending', - last_scheduled_at = $2, - next_run_at = $3, - updated_at = now() -where id = $1 AND status = 'completed'; - -- name: InsertJobRun :one insert into job_runs (job_id, worker_id, started_at, finished_at, exec_request_id, logs) diff --git a/internal/odin/db/tx_add_job.go b/internal/odin/db/tx_add_job.go index 94e5978f..8943bac9 100644 --- a/internal/odin/db/tx_add_job.go +++ b/internal/odin/db/tx_add_job.go @@ -2,11 +2,8 @@ package db import ( "context" - "fmt" "log" - "time" - "github.com/adhocore/gronx" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgtype" ) @@ -18,7 +15,6 @@ type AddJobTxParams struct { Args string Path string ProgrammingLanguage string - CronExpression string MaxRetries int } @@ -33,10 +29,8 @@ func (s *SQLStore) AddJobTx(ctx context.Context, arg AddJobTxParams) (AddJobTxRe execReq, err := s.GetExecRequestByHash(ctx, arg.Hash) var execId int32 if err != nil { - log.Printf("GetExecRequestByHash error: %v", err) switch err { case pgx.ErrNoRows: - log.Println("InsertExecRequest") execId, err = s.InsertExecRequest(ctx, InsertExecRequestParams{ Code: arg.Code, Flake: arg.Flake, @@ -58,32 +52,7 @@ func (s *SQLStore) AddJobTx(ctx context.Context, arg AddJobTxParams) (AddJobTxRe var jobParams InsertJobParams jobParams.ExecRequestID = pgtype.Int4{Int32: execId, Valid: true} - jobParams.LastScheduledAt = pgtype.Timestamptz{ - Time: time.Now(), - Valid: true, - } jobParams.MaxRetries = pgtype.Int4{Int32: int32(arg.MaxRetries), Valid: true} - if arg.CronExpression == "" { - jobParams.NextRunAt = pgtype.Timestamptz{ - Time: time.Now(), - Valid: true, - } - } else { - gron := gronx.New() - if !gron.IsValid(arg.CronExpression) { - return fmt.Errorf("invalid cron expression: %s", arg.CronExpression) - } - jobParams.CronExpression = pgtype.Text{String: arg.CronExpression, Valid: true} - nextRunAt, err := gronx.NextTick(arg.CronExpression, true) - if err != nil { - log.Printf("NextTick error: %v", err) - return err - } - jobParams.NextRunAt = pgtype.Timestamptz{ - Time: nextRunAt, - Valid: true, - } - } job, err := s.InsertJob(ctx, jobParams) if err != nil { log.Printf("InsertJob error: %v", err) diff --git a/internal/odin/db/tx_update_job.go b/internal/odin/db/tx_update_job.go index 97c9038c..31bf1cc0 100644 --- a/internal/odin/db/tx_update_job.go +++ b/internal/odin/db/tx_update_job.go @@ -4,18 +4,16 @@ import ( "context" "time" - "github.com/adhocore/gronx" "github.com/jackc/pgx/v5/pgtype" ) type UpdateJobResultTxParams struct { - StartTime time.Time - Job Job - WorkerId int32 - Message string - Success bool - CronExpression string - Retry bool + StartTime time.Time + Job Job + WorkerId int32 + Message string + Success bool + Retry bool } type UpdateJobTxResult struct { @@ -25,36 +23,21 @@ type UpdateJobTxResult struct { func (s *SQLStore) UpdateJobResultTx(ctx context.Context, arg UpdateJobResultTxParams) (UpdateJobTxResult, error) { var updateJobTxResult UpdateJobTxResult err := s.execTx(ctx, func(q *Queries) error { - if arg.CronExpression != "" { - nextRunAt, err := gronx.NextTickAfter(arg.CronExpression, arg.StartTime, true) - if err != nil { - return err - } - err = q.UpdateJobSchedule(ctx, UpdateJobScheduleParams{ - ID: arg.Job.ID, - LastScheduledAt: pgtype.Timestamptz{Time: arg.StartTime, Valid: true}, - NextRunAt: pgtype.Timestamptz{Time: nextRunAt, Valid: true}, - }) + if arg.Success { + err := q.UpdateJobCompleted(ctx, arg.Job.ID) if err != nil { return err } } else { - if arg.Success { - err := q.UpdateJobCompleted(ctx, arg.Job.ID) + if !arg.Retry { + err := q.CancelJob(ctx, arg.Job.ID) if err != nil { return err } } else { - if !arg.Retry { - err := q.CancelJob(ctx, arg.Job.ID) - if err != nil { - return err - } - } else { - err := q.RetryJob(ctx, arg.Job.ID) - if err != nil { - return err - } + err := q.RetryJob(ctx, arg.Job.ID) + if err != nil { + return err } } } diff --git a/internal/odin/provider/docker/execute.go b/internal/odin/provider/docker/execute.go index a569ab60..c046ffb4 100644 --- a/internal/odin/provider/docker/execute.go +++ b/internal/odin/provider/docker/execute.go @@ -228,13 +228,12 @@ func (d *DockerProvider) updateJob(ctx context.Context, job *db.Job, startTime t retry = false } if _, err := d.queries.UpdateJobResultTx(ctx, db.UpdateJobResultTxParams{ - StartTime: startTime, - Job: *job, - Message: message, - Success: success, - Retry: retry, - WorkerId: d.workerId, - CronExpression: job.CronExpression.String, + StartTime: startTime, + Job: *job, + Message: message, + Success: success, + Retry: retry, + WorkerId: d.workerId, }); err != nil { return err } diff --git a/internal/odin/provider/system/execute.go b/internal/odin/provider/system/execute.go index edab1e24..8609fdb3 100644 --- a/internal/odin/provider/system/execute.go +++ b/internal/odin/provider/system/execute.go @@ -154,13 +154,17 @@ func (s *SystemProvider) writeFiles(ctx context.Context, dir string, job db.Job) } func (s *SystemProvider) updateJob(ctx context.Context, job *db.Job, startTime time.Time, message string, success bool) error { + retry := true + if job.Retries.Int32+1 >= job.MaxRetries.Int32 || success { + retry = false + } if _, err := s.queries.UpdateJobResultTx(ctx, db.UpdateJobResultTxParams{ - StartTime: startTime, - Job: *job, - Message: message, - Success: success, - WorkerId: s.workerId, - CronExpression: job.CronExpression.String, + StartTime: startTime, + Job: *job, + Message: message, + Success: success, + WorkerId: s.workerId, + Retry: retry, }); err != nil { return err } diff --git a/internal/odin/services/execution/execute.go b/internal/odin/services/execution/execute.go index 82d9d47b..9958d794 100644 --- a/internal/odin/services/execution/execute.go +++ b/internal/odin/services/execution/execute.go @@ -122,9 +122,6 @@ func (s *ExecutionService) AddJob(ctx context.Context, req *api.ExecutionRequest jobParams.ProgrammingLanguage = req.Language jobParams.MaxRetries = req.MaxRetries.Value jobParams.Path = execReq.File.Name - if req.CronExpression.Set { - jobParams.CronExpression = req.CronExpression.Value - } hash := calculateHash(jobParams.Code, jobParams.ProgrammingLanguage, jobParams.Flake, jobParams.Path) jobParams.Hash = hash diff --git a/internal/odin/worker/worker.go b/internal/odin/worker/worker.go index 85bdb4c4..4d974a9d 100644 --- a/internal/odin/worker/worker.go +++ b/internal/odin/worker/worker.go @@ -63,19 +63,18 @@ func GetWorker(ctx context.Context, name string, envConfig *config.EnvConfig, ne queries: queries, envConfig: envConfig, logger: logger, - tp: tp, - mp: mp, + tp: tp, // trace provider + mp: mp, // metric provider otelShutdown: otelShutdown, } workerInfo, err := readWorkerInfo(envConfig.ODIN_WORKER_INFO_FILE, logger) if err != nil { - logger.Err(err).Msg("Failed to read worker info") switch err.(type) { case *WorkerInfoNotFoundError: - logger.Info().Msgf("Creating new worker") if name == "" { name = namesgenerator.GetRandomName(0) } + wrkr.Name = name wrkr.ID, err = wrkr.upsertWorker(ctx, name) if err != nil { logger.Err(err).Msg("Failed to create worker") @@ -85,7 +84,8 @@ func GetWorker(ctx context.Context, name string, envConfig *config.EnvConfig, ne } } if wrkr.ID == 0 && workerInfo != nil { - logger.Info().Msgf("Found worker info") + logger.Info().Str("workerName", workerInfo.Name).Int("workerID", workerInfo.ID).Msgf("Found worker info") + wrkr.Name = workerInfo.Name wrkr.ID, err = wrkr.upsertWorker(ctx, workerInfo.Name) if err != nil { logger.Err(err).Msg("Failed to get worker") @@ -96,7 +96,7 @@ func GetWorker(ctx context.Context, name string, envConfig *config.EnvConfig, ne logger.Err(err).Msg("Failed to get provider") } wrkr.provider = prvdr - logger.Info().Msgf("Starting worker %d", wrkr.ID) + logger.Info().Msgf("Starting worker %d with name %s", wrkr.ID, wrkr.Name) err = writeWorkerInfo(envConfig.ODIN_WORKER_INFO_FILE, wrkr) if err != nil { @@ -123,6 +123,7 @@ func (w *Worker) upsertWorker(ctx context.Context, name string) (int, error) { } func (w *Worker) Run(ctx context.Context, wg *sync.WaitGroup) error { + defer wg.Done() defer func() { var err error @@ -139,8 +140,16 @@ func (w *Worker) Run(ctx context.Context, wg *sync.WaitGroup) error { span.AddEvent("Acquiring lock on worker info") infLock := flock.New(w.envConfig.ODIN_WORKER_INFO_FILE) + locked, err := infLock.TryLock() + if err != nil { + w.logger.Err(err).Msg("Failed to acquire lock on worker info") + return err + } + if !locked { + w.logger.Info().Msg("Worker: failed to acquire lock on worker info, another worker is running") + return &WorkerError{Type: "Lock", Message: "Failed to acquire lock on worker info"} + } defer infLock.Unlock() - defer wg.Done() var swg concurrency.SafeWaitGroup ticker := time.NewTicker(time.Duration(w.envConfig.ODIN_WORKER_POLL_FREQ) * time.Second) for { diff --git a/oas/odin-schema.yml b/oas/odin-schema.yml index 57266bd4..07fc3011 100644 --- a/oas/odin-schema.yml +++ b/oas/odin-schema.yml @@ -380,12 +380,10 @@ components: type: string language: type: string - cron_expression: - type: string max_retries: type: integer format: int4 - default: 0 + default: 5 timeout: type: integer format: int64 diff --git a/pkg/odin/api/oas_json_gen.go b/pkg/odin/api/oas_json_gen.go index 4dd20db1..69e9ef5e 100644 --- a/pkg/odin/api/oas_json_gen.go +++ b/pkg/odin/api/oas_json_gen.go @@ -1376,12 +1376,6 @@ func (s *ExecutionRequest) encodeFields(e *jx.Encoder) { e.FieldStart("language") e.Str(s.Language) } - { - if s.CronExpression.Set { - e.FieldStart("cron_expression") - s.CronExpression.Encode(e) - } - } { if s.MaxRetries.Set { e.FieldStart("max_retries") @@ -1402,15 +1396,14 @@ func (s *ExecutionRequest) encodeFields(e *jx.Encoder) { } } -var jsonFieldsNameOfExecutionRequest = [8]string{ +var jsonFieldsNameOfExecutionRequest = [7]string{ 0: "environment", 1: "config", 2: "code", 3: "language", - 4: "cron_expression", - 5: "max_retries", - 6: "timeout", - 7: "priority", + 4: "max_retries", + 5: "timeout", + 6: "priority", } // Decode decodes ExecutionRequest from json. @@ -1467,16 +1460,6 @@ func (s *ExecutionRequest) Decode(d *jx.Decoder) error { }(); err != nil { return errors.Wrap(err, "decode field \"language\"") } - case "cron_expression": - if err := func() error { - s.CronExpression.Reset() - if err := s.CronExpression.Decode(d); err != nil { - return err - } - return nil - }(); err != nil { - return errors.Wrap(err, "decode field \"cron_expression\"") - } case "max_retries": if err := func() error { s.MaxRetries.Reset() diff --git a/pkg/odin/api/oas_schemas_gen.go b/pkg/odin/api/oas_schemas_gen.go index c5456740..e2115da5 100644 --- a/pkg/odin/api/oas_schemas_gen.go +++ b/pkg/odin/api/oas_schemas_gen.go @@ -407,14 +407,13 @@ func (s *ExecutionEnvironmentSpec) SetArgs(val OptString) { // Ref: #/components/schemas/ExecutionRequest type ExecutionRequest struct { - Environment OptExecutionRequestEnvironment `json:"environment"` - Config OptExecutionRequestConfig `json:"config"` - Code string `json:"code"` - Language string `json:"language"` - CronExpression OptString `json:"cron_expression"` - MaxRetries OptInt `json:"max_retries"` - Timeout OptInt64 `json:"timeout"` - Priority OptInt `json:"priority"` + Environment OptExecutionRequestEnvironment `json:"environment"` + Config OptExecutionRequestConfig `json:"config"` + Code string `json:"code"` + Language string `json:"language"` + MaxRetries OptInt `json:"max_retries"` + Timeout OptInt64 `json:"timeout"` + Priority OptInt `json:"priority"` } // GetEnvironment returns the value of Environment. @@ -437,11 +436,6 @@ func (s *ExecutionRequest) GetLanguage() string { return s.Language } -// GetCronExpression returns the value of CronExpression. -func (s *ExecutionRequest) GetCronExpression() OptString { - return s.CronExpression -} - // GetMaxRetries returns the value of MaxRetries. func (s *ExecutionRequest) GetMaxRetries() OptInt { return s.MaxRetries @@ -477,11 +471,6 @@ func (s *ExecutionRequest) SetLanguage(val string) { s.Language = val } -// SetCronExpression sets the value of CronExpression. -func (s *ExecutionRequest) SetCronExpression(val OptString) { - s.CronExpression = val -} - // SetMaxRetries sets the value of MaxRetries. func (s *ExecutionRequest) SetMaxRetries(val OptInt) { s.MaxRetries = val