Skip to content

PHOENIX-7562 Add HAGroupStore Metrics2 observability #2576

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
ritegarg wants to merge 1 commit into
base: PHOENIX-7562-feature-new
Choose a base branch
from
Open

PHOENIX-7562 Add HAGroupStore Metrics2 observability #2576

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
57 changes: 49 additions & 8 deletions phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HAGroupStoreClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,6 +78,8 @@
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;
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -278,6 +281,8 @@ public static List<String> getHAGroupNames(String zkUrl) throws SQLException {
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);
Expand All @@ -299,7 +304,10 @@ public static List<String> getHAGroupNames(String zkUrl) throws SQLException {
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 {
Expand Down Expand Up @@ -795,6 +803,7 @@ private void updateSystemTableHAGroupRecordSilently(String haGroupName,
pstmt.executeUpdate();
conn.commit();
} catch (Exception e) {
metricsSource.incrementSystemTableSyncFailedCount();
LOGGER.error(
"Failed to update system table on best" + "effort basis for HA group {}, error: {}",
haGroupName, e);
Expand Down Expand Up @@ -885,6 +894,14 @@ private void syncZKToSystemTable() {
}
}

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,
Expand All @@ -908,6 +925,7 @@ private PathChildrenCacheListener localCacheListener() {
// unreachable peer never blocks local event processing.
String peerZKUrl = record.getPeerZKUrl();
if (!Objects.equals(peerZKUrl, lastConfiguredPeerZKUrl)) {
metricsSource.setCurrentPeerState(HAGroupState.UNKNOWN);
lastConfiguredPeerZKUrl = peerZKUrl;
peerWatcher.reconfigureAsync(peerZKUrl);
}
Expand All @@ -916,12 +934,18 @@ private PathChildrenCacheListener localCacheListener() {
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;
Expand Down Expand Up @@ -1007,6 +1031,7 @@ public void onPeerStateChanged(HAGroupStoreRecord peerRecord, Stat stat) {
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,
Expand All @@ -1017,11 +1042,15 @@ public void onPeerStateChanged(HAGroupStoreRecord peerRecord, Stat stat) {

@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);
Expand Down Expand Up @@ -1060,6 +1089,8 @@ private boolean presentLocalDegradedStandbyIfStandby(HAGroupState priorEffective
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);
Expand All @@ -1086,6 +1117,7 @@ private void clearLocalDegradedStandbyIfStillStandby() {
}
local = readLocalRecordQuietly();
localDegradedStandbyActive = false;
metricsSource.setDegradedStandbyActive(false);
recover = local != null && local.getHAGroupState() == HAGroupState.STANDBY;
}
if (recover) {
Expand Down Expand Up @@ -1194,6 +1226,7 @@ private long validateTransitionAndGetWaitTime(HAGroupStoreRecord.HAGroupState cu
waitTime = waitTimeForSyncModeInMs;
}
} else {
metricsSource.incrementInvalidTransitionRejectedCount();
throw new InvalidClusterRoleTransitionException(
"Cannot transition from " + currentHAGroupState + " to " + newHAGroupState);
}
Expand Down Expand Up @@ -1439,6 +1472,9 @@ private void handleLocalStateChange(HAGroupStoreRecord newRecord, Stat newStat)
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) {
Expand Down Expand Up @@ -1499,15 +1535,20 @@ private void notifySubscribers(HAGroupState fromState, HAGroupState toState, lon
LOGGER.info("Notifying {} listeners of state transitionfor 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);
}
}
}
Expand Down
37 changes: 23 additions & 14 deletions phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HAGroupStoreRecord.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,15 +49,15 @@ public class HAGroupStoreRecord {
* Enum representing the HA group state with each state having a corresponding ClusterRole.
*/
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
Expand All @@ -66,13 +66,22 @@ public enum HAGroupState {
* 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;
}

public int getMetricCode() {
return metricCode;
}

/**
* Gets the corresponding ClusterRole for this HAGroupState.
Expand Down
Loading