From 82097d69d047f82ac3120d68be026b73a6a289f7 Mon Sep 17 00:00:00 2001 From: Julien Leicher Date: Mon, 7 Oct 2024 10:17:30 +0200 Subject: [PATCH] chore: remove scheduled_jobs.resource_id Use SQLite JSON functions to extract needed informations. --- cmd/serve/front/src/lib/resources/jobs.ts | 1 - .../front/src/routes/(main)/jobs/+page.svelte | 1 - .../deployment/app/cleanup_app/cleanup_app.go | 5 +-- .../app/cleanup_app/cleanup_app_test.go | 6 +-- .../cleanup_app/on_app_cleanup_requested.go | 33 ++++++++------- .../app/cleanup_target/cleanup_target.go | 7 ++-- .../app/cleanup_target/on_job_dismissed.go | 6 ++- .../app/configure_target/configure_target.go | 7 ++-- internal/deployment/app/deploy/deploy.go | 6 +-- ...728345059_add_app_cleanup_needed_on.up.sql | 32 ++++++++++++--- pkg/bus/embedded/get_jobs/get_jobs.go | 1 - pkg/bus/embedded/runner_test.go | 22 +++++----- pkg/bus/message.go | 3 +- pkg/bus/message_test.go | 5 +-- pkg/bus/scheduler.go | 2 +- pkg/bus/sqlite/jobs.go | 40 ++++++++++--------- .../1727345059_nullable_policy.up.sql | 31 ++++++++++++-- 17 files changed, 126 insertions(+), 82 deletions(-) diff --git a/cmd/serve/front/src/lib/resources/jobs.ts b/cmd/serve/front/src/lib/resources/jobs.ts index 32f0b0c..12d385d 100644 --- a/cmd/serve/front/src/lib/resources/jobs.ts +++ b/cmd/serve/front/src/lib/resources/jobs.ts @@ -4,7 +4,6 @@ import type { Paginated } from '$lib/pagination'; export type Job = { id: string; - resource_id: string; group: string; message_name: string; message_data: string; diff --git a/cmd/serve/front/src/routes/(main)/jobs/+page.svelte b/cmd/serve/front/src/routes/(main)/jobs/+page.svelte index 9ef633a..0d5b2e1 100644 --- a/cmd/serve/front/src/routes/(main)/jobs/+page.svelte +++ b/cmd/serve/front/src/routes/(main)/jobs/+page.svelte @@ -48,7 +48,6 @@ {:else if value === 'resource'}
{translateMessageName(item.message_name)}
-
{item.resource_id}
{/if} diff --git a/internal/deployment/app/cleanup_app/cleanup_app.go b/internal/deployment/app/cleanup_app/cleanup_app.go index 99248ac..1807a14 100644 --- a/internal/deployment/app/cleanup_app/cleanup_app.go +++ b/internal/deployment/app/cleanup_app/cleanup_app.go @@ -25,9 +25,8 @@ type Command struct { To time.Time `json:"to"` } -func (Command) Name_() string { return "deployment.command.cleanup_app" } -func (c Command) ResourceID() string { return c.AppID } -func (c Command) Group() string { return bus.Group(c.AppID, c.Environment, c.TargetID) } +func (Command) Name_() string { return "deployment.command.cleanup_app" } +func (c Command) Group() string { return bus.Group(c.AppID, c.Environment, c.TargetID) } func Handler( reader domain.TargetsReader, diff --git a/internal/deployment/app/cleanup_app/cleanup_app_test.go b/internal/deployment/app/cleanup_app/cleanup_app_test.go index 150faf6..4a3e431 100644 --- a/internal/deployment/app/cleanup_app/cleanup_app_test.go +++ b/internal/deployment/app/cleanup_app/cleanup_app_test.go @@ -108,7 +108,7 @@ func Test_CleanupApp(t *testing.T) { var provider mockProvider user := authfixture.User() target := fixture.Target(fixture.WithTargetCreatedBy(user.ID())) - target.Configured(target.CurrentVersion(), nil, nil) + assert.Nil(t, target.Configured(target.CurrentVersion(), nil, nil)) assert.Nil(t, target.RequestDelete(false, "uid")) handler, ctx := arrange(t, &provider, @@ -146,7 +146,7 @@ func Test_CleanupApp(t *testing.T) { var provider mockProvider user := authfixture.User() target := fixture.Target(fixture.WithTargetCreatedBy(user.ID())) - target.Configured(target.CurrentVersion(), nil, errors.New("configuration failed")) + assert.Nil(t, target.Configured(target.CurrentVersion(), nil, errors.New("configuration failed"))) app := fixture.App(fixture.WithAppCreatedBy(user.ID()), fixture.WithEnvironmentConfig( domain.NewEnvironmentConfig(target.ID()), @@ -181,7 +181,7 @@ func Test_CleanupApp(t *testing.T) { var provider mockProvider user := authfixture.User() target := fixture.Target(fixture.WithTargetCreatedBy(user.ID())) - target.Configured(target.CurrentVersion(), nil, nil) + assert.Nil(t, target.Configured(target.CurrentVersion(), nil, nil)) app := fixture.App(fixture.WithAppCreatedBy(user.ID()), fixture.WithEnvironmentConfig( domain.NewEnvironmentConfig(target.ID()), diff --git a/internal/deployment/app/cleanup_app/on_app_cleanup_requested.go b/internal/deployment/app/cleanup_app/on_app_cleanup_requested.go index 726a091..69bc6d5 100644 --- a/internal/deployment/app/cleanup_app/on_app_cleanup_requested.go +++ b/internal/deployment/app/cleanup_app/on_app_cleanup_requested.go @@ -13,22 +13,21 @@ func OnAppCleanupRequestedHandler(scheduler bus.Scheduler) bus.SignalHandler[dom return func(ctx context.Context, evt domain.AppCleanupRequested) error { now := time.Now().UTC() - if err := scheduler.Queue(ctx, Command{ - AppID: string(evt.ID), - Environment: string(domain.Production), - TargetID: string(evt.ProductionConfig.Target()), - From: evt.ProductionConfig.Version(), - To: now, - }); err != nil { - return err - } - - return scheduler.Queue(ctx, Command{ - AppID: string(evt.ID), - Environment: string(domain.Staging), - TargetID: string(evt.StagingConfig.Target()), - From: evt.StagingConfig.Version(), - To: now, - }) + return scheduler.Queue(ctx, + Command{ + AppID: string(evt.ID), + Environment: string(domain.Production), + TargetID: string(evt.ProductionConfig.Target()), + From: evt.ProductionConfig.Version(), + To: now, + }, + Command{ + AppID: string(evt.ID), + Environment: string(domain.Staging), + TargetID: string(evt.StagingConfig.Target()), + From: evt.StagingConfig.Version(), + To: now, + }, + ) } } diff --git a/internal/deployment/app/cleanup_target/cleanup_target.go b/internal/deployment/app/cleanup_target/cleanup_target.go index 9a5f8fb..d6b8e70 100644 --- a/internal/deployment/app/cleanup_target/cleanup_target.go +++ b/internal/deployment/app/cleanup_target/cleanup_target.go @@ -14,12 +14,11 @@ import ( type Command struct { bus.AsyncCommand - ID string `json:"id"` + ID string `json:"target_id"` } -func (Command) Name_() string { return "deployment.command.cleanup_target" } -func (c Command) ResourceID() string { return c.ID } -func (c Command) Group() string { return c.ID } +func (Command) Name_() string { return "deployment.command.cleanup_target" } +func (c Command) Group() string { return c.ID } func Handler( reader domain.TargetsReader, diff --git a/internal/deployment/app/cleanup_target/on_job_dismissed.go b/internal/deployment/app/cleanup_target/on_job_dismissed.go index 01eb9f0..f87eec9 100644 --- a/internal/deployment/app/cleanup_target/on_job_dismissed.go +++ b/internal/deployment/app/cleanup_target/on_job_dismissed.go @@ -15,11 +15,13 @@ func OnJobDismissedHandler( writer domain.TargetsWriter, ) bus.SignalHandler[embedded.JobDismissed] { return func(ctx context.Context, evt embedded.JobDismissed) error { - if _, isCleanupJob := evt.Command.(Command); !isCleanupJob { + cmd, isCleanupJob := evt.Command.(Command) + + if !isCleanupJob { return nil } - target, err := reader.GetByID(ctx, domain.TargetID(evt.Command.ResourceID())) + target, err := reader.GetByID(ctx, domain.TargetID(cmd.ID)) if err != nil { // Target deleted, no need to go further diff --git a/internal/deployment/app/configure_target/configure_target.go b/internal/deployment/app/configure_target/configure_target.go index 7ee5e81..569a1b0 100644 --- a/internal/deployment/app/configure_target/configure_target.go +++ b/internal/deployment/app/configure_target/configure_target.go @@ -14,13 +14,12 @@ import ( type Command struct { bus.AsyncCommand - ID string `json:"id"` + ID string `json:"target_id"` Version time.Time `json:"version"` } -func (Command) Name_() string { return "deployment.command.configure_target" } -func (c Command) ResourceID() string { return c.ID } -func (c Command) Group() string { return c.ID } +func (Command) Name_() string { return "deployment.command.configure_target" } +func (c Command) Group() string { return c.ID } func Handler( reader domain.TargetsReader, diff --git a/internal/deployment/app/deploy/deploy.go b/internal/deployment/app/deploy/deploy.go index 6fade1e..450a15d 100644 --- a/internal/deployment/app/deploy/deploy.go +++ b/internal/deployment/app/deploy/deploy.go @@ -3,7 +3,6 @@ package deploy import ( "context" "errors" - "strconv" "github.com/YuukanOO/seelf/internal/deployment/domain" "github.com/YuukanOO/seelf/pkg/apperr" @@ -20,9 +19,8 @@ type Command struct { TargetID string `json:"target_id"` } -func (Command) Name_() string { return "deployment.command.deploy" } -func (c Command) ResourceID() string { return c.AppID + "-" + strconv.Itoa(c.DeploymentNumber) } -func (c Command) Group() string { return bus.Group(c.AppID, c.Environment, c.TargetID) } +func (Command) Name_() string { return "deployment.command.deploy" } +func (c Command) Group() string { return bus.Group(c.AppID, c.Environment, c.TargetID) } // Handle the deployment process. // If an unexpected error occurs during this process, it uses the bus.PreserveOrder function diff --git a/internal/deployment/infra/sqlite/migrations/1728345059_add_app_cleanup_needed_on.up.sql b/internal/deployment/infra/sqlite/migrations/1728345059_add_app_cleanup_needed_on.up.sql index 4ff1094..dc317d9 100644 --- a/internal/deployment/infra/sqlite/migrations/1728345059_add_app_cleanup_needed_on.up.sql +++ b/internal/deployment/infra/sqlite/migrations/1728345059_add_app_cleanup_needed_on.up.sql @@ -12,18 +12,40 @@ ALTER TABLE apps ADD version DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP; ALTER TABLE registries ADD version DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP; ALTER TABLE targets ADD version DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP; -ALTER TABLE apps ADD history TEXT NOT NULL DEFAULT '{}'; +ALTER TABLE apps ADD history JSON NOT NULL DEFAULT '{}'; UPDATE apps SET history = '{"production": ["' || production_target || '"], "staging": ["'|| staging_target ||'"]}'; -- Since the group has changed for configure and cleanup, just update those jobs. UPDATE scheduled_jobs SET [group] = resource_id -WHERE message_name IN ('deployment.command.configure_target', 'deployment.command.cleanup_target'); + ,message_data = '{"target_id": "' || resource_id || '"}' +WHERE message_name = 'deployment.command.cleanup_target'; + +UPDATE scheduled_jobs +SET + [group] = resource_id + ,message_data = '{"target_id": "' || resource_id || '", "version": "' || (message_data ->> '$.version') || '"}' +WHERE message_name = 'deployment.command.configure_target'; -- Since those messages no longer exists. DELETE FROM scheduled_jobs WHERE message_name IN ('deployment.command.delete_target', 'deployment.command.delete_app'); +DROP TRIGGER IF EXISTS on_deployment_failed_cleanup_jobs; + +-- When a deployment is marked as failed, remove all pending jobs for that resource. +-- This is to avoid running jobs that are no longer needed. +CREATE TRIGGER IF NOT EXISTS on_deployment_failed_cleanup_jobs AFTER UPDATE ON deployments + WHEN OLD.state_status != NEW.state_status AND NEW.state_status = 2 -- Only when the deployment goes to the failed state +BEGIN + DELETE FROM scheduled_jobs + WHERE + message_name = 'deployment.command.deploy' + AND message_data ->> '$.app_id' = NEW.app_id + AND message_data ->> '$.deployment_number' = NEW.deployment_number + AND retrieved = 0; +END; + -- When a target is configured, no need to keep old configure jobs since they are outdated. CREATE TRIGGER IF NOT EXISTS on_target_configure_remove_outdated_jobs BEFORE INSERT ON scheduled_jobs @@ -31,7 +53,7 @@ WHEN NEW.message_name = 'deployment.command.configure_target' BEGIN DELETE FROM scheduled_jobs WHERE - resource_id = NEW.resource_id - AND message_name = 'deployment.command.configure_target' + message_name = 'deployment.command.configure_target' + AND message_data ->> '$.target_id' = NEW.message_data ->> '$.target_id' AND retrieved = 0; -END \ No newline at end of file +END; \ No newline at end of file diff --git a/pkg/bus/embedded/get_jobs/get_jobs.go b/pkg/bus/embedded/get_jobs/get_jobs.go index 28a8c2b..cdb2612 100644 --- a/pkg/bus/embedded/get_jobs/get_jobs.go +++ b/pkg/bus/embedded/get_jobs/get_jobs.go @@ -11,7 +11,6 @@ import ( type ( Job struct { ID string `json:"id"` - ResourceID string `json:"resource_id"` Group string `json:"group"` MessageName string `json:"message_name"` MessageData string `json:"message_data"` diff --git a/pkg/bus/embedded/runner_test.go b/pkg/bus/embedded/runner_test.go index 6156eff..7cec068 100644 --- a/pkg/bus/embedded/runner_test.go +++ b/pkg/bus/embedded/runner_test.go @@ -126,17 +126,15 @@ type returnCommand struct { err error } -func (r returnCommand) Name_() string { return "returnCommand" } -func (r returnCommand) ResourceID() string { return "" } -func (r returnCommand) Group() string { return "" } +func (r returnCommand) Name_() string { return "returnCommand" } +func (r returnCommand) Group() string { return "" } type unhandledCommand struct { bus.AsyncCommand } -func (u unhandledCommand) Name_() string { return "unhandledCommand" } -func (u unhandledCommand) ResourceID() string { return "" } -func (u unhandledCommand) Group() string { return "" } +func (u unhandledCommand) Name_() string { return "unhandledCommand" } +func (u unhandledCommand) Group() string { return "" } var ( _ embedded.JobsStore = (*adapter)(nil) @@ -145,12 +143,12 @@ var ( type job struct { id string - msg bus.AsyncRequest + cmd bus.AsyncRequest err error } func (j *job) ID() string { return j.id } -func (j *job) Command() bus.AsyncRequest { return j.msg } +func (j *job) Command() bus.AsyncRequest { return j.cmd } type adapter struct { wg sync.WaitGroup @@ -160,9 +158,13 @@ type adapter struct { done []*job } -func (a *adapter) Queue(ctx context.Context, msg bus.AsyncRequest) error { +func (a *adapter) Queue(ctx context.Context, requests ...bus.AsyncRequest) error { a.wg.Add(1) - a.jobs = append(a.jobs, &job{id: id.New[string](), msg: msg}) + + for _, req := range requests { + a.jobs = append(a.jobs, &job{id: id.New[string](), cmd: req}) + } + return nil } diff --git a/pkg/bus/message.go b/pkg/bus/message.go index 411d9ee..acd76c6 100644 --- a/pkg/bus/message.go +++ b/pkg/bus/message.go @@ -65,8 +65,7 @@ type ( AsyncRequest interface { MutateRequest TypedRequest[AsyncResult] - ResourceID() string // ID of the main resource processed by the request - Group() string // Work group for this request, at most one job per group is processed at any given time + Group() string // Work group for this request, at most one job per group is processed at any given time } // Request to query the system. diff --git a/pkg/bus/message_test.go b/pkg/bus/message_test.go index d851288..b997248 100644 --- a/pkg/bus/message_test.go +++ b/pkg/bus/message_test.go @@ -181,9 +181,8 @@ type asyncCommand struct { SomeValue int } -func (asyncCommand) Name_() string { return "AsyncCommand" } -func (asyncCommand) ResourceID() string { return "" } -func (asyncCommand) Group() string { return "" } +func (asyncCommand) Name_() string { return "AsyncCommand" } +func (asyncCommand) Group() string { return "" } type getQuery struct { bus.Query[int] diff --git a/pkg/bus/scheduler.go b/pkg/bus/scheduler.go index 4a1e60c..889785a 100644 --- a/pkg/bus/scheduler.go +++ b/pkg/bus/scheduler.go @@ -5,5 +5,5 @@ import "context" // Enable scheduled dispatching of a message. type Scheduler interface { // Queue a request to be dispatched asynchronously at a later time. - Queue(context.Context, AsyncRequest) error + Queue(context.Context, ...AsyncRequest) error } diff --git a/pkg/bus/sqlite/jobs.go b/pkg/bus/sqlite/jobs.go index c528b2b..07d9211 100644 --- a/pkg/bus/sqlite/jobs.go +++ b/pkg/bus/sqlite/jobs.go @@ -58,7 +58,6 @@ func (s *JobsStore) GetAllJobs(ctx context.Context, query get_jobs.Query) (stora return builder. Select[get_jobs.Job](` id - ,resource_id ,[group] ,message_name ,message_data @@ -124,27 +123,33 @@ func (s *JobsStore) DismissJob(ctx context.Context, cmd dismiss_job.Command) (bu func (s *JobsStore) Queue( ctx context.Context, - msg bus.AsyncRequest, + requests ...bus.AsyncRequest, ) error { now := time.Now().UTC() - msgData, err := storage.ValueJSON(msg) - if err != nil { - return err + for _, req := range requests { + cmdData, err := storage.ValueJSON(req) + + if err != nil { + return err + } + + if err = builder. + Insert("scheduled_jobs", builder.Values{ + "id": id.New[string](), + "[group]": req.Group(), + "message_name": req.Name_(), + "message_data": cmdData, + "queued_at": now, + "not_before": now, + "retrieved": false, + }). + Exec(s.db, ctx); err != nil { + return err + } } - return builder. - Insert("scheduled_jobs", builder.Values{ - "id": id.New[string](), - "resource_id": msg.ResourceID(), - "[group]": msg.Group(), - "message_name": msg.Name_(), - "message_data": msgData, - "queued_at": now, - "not_before": now, - "retrieved": false, - }). - Exec(s.db, ctx) + return nil } func (s *JobsStore) GetNextPendingJobs(ctx context.Context) ([]embedded.Job, error) { @@ -226,7 +231,6 @@ func jobMapper(scanner storage.Scanner) (embedded.Job, error) { func jobQueryMapper(scanner storage.Scanner) (j get_jobs.Job, err error) { err = scanner.Scan( &j.ID, - &j.ResourceID, &j.Group, &j.MessageName, &j.MessageData, diff --git a/pkg/bus/sqlite/migrations/1727345059_nullable_policy.up.sql b/pkg/bus/sqlite/migrations/1727345059_nullable_policy.up.sql index 77f94a2..d441124 100644 --- a/pkg/bus/sqlite/migrations/1727345059_nullable_policy.up.sql +++ b/pkg/bus/sqlite/migrations/1727345059_nullable_policy.up.sql @@ -1,7 +1,32 @@ -ALTER TABLE scheduled_jobs DROP COLUMN policy; +CREATE TEMP TABLE tmp_scheduled_jobs AS +SELECT * FROM scheduled_jobs; --- Make the policy column nullable since it's not used anymore. +DROP TABLE scheduled_jobs; + +-- Make the policy and resource_id columns nullable since it's not used anymore. -- Since migrations are executed by domain orders (scheduler, auth, deployment) and -- I have failed by making deployment rely on the scheduled_jobs table, I must keep -- it or else I have to update the migration (which I think is worse). -ALTER TABLE scheduled_jobs ADD policy INTEGER NULL; +CREATE TABLE scheduled_jobs +( + id TEXT NOT NULL, + resource_id TEXT NULL, + [group] TEXT NOT NULL, + message_name TEXT NOT NULL, + message_data JSON NOT NULL, + queued_at DATETIME NOT NULL, + not_before DATETIME NOT NULL, + policy INTEGER NULL, + errcode TEXT NULL, + retrieved BOOLEAN NOT NULL DEFAULT false, + + CONSTRAINT pk_scheduled_jobs PRIMARY KEY(id) +); + +INSERT INTO scheduled_jobs +SELECT * FROM tmp_scheduled_jobs; + +CREATE INDEX idx_scheduled_jobs_group ON scheduled_jobs([group]); +CREATE INDEX idx_scheduled_jobs_message_name ON scheduled_jobs(message_name); + +DROP TABLE tmp_scheduled_jobs; \ No newline at end of file