This is an automated email from the ASF dual-hosted git repository.
tkhurana pushed a commit to branch PHOENIX-7562-feature-new
in repository https://gitbox.apache.org/repos/asf/phoenix.git
The following commit(s) were added to refs/heads/PHOENIX-7562-feature-new by
this push:
new 956c7518be PHOENIX-7562 Add HAGroupStore Metrics2 observability (#2576)
956c7518be is described below
commit 956c7518be5e69018e76b2af3f7dccbbc11813bd
Author: ritegarg <[email protected]>
AuthorDate: Fri Jul 24 14:49:39 2026 -0700
PHOENIX-7562 Add HAGroupStore Metrics2 observability (#2576)
Expose a focused per-group health, state, failure, and listener-latency
contract without coupling metrics to HA decisions.
---
.../apache/phoenix/jdbc/HAGroupStoreClient.java | 98 ++++--
.../apache/phoenix/jdbc/HAGroupStoreRecord.java | 45 ++-
.../jdbc/metrics/HAGroupStoreMetricValues.java | 196 +++++++++++
.../jdbc/metrics/HAGroupStoreMetricsSource.java | 133 ++++++++
.../metrics/HAGroupStoreMetricsSourceFactory.java | 40 +++
.../metrics/HAGroupStoreMetricsSourceImpl.java | 171 ++++++++++
.../apache/phoenix/jdbc/HAGroupStoreMetricsIT.java | 358 +++++++++++++++++++++
.../phoenix/jdbc/HAGroupStoreRecordTest.java | 19 ++
.../metrics/HAGroupStoreMetricsSourceImplTest.java | 197 ++++++++++++
9 files changed, 1222 insertions(+), 35 deletions(-)
diff --git
a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HAGroupStoreClient.java
b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HAGroupStoreClient.java
index ec00184969..82c8460f78 100644
---
a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HAGroupStoreClient.java
+++
b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HAGroupStoreClient.java
@@ -78,6 +78,8 @@ import
org.apache.phoenix.exception.StaleHAGroupStoreRecordVersionException;
import org.apache.phoenix.jdbc.ClusterRoleRecord.ClusterRole;
import org.apache.phoenix.jdbc.ClusterRoleRecord.RegistryType;
import org.apache.phoenix.jdbc.HAGroupStoreRecord.HAGroupState;
+import org.apache.phoenix.jdbc.metrics.HAGroupStoreMetricsSource;
+import org.apache.phoenix.jdbc.metrics.HAGroupStoreMetricsSourceFactory;
import org.apache.phoenix.query.QueryServices;
import org.apache.phoenix.query.QueryServicesOptions;
import org.apache.phoenix.util.JDBCUtil;
@@ -118,6 +120,7 @@ public class HAGroupStoreClient implements Closeable {
new ConcurrentHashMap<>();
// HAGroupName for this instance
private final String haGroupName;
+ private final HAGroupStoreMetricsSource metricsSource;
// PathChildrenCache for current cluster and HAGroupName
private PathChildrenCache pathChildrenCache = null;
// Watches the peer cluster's HA record; owns the peer cache/admin, retry,
and visibility.
@@ -278,6 +281,8 @@ public class HAGroupStoreClient implements Closeable {
final String zkUrl) {
this.conf = conf;
this.haGroupName = haGroupName;
+ this.metricsSource =
HAGroupStoreMetricsSourceFactory.getInstanceForHAGroup(haGroupName);
+ resetMetricGauges();
this.zkUrl = zkUrl;
this.waitTimeForSyncModeInMs = (long) Math.ceil(
conf.getLong(ZK_SESSION_TIMEOUT, DEFAULT_ZK_SESSION_TIMEOUT) *
ZK_SESSION_TIMEOUT_MULTIPLIER);
@@ -299,7 +304,10 @@ public class HAGroupStoreClient implements Closeable {
initializeZNodeIfNeeded();
if (this.pathChildrenCache != null) {
this.isHealthy = true;
+ metricsSource.setLocalCacheHealthy(true);
HAGroupStoreRecord local = getHAGroupStoreRecord();
+ metricsSource.setCurrentLocalState(
+ local != null ? local.getHAGroupState() :
HAGroupStoreRecord.HAGroupState.UNKNOWN);
this.lastConfiguredPeerZKUrl = local != null ? local.getPeerZKUrl() :
null;
peerWatcher.reconfigure(this.lastConfiguredPeerZKUrl);
} else {
@@ -317,6 +325,7 @@ public class HAGroupStoreClient implements Closeable {
} catch (Exception e) {
this.isHealthy = false;
+ resetMetricGauges();
close();
LOGGER.error("Unexpected error occurred while initializing
HAGroupStoreClient, "
+ "marking cache as unhealthy", e);
@@ -686,8 +695,21 @@ public class HAGroupStoreClient implements Closeable {
// Update the system table on best effort basis for HA group
// In case of failure, we will log the error and continue.
- private void updateSystemTableHAGroupRecordSilently(String haGroupName,
- SystemTableHAGroupRecord record) throws SQLException {
+ private boolean updateSystemTableHAGroupRecordSilently(String haGroupName,
+ SystemTableHAGroupRecord record) {
+ try {
+ updateSystemTableHAGroupRecord(haGroupName, record);
+ return true;
+ } catch (Exception e) {
+ metricsSource.incrementSystemTableSyncFailedCount();
+ LOGGER.error("Failed to update system table on best effort basis for HA
group {}", haGroupName,
+ e);
+ return false;
+ }
+ }
+
+ private void updateSystemTableHAGroupRecord(String haGroupName,
SystemTableHAGroupRecord record)
+ throws SQLException {
StringBuilder updateQuery = new StringBuilder("UPSERT INTO " +
SYSTEM_HA_GROUP_NAME + " (");
StringBuilder valuesClause = new StringBuilder(" VALUES (");
List<Object> parameters = new ArrayList<>();
@@ -794,10 +816,6 @@ public class HAGroupStoreClient implements Closeable {
pstmt.executeUpdate();
conn.commit();
- } catch (Exception e) {
- LOGGER.error(
- "Failed to update system table on best" + "effort basis for HA group
{}, error: {}",
- haGroupName, e);
}
}
@@ -832,7 +850,8 @@ public class HAGroupStoreClient implements Closeable {
* Syncs data from ZooKeeper (source of truth) to the system table. This
method is called
* periodically to ensure consistency.
*/
- private void syncZKToSystemTable() {
+ @VisibleForTesting
+ void syncZKToSystemTable() {
if (!isHealthy) {
LOGGER.debug("HAGroupStoreClient is not healthy," + "skipping sync for
HA group {}",
haGroupName);
@@ -875,16 +894,26 @@ public class HAGroupStoreClient implements Closeable {
}
// Update system table with ZK data
- updateSystemTableHAGroupRecordSilently(haGroupName,
newSystemTableRecord);
- LOGGER.info("Successfully synced ZK data to system table for HA group
{}", haGroupName);
- } catch (IOException | SQLException e) {
+ if (updateSystemTableHAGroupRecordSilently(haGroupName,
newSystemTableRecord)) {
+ LOGGER.info("Successfully synced ZK data to system table for HA group
{}", haGroupName);
+ }
+ } catch (Exception e) {
+ metricsSource.incrementSystemTableSyncFailedCount();
long syncIntervalSeconds =
conf.getLong(HA_GROUP_STORE_SYNC_INTERVAL_SECONDS,
DEFAULT_HA_GROUP_STORE_SYNC_INTERVAL_SECONDS);
LOGGER.error("Failed to sync ZK data to system table for HA group on
best effort basis {},"
- + "retrying in {} seconds", haGroupName, syncIntervalSeconds);
+ + "retrying in {} seconds", haGroupName, syncIntervalSeconds, e);
}
}
+ private void resetMetricGauges() {
+ metricsSource.setLocalCacheHealthy(false);
+ metricsSource.setPeerVisible(false);
+ metricsSource.setDegradedStandbyActive(false);
+ metricsSource.setCurrentLocalState(HAGroupState.UNKNOWN);
+ metricsSource.setCurrentPeerState(HAGroupState.UNKNOWN);
+ }
+
/**
* Listener for the local cache: tracks local state transitions (suppressing
STANDBY while
* degraded), keeps the peer watcher pointed at the current peer ZK, drives
the legacy CRR sync,
@@ -908,6 +937,9 @@ public class HAGroupStoreClient implements Closeable {
// unreachable peer never blocks local event processing.
String peerZKUrl = record.getPeerZKUrl();
if (!Objects.equals(peerZKUrl, lastConfiguredPeerZKUrl)) {
+ // An in-flight callback from the prior peer can briefly
overwrite UNKNOWN. The new
+ // watcher self-heals this best-effort gauge; exact generation
fencing is unnecessary.
+ metricsSource.setCurrentPeerState(HAGroupState.UNKNOWN);
lastConfiguredPeerZKUrl = peerZKUrl;
peerWatcher.reconfigureAsync(peerZKUrl);
}
@@ -916,12 +948,18 @@ public class HAGroupStoreClient implements Closeable {
break;
case CONNECTION_LOST:
case CONNECTION_SUSPENDED:
+ boolean wasHealthy = isHealthy;
isHealthy = false;
+ metricsSource.setLocalCacheHealthy(false);
+ if (wasHealthy) {
+ metricsSource.incrementLocalZkConnectionLostCount();
+ }
LOGGER.warn("LOCAL HAGroupStoreClient cache connection
lost/suspended for HA group {}",
haGroupName);
break;
case CONNECTION_RECONNECTED:
isHealthy = true;
+ metricsSource.setLocalCacheHealthy(true);
LOGGER.info("LOCAL HAGroupStoreClient cache connection reconnected
for HA group {}",
haGroupName);
break;
@@ -1007,6 +1045,7 @@ public class HAGroupStoreClient implements Closeable {
HAGroupState fromState = lastKnownPeerState;
HAGroupState toState = peerRecord.getHAGroupState();
lastKnownPeerState = toState;
+ metricsSource.setCurrentPeerState(toState);
LOGGER.info("Detected state transition for HA group {} from {} to {} on
PEER cluster",
haGroupName, fromState, toState);
notifySubscribers(fromState, toState, stat != null ? stat.getMtime() :
0L, ClusterType.PEER,
@@ -1017,11 +1056,15 @@ public class HAGroupStoreClient implements Closeable {
@Override
public void onPeerVisible() {
+ metricsSource.setPeerVisible(true);
clearLocalDegradedStandbyIfStillStandby();
}
@Override
public void onPeerBlind() {
+ metricsSource.setPeerVisible(false);
+ metricsSource.setCurrentPeerState(HAGroupState.UNKNOWN);
+ metricsSource.incrementPeerBlindCount();
// Peer just went blind while local is STANDBY (present re-checks):
subscribers last saw the
// raw STANDBY, so that is the prior effective state.
presentLocalDegradedStandbyIfStandby(HAGroupState.STANDBY);
@@ -1060,6 +1103,8 @@ public class HAGroupStoreClient implements Closeable {
return false;
}
localDegradedStandbyActive = true;
+ metricsSource.setDegradedStandbyActive(true);
+ metricsSource.incrementDegradedStandbyPresentedCount();
}
LOGGER.warn("Peer not visible for HA group {}; presenting local
DEGRADED_STANDBY "
+ "(reason=peer-blind)", haGroupName);
@@ -1086,6 +1131,7 @@ public class HAGroupStoreClient implements Closeable {
}
local = readLocalRecordQuietly();
localDegradedStandbyActive = false;
+ metricsSource.setDegradedStandbyActive(false);
recover = local != null && local.getHAGroupState() ==
HAGroupState.STANDBY;
}
if (recover) {
@@ -1122,7 +1168,8 @@ public class HAGroupStoreClient implements Closeable {
instances.computeIfPresent(key, (k, v) -> v.isEmpty() ? null : v);
}
- private void shutdownSyncExecutor() {
+ @VisibleForTesting
+ void shutdownSyncExecutor() {
if (syncExecutor != null) {
MoreExecutors.shutdownAndAwaitTermination(syncExecutor, 5,
TimeUnit.SECONDS);
syncExecutor = null;
@@ -1194,6 +1241,7 @@ public class HAGroupStoreClient implements Closeable {
waitTime = waitTimeForSyncModeInMs;
}
} else {
+ metricsSource.incrementInvalidTransitionRejectedCount();
throw new InvalidClusterRoleTransitionException(
"Cannot transition from " + currentHAGroupState + " to " +
newHAGroupState);
}
@@ -1439,6 +1487,9 @@ public class HAGroupStoreClient implements Closeable {
synchronized (localTransitionLock) {
HAGroupState oldState = lastKnownLocalState;
lastKnownLocalState = newState;
+ metricsSource.setCurrentLocalState(newState);
+ metricsSource
+ .setDegradedStandbyActive(localDegradedStandbyActive && newState ==
HAGroupState.STANDBY);
// Fail closed for a final STANDBY while the peer is not visible.
STANDBY_TO_ACTIVE and
// ABORT_TO_STANDBY pass through because replay listens for those
failover signals.
if (newState == HAGroupState.STANDBY) {
@@ -1496,18 +1547,23 @@ public class HAGroupStoreClient implements Closeable {
// Notify all listeners with error isolation
if (!listenersToNotify.isEmpty()) {
- LOGGER.info("Notifying {} listeners of state transitionfor HA group {}
from {} to {} on {} "
+ LOGGER.info("Notifying {} listeners of state transition for HA group {}
from {} to {} on {} "
+ "cluster", listenersToNotify.size(), haGroupName, fromState,
toState, clusterType);
- for (HAGroupStateListener listener : listenersToNotify) {
- try {
- listener.onStateChange(haGroupName, fromState, toState,
modifiedTime, clusterType,
- lastSyncStateTimeInMs);
- } catch (Exception e) {
- LOGGER.error("Error notifying listener of state transition for HA
group {} from {} to "
- + "{} on {} cluster", haGroupName, fromState, toState,
clusterType, e);
- // Continue notifying other listeners
+ long startTimeNs = System.nanoTime();
+ try {
+ for (HAGroupStateListener listener : listenersToNotify) {
+ try {
+ listener.onStateChange(haGroupName, fromState, toState,
modifiedTime, clusterType,
+ lastSyncStateTimeInMs);
+ } catch (Exception e) {
+ metricsSource.incrementNotificationListenerErrorCount();
+ LOGGER.error("Error notifying listener of state transition for HA
group {} from {} to "
+ + "{} on {} cluster", haGroupName, fromState, toState,
clusterType, e);
+ }
}
+ } finally {
+ metricsSource.updateSubscriberNotifyTime(System.nanoTime() -
startTimeNs);
}
}
}
diff --git
a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HAGroupStoreRecord.java
b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HAGroupStoreRecord.java
index 0494ca9d34..be61a644dc 100644
---
a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HAGroupStoreRecord.java
+++
b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HAGroupStoreRecord.java
@@ -47,17 +47,21 @@ public class HAGroupStoreRecord {
/**
* Enum representing the HA group state with each state having a
corresponding ClusterRole.
+ * <p>
+ * Metric codes are a stable external monitoring contract. They are
independent of enum ordinal:
+ * never renumber or reuse an existing code, assign new states new codes,
and keep code {@code 0}
+ * reserved for {@link #UNKNOWN}.
*/
public enum HAGroupState {
- ABORT_TO_ACTIVE_IN_SYNC,
- ABORT_TO_ACTIVE_NOT_IN_SYNC,
- ABORT_TO_STANDBY,
- ACTIVE_IN_SYNC,
- ACTIVE_NOT_IN_SYNC,
- ACTIVE_NOT_IN_SYNC_TO_STANDBY,
- ACTIVE_NOT_IN_SYNC_WITH_OFFLINE_PEER,
- ACTIVE_IN_SYNC_TO_STANDBY,
- ACTIVE_WITH_OFFLINE_PEER,
+ ABORT_TO_ACTIVE_IN_SYNC(1),
+ ABORT_TO_ACTIVE_NOT_IN_SYNC(2),
+ ABORT_TO_STANDBY(3),
+ ACTIVE_IN_SYNC(4),
+ ACTIVE_NOT_IN_SYNC(5),
+ ACTIVE_NOT_IN_SYNC_TO_STANDBY(6),
+ ACTIVE_NOT_IN_SYNC_WITH_OFFLINE_PEER(7),
+ ACTIVE_IN_SYNC_TO_STANDBY(8),
+ ACTIVE_WITH_OFFLINE_PEER(9),
/**
* Degraded standby. Used two ways: (1) a persisted state for a
reader-degraded standby (see
* {@link HAGroupStoreManager#setReaderToDegraded(String)}), governed by
the allowedTransitions
@@ -66,13 +70,26 @@ public class HAGroupStoreRecord {
* cannot see its peer - that overlay is never persisted to ZK and is
cleared (not transitioned)
* when the peer becomes visible again.
*/
- DEGRADED_STANDBY,
- OFFLINE,
- STANDBY,
- STANDBY_TO_ACTIVE,
- UNKNOWN;
+ DEGRADED_STANDBY(10),
+ OFFLINE(11),
+ STANDBY(12),
+ STANDBY_TO_ACTIVE(13),
+ UNKNOWN(0);
private Set<HAGroupState> allowedTransitions;
+ private final int metricCode;
+
+ HAGroupState(int metricCode) {
+ this.metricCode = metricCode;
+ }
+
+ /**
+ * Returns the stable, append-only code exported by HAGroupStore state
gauges.
+ * @return non-ordinal monitoring code; {@code 0} is reserved for {@link
#UNKNOWN}
+ */
+ public int getMetricCode() {
+ return metricCode;
+ }
/**
* Gets the corresponding ClusterRole for this HAGroupState.
diff --git
a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/metrics/HAGroupStoreMetricValues.java
b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/metrics/HAGroupStoreMetricValues.java
new file mode 100644
index 0000000000..23f5787d3c
--- /dev/null
+++
b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/metrics/HAGroupStoreMetricValues.java
@@ -0,0 +1,196 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.jdbc.metrics;
+
+import
org.apache.phoenix.thirdparty.com.google.common.annotations.VisibleForTesting;
+
+/** Immutable snapshot of HAGroupStore metrics used by tests. */
+@VisibleForTesting
+public final class HAGroupStoreMetricValues {
+
+ private final long localCacheHealthStatus;
+ private final long peerVisibilityStatus;
+ private final long degradedStandbyActive;
+ private final long currentLocalState;
+ private final long currentPeerState;
+ private final long localZkConnectionLostCount;
+ private final long peerBlindCount;
+ private final long degradedStandbyPresentedCount;
+ private final long invalidTransitionRejectedCount;
+ private final long systemTableSyncFailedCount;
+ private final long notificationListenerErrorCount;
+ private final long subscriberNotifyTimeCount;
+ private final long subscriberNotifyTimeMaxMs;
+
+ private HAGroupStoreMetricValues(Builder builder) {
+ localCacheHealthStatus = builder.localCacheHealthStatus;
+ peerVisibilityStatus = builder.peerVisibilityStatus;
+ degradedStandbyActive = builder.degradedStandbyActive;
+ currentLocalState = builder.currentLocalState;
+ currentPeerState = builder.currentPeerState;
+ localZkConnectionLostCount = builder.localZkConnectionLostCount;
+ peerBlindCount = builder.peerBlindCount;
+ degradedStandbyPresentedCount = builder.degradedStandbyPresentedCount;
+ invalidTransitionRejectedCount = builder.invalidTransitionRejectedCount;
+ systemTableSyncFailedCount = builder.systemTableSyncFailedCount;
+ notificationListenerErrorCount = builder.notificationListenerErrorCount;
+ subscriberNotifyTimeCount = builder.subscriberNotifyTimeCount;
+ subscriberNotifyTimeMaxMs = builder.subscriberNotifyTimeMaxMs;
+ }
+
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ public long getLocalCacheHealthStatus() {
+ return localCacheHealthStatus;
+ }
+
+ public long getPeerVisibilityStatus() {
+ return peerVisibilityStatus;
+ }
+
+ public long getDegradedStandbyActive() {
+ return degradedStandbyActive;
+ }
+
+ public long getCurrentLocalState() {
+ return currentLocalState;
+ }
+
+ public long getCurrentPeerState() {
+ return currentPeerState;
+ }
+
+ public long getLocalZkConnectionLostCount() {
+ return localZkConnectionLostCount;
+ }
+
+ public long getPeerBlindCount() {
+ return peerBlindCount;
+ }
+
+ public long getDegradedStandbyPresentedCount() {
+ return degradedStandbyPresentedCount;
+ }
+
+ public long getInvalidTransitionRejectedCount() {
+ return invalidTransitionRejectedCount;
+ }
+
+ public long getSystemTableSyncFailedCount() {
+ return systemTableSyncFailedCount;
+ }
+
+ public long getNotificationListenerErrorCount() {
+ return notificationListenerErrorCount;
+ }
+
+ public long getSubscriberNotifyTimeCount() {
+ return subscriberNotifyTimeCount;
+ }
+
+ public long getSubscriberNotifyTimeMaxMs() {
+ return subscriberNotifyTimeMaxMs;
+ }
+
+ public static final class Builder {
+ private long localCacheHealthStatus;
+ private long peerVisibilityStatus;
+ private long degradedStandbyActive;
+ private long currentLocalState;
+ private long currentPeerState;
+ private long localZkConnectionLostCount;
+ private long peerBlindCount;
+ private long degradedStandbyPresentedCount;
+ private long invalidTransitionRejectedCount;
+ private long systemTableSyncFailedCount;
+ private long notificationListenerErrorCount;
+ private long subscriberNotifyTimeCount;
+ private long subscriberNotifyTimeMaxMs;
+
+ public Builder setLocalCacheHealthStatus(long value) {
+ localCacheHealthStatus = value;
+ return this;
+ }
+
+ public Builder setPeerVisibilityStatus(long value) {
+ peerVisibilityStatus = value;
+ return this;
+ }
+
+ public Builder setDegradedStandbyActive(long value) {
+ degradedStandbyActive = value;
+ return this;
+ }
+
+ public Builder setCurrentLocalState(long value) {
+ currentLocalState = value;
+ return this;
+ }
+
+ public Builder setCurrentPeerState(long value) {
+ currentPeerState = value;
+ return this;
+ }
+
+ public Builder setLocalZkConnectionLostCount(long value) {
+ localZkConnectionLostCount = value;
+ return this;
+ }
+
+ public Builder setPeerBlindCount(long value) {
+ peerBlindCount = value;
+ return this;
+ }
+
+ public Builder setDegradedStandbyPresentedCount(long value) {
+ degradedStandbyPresentedCount = value;
+ return this;
+ }
+
+ public Builder setInvalidTransitionRejectedCount(long value) {
+ invalidTransitionRejectedCount = value;
+ return this;
+ }
+
+ public Builder setSystemTableSyncFailedCount(long value) {
+ systemTableSyncFailedCount = value;
+ return this;
+ }
+
+ public Builder setNotificationListenerErrorCount(long value) {
+ notificationListenerErrorCount = value;
+ return this;
+ }
+
+ public Builder setSubscriberNotifyTimeCount(long value) {
+ subscriberNotifyTimeCount = value;
+ return this;
+ }
+
+ public Builder setSubscriberNotifyTimeMaxMs(long value) {
+ subscriberNotifyTimeMaxMs = value;
+ return this;
+ }
+
+ public HAGroupStoreMetricValues build() {
+ return new HAGroupStoreMetricValues(this);
+ }
+ }
+}
diff --git
a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/metrics/HAGroupStoreMetricsSource.java
b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/metrics/HAGroupStoreMetricsSource.java
new file mode 100644
index 0000000000..0af587c503
--- /dev/null
+++
b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/metrics/HAGroupStoreMetricsSource.java
@@ -0,0 +1,133 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.jdbc.metrics;
+
+import org.apache.hadoop.hbase.metrics.BaseSource;
+import org.apache.phoenix.jdbc.HAGroupStoreRecord.HAGroupState;
+
+import
org.apache.phoenix.thirdparty.com.google.common.annotations.VisibleForTesting;
+
+/**
+ * Metrics for one HAGroupStore group on a RegionServer.
+ * <p>
+ * Lifecycle gauges are best-effort and re-baselined when a client is
constructed. Counters and the
+ * histogram operation count are cumulative for the JVM lifetime; histogram
distributions cover the
+ * latest Metrics2 collection interval. Metrics are not transactionally
coordinated across client
+ * replacement or RegionServers and never feed HA decisions.
+ * <p>
+ * {@code haGroup} is the only source-specific tag. Monitoring systems must
supply cluster,
+ * environment, host, and RegionServer identity as external scrape-target
labels; role is mutable
+ * and is represented by the state gauges rather than a tag.
+ * <p>
+ * State gauge code mapping:
+ * <ul>
+ * <li>0 = UNKNOWN</li>
+ * <li>1 = ABORT_TO_ACTIVE_IN_SYNC</li>
+ * <li>2 = ABORT_TO_ACTIVE_NOT_IN_SYNC</li>
+ * <li>3 = ABORT_TO_STANDBY</li>
+ * <li>4 = ACTIVE_IN_SYNC</li>
+ * <li>5 = ACTIVE_NOT_IN_SYNC</li>
+ * <li>6 = ACTIVE_NOT_IN_SYNC_TO_STANDBY</li>
+ * <li>7 = ACTIVE_NOT_IN_SYNC_WITH_OFFLINE_PEER</li>
+ * <li>8 = ACTIVE_IN_SYNC_TO_STANDBY</li>
+ * <li>9 = ACTIVE_WITH_OFFLINE_PEER</li>
+ * <li>10 = DEGRADED_STANDBY</li>
+ * <li>11 = OFFLINE</li>
+ * <li>12 = STANDBY</li>
+ * <li>13 = STANDBY_TO_ACTIVE</li>
+ * </ul>
+ */
+public interface HAGroupStoreMetricsSource extends BaseSource {
+
+ String METRICS_NAME = "HAGroupStore";
+ String METRICS_CONTEXT = "phoenix";
+ String METRICS_DESCRIPTION = "Metrics for HAGroupStore operations";
+ // Server-oriented JMX identity consistent with sibling HBase/Phoenix
sources. Short-lived admin
+ // CLI invocations can create an incidental bean under the same context.
+ String METRICS_JMX_CONTEXT = "RegionServer,sub=" + METRICS_NAME;
+
+ String HA_GROUP_TAG_NAME = "haGroup";
+ String HA_GROUP_TAG_DESC = "HA group name";
+
+ String LOCAL_CACHE_HEALTH_STATUS = "haGroupStoreLocalCacheHealthStatus";
+ String LOCAL_CACHE_HEALTH_STATUS_DESC =
+ "Local HAGroupStore cache health status: 0 healthy, non-zero unhealthy";
+ String PEER_VISIBILITY_STATUS = "haGroupStorePeerVisibilityStatus";
+ String PEER_VISIBILITY_STATUS_DESC =
+ "Configured peer visibility status: 0 visible or unconfigured, non-zero
blind";
+ String DEGRADED_STANDBY_ACTIVE = "haGroupStoreDegradedStandbyActive";
+ String DEGRADED_STANDBY_ACTIVE_DESC =
+ "Local in-memory peer-blind fail-closed overlay: 0 inactive, 1 active on
raw STANDBY; "
+ + "never persisted to ZooKeeper and distinct from persisted
DEGRADED_STANDBY";
+ String CURRENT_LOCAL_STATE = "haGroupStoreCurrentLocalState";
+ String CURRENT_LOCAL_STATE_DESC = "Current raw local HA group state code";
+ String CURRENT_PEER_STATE = "haGroupStoreCurrentPeerState";
+ String CURRENT_PEER_STATE_DESC = "Current peer HA group state code";
+
+ String LOCAL_ZK_CONNECTION_LOST_COUNT =
"haGroupStoreLocalZkConnectionLostCount";
+ String LOCAL_ZK_CONNECTION_LOST_COUNT_DESC = "Number of local ZooKeeper
connection losses";
+ String PEER_BLIND_COUNT = "haGroupStorePeerBlindCount";
+ String PEER_BLIND_COUNT_DESC = "Number of transitions to peer-blind";
+ String DEGRADED_STANDBY_PRESENTED_COUNT =
"haGroupStoreDegradedStandbyPresentedCount";
+ String DEGRADED_STANDBY_PRESENTED_COUNT_DESC = "Number of degraded standby
overlay presentations";
+ String INVALID_TRANSITION_REJECTED_COUNT =
"haGroupStoreInvalidTransitionRejectedCount";
+ String INVALID_TRANSITION_REJECTED_COUNT_DESC =
+ "Number of invalid HA group state transitions rejected";
+ String SYSTEM_TABLE_SYNC_FAILED_COUNT =
"haGroupStoreSystemTableSyncFailedCount";
+ String SYSTEM_TABLE_SYNC_FAILED_COUNT_DESC =
+ "Number of failed SYSTEM.HA_GROUP synchronization attempts";
+ String NOTIFICATION_LISTENER_ERROR_COUNT =
"haGroupStoreNotificationListenerErrorCount";
+ String NOTIFICATION_LISTENER_ERROR_COUNT_DESC = "Number of HA group
notification listener errors";
+
+ // MutableTimeHistogram capitalizes its exported basename. The logical
registry key remains
+ // lowercase, while JMX/Metrics2 consumers use
SUBSCRIBER_NOTIFY_TIME_MS_EXPORTED_BASE plus the
+ // standard histogram suffixes.
+ String SUBSCRIBER_NOTIFY_TIME_MS = "haGroupStoreSubscriberNotifyTimeMs";
+ String SUBSCRIBER_NOTIFY_TIME_MS_EXPORTED_BASE =
"HaGroupStoreSubscriberNotifyTimeMs";
+ String SUBSCRIBER_NOTIFY_TIME_MS_DESC =
+ "Time spent synchronously notifying HA group subscribers in whole
milliseconds; "
+ + "durations below one millisecond are recorded as zero-valued samples
and still increment "
+ + "the operation count";
+
+ void setLocalCacheHealthy(boolean healthy);
+
+ void setPeerVisible(boolean visible);
+
+ void setDegradedStandbyActive(boolean active);
+
+ void setCurrentLocalState(HAGroupState state);
+
+ void setCurrentPeerState(HAGroupState state);
+
+ void incrementLocalZkConnectionLostCount();
+
+ void incrementPeerBlindCount();
+
+ void incrementDegradedStandbyPresentedCount();
+
+ void incrementInvalidTransitionRejectedCount();
+
+ void incrementSystemTableSyncFailedCount();
+
+ void incrementNotificationListenerErrorCount();
+
+ void updateSubscriberNotifyTime(long elapsedTimeNs);
+
+ @VisibleForTesting
+ HAGroupStoreMetricValues getCurrentMetricValues();
+}
diff --git
a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/metrics/HAGroupStoreMetricsSourceFactory.java
b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/metrics/HAGroupStoreMetricsSourceFactory.java
new file mode 100644
index 0000000000..2d61a42fd3
--- /dev/null
+++
b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/metrics/HAGroupStoreMetricsSourceFactory.java
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.jdbc.metrics;
+
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Factory for process-lifetime HAGroupStore metric sources.
+ * <p>
+ * Creates one source lazily per HA group and intentionally retains it for the
JVM lifetime so
+ * cumulative counters survive client replacement. HA-group names are expected
to be a small,
+ * stable set within a process.
+ */
+public final class HAGroupStoreMetricsSourceFactory {
+
+ private static final ConcurrentHashMap<String, HAGroupStoreMetricsSource>
SOURCES =
+ new ConcurrentHashMap<>();
+
+ private HAGroupStoreMetricsSourceFactory() {
+ }
+
+ public static HAGroupStoreMetricsSource getInstanceForHAGroup(String
haGroupName) {
+ return SOURCES.computeIfAbsent(haGroupName,
HAGroupStoreMetricsSourceImpl::new);
+ }
+}
diff --git
a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/metrics/HAGroupStoreMetricsSourceImpl.java
b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/metrics/HAGroupStoreMetricsSourceImpl.java
new file mode 100644
index 0000000000..f7d731fe76
--- /dev/null
+++
b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/metrics/HAGroupStoreMetricsSourceImpl.java
@@ -0,0 +1,171 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.jdbc.metrics;
+
+import java.util.concurrent.TimeUnit;
+import javax.management.ObjectName;
+import org.apache.hadoop.hbase.metrics.BaseSourceImpl;
+import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem;
+import org.apache.hadoop.metrics2.lib.Interns;
+import org.apache.hadoop.metrics2.lib.MutableFastCounter;
+import org.apache.hadoop.metrics2.lib.MutableGaugeLong;
+import org.apache.hadoop.metrics2.lib.MutableTimeHistogram;
+import org.apache.phoenix.jdbc.HAGroupStoreRecord.HAGroupState;
+
+import
org.apache.phoenix.thirdparty.com.google.common.annotations.VisibleForTesting;
+
+/** Hadoop Metrics2 implementation for one HAGroupStore group. */
+public class HAGroupStoreMetricsSourceImpl extends BaseSourceImpl
+ implements HAGroupStoreMetricsSource {
+
+ private final MutableGaugeLong localCacheHealthStatus;
+ private final MutableGaugeLong peerVisibilityStatus;
+ private final MutableGaugeLong degradedStandbyActive;
+ private final MutableGaugeLong currentLocalState;
+ private final MutableGaugeLong currentPeerState;
+
+ private final MutableFastCounter localZkConnectionLostCount;
+ private final MutableFastCounter peerBlindCount;
+ private final MutableFastCounter degradedStandbyPresentedCount;
+ private final MutableFastCounter invalidTransitionRejectedCount;
+ private final MutableFastCounter systemTableSyncFailedCount;
+ private final MutableFastCounter notificationListenerErrorCount;
+
+ private final MutableTimeHistogram subscriberNotifyTimeMs;
+
+ public HAGroupStoreMetricsSourceImpl(String haGroupName) {
+ this(METRICS_NAME, METRICS_DESCRIPTION, METRICS_CONTEXT,
METRICS_JMX_CONTEXT, haGroupName);
+ }
+
+ HAGroupStoreMetricsSourceImpl(String metricsName, String metricsDescription,
+ String metricsContext, String metricsJmxContext, String haGroupName) {
+ super(metricsName, metricsDescription, metricsContext,
+ metricsJmxContext + ",haGroup=" + ObjectName.quote(haGroupName));
+ getMetricsRegistry().tag(Interns.info(HA_GROUP_TAG_NAME,
HA_GROUP_TAG_DESC), haGroupName);
+
+ localCacheHealthStatus =
getMetricsRegistry().newGauge(LOCAL_CACHE_HEALTH_STATUS,
+ LOCAL_CACHE_HEALTH_STATUS_DESC, 1L);
+ peerVisibilityStatus =
getMetricsRegistry().newGauge(PEER_VISIBILITY_STATUS,
+ PEER_VISIBILITY_STATUS_DESC, 1L);
+ degradedStandbyActive =
+ getMetricsRegistry().newGauge(DEGRADED_STANDBY_ACTIVE,
DEGRADED_STANDBY_ACTIVE_DESC, 0L);
+ currentLocalState = getMetricsRegistry().newGauge(CURRENT_LOCAL_STATE,
CURRENT_LOCAL_STATE_DESC,
+ (long) HAGroupState.UNKNOWN.getMetricCode());
+ currentPeerState = getMetricsRegistry().newGauge(CURRENT_PEER_STATE,
CURRENT_PEER_STATE_DESC,
+ (long) HAGroupState.UNKNOWN.getMetricCode());
+
+ localZkConnectionLostCount =
getMetricsRegistry().newCounter(LOCAL_ZK_CONNECTION_LOST_COUNT,
+ LOCAL_ZK_CONNECTION_LOST_COUNT_DESC, 0L);
+ peerBlindCount = getMetricsRegistry().newCounter(PEER_BLIND_COUNT,
PEER_BLIND_COUNT_DESC, 0L);
+ degradedStandbyPresentedCount = getMetricsRegistry()
+ .newCounter(DEGRADED_STANDBY_PRESENTED_COUNT,
DEGRADED_STANDBY_PRESENTED_COUNT_DESC, 0L);
+ invalidTransitionRejectedCount = getMetricsRegistry()
+ .newCounter(INVALID_TRANSITION_REJECTED_COUNT,
INVALID_TRANSITION_REJECTED_COUNT_DESC, 0L);
+ systemTableSyncFailedCount =
getMetricsRegistry().newCounter(SYSTEM_TABLE_SYNC_FAILED_COUNT,
+ SYSTEM_TABLE_SYNC_FAILED_COUNT_DESC, 0L);
+ notificationListenerErrorCount = getMetricsRegistry()
+ .newCounter(NOTIFICATION_LISTENER_ERROR_COUNT,
NOTIFICATION_LISTENER_ERROR_COUNT_DESC, 0L);
+
+ subscriberNotifyTimeMs =
getMetricsRegistry().newTimeHistogram(SUBSCRIBER_NOTIFY_TIME_MS,
+ SUBSCRIBER_NOTIFY_TIME_MS_DESC);
+ }
+
+ @Override
+ public void setLocalCacheHealthy(boolean value) {
+ localCacheHealthStatus.set(value ? 0L : 1L);
+ }
+
+ @Override
+ public void setPeerVisible(boolean value) {
+ peerVisibilityStatus.set(value ? 0L : 1L);
+ }
+
+ @Override
+ public void setDegradedStandbyActive(boolean value) {
+ degradedStandbyActive.set(value ? 1L : 0L);
+ }
+
+ @Override
+ public void setCurrentLocalState(HAGroupState state) {
+ HAGroupState currentState = state != null ? state : HAGroupState.UNKNOWN;
+ currentLocalState.set(currentState.getMetricCode());
+ }
+
+ @Override
+ public void setCurrentPeerState(HAGroupState state) {
+ HAGroupState currentState = state != null ? state : HAGroupState.UNKNOWN;
+ currentPeerState.set(currentState.getMetricCode());
+ }
+
+ @Override
+ public void incrementLocalZkConnectionLostCount() {
+ localZkConnectionLostCount.incr();
+ }
+
+ @Override
+ public void incrementPeerBlindCount() {
+ peerBlindCount.incr();
+ }
+
+ @Override
+ public void incrementDegradedStandbyPresentedCount() {
+ degradedStandbyPresentedCount.incr();
+ }
+
+ @Override
+ public void incrementInvalidTransitionRejectedCount() {
+ invalidTransitionRejectedCount.incr();
+ }
+
+ @Override
+ public void incrementSystemTableSyncFailedCount() {
+ systemTableSyncFailedCount.incr();
+ }
+
+ @Override
+ public void incrementNotificationListenerErrorCount() {
+ notificationListenerErrorCount.incr();
+ }
+
+ @Override
+ public void updateSubscriberNotifyTime(long elapsedTimeNs) {
+ subscriberNotifyTimeMs.add(TimeUnit.NANOSECONDS.toMillis(elapsedTimeNs));
+ }
+
+ @Override
+ public HAGroupStoreMetricValues getCurrentMetricValues() {
+ return HAGroupStoreMetricValues.builder()
+ .setLocalCacheHealthStatus(localCacheHealthStatus.value())
+ .setPeerVisibilityStatus(peerVisibilityStatus.value())
+ .setDegradedStandbyActive(degradedStandbyActive.value())
+
.setCurrentLocalState(currentLocalState.value()).setCurrentPeerState(currentPeerState.value())
+ .setLocalZkConnectionLostCount(localZkConnectionLostCount.value())
+ .setPeerBlindCount(peerBlindCount.value())
+ .setDegradedStandbyPresentedCount(degradedStandbyPresentedCount.value())
+
.setInvalidTransitionRejectedCount(invalidTransitionRejectedCount.value())
+ .setSystemTableSyncFailedCount(systemTableSyncFailedCount.value())
+
.setNotificationListenerErrorCount(notificationListenerErrorCount.value())
+ .setSubscriberNotifyTimeCount(subscriberNotifyTimeMs.getCount())
+ .setSubscriberNotifyTimeMaxMs(subscriberNotifyTimeMs.getMax()).build();
+ }
+
+ @VisibleForTesting
+ public void unregisterForTesting() {
+ DefaultMetricsSystem.instance().unregisterSource(metricsJmxContext);
+ }
+}
diff --git
a/phoenix-core/src/it/java/org/apache/phoenix/jdbc/HAGroupStoreMetricsIT.java
b/phoenix-core/src/it/java/org/apache/phoenix/jdbc/HAGroupStoreMetricsIT.java
new file mode 100644
index 0000000000..80e1f7781a
--- /dev/null
+++
b/phoenix-core/src/it/java/org/apache/phoenix/jdbc/HAGroupStoreMetricsIT.java
@@ -0,0 +1,358 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.jdbc;
+
+import static
org.apache.phoenix.jdbc.HAGroupStoreClient.ZK_CONSISTENT_HA_GROUP_RECORD_NAMESPACE;
+import static org.apache.phoenix.jdbc.PhoenixHAAdmin.getLocalZkUrl;
+import static org.apache.phoenix.jdbc.PhoenixHAAdmin.toPath;
+import static
org.apache.phoenix.query.QueryServices.HA_GROUP_STORE_SYNC_INTERVAL_SECONDS;
+import static
org.apache.phoenix.replication.reader.ReplicationLogReplayService.PHOENIX_REPLICATION_REPLAY_ENABLED;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Predicate;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.client.Admin;
+import org.apache.phoenix.end2end.NeedsOwnMiniClusterTest;
+import org.apache.phoenix.exception.InvalidClusterRoleTransitionException;
+import org.apache.phoenix.jdbc.metrics.HAGroupStoreMetricValues;
+import org.apache.phoenix.jdbc.metrics.HAGroupStoreMetricsSourceFactory;
+import org.apache.phoenix.util.HAGroupStoreTestUtil;
+import org.apache.zookeeper.data.Stat;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.rules.TestName;
+
+@Category(NeedsOwnMiniClusterTest.class)
+public class HAGroupStoreMetricsIT extends HABaseIT {
+
+ private static final long EVENT_TIMEOUT_MS = 60000L;
+
+ @Rule
+ public final TestName testName = new TestName();
+
+ private PhoenixHAAdmin localAdmin;
+ private PhoenixHAAdmin peerAdmin;
+ private String zkUrl;
+ private String peerZkUrl;
+ private String masterUrl;
+ private String peerMasterUrl;
+ private HAGroupStoreClient clientToClose;
+
+ @BeforeClass
+ public static synchronized void setUpClass() throws Exception {
+ conf1.setBoolean(PHOENIX_REPLICATION_REPLAY_ENABLED, false);
+ conf2.setBoolean(PHOENIX_REPLICATION_REPLAY_ENABLED, false);
+ CLUSTERS.start();
+ }
+
+ @AfterClass
+ public static synchronized void tearDownClass() throws Exception {
+ CLUSTERS.close();
+ }
+
+ @Before
+ public void setUp() throws Exception {
+ zkUrl = getLocalZkUrl(CLUSTERS.getHBaseCluster1().getConfiguration());
+ peerZkUrl = getLocalZkUrl(CLUSTERS.getHBaseCluster2().getConfiguration());
+ masterUrl = CLUSTERS.getMasterAddress1();
+ peerMasterUrl = CLUSTERS.getMasterAddress2();
+ localAdmin = new
PhoenixHAAdmin(CLUSTERS.getHBaseCluster1().getConfiguration(),
+ ZK_CONSISTENT_HA_GROUP_RECORD_NAMESPACE);
+ peerAdmin = new
PhoenixHAAdmin(CLUSTERS.getHBaseCluster2().getConfiguration(),
+ ZK_CONSISTENT_HA_GROUP_RECORD_NAMESPACE);
+ deleteZNodes();
+ HAGroupStoreTestUtil.deleteHAGroupRecordInSystemTable(group(), zkUrl);
+ HAGroupStoreTestUtil.upsertHAGroupRecordInSystemTable(group(), zkUrl,
peerZkUrl, masterUrl,
+ peerMasterUrl, ClusterRoleRecord.ClusterRole.ACTIVE,
ClusterRoleRecord.ClusterRole.STANDBY,
+ null, CLUSTERS.getHdfsUrl1(), CLUSTERS.getHdfsUrl2());
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ if (clientToClose != null) {
+ clientToClose.close();
+ }
+ deleteZNodes();
+ HAGroupStoreTestUtil.deleteHAGroupRecordInSystemTable(group(), zkUrl);
+ localAdmin.close();
+ peerAdmin.close();
+ }
+
+ @Test
+ public void testInitialAndLocalStateMetrics() throws Exception {
+ writeLocal(HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC);
+ writePeer(HAGroupStoreRecord.HAGroupState.STANDBY);
+ HAGroupStoreClient client = localClient();
+
+ awaitMetrics(values -> values.getLocalCacheHealthStatus() == 0
+ && values.getCurrentLocalState()
+ == HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC.getMetricCode()
+ && values.getCurrentPeerState() ==
HAGroupStoreRecord.HAGroupState.STANDBY.getMetricCode());
+
+ assertEquals(0L,
+
client.setHAGroupStatusIfNeeded(HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC_TO_STANDBY));
+ awaitMetrics(values -> values.getCurrentLocalState()
+ ==
HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC_TO_STANDBY.getMetricCode());
+
+ assertEquals(0L,
client.setHAGroupStatusIfNeeded(HAGroupStoreRecord.HAGroupState.STANDBY));
+ awaitMetrics(values -> values.getCurrentLocalState()
+ == HAGroupStoreRecord.HAGroupState.STANDBY.getMetricCode());
+ }
+
+ @Test
+ public void testLocalZkHealthMetrics() throws Exception {
+ writeLocal(HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC);
+ HAGroupStoreClient client = localClient();
+ awaitMetrics(values -> values.getLocalCacheHealthStatus() == 0);
+ HAGroupStoreMetricValues before = metrics();
+ int port = Integer.parseInt(
+
CLUSTERS.getHBaseCluster1().getConfiguration().get("hbase.zookeeper.property.clientPort"));
+
+ CLUSTERS.getHBaseCluster1().shutdownMiniZKCluster();
+ try {
+ awaitMetrics(values -> values.getLocalCacheHealthStatus() == 1
+ && values.getLocalZkConnectionLostCount() ==
before.getLocalZkConnectionLostCount() + 1);
+ assertThrows(IOException.class, client::getHAGroupStoreRecord);
+ } finally {
+ CLUSTERS.getHBaseCluster1().startMiniZKCluster(1, port);
+ }
+ awaitMetrics(values -> values.getLocalCacheHealthStatus() == 0);
+ }
+
+ @Test
+ public void testLateInitializationFailureResetsGauges() throws Exception {
+ writeLocal(HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC);
+ Configuration invalidConf =
+ new Configuration(CLUSTERS.getHBaseCluster1().getConfiguration());
+ invalidConf.setLong(HA_GROUP_STORE_SYNC_INTERVAL_SECONDS, 0L);
+
+ assertNull(HAGroupStoreClient.getInstanceForZkUrl(invalidConf, group(),
zkUrl));
+
+ HAGroupStoreMetricValues values = metrics();
+ assertEquals(1L, values.getLocalCacheHealthStatus());
+ assertEquals(1L, values.getPeerVisibilityStatus());
+ assertEquals(0L, values.getDegradedStandbyActive());
+ assertEquals(HAGroupStoreRecord.HAGroupState.UNKNOWN.getMetricCode(),
+ values.getCurrentLocalState());
+ assertEquals(HAGroupStoreRecord.HAGroupState.UNKNOWN.getMetricCode(),
+ values.getCurrentPeerState());
+ }
+
+ @Test
+ public void testPeerVisibilityAndDegradedMetrics() throws Exception {
+ writeLocal(HAGroupStoreRecord.HAGroupState.STANDBY);
+ writePeer(HAGroupStoreRecord.HAGroupState.STANDBY);
+ localClient();
+ awaitMetrics(values -> values.getPeerVisibilityStatus() == 0
+ && values.getCurrentPeerState() ==
HAGroupStoreRecord.HAGroupState.STANDBY.getMetricCode());
+ HAGroupStoreMetricValues before = metrics();
+ int port = Integer.parseInt(
+
CLUSTERS.getHBaseCluster2().getConfiguration().get("hbase.zookeeper.property.clientPort"));
+
+ CLUSTERS.getHBaseCluster2().shutdownMiniZKCluster();
+ try {
+ awaitMetrics(values -> values.getPeerVisibilityStatus() == 1
+ && values.getCurrentPeerState() ==
HAGroupStoreRecord.HAGroupState.UNKNOWN.getMetricCode()
+ && values.getDegradedStandbyActive() == 1
+ && values.getPeerBlindCount() == before.getPeerBlindCount() + 1
+ && values.getDegradedStandbyPresentedCount()
+ == before.getDegradedStandbyPresentedCount() + 1);
+ } finally {
+ CLUSTERS.getHBaseCluster2().startMiniZKCluster(1, port);
+ }
+ awaitMetrics(values -> values.getPeerVisibilityStatus() == 0
+ && values.getCurrentPeerState() ==
HAGroupStoreRecord.HAGroupState.STANDBY.getMetricCode()
+ && values.getDegradedStandbyActive() == 0);
+ }
+
+ @Test
+ public void testPeerReconfigurationMetrics() throws Exception {
+ writeLocal(HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC);
+ writePeer(HAGroupStoreRecord.HAGroupState.STANDBY);
+ localClient();
+ awaitMetrics(values -> values.getCurrentPeerState()
+ == HAGroupStoreRecord.HAGroupState.STANDBY.getMetricCode());
+
+ writeLocal(HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC, null);
+ awaitMetrics(values -> values.getPeerVisibilityStatus() == 0
+ && values.getCurrentPeerState() ==
HAGroupStoreRecord.HAGroupState.UNKNOWN.getMetricCode());
+
+ writeLocal(HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC);
+ awaitMetrics(values -> values.getCurrentPeerState()
+ == HAGroupStoreRecord.HAGroupState.STANDBY.getMetricCode());
+
+ writeLocal(HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC, "invalidURL");
+ awaitMetrics(values -> values.getPeerVisibilityStatus() == 1
+ && values.getCurrentPeerState() ==
HAGroupStoreRecord.HAGroupState.UNKNOWN.getMetricCode());
+
+ writeLocal(HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC, null);
+ awaitMetrics(values -> values.getPeerVisibilityStatus() == 0
+ && values.getCurrentPeerState() ==
HAGroupStoreRecord.HAGroupState.UNKNOWN.getMetricCode());
+ }
+
+ @Test
+ public void testInvalidTransitionMetric() throws Exception {
+ writeLocal(HAGroupStoreRecord.HAGroupState.ACTIVE_NOT_IN_SYNC);
+ HAGroupStoreClient client = localClient();
+ HAGroupStoreMetricValues before = metrics();
+
+ assertThrows(InvalidClusterRoleTransitionException.class,
+ () ->
client.setHAGroupStatusIfNeeded(HAGroupStoreRecord.HAGroupState.STANDBY));
+ assertEquals(before.getInvalidTransitionRejectedCount() + 1,
+ metrics().getInvalidTransitionRejectedCount());
+ }
+
+ @Test
+ public void testSystemTableSyncFailureMetric() throws Exception {
+ writeLocal(HAGroupStoreRecord.HAGroupState.STANDBY);
+ HAGroupStoreClient client = localClient();
+ awaitMetrics(values -> values.getCurrentLocalState()
+ == HAGroupStoreRecord.HAGroupState.STANDBY.getMetricCode());
+ client.shutdownSyncExecutor();
+ HAGroupStoreMetricValues before = metrics();
+ Admin admin = CLUSTERS.getHBaseCluster1().getAdmin();
+
+
admin.disableTable(PhoenixDatabaseMetaData.SYSTEM_HA_GROUP_HBASE_TABLE_NAME);
+ try {
+ assertEquals(0L,
+
client.setHAGroupStatusIfNeeded(HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE));
+ assertEquals(before.getSystemTableSyncFailedCount() + 1,
+ metrics().getSystemTableSyncFailedCount());
+ Pair<HAGroupStoreRecord, Stat> updated =
localAdmin.getHAGroupStoreRecordInZooKeeper(group());
+ assertEquals(HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE,
+ updated.getLeft().getHAGroupState());
+ } finally {
+
admin.enableTable(PhoenixDatabaseMetaData.SYSTEM_HA_GROUP_HBASE_TABLE_NAME);
+ }
+ }
+
+ @Test
+ public void testPeriodicSystemTableSyncFailureBeforeWriteMetric() throws
Exception {
+ writeLocal(HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC);
+ HAGroupStoreClient client = localClient();
+ awaitMetrics(values -> values.getLocalCacheHealthStatus() == 0);
+ client.shutdownSyncExecutor();
+
+ HAGroupStoreRecord invalidPolicyRecord =
+ new HAGroupStoreRecord(HAGroupStoreRecord.DEFAULT_PROTOCOL_VERSION,
group(),
+ HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC, 0L, "INVALID_POLICY",
peerZkUrl, masterUrl,
+ peerMasterUrl, CLUSTERS.getHdfsUrl1(), CLUSTERS.getHdfsUrl2(), 0L);
+ createOrUpdate(localAdmin, invalidPolicyRecord);
+ long before = metrics().getSystemTableSyncFailedCount();
+
+ client.syncZKToSystemTable();
+
+ assertEquals(before + 1, metrics().getSystemTableSyncFailedCount());
+ }
+
+ @Test
+ public void testListenerMetrics() throws Exception {
+ writeLocal(HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC);
+ HAGroupStoreClient client = localClient();
+
client.subscribeToTargetState(HAGroupStoreRecord.HAGroupState.ACTIVE_NOT_IN_SYNC,
+ ClusterType.LOCAL, (group, from, to, mtime, cluster, lastSync) -> {
+ throw new IllegalStateException("expected test failure");
+ });
+ AtomicInteger successfulListeners = new AtomicInteger();
+
client.subscribeToTargetState(HAGroupStoreRecord.HAGroupState.ACTIVE_NOT_IN_SYNC,
+ ClusterType.LOCAL,
+ (group, from, to, mtime, cluster, lastSync) ->
successfulListeners.incrementAndGet());
+ HAGroupStoreMetricValues before = metrics();
+
+ writeLocal(HAGroupStoreRecord.HAGroupState.ACTIVE_NOT_IN_SYNC);
+ awaitMetrics(values -> values.getNotificationListenerErrorCount()
+ == before.getNotificationListenerErrorCount() + 1
+ && values.getSubscriberNotifyTimeCount() ==
before.getSubscriberNotifyTimeCount() + 1
+ && successfulListeners.get() == 1);
+ }
+
+ private String group() {
+ return testName.getMethodName();
+ }
+
+ private HAGroupStoreClient localClient() {
+ HAGroupStoreClient client = HAGroupStoreClient
+ .getInstanceForZkUrl(CLUSTERS.getHBaseCluster1().getConfiguration(),
group(), zkUrl);
+ if (client != null) {
+ clientToClose = client;
+ }
+ return client;
+ }
+
+ private HAGroupStoreMetricValues metrics() {
+ return
HAGroupStoreMetricsSourceFactory.getInstanceForHAGroup(group()).getCurrentMetricValues();
+ }
+
+ private void awaitMetrics(Predicate<HAGroupStoreMetricValues> condition)
throws Exception {
+ long deadline = System.currentTimeMillis() + EVENT_TIMEOUT_MS;
+ while (!condition.test(metrics()) && System.currentTimeMillis() <
deadline) {
+ Thread.sleep(100L);
+ }
+ assertTrue("Metric condition not met for " + group(),
condition.test(metrics()));
+ }
+
+ private HAGroupStoreRecord localRecord(HAGroupStoreRecord.HAGroupState
state, String peerUrl) {
+ return new HAGroupStoreRecord(HAGroupStoreRecord.DEFAULT_PROTOCOL_VERSION,
group(), state, 0L,
+ HighAvailabilityPolicy.FAILOVER.toString(), peerUrl, masterUrl,
peerMasterUrl,
+ CLUSTERS.getHdfsUrl1(), CLUSTERS.getHdfsUrl2(), 0L);
+ }
+
+ private HAGroupStoreRecord peerRecord(HAGroupStoreRecord.HAGroupState state)
{
+ return new HAGroupStoreRecord(HAGroupStoreRecord.DEFAULT_PROTOCOL_VERSION,
group(), state, 0L,
+ HighAvailabilityPolicy.FAILOVER.toString(), zkUrl, peerMasterUrl,
masterUrl,
+ CLUSTERS.getHdfsUrl2(), CLUSTERS.getHdfsUrl1(), 0L);
+ }
+
+ private void writeLocal(HAGroupStoreRecord.HAGroupState state) throws
Exception {
+ writeLocal(state, peerZkUrl);
+ }
+
+ private void writeLocal(HAGroupStoreRecord.HAGroupState state, String
peerUrl) throws Exception {
+ createOrUpdate(localAdmin, localRecord(state, peerUrl));
+ }
+
+ private void writePeer(HAGroupStoreRecord.HAGroupState state) throws
Exception {
+ createOrUpdate(peerAdmin, peerRecord(state));
+ }
+
+ private void createOrUpdate(PhoenixHAAdmin admin, HAGroupStoreRecord record)
throws Exception {
+ String path = toPath(group());
+ if (admin.getCurator().checkExists().forPath(path) == null) {
+ admin.createHAGroupStoreRecordInZooKeeper(record);
+ return;
+ }
+ Pair<HAGroupStoreRecord, Stat> current =
admin.getHAGroupStoreRecordInZooKeeper(group());
+ admin.updateHAGroupStoreRecordInZooKeeper(group(), record,
current.getRight().getVersion());
+ }
+
+ private void deleteZNodes() throws Exception {
+ localAdmin.getCurator().delete().quietly().forPath(toPath(group()));
+ peerAdmin.getCurator().delete().quietly().forPath(toPath(group()));
+ }
+}
diff --git
a/phoenix-core/src/test/java/org/apache/phoenix/jdbc/HAGroupStoreRecordTest.java
b/phoenix-core/src/test/java/org/apache/phoenix/jdbc/HAGroupStoreRecordTest.java
index 3fb77d4bd4..ee451d1207 100644
---
a/phoenix-core/src/test/java/org/apache/phoenix/jdbc/HAGroupStoreRecordTest.java
+++
b/phoenix-core/src/test/java/org/apache/phoenix/jdbc/HAGroupStoreRecordTest.java
@@ -252,6 +252,25 @@ public class HAGroupStoreRecordTest {
}
// Tests for HAGroupState enum
+ @Test
+ public void testHAGroupStateMetricContract() {
+ assertEquals(1,
HAGroupStoreRecord.HAGroupState.ABORT_TO_ACTIVE_IN_SYNC.getMetricCode());
+ assertEquals(2,
HAGroupStoreRecord.HAGroupState.ABORT_TO_ACTIVE_NOT_IN_SYNC.getMetricCode());
+ assertEquals(3,
HAGroupStoreRecord.HAGroupState.ABORT_TO_STANDBY.getMetricCode());
+ assertEquals(4,
HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC.getMetricCode());
+ assertEquals(5,
HAGroupStoreRecord.HAGroupState.ACTIVE_NOT_IN_SYNC.getMetricCode());
+ assertEquals(6,
HAGroupStoreRecord.HAGroupState.ACTIVE_NOT_IN_SYNC_TO_STANDBY.getMetricCode());
+ assertEquals(7,
+
HAGroupStoreRecord.HAGroupState.ACTIVE_NOT_IN_SYNC_WITH_OFFLINE_PEER.getMetricCode());
+ assertEquals(8,
HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC_TO_STANDBY.getMetricCode());
+ assertEquals(9,
HAGroupStoreRecord.HAGroupState.ACTIVE_WITH_OFFLINE_PEER.getMetricCode());
+ assertEquals(10,
HAGroupStoreRecord.HAGroupState.DEGRADED_STANDBY.getMetricCode());
+ assertEquals(11, HAGroupStoreRecord.HAGroupState.OFFLINE.getMetricCode());
+ assertEquals(12, HAGroupStoreRecord.HAGroupState.STANDBY.getMetricCode());
+ assertEquals(13,
HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE.getMetricCode());
+ assertEquals(0, HAGroupStoreRecord.HAGroupState.UNKNOWN.getMetricCode());
+ }
+
@Test
public void testHAGroupStateGetClusterRole() {
// Test that each HAGroupState maps to the correct ClusterRole
diff --git
a/phoenix-core/src/test/java/org/apache/phoenix/jdbc/metrics/HAGroupStoreMetricsSourceImplTest.java
b/phoenix-core/src/test/java/org/apache/phoenix/jdbc/metrics/HAGroupStoreMetricsSourceImplTest.java
new file mode 100644
index 0000000000..9eabdd97b9
--- /dev/null
+++
b/phoenix-core/src/test/java/org/apache/phoenix/jdbc/metrics/HAGroupStoreMetricsSourceImplTest.java
@@ -0,0 +1,197 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.jdbc.metrics;
+
+import static
org.apache.phoenix.jdbc.metrics.HAGroupStoreMetricsSource.CURRENT_LOCAL_STATE;
+import static
org.apache.phoenix.jdbc.metrics.HAGroupStoreMetricsSource.CURRENT_PEER_STATE;
+import static
org.apache.phoenix.jdbc.metrics.HAGroupStoreMetricsSource.DEGRADED_STANDBY_ACTIVE;
+import static
org.apache.phoenix.jdbc.metrics.HAGroupStoreMetricsSource.DEGRADED_STANDBY_PRESENTED_COUNT;
+import static
org.apache.phoenix.jdbc.metrics.HAGroupStoreMetricsSource.HA_GROUP_TAG_NAME;
+import static
org.apache.phoenix.jdbc.metrics.HAGroupStoreMetricsSource.INVALID_TRANSITION_REJECTED_COUNT;
+import static
org.apache.phoenix.jdbc.metrics.HAGroupStoreMetricsSource.LOCAL_CACHE_HEALTH_STATUS;
+import static
org.apache.phoenix.jdbc.metrics.HAGroupStoreMetricsSource.LOCAL_ZK_CONNECTION_LOST_COUNT;
+import static
org.apache.phoenix.jdbc.metrics.HAGroupStoreMetricsSource.NOTIFICATION_LISTENER_ERROR_COUNT;
+import static
org.apache.phoenix.jdbc.metrics.HAGroupStoreMetricsSource.PEER_BLIND_COUNT;
+import static
org.apache.phoenix.jdbc.metrics.HAGroupStoreMetricsSource.PEER_VISIBILITY_STATUS;
+import static
org.apache.phoenix.jdbc.metrics.HAGroupStoreMetricsSource.SUBSCRIBER_NOTIFY_TIME_MS;
+import static
org.apache.phoenix.jdbc.metrics.HAGroupStoreMetricsSource.SUBSCRIBER_NOTIFY_TIME_MS_EXPORTED_BASE;
+import static
org.apache.phoenix.jdbc.metrics.HAGroupStoreMetricsSource.SYSTEM_TABLE_SYNC_FAILED_COUNT;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertNotSame;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import javax.management.ObjectName;
+import org.apache.hadoop.metrics2.AbstractMetric;
+import org.apache.hadoop.metrics2.impl.MetricsCollectorImpl;
+import org.apache.hadoop.metrics2.impl.MetricsRecordImpl;
+import org.apache.hadoop.metrics2.lib.MutableFastCounter;
+import org.apache.hadoop.metrics2.lib.MutableGaugeLong;
+import org.apache.hadoop.metrics2.lib.MutableMetric;
+import org.apache.hadoop.metrics2.lib.MutableTimeHistogram;
+import org.apache.phoenix.jdbc.HAGroupStoreRecord.HAGroupState;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+public class HAGroupStoreMetricsSourceImplTest {
+
+ private HAGroupStoreMetricsSourceImpl source;
+
+ @Before
+ public void setUp() {
+ source = new HAGroupStoreMetricsSourceImpl("metrics-test-" +
System.nanoTime());
+ }
+
+ @After
+ public void tearDown() {
+ source.unregisterForTesting();
+ }
+
+ @Test
+ public void testMetricNamesAreUniqueAndPrefixed() {
+ Set<String> metricNames = new
HashSet<>(Arrays.asList(LOCAL_CACHE_HEALTH_STATUS,
+ PEER_VISIBILITY_STATUS, DEGRADED_STANDBY_ACTIVE, CURRENT_LOCAL_STATE,
CURRENT_PEER_STATE,
+ LOCAL_ZK_CONNECTION_LOST_COUNT, PEER_BLIND_COUNT,
DEGRADED_STANDBY_PRESENTED_COUNT,
+ INVALID_TRANSITION_REJECTED_COUNT, SYSTEM_TABLE_SYNC_FAILED_COUNT,
+ NOTIFICATION_LISTENER_ERROR_COUNT, SUBSCRIBER_NOTIFY_TIME_MS));
+
+ assertEquals(12, metricNames.size());
+ for (String metricName : metricNames) {
+ assertTrue(metricName, metricName.startsWith("haGroupStore"));
+ }
+ }
+
+ @Test
+ public void testMetricRegistryContract() {
+ Map<String, MutableMetric> metrics =
source.getMetricsRegistry().getMetricsMap();
+ assertEquals(12, metrics.size());
+
+ for (String gauge : Arrays.asList(LOCAL_CACHE_HEALTH_STATUS,
PEER_VISIBILITY_STATUS,
+ DEGRADED_STANDBY_ACTIVE, CURRENT_LOCAL_STATE, CURRENT_PEER_STATE)) {
+ assertTrue(gauge, metrics.get(gauge) instanceof MutableGaugeLong);
+ }
+ for (String counter : Arrays.asList(LOCAL_ZK_CONNECTION_LOST_COUNT,
PEER_BLIND_COUNT,
+ DEGRADED_STANDBY_PRESENTED_COUNT, INVALID_TRANSITION_REJECTED_COUNT,
+ SYSTEM_TABLE_SYNC_FAILED_COUNT, NOTIFICATION_LISTENER_ERROR_COUNT)) {
+ assertTrue(counter, metrics.get(counter) instanceof MutableFastCounter);
+ }
+ assertTrue(metrics.get(SUBSCRIBER_NOTIFY_TIME_MS) instanceof
MutableTimeHistogram);
+ }
+
+ @Test
+ public void testMetricsSnapshot() {
+ source.setLocalCacheHealthy(true);
+ source.setPeerVisible(true);
+ source.setDegradedStandbyActive(true);
+ source.setCurrentLocalState(HAGroupState.STANDBY_TO_ACTIVE);
+ source.setCurrentPeerState(HAGroupState.ACTIVE_IN_SYNC);
+ source.incrementLocalZkConnectionLostCount();
+ source.incrementPeerBlindCount();
+ source.incrementDegradedStandbyPresentedCount();
+ source.incrementInvalidTransitionRejectedCount();
+ source.incrementSystemTableSyncFailedCount();
+ source.incrementNotificationListenerErrorCount();
+ source.updateSubscriberNotifyTime(5_000_000L);
+
+ HAGroupStoreMetricValues values = source.getCurrentMetricValues();
+ assertEquals(0L, values.getLocalCacheHealthStatus());
+ assertEquals(0L, values.getPeerVisibilityStatus());
+ assertEquals(1L, values.getDegradedStandbyActive());
+ assertEquals(HAGroupState.STANDBY_TO_ACTIVE.getMetricCode(),
values.getCurrentLocalState());
+ assertEquals(HAGroupState.ACTIVE_IN_SYNC.getMetricCode(),
values.getCurrentPeerState());
+ assertEquals(1L, values.getLocalZkConnectionLostCount());
+ assertEquals(1L, values.getPeerBlindCount());
+ assertEquals(1L, values.getDegradedStandbyPresentedCount());
+ assertEquals(1L, values.getInvalidTransitionRejectedCount());
+ assertEquals(1L, values.getSystemTableSyncFailedCount());
+ assertEquals(1L, values.getNotificationListenerErrorCount());
+ assertEquals(1L, values.getSubscriberNotifyTimeCount());
+ assertEquals(5L, values.getSubscriberNotifyTimeMaxMs());
+ }
+
+ @Test
+ public void testSubMillisecondNotifyTimeRoundsDownToZero() {
+ source.updateSubscriberNotifyTime(999_999L);
+
+ HAGroupStoreMetricValues values = source.getCurrentMetricValues();
+ assertEquals(1L, values.getSubscriberNotifyTimeCount());
+ assertEquals(0L, values.getSubscriberNotifyTimeMaxMs());
+ }
+
+ @Test
+ public void testExportedHistogramNames() {
+ source.updateSubscriberNotifyTime(5_000_000L);
+ MetricsCollectorImpl collector = new MetricsCollectorImpl();
+ source.getMetrics(collector, true);
+
+ Set<String> exportedNames = new HashSet<>();
+ for (MetricsRecordImpl record : collector.getRecords()) {
+ for (AbstractMetric metric : record.metrics()) {
+ exportedNames.add(metric.name());
+ }
+ }
+
+ assertTrue(exportedNames.contains(SUBSCRIBER_NOTIFY_TIME_MS_EXPORTED_BASE
+ "_num_ops"));
+ assertTrue(exportedNames.contains(SUBSCRIBER_NOTIFY_TIME_MS_EXPORTED_BASE
+ "_max"));
+ assertTrue(
+ exportedNames.contains(SUBSCRIBER_NOTIFY_TIME_MS_EXPORTED_BASE +
"_99th_percentile"));
+ assertFalse(exportedNames.contains(SUBSCRIBER_NOTIFY_TIME_MS +
"_num_ops"));
+ }
+
+ @Test
+ public void testFactoryCachesSourceByGroup() {
+ String group = "factory-test-" + System.nanoTime();
+ assertSame(HAGroupStoreMetricsSourceFactory.getInstanceForHAGroup(group),
+ HAGroupStoreMetricsSourceFactory.getInstanceForHAGroup(group));
+ }
+
+ @Test
+ public void testFactoryIsolatesGroups() {
+ String prefix = "factory-isolation-" + System.nanoTime();
+ HAGroupStoreMetricsSource first =
+ HAGroupStoreMetricsSourceFactory.getInstanceForHAGroup(prefix + "-1");
+ HAGroupStoreMetricsSource second =
+ HAGroupStoreMetricsSourceFactory.getInstanceForHAGroup(prefix + "-2");
+
+ assertNotSame(first, second);
+ assertNotEquals(first.getMetricsJmxContext(),
second.getMetricsJmxContext());
+ first.incrementInvalidTransitionRejectedCount();
+ assertEquals(1L,
first.getCurrentMetricValues().getInvalidTransitionRejectedCount());
+ assertEquals(0L,
second.getCurrentMetricValues().getInvalidTransitionRejectedCount());
+ }
+
+ @Test
+ public void testGroupIdentityIsTaggedAndQuoted() {
+ String group = "group,one=" + System.nanoTime();
+ HAGroupStoreMetricsSourceImpl taggedSource = new
HAGroupStoreMetricsSourceImpl(group);
+ try {
+ assertEquals(group,
taggedSource.getMetricsRegistry().getTag(HA_GROUP_TAG_NAME).value());
+ assertTrue(
+ taggedSource.getMetricsJmxContext().endsWith(",haGroup=" +
ObjectName.quote(group)));
+ } finally {
+ taggedSource.unregisterForTesting();
+ }
+ }
+}
