Add Kafka Component #951

g7ed6e · 2023-11-19T15:25:03Z

Relates to #884

samples/kafkaBasic/Producer/ProducerWorker.cs

src/Aspire.Hosting/Kafka/KafkaBuilderExtensions.cs

samples/kafkaBasic/Producer/ProducerWorker.cs

davidfowl · 2023-11-20T08:53:20Z

src/Aspire.Hosting/Kafka/KafkaBuilderExtensions.cs

+             .WithEnvironment("KAFKA_LISTENERS", "PLAINTEXT://localhost:29092,CONTROLLER://localhost:29093,PLAINTEXT_HOST://0.0.0.0:9092")
+             .WithEnvironment("KAFKA_ADVERTISED_LISTENERS", $"PLAINTEXT://localhost:29092,PLAINTEXT_HOST://localhost:{port}");


Where do these ports come from? Is it worth leaving a comment about them?

I added comments and links. These ports are the defaults of the confluentinc/confluent-local image which are set when the container starts if the env var is not supplied. See https://github.com/confluentinc/kafka-images/blob/master/local/include/etc/confluent/docker/configureDefaults

davidfowl · 2023-11-20T09:38:53Z

samples/kafkaBasic/Consumer/ConsumerWorker.cs

+        return Task.Factory.StartNew(() =>
+        {
+            consumer.Subscribe("topic");
+            while (!stoppingToken.IsCancellationRequested)
+            {
+                var result = consumer.Consume(stoppingToken);
+                logger.LogInformation($"Received message '{result.Message.Value}'.");
+            }
+        }, stoppingToken, TaskCreationOptions.LongRunning, TaskScheduler.Current);


This API is very strange.. no async? Is it blocking?

Indeed it is. See https://github.com/confluentinc/confluent-kafka-dotnet/blob/master/examples/Consumer/Program.cs and https://docs.confluent.io/kafka-clients/dotnet/current/overview.html#the-consume-loop

Confluentic has bad API here. It prefetches messages (if available), but it will block if nothing is there. That would be a perfect fit for ValueTask 😉.

Cf. confluentinc/confluent-kafka-dotnet#487

samples/kafkaBasic/Producer/ProducerWorker.cs

src/Components/Aspire.Kafka.Consumer/AspireKafkaConsumerExtensions.cs

davidfowl · 2023-11-20T09:40:36Z

src/Components/Aspire.Kafka.Consumer/AspireKafkaConsumerExtensions.cs

