Strimzi Metrics Reporter integration with KafkaConnect and MirrorMaker2

[OwenCorrigan76]

Type of change

• Enhancement / new feature

Description

This patch adds support for the Strimzi Metrics Reporter to Kafka Connect and MirrorMaker2 as described by the following proposal:

https://github.com/strimzi/proposals/blob/main/064-prometheus-metrics-reporter.md

Related to #10753

We won’t initially support the CruiseControl component. To make it work, CC should be changed to expose metrics through its HTTP endpoint.

Documentation will be updated in a separate PR.

Checklist

• Write tests
• Make sure all tests pass
• [] Update documentation
• Try your changes from Pod inside your Kubernetes and OpenShift cluster, not just locally
• Reference relevant issue(s) and close them after merging
• Update CHANGELOG.md

[ppatierno]

Suggested change

	if (configurations instanceof KafkaConnectConfiguration) {
	if (configurations instanceof KafkaConnectConfiguration kcConfiguration) {

[ppatierno]

Suggested change

	configurations = new KafkaConnectConfiguration((KafkaConnectConfiguration) configurations);
	configurations = new KafkaConnectConfiguration(kcConfiguration);

[ppatierno]

Suggested change

	} else if (configurations instanceof KafkaMirrorMaker2Configuration) {
	} else if (configurations instanceof KafkaMirrorMaker2Configuration kmm2Configuration) {

[ppatierno]

Suggested change

	configurations = new KafkaMirrorMaker2Configuration((KafkaMirrorMaker2Configuration) configurations);
	configurations = new KafkaMirrorMaker2Configuration(kmm2Configuration);

[im-konge]

Why is this needed?
I'm not sure that it will do exactly what you want, we usually set it in the Kafka once it is "built".
In the past this didn't work.

But anyway, the KafkaTemplates class is there for things that are used multiple times, you are using it just once, which doesn't make sense to me to have it in the templates class.

[im-konge]

It's desired to use mixed pools? Also, should it be really just one node? These tests were always really unstable, so I wouldn't go with the mixed node pool and with just 1 replica (if you keep the mixed node pool).

Also, you don't need to use the -tgt, you can just use getTargetClusterName(), getTargetBrokerPoolName() from the testStorage.

[im-konge]

You should use things from the testStorage. It's easier than having hardcoded values like my-connect. Just go with the getClusterName() as we have in all other tests.

[im-konge]

Same here.

[fvaleri]

@im-konge this would result in a KubernetesClientException due to the operator attempt to patch the roleRef field of an existing RoleBinding, which is not allowed in Kubernetes

[im-konge]

We are using it everywhere else:

KafkaMirrorMaker2Templates.kafkaMirrorMaker2(testStorage.getNamespaceName(), testStorage.getClusterName(), testStorage.getSourceClusterName(), testStorage.getTargetClusterName(), 1, false)

Why it would not work here as well?

[fvaleri]

with your change, we are creating both KC and MM2 with the same name, and the role binding reconciliation logic is in common between these components

[im-konge]

Ah okay, I understand.
So anyway, instead of hardcoding it, it would be better to create some local variables for the class (similarly how it is done in MetricsST) for better manipulation in other tests.

Also, I'm thinking about leaving the creation of the MM2 and Connect (and possibly other components in the future) for the particular tests.
So in case that one test (for Kafka for example) fails, the suite will not try to recreate MM2 and Connect.

That's I guess the right way (plus for example in the MetricsST we can move the second Kafka cluster out of the before all, as there is no MM1 anymore) - and it will speed things up. WDYT?

[im-konge]

Isn't the issue because you have different namespaces in beforeall and in the test?

[fvaleri]

I'm always using testStorage.getNamespaceName(), and looking at logs it's the same namespace (test-suite-namespace). Uploading 2025-06-24-10-44-07.zip…

[im-konge]

Thanks, I will a have a look.
What you should do is that you will create class variable for the testStorage and use it everywhere.
Because the context is different in the before all and in the test.

So even if the namespace is same, the cluster name you are passing there is different in the test.

[im-konge]

You can check for example HttpBridgeST where we are using the suiteTestStorage

[fvaleri]

Yes, I'm doing exactly that, but still running some tests to confirm is all good.

[im-konge]

Same here.

[fvaleri]

@im-konge your comments should be addressed now. Thanks.

[im-konge]

@im-konge your comments should be addressed now. Thanks.

Thanks @fvaleri, I will have a look later today or tomorrow :)

[scholzj]

I don't think you should modify this in the first place.

[OwenCorrigan76]

What do you mean? Can you please give more details?

[scholzj]

I don't think you should modify this in the first place.

[OwenCorrigan76]

What do you mean? Can you please give more details?

[scholzj]

We should try to avoid modifying it in the first place. So I guess the main question is why is this needed in the first place.

[OwenCorrigan76]

Hi @scholzj.
The javadoc in these configuration classes and the comment in KafkaConnectConfigurationBuilder are pretty clear about why these changes are needed.
We are creating a defensive copy and utilising the newly created constructor to do so.

 // creating a defensive copy to avoid mutating the input config
        if (configurations instanceof KafkaConnectConfiguration kcConfiguration) {
            configurations = new KafkaConnectConfiguration(kcConfiguration);

Try to comment out this change in KafkaConnectConfiguration and run the KafkaConnectConfigurationBuilderTest to have an example of the kind of issues you may have. That's how I found the issue and prompted the need for that change

Also, this is the same strategy used here in KafkaBrokerConfigurationBuilder, for exactly the same reason.

[scholzj]

I'm pretty sure the Kafka Broker Configuration Builder does not insert anything into the user configuration and neither should you. If it does, it is wrong and should be fixed.

[ppatierno]

I got your point. +1 for me

[ppatierno]

I reviewed last changes and now I am less convinced that using a copy of the user configuration as cache was not that great (and sorry if I changed my mind).
With making a copy of the user configuration (which anyway we do in the broker and connect builder for removing stuff from there) we have just one consistent place where things are happening and where we build the configuration to be printed at the end (and it contains both strimzi and user ones).
With the latest changes we are anyway using a copy (because of the removing stuff) but then we are directly printing some other stuff so it's now a mix. The only advantage that current code is bringing is a clear separation of configuration sections (for user and strimzi) within the Kafka configuration, but from a code perspective we are losing clarity and maybe introducing side effects. Let's see what others think but imho we could come back to the previous solution, unless someone else strongly think (and convince me) that the current one is better.

[OwenCorrigan76]

Do any of the @strimzi/maintainers have any further opinion on this thread? Thanks

[katheris]

Looking back at previous changes, there was already logic in the KafkaBrokerConfigurationBuilder.withUserConfiguration method handling metric configs, so I can see why we added the part about the Strimzi Metrics Reporter there as well in the previous PR and it went in without anyone commenting on it.

However, I do think it is confusing to have a method called withUserConfiguration and then to actually add some other metrics that aren't directly user configurations. So I would prefer a dedicated method in KafkaConnectConfigurationBuilder called something like withMetricsConfiguration that handles the metrics configuration.

I think we might also want to consider changing the way we do it in the KafkaBrokerConfigurationBuilder, but I think that should be handled in a separate PR to make things easier to review.

[ppatierno]

I would agree that from a logical perspective, having two methods would make much more sense but I think it was done this ways because both configuration (from user and by strimzi) have an impact on the same property, the metric.reporters so you have to mix and match the reporters set by the user via the spec.kafka.config together with the ones added by Strimzi automatically when you enable the Strimzi Metrics reporter for example.
You can't write the property in each withXXXConfiguration method you should write it when everything is consistent in one object, thus the first/previous approach by cloning the user config object and using it to add the Strimzi ones to write everything at the end.
Of course, it doesn't mean that having two methods is not possible. We should pass a config object across them in order to fill what we need from both and the write it.

[scholzj]

These are not really model utils from my point of view and should not be here. (Plus as I said, you should not need these things in the first place.)

[OwenCorrigan76]

Can you suggest a better place please? Thanks

[scholzj]

Possibly in the configuration classes themself? But as I said in the other place, I do not think you should be adding to it in the first place?

[OwenCorrigan76]

Ok, I will use AbstractConfiguration to host that method. Thanks for the suggestion.

[OwenCorrigan76]

I have moved createOrAddListConfig whilst awaiting the outcome of the conversation here #11570 (comment)

[mimaison]

FYI there is an issue in metrics-reporter 0.2.0 with MirrorMaker that prevents it from retrieving metrics: strimzi/metrics-reporter#87
The issue happens when multiple tasks of MirrorMaker connectors run in the same runtime, which pretty much always happens when you run MirrorMaker.

[fvaleri]

FYI there is an issue in metrics-reporter 0.2.0 with MirrorMaker that prevents it from retrieving metrics: strimzi/metrics-reporter#87 The issue happens when multiple tasks of MirrorMaker connectors run in the same runtime, which pretty much always happens when you run MirrorMaker.

@mimaison thanks for the note. AFAICS, it should be easy to reproduce with this PR and, in order to move this PR forward, we would need metrics-reporter 0.2.1 or 0.3.0. What's the plan?

[mimaison]

I'd suggest releasing metrics-reporter 0.3.0 and using that in strimzi-kafka-operator. But obviously we need the input of Strimzi maintainers.

[fvaleri]

It looks like 0.2.0 is not affected by metrics-reporter/issues/87 (*) and we can proceed with this PR review. Most probably the issue originated from some other change done after the release.

@im-konge @scholzj @ppatierno can you please have another look? Thanks.

(*) In order to reproduce the issue you need to set the allowList to .* so that kafka_connect_mirror_kafka_metrics_count_count is not filtered out by the default allowList. You also need to run a producer and consumer to have running tasks on both source and checkpoint connectors.

Source connector status:

{
  "name": "my-cluster->my-cluster-tgt.MirrorSourceConnector",
  "connector": {
    "state": "RUNNING",
    "worker_id": "my-mm2-cluster-mirrormaker2-0.my-mm2-cluster-mirrormaker2.test.svc:8083"
  },
  "tasks": [
    {
      "id": 0,
      "state": "RUNNING",
      "worker_id": "my-mm2-cluster-mirrormaker2-0.my-mm2-cluster-mirrormaker2.test.svc:8083"
    },
    {
      "id": 1,
      "state": "RUNNING",
      "worker_id": "my-mm2-cluster-mirrormaker2-0.my-mm2-cluster-mirrormaker2.test.svc:8083"
    },
    {
      "id": 2,
      "state": "RUNNING",
      "worker_id": "my-mm2-cluster-mirrormaker2-0.my-mm2-cluster-mirrormaker2.test.svc:8083"
    },
    {
      "id": 3,
      "state": "RUNNING",
      "worker_id": "my-mm2-cluster-mirrormaker2-0.my-mm2-cluster-mirrormaker2.test.svc:8083"
    },
    {
      "id": 4,
      "state": "RUNNING",
      "worker_id": "my-mm2-cluster-mirrormaker2-0.my-mm2-cluster-mirrormaker2.test.svc:8083"
    }
  ],
  "type": "source"
}

Checkpoint connector status:

{
  "name": "my-cluster->my-cluster-tgt.MirrorCheckpointConnector",
  "connector": {
    "state": "RUNNING",
    "worker_id": "my-mm2-cluster-mirrormaker2-0.my-mm2-cluster-mirrormaker2.test.svc:8083"
  },
  "tasks": [
    {
      "id": 0,
      "state": "RUNNING",
      "worker_id": "my-mm2-cluster-mirrormaker2-0.my-mm2-cluster-mirrormaker2.test.svc:8083"
    },
    {
      "id": 1,
      "state": "RUNNING",
      "worker_id": "my-mm2-cluster-mirrormaker2-0.my-mm2-cluster-mirrormaker2.test.svc:8083"
    }
  ],
  "type": "source"
}

After that, I'm able to get all metrics by invoking the endpoint, including the offending one:

# TYPE kafka_connect_mirror_kafka_metrics_count_count gauge
kafka_connect_mirror_kafka_metrics_count_count 1.0

[mimaison]

Thanks @fvaleri for checking.

[im-konge]

LGTM from systemtests POV. Thanks

[fvaleri]

Nit: let's get rid of any change to this class.

[ppatierno]

why AbstractConfiguration and not KafkaConfiguration?

[ppatierno]

Ditto as above

[ppatierno]

AFAIU this part is about putting the user config, why are you also testing injectFlags.get(i) which is about the injection of "internal" reporters?

[fvaleri]

This is related to the following comment.

[ppatierno]

the above three lines of code doesn't use anything which is driving the for loop (i.e. reporterClasses, ...) why can't they be outside of the loop?

[fvaleri]

These have to be executed only if there is user config AND an injection needed (i.e. we don't mess with user config unless we have to). I reworked the code to make it more readable and less error-prone.

[ppatierno]

Maybe it's similar to the previous question but:
if there is a "kafka.metrics.reporters" provided by the user BUT injectStrimziMetricsReporter is false, we are not entering in the if so we are not adding the user config to the props collection. The hasUserMetricReporters remain false and when we go down the final code with don't print any configuration. The result we are not printing the user provided configuration, or?

[fvaleri]

In that case, the user configuration stays in the userConfig and it is printed at the end of withUserConfiguration.

[scholzj]

After #11638 is merged, this has to be moved to the 0.48.0 section to avoid the same problem we had with the Kafka brokers part.

[fvaleri]

@ppatierno @scholzj any feedback on this? We are already working on the Bridge support, but we would need this one to move on. Thanks.

[scholzj]

Why do we need to feed this into a list instead of just printing it as it comes?

[fvaleri]

This is a temporary list of 3 elements that is used as an optimization to make the code less verbose. The alternative would be to get rid of this data structure and replace the for loop with explicit if statements.

[scholzj]

Is it really less verbose? Is it really more readable which is probably more important quality? TBH, I do not think so. Why is it not possible to just assemble it as a String right away?

[katheris]

I find this code very hard to read. The code in KafkaConnectConfigurationBuilder.printMetricReporters is even more difficult to follow since there are two Lists and two a for loop within a for loop. I think we should aim for readability here over fewer lines of code.

[scholzj]

Same as above.

[fvaleri]

This is just an optimization where we are only printing the data at the end. It is one of the options you suggested in a previous comment:

We can either generate the configuration right away into the configuration file or if we really need to store something between the different steps, we can use the builder class for it.

[scholzj]

TBH, the configuration builder classes are all designed around just printing the configurations into the PrintWriter. When I suggested to store the state, I expected for example the user-configured metric reporters will be taken in the user configuration method, stored in the class variable and added later in the metrics method to the. Not that you generate everything and store it just to print it later. Not sure that makes sense. Don't the order properties have always only one entry anyway?

[scholzj]

The idea to move this into AbstractConfiguration made sense when this was invoked in the Configuration classes. But it does not seem to make much sense if you use it with different classes only.

[fvaleri]

We had this in ModelUtils and that wasn't ok. Where would you suggest to move it?

[scholzj]

It was not OK to have it in ModelUtils, and use it for configurations, yes. But the overall idea was to avoid as much as possible the modifications of the configurations. In particular inserting our own configurations into them. So this method should not be needed for the Configuration classes if that is where we moved.

[scholzj]

Do I read this correctly that we do not expect metric reporters in MM2 connectors? Can we make such an assumption?

[fvaleri]

No, we expect them.

MM2 connectors metrics are collected through this dedicated SMR instance into a shared Prometheus registry instance, so they can be exposed through the Kafka Connect SMR listener endpoint without enabling a new listener.

Comment updated for clarity.

[scholzj]

We insert our own metrics reporter. But what I meant is that from the code, it seems we do not expect any user-configured metrics reporters there and just insert our own. Is there some reason for that? Or did I misunderstand the code?

[fvaleri]

You are right. We also need to handle user config when present.

[scholzj]

Do we want to have these option in the Metrics Reporter itself and load it from there? Or do you think it is better to avoid it as a CO depedency and define them here?

[fvaleri]

This is a level of indirection, similar to what we have for CruiseControl. It would be great to have them from the SMR, but we would still need to complement with additional configurations (e.g. FQCNs), and it's handy to have them grouped here.

[scholzj]

Why would we need to complement them? Or what would we need to complement them with? I do not understand that.

[fvaleri]

We have the fully qualified class names in addition to configuration keys.

[fvaleri]

@scholzj we are ready with the following PR, so it would be helpful if you could have another look at this.

[katheris]

Am I right in understanding that these changes are here because we added the StrimziMetricsReporterConfig class in this PR?

[fvaleri]

@scholzj @katheris @ppatierno last commit should implement the strategy we discussed during the community meeting and address latest comments. Let us know if you are happy with it.