Function TestConfig_ValidateRingMode

modules/generator/config_test.go:302–341

(t *testing.T)

}

func TestConfig_ValidateRingMode(t *testing.T) {
	cfg := &Config{}
	cfg.RegisterFlagsAndApplyDefaults("", flag.NewFlagSet("", flag.PanicOnError))
	cfg.Storage.Path = t.TempDir()

	t.Run("valid partition ring mode", func(t *testing.T) {
		cfg := *cfg
		cfg.RingMode = RingModePartition
		require.NoError(t, cfg.Validate())
	})

	t.Run("valid generator ring mode without kafka", func(t *testing.T) {
		cfg := *cfg
		cfg.RingMode = RingModeGenerator
		require.NoError(t, cfg.Validate())
	})

	t.Run("valid generator ring mode with configured consumer group", func(t *testing.T) {
		cfg := *cfg
		cfg.ConsumeFromKafka = true
		cfg.RingMode = RingModeGenerator
		cfg.Ingest.Kafka.Topic = "tempo"
		cfg.Ingest.Kafka.ConsumerGroup = "generator-a"
		require.NoError(t, cfg.Validate())
	})

	t.Run("generator ring mode requires consumer group when consuming from kafka", func(t *testing.T) {
		cfg := *cfg
		cfg.ConsumeFromKafka = true
		cfg.RingMode = RingModeGenerator
		cfg.Ingest.Kafka.Topic = "tempo"
		require.EqualError(t, cfg.Validate(), "ingest.kafka.consumer_group must be configured when metrics-generator ring mode is generator")
	})

	t.Run("invalid ring mode", func(t *testing.T) {
		cfg := *cfg
		cfg.RingMode = RingMode("invalid")
		require.EqualError(t, cfg.Validate(), "invalid ring mode: invalid, valid values are partition and generator")
	})
}

func boolPtr(b bool) *bool {
	return &b