Ingester Processor #944
Ingester Processor #944
Conversation
davit-y
requested review from
black-adder,
jpkrohling,
pavolloffay,
vprithvi and
yurishkuro
as code owners
davit-y
force-pushed
the
ingester-processor
branch
from
d1565e5 to
38f5111
Compare
davit-y
force-pushed
the
ingester-processor
branch
from
38f5111 to
ee86e47
Compare
Codecov Report
@@ Coverage Diff @@
## master #944 +/- ##
=====================================
Coverage 100% 100%
=====================================
Files 136 138 +2
Lines 6322 6343 +21
=====================================
+ Hits 6322 6343 +21
Continue to review full report at Codecov.
|
davit-y
force-pushed
the
ingester-processor
branch
2 times, most recently
from
250f585 to
ece2341
Compare
| // NewSpanProcessor creates a new SpanProcessor | ||
| func NewSpanProcessor(writer spanstore.Writer) SpanProcessor { | ||
| return &spanProcessor{ | ||
| unmarshaller: kafka.NewProtobufUnmarshaller(), |
There was a problem hiding this comment.
I recommend making the unmarshaller configurable by using an Option.
That way, we can default to protobuf, but allow people to easily conveniently override it to anything else.
plugin/storage/kafka/unmarshaller.go
Outdated
| // JSONUnmarshaller implements Unmarshaller | ||
| type JSONUnmarshaller struct{} | ||
|
|
||
| // NewJSONUnmarshaller constructs a ProtobufUnmarshaller |
There was a problem hiding this comment.
if this was school, you'd be reprimanded
| func (s spanProcessor) Process(message Message) error { | ||
| mSpan, err := s.unmarshaller.Unmarshal(message.Value()) | ||
| if err != nil { | ||
| return errors.Wrap(err, "cannot read message") |
There was a problem hiding this comment.
this message is a bit too generic, maybe could not unmarshal kafka message?
davit-y
force-pushed
the
ingester-processor
branch
from
1c101c2 to
973195f
Compare
davit-y
force-pushed
the
ingester-processor
branch
from
973195f to
58fb9c2
Compare
| // Message contains the fields of the kafka message that the span processor uses | ||
| type Message interface { | ||
| Value() []byte | ||
| } | ||
|
|
||
| // NewSpanProcessor creates a new SpanProcessor | ||
| func NewSpanProcessor(writer spanstore.Writer, unmarshaller kafka.Unmarshaller) SpanProcessor { |
There was a problem hiding this comment.
Any reason not to use functional options here and default to protobuf?
There was a problem hiding this comment.
please don't use functional options. Pass an options struct if necessary.
There was a problem hiding this comment.
I will create a Params struct similar to the consumer
davit-y
force-pushed
the
ingester-processor
branch
3 times, most recently
from
990848b to
2017c2e
Compare
| } | ||
|
|
||
| // NewSpanProcessor creates a new KafkaSpanProcessor | ||
| func NewSpanProcessor(params SpanProcessorParams) KafkaSpanProcessor { |
There was a problem hiding this comment.
Could you return a pointer instead?
Signed-off-by: Davit Yeghshatyan <[email protected]>
Signed-off-by: Davit Yeghshatyan <[email protected]>
Signed-off-by: Davit Yeghshatyan <[email protected]>
Signed-off-by: Davit Yeghshatyan <[email protected]>
davit-y
force-pushed
the
ingester-processor
branch
from
2017c2e to
35210fd
Compare
Signed-off-by: Davit Yeghshatyan <[email protected]>
davit-y
force-pushed
the
ingester-processor
branch
from
35210fd to
0f34c30
Compare
Which problem is this PR solving?
Short description of the changes