Add Kafka receiver #1410

pavolloffay · 2020-07-22T08:35:32Z

Signed-off-by: Pavol Loffay [email protected]

Description: <Describe what has changed.

This patch adds Kafka receiver and Consumer. The implementaiton is using Sarama client https://github.com/Shopify/sarama.

After this is merged I will do a couple of improvements:

authentication
batch configuration
make encoding configurable/pluggable (zipkin, Jaeger) for easier migration

Link to tracking Issue:

Resolves: #1331
Resolves: open-telemetry/opentelemetry-collector-contrib#5

Related to : #1101

Testing: < Describe what testing was performed and which tests were added.>

Tested locally with Confluent 5.0.0 https://docs.confluent.io/5.0.0/installation/docker/docs/installation/single-node-client.html

docker run --rm -it \
    --net=confluent \
    --name=zookeeper \
    -e ZOOKEEPER_CLIENT_PORT=2181 \
    confluentinc/cp-zookeeper:5.0.0

docker run --rm -it \
    --net=confluent -p 9092:9092 \
    --name=kafka \
    -e KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181 \
    -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9092 \
    -e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 \
    confluentinc/cp-kafka:5.0.0

Documentation: < Describe the documentation added.>

Exporter and receiver readmes.

codecov · 2020-07-22T10:53:35Z

Codecov Report

Merging #1410 into master will increase coverage by 0.07%.
The diff coverage is 100.00%.

@@            Coverage Diff             @@
##           master    #1410      +/-   ##
==========================================
+ Coverage   91.42%   91.50%   +0.07%     
==========================================
  Files         248      252       +4     
  Lines       17219    17372     +153     
==========================================
+ Hits        15743    15896     +153     
  Misses       1064     1064              
  Partials      412      412

Impacted Files	Coverage Δ
receiver/kafkareceiver/factory.go	`100.00% <100.00%> (ø)`
receiver/kafkareceiver/kafka_receiver.go	`100.00% <100.00%> (ø)`
receiver/kafkareceiver/metrics.go	`100.00% <100.00%> (ø)`
receiver/kafkareceiver/unmarshaller.go	`100.00% <100.00%> (ø)`
service/defaultcomponents/defaults.go	`85.18% <100.00%> (+0.27%)`	⬆️
service/telemetry.go	`85.48% <100.00%> (+0.23%)`	⬆️

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update e68440e...df9a0bd. Read the comment docs.

bogdandrutu

It will be almost impossible for one of the maintainer to review 1500+ lines. Consider to split this PR into more readable PRs, as a rule of thumb 500 lines is considered a large PR.

Thank you for your understanding :)

pavolloffay · 2020-07-24T05:26:12Z

Alright I will split it to exporter and receiver.

pavolloffay · 2020-07-27T07:52:35Z

@bogdandrutu I have removed the exporter from this PR add moved it to #1439. This and the other PR is ready for review.

bogdandrutu · 2020-07-27T20:19:18Z

@pavolloffay I would mark this PR WIP until we make some decisions in the exporter (regarding to the message we add the Kafka) then remove that. Does it make sense?

bogdandrutu · 2020-07-29T14:35:15Z

Please update accordingly to the changes in the exporter.

bogdandrutu

Some preliminary feedback, waiting for the updated PR.

receiver/kafkareceiver/factory_test.go

receiver/kafkareceiver/kafka_receiver.go

pavolloffay · 2020-07-29T16:11:32Z

receiver/kafkareceiver/config.go

+
+	// Metadata is the namespace for metadata management properties used by the
+	// Client, and shared by the Producer/Consumer.
+	Metadata kafkaexporter.Metadata `mapstructure:"metadata"`


This is being reused between exporter and receiver shall I move it to ./config/configkafka?

We can do this in a separate PR.

Alright once this is merged I will submit a refactoring PR.

Please revert then :)

What should I revert?

There is nothing to revert, or do you mean that instead of importing kafkaexporter.Metadata redefine that struct in the receiver?

pavolloffay · 2020-07-29T16:12:45Z

I have rebased it and fixed a couple of issues that I remember from the exporter. @bogdandrutu could you please give it a second go?

receiver/kafkareceiver/kafka_receiver.go

bogdandrutu · 2020-08-04T21:47:04Z

