Skip to content

tsdb: decouple OOO compaction from in-order head compaction #18434

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
anhtranbk wants to merge 1 commit into
base: main
Choose a base branch
from
Open

tsdb: decouple OOO compaction from in-order head compaction #18434

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
38 changes: 37 additions & 1 deletion cmd/prometheus/main_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -684,6 +684,42 @@ func getMetricValue(t *testing.T, body io.Reader, metricType model.MetricType, m
return 0, errors.New("cannot get value")
}

// getMetricValueSum returns the sum of all metrics for a given metric name.
// This is useful for CounterVec or GaugeVec metrics where we want the total across all label values.
//
// Key difference from getMetricValue: This function returns 0 (not an error) when the metric family
// doesn't exist, which is appropriate for metrics that only appear after being incremented.
func getMetricValueSum(t *testing.T, body io.Reader, metricType model.MetricType, metricName string) (float64, error) {
t.Helper()

p := expfmt.NewTextParser(model.UTF8Validation)
metricFamilies, err := p.TextToMetricFamilies(body)
if err != nil {
return 0, err
}
metricFamily, ok := metricFamilies[metricName]
if !ok {
return 0, nil
}
metrics := metricFamily.GetMetric()
if len(metrics) == 0 {
return 0, nil
}

var sum float64
for _, metric := range metrics {
switch metricType {
case model.MetricTypeGauge:
sum += metric.GetGauge().GetValue()
case model.MetricTypeCounter:
sum += metric.GetCounter().GetValue()
default:
t.Fatalf("metric type %s not supported", metricType)
}
}
return sum, nil
}

