Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
1 change: 1 addition & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -86,6 +86,7 @@
* [BUGFIX] Distributor: Return HTTP 499 (Client Closed Request) instead of 500 when a remote-write or OTLP push is canceled by the client, so client-side cancellations are no longer counted as server-side errors. #7717
* [BUGFIX] Querier: Fix gRPC `codes.Canceled` errors being mapped to HTTP 500 instead of 499 when a client cancels a query. #7738
* [BUGFIX] Compactor: Fix spurious `bucket operation fail after retries` error logs emitted during partial block cleanup. #7749
* [BUGFIX] Parquet Converter: Fix `auto_forget_delay` having no effect. The ring lifecycler was created without the auto-forget delegate, so unhealthy instances were never automatically removed from the ring. #7752

## 1.21.1 2026-06-04

Expand Down
7 changes: 6 additions & 1 deletion pkg/parquetconverter/converter.go
Original file line number Diff line number Diff line change
Expand Up @@ -174,7 +174,12 @@ func newConverter(cfg Config, bkt objstore.InstrumentedBucket, storageCfg cortex
func (c *Converter) starting(ctx context.Context) error {
lifecyclerCfg := c.cfg.Ring.ToLifecyclerConfig()
var err error
c.ringLifecycler, err = ring.NewLifecycler(lifecyclerCfg, ring.NewNoopFlushTransferer(), "parquet-converter", ringKey, true, false, c.logger, prometheus.WrapRegistererWithPrefix("cortex_", c.reg))
var delegate ring.LifecyclerDelegate
delegate = &ring.DefaultLifecyclerDelegate{}
if c.cfg.Ring.AutoForgetDelay > 0 {
delegate = ring.NewLifecyclerAutoForgetDelegate(c.cfg.Ring.AutoForgetDelay, delegate, c.logger)
}
c.ringLifecycler, err = ring.NewLifecyclerWithDelegate(lifecyclerCfg, ring.NewNoopFlushTransferer(), "parquet-converter", ringKey, true, false, c.logger, prometheus.WrapRegistererWithPrefix("cortex_", c.reg), delegate)
if err != nil {
return errors.Wrap(err, "unable to initialize converter ring lifecycler")
}
Expand Down
54 changes: 54 additions & 0 deletions pkg/parquetconverter/converter_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -682,3 +682,57 @@ func TestConvertWithMaxNumColumns(t *testing.T) {
require.NoError(t, err)
require.Equal(t, 1, shards2, "expected single shard with high column limit")
}

func TestConverter_RingLifecyclerShouldAutoForgetUnhealthyInstances(t *testing.T) {
// Create a shared KV Store
kvstore, closer := consul.NewInMemoryClient(ring.GetCodec(), log.NewNopLogger(), nil)
t.Cleanup(func() { assert.NoError(t, closer.Close()) })

bucketClient := objstore.WithNoopInstr(objstore.NewInMemBucket())
limits := &validation.Limits{}
flagext.DefaultValues(limits)
limits.ParquetConverterEnabled = true

// Create two converters
var converters []*Converter
for i := range 2 {
cfg := prepareConfig()
cfg.Ring.InstanceID = fmt.Sprintf("parquet-converter-%d", i)
cfg.Ring.InstanceAddr = fmt.Sprintf("127.0.0.%d", i+1)
cfg.Ring.KVStore.Mock = kvstore
cfg.Ring.HeartbeatPeriod = 100 * time.Millisecond
cfg.Ring.HeartbeatTimeout = 200 * time.Millisecond
cfg.Ring.AutoForgetDelay = 400 * time.Millisecond
// UnregisterOnShutdown=false lets the instance stay in the ring
// after stopping, so we can verify auto-forget kicks it out.

c, _, _ := prepare(t, cfg, bucketClient, limits, nil)
converters = append(converters, c)
}

// Start both converters.
require.NoError(t, services.StartAndAwaitRunning(context.Background(), converters[0]))
require.NoError(t, services.StartAndAwaitRunning(context.Background(), converters[1]))

// Both should be healthy.
test.Poll(t, 5*time.Second, true, func() any {
healthy, unhealthy, _ := converters[0].ring.GetAllInstanceDescs(ring.Reporting)
return len(healthy) == 2 && len(unhealthy) == 0
})

// Override UnregisterOnShutdown so the instance stays in the ring after stop,
// simulating a crash or ungraceful shutdown.
converters[1].ringLifecycler.SetUnregisterOnShutdown(false)
// The converter running() returns ctx.Err() on stop, so context.Canceled is expected.
err := services.StopAndAwaitTerminated(context.Background(), converters[1])
require.True(t, err == nil || errors.Is(err, context.Canceled), "unexpected error stopping converter: %v", err)

// The stopped instance should appear unhealthy first, then be auto-forgotten.
test.Poll(t, 5*time.Second, true, func() any {
healthy, unhealthy, _ := converters[0].ring.GetAllInstanceDescs(ring.Reporting)
return len(healthy) == 1 && len(unhealthy) == 0
})

err = services.StopAndAwaitTerminated(context.Background(), converters[0])
require.True(t, err == nil || errors.Is(err, context.Canceled), "unexpected error stopping converter: %v", err)
}