diff --git a/job/controllers.go b/job/controllers.go index 1e124e203..a812472c1 100644 --- a/job/controllers.go +++ b/job/controllers.go @@ -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"` @@ -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, diff --git a/job/job.go b/job/job.go index 3e0135c88..36ee5fa2e 100644 --- a/job/job.go +++ b/job/job.go @@ -110,7 +110,6 @@ var RetentionHigh = Retention{ type Job struct { context.Context entryID *cron.EntryID - lock *sync.Mutex initialized bool unschedule func() statusRing StatusRing @@ -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 @@ -141,7 +140,8 @@ 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, @@ -149,11 +149,15 @@ func (j *Job) GetContext() map[string]any { } } -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 { @@ -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 } @@ -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{ @@ -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))) } if j.Timeout > 0 { @@ -449,23 +453,19 @@ 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 } } @@ -473,17 +473,47 @@ func (j *Job) GetProperty(property string) (string, bool) { } 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) @@ -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 { @@ -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 } @@ -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 diff --git a/job/job_test.go b/job/job_test.go index 79f4c0d04..18400ecc0 100644 --- a/job/job_test.go +++ b/job/job_test.go @@ -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" @@ -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 diff --git a/job/singleton.go b/job/singleton.go new file mode 100644 index 000000000..b6f6303f4 --- /dev/null +++ b/job/singleton.go @@ -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 +} diff --git a/tests/job_test.go b/tests/job_test.go index 3aaa7801e..a8ac8ecd3 100644 --- a/tests/job_test.go +++ b/tests/job_test.go @@ -1,12 +1,14 @@ package tests import ( + "sync" "sync/atomic" "time" "github.com/flanksource/duty/context" "github.com/flanksource/duty/job" "github.com/flanksource/duty/models" + "github.com/google/uuid" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" "github.com/samber/lo" @@ -47,6 +49,133 @@ var _ = Describe("Job", Ordered, func() { Expect(counter.Load()).To(Equal(current + 1)) }) + It("Prevents concurrent execution for separate job instances with the same resource identity", func() { + var firstRuns atomic.Int32 + var secondRuns atomic.Int32 + resourceID := uuid.NewString() + name := "keyed-singleton-" + uuid.NewString() + started := make(chan struct{}) + release := make(chan struct{}) + firstDone := make(chan struct{}) + var releaseOnce sync.Once + + DeferCleanup(func() { + releaseOnce.Do(func() { close(release) }) + Eventually(firstDone, "5s").Should(BeClosed()) + }) + + first := &job.Job{ + Name: name, + Aliases: []string{"scheduled"}, + Singleton: true, + Context: DefaultContext, + ResourceID: resourceID, + ResourceType: job.ResourceTypeScraper, + IgnoreSuccessHistory: true, + Fn: func(ctx job.JobRuntime) error { + firstRuns.Add(1) + close(started) + <-release + return nil + }, + } + second := &job.Job{ + Name: name, + Aliases: []string{"manual"}, + Singleton: true, + Context: DefaultContext, + ResourceID: resourceID, + ResourceType: job.ResourceTypeScraper, + IgnoreSuccessHistory: true, + Fn: func(ctx job.JobRuntime) error { + secondRuns.Add(1) + return nil + }, + } + + go func() { + defer close(firstDone) + first.Run() + }() + + Eventually(started, "5s").Should(BeClosed()) + second.Run() + + Expect(firstRuns.Load()).To(Equal(int32(1))) + Expect(secondRuns.Load()).To(BeZero()) + Expect(second.LastJob).ToNot(BeNil()) + Expect(second.LastJob.Status).To(Equal(models.StatusSkipped)) + + releaseOnce.Do(func() { close(release) }) + Eventually(firstDone, "5s").Should(BeClosed()) + }) + + It("Allows singleton jobs with different resource identities to run concurrently", func() { + var firstRuns atomic.Int32 + var secondRuns atomic.Int32 + name := "keyed-singleton-concurrent-" + uuid.NewString() + firstID := uuid.NewString() + secondID := uuid.NewString() + started := make(chan string, 2) + release := make(chan struct{}) + var wg sync.WaitGroup + var releaseOnce sync.Once + waitForJobs := func() { + done := make(chan struct{}) + go func() { + wg.Wait() + close(done) + }() + Eventually(done, "5s").Should(BeClosed()) + } + + DeferCleanup(func() { + releaseOnce.Do(func() { close(release) }) + waitForJobs() + }) + + makeJob := func(resourceID string, runs *atomic.Int32) *job.Job { + return &job.Job{ + Name: name, + Singleton: true, + Context: DefaultContext, + ResourceID: resourceID, + ResourceType: job.ResourceTypeScraper, + IgnoreSuccessHistory: true, + Fn: func(ctx job.JobRuntime) error { + runs.Add(1) + started <- resourceID + <-release + return nil + }, + } + } + + runAsync := func(j *job.Job) { + wg.Add(1) + go func() { + defer wg.Done() + j.Run() + }() + } + + first := makeJob(firstID, &firstRuns) + second := makeJob(secondID, &secondRuns) + + runAsync(first) + Eventually(started, "5s").Should(Receive(Equal(firstID))) + runAsync(second) + Eventually(started, "5s").Should(Receive(Equal(secondID))) + + Expect(firstRuns.Load()).To(Equal(int32(1))) + Expect(secondRuns.Load()).To(Equal(int32(1))) + + releaseOnce.Do(func() { close(release) }) + waitForJobs() + Expect(first.LastJob.Status).To(Equal(models.StatusSuccess)) + Expect(second.LastJob.Status).To(Equal(models.StatusSuccess)) + }) + It("Should skip disabled jobs", func() { var counter = atomic.Int32{} disabledJob := &job.Job{