receiver/kafkareceiver/kafka_receiver.go

+
+func (c *kafkaConsumer) Start(context.Context, component.Host) error {
+	ctx, cancel := context.WithCancel(context.Background())
+	c.cancel = cancel


Confused about this cancel, is this just to cancel the "consumerGroup.Consume"? Is that not cancelled by the Close?

It's used to interrupt the infinite loop that calls c.consumerGroup.Consume

bogdandrutu · 2020-08-11T14:40:49Z

Ping

Signed-off-by: Pavol Loffay <[email protected]>

pavolloffay · 2020-08-12T13:55:15Z

@bogdandrutu I have updated the PR

Signed-off-by: Pavol Loffay <[email protected]>

bogdandrutu · 2020-08-13T18:33:35Z

receiver/kafkareceiver/kafka_receiver.go

+		// `Consume` should be called inside an infinite loop, when a
+		// server-side rebalance happens, the consumer session will need to be
+		// recreated to get the new claims
+		if err := c.consumerGroup.Consume(ctx, c.topics, handler); err != nil {


Is it not a problem that most likely this will block forever if no messages are coming, so never gets closed?

When no messages are coming the call passes and the handler is ready to receive messages.

When the close on the receiver is called it closes the consumer group correctly and interrupts the infinite loop via the cancelConsumeLoop context.CancelFunc (the "cancel" question you had before).

The critical change was submitted

bogdandrutu · 2020-08-17T14:23:27Z

go.sum

@@ -10,6 +10,7 @@ cloud.google.com/go v0.50.0/go.mod h1:r9sluTvynVuxRIOHXQEHMFffphuXHOMZMycpNR5e6T
 cloud.google.com/go v0.51.0/go.mod h1:hWtGJ6gnXH+KgDv+V0zFGDvpi07n3z8ZNj3T1RW0Gcw=
 cloud.google.com/go v0.52.0/go.mod h1:pXajvRH/6o3+F9jDHZWQ5PbGhn+o8w9qiu/CffaVdO4=
 cloud.google.com/go v0.53.0/go.mod h1:fp/UouUEsRkN6ryDKNW/Upv/JBKnv6WDthjR6+vze6M=
+cloud.google.com/go v0.56.0 h1:WRz29PgAsVEyPSDHyk+0fpEkwEFyfhHn+JbksT6gIL4=


Why go sum changes without go mod changes?

I am not sure why but go mod tidy gives an empty output :)

If I revert it go mod tidy brings it back

…y#1410) Bumps [boto3](https://github.com/boto/boto3) from 1.21.27 to 1.21.28. - [Release notes](https://github.com/boto/boto3/releases) - [Changelog](https://github.com/boto/boto3/blob/develop/CHANGELOG.rst) - [Commits](boto/boto3@1.21.27...1.21.28) --- updated-dependencies: - dependency-name: boto3 dependency-type: direct:production update-type: version-update:semver-patch ... Signed-off-by: dependabot[bot] <[email protected]> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

…emetry#1410)

pavolloffay requested review from bogdandrutu, dmitryax, flands, james-bebbington, owais, pjanotti, rghetia and tigrannajaryan as code owners July 22, 2020 08:35

jrcamp assigned dmitryax Jul 22, 2020

pavolloffay force-pushed the kafka-receiver branch from 7db9a0d to a8118b6 Compare July 23, 2020 09:50

bogdandrutu reviewed Jul 23, 2020

View reviewed changes

pavolloffay force-pushed the kafka-receiver branch from dfbbd53 to d311aec Compare July 27, 2020 07:31

pavolloffay requested a review from nilebox as a code owner July 27, 2020 07:31

pavolloffay mentioned this pull request Jul 27, 2020

Add Kafka exporter #1439

Merged

pavolloffay changed the title ~~Add Kafka receiver and exporter~~ Add Kafka receiver Jul 27, 2020

pavolloffay force-pushed the kafka-receiver branch from d311aec to 36fe01d Compare July 27, 2020 07:51

bogdandrutu unassigned dmitryax Jul 27, 2020

bogdandrutu added the WIP label Jul 27, 2020

pavolloffay changed the title ~~Add Kafka receiver~~ WIP: Add Kafka receiver Jul 28, 2020

bogdandrutu removed the WIP label Jul 29, 2020

bogdandrutu previously requested changes Jul 29, 2020

View reviewed changes

receiver/kafkareceiver/factory_test.go Outdated Show resolved Hide resolved

receiver/kafkareceiver/kafka_receiver.go Outdated Show resolved Hide resolved

pavolloffay force-pushed the kafka-receiver branch from a263ad5 to a6294c6 Compare July 29, 2020 15:51

pavolloffay commented Jul 29, 2020

View reviewed changes

pavolloffay changed the title ~~WIP: Add Kafka receiver~~ Add Kafka receiver Jul 29, 2020

bogdandrutu reviewed Aug 4, 2020

View reviewed changes

bogdandrutu added the waiting-for-author label Aug 6, 2020

Add Kafka receiver

be70c90

Signed-off-by: Pavol Loffay <[email protected]>

pavolloffay force-pushed the kafka-receiver branch from ea1f646 to be70c90 Compare August 12, 2020 11:59

pavolloffay added 2 commits August 12, 2020 14:23

Rebase

4829891

Signed-off-by: Pavol Loffay <[email protected]>

Fix metrics

2eb276f

Signed-off-by: Pavol Loffay <[email protected]>

pavolloffay added 2 commits August 12, 2020 16:14

Fix metrics

c6c948a

Signed-off-by: Pavol Loffay <[email protected]>

Cancel consume loop

df9a0bd

Signed-off-by: Pavol Loffay <[email protected]>

bogdandrutu reviewed Aug 13, 2020

View reviewed changes

bogdandrutu approved these changes Aug 17, 2020

View reviewed changes

bogdandrutu merged commit 55d4579 into open-telemetry:master Aug 17, 2020

pavolloffay mentioned this pull request Oct 13, 2020

Ingester does not start properly when using OTEL image jaegertracing/jaeger-operator#1250

Closed

MovieStoreGuy pushed a commit to atlassian-forks/opentelemetry-collector that referenced this pull request Nov 11, 2021

Add @dashpole as a project Approver (open-telemetry#1410)

93d426a

Troels51 pushed a commit to Troels51/opentelemetry-collector that referenced this pull request Jul 5, 2024

fix: WaitOnSocket select error when sockfd above FD_SETSIZE (open-tel…

0ef9446

…emetry#1410)

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add Kafka receiver #1410

Add Kafka receiver #1410

pavolloffay commented Jul 22, 2020 •

edited

Loading

codecov bot commented Jul 22, 2020 •

edited

Loading

bogdandrutu left a comment

pavolloffay commented Jul 24, 2020

pavolloffay commented Jul 27, 2020

bogdandrutu commented Jul 27, 2020

bogdandrutu commented Jul 29, 2020

bogdandrutu left a comment

pavolloffay Jul 29, 2020

bogdandrutu Jul 31, 2020

pavolloffay Aug 3, 2020

bogdandrutu Aug 17, 2020

pavolloffay Aug 17, 2020

pavolloffay commented Jul 29, 2020

bogdandrutu Aug 4, 2020

pavolloffay Aug 12, 2020

bogdandrutu commented Aug 11, 2020

pavolloffay commented Aug 12, 2020

bogdandrutu Aug 13, 2020

pavolloffay Aug 17, 2020

bogdandrutu Aug 17, 2020

pavolloffay Aug 17, 2020

bogdandrutu Aug 17, 2020

pavolloffay Aug 17, 2020

Add Kafka receiver #1410

Add Kafka receiver #1410

Conversation

pavolloffay commented Jul 22, 2020 • edited Loading

codecov bot commented Jul 22, 2020 • edited Loading

Codecov Report

bogdandrutu left a comment

Choose a reason for hiding this comment

pavolloffay commented Jul 24, 2020

pavolloffay commented Jul 27, 2020

bogdandrutu commented Jul 27, 2020

bogdandrutu commented Jul 29, 2020

bogdandrutu left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pavolloffay commented Jul 29, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bogdandrutu commented Aug 11, 2020

pavolloffay commented Aug 12, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pavolloffay commented Jul 22, 2020 •

edited

Loading

codecov bot commented Jul 22, 2020 •

edited

Loading