Skip to content

fix(block-scheduler): one job per partition (local branch copy) #15579

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

Merged
merged 4 commits into from
Jan 6, 2025
Merged

fix(block-scheduler): one job per partition (local branch copy) #15579

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
1a300bb
fix: removes planned-jobs-per-interval config
owen-d Jan 2, 2025
ad9a9a6
enqueue next job in partition after previous completes
owen-d Jan 2, 2025
7235e37
bump ci
owen-d Jan 2, 2025
3ee784c
check match on enqueueing next job for partition
owen-d Jan 3, 2025
File filter

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Next Next commit
fix: removes planned-jobs-per-interval config
  • Loading branch information
@owen-d
owen-d committed Jan 2, 2025
commit 1a300bb13c5a2c3cc989ae57bdea2c8de613615b
4 changes: 0 additions & 4 deletions docs/sources/shared/configuration.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -232,10 +232,6 @@ block_scheduler:
# CLI flag: -block-scheduler.target-record-count
[target_record_count: <int> | default = 1000]

# Maximum number of jobs that the planner can return.
# CLI flag: -block-scheduler.max-jobs-planned-per-interval
[max_jobs_planned_per_interval: <int> | default = 100]

job_queue:
# Interval to check for expired job leases
# CLI flag: -jobqueue.lease-expiry-check-interval
Expand Down
21 changes: 7 additions & 14 deletions pkg/blockbuilder/scheduler/scheduler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,13 +25,12 @@ var (
)

type Config struct {
ConsumerGroup string `yaml:"consumer_group"`
Interval time.Duration `yaml:"interval"`
LookbackPeriod time.Duration `yaml:"lookback_period"`
Strategy string `yaml:"strategy"`
TargetRecordCount int64 `yaml:"target_record_count"`
MaxJobsPlannedPerInterval int `yaml:"max_jobs_planned_per_interval"`
JobQueueConfig JobQueueConfig `yaml:"job_queue"`
ConsumerGroup string `yaml:"consumer_group"`
Interval time.Duration `yaml:"interval"`
LookbackPeriod time.Duration `yaml:"lookback_period"`
Strategy string `yaml:"strategy"`
TargetRecordCount int64 `yaml:"target_record_count"`
JobQueueConfig JobQueueConfig `yaml:"job_queue"`
}

func (cfg *Config) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet) {
Expand All @@ -56,12 +55,6 @@ func (cfg *Config) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet) {
RecordCountStrategy,
),
)
f.IntVar(
&cfg.MaxJobsPlannedPerInterval,
prefix+"max-jobs-planned-per-interval",
100,
"Maximum number of jobs that the planner can return.",
)
cfg.JobQueueConfig.RegisterFlags(f)
}

Expand Down Expand Up @@ -155,7 +148,7 @@ func (s *BlockScheduler) runOnce(ctx context.Context) error {

s.publishLagMetrics(lag)

jobs, err := s.planner.Plan(ctx, s.cfg.MaxJobsPlannedPerInterval)
jobs, err := s.planner.Plan(ctx, 1) // TODO(owen-d): parallelize work within a partition
if err != nil {
level.Error(s.logger).Log("msg", "failed to plan jobs", "err", err)
}
Expand Down