Fixing Peak Sample Count Per Series

[Naman-B-Parlecha]

Resolve: #573

Currently, the peak sample count is incorrectly reported as 1 because currStepSamples is reset to 0 at every step iteration, preventing proper accumulation of peak samples across series.

To fix this we can maintain a vector to track sample count per step across all series and add up the sample at each step. Once all the all sample per step of all series are processed we can update the telemetry for all series

[Naman-B-Parlecha]

@yeya24 @MichaHoffmann PTAL!

[yeya24]

I think it is better to do #573 (comment).
We can decouple updating peak samples from IncrementSamplesAtTimestamp.

[Naman-B-Parlecha]

Sure refactoring!

[harry671003]

Thanks for the work on this!

Before we finalize the implementation, I'd like to clarify what we mean by peak samples in the context of the Thanos engine.

In the Prometheus engine, peak samples is defined as the highest number of samples held in memory at any moment during query execution, across all operators. This is used to enforce the --query.max-samples limit and prevent memory exhaustion.

A few open questions for alignment:

• Are we using the same definition of peak samples in the Thanos engine? That is, tracking the max number of samples in memory at any point during evaluation (across all operators)?
• Or are we maintaining per-operator peak samples separately? If so, how are we aggregating or enforcing limits?
• At what points in the engine are we calling UpdatePeak? Are we sure those points capture all relevant memory usage?

Having a shared definition with Prometheus helps make this limit consistent and easier to reason about. If we’re diverging, it would be good to document how and why.

Curious to hear thoughts.

[Naman-B-Parlecha]

Hey @harry671003 thanks for the clarification,
i m not entirely sure at what point we should be updating the peak but i believe we want to do something similar to what happens in prometheus?

@yeya24 is more qualified to answer this i believe as he has worked closely with the engine

[harry671003]

I thought a little bit more about this:

Let's define the Peak samples first. Peak samples for an operator is the total samples the operator processed in a single Next() call. Maximum Peak samples would be the peak samples across all operators.

One way to implement this is to use the telemetry operator
https://github.com/thanos-io/promql-engine/blob/main/execution/telemetry/telemetry.go#L204

func (t *Operator) Next(ctx context.Context) ([]model.StepVector, error) {
	start := time.Now()
	totalSamplesBefore := t.OperatorTelemetry.Samples().TotalSamples() // Get the samples before calling Next()
	defer func() { t.OperatorTelemetry.AddNextExecutionTime(time.Since(start)) }()
	out, err := t.inner.Next(ctx)
	if err != nil {
		return nil, err
	}
 	totalSamplesAfter :=  t.OperatorTelemetry.Samples().TotalSamples() // Get the samples after calling Next()
 	t.OperatorTelemetry.UpdatePeak(totalSamplesAfter - totalSamplesBefore) // The diff is the peak samples.
 	
 	return out, err
}

Wdyt @yeya24 @Naman-B-Parlecha ?

[Naman-B-Parlecha]

I thought a little bit more about this:

Let's define the Peak samples first. Peak samples for an operator is the total samples the operator processed in a single Next() call. Maximum Peak samples would be the peak samples across all operators.

One way to implement this is to use the telemetry operator https://github.com/thanos-io/promql-engine/blob/main/execution/telemetry/telemetry.go#L204

func (t *Operator) Next(ctx context.Context) ([]model.StepVector, error) {
	start := time.Now()
	totalSamplesBefore := t.OperatorTelemetry.Samples().TotalSamples() // Get the samples before calling Next()
	defer func() { t.OperatorTelemetry.AddNextExecutionTime(time.Since(start)) }()
	out, err := t.inner.Next(ctx)
	if err != nil {
		return nil, err
	}
 	totalSamplesAfter :=  t.OperatorTelemetry.Samples().TotalSamples() // Get the samples after calling Next()
 	t.OperatorTelemetry.UpdatePeak(totalSamplesAfter - totalSamplesBefore) // The diff is the peak samples.
 	
 	return out, err
}

Wdyt @yeya24 @Naman-B-Parlecha ?

great was looking into how prometheus is updating its peak
but curious about why is peak value the diff of after and before
shouldnt it be the highest in of the series? (this is how i m updating it currently in this pr not sure if i m forgetting something)

[harry671003]

@Naman-B-Parlecha

Yes, what you're doing is fundamentally the same. This is just a cleaner way of doing it.

• We decouple the IncrementSamples() and UpdatePeaks()
• We only do it in telemetry.Operator

[Naman-B-Parlecha]

@Naman-B-Parlecha

Yes, what you're doing is fundamentally the same. This is just a cleaner way of doing it.

• We decouple the IncrementSamples() and UpdatePeaks()
• We only do it in telemetry.Operator

Thanks for the clarification 🙌
Should I refactor the code in this pr or would u like to make a fresh pr and add changes to it?

[harry671003]

Use the same PR.
Also include unit tests.

[harry671003]

Great work! Just some comments.

[harry671003]

Is the peak samples correct here?
For the selector {[__name__="http_requests_total"]}[10m0s]

Conditions:

• Steps batch is 10.
• Selectors shards = 2

Calculation:

• Each Next() call has 10 steps.

• Each step has 1 series (due to sharding)

• For each step:

  • Each series has a sample every 30 seconds.
  • In [10m] range, there will be 10*60s/30s = 20 samples

• Total samples in each step = 10 steps * 1 series * 20 samples = 200

Looks correct.

[harry671003]

LGTM! Great work. Thank you.

[Naman-B-Parlecha]

Thanks for the helping me out @harry671003 @yeya24 🙌

[yeya24]

Thanks @Naman-B-Parlecha and @harry671003! I like how this is implemented now and it seems clean.

Let's fix lint. Unit tests also failed so maybe need to take another look

[Naman-B-Parlecha]

@yeya24 PTAL