Implement auto-forget in Parquet converter lifecycle (cortexproject#7752)

* implement auto-forget in parquet converter ring lifecycler

Signed-off-by: SungJin1212 <tjdwls1201@gmail.com>

* clean annotations

Signed-off-by: SungJin1212 <tjdwls1201@gmail.com>

---------

Signed-off-by: SungJin1212 <tjdwls1201@gmail.com>

Author: SungJin1212

CHANGELOG.md

@@ -87,6 +87,7 @@
 * [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] Alertmanager: Fix panic in `validateAlertmanagerConfig` when receiver config traversal encounters nil interface values. #7751
+* [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
 

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")
 	}

pkg/parquetconverter/converter_test.go

@@ -682,3 +682,55 @@ 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
+
+		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)
+}