func TestRuntimeGOGCConfig(t *testing.T) {
if testing.Short() {
t.Skip("skipping test in short mode.")
Expand Down Expand Up @@ -870,7 +906,7 @@ scrape_configs:
require.NotZero(t, series)

// No compaction must have failed
failures, err := getMetricValue(t, bytes.NewReader(metrics), model.MetricTypeCounter, "prometheus_tsdb_compactions_failed_total")
failures, err := getMetricValueSum(t, bytes.NewReader(metrics), model.MetricTypeCounter, "prometheus_tsdb_compactions_failed_total")
require.NoError(t, err)
require.Zero(t, failures)
return true
Expand Down
106 changes: 55 additions & 51 deletions cmd/prometheus/main_upgrade_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -92,7 +92,7 @@ func fetchLatestLTSRelease(t *testing.T, prefix string) (version, assetURL strin
return "", ""
}

func getPrometheusMetricValue(t *testing.T, port int, metricType model.MetricType, metricName string) (float64, error) {
func getPrometheusMetricValue(t *testing.T, port int, metricType model.MetricType, metricName string, isVecMetric bool) (float64, error) {
t.Helper()

resp, err := http.Get(fmt.Sprintf("http://127.0.0.1:%d/metrics", port))
Expand All @@ -103,6 +103,9 @@ func getPrometheusMetricValue(t *testing.T, port int, metricType model.MetricTyp
if resp.StatusCode != http.StatusOK {
return 0, fmt.Errorf("bad status code %d", resp.StatusCode)
}
if isVecMetric {
return getMetricValueSum(t, resp.Body, metricType, metricName)
}
return getMetricValue(t, resp.Body, metricType, metricName)
}

Expand Down Expand Up @@ -159,59 +162,60 @@ func (c versionChangeTest) ensureHealthyMetrics(t *testing.T) {
checkStartTime := time.Now()

for _, mc := range []struct {
mType model.MetricType
mName string
check func(float64) bool
mType model.MetricType
mName string
isVecMetric bool // true if the metric is a vector type (e.g., CounterVec or GaugeVec).
check func(float64) bool
}{
{model.MetricTypeGauge, "prometheus_ready", func(v float64) bool { return v == 1 }},
{model.MetricTypeGauge, "prometheus_config_last_reload_successful", func(v float64) bool { return v == 1 }},

{model.MetricTypeCounter, "prometheus_target_scrape_pools_total", func(v float64) bool { return v == 3 }},
{model.MetricTypeCounter, "prometheus_target_scrape_pools_failed_total", func(v float64) bool { return v == 0 }},
{model.MetricTypeCounter, "prometheus_target_scrape_pool_reloads_failed_total", func(v float64) bool { return v == 0 }},

{model.MetricTypeGauge, "prometheus_remote_storage_highest_timestamp_in_seconds", func(v float64) bool { return v > float64(checkStartTime.Unix()) }},

{model.MetricTypeGauge, "prometheus_rule_group_rules", func(v float64) bool { return v == 2 }},
{model.MetricTypeCounter, "prometheus_rule_evaluations_total", func(v float64) bool { return v >= 1 }},
{model.MetricTypeCounter, "prometheus_rule_evaluation_failures_total", func(v float64) bool { return v == 0 }},

{model.MetricTypeCounter, "prometheus_tsdb_compactions_triggered_total", func(v float64) bool { return v >= 1 }},
{model.MetricTypeCounter, "prometheus_tsdb_compactions_total", func(v float64) bool { return v >= 1 }},
{model.MetricTypeCounter, "prometheus_tsdb_compactions_failed_total", func(v float64) bool { return v == 0 }},

{model.MetricTypeGauge, "prometheus_tsdb_head_series", func(v float64) bool { return v >= 1 }},
{model.MetricTypeGauge, "prometheus_tsdb_head_chunks", func(v float64) bool { return v >= 1 }},
{model.MetricTypeGauge, "prometheus_tsdb_head_chunks_storage_size_bytes", func(v float64) bool { return v >= 1 }},
{model.MetricTypeCounter, "prometheus_tsdb_head_series_created_total", func(v float64) bool { return v >= 1 }},
{model.MetricTypeGauge, "prometheus_tsdb_blocks_loaded", func(v float64) bool { return v >= 1 }},
{model.MetricTypeGauge, "prometheus_tsdb_storage_blocks_bytes", func(v float64) bool { return v >= 1 }},
{model.MetricTypeCounter, "prometheus_tsdb_reloads_total", func(v float64) bool { return v >= 1 }},
{model.MetricTypeCounter, "prometheus_tsdb_reloads_failures_total", func(v float64) bool { return v == 0 }},

{model.MetricTypeCounter, "prometheus_tsdb_head_chunks_created_total", func(v float64) bool { return v >= 1 }},
{model.MetricTypeCounter, "prometheus_tsdb_head_chunks_removed_total", func(v float64) bool { return v >= 1 }},
{model.MetricTypeCounter, "prometheus_tsdb_mmap_chunks_total", func(v float64) bool { return v >= 1 }},
{model.MetricTypeCounter, "prometheus_tsdb_mmap_chunk_corruptions_total", func(v float64) bool { return v == 0 }},

{model.MetricTypeCounter, "prometheus_tsdb_checkpoint_creations_total", func(v float64) bool { return v >= 1 }},
{model.MetricTypeCounter, "prometheus_tsdb_checkpoint_creations_failed_total", func(v float64) bool { return v == 0 }},
{model.MetricTypeCounter, "prometheus_tsdb_checkpoint_deletions_total", func(v float64) bool { return v >= 1 }},
{model.MetricTypeCounter, "prometheus_tsdb_checkpoint_deletions_failed_total", func(v float64) bool { return v == 0 }},

{model.MetricTypeCounter, "prometheus_tsdb_head_truncations_total", func(v float64) bool { return v >= 1 }},
{model.MetricTypeCounter, "prometheus_tsdb_head_truncations_failed_total", func(v float64) bool { return v == 0 }},

{model.MetricTypeGauge, "prometheus_tsdb_wal_storage_size_bytes", func(v float64) bool { return v >= 1 }},
{model.MetricTypeCounter, "prometheus_tsdb_wal_completed_pages_total", func(v float64) bool { return v >= 1 }},
{model.MetricTypeCounter, "prometheus_tsdb_wal_page_flushes_total", func(v float64) bool { return v >= 1 }},
{model.MetricTypeCounter, "prometheus_tsdb_wal_truncations_total", func(v float64) bool { return v >= 1 }},
{model.MetricTypeCounter, "prometheus_tsdb_wal_writes_failed_total", func(v float64) bool { return v == 0 }},
{model.MetricTypeCounter, "prometheus_tsdb_wal_corruptions_total", func(v float64) bool { return v == 0 }},
{model.MetricTypeCounter, "prometheus_tsdb_wal_truncations_failed_total", func(v float64) bool { return v == 0 }},
{model.MetricTypeGauge, "prometheus_ready", false, func(v float64) bool { return v == 1 }},
{model.MetricTypeGauge, "prometheus_config_last_reload_successful", false, func(v float64) bool { return v == 1 }},

{model.MetricTypeCounter, "prometheus_target_scrape_pools_total", false, func(v float64) bool { return v == 3 }},
{model.MetricTypeCounter, "prometheus_target_scrape_pools_failed_total", false, func(v float64) bool { return v == 0 }},
{model.MetricTypeCounter, "prometheus_target_scrape_pool_reloads_failed_total", false, func(v float64) bool { return v == 0 }},

{model.MetricTypeGauge, "prometheus_remote_storage_highest_timestamp_in_seconds", false, func(v float64) bool { return v > float64(checkStartTime.Unix()) }},

{model.MetricTypeGauge, "prometheus_rule_group_rules", false, func(v float64) bool { return v == 2 }},
{model.MetricTypeCounter, "prometheus_rule_evaluations_total", false, func(v float64) bool { return v >= 1 }},
{model.MetricTypeCounter, "prometheus_rule_evaluation_failures_total", false, func(v float64) bool { return v == 0 }},

{model.MetricTypeCounter, "prometheus_tsdb_compactions_triggered_total", true, func(v float64) bool { return v >= 1 }},
{model.MetricTypeCounter, "prometheus_tsdb_compactions_total", false, func(v float64) bool { return v >= 1 }},
{model.MetricTypeCounter, "prometheus_tsdb_compactions_failed_total", true, func(v float64) bool { return v == 0 }},

{model.MetricTypeGauge, "prometheus_tsdb_head_series", false, func(v float64) bool { return v >= 1 }},
{model.MetricTypeGauge, "prometheus_tsdb_head_chunks", false, func(v float64) bool { return v >= 1 }},
{model.MetricTypeGauge, "prometheus_tsdb_head_chunks_storage_size_bytes", false, func(v float64) bool { return v >= 1 }},
{model.MetricTypeCounter, "prometheus_tsdb_head_series_created_total", false, func(v float64) bool { return v >= 1 }},
{model.MetricTypeGauge, "prometheus_tsdb_blocks_loaded", false, func(v float64) bool { return v >= 1 }},
{model.MetricTypeGauge, "prometheus_tsdb_storage_blocks_bytes", false, func(v float64) bool { return v >= 1 }},
{model.MetricTypeCounter, "prometheus_tsdb_reloads_total", false, func(v float64) bool { return v >= 1 }},
{model.MetricTypeCounter, "prometheus_tsdb_reloads_failures_total", false, func(v float64) bool { return v == 0 }},

{model.MetricTypeCounter, "prometheus_tsdb_head_chunks_created_total", false, func(v float64) bool { return v >= 1 }},
{model.MetricTypeCounter, "prometheus_tsdb_head_chunks_removed_total", false, func(v float64) bool { return v >= 1 }},
{model.MetricTypeCounter, "prometheus_tsdb_mmap_chunks_total", false, func(v float64) bool { return v >= 1 }},
{model.MetricTypeCounter, "prometheus_tsdb_mmap_chunk_corruptions_total", false, func(v float64) bool { return v == 0 }},

{model.MetricTypeCounter, "prometheus_tsdb_checkpoint_creations_total", false, func(v float64) bool { return v >= 1 }},
{model.MetricTypeCounter, "prometheus_tsdb_checkpoint_creations_failed_total", false, func(v float64) bool { return v == 0 }},
{model.MetricTypeCounter, "prometheus_tsdb_checkpoint_deletions_total", false, func(v float64) bool { return v >= 1 }},
{model.MetricTypeCounter, "prometheus_tsdb_checkpoint_deletions_failed_total", false, func(v float64) bool { return v == 0 }},

{model.MetricTypeCounter, "prometheus_tsdb_head_truncations_total", false, func(v float64) bool { return v >= 1 }},
{model.MetricTypeCounter, "prometheus_tsdb_head_truncations_failed_total", false, func(v float64) bool { return v == 0 }},

{model.MetricTypeGauge, "prometheus_tsdb_wal_storage_size_bytes", false, func(v float64) bool { return v >= 1 }},
{model.MetricTypeCounter, "prometheus_tsdb_wal_completed_pages_total", false, func(v float64) bool { return v >= 1 }},
{model.MetricTypeCounter, "prometheus_tsdb_wal_page_flushes_total", false, func(v float64) bool { return v >= 1 }},
{model.MetricTypeCounter, "prometheus_tsdb_wal_truncations_total", false, func(v float64) bool { return v >= 1 }},
{model.MetricTypeCounter, "prometheus_tsdb_wal_writes_failed_total", false, func(v float64) bool { return v == 0 }},
{model.MetricTypeCounter, "prometheus_tsdb_wal_corruptions_total", false, func(v float64) bool { return v == 0 }},
{model.MetricTypeCounter, "prometheus_tsdb_wal_truncations_failed_total", false, func(v float64) bool { return v == 0 }},
} {
require.Eventually(t, func() bool {
val, err := getPrometheusMetricValue(t, c.prometheusPort, mc.mType, mc.mName)
val, err := getPrometheusMetricValue(t, c.prometheusPort, mc.mType, mc.mName, mc.isVecMetric)
if err != nil {
return false
}
Expand Down
10 changes: 5 additions & 5 deletions tsdb/compact_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1340,13 +1340,13 @@ func TestDisableAutoCompactions(t *testing.T) {
}

for range 10 {
if prom_testutil.ToFloat64(db.metrics.compactionsSkipped) > 0.0 {
if prom_testutil.ToFloat64(db.metrics.compactionsSkipped.WithLabelValues("head")) > 0.0 {
break
}
time.Sleep(10 * time.Millisecond)
}

require.Greater(t, prom_testutil.ToFloat64(db.metrics.compactionsSkipped), 0.0, "No compaction was skipped after the set timeout.")
require.Greater(t, prom_testutil.ToFloat64(db.metrics.compactionsSkipped.WithLabelValues("head")), 0.0, "No compaction was skipped after the set timeout.")
require.Empty(t, db.blocks)

// Enable the compaction, trigger it and check that the block is persisted.
Expand Down Expand Up @@ -1489,15 +1489,15 @@ func TestDeleteCompactionBlockAfterFailedReload(t *testing.T) {

require.Equal(t, 0.0, prom_testutil.ToFloat64(db.metrics.reloadsFailed), "initial 'failed db reloadBlocks' count metrics mismatch")
require.Equal(t, 0.0, prom_testutil.ToFloat64(db.compactor.(*LeveledCompactor).metrics.Ran), "initial `compactions` count metric mismatch")
require.Equal(t, 0.0, prom_testutil.ToFloat64(db.metrics.compactionsFailed), "initial `compactions failed` count metric mismatch")
require.Equal(t, 0.0, prom_testutil.ToFloat64(db.metrics.compactionsFailed.WithLabelValues("head")), "initial `compactions failed` count metric mismatch")

// Do the compaction and check the metrics.
// Compaction should succeed, but the reloadBlocks should fail and
// the new block created from the compaction should be deleted.
require.Error(t, db.Compact(ctx))
require.Equal(t, 1.0, prom_testutil.ToFloat64(db.metrics.reloadsFailed), "'failed db reloadBlocks' count metrics mismatch")
require.Equal(t, 1.0, prom_testutil.ToFloat64(db.compactor.(*LeveledCompactor).metrics.Ran), "`compaction` count metric mismatch")
require.Equal(t, 1.0, prom_testutil.ToFloat64(db.metrics.compactionsFailed), "`compactions failed` count metric mismatch")
require.Equal(t, 1.0, prom_testutil.ToFloat64(db.metrics.compactionsFailed.WithLabelValues("head")), "`compactions failed` count metric mismatch")

actBlocks, err = blockDirs(db.Dir())
require.NoError(t, err)
Expand Down Expand Up @@ -2054,7 +2054,7 @@ func TestDelayedCompaction(t *testing.T) {
require.Equal(t, 1.0, compactorRanCount(db))
// The compaction delay doesn't block or wait on the trigger;
// 3 triggers were processed above without blocking.
require.GreaterOrEqual(t, prom_testutil.ToFloat64(db.metrics.compactionsTriggered)-prom_testutil.ToFloat64(db.metrics.compactionsSkipped), 3.0)
require.GreaterOrEqual(t, prom_testutil.ToFloat64(db.metrics.compactionsTriggered.WithLabelValues("head"))-prom_testutil.ToFloat64(db.metrics.compactionsSkipped.WithLabelValues("head")), 3.0)
// The delay doesn't change the head blocks alignment.
require.Equal(t, db.compactor.(*LeveledCompactor).ranges[0]+1, db.head.MinTime())

Expand Down
Loading
Loading