Skip to content

PHOENIX-7872 Addendum record HA failover duration on the CRR-write path and add a connection-failed counter #2605

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

PHOENIX-7872 Addendum record HA failover duration on the CRR-write path and add a connection-failed counter #2605

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
93 changes: 42 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,6 @@

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_MUTATION_BLOCKED_COUNT;
import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HA_STALE_CRR_DETECTED_COUNT;

Expand Down Expand Up @@ -175,63 +174,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
104 changes: 61 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,9 @@

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_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.query.QueryServicesOptions.DEFAULT_CLIENT_CONNECTION_CACHE_MAX_DURATION;
import static org.apache.phoenix.util.PhoenixRuntime.JDBC_PROTOCOL_SEPARATOR;

Expand Down Expand Up @@ -700,6 +702,10 @@ 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();
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 @@ -1213,53 +1219,65 @@ 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);
// 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();
}
// 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 {
GLOBAL_HA_FAILOVER_DURATION_MS.update(System.currentTimeMillis() - transitionStartMs);
}
// 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
2 changes: 2 additions & 0 deletions phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/GlobalClientMetrics.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@
import static org.apache.phoenix.monitoring.MetricType.COUNT_SCANNED_REGIONS;
import static org.apache.phoenix.monitoring.MetricType.HA_CRR_CACHE_AGE_MS;
import static org.apache.phoenix.monitoring.MetricType.HA_CRR_REFRESH_COUNT;
import static org.apache.phoenix.monitoring.MetricType.HA_FAILOVER_CONNECTION_FAILED_COUNTER;
import static org.apache.phoenix.monitoring.MetricType.HA_FAILOVER_COUNT;
import static org.apache.phoenix.monitoring.MetricType.HA_FAILOVER_DURATION_MS;
import static org.apache.phoenix.monitoring.MetricType.HA_MUTATION_BLOCKED_COUNT;
Expand Down Expand Up @@ -173,6 +174,7 @@ public enum GlobalClientMetrics {

GLOBAL_HA_FAILOVER_COUNT(HA_FAILOVER_COUNT),
GLOBAL_HA_FAILOVER_DURATION_MS(HA_FAILOVER_DURATION_MS),
GLOBAL_HA_FAILOVER_CONNECTION_FAILED_COUNTER(HA_FAILOVER_CONNECTION_FAILED_COUNTER),
GLOBAL_HA_MUTATION_BLOCKED_COUNT(HA_MUTATION_BLOCKED_COUNT),
GLOBAL_HA_STALE_CRR_DETECTED_COUNT(HA_STALE_CRR_DETECTED_COUNT),
GLOBAL_HA_CRR_REFRESH_COUNT(HA_CRR_REFRESH_COUNT),
Expand Down
8 changes: 6 additions & 2 deletions phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/MetricType.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -359,8 +359,12 @@ public enum MetricType {
+ "from no-active state) recorded at the CRR write site",
LogLevel.DEBUG, PLong.INSTANCE),
HA_FAILOVER_DURATION_MS("hafd",
"Total time in milliseconds spent in connection-level failover transitions, summed across "
+ "all observing connections (per-connection observation, not per-cluster-event)",
"Total time in milliseconds spent dispatching cluster-role transitions, recorded at the CRR "
+ "write site (refreshClusterRoleRecord) on every transition exit",
LogLevel.DEBUG, PLong.INSTANCE),
HA_FAILOVER_CONNECTION_FAILED_COUNTER("hafcf",
"Counter for failed attempts to connect to the active cluster in an HA group, recorded at the "
+ "connectActive throw site (no active cluster, demoted mid-connect, or connect error)",
LogLevel.DEBUG, PLong.INSTANCE),
HA_MUTATION_BLOCKED_COUNT("hambc",
"Counter for MutationBlockedIOException surfaces caught by wrapActionDuringFailover",
Expand Down
Loading