Skip to content
This repository was archived by the owner on Oct 7, 2022. It is now read-only.

Commit 3389d07

Browse files
racevedooracevedoo
authored
Merge pull request #4 from inloco/refactor/meetup
Refactor: separate metrics logic from consumer
2 parents 756be58 + e3991e5 commit 3389d07

File tree

7 files changed

+179
-137
lines changed

7 files changed

+179
-137
lines changed

‎cmd/injector.go‎

Lines changed: 9 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -9,13 +9,13 @@ import (
99
"os/signal"
1010
"syscall"
1111

12+
"github.com/go-kit/kit/log/level"
1213
"github.com/inloco/kafka-elasticsearch-injector/src/injector"
1314
"github.com/inloco/kafka-elasticsearch-injector/src/kafka"
1415
"github.com/inloco/kafka-elasticsearch-injector/src/logger_builder"
15-
"github.com/inloco/kafka-elasticsearch-injector/src/metrics_instrumenter"
16+
"github.com/inloco/kafka-elasticsearch-injector/src/metrics"
1617
"github.com/inloco/kafka-elasticsearch-injector/src/probes"
1718
"github.com/inloco/kafka-elasticsearch-injector/src/schema_registry"
18-
"github.com/go-kit/kit/log/level"
1919
)
2020

