Merge pull request #1740 from openmeterio/refactor/topic-provisioning

refactor(kafka): topic provisioning

Author: chrisgacsal

app/common/kafka.go

@@ -65,11 +65,12 @@ func NewKafkaTopicProvisionerConfig(
 	settings config.TopicProvisionerConfig,
 ) pkgkafka.TopicProvisionerConfig {
 	return pkgkafka.TopicProvisionerConfig{
-		AdminClient: adminClient,
-		Logger:      logger,
-		Meter:       meter,
-		CacheSize:   settings.CacheSize,
-		CacheTTL:    settings.CacheTTL,
+		AdminClient:     adminClient,
+		Logger:          logger,
+		Meter:           meter,
+		CacheSize:       settings.CacheSize,
+		CacheTTL:        settings.CacheTTL,
+		ProtectedTopics: settings.ProtectedTopics,
 	}
 }
 

app/common/openmeter.go

@@ -123,12 +123,13 @@ func NewIngestCollector(
 func NewKafkaNamespaceHandler(
 	topicResolver topicresolver.Resolver,
 	topicProvisioner pkgkafka.TopicProvisioner,
-	conf config.Configuration,
+	conf config.KafkaIngestConfiguration,
 ) (*kafkaingest.NamespaceHandler, error) {
 	return &kafkaingest.NamespaceHandler{
 		TopicResolver:    topicResolver,
 		TopicProvisioner: topicProvisioner,
-		Partitions:       conf.Ingest.Kafka.Partitions,
+		Partitions:       conf.Partitions,
+		DeletionEnabled:  conf.NamespaceDeletionEnabled,
 	}, nil
 }
 

app/config/config_test.go

@@ -104,6 +104,10 @@ func TestComplete(t *testing.T) {
 				TopicProvisionerConfig: TopicProvisionerConfig{
 					CacheSize: 200,
 					CacheTTL:  15 * time.Minute,
+					ProtectedTopics: []string{
+						"protected-topic-1",
+						"protected-topic-2",
+					},
 				},
 			},
 		},

app/config/ingest.go

@@ -34,6 +34,9 @@ type KafkaIngestConfiguration struct {
 
 	Partitions          int
 	EventsTopicTemplate string
+
+	// NamespaceDeletionEnabled defines whether deleting namespaces are allowed or not.
+	NamespaceDeletionEnabled bool
 }
 
 // Validate validates the configuration.
@@ -166,6 +169,7 @@ func ConfigureIngest(v *viper.Viper) {
 	v.SetDefault("ingest.kafka.saslPassword", "")
 	v.SetDefault("ingest.kafka.partitions", 1)
 	v.SetDefault("ingest.kafka.eventsTopicTemplate", "om_%s_events")
+	v.SetDefault("ingest.kafka.namespaceDeletionEnabled", false)
 
 	ConfigureTopicProvisioner(v, "ingest", "kafka")
 }

app/config/testdata/complete.yaml

@@ -45,6 +45,9 @@ ingest:
       - consumer
     cacheSize: 200
     cacheTTL: 15m
+    protectedTopics:
+      - "protected-topic-1"
+      - "protected-topic-2"
 
 aggregation:
   clickhouse:

app/config/topicprovisioner.go

