Update kafka-go settings to be compatible to latest Kafka Helm Charts (#459)

* Update kafka-go settings to be compatible to latest Kafka Helm Charts

Signed-off-by: Arrobo, Gabriel <[email protected]>

* Address Copilot''s comments

Signed-off-by: Arrobo, Gabriel <[email protected]>

---------

Signed-off-by: Arrobo, Gabriel <[email protected]>

Author: gab-arrobo

VERSION

@@ -1 +1 @@
-3.0.1-dev
+3.0.1

metrics/kafka.go

@@ -2,7 +2,6 @@
 //
 // SPDX-License-Identifier: Apache-2.0
 
-// Kafka metric Producer
 package metrics
 
 import (
@@ -25,35 +24,38 @@ var StatWriter Writer
 
 func InitialiseKafkaStream(config *factory.Configuration) error {
 	if !*config.KafkaInfo.EnableKafka {
-		logger.KafkaLog.Info("Kafka disabled")
+		logger.KafkaLog.Warnln("Kafka is disabled")
 		return nil
 	}
 
-	brokerUrl := "sd-core-kafka-headless:9092"
+	brokerUrl := "kafka:9092"
 	topicName := "sdcore-data-source-smf"
 
 	if config.KafkaInfo.BrokerUri != "" && config.KafkaInfo.BrokerPort != 0 {
 		brokerUrl = fmt.Sprintf("%s:%d", config.KafkaInfo.BrokerUri, config.KafkaInfo.BrokerPort)
 	}
 
-	logger.KafkaLog.Debugf("initialise kafka broker url [%v]", brokerUrl)
+	logger.KafkaLog.Debugf("initialise kafka broker url: %s", brokerUrl)
 
 	if config.KafkaInfo.Topic != "" {
 		topicName = config.KafkaInfo.Topic
 	}
 
-	logger.KafkaLog.Debugf("initialise kafka Topic [%v]", config.KafkaInfo.Topic)
+	logger.KafkaLog.Debugf("initialise kafka Topic: %s", topicName)
 
 	producer := kafka.Writer{
-		Addr:         kafka.TCP(brokerUrl),
-		Topic:        topicName,
-		Balancer:     &kafka.LeastBytes{},
-		BatchTimeout: 10 * time.Millisecond,
+		Addr:                   kafka.TCP(brokerUrl),
+		Topic:                  topicName,
+		AllowAutoTopicCreation: true,
+		Balancer:               &kafka.LeastBytes{},
+		BatchTimeout:           10 * time.Millisecond,
 	}
 
 	StatWriter = Writer{
 		kafkaWriter: &producer,
 	}
+
+	logger.KafkaLog.Debugf("initialising kafka stream with url[%s], topic[%s]", brokerUrl, topicName)
 	return nil
 }
 
@@ -67,26 +69,25 @@ func (writer Writer) SendMessage(message []byte) error {
 	}
 	msg := kafka.Message{Value: message}
 	if err := writer.kafkaWriter.WriteMessages(context.Background(), msg); err != nil {
-		logger.KafkaLog.Errorf("kafka send message write error: [%v] ", err.Error())
+		logger.KafkaLog.Errorf("kafka send message write error: %s", err.Error())
 		return err
 	}
 	return nil
 }
 
-// caller should make sure kafka is enabled.
 func (writer Writer) PublishPduSessEvent(ctxt mi.CoreSubscriber, op mi.SubscriberOp) error {
 	smKafkaEvt := mi.MetricEvent{
 		EventType:      mi.CSubscriberEvt,
 		SubscriberData: mi.CoreSubscriberData{Subscriber: ctxt, Operation: op},
 	}
 	if msg, err := json.Marshal(smKafkaEvt); err != nil {
-		logger.KafkaLog.Errorf("publishing pdu sess event marshal error [%v] ", err.Error())
+		logger.KafkaLog.Errorf("publishing pdu sess event marshal error: %s", err.Error())
 		return err
 	} else {
-		logger.KafkaLog.Debugf("publishing pdu sess event[%s] ", msg)
+		logger.KafkaLog.Debugf("publishing pdu sess event[%s]", string(msg))
 		err := StatWriter.SendMessage(msg)
 		if err != nil {
-			logger.KafkaLog.Errorf("publishing pdu sess event error [%v] ", err.Error())
+			logger.KafkaLog.Errorf("publishing pdu sess event error: %s", err.Error())
 		}
 	}
 	return nil
@@ -105,28 +106,26 @@ func PublishMsgEvent(msgType mi.SmfMsgType) error {
 	}
 	smKafkaMsgEvt := mi.MetricEvent{EventType: mi.CMsgTypeEvt, MsgType: mi.CoreMsgType{MsgType: msgType.String(), SourceNfId: nfInstanceId}}
 	if msg, err := json.Marshal(smKafkaMsgEvt); err != nil {
-		logger.KafkaLog.Errorf("publishing msg event marshal error [%v] ", err.Error())
+		logger.KafkaLog.Errorf("publishing msg event marshal error: %s", err.Error())
 		return err
 	} else {
-		logger.KafkaLog.Debugf("publishing msg event[%s] ", msg)
+		logger.KafkaLog.Debugf("publishing msg event: %s", string(msg))
 		err := StatWriter.SendMessage(msg)
 		if err != nil {
-			logger.KafkaLog.Errorf("publishing msg event error [%v] ", err.Error())
+			logger.KafkaLog.Errorf("publishing msg event error: %s", err.Error())
 		}
 	}
 	return nil
 }
 
-// caller should make sure kafka is enabled.
 func (writer Writer) PublishNfStatusEvent(msgEvent mi.MetricEvent) error {
 	if msg, err := json.Marshal(msgEvent); err != nil {
-		logger.KafkaLog.Errorf("publishing nf status marshal error [%v] ", err.Error())
+		logger.KafkaLog.Errorf("publishing nf status marshal error: %s", err.Error())
 		return err
 	} else {
-		logger.KafkaLog.Debugf("publishing nf status event[%s] ", msg)
-		err := StatWriter.SendMessage(msg)
-		if err != nil {
-			logger.KafkaLog.Errorf("publishing nf status event error [%v] ", err.Error())
+		logger.KafkaLog.Debugf("publishing nf status event: %s", string(msg))
+		if err := StatWriter.SendMessage(msg); err != nil {
+			logger.KafkaLog.Errorf("publishing nf status event error: %s", err.Error())
 		}
 	}
 	return nil