2121
func main() {
@@ -30,7 +30,7 @@ func main() {
3030
"message", fmt.Sprintf("Initializing kubernetes probes at %s", probesPort),
3131
)
3232
go p.Serve()
33-
metrics_instrumenter.Register()
33+
metrics.Register()
3434
schemaRegistry, err := schema_registry.NewSchemaRegistry(os.Getenv("SCHEMA_REGISTRY_URL"))
3535
if err != nil {
3636
level.Error(logger).Log("err", err, "message", "failed to create schema registry client")
@@ -64,8 +64,12 @@ func main() {
6464
notifications := make(chan kafka.Notification, 10)
6565
go func() {
6666
for {
67-
select {
68-
case _ = <-notifications: //ignore notifications
67+
ntf := <-notifications
68+
switch ntf {
69+
case kafka.Ready:
70+
level.Info(logger).Log("message", "kafka consumer ready")
71+
case kafka.Inserted:
72+
level.Info(logger).Log("message", fmt.Sprintf("inserted records"))
6973
}
7074
}
7175
}()

‎src/elasticsearch/elasticsearch.go‎

Lines changed: 29 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -50,27 +50,9 @@ func (d recordDatabase) CloseClient() {
5050
}
5151

5252
func (d recordDatabase) Insert(records []*models.Record) error {
53-
bulkRequest := d.GetClient().Bulk()
54-
for _, record := range records {
55-
indexName := d.config.Index
56-
if indexName == "" {
57-
indexName = record.Topic
58-
}
59-
indexColumn := d.config.IndexColumn
60-
indexColumnValue := record.FormatTimestamp()
61-
if indexColumn != "" {
62-
newIndexColumnValue, err := record.GetValueForField(indexColumn)
63-
if err != nil {
64-
level.Error(d.logger).Log("err", err, "message", "Could not get column value from record.")
65-
return err
66-
}
67-
indexColumnValue = newIndexColumnValue
68-
}
69-
index := fmt.Sprintf("%s-%s", indexName, indexColumnValue)
70-
bulkRequest.Add(elastic.NewBulkIndexRequest().Index(index).
71-
Type(record.Topic).
72-
Id(record.GetId()).
73-
Doc(record.FilteredFieldsJSON(d.config.BlacklistedColumns)))
53+
bulkRequest, err := d.buildBulkRequest(records)
54+
if err != nil {
55+
return err
7456
}
7557
timeout := d.config.BulkTimeout
7658
ctx, cancel := context.WithTimeout(context.Background(), timeout)
@@ -97,6 +79,32 @@ func (d recordDatabase) ReadinessCheck() bool {
9779
return true
9880
}
9981

82+
func (d recordDatabase) buildBulkRequest(records []*models.Record) (*elastic.BulkService, error) {
83+
bulkRequest := d.GetClient().Bulk()
84+
for _, record := range records {
85+
indexName := d.config.Index
86+
if indexName == "" {
87+
indexName = record.Topic
88+
}
89+
indexColumn := d.config.IndexColumn
90+
indexColumnValue := record.FormatTimestamp()
91+
if indexColumn != "" {
92+
newIndexColumnValue, err := record.GetValueForField(indexColumn)
93+
if err != nil {
94+
level.Error(d.logger).Log("err", err, "message", "Could not get column value from record.")
95+
return nil, err
96+
}
97+
indexColumnValue = newIndexColumnValue
98+
}
99+
index := fmt.Sprintf("%s-%s", indexName, indexColumnValue)
100+
bulkRequest.Add(elastic.NewBulkIndexRequest().Index(index).
101+
Type(record.Topic).
102+
Id(record.GetId()).
103+
Doc(record.FilteredFieldsJSON(d.config.BlacklistedColumns)))
104+
}
105+
return bulkRequest, nil
106+
}
107+
100108
func NewDatabase(logger log.Logger, config Config) RecordDatabase {
101109
return recordDatabase{logger: logger, config: config}
102110
}

‎src/kafka/consumer.go‎

Lines changed: 59 additions & 80 deletions
Original file line numberDiff line numberDiff line change
@@ -6,13 +6,12 @@ import (
66

77
"time"
88

9-
"sync"
10-
119
"github.com/Shopify/sarama"
1210
"github.com/bsm/sarama-cluster"
1311
"github.com/go-kit/kit/endpoint"
1412
"github.com/go-kit/kit/log"
1513
"github.com/go-kit/kit/log/level"
14+
"github.com/inloco/kafka-elasticsearch-injector/src/metrics"
1615
"github.com/inloco/kafka-elasticsearch-injector/src/models"
1716
)
1817

@@ -24,9 +23,12 @@ const (
2423
)
2524

2625
type kafka struct {
27-
consumer Consumer
28-
config *cluster.Config
29-
brokers []string
26+
consumer Consumer
27+
consumerCh chan *sarama.ConsumerMessage
28+
offsetCh chan *topicPartitionOffset
29+
config *cluster.Config
30+
brokers []string
31+
metricsPublisher metrics.MetricsPublisher
3032
}
3133

3234
type Consumer struct {
@@ -56,13 +58,16 @@ func NewKafka(address string, consumer Consumer) kafka {
5658
config.Version = sarama.V0_10_0_0
5759

5860
return kafka{
59-
brokers: brokers,
60-
config: config,
61-
consumer: consumer,
61+
brokers: brokers,
62+
config: config,
63+
consumer: consumer,
64+
metricsPublisher: metrics.NewMetricsPublisher(),
65+
consumerCh: make(chan *sarama.ConsumerMessage, consumer.BufferSize),
66+
offsetCh: make(chan *topicPartitionOffset),
6267
}
6368
}
6469

65-
func (k *kafka) Start(signals chan os.Signal, notifications chan Notification) {
70+
func (k *kafka) Start(signals chan os.Signal, notifications chan<- Notification) {
6671
topics := k.consumer.Topics
6772
concurrency := k.consumer.Concurrency
6873
consumer, err := cluster.NewConsumer(k.brokers, k.consumer.Group, topics, k.config)
@@ -72,83 +77,19 @@ func (k *kafka) Start(signals chan os.Signal, notifications chan Notification) {
7277
defer consumer.Close()
7378

7479
buffSize := k.consumer.BatchSize
75-
// Fan-out channel
76-
consumerCh := make(chan *sarama.ConsumerMessage, k.consumer.BufferSize)
77-
// Update offset channel
78-
offsetCh := make(chan *topicPartitionOffset)
7980
for i := 0; i < concurrency; i++ {
80-
go func() {
81-
buf := make([]*sarama.ConsumerMessage, buffSize)
82-
var decoded []*models.Record
83-
idx := 0
84-
for {
85-
kafkaMsg := <-consumerCh
86-
buf[idx] = kafkaMsg
87-
idx++
88-
for idx == buffSize {
89-
if decoded == nil {
90-
for _, msg := range buf {
91-
req, err := k.consumer.Decoder(nil, msg)
92-
if err != nil {
93-
level.Error(k.consumer.Logger).Log(
94-
"message", "Error decoding message",
95-
"err", err.Error(),
96-
)
97-
continue
98-
}
99-
decoded = append(decoded, req)
100-
}
101-
}
102-
if res, err := k.consumer.Endpoint(context.Background(), decoded); err != nil {
103-
level.Error(k.consumer.Logger).Log("message", "error on endpoint call", "err", err.Error())
104-
var _ = res // ignore res (for now)
105-
continue
106-
}
107-
notifications <- Inserted
108-
incrementRecordsConsumed(buffSize)
109-
for _, msg := range buf {
110-
offsetCh <- &topicPartitionOffset{msg.Topic, msg.Partition, msg.Offset}
111-
consumer.MarkOffset(msg, "") // mark message as processed
112-
}
113-
decoded = nil
114-
idx = 0
115-
}
116-
}
117-
}()
81+
go k.worker(consumer, buffSize, notifications)
11882
}
119-
lock := sync.RWMutex{}
120-
topicPartitionToOffset := make(map[string]map[int32]int64)
12183
go func() {
12284
for {
123-
offset := <-offsetCh
124-
lock.Lock()
125-
currentOffset, exists := topicPartitionToOffset[offset.topic][offset.partition]
126-
if !exists || offset.offset > currentOffset {
127-
_, exists := topicPartitionToOffset[offset.topic]
128-
if !exists {
129-
topicPartitionToOffset[offset.topic] = make(map[int32]int64)
130-
}
131-
topicPartitionToOffset[offset.topic][offset.partition] = offset.offset
132-
}
133-
lock.Unlock()
85+
offset := <-k.offsetCh
86+
k.metricsPublisher.UpdateOffset(offset.topic, offset.partition, offset.offset)
13487
}
13588
}()
13689

13790
go func() {
13891
for range time.Tick(k.consumer.MetricsUpdateInterval) {
139-
for topic, partitions := range consumer.HighWaterMarks() {
140-
for partition, maxOffset := range partitions {
141-
lock.RLock()
142-
offset, ok := topicPartitionToOffset[topic][partition]
143-
lock.RUnlock()
144-
if ok {
145-
delay := maxOffset - offset
146-
level.Info(k.consumer.Logger).Log("message", "updating partition offset metric",
147-
"partition", partition, "maxOffset", maxOffset, "current", offset, "delay", delay)
148-
updateOffset(topic, partition, delay)
149-
}
150-
}
151-
}
92+
k.metricsPublisher.PublishOffsetMetrics(consumer.HighWaterMarks())
15293
}
15394
}()
15495

@@ -157,13 +98,13 @@ func (k *kafka) Start(signals chan os.Signal, notifications chan Notification) {
15798
select {
15899
case msg, more := <-consumer.Messages():
159100
if more {
160-
if len(consumerCh) >= cap(consumerCh) {
101+
if len(k.consumerCh) >= cap(k.consumerCh) {
161102
level.Warn(k.consumer.Logger).Log(
162103
"message", "Buffer is full ",
163-
"channelSize", cap(consumerCh),
104+
"channelSize", cap(k.consumerCh),
164105
)
165106
}
166-
consumerCh <- msg
107+
k.consumerCh <- msg
167108
}
168109
case err, more := <-consumer.Errors():
169110
if more {
@@ -186,5 +127,43 @@ func (k *kafka) Start(signals chan os.Signal, notifications chan Notification) {
186127
return
187128
}
188129
}
130+
}
189131

132+
func (k *kafka) worker(consumer *cluster.Consumer, buffSize int, notifications chan<- Notification) {
133+
buf := make([]*sarama.ConsumerMessage, buffSize)
134+
var decoded []*models.Record
135+
idx := 0
136+
for {
137+
kafkaMsg := <-k.consumerCh
138+
buf[idx] = kafkaMsg
139+
idx++
140+
for idx == buffSize {
141+
if decoded == nil {
142+
for _, msg := range buf {
143+
req, err := k.consumer.Decoder(nil, msg)
144+
if err != nil {
145+
level.Error(k.consumer.Logger).Log(
146+
"message", "Error decoding message",
147+
"err", err.Error(),
148+
)
149+
continue
150+
}
151+
decoded = append(decoded, req)
152+
}
153+
}
154+
if res, err := k.consumer.Endpoint(context.Background(), decoded); err != nil {
155+
level.Error(k.consumer.Logger).Log("message", "error on endpoint call", "err", err.Error())
156+
var _ = res // ignore res (for now)
157+
continue
158+
}
159+
notifications <- Inserted
160+
k.metricsPublisher.IncrementRecordsConsumed(buffSize)
161+
for _, msg := range buf {
162+
k.offsetCh <- &topicPartitionOffset{msg.Topic, msg.Partition, msg.Offset}
163+
consumer.MarkOffset(msg, "") // mark message as processed
164+
}
165+
decoded = nil
166+
idx = 0
167+
}
168+
}
190169
}

‎src/kafka/consumer_test.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -122,7 +122,7 @@ func TestKafka_Start(t *testing.T) {
122122
err = json.Unmarshal(*res.Source, &r)
123123
if assert.NoError(t, err) {
124124
assert.Equal(t, rec.Id, r.Id)
125-
assert.InDelta(t, expectedTimestamp, r.Timestamp, 1000.0)
125+
assert.InDelta(t, expectedTimestamp, r.Timestamp, 5000.0)
126126
}
127127
}
128128
signals <- os.Interrupt

‎src/kafka/metrics.go‎

Lines changed: 0 additions & 29 deletions
This file was deleted.

0 commit comments

Comments
 (0)