Skip to content
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

chore(blockbuilder): mark a job as complete only after stopping sync updates #15670

Merged
merged 5 commits into from
Jan 9, 2025
+10 −11
Merged

chore(blockbuilder): mark a job as complete only after stopping sync updates #15670

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
53068d8
remove unused consumergroup flag
ashwanthgoli Jan 8, 2025
44b515f
send consistent status update to scheduler
ashwanthgoli Jan 9, 2025
aa59ac1
fixup! send consistent status update to scheduler
ashwanthgoli Jan 9, 2025
8d9bdfc
add duration to summary log
ashwanthgoli Jan 9, 2025
9cb0361
fixup! add duration to summary log
ashwanthgoli Jan 9, 2025
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
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 @@ -210,10 +210,6 @@ block_builder:
[scheduler_grpc_client_config: <grpc_client>]

block_scheduler:
# Consumer group used by block scheduler to track the last consumed offset.
# CLI flag: -block-scheduler.consumer-group
[consumer_group: <string> | default = "block-scheduler"]

# How often the scheduler should plan jobs.
# CLI flag: -block-scheduler.interval
[interval: <duration> | default = 15m]
Expand Down
14 changes: 9 additions & 5 deletions pkg/blockbuilder/builder/builder.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -279,6 +279,12 @@ func (i *BlockBuilder) runOne(ctx context.Context, c *kgo.Client, workerID strin
completion.Success = false
}

// remove from inflight jobs to stop sending sync requests
i.jobsMtx.Lock()
delete(i.inflightJobs, job.ID())
i.metrics.inflightJobs.Set(float64(len(i.inflightJobs)))
i.jobsMtx.Unlock()

if _, err := withBackoff(
ctx,
i.cfg.Backoff,
Expand All @@ -292,16 +298,12 @@ func (i *BlockBuilder) runOne(ctx context.Context, c *kgo.Client, workerID strin
return true, err
}

i.jobsMtx.Lock()
delete(i.inflightJobs, job.ID())
i.metrics.inflightJobs.Set(float64(len(i.inflightJobs)))
i.jobsMtx.Unlock()

return true, err
}

func (i *BlockBuilder) processJob(ctx context.Context, c *kgo.Client, job *types.Job, logger log.Logger) (lastOffsetConsumed int64, err error) {
level.Debug(logger).Log("msg", "beginning job")
start := time.Now()

indexer := newTsdbCreator()
appender := newAppender(i.id,
Expand Down Expand Up @@ -505,6 +507,8 @@ func (i *BlockBuilder) processJob(ctx context.Context, c *kgo.Client, job *types
level.Info(logger).Log(
"msg", "successfully processed job",
"last_offset", lastOffset,
"duration", time.Since(start),
"records", lastOffset-job.Offsets().Min,
)

return lastOffset, nil
Expand Down
1 change: 1 addition & 0 deletions pkg/blockbuilder/scheduler/queue.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -332,6 +332,7 @@ func (q *JobQueue) SyncJob(jobID string, job *types.Job) {
case types.JobStatusInProgress:
case types.JobStatusComplete, types.JobStatusFailed, types.JobStatusExpired:
// Job already completed, re-enqueue a new one
level.Warn(q.logger).Log("msg", "job already completed, re-enqueuing", "job", jobID, "status", jobMeta.Status)
registerInProgress()
return
default:
Expand Down
2 changes: 0 additions & 2 deletions pkg/blockbuilder/scheduler/scheduler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,6 @@ var (
)

type Config struct {
ConsumerGroup string `yaml:"consumer_group"`
Interval time.Duration `yaml:"interval"`
LookbackPeriod time.Duration `yaml:"lookback_period"`
Strategy string `yaml:"strategy"`
Expand All @@ -36,7 +35,6 @@ type Config struct {

func (cfg *Config) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet) {
f.DurationVar(&cfg.Interval, prefix+"interval", 15*time.Minute, "How often the scheduler should plan jobs.")
f.StringVar(&cfg.ConsumerGroup, prefix+"consumer-group", "block-scheduler", "Consumer group used by block scheduler to track the last consumed offset.")
f.DurationVar(&cfg.LookbackPeriod, prefix+"lookback-period", 0, "Lookback period used by the scheduler to plan jobs when the consumer group has no commits. 0 consumes from the start of the partition.")
f.StringVar(
&cfg.Strategy,
Expand Down
Loading