+        AddKafkaConsumer(builder, $"{DefaultConfigSectionName}:{name}", configureConsumerBuilder, configureConsumerConfig, connectionName: name, serviceKey: name);
+    }
+
+    private static void AddKafkaConsumer<TKey, TValue>(


Any health checks, telemetry etc?

Not yet. I'm thinking to pull logs and metrics from librdkafka using the below delegates that can be registered within ProducerBuilder<TKey, TValue> and ConsumerBuilder<TKey, TValue> SetLogHandler, SetErrorHandler, SetStatisticsHandler. The STATISTICS.json contains metrics.

src/Aspire.Hosting/Kafka/KafkaBuilderExtensions.cs

samples/kafkaBasic/Producer/ProducerWorker.cs

mitchdenny · 2023-11-20T22:19:52Z

Overall, the app model side of this is looking pretty good. I think we need some end to end integration tests for this.

src/Components/Aspire.Kafka.Consumer/IConsumerConnection.cs

mitchdenny · 2023-11-23T22:30:49Z

Once health checks are implemented, please wire-up the Kafa component inside our smoke test (new):

Add container to app host here:

aspire/tests/testproject/TestProject.AppHost/TestProgram.cs

Line 18 in 60bd2b0

IntegrationServiceA = AppBuilder.AddProject<Projects.IntegrationServiceA>("integrationservicea")
Add component to the integration test project:

aspire/tests/testproject/TestProject.IntegrationServiceA/Program.cs

Line 5 in 60bd2b0

builder.AddSqlServerClient("sql");

g7ed6e · 2023-11-24T15:48:17Z

@mitchdenny Can you add Confluent.Kafka 2.3.0 and AspNetCore.HealthChecks.Kafka 7.0.0 ?

davidfowl · 2023-11-24T16:16:23Z

I think we want a single component for kafka, not a package for the consumer and produce. Also, the component should be Aspire.Confluent.Kafka since that's the name of the underlying client library we're using.

PS: People are out of vacation this week so it's slow 😄

riverar · 2023-11-25T04:49:15Z

Side comment: As additional components get developed/added, I wonder if we'll need an ingredients label so developers know which components (and which versions) do and do not have health checks, telemetry, etc.

mitchdenny · 2023-11-26T22:39:35Z

I wonder if we'll need an ingredients label so developers know which components (and which versions) do and do not have health checks, telemetry, etc.

We probably shouldn't be merging it if it doesn't have telemetry, health-checks etc. In terms of dependency graph, I think we get that for free from NuGet so I don't think we need to do anything extra there.

mitchdenny · 2023-11-26T23:31:31Z

@mitchdenny Can you add Confluent.Kafka 2.3.0 and AspNetCore.HealthChecks.Kafka 7.0.0 ?

They should be in dotnet-public shortly.

g7ed6e · 2023-11-27T06:17:12Z

/azp run

azure-pipelines · 2023-11-27T06:17:18Z

Commenter does not have sufficient privileges for PR 951 in repo dotnet/aspire

eerhardt

We should also add a README and ConfigurationSchema.json files

tests/testproject/TestProject.IntegrationServiceA/Program.cs

src/Components/Aspire.Confluent.Kafka/Aspire.Confluent.Kafka.csproj

src/Components/Aspire.Confluent.Kafka/KafkaProducerSettings.cs

src/Components/Aspire.Confluent.Kafka/AspireKafkaConsumerExtensions.cs

src/Components/Aspire.Confluent.Kafka/ConsumerConnectionFactory.cs

eerhardt · 2023-11-27T19:21:46Z

src/Components/Aspire.Confluent.Kafka/ConfluentKafkaMetrics.cs

+
+    public static class Counters
+    {
+        public const string ReplyQueue = "replyq";


Should these names follow the same naming patterns as https://github.com/open-telemetry/semantic-conventions/blob/main/docs/messaging/kafka.md?

I'm not satisfied either by these names (basically same names as the ones from librdkafka statistics.json), but the naming pattern you linked seems to concern spans not metrics.
As these constants are used to expose specific metrics related to a given producer or consumer instance, I think we should instead follow a metric naming pattern.

edit: @eerhardt would prefixing these constants with messaging.kafka be sufficient ?

@lmolkova you are needed!

for the time being I prefixed metrics with messaging.kafka

Kafka metrics are in a weird place.
OTel is in the process of adding general messaging metrics into the spec - open-telemetry/semantic-conventions#163 (will be merged soon).

However Kafka as a project decided (AFAIK) to report a set of client metrics (KIP-714) that would be sent by default back to the broker and users would consume them from it. They decided not to follow otel semconv for the time being because of some back-compat reasons and limitations of existing metric impl on the broker side. Since they don't follow otel conventions, they also don't prefix with messaging. namespace.

I have a couple of suggestion

let's get an opinion from Kafka folks to understand the bigger picture. @forlack @AndrewJSchofield what are your thoughts on Kafka integration component for .NET Aspire emitting it's own set of OTel-compatible metrics different from what's defined in KIP-714? Also, could you confirm if I remember the decision (above) correctly or maybe something has changed? I envision the same conversation in OTel instrumentation world in Java, Python and other languages.

in the short term, WRT metrics added here, I'd recommend aligning them with otel messaging semantic conventions when applicable and following otel general naming conventions otherwise. I'll go through the changes and add my suggestions.

/cc @pyohannes

I think we should align with the semantic conventions as well, even if they are not final. Maybe we can help push towards that goal.

we can't because we just don't get the right data from the kafka client library.

We're going to discuss naming/approach with otel semconv community - https://github.com/open-telemetry/semantic-conventions/issues/578, I should have an update soon.

@g7ed6e would you be able to take a quick look at https://github.com/open-telemetry/opentelemetry-java-instrumentation/blob/bbfe950ad0ace8123a5e6817fb3767e27a1a2cee/instrumentation/kafka/kafka-clients/kafka-clients-2.6/library/README.md and check if any of those metrics apply here and if we can unify the naming? (just discovered they exist).

In any case, metric naming and OTEL community discussion should not block this PR and can be changed in the follow-ups.

in the short term, WRT metrics added here, I'd recommend aligning them with otel messaging semantic conventions when applicable and following otel general naming conventions otherwise. I'll go through the changes and add my suggestions.

💯 Agreed.

@lmolkova Ideally this could be modeled on top of generic messaging metrics that were recently defined, but it seems to me the overlap is quite small:

messaging.kafka.message.transmitted could become messaging.publish.messages

messaging.kafka.message.received could become messaging.receive.messages

Besides that we could validate on what attributes we can converge, I'll try to have a closer look.

@pyohannes : I updated and renamed messaging.kafka.message.tx to messaging.publish.messages and
messaging.kafka.message.rx to messaging.receive.messages.

@lmolkova : I had a look to the linked implementation. Currently (in this Aspire PR) we do not intercept messages between user code and kafka client library. Instead we use a librdkafka feature which perdiodically fetch statistics (please notice we only parse the top level properties from the statistics json for now, there should be other metrics we may probably use and map to metrics in the future).
Regarding the interceptor / decorator approach we may implement it, but before I would like to ensure that we want to do this in Aspire component as having a look to the other components they do not hold too much code/logic and only exposes existing clients.

@eerhardt / @davidfowl : What are your thoughts about implementing otel distributed tracing with an internal wrapper around IProducer<TKey, TValue> / IConsumer<TKey, TValue> living in Aspire.Confluent.Kafka.

@eerhardt / @davidfowl : What are your thoughts about implementing otel distributed tracing with an internal wrapper around IProducer<TKey, TValue> / IConsumer<TKey, TValue> living in Aspire.Confluent.Kafka.

Yes, this sounds reasonable as a way to drive the conversation forward for the default .NET Kafka library.

@eerhardt I know we don't do this today with rabbit MQ (I think it's harder there because there are more APIs to wrap).

src/Components/Aspire.Confluent.Kafka/AspireKafkaConsumerExtensions.cs

src/Components/Aspire.Confluent.Kafka/KafkaConsumerSettings.cs

eerhardt · 2023-11-27T19:48:39Z

samples/kafkaBasic/KafkaBasic.sln

@@ -0,0 +1,42 @@
+Microsoft Visual Studio Solution File, Format Version 12.00


Can this sample be moved to https://github.com/dotnet/aspire-samples?

I'm planning a follow up PR to add more metrics and (maybe) implement distributed tracing support by intercepting calls between user code and Confluent.Kafka. Having the sample in the solution highly ease debugging things. I propose to move the sample to the samples repo after that.

eerhardt · 2023-11-27T20:17:40Z

Also can you update

mitchdenny · 2023-12-19T01:02:42Z

@g7ed6e importing now.

g7ed6e · 2023-12-19T09:24:45Z

@g7ed6e importing now.

Thanks @mitchdenny CI now completes successfully

eerhardt · 2024-01-09T00:23:54Z

I pushed 2 commits to update the ConfigurationSchema.json file to be more complete (sorted and using the generator from #1383). I put them in 2 commits for easy reviewing - first just sorts the existing properties. The second can be diff'd by itself to see what the generator changed.

Unfortunately I hit a bug with the ConfigurationBinder source generator - dotnet/runtime#96652. We aren't able to automate updating the ConfigurationSchema.json file automatically until we have a fix for that issue.

eerhardt

This is looking pretty good. Thanks for the contribution. I have a few comments/questions. Once they are addressed I believe we can get this merged.

samples/kafkaBasic/KafkaBasic.AppHost/KafkaBasic.AppHost.csproj

src/Aspire.Hosting/Kafka/KafkaServerResource.cs

src/Aspire.Hosting/Kafka/KafkaBuilderExtensions.cs

src/Components/Aspire.Confluent.Kafka/Aspire.Confluent.Kafka.csproj

eerhardt · 2024-01-09T17:57:25Z

src/Components/Aspire.Confluent.Kafka/AspireKafkaConsumerExtensions.cs

+
+        ConsumerConfig config = new();
+        configSection.GetSection(nameof(KafkaConsumerSettings.Config)).Bind(config);
+        settings.Config = config;


The typical pattern we have is that the "Settings" object and the underlying library's "Options" object (in this case ConsumerConfig) are separate. There isn't a reference between them. Can we remove this property on KafkaConsumerSettings (and the producer as well)?

That way we can use the IOptions pattern for ConsumerConfig, which is what the Redis component does for its ConfigurationOptions object.

Looking more at Kafka, there is a bit of a special pattern here in that the ConsumerConfig is separate from the ConsumerBuilder. We don't really have that pattern in the other components - typically there is just 1 object - ConfigurationOptions in Redis, IConnectionFactory in RabbitMQ.

Maybe what we have here is fine for now. It would be awkward to have 3 "configure" callbacks - configureSettings, configureConfig, configureBuilder.

@eerhardt The reference to the underlying library configuration object in the Aspire's "Settings" object lets customer configure a dedicated serializer / deserialize per pair of generic parameters provided to the underlying confluent XXXBuilder. Using IOptions pattern here would defeat it.
I "reproduced" the pattern I observed in the RabbitMQ implementation.

src/Components/Aspire.Confluent.Kafka/README.md

src/Components/Aspire.Confluent.Kafka/KafkaConsumerSettings.cs

src/Components/Aspire.Confluent.Kafka/MetricsService.cs

src/Components/Aspire.Confluent.Kafka/AspireKafkaConsumerExtensions.cs

src/Aspire.Hosting/Kafka/KafkaBuilderExtensions.cs

mitchdenny · 2024-01-10T09:24:12Z

tests/Aspire.Hosting.Tests/ManifestGenerationTests.cs

@@ -331,6 +331,29 @@ public void EnsureAllRabitMQManifestTypesHaveVersion0Suffix()
        Assert.Equal("container.v0", server.GetProperty("type").GetString());
    }

+    [Fact]
+    public void EnsureAllKafkaManifestTypesHaveVersion0Suffix()


Do we need a test to makes sure the connectionString field is being emitted.

mitchdenny · 2024-01-10T09:26:44Z

I think this is really close. A question or two from my perspective, but from an app model point of view this looks good.

eerhardt · 2024-01-11T17:02:35Z

src/Components/Aspire.Confluent.Kafka/MetricsService.cs

+        }
+    }
+
+    [LoggerMessage(LogLevel.Warning, EventId = 42, Message = "Invalid statistics json payload received: `{json}`")]


