feat : emit total delivery message count

Author: siddhanta-rath

app/server.go

@@ -37,6 +37,7 @@ func StartServer(ctx context.Context, cancel context.CancelFunc, shutdown chan b
 	workerPool.StartWorkers()
 	go kPublisher.ReportStats()
 	go reportProcMetrics()
+	go kPublisher.ReportDeliveryEventCount()
 	// create signal channel at startup
 	signalChan := make(chan os.Signal, 1)
 	signal.Notify(signalChan, syscall.SIGHUP, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT)

proto/TotalEventCount.pb.go

@@ -0,0 +1,11 @@
+syntax = "proto3";
+import "google/protobuf/timestamp.proto";
+
+package github.com.goto.raccoon;
+
+option go_package = "github.com/goto/raccoon/proto";
+
+message TotalEventCountMessage {
+  google.protobuf.Timestamp event_timestamp = 1; //partition column
+  int32 event_count = 2;
+}

proto/TotalEventCount.proto

@@ -3,7 +3,12 @@ package publisher
 import (
 	"encoding/json"
 	"fmt"
+	"github.com/goto/raccoon/proto"
+	"github.com/goto/raccoon/serialization"
+	"google.golang.org/protobuf/types/known/timestamppb"
 	"strings"
+	"sync/atomic"
+	"time"
 
 	"gopkg.in/confluentinc/confluent-kafka-go.v1/kafka"
 	// Importing librd to make it work on vendor mode
@@ -20,6 +25,8 @@ const (
 	errLargeMessageSize = "Broker: Message size too large" //error msg while producing a message which is larger than message.max.bytes config
 )
 
+var DeliveryEventCount atomic.Int64
+
 // KafkaProducer Produce data to kafka synchronously
 type KafkaProducer interface {
 	// ProduceBulk message to kafka. Block until all messages are sent. Return array of error. Order is not guaranteed.
@@ -100,6 +107,7 @@ func (pr *Kafka) ProduceBulk(events []*pb.Event, connGroup string, deliveryChann
 			order := m.Opaque.(int)
 			errors[order] = m.TopicPartition.Error
 		}
+		DeliveryEventCount.Add(1)
 	}
 
 	if allNil(errors) {
@@ -108,6 +116,17 @@ func (pr *Kafka) ProduceBulk(events []*pb.Event, connGroup string, deliveryChann
 	return BulkError{Errors: errors}
 }
 
+func (pr *Kafka) ProduceTotalEventMessage(topicName string, event *proto.TotalEventCountMessage) error {
+	value, err := serialization.SerializeProto(event)
+	if err != nil {
+		return fmt.Errorf("failed to serialize proto: %w", err)
+	}
+	return pr.kp.Produce(&kafka.Message{
+		TopicPartition: kafka.TopicPartition{Topic: &topicName, Partition: kafka.PartitionAny},
+		Value:          value,
+	}, nil)
+}
+
 func (pr *Kafka) ReportStats() {
 	for v := range pr.kp.Events() {
 		switch e := v.(type) {
@@ -141,6 +160,24 @@ func (pr *Kafka) ReportStats() {
 	}
 }
 
+func (pr *Kafka) ReportDeliveryEventCount() {
+	ticker := time.NewTicker(1 * time.Minute)
+	defer ticker.Stop()
+
+	for range ticker.C {
+		fmt.Println("get the current value:", DeliveryEventCount.Load())
+		//build kafka message
+		msg := &proto.TotalEventCountMessage{
+			EventTimestamp: timestamppb.Now(),
+			EventCount:     int32(DeliveryEventCount.Load()),
+		}
+		//produce to kafka
+		pr.ProduceTotalEventMessage("clickstream-total-event", msg)
+		//reset the counter
+		DeliveryEventCount.Store(0)
+	}
+}
+
 // Close wait for outstanding messages to be delivered within given flush interval timeout.
 func (pr *Kafka) Close() int {
 	remaining := pr.kp.Flush(pr.flushInterval)