Skip to content

PHOENIX-7995 Tag client HA metrics with the HA group and add missing HA/CRR/failover client metrics #2609

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

Draft
lokiore wants to merge 2 commits into
base: PHOENIX-7562-feature-new
Choose a base branch
from
Draft

PHOENIX-7995 Tag client HA metrics with the HA group and add missing HA/CRR/failover client metrics #2609

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
d54f74a
PHOENIX-7872 Addendum record HA failover duration on the CRR-write pa…
lokiore Aug 20, 2026
4d58ed6
PHOENIX-7995 Tag client HA metrics with the HA group and add missing …
Aug 25, 2026
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
104 changes: 53 additions & 51 deletions phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/FailoverPhoenixConnection.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@

import static org.apache.phoenix.jdbc.HighAvailabilityUtil.isMutationBlockedIOExceptionExistsInThrowable;
import static org.apache.phoenix.jdbc.HighAvailabilityUtil.isStaleClusterRoleRecordExceptionExistsInThrowable;
import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HA_FAILOVER_DURATION_MS;
import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HA_FAILOVER_CONNECTION_CREATED_COUNTER;
import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HA_MUTATION_BLOCKED_COUNT;
import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HA_STALE_CRR_DETECTED_COUNT;

Expand All @@ -44,6 +44,7 @@
import org.apache.phoenix.exception.FailoverSQLException;
import org.apache.phoenix.exception.SQLExceptionCode;
import org.apache.phoenix.exception.SQLExceptionInfo;
import org.apache.phoenix.monitoring.HAGroupMetricsManager;
import org.apache.phoenix.monitoring.MetricType;
import org.apache.phoenix.util.EnvironmentEdgeManager;
import org.slf4j.Logger;
Expand Down Expand Up @@ -111,6 +112,11 @@ public FailoverPhoenixConnection(FailoverPhoenixContext context) throws SQLExcep
this.isClosed = false;
this.connection =
context.getHAGroup().connectActive(context.getProperties(), context.getHAURLInfo());
// A FailoverPhoenixConnection was successfully created against the active cluster. Pairs with
// HA_FAILOVER_CONNECTION_FAILED_COUNTER, which connectActive increments on its throw funnel.
GLOBAL_HA_FAILOVER_CONNECTION_CREATED_COUNTER.increment();
HAGroupMetricsManager.increment(context.getHAGroup().getName(),
MetricType.HA_FAILOVER_CONNECTION_CREATED_COUNTER);
}