which value would you suggest ?
edit: Is there an already existing source file containing event id ? I did not find one yet.

Looking at other usages around dotnet/aspnetcore, it looks like the convention is to start at 1.

src/Components/Aspire.Confluent.Kafka/MetricsService.cs

src/Components/Aspire.Confluent.Kafka/README.md

tests/Aspire.Confluent.Kafka.Tests/MetricsTests.cs

tests/Aspire.Confluent.Kafka.Tests/ProducerConfigurationTests.cs

eerhardt · 2024-01-11T17:59:43Z

src/Components/Aspire.Confluent.Kafka/AspireKafkaConsumerExtensions.cs

+
+        ConsumerConfig config = new();
+        configSection.GetSection(nameof(KafkaConsumerSettings.Config)).Bind(config);
+        settings.Config = config;


Looking more at Kafka, there is a bit of a special pattern here in that the ConsumerConfig is separate from the ConsumerBuilder. We don't really have that pattern in the other components - typically there is just 1 object - ConfigurationOptions in Redis, IConnectionFactory in RabbitMQ.

Maybe what we have here is fine for now. It would be awkward to have 3 "configure" callbacks - configureSettings, configureConfig, configureBuilder.

apply pr suggestions apply pr suggestions apply pr suggestions Sort ConfigurationSchema.json properties Update ConfigurationSchema.json using ConfigSchemaGenerator apply pr suggestions apply pr suggestions apply pr suggesstions apply pr suggesstions apply pr suggestions apply pr suggestions drop kafka sample from this repo apply pr suggestions

g7ed6e · 2024-01-11T20:42:16Z

@eerhardt I just squashed end rebased until the Orleans support adding.

- Make Kafka Settings Config object non-nullable and not settable from user code. - Use EventId 1 in Logging - Log response body in Functional tests

eerhardt

This is great! Thanks, @g7ed6e, for this contribution and all your great work.

I pushed some review feedback changes. Let me know if you have any issues with the changes.

Unless anyone else has any objections, I will merge this once CI passes.

…he type can be created publicly.

davidfowl · 2024-01-13T02:08:22Z

Thanks so much @g7ed6e!

dotnet-issue-labeler bot added the area-integrations Issues pertaining to Aspire Integrations packages label Nov 19, 2023

gfoidl reviewed Nov 19, 2023

View reviewed changes

samples/kafkaBasic/Producer/ProducerWorker.cs Outdated Show resolved Hide resolved

src/Aspire.Hosting/Kafka/KafkaBuilderExtensions.cs Outdated Show resolved Hide resolved