@@ -16,6 +16,9 @@ type TopicProvisionerConfig struct {
 
 	// The maximum time an entries is kept in cache before being evicted
 	CacheTTL time.Duration
+
+	// ProtectedTopics defines a list of topics which are protected from deletion.
+	ProtectedTopics []string
 }
 
 func (c TopicProvisionerConfig) Validate() error {
@@ -38,4 +41,5 @@ func ConfigureTopicProvisioner(v *viper.Viper, prefixes ...string) {
 
 	v.SetDefault(prefixer("cacheSize"), 250)
 	v.SetDefault(prefixer("cacheTTL"), "5m")
+	v.SetDefault(prefixer("protectedTopics"), nil)
 }

cmd/server/wire_gen.go

@@ -15,6 +15,9 @@ type NamespaceHandler struct {
 	TopicProvisioner pkgkafka.TopicProvisioner
 
 	Partitions int
+
+	// DeletionEnabled defines whether deleting namespaces are allowed or not.
+	DeletionEnabled bool
 }
 
 // CreateNamespace implements the namespace handler interface.
@@ -41,6 +44,10 @@ func (h NamespaceHandler) CreateNamespace(ctx context.Context, namespace string)
 
 // DeleteNamespace implements the namespace handler interface.
 func (h NamespaceHandler) DeleteNamespace(ctx context.Context, namespace string) error {
+	if !h.DeletionEnabled {
+		return nil
+	}
+
 	if h.TopicResolver == nil {
 		return errors.New("topic name resolver must not be nil")
 	}

openmeter/ingest/kafkaingest/namespace.go

@@ -37,8 +37,13 @@ type TopicProvisioner interface {
 	DeProvision(ctx context.Context, topics ...string) error
 }
 
+type AdminClient interface {
+	CreateTopics(ctx context.Context, topics []kafka.TopicSpecification, options ...kafka.CreateTopicsAdminOption) ([]kafka.TopicResult, error)
+	DeleteTopics(ctx context.Context, topics []string, options ...kafka.DeleteTopicsAdminOption) ([]kafka.TopicResult, error)
+}
+
 type TopicProvisionerConfig struct {
-	AdminClient *kafka.AdminClient
+	AdminClient AdminClient
 	Logger      *slog.Logger
 	Meter       metric.Meter
 
@@ -49,6 +54,9 @@ type TopicProvisionerConfig struct {
 	// CacheTTL stores maximum time an entries is kept in cache before being evicted.
 	// Setting it to 0 disables cache entry expiration.
 	CacheTTL time.Duration
+
+	// ProtectedTopics defines a list of topics which are protected from deletion.
+	ProtectedTopics []string
 }
 
 // NewTopicProvisioner returns a new TopicProvisioner.
@@ -65,9 +73,15 @@ func NewTopicProvisioner(config TopicProvisionerConfig) (TopicProvisioner, error
 		return nil, errors.New("meter is required")
 	}
 
+	protectedTopics := make(map[string]struct{}, len(config.ProtectedTopics))
+	for _, protectedTopic := range config.ProtectedTopics {
+		protectedTopics[protectedTopic] = struct{}{}
+	}
+
 	provisioner := &topicProvisioner{
-		client: config.AdminClient,
-		logger: config.Logger,
+		client:          config.AdminClient,
+		logger:          config.Logger,
+		protectedTopics: protectedTopics,
 	}
 
 	provisioner.cache = expirable.NewLRU[string, struct{}](config.CacheSize, provisioner.evictCallback, config.CacheTTL)
@@ -128,10 +142,12 @@ var _ TopicProvisioner = (*topicProvisioner)(nil)
 // to keep track of topics which have been already provisioned. This allows the provisioner being called multiple times with the same
 // set of topics without need for extra round-trip to Kafka brokers for sub-sequential calls.
 type topicProvisioner struct {
-	client *kafka.AdminClient
+	client AdminClient
 	logger *slog.Logger
 	cache  *expirable.LRU[string, struct{}]
 
+	protectedTopics map[string]struct{}
+
 	metrics struct {
 		// Errors
 		Errors metric.Int64Counter
@@ -233,13 +249,25 @@ func (p *topicProvisioner) DeProvision(ctx context.Context, topics ...string) er
 		return nil
 	}
 
+	topicsToDelete := make([]string, 0, len(topics))
 	for _, topic := range topics {
 		if topic == "" {
-			return fmt.Errorf("invalid topic name %q", topic)
+			p.logger.Warn("skip topic: empty topic name", "topic", topic)
+
+			continue
+		}
+
+		// Skip protected topics to avoid accidental deletion
+		if _, ok := p.protectedTopics[topic]; ok {
+			p.logger.Info("skip topic: protected topic", "topic", topic)
+
+			continue
 		}
+
+		topicsToDelete = append(topicsToDelete, topic)
 	}
 
-	results, err := p.client.DeleteTopics(ctx, topics)
+	results, err := p.client.DeleteTopics(ctx, topicsToDelete)
 	if err != nil {
 		p.metrics.Errors.Add(ctx, 1, metric.WithAttributes(attribute.String("scope", "client")))
 

pkg/kafka/topicprovisioner.go

@@ -0,0 +1,149 @@
+package kafka
+
+import (
+	"context"
+	"log/slog"
+	"testing"
+	"time"
+
+	"github.com/confluentinc/confluent-kafka-go/v2/kafka"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/otel/metric/noop"
+)
+
+// FIXME(chrisgacsal): move discardHandler to 'testutils' pkg after import cycle is resolved.
+// discardHandler is a slog.Handler implementation which does not emit log messages
+// See: https://go-review.googlesource.com/c/go/+/548335/5/src/log/slog/example_discard_test.go#14
+type discardHandler struct {
+	slog.JSONHandler
+}
+
+func (d *discardHandler) Enabled(context.Context, slog.Level) bool { return false }
+
+func NewDiscardLogger(t testing.TB) *slog.Logger {
+	t.Helper()
+
+	return slog.New(&discardHandler{})
+}
+
+var _ AdminClient = (*mockTopicProvisioner)(nil)
+
+type mockTopicProvisioner struct {
+	added   []string
+	removed []string
+}
+
+func (m *mockTopicProvisioner) CreateTopics(_ context.Context, topics []kafka.TopicSpecification, _ ...kafka.CreateTopicsAdminOption) ([]kafka.TopicResult, error) {
+	result := make([]kafka.TopicResult, 0, len(topics))
+
+	for _, topic := range topics {
+		m.added = append(m.added, topic.Topic)
+
+		result = append(result, kafka.TopicResult{
+			Topic: topic.Topic,
+			Error: kafka.NewError(kafka.ErrNoError, "", false),
+		})
+	}
+
+	return result, nil
+}
+
+func (m *mockTopicProvisioner) DeleteTopics(_ context.Context, topics []string, _ ...kafka.DeleteTopicsAdminOption) ([]kafka.TopicResult, error) {
+	result := make([]kafka.TopicResult, 0, len(topics))
+
+	for _, topic := range topics {
+		m.removed = append(m.removed, topic)
+
+		result = append(result, kafka.TopicResult{
+			Topic: topic,
+			Error: kafka.NewError(kafka.ErrNoError, "", false),
+		})
+	}
+
+	return result, nil
+}
+
+func (m *mockTopicProvisioner) reset() {
+	m.added, m.removed = []string{}, []string{}
+}
+
+func TestTopicProvisioner(t *testing.T) {
+	tests := []struct {
+		Name string
+
+		AddTopics    []TopicConfig
+		RemoveTopics []string
+
+		ExpectedError         error
+		ExpectedAddedTopics   []string
+		ExpectedRemovedTopics []string
+	}{
+		{
+			Name: "Add topics",
+			AddTopics: []TopicConfig{
+				{
+					Name:       "topic-1",
+					Partitions: 1,
+				},
+
+				{
+					Name:       "topic-2",
+					Partitions: 1,
+				},
+			},
+			ExpectedError:         nil,
+			ExpectedAddedTopics:   []string{"topic-1", "topic-2"},
+			ExpectedRemovedTopics: []string{},
+		},
+		{
+			Name:                  "Remove topics",
+			RemoveTopics:          []string{"topic-1", "topic-2"},
+			ExpectedError:         nil,
+			ExpectedAddedTopics:   []string{},
+			ExpectedRemovedTopics: []string{"topic-1", "topic-2"},
+		},
+		{
+			Name:                  "Remove protected topics",
+			RemoveTopics:          []string{"protected-topic-1", "protected-topic-2"},
+			ExpectedError:         nil,
+			ExpectedAddedTopics:   []string{},
+			ExpectedRemovedTopics: []string{},
+		},
+	}
+
+	adminClient := &mockTopicProvisioner{}
+	meter := noop.NewMeterProvider().Meter("test")
+	logger := NewDiscardLogger(t)
+
+	provisioner, err := NewTopicProvisioner(TopicProvisionerConfig{
+		AdminClient: adminClient,
+		Logger:      logger,
+		Meter:       meter,
+		CacheSize:   200,
+		CacheTTL:    5 * time.Second,
+		ProtectedTopics: []string{
+			"protected-topic-1",
+			"protected-topic-2",
+		},
+	})
+	require.NoError(t, err, "initializing new topic provisioner should not fail")
+
+	for _, test := range tests {
+		t.Run(test.Name, func(t *testing.T) {
+			adminClient.reset()
+
+			ctx := context.TODO()
+
+			err = provisioner.Provision(ctx, test.AddTopics...)
+			assert.NoError(t, err, "provisioning topics must not fail")
+
+			assert.ElementsMatch(t, test.ExpectedAddedTopics, adminClient.added, "provisioned topics must match")
+
+			err = provisioner.DeProvision(ctx, test.RemoveTopics...)
+			assert.NoError(t, err, "de-provisioning topics must not fail")
+
+			assert.ElementsMatch(t, test.ExpectedRemovedTopics, adminClient.removed, "de-provisioned topics must match")
+		})
+	}
+}