Skip to content

NIFI-16298 Expose Primary Node and Cluster Coordinator role as Prometheus metrics #11628

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
abij wants to merge 1 commit into
base: main
Choose a base branch
from
Open

NIFI-16298 Expose Primary Node and Cluster Coordinator role as Prometheus metrics #11628

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions ...web/nifi-web-api/src/main/java/org/apache/nifi/prometheusutil/ClusterMetricsRegistry.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,5 +48,17 @@ public ClusterMetricsRegistry() {
.help("The total number of nodes in this cluster")
.labelNames("instance")
.register(registry));

nameToGaugeMap.put("IS_PRIMARY_NODE", Gauge.build()
.name("cluster_is_primary_node")
.help("Whether this NiFi instance is the Primary Node. Values are 0 or 1")
.labelNames("instance")
.register(registry));

nameToGaugeMap.put("IS_CLUSTER_COORDINATOR", Gauge.build()
.name("cluster_is_cluster_coordinator")
.help("Whether this NiFi instance is the Cluster Coordinator. Values are 0 or 1")
.labelNames("instance")
.register(registry));
}
}
5 changes: 4 additions & 1 deletion ...-web/nifi-web-api/src/main/java/org/apache/nifi/prometheusutil/PrometheusMetricsUtil.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -500,12 +500,15 @@ public static void createVersionInfoMetrics(final VersionInfoRegistry versionInf
}

