From d54f74a407d6df43cca312e1f0113a71bf9fc74a Mon Sep 17 00:00:00 2001 From: lokiore Date: Thu, 20 Aug 2026 13:21:02 -0700 Subject: [PATCH 1/2] PHOENIX-7872 Addendum record HA failover duration on the CRR-write path and add a connection-failed counter The HA failover observability metrics added under PHOENIX-7872 recorded HA_FAILOVER_DURATION_MS inside FailoverPhoenixConnection.failover(long). That method is only reached through wrapActionDuringFailover -> FailoverPolicy .shouldFailover(), which returns false under the default ExplicitFailoverPolicy, or through the explicit static failover(Connection, long) helper. Neither runs during an autonomous, CRR-driven failover, so the duration metric never moved in production. Move the duration measurement to the path that actually drives failovers: refreshClusterRoleRecord, where the cluster-role transition is dispatched and where HA_FAILOVER_COUNT is already gated by shouldCountFailover. The dispatch block is wrapped in a try/finally so the duration is recorded on every exit (success, timeout, policy failure, or interrupt), avoiding a silent metric miss if a future exit path is added. The now-dead timing in failover(long) is removed. Add HA_FAILOVER_CONNECTION_FAILED_COUNTER, incremented at the single SQLException throw funnel in connectActive (no active cluster, cluster demoted mid-connect, or the underlying connect threw). This tracks real active-cluster connection failures regardless of the configured failover policy. Tests: three unit tests in HighAvailabilityGroupTest -- a counted role-flip transition records both HA_FAILOVER_COUNT and an HA_FAILOVER_DURATION_MS sample on the CRR-write path; a failed connectActive increments the connection-failed counter; a successful connectActive leaves it unchanged. HighAvailabilityGroupTest 16/16, FailoverPhoenixConnectionTest 8/8. Generated-by: Claude Code (Opus 4.8) Co-Authored-By: Claude Opus 4.8 (1M context) --- .../jdbc/FailoverPhoenixConnection.java | 93 +++++++-------- .../phoenix/jdbc/HighAvailabilityGroup.java | 104 ++++++++++------- .../monitoring/GlobalClientMetrics.java | 2 + .../apache/phoenix/monitoring/MetricType.java | 8 +- .../jdbc/HighAvailabilityGroupTest.java | 106 ++++++++++++++++++ 5 files changed, 217 insertions(+), 96 deletions(-) diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/FailoverPhoenixConnection.java b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/FailoverPhoenixConnection.java index 5916ea86c4d..387c85e40b1 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/FailoverPhoenixConnection.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/FailoverPhoenixConnection.java @@ -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; @@ -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()); } /** diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HighAvailabilityGroup.java b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HighAvailabilityGroup.java index 05ce2787644..db4408fdbf2 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HighAvailabilityGroup.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HighAvailabilityGroup.java @@ -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; @@ -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(); @@ -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(); } diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/GlobalClientMetrics.java b/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/GlobalClientMetrics.java index bbd23bcbc52..816a142cfa9 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/GlobalClientMetrics.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/GlobalClientMetrics.java @@ -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; @@ -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), diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/MetricType.java b/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/MetricType.java index 22a9ecbdcf3..ecb3d314471 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/MetricType.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/MetricType.java @@ -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", diff --git a/phoenix-core/src/test/java/org/apache/phoenix/jdbc/HighAvailabilityGroupTest.java b/phoenix-core/src/test/java/org/apache/phoenix/jdbc/HighAvailabilityGroupTest.java index 5bf956fa88e..f323f183576 100644 --- a/phoenix-core/src/test/java/org/apache/phoenix/jdbc/HighAvailabilityGroupTest.java +++ b/phoenix-core/src/test/java/org/apache/phoenix/jdbc/HighAvailabilityGroupTest.java @@ -27,6 +27,7 @@ import java.io.IOException; import java.io.InterruptedIOException; import java.sql.SQLException; +import java.util.HashSet; import java.util.Properties; import org.apache.phoenix.exception.SQLExceptionCode; import org.apache.phoenix.jdbc.ClusterRoleRecord.ClusterRole; @@ -618,4 +619,109 @@ public void testEndpointRestoresInterruptStatusOnFallback() throws Exception { Thread.interrupted(); } } + + /** + * A real counted role-flip transition driven through {@code refreshClusterRoleRecord} must both + * increment {@code HA_FAILOVER_COUNT} and record a sample on {@code HA_FAILOVER_DURATION_MS}. + * This pins the duration metric to the CRR-write transition path (the path that autonomous + * failovers actually take) rather than the connection-level + * {@code FailoverPhoenixConnection.failover()} path, which is never auto-invoked under the + * default {@code ExplicitFailoverPolicy}. Active URL flips from url1 to url2 so + * {@code shouldCountFailover} returns true; the {@code URLS} entry is seeded empty so the policy + * transition is a clean no-op (no real connections needed). + */ + @Test + public void testCountedTransitionRecordsFailoverCountAndDuration() throws Exception { + String haGroupName = "testCountedTransitionRecordsFailoverCountAndDuration"; + String url1 = "host1\\:60010"; + String url2 = "host2\\:60010"; + ClusterRoleRecord aActiveBStandby = new ClusterRoleRecord(haGroupName, + HighAvailabilityPolicy.FAILOVER, url1, ClusterRole.ACTIVE, url2, ClusterRole.STANDBY, 10L); + ClusterRoleRecord aStandbyBActive = new ClusterRoleRecord(haGroupName, + HighAvailabilityPolicy.FAILOVER, url1, ClusterRole.STANDBY, url2, ClusterRole.ACTIVE, 11L); + + HAGroupInfo info = new HAGroupInfo(haGroupName, url1, url2); + // Seed an empty URL set so the policy-side transition iterates nothing and is a clean no-op. + HighAvailabilityGroup.URLS.put(info, new HashSet<>()); + try { + HighAvailabilityGroup group = Mockito + .spy(new HighAvailabilityGroup(info, new Properties(), aActiveBStandby, State.READY)); + Mockito.doReturn(aStandbyBActive).when(group).getClusterRoleRecordFromEndpoint(); + + long countBefore = GlobalClientMetrics.GLOBAL_HA_FAILOVER_COUNT.getMetric().getValue(); + long durationSamplesBefore = + GlobalClientMetrics.GLOBAL_HA_FAILOVER_DURATION_MS.getMetric().getNumberOfSamples(); + + assertTrue("A real role-flip transition must apply and return true", + group.refreshClusterRoleRecord(true)); + assertSame("The new record must be applied after the transition", aStandbyBActive, + group.getRoleRecord()); + + assertEquals("An active-URL flip must increment HA_FAILOVER_COUNT", countBefore + 1, + GlobalClientMetrics.GLOBAL_HA_FAILOVER_COUNT.getMetric().getValue()); + assertEquals( + "The transition must record a HA_FAILOVER_DURATION_MS sample on the CRR-write " + "path", + durationSamplesBefore + 1, + GlobalClientMetrics.GLOBAL_HA_FAILOVER_DURATION_MS.getMetric().getNumberOfSamples()); + } finally { + HighAvailabilityGroup.URLS.remove(info); + } + } + + /** + * A failed {@code connectActive} (no active cluster in the record) must increment + * {@code HA_FAILOVER_CONNECTION_FAILED_COUNTER} on its single SQLException throw funnel. + */ + @Test + public void testConnectActiveFailureIncrementsFailedCounter() { + String haGroupName = "testConnectActiveFailureIncrementsFailedCounter"; + String url1 = "host1\\:60010"; + String url2 = "host2\\:60010"; + // Both STANDBY → no active URL → connectActive takes the HA_NO_ACTIVE_CLUSTER throw path. + ClusterRoleRecord bothStandby = new ClusterRoleRecord(haGroupName, + HighAvailabilityPolicy.FAILOVER, url1, ClusterRole.STANDBY, url2, ClusterRole.STANDBY, 10L); + HAGroupInfo info = new HAGroupInfo(haGroupName, url1, url2); + HighAvailabilityGroup group = + new HighAvailabilityGroup(info, new Properties(), bothStandby, State.READY); + + long failedBefore = + GlobalClientMetrics.GLOBAL_HA_FAILOVER_CONNECTION_FAILED_COUNTER.getMetric().getValue(); + try { + group.connectActive(new Properties(), new HAURLInfo(haGroupName)); + fail("connectActive must throw when the HA group has no active cluster"); + } catch (SQLException e) { + assertEquals(SQLExceptionCode.CANNOT_ESTABLISH_CONNECTION.getErrorCode(), e.getErrorCode()); + } + assertEquals("A failed connectActive must increment the failed counter", failedBefore + 1, + GlobalClientMetrics.GLOBAL_HA_FAILOVER_CONNECTION_FAILED_COUNTER.getMetric().getValue()); + } + + /** + * A successful {@code connectActive} must NOT increment + * {@code HA_FAILOVER_CONNECTION_FAILED_COUNTER}. Guards against the counter being placed on a + * path that also runs on success (a non-vacuous negative assertion). + */ + @Test + public void testConnectActiveSuccessLeavesFailedCounterUnchanged() throws Exception { + String haGroupName = "testConnectActiveSuccessLeavesFailedCounterUnchanged"; + String url1 = "host1\\:60010"; + String url2 = "host2\\:60010"; + ClusterRoleRecord aActiveBStandby = new ClusterRoleRecord(haGroupName, + HighAvailabilityPolicy.FAILOVER, url1, ClusterRole.ACTIVE, url2, ClusterRole.STANDBY, 10L); + HAGroupInfo info = new HAGroupInfo(haGroupName, url1, url2); + HighAvailabilityGroup group = + Mockito.spy(new HighAvailabilityGroup(info, new Properties(), aActiveBStandby, State.READY)); + + PhoenixConnection conn = Mockito.mock(PhoenixConnection.class); + Mockito.doReturn(conn).when(group).connectToOneCluster(Mockito.any(String.class), + Mockito.any(Properties.class), Mockito.any(HAURLInfo.class)); + Mockito.doReturn(true).when(group).isActive(conn); + + long failedBefore = + GlobalClientMetrics.GLOBAL_HA_FAILOVER_CONNECTION_FAILED_COUNTER.getMetric().getValue(); + assertSame("connectActive must return the established connection", conn, + group.connectActive(new Properties(), new HAURLInfo(haGroupName))); + assertEquals("A successful connectActive must not increment the failed counter", failedBefore, + GlobalClientMetrics.GLOBAL_HA_FAILOVER_CONNECTION_FAILED_COUNTER.getMetric().getValue()); + } } From 4d58ed6191c3e1859f6d5a8a6869ef976a76ca9c Mon Sep 17 00:00:00 2001 From: lokesh-khurana Date: Mon, 24 Aug 2026 17:05:52 -0700 Subject: [PATCH 2/2] PHOENIX-7995 Tag client HA metrics with the HA group and add missing HA/CRR/failover client metrics The ZK-less HA client emits all HA/CRR/failover metrics as JVM-global GLOBAL_HA_* counters, so a JVM connected to more than one HA group cannot attribute a failover, stale-CRR detection, or poller failure to a specific group. It also lacks client counters for several HA/CRR/failover events that are relevant on the ZK-less path. Per-HA-group tagging -------------------- Add HAGroupClientMetricsSource, a per-group Hadoop Metrics2 source (one per HA group name) that stamps a "haGroup" tag carrying the group name and appends the quoted group name to its JMX context so each group registers as a distinct source/MBean. HAGroupMetricsManager is the process-wide registry: it lazily creates a source per group (only when global client metrics are enabled), routes per-group increment/update, and detaches the source on HA-group close so the same group can re-register later. Emission is dual: every group-attributable GLOBAL_HA_* increment is mirrored to the group's source, and the JVM-global counters continue to emit unchanged. The tag key lives in a new module-neutral MetricConstants.HA_GROUP_TAG_NAME ("haGroup") referenced by both this client source and the server-side HAGroupStoreMetricsSource, so both sides tag with the same key and can be filtered together downstream. New client counters ------------------- - HA_FAILOVER_CONNECTION_CREATED_COUNTER: FailoverPhoenixConnection instances successfully created against the active cluster (pairs with the existing connection-failed counter). - HA_ROLE_TRANSITION_FAILED_COUNTER: cluster-role-transition dispatch failures (execution error or timeout) on the CRR-write path. - CRR_TRANSITION_COUNT: cluster-role-record transitions applied per HA policy, including transitions into a no-active state (distinct from HA_FAILOVER_COUNT, which counts only transitions that establish/move an ACTIVE cluster). Group-attributable emission is wired at the existing sites in HighAvailabilityGroup (failover count/duration, CRR refresh, connect-failed, and the two new transition metrics), FailoverPhoenixConnection (connection created, stale-CRR, mutation-blocked), HighAvailabilityPolicy (parallel fallback), ParallelPhoenixContext/ParallelPhoenixUtil (parallel connection created/error, task timeout), and GetClusterRoleRecordUtil (poller tick count/failures). The JVM-shared parallel-executor pool metrics and the HA_CRR_CACHE_AGE_MS gauge are deliberately excluded from the per-group source (JVM-shared / not a counter). Tests: HAGroupClientMetricsSourceTest (6) and HAGroupMetricsManagerTest (7) cover the haGroup tag/quoting, per-counter increment/update, per-group isolation, distinct registered sources, and detach/re-create. 13/13 pass. Generated-by: Claude Code (Opus 4.8) Co-Authored-By: Claude Opus 4.8 (1M context) --- .../jdbc/FailoverPhoenixConnection.java | 11 ++ .../phoenix/jdbc/HighAvailabilityGroup.java | 28 ++- .../phoenix/jdbc/HighAvailabilityPolicy.java | 4 + .../phoenix/jdbc/ParallelPhoenixContext.java | 5 + .../phoenix/jdbc/ParallelPhoenixUtil.java | 4 + .../metrics/HAGroupStoreMetricsSource.java | 7 +- .../phoenix/metrics/MetricConstants.java | 39 +++++ .../monitoring/GlobalClientMetrics.java | 6 + .../HAGroupClientMetricsSource.java | 131 ++++++++++++++ .../monitoring/HAGroupMetricsManager.java | 126 ++++++++++++++ .../apache/phoenix/monitoring/MetricType.java | 12 ++ .../util/GetClusterRoleRecordUtil.java | 4 + .../HAGroupClientMetricsSourceTest.java | 130 ++++++++++++++ .../monitoring/HAGroupMetricsManagerTest.java | 164 ++++++++++++++++++ 14 files changed, 668 insertions(+), 3 deletions(-) create mode 100644 phoenix-core-client/src/main/java/org/apache/phoenix/metrics/MetricConstants.java create mode 100644 phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/HAGroupClientMetricsSource.java create mode 100644 phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/HAGroupMetricsManager.java create mode 100644 phoenix-core/src/test/java/org/apache/phoenix/monitoring/HAGroupClientMetricsSourceTest.java create mode 100644 phoenix-core/src/test/java/org/apache/phoenix/monitoring/HAGroupMetricsManagerTest.java diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/FailoverPhoenixConnection.java b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/FailoverPhoenixConnection.java index 387c85e40b1..b8761a106a2 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/FailoverPhoenixConnection.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/FailoverPhoenixConnection.java @@ -19,6 +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_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; @@ -43,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; @@ -110,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); } /** @@ -328,6 +335,8 @@ T wrapActionDuringFailover(SupplierWithSQLException 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 @@ -354,6 +363,8 @@ T wrapActionDuringFailover(SupplierWithSQLException 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); diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HighAvailabilityGroup.java b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HighAvailabilityGroup.java index db4408fdbf2..2271f697361 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HighAvailabilityGroup.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HighAvailabilityGroup.java @@ -19,9 +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; @@ -63,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; @@ -637,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()); } /** @@ -706,6 +714,7 @@ PhoenixConnection connectActive(final Properties properties, final HAURLInfo hau // 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(); @@ -795,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 @@ -1166,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. @@ -1205,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 { @@ -1239,6 +1256,11 @@ public boolean refreshClusterRoleRecord(boolean forceRefresh) throws SQLExceptio 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 ( @@ -1266,6 +1288,7 @@ public boolean refreshClusterRoleRecord(boolean forceRefresh) throws SQLExceptio // 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; @@ -1276,7 +1299,10 @@ public boolean refreshClusterRoleRecord(boolean forceRefresh) throws SQLExceptio LOG.debug("HA group is ready: {}", this); return true; } finally { - GLOBAL_HA_FAILOVER_DURATION_MS.update(System.currentTimeMillis() - transitionStartMs); + long transitionDurationMs = System.currentTimeMillis() - transitionStartMs; + GLOBAL_HA_FAILOVER_DURATION_MS.update(transitionDurationMs); + HAGroupMetricsManager.update(getName(), MetricType.HA_FAILOVER_DURATION_MS, + transitionDurationMs); } } finally { writeLock.unlock(); diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HighAvailabilityPolicy.java b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HighAvailabilityPolicy.java index 6b898ebeb5f..34aeb56a813 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HighAvailabilityPolicy.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HighAvailabilityPolicy.java @@ -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; @@ -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); } } diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/ParallelPhoenixContext.java b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/ParallelPhoenixContext.java index 928ad2923bd..3472c62c0ba 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/ParallelPhoenixContext.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/ParallelPhoenixContext.java @@ -32,6 +32,7 @@ import java.util.stream.Collectors; 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.query.QueryServices; import org.apache.phoenix.query.QueryServicesOptions; @@ -80,6 +81,8 @@ public class ParallelPhoenixContext { Preconditions.checkArgument(executors.size() >= 2, "Expected 2 executor pairs, one for each connection with a normal/close executor"); GLOBAL_HA_PARALLEL_CONNECTION_CREATED_COUNTER.increment(); + HAGroupMetricsManager.increment(haGroup.getName(), + MetricType.HA_PARALLEL_CONNECTION_CREATED_COUNTER); this.properties = properties; this.haGroup = haGroup; this.haurlInfo = haurlInfo; @@ -176,6 +179,8 @@ public void close() { isClosed = true; if (isErrored) { GLOBAL_HA_PARALLEL_CONNECTION_ERROR_COUNTER.increment(); + HAGroupMetricsManager.increment(this.haGroup.getName(), + MetricType.HA_PARALLEL_CONNECTION_ERROR_COUNTER); } } diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/ParallelPhoenixUtil.java b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/ParallelPhoenixUtil.java index b223e3d9fb9..bead6b47450 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/ParallelPhoenixUtil.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/ParallelPhoenixUtil.java @@ -34,7 +34,9 @@ import org.apache.hadoop.hbase.util.PairOfSameType; import org.apache.phoenix.exception.SQLExceptionCode; import org.apache.phoenix.exception.SQLExceptionInfo; +import org.apache.phoenix.monitoring.HAGroupMetricsManager; import org.apache.phoenix.monitoring.Metric; +import org.apache.phoenix.monitoring.MetricType; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -126,6 +128,8 @@ public Object getAnyOfNonExceptionally(List> if (timedout) { GLOBAL_HA_PARALLEL_TASK_TIMEOUT_COUNTER.increment(); + HAGroupMetricsManager.increment(context.getHaGroup().getName(), + MetricType.HA_PARALLEL_TASK_TIMEOUT_COUNTER); if (futures.isEmpty()) { LOGGER.warn("Unexpected race between timeout and failure occurred."); 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 index 0af587c5037..1a1ad0fc605 100644 --- 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 @@ -19,6 +19,7 @@ import org.apache.hadoop.hbase.metrics.BaseSource; import org.apache.phoenix.jdbc.HAGroupStoreRecord.HAGroupState; +import org.apache.phoenix.metrics.MetricConstants; import org.apache.phoenix.thirdparty.com.google.common.annotations.VisibleForTesting; @@ -61,8 +62,10 @@ public interface HAGroupStoreMetricsSource extends BaseSource { // 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"; + // Shared with the client HA-group source via the module-neutral MetricConstants so both + // reference the same tag key ("haGroup") and can be filtered together. + String HA_GROUP_TAG_NAME = MetricConstants.HA_GROUP_TAG_NAME; + String HA_GROUP_TAG_DESC = MetricConstants.HA_GROUP_TAG_DESC; String LOCAL_CACHE_HEALTH_STATUS = "haGroupStoreLocalCacheHealthStatus"; String LOCAL_CACHE_HEALTH_STATUS_DESC = diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/metrics/MetricConstants.java b/phoenix-core-client/src/main/java/org/apache/phoenix/metrics/MetricConstants.java new file mode 100644 index 00000000000..39600a14610 --- /dev/null +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/metrics/MetricConstants.java @@ -0,0 +1,39 @@ +/* + * 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.metrics; + +/** + * Shared Hadoop Metrics2 constant definitions used across Phoenix metric sources. + *

+ * Holds the {@code haGroup} tag registered by the per-HA-group sources so their series can be + * sliced per HA group downstream. Lives in this module-neutral package so both the client source + * ({@code HAGroupClientMetricsSource}) and the server source + * ({@code org.apache.phoenix.jdbc.metrics.HAGroupStoreMetricsSource}) reference the same tag key + * and a query can filter both with a single label. + */ +public final class MetricConstants { + + /** Metrics2 tag name carrying the HA group name. */ + public static final String HA_GROUP_TAG_NAME = "haGroup"; + + /** Description for the {@link #HA_GROUP_TAG_NAME} tag. */ + public static final String HA_GROUP_TAG_DESC = "HA group name"; + + private MetricConstants() { + } +} diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/GlobalClientMetrics.java b/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/GlobalClientMetrics.java index 816a142cfa9..18e79e67cf3 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/GlobalClientMetrics.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/GlobalClientMetrics.java @@ -35,8 +35,10 @@ import static org.apache.phoenix.monitoring.MetricType.COUNT_RPC_CALLS; import static org.apache.phoenix.monitoring.MetricType.COUNT_RPC_RETRIES; import static org.apache.phoenix.monitoring.MetricType.COUNT_SCANNED_REGIONS; +import static org.apache.phoenix.monitoring.MetricType.CRR_TRANSITION_COUNT; 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_CREATED_COUNTER; 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; @@ -57,6 +59,7 @@ import static org.apache.phoenix.monitoring.MetricType.HA_PARALLEL_TASK_TIMEOUT_COUNTER; import static org.apache.phoenix.monitoring.MetricType.HA_POLLER_TICK_COUNT; import static org.apache.phoenix.monitoring.MetricType.HA_POLLER_TICK_FAILURES; +import static org.apache.phoenix.monitoring.MetricType.HA_ROLE_TRANSITION_FAILED_COUNTER; import static org.apache.phoenix.monitoring.MetricType.HA_STALE_CRR_DETECTED_COUNT; import static org.apache.phoenix.monitoring.MetricType.HCONNECTIONS_COUNTER; import static org.apache.phoenix.monitoring.MetricType.INDEX_COMMIT_FAILURE_SIZE; @@ -175,6 +178,9 @@ 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_FAILOVER_CONNECTION_CREATED_COUNTER(HA_FAILOVER_CONNECTION_CREATED_COUNTER), + GLOBAL_HA_ROLE_TRANSITION_FAILED_COUNTER(HA_ROLE_TRANSITION_FAILED_COUNTER), + GLOBAL_HA_CRR_TRANSITION_COUNT(CRR_TRANSITION_COUNT), 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), diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/HAGroupClientMetricsSource.java b/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/HAGroupClientMetricsSource.java new file mode 100644 index 00000000000..708dbebd486 --- /dev/null +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/HAGroupClientMetricsSource.java @@ -0,0 +1,131 @@ +/* + * 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.monitoring; + +import static org.apache.phoenix.monitoring.MetricType.CRR_TRANSITION_COUNT; +import static org.apache.phoenix.monitoring.MetricType.HA_CRR_REFRESH_COUNT; +import static org.apache.phoenix.monitoring.MetricType.HA_FAILOVER_CONNECTION_CREATED_COUNTER; +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; +import static org.apache.phoenix.monitoring.MetricType.HA_PARALLEL_CONNECTION_CREATED_COUNTER; +import static org.apache.phoenix.monitoring.MetricType.HA_PARALLEL_CONNECTION_ERROR_COUNTER; +import static org.apache.phoenix.monitoring.MetricType.HA_PARALLEL_CONNECTION_FALLBACK_COUNTER; +import static org.apache.phoenix.monitoring.MetricType.HA_PARALLEL_TASK_TIMEOUT_COUNTER; +import static org.apache.phoenix.monitoring.MetricType.HA_POLLER_TICK_COUNT; +import static org.apache.phoenix.monitoring.MetricType.HA_POLLER_TICK_FAILURES; +import static org.apache.phoenix.monitoring.MetricType.HA_ROLE_TRANSITION_FAILED_COUNTER; +import static org.apache.phoenix.monitoring.MetricType.HA_STALE_CRR_DETECTED_COUNT; + +import java.util.EnumMap; +import java.util.Map; +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.phoenix.metrics.MetricConstants; + +import org.apache.phoenix.thirdparty.com.google.common.annotations.VisibleForTesting; + +/** + * Per-HA-group Hadoop Metrics2 source for the ZK-less HA client's HA/CRR/failover metrics. + *

+ * All groups share one metrics name/context; each instance appends {@code ,haGroup=} + * to the JMX context so it registers as a distinct metrics2 source / MBean, and stamps a + * {@code haGroup} tag ({@link MetricConstants#HA_GROUP_TAG_NAME}) carrying the (unquoted) group + * name so the series can be sliced per HA group downstream. This mirrors the server-side + * {@code HAGroupStoreMetricsSourceImpl} tagging pattern; both reference the same tag key. + *

+ * Each metric in {@link #METRIC_TYPES} is a monotonic {@link MutableFastCounter}. Accumulating + * metrics such as {@code HA_FAILOVER_DURATION_MS} add total milliseconds via {@link #update}, which + * matches the semantics of the JVM-global {@code GLOBAL_HA_*} counters; those continue to emit + * unchanged alongside this per-group source (dual emit). + *

+ * The set is intentionally limited to metrics attributable to a single HA group (each emission site + * has a {@code HighAvailabilityGroup} in scope). The shared parallel-executor pool metrics + * ({@code HA_PARALLEL_POOL1_*}/{@code HA_PARALLEL_POOL2_*}) and the {@code HA_CRR_CACHE_AGE_MS} + * gauge are excluded: the pools are JVM-shared and the gauge is not a counter. + */ +public class HAGroupClientMetricsSource extends BaseSourceImpl { + + static final String METRICS_NAME = "HAGroupClient"; + static final String METRICS_DESC = "Phoenix HA Group Client Metrics"; + static final String METRICS_CONTEXT = "phoenix"; + static final String METRICS_JMX_CONTEXT = "Phoenix,sub=" + METRICS_NAME; + + /** + * The HA metrics attributable to a single HA group; each emission site has a + * {@code HighAvailabilityGroup} in scope. + */ + public static final MetricType[] METRIC_TYPES = new MetricType[] { HA_FAILOVER_COUNT, + HA_FAILOVER_DURATION_MS, HA_FAILOVER_CONNECTION_CREATED_COUNTER, + HA_FAILOVER_CONNECTION_FAILED_COUNTER, HA_STALE_CRR_DETECTED_COUNT, HA_MUTATION_BLOCKED_COUNT, + HA_CRR_REFRESH_COUNT, HA_ROLE_TRANSITION_FAILED_COUNTER, CRR_TRANSITION_COUNT, + HA_PARALLEL_CONNECTION_FALLBACK_COUNTER, HA_PARALLEL_CONNECTION_CREATED_COUNTER, + HA_PARALLEL_CONNECTION_ERROR_COUNTER, HA_PARALLEL_TASK_TIMEOUT_COUNTER, HA_POLLER_TICK_COUNT, + HA_POLLER_TICK_FAILURES }; + + private final Map counters = new EnumMap<>(MetricType.class); + + public HAGroupClientMetricsSource(String haGroupName) { + super(METRICS_NAME, METRICS_DESC, METRICS_CONTEXT, + METRICS_JMX_CONTEXT + ",haGroup=" + ObjectName.quote(haGroupName)); + getMetricsRegistry().tag( + Interns.info(MetricConstants.HA_GROUP_TAG_NAME, MetricConstants.HA_GROUP_TAG_DESC), + haGroupName); + for (MetricType type : METRIC_TYPES) { + counters.put(type, + getMetricsRegistry().newCounter(type.columnName(), type.description(), 0L)); + } + } + + /** Increment the group's counter for the given HA metric type. */ + public void increment(MetricType type) { + MutableFastCounter counter = counters.get(type); + if (counter != null) { + counter.incr(); + } + } + + /** + * Add {@code value} to the group's accumulating counter for the given HA metric type (used for + * {@code HA_FAILOVER_DURATION_MS}). + */ + public void update(MetricType type, long value) { + MutableFastCounter counter = counters.get(type); + if (counter != null) { + counter.incr(value); + } + } + + /** + * Detach this source from the metrics system on HA-group close, freeing its JMX-context name in + * {@link DefaultMetricsSystem} so the same group can register again if later re-created. + */ + public void unregister() { + DefaultMetricsSystem.instance().unregisterSource(metricsJmxContext); + } + + @VisibleForTesting + long getCounterValue(MetricType type) { + MutableFastCounter counter = counters.get(type); + return counter == null ? -1L : counter.value(); + } +} diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/HAGroupMetricsManager.java b/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/HAGroupMetricsManager.java new file mode 100644 index 00000000000..58538ef1382 --- /dev/null +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/HAGroupMetricsManager.java @@ -0,0 +1,126 @@ +/* + * 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.monitoring; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import org.apache.commons.lang3.StringUtils; +import org.apache.phoenix.query.QueryServicesOptions; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.phoenix.thirdparty.com.google.common.annotations.VisibleForTesting; + +/** + * Central registry of {@link HAGroupClientMetricsSource}, one per HA group name. Each group's + * source registers with Hadoop's metrics2 {@code DefaultMetricsSystem} tagged + * {@code haGroup=} so the HA/CRR/failover client metrics can be sliced per HA group alongside + * the JVM-global {@code GLOBAL_HA_*} counters (which continue to emit unchanged). + *

+ * Sources are created only when global client metrics are enabled (constructing a source registers + * it with the metrics system). {@link #remove(String)} detaches a group's source on HA-group close. + */ +public class HAGroupMetricsManager { + + private static final Logger LOGGER = LoggerFactory.getLogger(HAGroupMetricsManager.class); + + private static final boolean IS_GLOBAL_METRICS_ENABLED = + QueryServicesOptions.withDefaults().isGlobalMetricsEnabled(); + + private static final Map GROUP_SOURCES = + new ConcurrentHashMap<>(); + + private HAGroupMetricsManager() { + } + + /** + * Get (registering a metrics2 source on first sight) the metrics source for an HA group. Safe to + * call repeatedly for the same name. Returns {@code null} for a null/empty group name or when + * global client metrics are disabled. + */ + public static HAGroupClientMetricsSource getOrCreate(String haGroupName) { + if (StringUtils.isEmpty(haGroupName) || !IS_GLOBAL_METRICS_ENABLED) { + return null; + } + HAGroupClientMetricsSource source = GROUP_SOURCES.get(haGroupName); + if (source != null) { + return source; + } + synchronized (HAGroupMetricsManager.class) { + source = GROUP_SOURCES.get(haGroupName); + if (source == null) { + source = new HAGroupClientMetricsSource(haGroupName); + GROUP_SOURCES.put(haGroupName, source); + LOGGER.info("Created HA-group client metrics source for group '{}'", haGroupName); + } + } + return source; + } + + /** + * Increment the per-group counter for an HA metric type. Registers the group's source if needed. + * Metric emission is best-effort and never propagates an exception to the caller's request path. + */ + public static void increment(String haGroupName, MetricType type) { + try { + HAGroupClientMetricsSource source = getOrCreate(haGroupName); + if (source != null) { + source.increment(type); + } + } catch (Exception e) { + LOGGER.error("Failed incrementing HA-group metric {} for group '{}'", type, haGroupName, e); + } + } + + /** + * Add a sample to a per-group accumulating HA metric (e.g. {@code HA_FAILOVER_DURATION_MS}). + * Registers the group's source if needed. Best-effort; never propagates an exception. + */ + public static void update(String haGroupName, MetricType type, long value) { + try { + HAGroupClientMetricsSource source = getOrCreate(haGroupName); + if (source != null) { + source.update(type, value); + } + } catch (Exception e) { + LOGGER.error("Failed updating HA-group metric {} for group '{}'", type, haGroupName, e); + } + } + + /** + * Tear down a group's metrics source on HA-group close, freeing its metrics2 source name so the + * same group can be re-created later. + */ + public static void remove(String haGroupName) { + if (StringUtils.isEmpty(haGroupName)) { + return; + } + synchronized (HAGroupMetricsManager.class) { + HAGroupClientMetricsSource source = GROUP_SOURCES.remove(haGroupName); + if (source != null) { + source.unregister(); + LOGGER.info("Removed HA-group client metrics source for group '{}'", haGroupName); + } + } + } + + @VisibleForTesting + public static HAGroupClientMetricsSource getIfPresent(String haGroupName) { + return GROUP_SOURCES.get(haGroupName); + } +} diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/MetricType.java b/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/MetricType.java index ecb3d314471..37db8e80a48 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/MetricType.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/MetricType.java @@ -366,6 +366,18 @@ public enum MetricType { "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_FAILOVER_CONNECTION_CREATED_COUNTER("hafcc", + "Counter for FailoverPhoenixConnection instances successfully created (connected to the active " + + "cluster); pairs with HA_FAILOVER_CONNECTION_FAILED_COUNTER", + LogLevel.DEBUG, PLong.INSTANCE), + HA_ROLE_TRANSITION_FAILED_COUNTER("hrtf", + "Counter for cluster-role-transition dispatch failures (ExecutionException/TimeoutException) " + + "when applying a new CRR per HA policy at the CRR-write path", + LogLevel.DEBUG, PLong.INSTANCE), + CRR_TRANSITION_COUNT("crtc", + "Counter for cluster role record transitions applied per HA policy (a role/url change that " + + "triggers a transit), including transitions to a no-active state", + LogLevel.DEBUG, PLong.INSTANCE), HA_MUTATION_BLOCKED_COUNT("hambc", "Counter for MutationBlockedIOException surfaces caught by wrapActionDuringFailover", LogLevel.DEBUG, PLong.INSTANCE), diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/util/GetClusterRoleRecordUtil.java b/phoenix-core-client/src/main/java/org/apache/phoenix/util/GetClusterRoleRecordUtil.java index b25d5f46f21..1bc202d87ec 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/util/GetClusterRoleRecordUtil.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/util/GetClusterRoleRecordUtil.java @@ -43,6 +43,8 @@ import org.apache.phoenix.jdbc.HighAvailabilityPolicy; import org.apache.phoenix.jdbc.PhoenixConnection; import org.apache.phoenix.monitoring.GlobalClientMetrics; +import org.apache.phoenix.monitoring.HAGroupMetricsManager; +import org.apache.phoenix.monitoring.MetricType; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -223,6 +225,7 @@ private static void schedulePoller(String url1, String url2, String haGroupName, // Increment unconditionally so a failed tick still alternates next iteration. long tick = tickCount.getAndIncrement(); GlobalClientMetrics.GLOBAL_HA_POLLER_TICK_COUNT.increment(); + HAGroupMetricsManager.increment(haGroupName, MetricType.HA_POLLER_TICK_COUNT); // Sample current CRR cache age into the gauge each tick. Without this, the // HA_CRR_CACHE_AGE_MS counter-backed gauge is only updated on connect() and would // not advance during idle periods between connects, making it look fresher than it @@ -271,6 +274,7 @@ private static void schedulePoller(String url1, String url2, String haGroupName, } } catch (SQLException e) { GlobalClientMetrics.GLOBAL_HA_POLLER_TICK_FAILURES.increment(); + HAGroupMetricsManager.increment(haGroupName, MetricType.HA_POLLER_TICK_FAILURES); LOGGER.error( "Exception found while polling for ClusterRoleRecord on {} for HA group" + " {}: {}", tickUrl, haGroupName, e.getMessage()); diff --git a/phoenix-core/src/test/java/org/apache/phoenix/monitoring/HAGroupClientMetricsSourceTest.java b/phoenix-core/src/test/java/org/apache/phoenix/monitoring/HAGroupClientMetricsSourceTest.java new file mode 100644 index 00000000000..c08e18d1766 --- /dev/null +++ b/phoenix-core/src/test/java/org/apache/phoenix/monitoring/HAGroupClientMetricsSourceTest.java @@ -0,0 +1,130 @@ +/* + * 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.monitoring; + +import static org.apache.phoenix.metrics.MetricConstants.HA_GROUP_TAG_NAME; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +import java.util.ArrayList; +import java.util.List; +import javax.management.ObjectName; +import org.apache.hadoop.metrics2.MetricsTag; +import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem; +import org.junit.After; +import org.junit.Test; + +/** + * Unit tests for {@link HAGroupClientMetricsSource}: the per-HA-group metrics2 source stamps a + * {@code haGroup} tag carrying the group name, keeps per-group counters, and detaches cleanly on + * {@link HAGroupClientMetricsSource#unregister()}. + *

+ * Constructing a source registers it with the process-global {@link DefaultMetricsSystem}, so each + * test uses a unique group name and {@link #tearDown()} unregisters every source it created. + */ +public class HAGroupClientMetricsSourceTest { + + private final List created = new ArrayList<>(); + + private HAGroupClientMetricsSource newSource(String group) { + HAGroupClientMetricsSource source = new HAGroupClientMetricsSource(group); + created.add(source); + return source; + } + + @After + public void tearDown() { + for (HAGroupClientMetricsSource source : created) { + source.unregister(); + } + created.clear(); + } + + @Test + public void testTagCarriesUnquotedGroupName() { + String group = "srcTag"; + HAGroupClientMetricsSource source = newSource(group); + MetricsTag tag = source.getMetricsRegistry().getTag(HA_GROUP_TAG_NAME); + assertNotNull("the haGroup tag must be present on the source", tag); + assertEquals("haGroup tag must carry the unquoted group name", group, tag.value()); + } + + @Test + public void testJmxContextIsQuotedPerGroup() { + String group = "group,one=weird"; + HAGroupClientMetricsSource source = newSource(group); + assertEquals(group, source.getMetricsRegistry().getTag(HA_GROUP_TAG_NAME).value()); + assertTrue(source.getMetricsJmxContext().endsWith(",haGroup=" + ObjectName.quote(group))); + } + + @Test + public void testIncrementAndUpdateArePerCounter() { + HAGroupClientMetricsSource source = newSource("srcCounters"); + source.increment(MetricType.HA_FAILOVER_CONNECTION_FAILED_COUNTER); + source.increment(MetricType.HA_FAILOVER_CONNECTION_FAILED_COUNTER); + // The duration metric accumulates via update(); other counters are untouched by it. + source.update(MetricType.HA_FAILOVER_DURATION_MS, 40L); + source.update(MetricType.HA_FAILOVER_DURATION_MS, 60L); + + assertEquals(2L, source.getCounterValue(MetricType.HA_FAILOVER_CONNECTION_FAILED_COUNTER)); + assertEquals(100L, source.getCounterValue(MetricType.HA_FAILOVER_DURATION_MS)); + assertEquals("an untouched counter stays at zero", 0L, + source.getCounterValue(MetricType.HA_STALE_CRR_DETECTED_COUNT)); + } + + @Test + public void testUnknownMetricTypeIsIgnored() { + HAGroupClientMetricsSource source = newSource("srcUnknown"); + // A type that is not in METRIC_TYPES has no backing counter: increment/update are no-ops and + // getCounterValue reports -1 to distinguish "absent" from a present-but-zero counter. + source.increment(MetricType.MUTATION_BATCH_SIZE); + source.update(MetricType.MUTATION_BATCH_SIZE, 5L); + assertEquals(-1L, source.getCounterValue(MetricType.MUTATION_BATCH_SIZE)); + } + + @Test + public void testEachGroupIsADistinctRegisteredSource() { + HAGroupClientMetricsSource a = newSource("srcDistinctA"); + HAGroupClientMetricsSource b = newSource("srcDistinctB"); + assertNotEquals("distinct groups must get distinct JMX contexts", a.getMetricsJmxContext(), + b.getMetricsJmxContext()); + assertNotNull(DefaultMetricsSystem.instance().getSource(a.getMetricsJmxContext())); + assertNotNull(DefaultMetricsSystem.instance().getSource(b.getMetricsJmxContext())); + } + + @Test + public void testUnregisterFreesTheSourceNameForReuse() { + String group = "srcReuse"; + HAGroupClientMetricsSource first = new HAGroupClientMetricsSource(group); + String jmxContext = first.getMetricsJmxContext(); + assertNotNull(DefaultMetricsSystem.instance().getSource(jmxContext)); + + first.unregister(); + assertNull("unregister must free the source name in DefaultMetricsSystem", + DefaultMetricsSystem.instance().getSource(jmxContext)); + + // The same group name must be registrable again (would throw "source already exists" if the + // prior source were still attached). + HAGroupClientMetricsSource second = newSource(group); + assertNotNull(DefaultMetricsSystem.instance().getSource(second.getMetricsJmxContext())); + assertEquals(jmxContext, second.getMetricsJmxContext()); + } +} diff --git a/phoenix-core/src/test/java/org/apache/phoenix/monitoring/HAGroupMetricsManagerTest.java b/phoenix-core/src/test/java/org/apache/phoenix/monitoring/HAGroupMetricsManagerTest.java new file mode 100644 index 00000000000..388cf378267 --- /dev/null +++ b/phoenix-core/src/test/java/org/apache/phoenix/monitoring/HAGroupMetricsManagerTest.java @@ -0,0 +1,164 @@ +/* + * 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.monitoring; + +import static org.apache.phoenix.metrics.MetricConstants.HA_GROUP_TAG_NAME; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; + +import java.util.ArrayList; +import java.util.List; +import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem; +import org.apache.phoenix.query.QueryServicesOptions; +import org.junit.After; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; + +/** + * Unit tests for {@link HAGroupMetricsManager}: the per-HA-group source registry. Covers per-group + * isolation (two groups never cross-count), that each group registers as its own + * {@code haGroup}-tagged metrics2 source, and that {@link HAGroupMetricsManager#remove(String)} + * detaches the source so the same group can be re-created fresh. + *

+ * These tests mutate process-global state ({@link DefaultMetricsSystem} and the manager's static + * map), so each group name is unique per test and {@link #tearDown()} removes every group this + * class created to avoid leaking sources into sibling tests. + */ +public class HAGroupMetricsManagerTest { + + private final List createdGroups = new ArrayList<>(); + + @BeforeClass + public static void assumeGlobalMetricsEnabled() { + // The manager creates sources only when global metrics are enabled (the default). If a + // surrounding config disabled them, getOrCreate would return null and these assertions would + // not apply. + assertTrue("these tests assume the default global-metrics-enabled=true", + QueryServicesOptions.withDefaults().isGlobalMetricsEnabled()); + } + + @Before + public void setUp() { + createdGroups.clear(); + } + + @After + public void tearDown() { + for (String group : createdGroups) { + HAGroupMetricsManager.remove(group); + } + } + + private String track(String group) { + createdGroups.add(group); + return group; + } + + @Test + public void testGetOrCreateIsIdempotentAndRegistersSource() { + String group = track("mgrCreate"); + HAGroupClientMetricsSource first = HAGroupMetricsManager.getOrCreate(group); + HAGroupClientMetricsSource second = HAGroupMetricsManager.getOrCreate(group); + assertNotNull(first); + assertSame("getOrCreate must be idempotent for a group name", first, second); + assertNotNull("the group's source must be registered with DefaultMetricsSystem", + DefaultMetricsSystem.instance().getSource(first.getMetricsJmxContext())); + assertEquals("the source must carry the haGroup tag", group, + first.getMetricsRegistry().getTag(HA_GROUP_TAG_NAME).value()); + } + + @Test + public void testNullAndEmptyGroupNameIsNoOp() { + assertNull(HAGroupMetricsManager.getOrCreate(null)); + assertNull(HAGroupMetricsManager.getOrCreate("")); + // remove must tolerate null/empty without throwing. + HAGroupMetricsManager.remove(null); + HAGroupMetricsManager.remove(""); + } + + @Test + public void testTwoGroupsDoNotCrossCount() { + String groupA = track("mgrIsoA"); + String groupB = track("mgrIsoB"); + MetricType type = MetricType.HA_FAILOVER_CONNECTION_FAILED_COUNTER; + + HAGroupMetricsManager.increment(groupA, type); + HAGroupMetricsManager.increment(groupA, type); + HAGroupMetricsManager.increment(groupB, type); + + assertEquals(2L, HAGroupMetricsManager.getIfPresent(groupA).getCounterValue(type)); + assertEquals("each HA group must maintain independent counters", 1L, + HAGroupMetricsManager.getIfPresent(groupB).getCounterValue(type)); + } + + @Test + public void testEachGroupGetsADistinctTaggedSource() { + String groupA = track("mgrDistinctA"); + String groupB = track("mgrDistinctB"); + HAGroupClientMetricsSource regA = HAGroupMetricsManager.getOrCreate(groupA); + HAGroupClientMetricsSource regB = HAGroupMetricsManager.getOrCreate(groupB); + + assertNotEquals("distinct groups must map to distinct JMX contexts", + regA.getMetricsJmxContext(), regB.getMetricsJmxContext()); + assertEquals(groupA, regA.getMetricsRegistry().getTag(HA_GROUP_TAG_NAME).value()); + assertEquals(groupB, regB.getMetricsRegistry().getTag(HA_GROUP_TAG_NAME).value()); + } + + @Test + public void testUpdateAccumulatesPerGroupDuration() { + String group = track("mgrDuration"); + MetricType type = MetricType.HA_FAILOVER_DURATION_MS; + HAGroupMetricsManager.update(group, type, 40L); + HAGroupMetricsManager.update(group, type, 60L); + assertEquals(100L, HAGroupMetricsManager.getIfPresent(group).getCounterValue(type)); + } + + @Test + public void testRemoveDetachesSource() { + String group = "mgrRemove"; // not tracked: this test removes it itself + HAGroupClientMetricsSource source = HAGroupMetricsManager.getOrCreate(group); + String jmxContext = source.getMetricsJmxContext(); + assertNotNull(HAGroupMetricsManager.getIfPresent(group)); + assertNotNull(DefaultMetricsSystem.instance().getSource(jmxContext)); + + HAGroupMetricsManager.remove(group); + + assertNull("holder must be gone after remove", HAGroupMetricsManager.getIfPresent(group)); + assertNull("source must be detached from DefaultMetricsSystem after remove", + DefaultMetricsSystem.instance().getSource(jmxContext)); + } + + @Test + public void testGetOrCreateAfterRemoveRebuildsFresh() { + String group = track("mgrRebuild"); + MetricType type = MetricType.HA_STALE_CRR_DETECTED_COUNT; + HAGroupMetricsManager.increment(group, type); + assertEquals(1L, HAGroupMetricsManager.getIfPresent(group).getCounterValue(type)); + + HAGroupMetricsManager.remove(group); + // A re-create after remove must yield a fresh source with zeroed counters. + HAGroupClientMetricsSource rebuilt = HAGroupMetricsManager.getOrCreate(group); + assertNotNull(rebuilt); + assertEquals("re-created group must start from zero", 0L, rebuilt.getCounterValue(type)); + } +}