/**
Expand Down Expand Up @@ -175,63 +181,55 @@ void failover(long timeoutMs) throws SQLException {
return;
}

final long failoverStartMs = EnvironmentEdgeManager.currentTimeMillis();
try {
PhoenixConnection newConn = null;
SQLException cause = null;
final long startTime = EnvironmentEdgeManager.currentTimeMillis();
while (
newConn == null && EnvironmentEdgeManager.currentTimeMillis() < startTime + timeoutMs
) {
PhoenixConnection newConn = null;
SQLException cause = null;
final long startTime = EnvironmentEdgeManager.currentTimeMillis();
while (newConn == null && EnvironmentEdgeManager.currentTimeMillis() < startTime + timeoutMs) {
try {
newConn =
context.getHAGroup().connectActive(context.getProperties(), context.getHAURLInfo());
} catch (SQLException e) {
cause = e;
LOG.info("Got exception when trying to connect to active cluster.", e);
try {
newConn =
context.getHAGroup().connectActive(context.getProperties(), context.getHAURLInfo());
} catch (SQLException e) {
cause = e;
LOG.info("Got exception when trying to connect to active cluster.", e);
try {
Thread.sleep(100); // TODO: be smart than this
} catch (InterruptedException ie) {
Thread.currentThread().interrupt();
throw new SQLException("Got interrupted waiting for connection failover", e);
}
Thread.sleep(100); // TODO: be smart than this
} catch (InterruptedException ie) {
Thread.currentThread().interrupt();
throw new SQLException("Got interrupted waiting for connection failover", e);
}
}
if (newConn == null) {
throw new FailoverSQLException("Can not failover connection",
context.getHAGroup().getGroupInfo().toString(), cause);
}
}
if (newConn == null) {
throw new FailoverSQLException("Can not failover connection",
context.getHAGroup().getGroupInfo().toString(), cause);
}

final PhoenixConnection oldConn = connection;
connection = newConn;
if (oldConn != null) {
// aggregate metrics
previousMutationMetrics = oldConn.getMutationMetrics();
previousReadMetrics = oldConn.getReadMetrics();
oldConn.clearMetrics();

// close old connection
if (!oldConn.isClosed()) {
// TODO: what happens to in-flight edits/mutations?
// Can we copy into the new connection we do not allow this failover?
// MutationState state = oldConn.getMutationState();
try {
oldConn.close(new SQLExceptionInfo.Builder(SQLExceptionCode.HA_CLOSED_AFTER_FAILOVER)
.setMessage("Phoenix connection got closed due to failover")
.setHaGroupInfo(context.getHAGroup().getGroupInfo().toString()).build()
.buildException());
} catch (SQLException e) {
LOG.error("Failed to close old connection after failover: {}", e.getMessage());
LOG.info("Full stack when closing old connection after failover", e);
}
final PhoenixConnection oldConn = connection;
connection = newConn;
if (oldConn != null) {
// aggregate metrics
previousMutationMetrics = oldConn.getMutationMetrics();
previousReadMetrics = oldConn.getReadMetrics();
oldConn.clearMetrics();

// close old connection
if (!oldConn.isClosed()) {
// TODO: what happens to in-flight edits/mutations?
// Can we copy into the new connection we do not allow this failover?
// MutationState state = oldConn.getMutationState();
try {
oldConn.close(new SQLExceptionInfo.Builder(SQLExceptionCode.HA_CLOSED_AFTER_FAILOVER)
.setMessage("Phoenix connection got closed due to failover")
.setHaGroupInfo(context.getHAGroup().getGroupInfo().toString()).build()
.buildException());
} catch (SQLException e) {
LOG.error("Failed to close old connection after failover: {}", e.getMessage());
LOG.info("Full stack when closing old connection after failover", e);
}
}
LOG.info("Connection {} failed over to {}", context.getHAGroup().getGroupInfo(),
connection.getURL());
} finally {
GLOBAL_HA_FAILOVER_DURATION_MS
.update(EnvironmentEdgeManager.currentTimeMillis() - failoverStartMs);
}
LOG.info("Connection {} failed over to {}", context.getHAGroup().getGroupInfo(),
connection.getURL());
}

/**
Expand Down Expand Up @@ -337,6 +335,8 @@ <T> T wrapActionDuringFailover(SupplierWithSQLException<T> s) throws SQLExceptio
} catch (Exception e) {
if (isStaleClusterRoleRecordExceptionExistsInThrowable(e)) {
GLOBAL_HA_STALE_CRR_DETECTED_COUNT.increment();
HAGroupMetricsManager.increment(context.getHAGroup().getName(),
MetricType.HA_STALE_CRR_DETECTED_COUNT);
// If we receive StaleClusterRoleRecordException, that means Operation was
// supposed to be executed on Active Cluster but was in reality was sent to
// STANDBY Cluster, that can happen only when Failover is in Progress, So we
Expand All @@ -363,6 +363,8 @@ <T> T wrapActionDuringFailover(SupplierWithSQLException<T> s) throws SQLExceptio
}
if (isMutationBlockedIOExceptionExistsInThrowable(e)) {
GLOBAL_HA_MUTATION_BLOCKED_COUNT.increment();
HAGroupMetricsManager.increment(context.getHAGroup().getName(),
MetricType.HA_MUTATION_BLOCKED_COUNT);
}
if (policy.shouldFailover(e, ++failoverCount)) {
failover(timeoutMs);
Expand Down
130 changes: 87 additions & 43 deletions phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HighAvailabilityGroup.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,11 @@

import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HA_CRR_CACHE_AGE_MS;
import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HA_CRR_REFRESH_COUNT;
import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HA_CRR_TRANSITION_COUNT;
import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HA_FAILOVER_CONNECTION_FAILED_COUNTER;
import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HA_FAILOVER_COUNT;
import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HA_FAILOVER_DURATION_MS;
import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HA_ROLE_TRANSITION_FAILED_COUNTER;
import static org.apache.phoenix.query.QueryServicesOptions.DEFAULT_CLIENT_CONNECTION_CACHE_MAX_DURATION;
import static org.apache.phoenix.util.PhoenixRuntime.JDBC_PROTOCOL_SEPARATOR;

Expand Down Expand Up @@ -61,6 +65,8 @@
import org.apache.phoenix.exception.SQLExceptionInfo;
import org.apache.phoenix.jdbc.ClusterRoleRecord.ClusterRole;
import org.apache.phoenix.jdbc.ClusterRoleRecord.RegistryType;
import org.apache.phoenix.monitoring.HAGroupMetricsManager;
import org.apache.phoenix.monitoring.MetricType;
import org.apache.phoenix.query.HBaseFactoryProvider;
import org.apache.phoenix.util.GetClusterRoleRecordUtil;
import org.apache.phoenix.util.JDBCUtil;
Expand Down Expand Up @@ -635,6 +641,10 @@ public void init() throws IOException, SQLException {
roleRecord = roleRecordFromEndpoint;
lastClusterRoleRecordRefreshTime = System.currentTimeMillis();
state = State.READY;
// Pre-register the per-group metrics2 source so the haGroup-tagged series exists as soon as the
// group is ready, rather than only after the first HA metric fires. No-op when global client
// metrics are disabled.
HAGroupMetricsManager.getOrCreate(getName());
}

/**
Expand Down Expand Up @@ -700,6 +710,11 @@ PhoenixConnection connectActive(final Properties properties, final HAURLInfo hau
} catch (SQLException e) {
LOG.error("Failed to connect to active cluster in HA group {}, record: {}", info, roleRecord,
e);
// Single throw funnel for a failed active-cluster connection (no active cluster, cluster
// demoted mid-connect, or the underlying connect threw). Counted here so it tracks real
// production failures regardless of failover policy.
GLOBAL_HA_FAILOVER_CONNECTION_FAILED_COUNTER.increment();
HAGroupMetricsManager.increment(getName(), MetricType.HA_FAILOVER_CONNECTION_FAILED_COUNTER);
throw new SQLExceptionInfo.Builder(SQLExceptionCode.CANNOT_ESTABLISH_CONNECTION)
.setMessage("Failed to connect to active cluster in HA group")
.setHaGroupInfo(info.toString()).setRootCause(e).build().buildException();
Expand Down Expand Up @@ -789,6 +804,8 @@ State getStateForTesting() {
*/
void close() {
state = State.CLOSED;
// Detach the per-group metrics2 source so its JMX-context name is freed for possible re-create.
HAGroupMetricsManager.remove(getName());
}

