From 5cac82ea681c4de1d001113e3df2d4856922b753 Mon Sep 17 00:00:00 2001 From: Douglas-Lee Date: Tue, 4 Aug 2026 16:44:55 +0800 Subject: [PATCH 1/2] feat(retention): data retention --- app/app.go | 22 ++- config.yml | 15 ++ config/config.go | 8 + config/modules/retention.go | 13 ++ db/dao/attempt_dao.go | 19 +++ db/dao/daos.go | 2 + db/dao/event_dao.go | 19 +++ db/migrations/1785833137_retention.down.sql | 4 + db/migrations/1785833137_retention.up.sql | 4 + services/retention/retention.go | 167 ++++++++++++++++++++ test/cmd/db_test.go | 8 +- test/retention/retention_test.go | 126 +++++++++++++++ utils/fmt.go | 18 +++ 13 files changed, 421 insertions(+), 4 deletions(-) create mode 100644 config/modules/retention.go create mode 100644 db/migrations/1785833137_retention.down.sql create mode 100644 db/migrations/1785833137_retention.up.sql create mode 100644 services/retention/retention.go create mode 100644 test/retention/retention_test.go create mode 100644 utils/fmt.go diff --git a/app/app.go b/app/app.go index eda85977..d54b4d6c 100644 --- a/app/app.go +++ b/app/app.go @@ -42,6 +42,7 @@ import ( "github.com/webhookx-io/webhookx/services" "github.com/webhookx-io/webhookx/services/distributed" "github.com/webhookx-io/webhookx/services/eventbus" + "github.com/webhookx-io/webhookx/services/retention" "github.com/webhookx-io/webhookx/services/schedule" "github.com/webhookx-io/webhookx/services/task" tracingservice "github.com/webhookx-io/webhookx/services/tracing" @@ -198,6 +199,16 @@ func (app *Application) initialize() error { return err } + if err := app.initRetention(&cfg.Retention, services); err != nil { + return err + } + + return nil +} + +func (app *Application) initRetention(cfg *modules.RetentionConfig, services *services.Services) error { + retentionService := retention.NewRetentionService(*cfg, app.db, app.log, services.Scheduler) + app.registerService(retentionService) return nil } @@ -549,6 +560,10 @@ func (app *Application) Start() error { app.log.Info("anonymous reports is disabled") } + if app.cfg.Retention.Enabled && app.cfg.Role.IsDataPlane() { + app.log.Warnf("retention configuration is ignored on data-plane nodes") + } + services := []services.Service{ app.getService("eventbus"), app.getService("metrics"), @@ -558,6 +573,11 @@ func (app *Application) Start() error { app.getService("status"), app.getService("schedule"), } + + if !app.cfg.Role.IsDataPlane() { + services = append(services, app.getService("retention")) + } + return startServices(services...) } @@ -582,7 +602,7 @@ func (app *Application) stop(ctx context.Context) error { errs = append(errs, stopServices(ctx, 0, app.getService("eventbus"))) errs = append(errs, stopServices(ctx, 0, app.getService("worker"))) errs = append(errs, stopServices(ctx, time.Second*5, app.getService("metrics"), app.getService("tracing"))) - errs = append(errs, stopServices(ctx, 0, app.getService("schedule"))) + errs = append(errs, stopServices(ctx, 0, app.getService("schedule"), app.getService("retention"))) errs = append(errs, app.db.Close()) app.log.Info("exit") diff --git a/config.yml b/config.yml index 38d190ee..db575e04 100644 --- a/config.yml +++ b/config.yml @@ -263,3 +263,18 @@ tracing: # kubernetes: # role: # Vault role bound to the Kubernetes service account. # token_path: # Path to JWT token file. + + +#------------------------------------------------------------------------------ +# Data Retention +# +# Data Retention automatically remove expired objects from the database +# periodically to reduce storage usage and improve query performance. +# +# This service only runs on control-plane or standalone nodes. +# Data-plane nodes ignore this configuration. +#------------------------------------------------------------------------------ +#retention: +# enabled: true # Whether to enable data retention. Defaults to false. +# events: 30 # Retention period (in days) for events. Set to 0 to disable event cleanup. +# attempts: 60 # Retention period (in days) for delivery attempts. Set to 0 to disable attempt cleanup. diff --git a/config/config.go b/config/config.go index 50926b9e..f5a12f84 100644 --- a/config/config.go +++ b/config/config.go @@ -19,6 +19,10 @@ const ( RoleDPProxy Role = "dp_proxy" ) +func (r Role) IsDataPlane() bool { + return r == RoleDPWorker || r == RoleDPProxy +} + var _ types.Config = &Config{} // Config Configuration @@ -37,6 +41,7 @@ type Config struct { Role Role `yaml:"role" json:"role" envconfig:"ROLE" default:"standalone"` AnonymousReports bool `yaml:"anonymous_reports" json:"anonymous_reports" envconfig:"ANONYMOUS_REPORTS" default:"true"` Secret modules.SecretConfig `yaml:"secret" json:"secret" envconfig:"SECRET"` + Retention modules.RetentionConfig `yaml:"retention" json:"retention" envconfig:"RETENTION"` } func (cfg *Config) PostProcess() error { @@ -106,6 +111,9 @@ func (cfg Config) Validate() error { if err := cfg.Secret.Validate(); err != nil { return err } + if err := cfg.Retention.Validate(); err != nil { + return err + } return nil } diff --git a/config/modules/retention.go b/config/modules/retention.go new file mode 100644 index 00000000..ea57259c --- /dev/null +++ b/config/modules/retention.go @@ -0,0 +1,13 @@ +package modules + +type RetentionConfig struct { + BaseConfig + + Enabled bool `yaml:"enabled" json:"enabled" default:"false"` + Events uint32 `yaml:"events" json:"events" default:"0"` + Attempts uint32 `yaml:"attempts" json:"attempts" default:"0"` +} + +func (cfg RetentionConfig) Validate() error { + return nil +} diff --git a/db/dao/attempt_dao.go b/db/dao/attempt_dao.go index 21efe23f..8e8f61eb 100644 --- a/db/dao/attempt_dao.go +++ b/db/dao/attempt_dao.go @@ -93,6 +93,25 @@ func (dao *attemptDao) ListUnqueuedForUpdate(ctx context.Context, maxScheduledAt return } +func (dao *attemptDao) DeleteTTL(ctx context.Context, ttl time.Duration, limit int) (int64, error) { + ctx, span := dao.trace(ctx, fmt.Sprintf("dao.%s.delete_ttl", dao.opts.Table)) + defer span.End() + + sql := fmt.Sprintf( + `DELETE FROM attempts WHERE id IN ( + SELECT id FROM attempts WHERE created_at < now() - INTERVAL '%s' ORDER BY created_at ASC LIMIT $1 + )`, + fmt.Sprintf("%d hours", int(ttl.Hours())), + ) + + dao.debugSQL(sql, []interface{}{limit}) + res, err := dao.DB(ctx).ExecContext(ctx, sql, limit) + if err != nil { + return 0, err + } + return res.RowsAffected() +} + type AttemptQuery struct { Query diff --git a/db/dao/daos.go b/db/dao/daos.go index 6a24f323..fe163bd9 100644 --- a/db/dao/daos.go +++ b/db/dao/daos.go @@ -36,6 +36,7 @@ type EventDAO interface { BaseDAO[entities.Event] BatchInsertIgnoreConflict(ctx context.Context, events []*entities.Event) ([]string, error) ListExistingUniqueIDs(ctx context.Context, uniques []string) ([]string, error) + DeleteTTL(ctx context.Context, ttl time.Duration, limit int) (int64, error) } type AttemptDAO interface { @@ -44,6 +45,7 @@ type AttemptDAO interface { UpdateErrorCode(ctx context.Context, id string, status entities.AttemptStatus, code entities.AttemptErrorCode) error UpdateDelivery(ctx context.Context, result *AttemptResult) error ListUnqueuedForUpdate(ctx context.Context, maxScheduledAt time.Time, limit int) (list []*entities.Attempt, err error) + DeleteTTL(ctx context.Context, ttl time.Duration, limit int) (int64, error) } type SourceDAO interface { diff --git a/db/dao/event_dao.go b/db/dao/event_dao.go index 59afd252..21f22222 100644 --- a/db/dao/event_dao.go +++ b/db/dao/event_dao.go @@ -3,6 +3,7 @@ package dao import ( "context" "fmt" + "time" sq "github.com/Masterminds/squirrel" "github.com/jmoiron/sqlx" @@ -69,6 +70,24 @@ func (dao *eventDao) BatchInsertIgnoreConflict(ctx context.Context, events []*en return inserteds, rows.Err() } +func (dao *eventDao) DeleteTTL(ctx context.Context, ttl time.Duration, limit int) (int64, error) { + ctx, span := dao.trace(ctx, fmt.Sprintf("dao.%s.delete_ttl", dao.opts.Table)) + defer span.End() + + sql := fmt.Sprintf( + `DELETE FROM events WHERE id IN ( + SELECT id FROM events WHERE created_at < now() - INTERVAL '%s' ORDER BY created_at ASC LIMIT $1 + )`, + fmt.Sprintf("%d hours", int(ttl.Hours())), + ) + + dao.debugSQL(sql, []interface{}{limit}) + res, err := dao.DB(ctx).ExecContext(ctx, sql, limit) + if err != nil { + return 0, err + } + return res.RowsAffected() +} type EventQuery struct { Query diff --git a/db/migrations/1785833137_retention.down.sql b/db/migrations/1785833137_retention.down.sql new file mode 100644 index 00000000..460acb04 --- /dev/null +++ b/db/migrations/1785833137_retention.down.sql @@ -0,0 +1,4 @@ +DROP INDEX IF EXISTS idx_events_created_at; +DROP INDEX IF EXISTS idx_attempts_created_at; +ALTER TABLE attempts ADD CONSTRAINT attempts_event_id_fkey FOREIGN KEY (event_id) REFERENCES events(id) ON DELETE CASCADE; +ALTER TABLE attempts ADD CONSTRAINT attempts_endpoint_id_fkey FOREIGN KEY (endpoint_id) REFERENCES endpoints(id) ON DELETE CASCADE; diff --git a/db/migrations/1785833137_retention.up.sql b/db/migrations/1785833137_retention.up.sql new file mode 100644 index 00000000..b38c5061 --- /dev/null +++ b/db/migrations/1785833137_retention.up.sql @@ -0,0 +1,4 @@ +CREATE INDEX IF NOT EXISTS idx_events_created_at ON events (created_at); +ALTER TABLE attempts DROP CONSTRAINT IF EXISTS attempts_endpoint_id_fkey; +ALTER TABLE attempts DROP CONSTRAINT IF EXISTS attempts_event_id_fkey; +CREATE INDEX IF NOT EXISTS idx_attempts_created_at ON attempts (created_at); diff --git a/services/retention/retention.go b/services/retention/retention.go new file mode 100644 index 00000000..8b8a4c45 --- /dev/null +++ b/services/retention/retention.go @@ -0,0 +1,167 @@ +package retention + +import ( + "context" + "fmt" + "time" + + "github.com/webhookx-io/webhookx/config/modules" + "github.com/webhookx-io/webhookx/db" + "github.com/webhookx-io/webhookx/services/schedule" + "github.com/webhookx-io/webhookx/utils" + "go.uber.org/zap" +) + +const ( + Interval = time.Hour * 24 + BatchSize = 1000 + + EventsKey = "events" + AttemptsKey = "attempts" +) + +type RetentionService struct { + cfg modules.RetentionConfig + db *db.DB + log *zap.SugaredLogger + scheduler schedule.Scheduler + ttls map[string]time.Duration +} + +func NewRetentionService( + cfg modules.RetentionConfig, + db *db.DB, + log *zap.SugaredLogger, + scheduler schedule.Scheduler, +) *RetentionService { + + ttls := make(map[string]time.Duration) + ttls[EventsKey] = time.Hour * 24 * time.Duration(cfg.Events) + ttls[AttemptsKey] = time.Hour * 24 * time.Duration(cfg.Attempts) + + return &RetentionService{ + cfg: cfg, + db: db, + log: log.Named("retention"), + scheduler: scheduler, + ttls: ttls, + } +} + +func (s *RetentionService) Name() string { + return "retention" +} + +func (s *RetentionService) Start() error { + if !s.cfg.Enabled { + return nil + } + + s.log.Infow("service started", + "purge_interval", utils.FormatDuration(Interval), + "batch_size", BatchSize, + zap.Any("ttl", map[string]string{ + "events": fmt.Sprintf("%dd", s.cfg.Events), + "attempts": fmt.Sprintf("%dd", s.cfg.Attempts), + })) + + s.scheduler.Schedule(schedule.Task{ + Name: "retention", + Scheduled: schedule.NewIntervalSchedule(0, Interval), + Run: func(ctx context.Context) error { + return s.run(ctx) + }, + }) + + return nil +} + +func (s *RetentionService) Stop(ctx context.Context) error { + return nil +} + +func (s *RetentionService) run(ctx context.Context) error { + if s.ttls[EventsKey] > 0 { + ttl := s.ttls[EventsKey] + count, err := s.purgeEvents(ctx, ttl) + if err != nil { + return err + } + if count > 0 { + s.log.Infof("deleted %d expired events", count) + } + } + + if s.ttls[AttemptsKey] > 0 { + ttl := s.ttls[AttemptsKey] + count, err := s.purgeAttempts(ctx, ttl) + if err != nil { + return err + } + if count > 0 { + s.log.Infof("deleted %d expired attempts", count) + } + } + + return nil +} + +func (s *RetentionService) purgeEvents(ctx context.Context, ttl time.Duration) (int64, error) { + batchSize := BatchSize + + s.log.Debugw("deleting expired events", "ttl", utils.FormatDuration(ttl), "batch_size", batchSize) + + var total int64 + for { + select { + case <-ctx.Done(): + return total, ctx.Err() + default: + } + + deleted, err := s.db.Events.DeleteTTL(ctx, ttl, batchSize) + if err != nil { + return total, fmt.Errorf("failed to delete expired events: %w", err) + } + + total += deleted + + if deleted < int64(batchSize) { + break + } + + time.Sleep(10 * time.Millisecond) + } + + return total, nil +} + +func (s *RetentionService) purgeAttempts(ctx context.Context, ttl time.Duration) (int64, error) { + batchSize := BatchSize + + s.log.Debugw("deleting expired attempts", "ttl", utils.FormatDuration(ttl), "batch_size", batchSize) + + var total int64 + for { + select { + case <-ctx.Done(): + return total, ctx.Err() + default: + } + + deleted, err := s.db.Attempts.DeleteTTL(ctx, ttl, batchSize) + if err != nil { + return total, fmt.Errorf("failed to delete expired attempts: %w", err) + } + + total += deleted + + if deleted < int64(batchSize) { + break + } + + time.Sleep(10 * time.Millisecond) + } + + return total, nil +} diff --git a/test/cmd/db_test.go b/test/cmd/db_test.go index 32708963..12dd5a2c 100644 --- a/test/cmd/db_test.go +++ b/test/cmd/db_test.go @@ -21,11 +21,12 @@ var statusOutputInit = `1 init (⏳ pending) 10 ratelimit (⏳ pending) 11 event_unique_id (⏳ pending) 1762423418 source_config (⏳ pending) +1785833137 retention (⏳ pending) Summary: Current version: 0 Dirty: false Executed: 0 - Pending: 12 + Pending: 13 ` var statusOutputDone = `1 init (✅ executed) @@ -40,10 +41,11 @@ var statusOutputDone = `1 init (✅ executed) 10 ratelimit (✅ executed) 11 event_unique_id (✅ executed) 1762423418 source_config (✅ executed) +1785833137 retention (✅ executed) Summary: - Current version: 1762423418 + Current version: 1785833137 Dirty: false - Executed: 12 + Executed: 13 Pending: 0 ` diff --git a/test/retention/retention_test.go b/test/retention/retention_test.go new file mode 100644 index 00000000..7114ca26 --- /dev/null +++ b/test/retention/retention_test.go @@ -0,0 +1,126 @@ +package retention + +import ( + "context" + "testing" + "time" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + "github.com/stretchr/testify/assert" + "github.com/webhookx-io/webhookx/app" + "github.com/webhookx-io/webhookx/db/dao" + "github.com/webhookx-io/webhookx/db/entities" + "github.com/webhookx-io/webhookx/test/helper" + "github.com/webhookx-io/webhookx/test/helper/factory" + "github.com/webhookx-io/webhookx/utils" +) + +var _ = Describe("retention", Ordered, func() { + + Context("sanity", func() { + var app *app.Application + + BeforeAll(func() { + db := helper.InitDB(true, nil) + ws, err := helper.GetDeafultWorkspace() + assert.NoError(GinkgoT(), err) + + // Expired event + expiredEvent := factory.EventWS(ws.ID) + assert.NoError(GinkgoT(), db.Events.Insert(context.TODO(), expiredEvent)) + _, err = db.SqlDB().Exec("UPDATE events SET created_at = $1 WHERE id = $2", time.Now().Add(-24*time.Hour), expiredEvent.ID) + assert.NoError(GinkgoT(), err) + + // Active event + activeEvent := factory.EventWS(ws.ID) + assert.NoError(GinkgoT(), db.Events.Insert(context.TODO(), activeEvent)) + + // Expired attempt + expiredAttempt := &entities.Attempt{ + ID: utils.KSUID(), + EventId: activeEvent.ID, + Status: entities.AttemptStatusSuccess, + AttemptNumber: 1, + BaseModel: entities.BaseModel{ + WorkspaceId: ws.ID, + }, + } + assert.NoError(GinkgoT(), db.Attempts.Insert(context.TODO(), expiredAttempt)) + _, err = db.SqlDB().Exec("UPDATE attempts SET created_at = $1 WHERE id = $2", time.Now().Add(-24*time.Hour), expiredAttempt.ID) + assert.NoError(GinkgoT(), err) + + // Active attempt + activeAttempt := &entities.Attempt{ + ID: utils.KSUID(), + EventId: activeEvent.ID, + Status: entities.AttemptStatusSuccess, + AttemptNumber: 1, + BaseModel: entities.BaseModel{ + WorkspaceId: ws.ID, + }, + } + assert.NoError(GinkgoT(), db.Attempts.Insert(context.TODO(), activeAttempt)) + + app = utils.Must(helper.Start(map[string]string{ + "WEBHOOKX_RETENTION_ENABLED": "true", + "WEBHOOKX_RETENTION_EVENTS": "1", + "WEBHOOKX_RETENTION_ATTEMPTS": "1", + })) + }) + + AfterAll(func() { + app.Stop() + }) + + It("retention service should start", func() { + assert.Eventually(GinkgoT(), func() bool { + matched, err := helper.FileHasLine(helper.LogFile, "^.*\\[retention\\]\\s+service started.*$") + return err == nil && matched + }, time.Second*5, time.Second) + }) + + It("should purge expired events and attempts when triggered", func() { + app.Scheduler().RunNow("retention") + + db := app.DB() + + // verify expired event is deleted, active event remains + events, err := db.Events.List(context.TODO(), &dao.Query{}) + assert.NoError(GinkgoT(), err) + assert.Equal(GinkgoT(), 1, len(events)) + + // verify expired attempt is deleted, active attempt remains + attempts, err := db.Attempts.List(context.TODO(), &dao.Query{}) + assert.NoError(GinkgoT(), err) + assert.Equal(GinkgoT(), 1, len(attempts)) + }) + }) + + Context("errors", func() { + var app *app.Application + + BeforeAll(func() { + app = utils.Must(helper.Start(map[string]string{ + "WEBHOOKX_RETENTION_ENABLED": "true", + "WEBHOOKX_ROLE": "dp_proxy", + })) + }) + + AfterAll(func() { + app.Stop() + }) + + It("ignore retention configuration on data-planes", func() { + assert.Eventually(GinkgoT(), func() bool { + matched, err := helper.FileHasLine(helper.LogFile, "retention configuration is ignored on data-plane nodes") + return err == nil && matched + }, time.Second*5, time.Second) + }) + }) +}) + +func Test(t *testing.T) { + RegisterFailHandler(Fail) + RunSpecs(t, "Retention Suite") +} diff --git a/utils/fmt.go b/utils/fmt.go new file mode 100644 index 00000000..64126ab4 --- /dev/null +++ b/utils/fmt.go @@ -0,0 +1,18 @@ +package utils + +import ( + "fmt" + "time" +) + +func FormatDuration(d time.Duration) string { + if d%(24*time.Hour) == 0 { + return fmt.Sprintf("%dd", d/(24*time.Hour)) + } + + if d%time.Hour == 0 { + return fmt.Sprintf("%dh", d/time.Hour) + } + + return d.String() +} From 55a05333850c9c1d5cc00d1d9d6e8aa62d424489 Mon Sep 17 00:00:00 2001 From: Douglas-Lee Date: Tue, 4 Aug 2026 17:15:07 +0800 Subject: [PATCH 2/2] add tests --- test/retention/retention_test.go | 43 ++++++++++++++++-------------- utils/fmt.go | 45 +++++++++++++++++++++++++++----- utils/fmt_test.go | 18 +++++++++++++ 3 files changed, 81 insertions(+), 25 deletions(-) create mode 100644 utils/fmt_test.go diff --git a/test/retention/retention_test.go b/test/retention/retention_test.go index 7114ca26..896e6680 100644 --- a/test/retention/retention_test.go +++ b/test/retention/retention_test.go @@ -11,6 +11,7 @@ import ( "github.com/webhookx-io/webhookx/app" "github.com/webhookx-io/webhookx/db/dao" "github.com/webhookx-io/webhookx/db/entities" + "github.com/webhookx-io/webhookx/services/retention" "github.com/webhookx-io/webhookx/test/helper" "github.com/webhookx-io/webhookx/test/helper/factory" "github.com/webhookx-io/webhookx/utils" @@ -26,29 +27,33 @@ var _ = Describe("retention", Ordered, func() { ws, err := helper.GetDeafultWorkspace() assert.NoError(GinkgoT(), err) - // Expired event - expiredEvent := factory.EventWS(ws.ID) - assert.NoError(GinkgoT(), db.Events.Insert(context.TODO(), expiredEvent)) - _, err = db.SqlDB().Exec("UPDATE events SET created_at = $1 WHERE id = $2", time.Now().Add(-24*time.Hour), expiredEvent.ID) - assert.NoError(GinkgoT(), err) + // add 1001 expired events + for i := 0; i < retention.BatchSize+1; i++ { + expiredEvent := factory.EventWS(ws.ID) + assert.NoError(GinkgoT(), db.Events.Insert(context.TODO(), expiredEvent)) + _, err = db.SqlDB().Exec("UPDATE events SET created_at = $1 WHERE id = $2", time.Now().Add(-24*time.Hour), expiredEvent.ID) + assert.NoError(GinkgoT(), err) + } // Active event activeEvent := factory.EventWS(ws.ID) assert.NoError(GinkgoT(), db.Events.Insert(context.TODO(), activeEvent)) - // Expired attempt - expiredAttempt := &entities.Attempt{ - ID: utils.KSUID(), - EventId: activeEvent.ID, - Status: entities.AttemptStatusSuccess, - AttemptNumber: 1, - BaseModel: entities.BaseModel{ - WorkspaceId: ws.ID, - }, + // add 1001 expired attempts + for i := 0; i < retention.BatchSize+1; i++ { + expiredAttempt := &entities.Attempt{ + ID: utils.KSUID(), + EventId: activeEvent.ID, + Status: entities.AttemptStatusSuccess, + AttemptNumber: 1, + BaseModel: entities.BaseModel{ + WorkspaceId: ws.ID, + }, + } + assert.NoError(GinkgoT(), db.Attempts.Insert(context.TODO(), expiredAttempt)) + _, err = db.SqlDB().Exec("UPDATE attempts SET created_at = $1 WHERE id = $2", time.Now().Add(-24*time.Hour), expiredAttempt.ID) + assert.NoError(GinkgoT(), err) } - assert.NoError(GinkgoT(), db.Attempts.Insert(context.TODO(), expiredAttempt)) - _, err = db.SqlDB().Exec("UPDATE attempts SET created_at = $1 WHERE id = $2", time.Now().Add(-24*time.Hour), expiredAttempt.ID) - assert.NoError(GinkgoT(), err) // Active attempt activeAttempt := &entities.Attempt{ @@ -77,7 +82,7 @@ var _ = Describe("retention", Ordered, func() { assert.Eventually(GinkgoT(), func() bool { matched, err := helper.FileHasLine(helper.LogFile, "^.*\\[retention\\]\\s+service started.*$") return err == nil && matched - }, time.Second*5, time.Second) + }, time.Second*5, time.Microsecond*100) }) It("should purge expired events and attempts when triggered", func() { @@ -115,7 +120,7 @@ var _ = Describe("retention", Ordered, func() { assert.Eventually(GinkgoT(), func() bool { matched, err := helper.FileHasLine(helper.LogFile, "retention configuration is ignored on data-plane nodes") return err == nil && matched - }, time.Second*5, time.Second) + }, time.Second*5, time.Microsecond*100) }) }) }) diff --git a/utils/fmt.go b/utils/fmt.go index 64126ab4..8f1d0b42 100644 --- a/utils/fmt.go +++ b/utils/fmt.go @@ -1,18 +1,51 @@ package utils import ( - "fmt" + "strconv" + "strings" "time" ) func FormatDuration(d time.Duration) string { - if d%(24*time.Hour) == 0 { - return fmt.Sprintf("%dd", d/(24*time.Hour)) + if d == 0 { + return "0s" } - if d%time.Hour == 0 { - return fmt.Sprintf("%dh", d/time.Hour) + d = d.Abs() + var sb strings.Builder + + day := 24 * time.Hour + if d >= day { + days := d / day + sb.WriteString(strconv.FormatInt(int64(days), 10)) + sb.WriteByte('d') + d %= day + } + + if d >= time.Hour { + hours := d / time.Hour + sb.WriteString(strconv.FormatInt(int64(hours), 10)) + sb.WriteByte('h') + d %= time.Hour + } + + if d >= time.Minute { + minutes := d / time.Minute + sb.WriteString(strconv.FormatInt(int64(minutes), 10)) + sb.WriteByte('m') + d %= time.Minute + } + + if d >= time.Second { + seconds := d / time.Second + sb.WriteString(strconv.FormatInt(int64(seconds), 10)) + sb.WriteByte('s') + d %= time.Second + } + + if d > 0 { + sb.WriteString(d.String()) } - return d.String() + return sb.String() } diff --git a/utils/fmt_test.go b/utils/fmt_test.go new file mode 100644 index 00000000..8cdabc5c --- /dev/null +++ b/utils/fmt_test.go @@ -0,0 +1,18 @@ +package utils + +import ( + "testing" + "time" + + "github.com/stretchr/testify/assert" +) + +func TestFormatDuration(t *testing.T) { + assert.Equal(t, "1d", FormatDuration(time.Hour*24)) + assert.Equal(t, "1h", FormatDuration(time.Hour)) + assert.Equal(t, "1m", FormatDuration(time.Minute)) + assert.Equal(t, "1s", FormatDuration(time.Second)) + assert.Equal(t, "0s", FormatDuration(0)) + assert.Equal(t, "1d2h3m4s", FormatDuration(24*time.Hour+2*time.Hour+3*time.Minute+4*time.Second)) + assert.Equal(t, "1ms", FormatDuration(time.Millisecond)) +}