@@ -5,75 +5,100 @@ import (
55 "sync"
66 "time"
77
8- "github.com/confluentinc/confluent-kafka-go/kafka"
98 "github.com/raystack/raccoon/collection"
109 "github.com/raystack/raccoon/logger"
1110 "github.com/raystack/raccoon/metrics"
11+ pb "github.com/raystack/raccoon/proto"
1212 "github.com/raystack/raccoon/publisher"
1313)
1414
15+ // Producer produces data to sink
16+ type Producer interface {
17+ // ProduceBulk message to a sink. Blocks until all messages are sent. Returns slice of error.
18+ ProduceBulk (events []* pb.Event , connGroup string ) error
19+ }
20+
1521// Pool spawn goroutine as much as Size that will listen to EventsChannel. On Close, wait for all data in EventsChannel to be processed.
1622type Pool struct {
1723 Size int
1824 deliveryChannelSize int
1925 EventsChannel <- chan collection.CollectRequest
20- kafkaProducer publisher. KafkaProducer
26+ producer Producer
2127 wg sync.WaitGroup
2228}
2329
2430// CreateWorkerPool create new Pool struct given size and EventsChannel worker.
25- func CreateWorkerPool (size int , eventsChannel <- chan collection.CollectRequest , deliveryChannelSize int , kafkaProducer publisher. KafkaProducer ) * Pool {
31+ func CreateWorkerPool (size int , eventsChannel <- chan collection.CollectRequest , deliveryChannelSize int , producer Producer ) * Pool {
2632 return & Pool {
2733 Size : size ,
2834 deliveryChannelSize : deliveryChannelSize ,
2935 EventsChannel : eventsChannel ,
30- kafkaProducer : kafkaProducer ,
36+ producer : producer ,
3137 wg : sync.WaitGroup {},
3238 }
3339}
3440
35- // StartWorkers initialize worker pool as much as Pool.Size
36- func (w * Pool ) StartWorkers () {
37- w .wg .Add (w .Size )
38- for i := 0 ; i < w .Size ; i ++ {
39- go func (workerName string ) {
40- logger .Info ("Running worker: " + workerName )
41- deliveryChan := make (chan kafka.Event , w .deliveryChannelSize )
42- for request := range w .EventsChannel {
43- metrics .Histogram ("batch_idle_in_channel_milliseconds" , (time .Now ().Sub (request .TimePushed )).Milliseconds (), map [string ]string {"worker" : workerName })
44- batchReadTime := time .Now ()
45- //@TODO - Should add integration tests to prove that the worker receives the same message that it produced, on the delivery channel it created
41+ func (w * Pool ) newWorker (name string ) {
4642
47- err := w .kafkaProducer .ProduceBulk (request .GetEvents (), request .ConnectionIdentifier .Group , deliveryChan )
43+ logger .Info ("Running worker: " + name )
44+ for request := range w .EventsChannel {
4845
49- produceTime := time .Since (batchReadTime )
50- metrics .Histogram ("kafka_producebulk_tt_ms" , produceTime .Milliseconds (), map [string ]string {})
46+ metrics .Histogram (
47+ "batch_idle_in_channel_milliseconds" ,
48+ time .Since (request .TimePushed ).Milliseconds (),
49+ map [string ]string {"worker" : name })
5150
52- if request .AckFunc != nil {
53- request .AckFunc (err )
54- }
51+ batchReadTime := time .Now ()
52+ //@TODO - Should add integration tests to prove that the worker receives the same message that it produced, on the delivery channel it created
53+
54+ err := w .producer .ProduceBulk (request .GetEvents (), request .ConnectionIdentifier .Group )
55+
56+ // TODO(turtledev): instrument this for individual sinks
57+ // produceTime := time.Since(batchReadTime)
58+ // metrics.Histogram("kafka_producebulk_tt_ms", produceTime.Milliseconds(), map[string]string{})
5559
56- totalErr := 0
60+ if request .AckFunc != nil {
61+ request .AckFunc (err )
62+ }
63+
64+ totalErr := 0
65+ if err != nil {
66+ // WARN(turtledev): this can panic if returned error is not of
67+ // type publisher.BulkError
68+ for _ , err := range err .(publisher.BulkError ).Errors {
5769 if err != nil {
58- for _ , err := range err .(publisher.BulkError ).Errors {
59- if err != nil {
60- logger .Errorf ("[worker] Fail to publish message to kafka %v" , err )
61- totalErr ++
62- }
63- }
64- }
65- lenBatch := int64 (len (request .GetEvents ()))
66- logger .Debug (fmt .Sprintf ("Success sending messages, %v" , lenBatch - int64 (totalErr )))
67- if lenBatch > 0 {
68- eventTimingMs := time .Since (request .GetSentTime ().AsTime ()).Milliseconds () / lenBatch
69- metrics .Histogram ("event_processing_duration_milliseconds" , eventTimingMs , map [string ]string {"conn_group" : request .ConnectionIdentifier .Group })
70- now := time .Now ()
71- metrics .Histogram ("worker_processing_duration_milliseconds" , (now .Sub (batchReadTime ).Milliseconds ())/ lenBatch , map [string ]string {"worker" : workerName })
72- metrics .Histogram ("server_processing_latency_milliseconds" , (now .Sub (request .TimeConsumed )).Milliseconds ()/ lenBatch , map [string ]string {"conn_group" : request .ConnectionIdentifier .Group })
70+ logger .Errorf ("[worker] Fail to publish message to kafka %v" , err )
71+ totalErr ++
7372 }
7473 }
75- w .wg .Done ()
76- }(fmt .Sprintf ("worker-%d" , i ))
74+ }
75+ lenBatch := int64 (len (request .GetEvents ()))
76+ logger .Debug (fmt .Sprintf ("Success sending messages, %v" , lenBatch - int64 (totalErr )))
77+ if lenBatch > 0 {
78+ eventTimingMs := time .Since (request .GetSentTime ().AsTime ()).Milliseconds () / lenBatch
79+ metrics .Histogram (
80+ "event_processing_duration_milliseconds" ,
81+ eventTimingMs ,
82+ map [string ]string {"conn_group" : request .ConnectionIdentifier .Group })
83+ now := time .Now ()
84+ metrics .Histogram (
85+ "worker_processing_duration_milliseconds" ,
86+ (now .Sub (batchReadTime ).Milliseconds ())/ lenBatch ,
87+ map [string ]string {"worker" : name })
88+ metrics .Histogram (
89+ "server_processing_latency_milliseconds" ,
90+ (now .Sub (request .TimeConsumed )).Milliseconds ()/ lenBatch ,
91+ map [string ]string {"conn_group" : request .ConnectionIdentifier .Group })
92+ }
93+ }
94+ w .wg .Done ()
95+ }
96+
97+ // StartWorkers initialize worker pool as much as Pool.Size
98+ func (w * Pool ) StartWorkers () {
99+ w .wg .Add (w .Size )
100+ for i := 0 ; i < w .Size ; i ++ {
101+ w .newWorker (fmt .Sprintf ("worker-%d" , i ))
77102 }
78103}
79104
0 commit comments