Skip to content

feat: Added task_id-related TTL for cache plugin #635

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
DrRebus wants to merge 1 commit into
base: master
Choose a base branch
from
Open

feat: Added task_id-related TTL for cache plugin #635

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion db/migration.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@ import (
)

const (
expectedVersion = "v1.21.1-migration011"
expectedVersion = "v1.21.1-migration012"
)

var (
Expand Down
71 changes: 63 additions & 8 deletions engine/collector_cache.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,9 @@ import (
"github.com/ovh/utask/pkg/now"
)

// How many orphan entries we can fetch to purge at once
const maxOrphanEntriesToPurge uint64 = 100

// CacheCollector launches a process that cleans up expired entries from the cache plugin
func CacheCollector(ctx context.Context) error {
dbp, err := zesty.NewDBProvider(utask.DBName)
Expand All @@ -26,10 +29,9 @@ func CacheCollector(ctx context.Context) error {
// Delete expired entries from the cache plugin
go func() {
// Run it immediately and wait for new tick
if purged, err := purgeExpiredEntries(dbp); err != nil {
log.Printf("CacheCollector: failed to trash expired entries: %s", err)
} else if purged > 0 {
log.Printf("CacheCollector: purged %d expired entries at startup", purged)
expired, orphan := purgeCache(dbp)
if expired+orphan > 0 {
log.Printf("CacheCollector: purged %d expired and %d orphan entrie(s) at startup", expired, orphan)
}

for running := true; running; {
Expand All @@ -39,10 +41,9 @@ func CacheCollector(ctx context.Context) error {
case <-ctx.Done():
running = false
default:
if purged, err := purgeExpiredEntries(dbp); err != nil {
log.Printf("CacheCollector: failed to trash expired entries: %s", err)
} else if purged > 0 {
log.Printf("CacheCollector: purged %d expired entries", purged)
expired, orphan := purgeCache(dbp)
if expired+orphan > 0 {
log.Printf("CacheCollector: purged %d expired and %d orphan entrie(s)", expired, orphan)
}
}
}
Expand All @@ -51,6 +52,21 @@ func CacheCollector(ctx context.Context) error {
return nil
}

// purgeCache purges all expired and orphan entries in the "cache" table (See the "cache" plugin)
func purgeCache(dbp zesty.DBProvider) (int64, int64) {
purgedExpired, err := purgeExpiredEntries(dbp)
if err != nil {
log.Printf("CacheCollector: failed to trash expired entries: %s", err)
}

purgedOrdphan, err := purgeOrphanEntries(dbp)
if err != nil {
log.Printf("CacheCollector: failed to trash orphan entries: %s", err)
}

return purgedExpired, purgedOrdphan
}

func purgeExpiredEntries(dbp zesty.DBProvider) (int64, error) {
query, args, err := sqlgenerator.PGsql.
Delete(`"cache"`).
Expand All @@ -70,3 +86,42 @@ func purgeExpiredEntries(dbp zesty.DBProvider) (int64, error) {

return res.RowsAffected()
}

func purgeOrphanEntries(dbp zesty.DBProvider) (int64, error) {
// Fetch orphan entries
query, args, err := sqlgenerator.PGsql.
Select(`"cache"."key"`).
From(`"cache"`).
LeftJoin(`"task" on "cache"."owner" = "task"."public_id"`).
Where(squirrel.And{
squirrel.Eq{`"cache"."expires_at"`: nil},
squirrel.NotEq{`"cache"."owner"`: nil},
squirrel.Eq{`"task"."public_id"`: nil},
}).
Limit(maxOrphanEntriesToPurge).
ToSql()
if err != nil {
return 0, err
}

cacheKeys := make([]string, 0)
_, err = dbp.DB().Select(&cacheKeys, query, args...)
if err != nil {
return 0, pgjuju.Interpret(err)
}

query, args, err = sqlgenerator.PGsql.
Delete(`"cache"`).
Where(squirrel.Eq{`"key"`: cacheKeys}).
ToSql()
if err != nil {
return 0, err
}

res, err := dbp.DB().Exec(query, args...)
if err != nil {
return 0, pgjuju.Interpret(err)
}

return res.RowsAffected()
}
7 changes: 5 additions & 2 deletions pkg/plugins/builtin/cache/cache.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,12 +21,15 @@ var (
// Action: "set", "get", or "delete"
// Key: the cache key (required)
// Value: the value to store (required for "set", ignored otherwise)
// TTL: time-to-live in seconds (0 means no expiration, only used with "set")
// TTL: time-to-live in seconds (0 means no expiration, only used with "set", mutually exclusive with Owner).
// Owner: Public ID of a Task that owns this cache-entry; the entry will be kept until the owner task is deleted (only
// used with "set", mutually exclusive with TTL).
type Config struct {
Action string `json:"action"`
Key string `json:"key"`
Value interface{} `json:"value,omitempty"`
TTL int64 `json:"ttl,omitempty"`
Owner string `json:"owner"`
}

func validConfig(config interface{}) error {
Expand Down Expand Up @@ -74,7 +77,7 @@ func execSet(dbp zesty.DBProvider, cfg *Config) (interface{}, interface{}, error
return nil, nil, fmt.Errorf("failed to marshal value: %s", err)
}

if err := setCacheEntry(dbp, cfg.Key, valueBytes, cfg.TTL); err != nil {
if err := setCacheEntry(dbp, cfg.Key, valueBytes, cfg.TTL, cfg.Owner); err != nil {
return nil, nil, err
}

Expand Down
22 changes: 18 additions & 4 deletions pkg/plugins/builtin/cache/models.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,18 +23,32 @@ func (c *cacheEntry) isExpired() bool {
return c.ExpiresAt != nil && now.Get().After(*c.ExpiresAt)
}

func setCacheEntry(dbp zesty.DBProvider, key string, value []byte, ttl int64) error {
func setCacheEntry(dbp zesty.DBProvider, key string, value []byte, ttl int64, owner string) error {
if ttl > 0 && owner != "" {
return errors.BadRequestf("can't use both 'ttl' and 'owner' as expiration condition")
}

var expiresAt *time.Time
if ttl > 0 {
t := now.Get().Add(time.Duration(ttl) * time.Second)
expiresAt = &t
}

ownerPtr := &owner
if owner == "" {
ownerPtr = nil
}

query, args, err := sqlgenerator.PGsql.
Insert(`"cache"`).
Columns(`"key"`, `"value"`, `"expires_at"`).
Values(key, value, expiresAt).
Suffix(`ON CONFLICT ("key") DO UPDATE SET "value" = EXCLUDED."value", "expires_at" = EXCLUDED."expires_at"`).
Columns(`"key"`, `"value"`, `"expires_at"`, `"owner"`).
Values(key, value, expiresAt, ownerPtr).
Suffix(
`ON CONFLICT ("key") DO UPDATE SET
"value" = EXCLUDED."value",
"expires_at" = EXCLUDED."expires_at",
"owner" = EXCLUDED."owner"`,
).
ToSql()
if err != nil {
return err
Expand Down
13 changes: 13 additions & 0 deletions sql/migrations/012_cache_task_id.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,13 @@
-- +migrate Up

ALTER TABLE "cache" ADD "owner" UUID;
CREATE INDEX "cache_owner_idx" ON "cache" ("owner");

INSERT INTO "utask_sql_migrations" VALUES ('v1.21.1-migration012');

-- +migrate Down

DROP INDEX "cache_owner_idx";
ALTER TABLE "cache" DROP COLUMN "owner" UUID;

DELETE FROM "utask_sql_migrations" WHERE current_migration_applied = 'v1.21.1-migration012';
6 changes: 4 additions & 2 deletions sql/schema.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -137,11 +137,13 @@ CREATE TABLE "utask_sql_migrations" (
CREATE TABLE "cache" (
"key" TEXT PRIMARY KEY,
"value" BYTEA NOT NULL,
"expires_at" TIMESTAMP WITH TIME ZONE
"expires_at" TIMESTAMP WITH TIME ZONE,
"owner" UUID
);

CREATE INDEX "cache_expires_at_idx" ON "cache" ("expires_at") WHERE "expires_at" IS NOT NULL;
CREATE INDEX "cache_owner_idx" ON "cache" ("owner");

INSERT INTO "utask_sql_migrations" VALUES ('v1.21.1-migration011');
INSERT INTO "utask_sql_migrations" VALUES ('v1.21.1-migration012');

END;