Skip to content

fix(job): key singleton locks by job identity #2042

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
adityathebe wants to merge 1 commit into
base: main
Choose a base branch
from
+338 −46
Open

fix(job): key singleton locks by job identity #2042

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
6 changes: 4 additions & 2 deletions job/controllers.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,8 +10,9 @@ import (
)

type JobCronEntry struct {
ID string `json:"id"`
JobID string `json:"id"`
Name string `json:"name"`
Aliases []string `json:"aliases,omitempty"`
Schedule string `json:"schedule"`
ResourceID string `json:"resource_id,omitempty"`
ResourceType string `json:"resource_type,omitempty"`
Expand All @@ -34,7 +35,8 @@ func CronDetailsHandler(crons ...*cron.Cron) func(c echo.Context) error {
}

return JobCronEntry{
ID: j.ID,
JobID: j.ID(),
Aliases: j.Aliases,
ResourceID: j.ResourceID,
ResourceType: j.ResourceType,
Name: j.Name,
Expand Down
118 changes: 74 additions & 44 deletions job/job.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -110,7 +110,6 @@ var RetentionHigh = Retention{
type Job struct {
context.Context
entryID *cron.EntryID
lock *sync.Mutex
initialized bool
unschedule func()
statusRing StatusRing
Expand All @@ -124,7 +123,7 @@ type Job struct {
Fn func(ctx JobRuntime) error
JobHistory bool
RunNow bool
ID string
Aliases []string
ResourceID, ResourceType string
IgnoreSuccessHistory bool
lastHistoryCleanup time.Time
Expand All @@ -141,19 +140,24 @@ type Job struct {

func (j *Job) GetContext() map[string]any {
return map[string]any{
"id": j.ID,
"id": j.ID(),
"aliases": j.Aliases,
"resourceID": j.ResourceID,
"resourceType": j.ResourceType,
"name": j.Name,
"schedule": j.Schedule,
}
}

func (j *Job) PK() string {
return strings.TrimSuffix(
strings.TrimSpace(fmt.Sprintf("%s/%s", j.Name, lo.CoalesceOrEmpty(j.ID, j.ResourceID))),
"/",
)
// ID returns the canonical logical identity for a job.
// Aliases are intentionally excluded; they are configuration/display names.
func (j *Job) ID() string {
name := strings.TrimSpace(j.Name)
resourceID := strings.TrimSpace(j.ResourceID)
if resourceID == "" {
return name
}
return fmt.Sprintf("%s::%s/%s", name, strings.TrimSpace(j.ResourceType), resourceID)
}

type StatusRing struct {
Expand Down Expand Up @@ -352,8 +356,9 @@ func (j *Job) Retain(r Retention) *Job {
return j
}

func (j *Job) SetID(id string) *Job {
j.ID = id
// SetAliases sets friendly names used for job properties and display.
func (j *Job) SetAliases(aliases ...string) *Job {
j.Aliases = aliases
return j
}

Expand All @@ -379,7 +384,7 @@ func (j *Job) Run() {
}

ctx, span := j.Context.StartSpan(j.Name)
ctx = ctx.WithName("job." + j.PK())
ctx = ctx.WithName("job." + j.ID())
defer span.End()

r := JobRuntime{
Expand All @@ -406,32 +411,31 @@ func (j *Job) Run() {
r.start()
defer r.end()
if j.Singleton {
ctx.Logger.V(4).Infof("acquiring lock")
key := j.ID()
ctx.Logger.V(4).Infof("acquiring lock %s", key)

if j.lock == nil {
j.lock = &sync.Mutex{}
}
if !j.lock.TryLock() {
unlock, ok := singletonLocks.TryLock(key)
if !ok {
r.History.Status = models.StatusSkipped
ctx.Tracef("failed to acquire lock")
ctx.Tracef("failed to acquire lock %s", key)
r.Skipped("job already running, skipping")
return
}
defer j.lock.Unlock()
defer unlock()
}

for i, lock := range j.Semaphores {
ctx.Logger.V(6).Infof("[%s] acquiring sempahore [%d/%d]", j.ID, i+1, len(j.Semaphores))
ctx.Logger.V(6).Infof("[%s] acquiring sempahore [%d/%d]", j.ID(), i+1, len(j.Semaphores))
if err := lock.Acquire(ctx, 1); err != nil {
r.Skipped("too many concurrent jobs, skipping")
return
}
ctx.Logger.V(7).Infof("[%s] acquired sempahore [%d/%d]", j.ID, i+1, len(j.Semaphores))
ctx.Logger.V(7).Infof("[%s] acquired sempahore [%d/%d]", j.ID(), i+1, len(j.Semaphores))

defer func(s *semaphore.Weighted, msg string) {
s.Release(1)
ctx.Logger.V(6).Infof(msg)
}(lock, fmt.Sprintf("[%s] released sempahore [%d/%d]", j.ID, i+1, len(j.Semaphores)))
}(lock, fmt.Sprintf("[%s] released sempahore [%d/%d]", j.ID(), i+1, len(j.Semaphores)))
Comment thread
adityathebe marked this conversation as resolved.
}

if j.Timeout > 0 {
Expand All @@ -449,41 +453,67 @@ func (j *Job) Run() {
}

func (j *Job) getPropertyNames(key string) []string {
if j.ID == "" {
return []string{
fmt.Sprintf("jobs.%s.%s", j.Name, key),
fmt.Sprintf("jobs.%s", key)}
names := make([]string, 0, len(j.Aliases)+2)
for _, alias := range j.aliases() {
names = append(names, fmt.Sprintf("jobs.%s.%s.%s", j.Name, alias, key))
}
return []string{
fmt.Sprintf("jobs.%s.%s.%s", j.Name, j.ID, key),
return append(names,
fmt.Sprintf("jobs.%s.%s", j.Name, key),
fmt.Sprintf("jobs.%s", key)}
fmt.Sprintf("jobs.%s", key),
)
}

func (j *Job) GetProperty(property string) (string, bool) {
if val := j.Context.Properties().String("jobs."+j.Name+"."+property, ""); val != "" {
return val, true
}
if j.ID != "" {
if val := j.Context.Properties().String(fmt.Sprintf("jobs.%s.%s.%s", j.Name, j.ID, property), ""); val != "" {
for _, name := range j.getPropertyLookupNames(property) {
if val := j.Context.Properties().String(name, ""); val != "" {
return val, true
}
}
return "", false
}

func (j *Job) GetPropertyInt(property string, def int) int {
if val := j.Context.Properties().Int("jobs."+j.Name+"."+property, def); val != def {
return val
}
if j.ID != "" {
if val := j.Context.Properties().Int(fmt.Sprintf("jobs.%s.%s.%s", j.Name, j.ID, property), def); val != def {
for _, name := range j.getPropertyLookupNames(property) {
if val := j.Context.Properties().Int(name, def); val != def {
return val
}
}
return def
}

func (j *Job) getPropertyLookupNames(key string) []string {
names := []string{fmt.Sprintf("jobs.%s.%s", j.Name, key)}
for _, alias := range j.aliases() {
names = append(names, fmt.Sprintf("jobs.%s.%s.%s", j.Name, alias, key))
}
return names
}

func (j *Job) aliases() []string {
aliases := make([]string, 0, len(j.Aliases))
seen := make(map[string]struct{}, len(j.Aliases))
for _, alias := range j.Aliases {
alias = strings.TrimSpace(alias)
if alias == "" {
continue
}
if _, ok := seen[alias]; ok {
continue
}
seen[alias] = struct{}{}
aliases = append(aliases, alias)
}
return aliases
}

func (j *Job) primaryAlias() string {
aliases := j.aliases()
if len(aliases) == 0 {
return ""
}
return aliases[0]
}

func (j *Job) init() error {
StartJobHistoryEvictor(j.Context)

Expand Down Expand Up @@ -544,8 +574,8 @@ func (j *Job) init() error {
j.Context = j.Context.WithDBLogLevel(dbLevel)
}

if j.ID != "" {
j.Context = j.Context.WithName(fmt.Sprintf("%s.%s", strings.ToLower(j.Name), j.ID))
if alias := j.primaryAlias(); alias != "" {
j.Context = j.Context.WithName(fmt.Sprintf("%s.%s", strings.ToLower(j.Name), alias))
} else if j.ResourceID != "" {
j.Context = j.Context.WithName(fmt.Sprintf("%s.%s", strings.ToLower(j.Name), j.ResourceID))
} else {
Expand All @@ -564,8 +594,8 @@ func (j *Job) init() error {
}

func (j *Job) Label() string {
if j.ID != "" {
return fmt.Sprintf("%s/%s", j.Name, j.ID)
if alias := j.primaryAlias(); alias != "" {
return fmt.Sprintf("%s/%s", j.Name, alias)
}
return j.Name
}
Expand All @@ -582,8 +612,8 @@ func (j *Job) String() string {
}

func (j *Job) GetResourcedName() string {
if j.ID != "" {
return fmt.Sprintf("%s [%s]", j.Name, j.ID)
if alias := j.primaryAlias(); alias != "" {
return fmt.Sprintf("%s [%s]", j.Name, alias)
}

return j.Name
Expand Down
93 changes: 93 additions & 0 deletions job/job_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,8 @@ import (
"testing"
"time"

"github.com/flanksource/commons/properties"
dutycontext "github.com/flanksource/duty/context"
. "github.com/onsi/ginkgo/v2"
. "github.com/onsi/gomega"

Expand All @@ -19,6 +21,97 @@ func TestJob(t *testing.T) {
RunSpecs(t, "Job Suite")
}

var _ = Describe("Job", func() {
It("builds IDs from logical job identity", func() {
cases := []struct {
name string
job *Job
want string
}{
{
name: "resource identity",
job: &Job{
Name: "Scraper",
ResourceType: ResourceTypeScraper,
ResourceID: "scraper-1",
Aliases: []string{"default/scraper"},
},
want: "Scraper::config_scraper/scraper-1",
},
{
name: "aliases are not identity",
job: &Job{
Name: "Scraper",
Aliases: []string{"manual", "scheduled"},
},
want: "Scraper",
},
{
name: "name fallback",
job: &Job{
Name: "Scraper",
},
want: "Scraper",
},
}

for _, tt := range cases {
Expect(tt.job.ID()).To(Equal(tt.want), tt.name)
}
})

It("keeps property lookup backward compatible", func() {
props := map[string]string{
"jobs.compat-job.schedule": "general",
"jobs.compat-job.alias.schedule": "specific",
"jobs.compat-job.retention.success": "10",
"jobs.compat-job.alias.retention.success": "20",
"jobs.compat-alias-only.alias.schedule": "specific",
"jobs.compat-alias-only.alias.retention.success": "20",
"jobs.schedule": "global",
"jobs.retention.success": "30",
}
for key, value := range props {
properties.Global.Set(key, value)
}
DeferCleanup(func() {
for key := range props {
properties.Global.Set(key, "")
}
})

j := &Job{
Name: "compat-job",
Aliases: []string{"alias"},
Context: dutycontext.NewContext(context.Background()),
}
schedule, ok := j.GetProperty("schedule")
Expect(ok).To(BeTrue())
Expect(schedule).To(Equal("general"))
Expect(j.GetPropertyInt("retention.success", 1)).To(Equal(10))

aliasOnly := &Job{
Name: "compat-alias-only",
Aliases: []string{"alias"},
Context: dutycontext.NewContext(context.Background()),
}
schedule, ok = aliasOnly.GetProperty("schedule")
Expect(ok).To(BeTrue())
Expect(schedule).To(Equal("specific"))
Expect(aliasOnly.GetPropertyInt("retention.success", 1)).To(Equal(20))

globalOnly := &Job{
Name: "compat-global-only",
Aliases: []string{"alias"},
Context: dutycontext.NewContext(context.Background()),
}
schedule, ok = globalOnly.GetProperty("schedule")
Expect(ok).To(BeFalse())
Expect(schedule).To(BeEmpty())
Expect(globalOnly.GetPropertyInt("retention.success", 1)).To(Equal(1))
})
})

var _ = Describe("StatusRing", Label("slow"), func() {
var ch chan uuid.UUID

Expand Down
38 changes: 38 additions & 0 deletions job/singleton.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,38 @@
package job

import (
"sync"
)

var singletonLocks = newSingletonLockRegistry()

type singletonLockRegistry struct {
mu sync.Mutex
running map[string]struct{}
}

func newSingletonLockRegistry() *singletonLockRegistry {
return &singletonLockRegistry{
running: make(map[string]struct{}),
}
}

func (r *singletonLockRegistry) TryLock(key string) (func(), bool) {
r.mu.Lock()
defer r.mu.Unlock()

if _, ok := r.running[key]; ok {
return nil, false
}

r.running[key] = struct{}{}

var once sync.Once
return func() {
once.Do(func() {
r.mu.Lock()
defer r.mu.Unlock()
delete(r.running, key)
})
}, true
}
Loading
Loading