Compare commits
1 Commits
ea552ec448
...
072e908cf3
Author | SHA1 | Date | |
---|---|---|---|
|
072e908cf3 |
@ -9,7 +9,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
|
|||||||
## [0.1.0] - 2025-01-18
|
## [0.1.0] - 2025-01-18
|
||||||
|
|
||||||
### Added
|
### Added
|
||||||
- move schedules to registered workers
|
|
||||||
- prune old workers
|
- prune old workers
|
||||||
- deregister worker on close
|
- deregister worker on close
|
||||||
- add worker distributor and model registry
|
- add worker distributor and model registry
|
||||||
|
@ -39,7 +39,6 @@ func NewExecutor(
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (e *Executor) DispatchEvents(ctx context.Context) error {
|
func (e *Executor) DispatchEvents(ctx context.Context) error {
|
||||||
|
|
||||||
e.logger.InfoContext(ctx, "dispatching events")
|
e.logger.InfoContext(ctx, "dispatching events")
|
||||||
|
|
||||||
start := time.Now().Add(-time.Second * 30)
|
start := time.Now().Add(-time.Second * 30)
|
||||||
@ -55,11 +54,6 @@ func (e *Executor) DispatchEvents(ctx context.Context) error {
|
|||||||
return fmt.Errorf("failed to find workers: %w", err)
|
return fmt.Errorf("failed to find workers: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
e.logger.InfoContext(ctx, "moving unattended events")
|
|
||||||
if err := e.workerscheduler.GetUnattended(ctx, registeredWorkers); err != nil {
|
|
||||||
return fmt.Errorf("failed to move unattended events: %w", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
workers, err := e.workerscheduler.GetWorkers(ctx, registeredWorkers)
|
workers, err := e.workerscheduler.GetWorkers(ctx, registeredWorkers)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("failed to find workers: %w", err)
|
return fmt.Errorf("failed to find workers: %w", err)
|
||||||
|
@ -49,7 +49,7 @@ func (s *Scheduler) Execute(ctx context.Context) error {
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
ticker := time.NewTicker(1 * time.Second)
|
ticker := time.NewTicker(5 * time.Second)
|
||||||
defer ticker.Stop()
|
defer ticker.Stop()
|
||||||
for {
|
for {
|
||||||
select {
|
select {
|
||||||
|
@ -30,4 +30,4 @@ WHERE
|
|||||||
-- name: PruneWorker :exec
|
-- name: PruneWorker :exec
|
||||||
DELETE FROM worker_register
|
DELETE FROM worker_register
|
||||||
WHERE
|
WHERE
|
||||||
heart_beat <= now() - INTERVAL '1 minutes';
|
heart_beat <= now() - INTERVAL '10 minutes';
|
||||||
|
@ -69,7 +69,7 @@ func (q *Queries) Ping(ctx context.Context) (int32, error) {
|
|||||||
const pruneWorker = `-- name: PruneWorker :exec
|
const pruneWorker = `-- name: PruneWorker :exec
|
||||||
DELETE FROM worker_register
|
DELETE FROM worker_register
|
||||||
WHERE
|
WHERE
|
||||||
heart_beat <= now() - INTERVAL '1 minutes'
|
heart_beat <= now() - INTERVAL '10 minutes'
|
||||||
`
|
`
|
||||||
|
|
||||||
func (q *Queries) PruneWorker(ctx context.Context) error {
|
func (q *Queries) PruneWorker(ctx context.Context) error {
|
||||||
|
@ -54,23 +54,3 @@ SET
|
|||||||
WHERE
|
WHERE
|
||||||
schedule_id = $1;
|
schedule_id = $1;
|
||||||
|
|
||||||
-- name: GetUnattended :many
|
|
||||||
SELECT
|
|
||||||
*
|
|
||||||
FROM
|
|
||||||
work_schedule
|
|
||||||
WHERE
|
|
||||||
worker_id NOT IN (SELECT unnest(@worker_ids::uuid[]))
|
|
||||||
AND state <> 'archived'
|
|
||||||
--AND updated_at <= now() - INTERVAL '10 minutes'
|
|
||||||
ORDER BY updated_at DESC
|
|
||||||
LIMIT @amount::integer;
|
|
||||||
|
|
||||||
-- name: UpdateSchdule :exec
|
|
||||||
UPDATE work_schedule
|
|
||||||
SET
|
|
||||||
state = 'pending'
|
|
||||||
, worker_id = $1
|
|
||||||
, updated_at = now()
|
|
||||||
WHERE
|
|
||||||
schedule_id = $2;
|
|
||||||
|
@ -14,11 +14,9 @@ type Querier interface {
|
|||||||
Archive(ctx context.Context, scheduleID uuid.UUID) error
|
Archive(ctx context.Context, scheduleID uuid.UUID) error
|
||||||
GetCurrentQueueSize(ctx context.Context, workerID uuid.UUID) (int64, error)
|
GetCurrentQueueSize(ctx context.Context, workerID uuid.UUID) (int64, error)
|
||||||
GetNext(ctx context.Context, workerID uuid.UUID) (*WorkSchedule, error)
|
GetNext(ctx context.Context, workerID uuid.UUID) (*WorkSchedule, error)
|
||||||
GetUnattended(ctx context.Context, arg *GetUnattendedParams) ([]*WorkSchedule, error)
|
|
||||||
InsertQueueItem(ctx context.Context, arg *InsertQueueItemParams) error
|
InsertQueueItem(ctx context.Context, arg *InsertQueueItemParams) error
|
||||||
Ping(ctx context.Context) (int32, error)
|
Ping(ctx context.Context) (int32, error)
|
||||||
StartProcessing(ctx context.Context, scheduleID uuid.UUID) error
|
StartProcessing(ctx context.Context, scheduleID uuid.UUID) error
|
||||||
UpdateSchdule(ctx context.Context, arg *UpdateSchduleParams) error
|
|
||||||
}
|
}
|
||||||
|
|
||||||
var _ Querier = (*Queries)(nil)
|
var _ Querier = (*Queries)(nil)
|
||||||
|
@ -68,51 +68,6 @@ func (q *Queries) GetNext(ctx context.Context, workerID uuid.UUID) (*WorkSchedul
|
|||||||
return &i, err
|
return &i, err
|
||||||
}
|
}
|
||||||
|
|
||||||
const getUnattended = `-- name: GetUnattended :many
|
|
||||||
SELECT
|
|
||||||
schedule_id, worker_id, start_run, end_run, updated_at, state
|
|
||||||
FROM
|
|
||||||
work_schedule
|
|
||||||
WHERE
|
|
||||||
worker_id NOT IN (SELECT unnest($1::uuid[]))
|
|
||||||
AND state <> 'archived'
|
|
||||||
--AND updated_at <= now() - INTERVAL '10 minutes'
|
|
||||||
ORDER BY updated_at DESC
|
|
||||||
LIMIT $2::integer
|
|
||||||
`
|
|
||||||
|
|
||||||
type GetUnattendedParams struct {
|
|
||||||
WorkerIds []uuid.UUID `json:"worker_ids"`
|
|
||||||
Amount int32 `json:"amount"`
|
|
||||||
}
|
|
||||||
|
|
||||||
func (q *Queries) GetUnattended(ctx context.Context, arg *GetUnattendedParams) ([]*WorkSchedule, error) {
|
|
||||||
rows, err := q.db.Query(ctx, getUnattended, arg.WorkerIds, arg.Amount)
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
defer rows.Close()
|
|
||||||
items := []*WorkSchedule{}
|
|
||||||
for rows.Next() {
|
|
||||||
var i WorkSchedule
|
|
||||||
if err := rows.Scan(
|
|
||||||
&i.ScheduleID,
|
|
||||||
&i.WorkerID,
|
|
||||||
&i.StartRun,
|
|
||||||
&i.EndRun,
|
|
||||||
&i.UpdatedAt,
|
|
||||||
&i.State,
|
|
||||||
); err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
items = append(items, &i)
|
|
||||||
}
|
|
||||||
if err := rows.Err(); err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
return items, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
const insertQueueItem = `-- name: InsertQueueItem :exec
|
const insertQueueItem = `-- name: InsertQueueItem :exec
|
||||||
INSERT INTO work_schedule
|
INSERT INTO work_schedule
|
||||||
(
|
(
|
||||||
@ -172,23 +127,3 @@ func (q *Queries) StartProcessing(ctx context.Context, scheduleID uuid.UUID) err
|
|||||||
_, err := q.db.Exec(ctx, startProcessing, scheduleID)
|
_, err := q.db.Exec(ctx, startProcessing, scheduleID)
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
const updateSchdule = `-- name: UpdateSchdule :exec
|
|
||||||
UPDATE work_schedule
|
|
||||||
SET
|
|
||||||
state = 'pending'
|
|
||||||
, worker_id = $1
|
|
||||||
, updated_at = now()
|
|
||||||
WHERE
|
|
||||||
schedule_id = $2
|
|
||||||
`
|
|
||||||
|
|
||||||
type UpdateSchduleParams struct {
|
|
||||||
WorkerID uuid.UUID `json:"worker_id"`
|
|
||||||
ScheduleID uuid.UUID `json:"schedule_id"`
|
|
||||||
}
|
|
||||||
|
|
||||||
func (q *Queries) UpdateSchdule(ctx context.Context, arg *UpdateSchduleParams) error {
|
|
||||||
_, err := q.db.Exec(ctx, updateSchdule, arg.WorkerID, arg.ScheduleID)
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
@ -177,47 +177,3 @@ func (w *WorkScheduler) Archive(ctx context.Context, scheduleID uuid.UUID) error
|
|||||||
|
|
||||||
return repo.Archive(ctx, scheduleID)
|
return repo.Archive(ctx, scheduleID)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (w *WorkScheduler) GetUnattended(ctx context.Context, registeredWorkers *worker.Workers) error {
|
|
||||||
if len(registeredWorkers.Instances) == 0 {
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
repo := repositories.New(w.db)
|
|
||||||
|
|
||||||
schedules, err := repo.GetUnattended(ctx, &repositories.GetUnattendedParams{
|
|
||||||
WorkerIds: workerIDs(registeredWorkers),
|
|
||||||
Amount: 100,
|
|
||||||
})
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("failed to get unattended workers: %w", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
for i, schedule := range schedules {
|
|
||||||
worker := registeredWorkers.Instances[i%len(registeredWorkers.Instances)].WorkerID
|
|
||||||
|
|
||||||
w.logger.InfoContext(ctx, "dispatching schedule for worker", "worker", worker, "schedule", schedule.ScheduleID)
|
|
||||||
|
|
||||||
if err := repo.UpdateSchdule(
|
|
||||||
ctx,
|
|
||||||
&repositories.UpdateSchduleParams{
|
|
||||||
WorkerID: worker,
|
|
||||||
ScheduleID: schedule.ScheduleID,
|
|
||||||
},
|
|
||||||
); err != nil {
|
|
||||||
return fmt.Errorf("failed to update schedule: %w", err)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func workerIDs(registeredWorkers *worker.Workers) []uuid.UUID {
|
|
||||||
uuids := make([]uuid.UUID, 0, len(registeredWorkers.Instances))
|
|
||||||
|
|
||||||
for _, registeredWorker := range registeredWorkers.Instances {
|
|
||||||
uuids = append(uuids, registeredWorker.WorkerID)
|
|
||||||
}
|
|
||||||
|
|
||||||
return uuids
|
|
||||||
}
|
|
||||||
|
Loading…
x
Reference in New Issue
Block a user