This is an automated email from the ASF dual-hosted git repository.
pvillard31 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new 6d5d007b951 NIFI-16298 Expose Primary Node and Cluster Coordinator
role as Prometheus metrics (#11628)
6d5d007b951 is described below
commit 6d5d007b951587aa044beeb9019d9fa82b563a77
Author: Alexander <[email protected]>
AuthorDate: Fri Sep 25 15:22:35 2026 +0200
NIFI-16298 Expose Primary Node and Cluster Coordinator role as Prometheus
metrics (#11628)
---
.../prometheusutil/ClusterMetricsRegistry.java | 12 ++++
.../nifi/prometheusutil/PrometheusMetricsUtil.java | 5 +-
.../apache/nifi/web/StandardNiFiServiceFacade.java | 3 +-
.../nifi/web/controller/ControllerFacade.java | 14 +++++
.../org/apache/nifi/web/api/TestFlowResource.java | 64 +++++++++++++++-------
5 files changed, 76 insertions(+), 22 deletions(-)
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/prometheusutil/ClusterMetricsRegistry.java
b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/prometheusutil/ClusterMetricsRegistry.java
index 462b2328812..787b6e22bc5 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/prometheusutil/ClusterMetricsRegistry.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/prometheusutil/ClusterMetricsRegistry.java
@@ -48,5 +48,17 @@ public class ClusterMetricsRegistry extends
AbstractMetricsRegistry {
.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));
}
}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/prometheusutil/PrometheusMetricsUtil.java
b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/prometheusutil/PrometheusMetricsUtil.java
index 60c7d4d223a..1d799a0ab69 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/prometheusutil/PrometheusMetricsUtil.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/prometheusutil/PrometheusMetricsUtil.java
@@ -500,12 +500,15 @@ public class PrometheusMetricsUtil {
}
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();
}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java
b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java
index 8e2236e34ce..ad76675bf18 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java
@@ -8236,7 +8236,8 @@ public class StandardNiFiServiceFacade implements
NiFiServiceFacade {
}
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,
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/controller/ControllerFacade.java
b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/controller/ControllerFacade.java
index e9a2436aa61..d33b4156dfc 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/controller/ControllerFacade.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/controller/ControllerFacade.java
@@ -494,6 +494,20 @@ public class ControllerFacade implements Authorizable {
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.
*
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/api/TestFlowResource.java
b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/api/TestFlowResource.java
index 3841ed70afe..602dd311304 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/api/TestFlowResource.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/api/TestFlowResource.java
@@ -139,6 +139,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";
@@ -295,13 +301,20 @@ public class TestFlowResource {
assertTrue(metrics.containsKey(ROOT_FIELD_NAME));
final List<Sample> registryList = metrics.get(ROOT_FIELD_NAME);
- assertEquals(13, registryList.size());
-
- 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));
+ assertEquals(15, registryList.size());
+
+ 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
@@ -320,8 +333,8 @@ public class TestFlowResource {
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
@@ -340,8 +353,8 @@ public class TestFlowResource {
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
@@ -360,8 +373,8 @@ public class TestFlowResource {
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
@@ -381,9 +394,9 @@ public class TestFlowResource {
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
@@ -1004,14 +1017,14 @@ public class TestFlowResource {
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);
}
@@ -1021,6 +1034,15 @@ public class TestFlowResource {
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<>();
@@ -1087,6 +1109,8 @@ public class TestFlowResource {
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();
}
