Skip to content

Commit 933efb3

Browse files
vprithvivprithvi
authored
Add files for ingester (#940)
This commit contains core functionality for asynchronous ingestion, future commits will make use of it. Signed-off-by: Prithvi Raj <p.r@uber.com>
1 parent 214c133 commit 933efb3

26 files changed

+2045
-6
lines changed

‎cmd/ingester/app/consumer/committing_processor.go

+51
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,51 @@
1+
// Copyright (c) 2018 The Jaeger Authors.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package consumer
16+
17+
import (
18+
"errors"
19+
"io"
20+
21+
"github.com/jaegertracing/jaeger/cmd/ingester/app/processor"
22+
)
23+
24+
type comittingProcessor struct {
25+
processor processor.SpanProcessor
26+
marker offsetMarker
27+
io.Closer
28+
}
29+
30+
type offsetMarker interface {
31+
MarkOffset(int64)
32+
}
33+
34+
// NewCommittingProcessor returns a processor that commits message offsets to Kafka
35+
func NewCommittingProcessor(processor processor.SpanProcessor, marker offsetMarker) processor.SpanProcessor {
36+
return &comittingProcessor{
37+
processor: processor,
38+
marker: marker,
39+
}
40+
}
41+
42+
func (d *comittingProcessor) Process(message processor.Message) error {
43+
if msg, ok := message.(Message); ok {
44+
err := d.processor.Process(message)
45+
if err == nil {
46+
d.marker.MarkOffset(msg.Offset())
47+
}
48+
return err
49+
}
50+
return errors.New("committing processor used with non-kafka message")
51+
}

‎cmd/ingester/app/consumer/committing_processor_test.go

+75
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,75 @@
1+
// Copyright (c) 2018 The Jaeger Authors.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package consumer
16+
17+
import (
18+
"errors"
19+
"testing"
20+
21+
"github.com/stretchr/testify/assert"
22+
"github.com/stretchr/testify/mock"
23+
24+
kafka "github.com/jaegertracing/jaeger/cmd/ingester/app/consumer/mocks"
25+
"github.com/jaegertracing/jaeger/cmd/ingester/app/processor/mocks"
26+
)
27+
28+
type fakeOffsetMarker struct {
29+
capturedOffset int64
30+
}
31+
32+
func (f *fakeOffsetMarker) MarkOffset(o int64) {
33+
f.capturedOffset = o
34+
}
35+
36+
func TestNewCommittingProcessor(t *testing.T) {
37+
msgOffset := int64(123)
38+
offsetMarker := &fakeOffsetMarker{}
39+
spanProcessor := &mocks.SpanProcessor{}
40+
spanProcessor.On("Process", mock.Anything).Return(nil)
41+
committingProcessor := NewCommittingProcessor(spanProcessor, offsetMarker)
42+
43+
msg := &kafka.Message{}
44+
msg.On("Offset").Return(msgOffset)
45+
46+
assert.NoError(t, committingProcessor.Process(msg))
47+
48+
spanProcessor.AssertExpectations(t)
49+
assert.Equal(t, msgOffset, offsetMarker.capturedOffset)
50+
}
51+
52+
func TestNewCommittingProcessorError(t *testing.T) {
53+
offsetMarker := &fakeOffsetMarker{}
54+
spanProcessor := &mocks.SpanProcessor{}
55+
spanProcessor.On("Process", mock.Anything).Return(errors.New("boop"))
56+
committingProcessor := NewCommittingProcessor(spanProcessor, offsetMarker)
57+
msg := &kafka.Message{}
58+
59+
assert.Error(t, committingProcessor.Process(msg))
60+
61+
spanProcessor.AssertExpectations(t)
62+
assert.Equal(t, int64(0), offsetMarker.capturedOffset)
63+
}
64+
65+
type fakeProcessorMessage struct{}
66+
67+
func (f fakeProcessorMessage) Value() []byte {
68+
return nil
69+
}
70+
71+
func TestNewCommittingProcessorErrorNoKafkaMessage(t *testing.T) {
72+
committingProcessor := NewCommittingProcessor(&mocks.SpanProcessor{}, &fakeOffsetMarker{})
73+
74+
assert.Error(t, committingProcessor.Process(fakeProcessorMessage{}))
75+
}

‎cmd/ingester/app/consumer/consumer.go

+111
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,111 @@
1+
// Copyright (c) 2018 The Jaeger Authors.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package consumer
16+
17+
import (
18+
"io"
19+
"sync"
20+
21+
"github.com/Shopify/sarama"
22+
sc "github.com/bsm/sarama-cluster"
23+
"github.com/uber/jaeger-lib/metrics"
24+
"go.uber.org/zap"
25+
26+
"github.com/jaegertracing/jaeger/cmd/ingester/app/processor"
27+
)
28+
29+
type consumer struct {
30+
metricsFactory metrics.Factory
31+
logger *zap.Logger
32+
processorFactory processorFactory
33+
34+
close chan struct{}
35+
isClosed sync.WaitGroup
36+
37+
SaramaConsumer
38+
}
39+
40+
// SaramaConsumer is an interface to features of Sarama that we use
41+
type SaramaConsumer interface {
42+
Partitions() <-chan sc.PartitionConsumer
43+
MarkPartitionOffset(topic string, partition int32, offset int64, metadata string)
44+
io.Closer
45+
}
46+
47+
func (c *consumer) mainLoop() {
48+
c.isClosed.Add(1)
49+
c.logger.Info("Starting main loop")
50+
go func() {
51+
for {
52+
select {
53+
case pc := <-c.Partitions():
54+
c.isClosed.Add(2)
55+
56+
go c.handleMessages(pc)
57+
go c.handleErrors(pc.Partition(), pc.Errors())
58+
59+
case <-c.close:
60+
c.isClosed.Done()
61+
return
62+
}
63+
}
64+
}()
65+
}
66+
67+
func (c *consumer) handleMessages(pc sc.PartitionConsumer) {
68+
c.logger.Info("Starting message handler")
69+
defer c.isClosed.Done()
70+
defer c.closePartition(pc)
71+
72+
msgMetrics := c.newMsgMetrics(pc.Partition())
73+
var msgProcessor processor.SpanProcessor
74+
75+
for msg := range pc.Messages() {
76+
c.logger.Debug("Got msg", zap.Any("msg", msg))
77+
msgMetrics.counter.Inc(1)
78+
msgMetrics.offsetGauge.Update(msg.Offset)
79+
msgMetrics.lagGauge.Update(pc.HighWaterMarkOffset() - msg.Offset - 1)
80+
81+
if msgProcessor == nil {
82+
msgProcessor = c.processorFactory.new(pc.Partition(), msg.Offset-1)
83+
defer msgProcessor.Close()
84+
}
85+
86+
msgProcessor.Process(&saramaMessageWrapper{msg})
87+
}
88+
}
89+
90+
func (c *consumer) closePartition(partitionConsumer sc.PartitionConsumer) {
91+
c.logger.Info("Closing partition consumer", zap.Int32("partition", partitionConsumer.Partition()))
92+
partitionConsumer.Close() // blocks until messages channel is drained
93+
c.logger.Info("Closed partition consumer", zap.Int32("partition", partitionConsumer.Partition()))
94+
}
95+
96+
func (c *consumer) handleErrors(partition int32, errChan <-chan *sarama.ConsumerError) {
97+
c.logger.Info("Starting error handler")
98+
defer c.isClosed.Done()
99+
100+
errMetrics := c.newErrMetrics(partition)
101+
for err := range errChan {
102+
errMetrics.errCounter.Inc(1)
103+
c.logger.Error("Error consuming from Kafka", zap.Error(err))
104+
}
105+
}
106+
107+
func (c *consumer) Close() error {
108+
close(c.close)
109+
c.isClosed.Wait()
110+
return c.SaramaConsumer.Close()
111+
}

‎cmd/ingester/app/consumer/consumer_metrics.go

+46
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,46 @@
1+
// Copyright (c) 2018 The Jaeger Authors.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package consumer
16+
17+
import (
18+
"strconv"
19+
20+
"github.com/uber/jaeger-lib/metrics"
21+
)
22+
23+
type msgMetrics struct {
24+
counter metrics.Counter
25+
offsetGauge metrics.Gauge
26+
lagGauge metrics.Gauge
27+
}
28+
29+
type errMetrics struct {
30+
errCounter metrics.Counter
31+
}
32+
33+
func (c *consumer) newMsgMetrics(partition int32) msgMetrics {
34+
f := c.metricsFactory.Namespace("sarama-consumer", map[string]string{"partition": strconv.Itoa(int(partition))})
35+
return msgMetrics{
36+
counter: f.Counter("messages", nil),
37+
offsetGauge: f.Gauge("current-offset", nil),
38+
lagGauge: f.Gauge("offset-lag", nil),
39+
}
40+
}
41+
42+
func (c *consumer) newErrMetrics(partition int32) errMetrics {
43+
f := c.metricsFactory.Namespace("sarama-consumer", map[string]string{"partition": strconv.Itoa(int(partition))})
44+
return errMetrics{errCounter: f.Counter("errors", nil)}
45+
46+
}

0 commit comments

Comments
 (0)