public static CollectorRegistry createClusterMetrics(final ClusterMetricsRegistry clusterMetricsRegistry, final String instId, final boolean isClustered, final boolean isConnectedToCluster,
final int connectedNodeCount, final int totalNodeCount) {
final int connectedNodeCount, final int totalNodeCount,
final boolean isPrimaryNode, final boolean isClusterCoordinator) {
final String instanceId = StringUtils.isEmpty(instId) ? DEFAULT_LABEL_STRING : instId;
clusterMetricsRegistry.setDataPoint(isClustered ? 1 : 0, "IS_CLUSTERED", instanceId);
clusterMetricsRegistry.setDataPoint(isConnectedToCluster ? 1 : 0, "IS_CONNECTED_TO_CLUSTER", instanceId);
clusterMetricsRegistry.setDataPoint(connectedNodeCount, "CONNECTED_NODE_COUNT", instanceId);
clusterMetricsRegistry.setDataPoint(totalNodeCount, "TOTAL_NODE_COUNT", instanceId);
clusterMetricsRegistry.setDataPoint(isPrimaryNode ? 1 : 0, "IS_PRIMARY_NODE", instanceId);
clusterMetricsRegistry.setDataPoint(isClusterCoordinator ? 1 : 0, "IS_CLUSTER_COORDINATOR", instanceId);

return clusterMetricsRegistry.getRegistry();
}
Expand Down
3 changes: 2 additions & 1 deletion ...rk/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8175,7 +8175,8 @@ protected Collection<AbstractMetricsRegistry> populateFlowMetrics(FlowMetricsRep
}
final boolean isClustered = clusterCoordinator != null;
final boolean isConnectedToCluster = isClustered() && clusterCoordinator.isConnected();
PrometheusMetricsUtil.createClusterMetrics(clusterMetricsRegistry, instanceId, isClustered, isConnectedToCluster, connectedNodeCount, totalNodeCount);
PrometheusMetricsUtil.createClusterMetrics(clusterMetricsRegistry, instanceId, isClustered, isConnectedToCluster, connectedNodeCount, totalNodeCount,
controllerFacade.isPrimary(), controllerFacade.isClusterCoordinator());
Collection<AbstractMetricsRegistry> metricsRegistries = Arrays.asList(
nifiMetricsRegistry,
jvmMetricsRegistry,
Expand Down
14 changes: 14 additions & 0 deletions .../nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/controller/ControllerFacade.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -494,6 +494,20 @@ public boolean isClustered() {
return flowController.isClustered();
}

/**
* @return true if this node is the Primary Node
*/
public boolean isPrimary() {
return flowController.isPrimary();
}

/**
* @return true if this node is the Cluster Coordinator
*/
public boolean isClusterCoordinator() {
return flowController.isClusterCoordinator();
}

/**
* Gets the name of this controller.
*
Expand Down
62 changes: 43 additions & 19 deletions ...amework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/api/TestFlowResource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -124,6 +124,12 @@ public class TestFlowResource {
private static final int COMPONENT_TYPE_VALUE_INDEX = 1;
private static final String CLUSTER_TYPE_LABEL = "cluster";
private static final String CLUSTER_LABEL_KEY = "instance";
private static final String IS_CLUSTERED_LABEL_KEY = "cluster_is_clustered";
private static final String IS_CONNECTED_TO_CLUSTER_LABEL_KEY = "cluster_is_connected_to_cluster";
private static final String CONNECTED_NODE_COUNT_LABEL_KEY = "cluster_connected_node_count";
private static final String TOTAL_NODE_COUNT_LABEL_KEY = "cluster_total_node_count";
private static final String IS_PRIMARY_NODE_LABEL_KEY = "cluster_is_primary_node";
private static final String IS_CLUSTER_COORDINATOR_LABEL_KEY = "cluster_is_cluster_coordinator";
private static final String SAMPLE_REGISTRY_ID = "0e87642a-7720-4799-a3bd-04db74b86e85";
private static final String SAMPLE_BRANCH_ID_A = "c302f541-976e-4c51-952d-345516444e3d";
private static final String SAMPLE_BUCKET_ID_A = "23da421d-a8da-4fa3-939e-658d8f35b972";
Expand Down Expand Up @@ -276,13 +282,20 @@ public void testGetFlowMetricsPrometheusAsJson() throws IOException {
assertTrue(metrics.containsKey(ROOT_FIELD_NAME));

final List<Sample> registryList = metrics.get(ROOT_FIELD_NAME);
assertEquals(13, registryList.size());
assertEquals(15, registryList.size());

@pvillard31 pvillard31 Sep 8, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we assert the names and values of both new metrics instead of only updating the sample counts? The current assertions would pass if either metric were renamed or its boolean value were inverted.

@abij abij Sep 8, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I've added asserts for the 6 cluster metrics, checking the first sample matching the name.
And for clarification renamed the getResult into getSampleCounts because its counting based on the keys.


final Map<String, Long> result = getResult(registryList);
assertEquals(3L, result.get(SAMPLE_NAME_JVM));
assertEquals(4L, result.get(SAMPLE_LABEL_VALUES_PROCESS_GROUP));
assertEquals(2L, result.get(SAMPLE_LABEL_VALUES_ROOT_PROCESS_GROUP));
assertEquals(4L, result.get(CLUSTER_LABEL_KEY));
final Map<String, Long> sampleCounts = getSampleCounts(registryList);
assertEquals(3L, sampleCounts.get(SAMPLE_NAME_JVM));
assertEquals(4L, sampleCounts.get(SAMPLE_LABEL_VALUES_PROCESS_GROUP));
assertEquals(2L, sampleCounts.get(SAMPLE_LABEL_VALUES_ROOT_PROCESS_GROUP));
assertEquals(6L, sampleCounts.get(CLUSTER_LABEL_KEY));

assertSample(registryList, IS_CLUSTERED_LABEL_KEY, 1.0);
assertSample(registryList, IS_CONNECTED_TO_CLUSTER_LABEL_KEY, 1.0);
assertSample(registryList, CONNECTED_NODE_COUNT_LABEL_KEY, 2.0);
assertSample(registryList, TOTAL_NODE_COUNT_LABEL_KEY, 3.0);
assertSample(registryList, IS_PRIMARY_NODE_LABEL_KEY, 1.0);
assertSample(registryList, IS_CLUSTER_COORDINATOR_LABEL_KEY, 0.0);
}

@Test
Expand All @@ -301,8 +314,8 @@ public void testGetFlowMetricsPrometheusAsJsonSampleName() throws IOException {
final List<Sample> registryList = metrics.get(ROOT_FIELD_NAME);
assertEquals(3, registryList.size());

final Map<String, Long> result = getResult(registryList);
assertEquals(3L, result.get(SAMPLE_NAME_JVM));
final Map<String, Long> sampleCounts = getSampleCounts(registryList);
assertEquals(3L, sampleCounts.get(SAMPLE_NAME_JVM));
}

@Test
Expand All @@ -321,8 +334,8 @@ public void testGetFlowMetricsPrometheusAsJsonSampleNameStartsWithPattern() thro
final List<Sample> registryList = metrics.get(ROOT_FIELD_NAME);
assertEquals(2, registryList.size());

final Map<String, Long> result = getResult(registryList);
assertEquals(2L, result.get(SAMPLE_NAME_JVM));
final Map<String, Long> sampleCounts = getSampleCounts(registryList);
assertEquals(2L, sampleCounts.get(SAMPLE_NAME_JVM));
}

@Test
Expand All @@ -341,8 +354,8 @@ public void testGetFlowMetricsPrometheusAsJsonSampleLabelValue() throws IOExcept
final List<Sample> registryList = metrics.get(ROOT_FIELD_NAME);
assertEquals(2, registryList.size());

final Map<String, Long> result = getResult(registryList);
assertEquals(2L, result.get(SAMPLE_LABEL_VALUES_ROOT_PROCESS_GROUP));
final Map<String, Long> sampleCounts = getSampleCounts(registryList);
assertEquals(2L, sampleCounts.get(SAMPLE_LABEL_VALUES_ROOT_PROCESS_GROUP));
}

@Test
Expand All @@ -362,9 +375,9 @@ public void testGetFlowMetricsPrometheusAsJsonSampleNameAndSampleLabelValue() th
final List<Sample> registryList = metrics.get(ROOT_FIELD_NAME);
assertEquals(5, registryList.size());

final Map<String, Long> result = getResult(registryList);
assertEquals(3L, result.get(SAMPLE_NAME_JVM));
assertEquals(2L, result.get(SAMPLE_LABEL_VALUES_ROOT_PROCESS_GROUP));
final Map<String, Long> sampleCounts = getSampleCounts(registryList);
assertEquals(3L, sampleCounts.get(SAMPLE_NAME_JVM));
assertEquals(2L, sampleCounts.get(SAMPLE_LABEL_VALUES_ROOT_PROCESS_GROUP));
}

@Test
Expand Down Expand Up @@ -753,14 +766,14 @@ private Map<String, List<Sample>> convertJsonResponseToMap(final Response respon
return mapper.readValue(json, typeReference);
}

private Map<String, Long> getResult(final List<Sample> registries) {
return registries.stream()
private Map<String, Long> getSampleCounts(final List<Sample> samples) {
return samples.stream()
.collect(Collectors.groupingBy(
sample -> getResultKey(sample),
sample -> getSampleKey(sample),
Collectors.counting()));
}

private String getResultKey(final Sample sample) {
private String getSampleKey(final Sample sample) {
if (sample.labelNames.contains(COMPONENT_TYPE_LABEL)) {
return sample.labelValues.get(COMPONENT_TYPE_VALUE_INDEX);
}
Expand All @@ -770,6 +783,15 @@ private String getResultKey(final Sample sample) {
return SAMPLE_NAME_JVM;
}

private void assertSample(final List<Sample> samples, final String sampleName, final double expectedValue) {
final Sample sample = samples.stream()
.filter(s -> sampleName.equals(s.name))
.findFirst()
.orElse(null);
assertNotNull(sample, String.format("Sample [%s] not found", sampleName));
assertEquals(expectedValue, sample.value);
}

private static List<CollectorRegistry> getCollectorRegistriesForJson() {
final List<CollectorRegistry> registryList = new ArrayList<>();

Expand Down Expand Up @@ -836,6 +858,8 @@ private static CollectorRegistry getClusterMetricsRegistry() {
clusterMetricsRegistry.setDataPoint(1, "IS_CONNECTED_TO_CLUSTER", "B1Id");
clusterMetricsRegistry.setDataPoint(2, "CONNECTED_NODE_COUNT", "B1Id");
clusterMetricsRegistry.setDataPoint(3, "TOTAL_NODE_COUNT", "B1Id");
clusterMetricsRegistry.setDataPoint(1, "IS_PRIMARY_NODE", "B1Id");
clusterMetricsRegistry.setDataPoint(0, "IS_CLUSTER_COORDINATOR", "B1Id");

return clusterMetricsRegistry.getRegistry();
}
Expand Down
Loading