(feat) return a default topic when tenant_id is not present (#304)

* (feat) return a default topic when tenant_id is not present

* (refactor) rename the function to getKafkaTopicPrefixFromClientMetadata making it consistent with what the function is doing

.gitignore

@@ -12,7 +12,7 @@
 *.out
 
 # Dependency directories (remove the comment below to include it)
-# vendor/
+vendor/
 
 .vscode
 .DS_Store
@@ -22,4 +22,4 @@
 # build folder
 .build
 *.rollback
-signoz-collector
+signoz-collector

exporter/signozkafkaexporter/kafka_exporter.go

@@ -39,10 +39,8 @@ func (ke kafkaErrors) Error() string {
 }
 
 func (e *kafkaTracesProducer) tracesPusher(ctx context.Context, td ptrace.Traces) error {
-	kafkaTopicPrefix, err := getKafkaTopicFromClientMetadata(client.FromContext(ctx).Metadata)
-	if err != nil {
-		return consumererror.NewPermanent(err)
-	}
+	kafkaTopicPrefix := getKafkaTopicPrefixFromClientMetadata(client.FromContext(ctx).Metadata)
+
 	kafkaTopic := fmt.Sprintf("%s_traces", kafkaTopicPrefix)
 	messages, err := e.marshaler.Marshal(td, kafkaTopic)
 	if err != nil {
@@ -74,10 +72,8 @@ type kafkaMetricsProducer struct {
 }
 
 func (e *kafkaMetricsProducer) metricsDataPusher(ctx context.Context, md pmetric.Metrics) error {
-	kafkaTopicPrefix, err := getKafkaTopicFromClientMetadata(client.FromContext(ctx).Metadata)
-	if err != nil {
-		return consumererror.NewPermanent(err)
-	}
+	kafkaTopicPrefix := getKafkaTopicPrefixFromClientMetadata(client.FromContext(ctx).Metadata)
+
 	kafkaTopic := fmt.Sprintf("%s_metrics", kafkaTopicPrefix)
 	messages, err := e.marshaler.Marshal(md, kafkaTopic)
 	if err != nil {
@@ -111,10 +107,8 @@ type kafkaLogsProducer struct {
 func (e *kafkaLogsProducer) logsDataPusher(ctx context.Context, ld plog.Logs) error {
 	e.normalizeLogData(&ld)
 
-	kafkaTopicPrefix, err := getKafkaTopicFromClientMetadata(client.FromContext(ctx).Metadata)
-	if err != nil {
-		return consumererror.NewPermanent(err)
-	}
+	kafkaTopicPrefix := getKafkaTopicPrefixFromClientMetadata(client.FromContext(ctx).Metadata)
+
 	kafkaTopic := fmt.Sprintf("%s_logs", kafkaTopicPrefix)
 	messages, err := e.marshaler.Marshal(ld, kafkaTopic)
 	if err != nil {

exporter/signozkafkaexporter/utils.go

@@ -1,17 +1,20 @@
 package signozkafkaexporter
 
 import (
-	"fmt"
-
 	"go.opentelemetry.io/collector/client"
 )
 
+const (
+	DefaultKafkaTopicPrefix = "default"
+)
+
 // getKafkaTopicFromClientMetadata returns the kafka topic from client metadata
-func getKafkaTopicFromClientMetadata(md client.Metadata) (string, error) {
+func getKafkaTopicPrefixFromClientMetadata(md client.Metadata) string {
 	// return default topic if no tenant id is found in client metadata
 	signozTenantId := md.Get("signoz_tenant_id")
-	if len(signozTenantId) == 0 {
-		return "", fmt.Errorf("signoz_tenant_id not found in client metadata")
+	if len(signozTenantId) != 0 {
+		return signozTenantId[0]
 	}
-	return signozTenantId[0], nil
+
+	return DefaultKafkaTopicPrefix
 }