@Override
Expand Down Expand Up @@ -1160,6 +1177,7 @@ public boolean refreshClusterRoleRecord(boolean forceRefresh) throws SQLExceptio
// otherwise inflate this counter against its name (a "refresh" with no fetch is a no-op
// from a CRR-state perspective).
GLOBAL_HA_CRR_REFRESH_COUNT.increment();
HAGroupMetricsManager.increment(getName(), MetricType.HA_CRR_REFRESH_COUNT);
if (roleRecord == null) {
// First-load init path: no prior cache state to compare against, so this is not a
// failover transition and HA_FAILOVER_COUNT is intentionally NOT incremented here.
Expand Down Expand Up @@ -1199,6 +1217,11 @@ public boolean refreshClusterRoleRecord(boolean forceRefresh) throws SQLExceptio

final ClusterRoleRecord oldRecord = roleRecord;
state = State.IN_TRANSITION;
// Count every applied CRR transition, including transitions into a no-active state. Distinct
// from HA_FAILOVER_COUNT, which counts only transitions that establish/move an ACTIVE
// cluster.
GLOBAL_HA_CRR_TRANSITION_COUNT.increment();
HAGroupMetricsManager.increment(getName(), MetricType.CRR_TRANSITION_COUNT);
LOG.info("HA group {} is in {} to set V{} record", info, state, newRoleRecord.getVersion());
Future<?> future = crrChangedExecutor.submit(() -> {
try {
Expand All @@ -1213,53 +1236,74 @@ public boolean refreshClusterRoleRecord(boolean forceRefresh) throws SQLExceptio
long maxTransitionTimeMs = StringUtils.isNotEmpty(transitionTimeoutProp)
? Long.parseLong(transitionTimeoutProp)
: PHOENIX_HA_TRANSITION_TIMEOUT_MS_DEFAULT;
boolean transitionSucceeded = false;
// Time the cluster-transition dispatch on this CRR-write path, which is where autonomous
// failovers are actually driven. The duration is recorded on every exit (success, timeout,
// policy failure, or interrupt) via the finally block below so it tracks time spent handling
// detected CRR transitions rather than the connection-level failover() path, which is never
// auto-invoked under the default ExplicitFailoverPolicy.
final long transitionStartMs = System.currentTimeMillis();
try {
future.get(maxTransitionTimeMs, TimeUnit.MILLISECONDS);
transitionSucceeded = true;
} catch (InterruptedException ie) {
LOG.error("Got interrupted when transiting cluster roles for HA group {}", info, ie);
future.cancel(true);
Thread.currentThread().interrupt();
return false;
} catch (ExecutionException | TimeoutException e) {
LOG.error("HA group {} failed to transit cluster roles per policy {} to new " + "record {}",
info, roleRecord.getPolicy(), newRoleRecord, e);
// Rethrow the Role transitions not allowed exceptions
if (e.getCause() != null && e.getCause().getCause() != null) {
if (
e.getCause().getCause() instanceof SQLException
&& ((SQLException) e.getCause().getCause()).getErrorCode()
== SQLExceptionCode.HA_ROLE_TRANSITION_NOT_ALLOWED.getErrorCode()
) {
state = State.READY;
throw (SQLException) e.getCause().getCause();
boolean transitionSucceeded = false;
try {
future.get(maxTransitionTimeMs, TimeUnit.MILLISECONDS);
transitionSucceeded = true;
} catch (InterruptedException ie) {
LOG.error("Got interrupted when transiting cluster roles for HA group {}", info, ie);
future.cancel(true);
Thread.currentThread().interrupt();
return false;
} catch (ExecutionException | TimeoutException e) {
LOG.error(
"HA group {} failed to transit cluster roles per policy {} to new " + "record {}", info,
roleRecord.getPolicy(), newRoleRecord, e);
// Dispatch of the policy-side cluster-role transition failed (execution error or timed
// out). Count every such failure, including the HA_ROLE_TRANSITION_NOT_ALLOWED case
// rethrown just below.
GLOBAL_HA_ROLE_TRANSITION_FAILED_COUNTER.increment();
HAGroupMetricsManager.increment(getName(), MetricType.HA_ROLE_TRANSITION_FAILED_COUNTER);
// Rethrow the Role transitions not allowed exceptions
if (e.getCause() != null && e.getCause().getCause() != null) {
if (
e.getCause().getCause() instanceof SQLException
&& ((SQLException) e.getCause().getCause()).getErrorCode()
== SQLExceptionCode.HA_ROLE_TRANSITION_NOT_ALLOWED.getErrorCode()
) {
state = State.READY;
throw (SQLException) e.getCause().getCause();
}
}
// Calling back HA policy function for cluster switch is conducted with best effort.
// HA group continues transition when its HA policy fails to deal with context switch
// (e.g. to close existing connections)
// The goal here is to gain higher availability even though existing resources against
// previous ACTIVE cluster may have not been closed cleanly.
}
// Calling back HA policy function for cluster switch is conducted with best effort.
// HA group continues transition when its HA policy fails to deal with context switch
// (e.g. to close existing connections)
// The goal here is to gain higher availability even though existing resources against
// previous ACTIVE cluster may have not been closed cleanly.
}
// Count the transition as a failover only when the policy-side transition actually
// succeeded AND an active cluster is established or moves between peers. Operator-driven
// transitions to a no-active state (both clusters STANDBY) are not counted as failovers;
// recovery from no-active back to having an ACTIVE peer is counted. Transitions where
// future.get() failed (ExecutionException/TimeoutException) are best-effort fall-through
// per the comment above, but they are NOT counted as successful failovers. Gate decision
// factored into the package-private static {@link #shouldCountFailover} so it can be
// unit-tested directly without driving a full mini-cluster transition.
if (shouldCountFailover(transitionSucceeded, oldRecord, newRoleRecord)) {
GLOBAL_HA_FAILOVER_COUNT.increment();
// Count the transition as a failover only when the policy-side transition actually
// succeeded AND an active cluster is established or moves between peers. Operator-driven
// transitions to a no-active state (both clusters STANDBY) are not counted as failovers;
// recovery from no-active back to having an ACTIVE peer is counted. Transitions where
// future.get() failed (ExecutionException/TimeoutException) are best-effort fall-through
// per the comment above, but they are NOT counted as successful failovers. Gate decision
// factored into the package-private static {@link #shouldCountFailover} so it can be
// unit-tested directly without driving a full mini-cluster transition.
if (shouldCountFailover(transitionSucceeded, oldRecord, newRoleRecord)) {
GLOBAL_HA_FAILOVER_COUNT.increment();
HAGroupMetricsManager.increment(getName(), MetricType.HA_FAILOVER_COUNT);
}
// Update the role record and the last refresh time
roleRecord = newRoleRecord;
lastClusterRoleRecordRefreshTime = System.currentTimeMillis();
state = State.READY;
LOG.info("HA group {} is in {} state, Old: {}, new: {}", info, state, oldRecord,
roleRecord);
LOG.debug("HA group is ready: {}", this);
return true;
} finally {
long transitionDurationMs = System.currentTimeMillis() - transitionStartMs;
GLOBAL_HA_FAILOVER_DURATION_MS.update(transitionDurationMs);
HAGroupMetricsManager.update(getName(), MetricType.HA_FAILOVER_DURATION_MS,
transitionDurationMs);
}
// Update the role record and the last refresh time
roleRecord = newRoleRecord;
lastClusterRoleRecordRefreshTime = System.currentTimeMillis();
state = State.READY;
LOG.info("HA group {} is in {} state, Old: {}, new: {}", info, state, oldRecord, roleRecord);
LOG.debug("HA group is ready: {}", this);
return true;
} finally {
writeLock.unlock();
}
Expand Down
4 changes: 4 additions & 0 deletions phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HighAvailabilityPolicy.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,8 @@
import org.apache.phoenix.exception.SQLExceptionCode;
import org.apache.phoenix.exception.SQLExceptionInfo;
import org.apache.phoenix.monitoring.GlobalClientMetrics;
import org.apache.phoenix.monitoring.HAGroupMetricsManager;
import org.apache.phoenix.monitoring.MetricType;
import org.apache.phoenix.query.ConnectionQueryServices;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Expand Down Expand Up @@ -194,6 +196,8 @@ public Connection provide(HighAvailabilityGroup haGroup, Properties info, HAURLI
// Give regular connection or a failover connection?
LOG.warn("Falling back to single phoenix connection due to resource constraints");
GlobalClientMetrics.GLOBAL_HA_PARALLEL_CONNECTION_FALLBACK_COUNTER.increment();
HAGroupMetricsManager.increment(haGroup.getName(),
MetricType.HA_PARALLEL_CONNECTION_FALLBACK_COUNTER);
return haGroup.connectActive(info, haURLInfo);
}
}
Expand Down
Loading