diff --git a/CHANGELOG.md b/CHANGELOG.md index e35a4f42ab1..9c098bdc97d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/pkg/parquetconverter/converter.go b/pkg/parquetconverter/converter.go index 187ee0f22cd..5184fe9f68e 100644 --- a/pkg/parquetconverter/converter.go +++ b/pkg/parquetconverter/converter.go @@ -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") } diff --git a/pkg/parquetconverter/converter_test.go b/pkg/parquetconverter/converter_test.go index 5dd79bb60b9..73608e0d524 100644 --- a/pkg/parquetconverter/converter_test.go +++ b/pkg/parquetconverter/converter_test.go @@ -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) +}