diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HAGroupStoreClient.java b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HAGroupStoreClient.java index ec001849696..a0fea14b65c 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HAGroupStoreClient.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HAGroupStoreClient.java @@ -78,6 +78,8 @@ import org.apache.phoenix.jdbc.ClusterRoleRecord.ClusterRole; import org.apache.phoenix.jdbc.ClusterRoleRecord.RegistryType; import org.apache.phoenix.jdbc.HAGroupStoreRecord.HAGroupState; +import org.apache.phoenix.jdbc.metrics.HAGroupStoreMetricsSource; +import org.apache.phoenix.jdbc.metrics.HAGroupStoreMetricsSourceFactory; import org.apache.phoenix.query.QueryServices; import org.apache.phoenix.query.QueryServicesOptions; import org.apache.phoenix.util.JDBCUtil; @@ -118,6 +120,7 @@ public class HAGroupStoreClient implements Closeable { new ConcurrentHashMap<>(); // HAGroupName for this instance private final String haGroupName; + private final HAGroupStoreMetricsSource metricsSource; // PathChildrenCache for current cluster and HAGroupName private PathChildrenCache pathChildrenCache = null; // Watches the peer cluster's HA record; owns the peer cache/admin, retry, and visibility. @@ -278,6 +281,8 @@ public static List getHAGroupNames(String zkUrl) throws SQLException { final String zkUrl) { this.conf = conf; this.haGroupName = haGroupName; + this.metricsSource = HAGroupStoreMetricsSourceFactory.getInstanceForHAGroup(haGroupName); + resetMetricGauges(); this.zkUrl = zkUrl; this.waitTimeForSyncModeInMs = (long) Math.ceil( conf.getLong(ZK_SESSION_TIMEOUT, DEFAULT_ZK_SESSION_TIMEOUT) * ZK_SESSION_TIMEOUT_MULTIPLIER); @@ -299,7 +304,10 @@ public static List getHAGroupNames(String zkUrl) throws SQLException { initializeZNodeIfNeeded(); if (this.pathChildrenCache != null) { this.isHealthy = true; + metricsSource.setLocalCacheHealthy(true); HAGroupStoreRecord local = getHAGroupStoreRecord(); + metricsSource.setCurrentLocalState( + local != null ? local.getHAGroupState() : HAGroupStoreRecord.HAGroupState.UNKNOWN); this.lastConfiguredPeerZKUrl = local != null ? local.getPeerZKUrl() : null; peerWatcher.reconfigure(this.lastConfiguredPeerZKUrl); } else { @@ -795,6 +803,7 @@ private void updateSystemTableHAGroupRecordSilently(String haGroupName, pstmt.executeUpdate(); conn.commit(); } catch (Exception e) { + metricsSource.incrementSystemTableSyncFailedCount(); LOGGER.error( "Failed to update system table on best" + "effort basis for HA group {}, error: {}", haGroupName, e); @@ -885,6 +894,14 @@ private void syncZKToSystemTable() { } } + private void resetMetricGauges() { + metricsSource.setLocalCacheHealthy(false); + metricsSource.setPeerVisible(false); + metricsSource.setDegradedStandbyActive(false); + metricsSource.setCurrentLocalState(HAGroupState.UNKNOWN); + metricsSource.setCurrentPeerState(HAGroupState.UNKNOWN); + } + /** * Listener for the local cache: tracks local state transitions (suppressing STANDBY while * degraded), keeps the peer watcher pointed at the current peer ZK, drives the legacy CRR sync, @@ -908,6 +925,7 @@ private PathChildrenCacheListener localCacheListener() { // unreachable peer never blocks local event processing. String peerZKUrl = record.getPeerZKUrl(); if (!Objects.equals(peerZKUrl, lastConfiguredPeerZKUrl)) { + metricsSource.setCurrentPeerState(HAGroupState.UNKNOWN); lastConfiguredPeerZKUrl = peerZKUrl; peerWatcher.reconfigureAsync(peerZKUrl); } @@ -916,12 +934,18 @@ private PathChildrenCacheListener localCacheListener() { break; case CONNECTION_LOST: case CONNECTION_SUSPENDED: + boolean wasHealthy = isHealthy; isHealthy = false; + metricsSource.setLocalCacheHealthy(false); + if (wasHealthy) { + metricsSource.incrementLocalZkConnectionLostCount(); + } LOGGER.warn("LOCAL HAGroupStoreClient cache connection lost/suspended for HA group {}", haGroupName); break; case CONNECTION_RECONNECTED: isHealthy = true; + metricsSource.setLocalCacheHealthy(true); LOGGER.info("LOCAL HAGroupStoreClient cache connection reconnected for HA group {}", haGroupName); break; @@ -1007,6 +1031,7 @@ public void onPeerStateChanged(HAGroupStoreRecord peerRecord, Stat stat) { HAGroupState fromState = lastKnownPeerState; HAGroupState toState = peerRecord.getHAGroupState(); lastKnownPeerState = toState; + metricsSource.setCurrentPeerState(toState); LOGGER.info("Detected state transition for HA group {} from {} to {} on PEER cluster", haGroupName, fromState, toState); notifySubscribers(fromState, toState, stat != null ? stat.getMtime() : 0L, ClusterType.PEER, @@ -1017,11 +1042,15 @@ public void onPeerStateChanged(HAGroupStoreRecord peerRecord, Stat stat) { @Override public void onPeerVisible() { + metricsSource.setPeerVisible(true); clearLocalDegradedStandbyIfStillStandby(); } @Override public void onPeerBlind() { + metricsSource.setPeerVisible(false); + metricsSource.setCurrentPeerState(HAGroupState.UNKNOWN); + metricsSource.incrementPeerBlindCount(); // Peer just went blind while local is STANDBY (present re-checks): subscribers last saw the // raw STANDBY, so that is the prior effective state. presentLocalDegradedStandbyIfStandby(HAGroupState.STANDBY); @@ -1060,6 +1089,8 @@ private boolean presentLocalDegradedStandbyIfStandby(HAGroupState priorEffective return false; } localDegradedStandbyActive = true; + metricsSource.setDegradedStandbyActive(true); + metricsSource.incrementDegradedStandbyPresentedCount(); } LOGGER.warn("Peer not visible for HA group {}; presenting local DEGRADED_STANDBY " + "(reason=peer-blind)", haGroupName); @@ -1086,6 +1117,7 @@ private void clearLocalDegradedStandbyIfStillStandby() { } local = readLocalRecordQuietly(); localDegradedStandbyActive = false; + metricsSource.setDegradedStandbyActive(false); recover = local != null && local.getHAGroupState() == HAGroupState.STANDBY; } if (recover) { @@ -1194,6 +1226,7 @@ private long validateTransitionAndGetWaitTime(HAGroupStoreRecord.HAGroupState cu waitTime = waitTimeForSyncModeInMs; } } else { + metricsSource.incrementInvalidTransitionRejectedCount(); throw new InvalidClusterRoleTransitionException( "Cannot transition from " + currentHAGroupState + " to " + newHAGroupState); } @@ -1439,6 +1472,9 @@ private void handleLocalStateChange(HAGroupStoreRecord newRecord, Stat newStat) synchronized (localTransitionLock) { HAGroupState oldState = lastKnownLocalState; lastKnownLocalState = newState; + metricsSource.setCurrentLocalState(newState); + metricsSource + .setDegradedStandbyActive(localDegradedStandbyActive && newState == HAGroupState.STANDBY); // Fail closed for a final STANDBY while the peer is not visible. STANDBY_TO_ACTIVE and // ABORT_TO_STANDBY pass through because replay listens for those failover signals. if (newState == HAGroupState.STANDBY) { @@ -1499,15 +1535,20 @@ private void notifySubscribers(HAGroupState fromState, HAGroupState toState, lon LOGGER.info("Notifying {} listeners of state transitionfor HA group {} from {} to {} on {} " + "cluster", listenersToNotify.size(), haGroupName, fromState, toState, clusterType); - for (HAGroupStateListener listener : listenersToNotify) { - try { - listener.onStateChange(haGroupName, fromState, toState, modifiedTime, clusterType, - lastSyncStateTimeInMs); - } catch (Exception e) { - LOGGER.error("Error notifying listener of state transition for HA group {} from {} to " - + "{} on {} cluster", haGroupName, fromState, toState, clusterType, e); - // Continue notifying other listeners + long startTimeNs = System.nanoTime(); + try { + for (HAGroupStateListener listener : listenersToNotify) { + try { + listener.onStateChange(haGroupName, fromState, toState, modifiedTime, clusterType, + lastSyncStateTimeInMs); + } catch (Exception e) { + metricsSource.incrementNotificationListenerErrorCount(); + LOGGER.error("Error notifying listener of state transition for HA group {} from {} to " + + "{} on {} cluster", haGroupName, fromState, toState, clusterType, e); + } } + } finally { + metricsSource.updateSubscriberNotifyTime(System.nanoTime() - startTimeNs); } } } diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HAGroupStoreRecord.java b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HAGroupStoreRecord.java index 0494ca9d344..dd5e6380fda 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HAGroupStoreRecord.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HAGroupStoreRecord.java @@ -49,15 +49,15 @@ public class HAGroupStoreRecord { * Enum representing the HA group state with each state having a corresponding ClusterRole. */ public enum HAGroupState { - ABORT_TO_ACTIVE_IN_SYNC, - ABORT_TO_ACTIVE_NOT_IN_SYNC, - ABORT_TO_STANDBY, - ACTIVE_IN_SYNC, - ACTIVE_NOT_IN_SYNC, - ACTIVE_NOT_IN_SYNC_TO_STANDBY, - ACTIVE_NOT_IN_SYNC_WITH_OFFLINE_PEER, - ACTIVE_IN_SYNC_TO_STANDBY, - ACTIVE_WITH_OFFLINE_PEER, + ABORT_TO_ACTIVE_IN_SYNC(1), + ABORT_TO_ACTIVE_NOT_IN_SYNC(2), + ABORT_TO_STANDBY(3), + ACTIVE_IN_SYNC(4), + ACTIVE_NOT_IN_SYNC(5), + ACTIVE_NOT_IN_SYNC_TO_STANDBY(6), + ACTIVE_NOT_IN_SYNC_WITH_OFFLINE_PEER(7), + ACTIVE_IN_SYNC_TO_STANDBY(8), + ACTIVE_WITH_OFFLINE_PEER(9), /** * Degraded standby. Used two ways: (1) a persisted state for a reader-degraded standby (see * {@link HAGroupStoreManager#setReaderToDegraded(String)}), governed by the allowedTransitions @@ -66,13 +66,22 @@ public enum HAGroupState { * cannot see its peer - that overlay is never persisted to ZK and is cleared (not transitioned) * when the peer becomes visible again. */ - DEGRADED_STANDBY, - OFFLINE, - STANDBY, - STANDBY_TO_ACTIVE, - UNKNOWN; + DEGRADED_STANDBY(10), + OFFLINE(11), + STANDBY(12), + STANDBY_TO_ACTIVE(13), + UNKNOWN(0); private Set allowedTransitions; + private final int metricCode; + + HAGroupState(int metricCode) { + this.metricCode = metricCode; + } + + public int getMetricCode() { + return metricCode; + } /** * Gets the corresponding ClusterRole for this HAGroupState. diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/metrics/HAGroupStoreMetricValues.java b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/metrics/HAGroupStoreMetricValues.java new file mode 100644 index 00000000000..389f0bac482 --- /dev/null +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/metrics/HAGroupStoreMetricValues.java @@ -0,0 +1,193 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.phoenix.jdbc.metrics; + +/** Immutable snapshot of HAGroupStore metrics used by tests. */ +public final class HAGroupStoreMetricValues { + + private final long localCacheHealthStatus; + private final long peerVisibilityStatus; + private final long degradedStandbyActive; + private final long currentLocalState; + private final long currentPeerState; + private final long localZkConnectionLostCount; + private final long peerBlindCount; + private final long degradedStandbyPresentedCount; + private final long invalidTransitionRejectedCount; + private final long systemTableSyncFailedCount; + private final long notificationListenerErrorCount; + private final long subscriberNotifyTimeCount; + private final long subscriberNotifyTimeMaxMs; + + private HAGroupStoreMetricValues(Builder builder) { + localCacheHealthStatus = builder.localCacheHealthStatus; + peerVisibilityStatus = builder.peerVisibilityStatus; + degradedStandbyActive = builder.degradedStandbyActive; + currentLocalState = builder.currentLocalState; + currentPeerState = builder.currentPeerState; + localZkConnectionLostCount = builder.localZkConnectionLostCount; + peerBlindCount = builder.peerBlindCount; + degradedStandbyPresentedCount = builder.degradedStandbyPresentedCount; + invalidTransitionRejectedCount = builder.invalidTransitionRejectedCount; + systemTableSyncFailedCount = builder.systemTableSyncFailedCount; + notificationListenerErrorCount = builder.notificationListenerErrorCount; + subscriberNotifyTimeCount = builder.subscriberNotifyTimeCount; + subscriberNotifyTimeMaxMs = builder.subscriberNotifyTimeMaxMs; + } + + public static Builder builder() { + return new Builder(); + } + + public long getLocalCacheHealthStatus() { + return localCacheHealthStatus; + } + + public long getPeerVisibilityStatus() { + return peerVisibilityStatus; + } + + public long getDegradedStandbyActive() { + return degradedStandbyActive; + } + + public long getCurrentLocalState() { + return currentLocalState; + } + + public long getCurrentPeerState() { + return currentPeerState; + } + + public long getLocalZkConnectionLostCount() { + return localZkConnectionLostCount; + } + + public long getPeerBlindCount() { + return peerBlindCount; + } + + public long getDegradedStandbyPresentedCount() { + return degradedStandbyPresentedCount; + } + + public long getInvalidTransitionRejectedCount() { + return invalidTransitionRejectedCount; + } + + public long getSystemTableSyncFailedCount() { + return systemTableSyncFailedCount; + } + + public long getNotificationListenerErrorCount() { + return notificationListenerErrorCount; + } + + public long getSubscriberNotifyTimeCount() { + return subscriberNotifyTimeCount; + } + + public long getSubscriberNotifyTimeMaxMs() { + return subscriberNotifyTimeMaxMs; + } + + public static final class Builder { + private long localCacheHealthStatus; + private long peerVisibilityStatus; + private long degradedStandbyActive; + private long currentLocalState; + private long currentPeerState; + private long localZkConnectionLostCount; + private long peerBlindCount; + private long degradedStandbyPresentedCount; + private long invalidTransitionRejectedCount; + private long systemTableSyncFailedCount; + private long notificationListenerErrorCount; + private long subscriberNotifyTimeCount; + private long subscriberNotifyTimeMaxMs; + + public Builder setLocalCacheHealthStatus(long value) { + localCacheHealthStatus = value; + return this; + } + + public Builder setPeerVisibilityStatus(long value) { + peerVisibilityStatus = value; + return this; + } + + public Builder setDegradedStandbyActive(long value) { + degradedStandbyActive = value; + return this; + } + + public Builder setCurrentLocalState(long value) { + currentLocalState = value; + return this; + } + + public Builder setCurrentPeerState(long value) { + currentPeerState = value; + return this; + } + + public Builder setLocalZkConnectionLostCount(long value) { + localZkConnectionLostCount = value; + return this; + } + + public Builder setPeerBlindCount(long value) { + peerBlindCount = value; + return this; + } + + public Builder setDegradedStandbyPresentedCount(long value) { + degradedStandbyPresentedCount = value; + return this; + } + + public Builder setInvalidTransitionRejectedCount(long value) { + invalidTransitionRejectedCount = value; + return this; + } + + public Builder setSystemTableSyncFailedCount(long value) { + systemTableSyncFailedCount = value; + return this; + } + + public Builder setNotificationListenerErrorCount(long value) { + notificationListenerErrorCount = value; + return this; + } + + public Builder setSubscriberNotifyTimeCount(long value) { + subscriberNotifyTimeCount = value; + return this; + } + + public Builder setSubscriberNotifyTimeMaxMs(long value) { + subscriberNotifyTimeMaxMs = value; + return this; + } + + public HAGroupStoreMetricValues build() { + return new HAGroupStoreMetricValues(this); + } + } +} diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/metrics/HAGroupStoreMetricsSource.java b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/metrics/HAGroupStoreMetricsSource.java new file mode 100644 index 00000000000..0734db4ec7a --- /dev/null +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/metrics/HAGroupStoreMetricsSource.java @@ -0,0 +1,106 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.phoenix.jdbc.metrics; + +import org.apache.hadoop.hbase.metrics.BaseSource; +import org.apache.phoenix.jdbc.HAGroupStoreRecord.HAGroupState; + +import org.apache.phoenix.thirdparty.com.google.common.annotations.VisibleForTesting; + +/** + * Metrics for one HAGroupStore group on a RegionServer. + *

+ * Lifecycle gauges and distributed reaction counters are best-effort diagnostics. They are not + * transactionally coordinated across client replacement or RegionServers and never feed HA + * decisions. + */ +public interface HAGroupStoreMetricsSource extends BaseSource { + + String METRICS_NAME = "HAGroupStore"; + String METRICS_CONTEXT = "phoenix"; + String METRICS_DESCRIPTION = "Metrics for HAGroupStore operations"; + String METRICS_JMX_CONTEXT = "RegionServer,sub=" + METRICS_NAME; + + String HA_GROUP_TAG_NAME = "haGroup"; + String HA_GROUP_TAG_DESC = "HA group name"; + + String LOCAL_CACHE_HEALTH_STATUS = "haGroupStoreLocalCacheHealthStatus"; + String LOCAL_CACHE_HEALTH_STATUS_DESC = + "Local HAGroupStore cache health status: 0 healthy, non-zero unhealthy"; + String PEER_VISIBILITY_STATUS = "haGroupStorePeerVisibilityStatus"; + String PEER_VISIBILITY_STATUS_DESC = + "Configured peer visibility status: 0 visible or unconfigured, non-zero blind"; + String DEGRADED_STANDBY_ACTIVE = "haGroupStoreDegradedStandbyActive"; + String DEGRADED_STANDBY_ACTIVE_DESC = + "Whether the local peer-blind degraded standby overlay is active"; + String CURRENT_LOCAL_STATE = "haGroupStoreCurrentLocalState"; + String CURRENT_LOCAL_STATE_DESC = "Current raw local HA group state code"; + String CURRENT_PEER_STATE = "haGroupStoreCurrentPeerState"; + String CURRENT_PEER_STATE_DESC = "Current peer HA group state code"; + + String LOCAL_ZK_CONNECTION_LOST_COUNT = "haGroupStoreLocalZkConnectionLostCount"; + String LOCAL_ZK_CONNECTION_LOST_COUNT_DESC = "Number of local ZooKeeper connection losses"; + String PEER_BLIND_COUNT = "haGroupStorePeerBlindCount"; + String PEER_BLIND_COUNT_DESC = "Number of transitions to peer-blind"; + String DEGRADED_STANDBY_PRESENTED_COUNT = "haGroupStoreDegradedStandbyPresentedCount"; + String DEGRADED_STANDBY_PRESENTED_COUNT_DESC = "Number of degraded standby overlay presentations"; + String INVALID_TRANSITION_REJECTED_COUNT = "haGroupStoreInvalidTransitionRejectedCount"; + String INVALID_TRANSITION_REJECTED_COUNT_DESC = + "Number of invalid HA group state transitions rejected"; + String SYSTEM_TABLE_SYNC_FAILED_COUNT = "haGroupStoreSystemTableSyncFailedCount"; + String SYSTEM_TABLE_SYNC_FAILED_COUNT_DESC = + "Number of failed SYSTEM.HA_GROUP synchronization writes"; + String NOTIFICATION_LISTENER_ERROR_COUNT = "haGroupStoreNotificationListenerErrorCount"; + String NOTIFICATION_LISTENER_ERROR_COUNT_DESC = "Number of HA group notification listener errors"; + + // MutableTimeHistogram capitalizes its exported basename. The logical registry key remains + // lowercase, while JMX/Metrics2 consumers use SUBSCRIBER_NOTIFY_TIME_MS_EXPORTED_BASE plus the + // standard histogram suffixes. + String SUBSCRIBER_NOTIFY_TIME_MS = "haGroupStoreSubscriberNotifyTimeMs"; + String SUBSCRIBER_NOTIFY_TIME_MS_EXPORTED_BASE = "HaGroupStoreSubscriberNotifyTimeMs"; + String SUBSCRIBER_NOTIFY_TIME_MS_DESC = + "Time spent synchronously notifying HA group subscribers in whole milliseconds; " + + "durations below one millisecond are recorded as zero"; + + void setLocalCacheHealthy(boolean healthy); + + void setPeerVisible(boolean visible); + + void setDegradedStandbyActive(boolean active); + + void setCurrentLocalState(HAGroupState state); + + void setCurrentPeerState(HAGroupState state); + + void incrementLocalZkConnectionLostCount(); + + void incrementPeerBlindCount(); + + void incrementDegradedStandbyPresentedCount(); + + void incrementInvalidTransitionRejectedCount(); + + void incrementSystemTableSyncFailedCount(); + + void incrementNotificationListenerErrorCount(); + + void updateSubscriberNotifyTime(long elapsedTimeNs); + + @VisibleForTesting + HAGroupStoreMetricValues getCurrentMetricValues(); +} diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/metrics/HAGroupStoreMetricsSourceFactory.java b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/metrics/HAGroupStoreMetricsSourceFactory.java new file mode 100644 index 00000000000..c1e884be44b --- /dev/null +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/metrics/HAGroupStoreMetricsSourceFactory.java @@ -0,0 +1,34 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.phoenix.jdbc.metrics; + +import java.util.concurrent.ConcurrentHashMap; + +/** Factory for process-lifetime HAGroupStore metric sources. */ +public final class HAGroupStoreMetricsSourceFactory { + + private static final ConcurrentHashMap SOURCES = + new ConcurrentHashMap<>(); + + private HAGroupStoreMetricsSourceFactory() { + } + + public static HAGroupStoreMetricsSource getInstanceForHAGroup(String haGroupName) { + return SOURCES.computeIfAbsent(haGroupName, HAGroupStoreMetricsSourceImpl::new); + } +} diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/metrics/HAGroupStoreMetricsSourceImpl.java b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/metrics/HAGroupStoreMetricsSourceImpl.java new file mode 100644 index 00000000000..f7d731fe763 --- /dev/null +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/metrics/HAGroupStoreMetricsSourceImpl.java @@ -0,0 +1,171 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.phoenix.jdbc.metrics; + +import java.util.concurrent.TimeUnit; +import javax.management.ObjectName; +import org.apache.hadoop.hbase.metrics.BaseSourceImpl; +import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem; +import org.apache.hadoop.metrics2.lib.Interns; +import org.apache.hadoop.metrics2.lib.MutableFastCounter; +import org.apache.hadoop.metrics2.lib.MutableGaugeLong; +import org.apache.hadoop.metrics2.lib.MutableTimeHistogram; +import org.apache.phoenix.jdbc.HAGroupStoreRecord.HAGroupState; + +import org.apache.phoenix.thirdparty.com.google.common.annotations.VisibleForTesting; + +/** Hadoop Metrics2 implementation for one HAGroupStore group. */ +public class HAGroupStoreMetricsSourceImpl extends BaseSourceImpl + implements HAGroupStoreMetricsSource { + + private final MutableGaugeLong localCacheHealthStatus; + private final MutableGaugeLong peerVisibilityStatus; + private final MutableGaugeLong degradedStandbyActive; + private final MutableGaugeLong currentLocalState; + private final MutableGaugeLong currentPeerState; + + private final MutableFastCounter localZkConnectionLostCount; + private final MutableFastCounter peerBlindCount; + private final MutableFastCounter degradedStandbyPresentedCount; + private final MutableFastCounter invalidTransitionRejectedCount; + private final MutableFastCounter systemTableSyncFailedCount; + private final MutableFastCounter notificationListenerErrorCount; + + private final MutableTimeHistogram subscriberNotifyTimeMs; + + public HAGroupStoreMetricsSourceImpl(String haGroupName) { + this(METRICS_NAME, METRICS_DESCRIPTION, METRICS_CONTEXT, METRICS_JMX_CONTEXT, haGroupName); + } + + HAGroupStoreMetricsSourceImpl(String metricsName, String metricsDescription, + String metricsContext, String metricsJmxContext, String haGroupName) { + super(metricsName, metricsDescription, metricsContext, + metricsJmxContext + ",haGroup=" + ObjectName.quote(haGroupName)); + getMetricsRegistry().tag(Interns.info(HA_GROUP_TAG_NAME, HA_GROUP_TAG_DESC), haGroupName); + + localCacheHealthStatus = getMetricsRegistry().newGauge(LOCAL_CACHE_HEALTH_STATUS, + LOCAL_CACHE_HEALTH_STATUS_DESC, 1L); + peerVisibilityStatus = getMetricsRegistry().newGauge(PEER_VISIBILITY_STATUS, + PEER_VISIBILITY_STATUS_DESC, 1L); + degradedStandbyActive = + getMetricsRegistry().newGauge(DEGRADED_STANDBY_ACTIVE, DEGRADED_STANDBY_ACTIVE_DESC, 0L); + currentLocalState = getMetricsRegistry().newGauge(CURRENT_LOCAL_STATE, CURRENT_LOCAL_STATE_DESC, + (long) HAGroupState.UNKNOWN.getMetricCode()); + currentPeerState = getMetricsRegistry().newGauge(CURRENT_PEER_STATE, CURRENT_PEER_STATE_DESC, + (long) HAGroupState.UNKNOWN.getMetricCode()); + + localZkConnectionLostCount = getMetricsRegistry().newCounter(LOCAL_ZK_CONNECTION_LOST_COUNT, + LOCAL_ZK_CONNECTION_LOST_COUNT_DESC, 0L); + peerBlindCount = getMetricsRegistry().newCounter(PEER_BLIND_COUNT, PEER_BLIND_COUNT_DESC, 0L); + degradedStandbyPresentedCount = getMetricsRegistry() + .newCounter(DEGRADED_STANDBY_PRESENTED_COUNT, DEGRADED_STANDBY_PRESENTED_COUNT_DESC, 0L); + invalidTransitionRejectedCount = getMetricsRegistry() + .newCounter(INVALID_TRANSITION_REJECTED_COUNT, INVALID_TRANSITION_REJECTED_COUNT_DESC, 0L); + systemTableSyncFailedCount = getMetricsRegistry().newCounter(SYSTEM_TABLE_SYNC_FAILED_COUNT, + SYSTEM_TABLE_SYNC_FAILED_COUNT_DESC, 0L); + notificationListenerErrorCount = getMetricsRegistry() + .newCounter(NOTIFICATION_LISTENER_ERROR_COUNT, NOTIFICATION_LISTENER_ERROR_COUNT_DESC, 0L); + + subscriberNotifyTimeMs = getMetricsRegistry().newTimeHistogram(SUBSCRIBER_NOTIFY_TIME_MS, + SUBSCRIBER_NOTIFY_TIME_MS_DESC); + } + + @Override + public void setLocalCacheHealthy(boolean value) { + localCacheHealthStatus.set(value ? 0L : 1L); + } + + @Override + public void setPeerVisible(boolean value) { + peerVisibilityStatus.set(value ? 0L : 1L); + } + + @Override + public void setDegradedStandbyActive(boolean value) { + degradedStandbyActive.set(value ? 1L : 0L); + } + + @Override + public void setCurrentLocalState(HAGroupState state) { + HAGroupState currentState = state != null ? state : HAGroupState.UNKNOWN; + currentLocalState.set(currentState.getMetricCode()); + } + + @Override + public void setCurrentPeerState(HAGroupState state) { + HAGroupState currentState = state != null ? state : HAGroupState.UNKNOWN; + currentPeerState.set(currentState.getMetricCode()); + } + + @Override + public void incrementLocalZkConnectionLostCount() { + localZkConnectionLostCount.incr(); + } + + @Override + public void incrementPeerBlindCount() { + peerBlindCount.incr(); + } + + @Override + public void incrementDegradedStandbyPresentedCount() { + degradedStandbyPresentedCount.incr(); + } + + @Override + public void incrementInvalidTransitionRejectedCount() { + invalidTransitionRejectedCount.incr(); + } + + @Override + public void incrementSystemTableSyncFailedCount() { + systemTableSyncFailedCount.incr(); + } + + @Override + public void incrementNotificationListenerErrorCount() { + notificationListenerErrorCount.incr(); + } + + @Override + public void updateSubscriberNotifyTime(long elapsedTimeNs) { + subscriberNotifyTimeMs.add(TimeUnit.NANOSECONDS.toMillis(elapsedTimeNs)); + } + + @Override + public HAGroupStoreMetricValues getCurrentMetricValues() { + return HAGroupStoreMetricValues.builder() + .setLocalCacheHealthStatus(localCacheHealthStatus.value()) + .setPeerVisibilityStatus(peerVisibilityStatus.value()) + .setDegradedStandbyActive(degradedStandbyActive.value()) + .setCurrentLocalState(currentLocalState.value()).setCurrentPeerState(currentPeerState.value()) + .setLocalZkConnectionLostCount(localZkConnectionLostCount.value()) + .setPeerBlindCount(peerBlindCount.value()) + .setDegradedStandbyPresentedCount(degradedStandbyPresentedCount.value()) + .setInvalidTransitionRejectedCount(invalidTransitionRejectedCount.value()) + .setSystemTableSyncFailedCount(systemTableSyncFailedCount.value()) + .setNotificationListenerErrorCount(notificationListenerErrorCount.value()) + .setSubscriberNotifyTimeCount(subscriberNotifyTimeMs.getCount()) + .setSubscriberNotifyTimeMaxMs(subscriberNotifyTimeMs.getMax()).build(); + } + + @VisibleForTesting + public void unregisterForTesting() { + DefaultMetricsSystem.instance().unregisterSource(metricsJmxContext); + } +} diff --git a/phoenix-core/src/it/java/org/apache/phoenix/jdbc/HAGroupStoreMetricsIT.java b/phoenix-core/src/it/java/org/apache/phoenix/jdbc/HAGroupStoreMetricsIT.java new file mode 100644 index 00000000000..5a9d4a501b9 --- /dev/null +++ b/phoenix-core/src/it/java/org/apache/phoenix/jdbc/HAGroupStoreMetricsIT.java @@ -0,0 +1,316 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.phoenix.jdbc; + +import static org.apache.phoenix.jdbc.HAGroupStoreClient.ZK_CONSISTENT_HA_GROUP_RECORD_NAMESPACE; +import static org.apache.phoenix.jdbc.PhoenixHAAdmin.getLocalZkUrl; +import static org.apache.phoenix.jdbc.PhoenixHAAdmin.toPath; +import static org.apache.phoenix.replication.reader.ReplicationLogReplayService.PHOENIX_REPLICATION_REPLAY_ENABLED; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; + +import java.io.IOException; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Predicate; +import org.apache.commons.lang3.tuple.Pair; +import org.apache.hadoop.hbase.client.Admin; +import org.apache.phoenix.end2end.NeedsOwnMiniClusterTest; +import org.apache.phoenix.exception.InvalidClusterRoleTransitionException; +import org.apache.phoenix.jdbc.metrics.HAGroupStoreMetricValues; +import org.apache.phoenix.jdbc.metrics.HAGroupStoreMetricsSourceFactory; +import org.apache.phoenix.util.HAGroupStoreTestUtil; +import org.apache.zookeeper.data.Stat; +import org.junit.After; +import org.junit.AfterClass; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Rule; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.rules.TestName; + +@Category(NeedsOwnMiniClusterTest.class) +public class HAGroupStoreMetricsIT extends HABaseIT { + + private static final long EVENT_TIMEOUT_MS = 60000L; + + @Rule + public final TestName testName = new TestName(); + + private PhoenixHAAdmin localAdmin; + private PhoenixHAAdmin peerAdmin; + private String zkUrl; + private String peerZkUrl; + private String masterUrl; + private String peerMasterUrl; + private HAGroupStoreClient clientToClose; + + @BeforeClass + public static synchronized void setUpClass() throws Exception { + conf1.setBoolean(PHOENIX_REPLICATION_REPLAY_ENABLED, false); + conf2.setBoolean(PHOENIX_REPLICATION_REPLAY_ENABLED, false); + CLUSTERS.start(); + } + + @AfterClass + public static synchronized void tearDownClass() throws Exception { + CLUSTERS.close(); + } + + @Before + public void setUp() throws Exception { + zkUrl = getLocalZkUrl(CLUSTERS.getHBaseCluster1().getConfiguration()); + peerZkUrl = getLocalZkUrl(CLUSTERS.getHBaseCluster2().getConfiguration()); + masterUrl = CLUSTERS.getMasterAddress1(); + peerMasterUrl = CLUSTERS.getMasterAddress2(); + localAdmin = new PhoenixHAAdmin(CLUSTERS.getHBaseCluster1().getConfiguration(), + ZK_CONSISTENT_HA_GROUP_RECORD_NAMESPACE); + peerAdmin = new PhoenixHAAdmin(CLUSTERS.getHBaseCluster2().getConfiguration(), + ZK_CONSISTENT_HA_GROUP_RECORD_NAMESPACE); + deleteZNodes(); + HAGroupStoreTestUtil.deleteHAGroupRecordInSystemTable(group(), zkUrl); + HAGroupStoreTestUtil.upsertHAGroupRecordInSystemTable(group(), zkUrl, peerZkUrl, masterUrl, + peerMasterUrl, ClusterRoleRecord.ClusterRole.ACTIVE, ClusterRoleRecord.ClusterRole.STANDBY, + null, CLUSTERS.getHdfsUrl1(), CLUSTERS.getHdfsUrl2()); + } + + @After + public void tearDown() throws Exception { + if (clientToClose != null) { + clientToClose.close(); + } + deleteZNodes(); + HAGroupStoreTestUtil.deleteHAGroupRecordInSystemTable(group(), zkUrl); + localAdmin.close(); + peerAdmin.close(); + } + + @Test + public void testInitialAndLocalStateMetrics() throws Exception { + writeLocal(HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC); + writePeer(HAGroupStoreRecord.HAGroupState.STANDBY); + HAGroupStoreClient client = localClient(); + + awaitMetrics(values -> values.getLocalCacheHealthStatus() == 0 + && values.getCurrentLocalState() + == HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC.getMetricCode() + && values.getCurrentPeerState() == HAGroupStoreRecord.HAGroupState.STANDBY.getMetricCode()); + + assertEquals(0L, + client.setHAGroupStatusIfNeeded(HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC_TO_STANDBY)); + awaitMetrics(values -> values.getCurrentLocalState() + == HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC_TO_STANDBY.getMetricCode()); + + assertEquals(0L, client.setHAGroupStatusIfNeeded(HAGroupStoreRecord.HAGroupState.STANDBY)); + awaitMetrics(values -> values.getCurrentLocalState() + == HAGroupStoreRecord.HAGroupState.STANDBY.getMetricCode()); + } + + @Test + public void testLocalZkHealthMetrics() throws Exception { + writeLocal(HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC); + HAGroupStoreClient client = localClient(); + awaitMetrics(values -> values.getLocalCacheHealthStatus() == 0); + HAGroupStoreMetricValues before = metrics(); + int port = Integer.parseInt( + CLUSTERS.getHBaseCluster1().getConfiguration().get("hbase.zookeeper.property.clientPort")); + + CLUSTERS.getHBaseCluster1().shutdownMiniZKCluster(); + try { + awaitMetrics(values -> values.getLocalCacheHealthStatus() == 1 + && values.getLocalZkConnectionLostCount() == before.getLocalZkConnectionLostCount() + 1); + assertThrows(IOException.class, client::getHAGroupStoreRecord); + } finally { + CLUSTERS.getHBaseCluster1().startMiniZKCluster(1, port); + } + awaitMetrics(values -> values.getLocalCacheHealthStatus() == 0); + } + + @Test + public void testPeerVisibilityAndDegradedMetrics() throws Exception { + writeLocal(HAGroupStoreRecord.HAGroupState.STANDBY); + writePeer(HAGroupStoreRecord.HAGroupState.STANDBY); + localClient(); + awaitMetrics(values -> values.getPeerVisibilityStatus() == 0 + && values.getCurrentPeerState() == HAGroupStoreRecord.HAGroupState.STANDBY.getMetricCode()); + HAGroupStoreMetricValues before = metrics(); + int port = Integer.parseInt( + CLUSTERS.getHBaseCluster2().getConfiguration().get("hbase.zookeeper.property.clientPort")); + + CLUSTERS.getHBaseCluster2().shutdownMiniZKCluster(); + try { + awaitMetrics(values -> values.getPeerVisibilityStatus() == 1 + && values.getCurrentPeerState() == HAGroupStoreRecord.HAGroupState.UNKNOWN.getMetricCode() + && values.getDegradedStandbyActive() == 1 + && values.getPeerBlindCount() == before.getPeerBlindCount() + 1 + && values.getDegradedStandbyPresentedCount() + == before.getDegradedStandbyPresentedCount() + 1); + } finally { + CLUSTERS.getHBaseCluster2().startMiniZKCluster(1, port); + } + awaitMetrics(values -> values.getPeerVisibilityStatus() == 0 + && values.getCurrentPeerState() == HAGroupStoreRecord.HAGroupState.STANDBY.getMetricCode() + && values.getDegradedStandbyActive() == 0); + } + + @Test + public void testPeerReconfigurationMetrics() throws Exception { + writeLocal(HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC); + writePeer(HAGroupStoreRecord.HAGroupState.STANDBY); + localClient(); + awaitMetrics(values -> values.getCurrentPeerState() + == HAGroupStoreRecord.HAGroupState.STANDBY.getMetricCode()); + + writeLocal(HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC, null); + awaitMetrics(values -> values.getPeerVisibilityStatus() == 0 + && values.getCurrentPeerState() == HAGroupStoreRecord.HAGroupState.UNKNOWN.getMetricCode()); + + writeLocal(HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC); + awaitMetrics(values -> values.getCurrentPeerState() + == HAGroupStoreRecord.HAGroupState.STANDBY.getMetricCode()); + + writeLocal(HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC, "invalidURL"); + awaitMetrics(values -> values.getPeerVisibilityStatus() == 1 + && values.getCurrentPeerState() == HAGroupStoreRecord.HAGroupState.UNKNOWN.getMetricCode()); + + writeLocal(HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC, null); + awaitMetrics(values -> values.getPeerVisibilityStatus() == 0 + && values.getCurrentPeerState() == HAGroupStoreRecord.HAGroupState.UNKNOWN.getMetricCode()); + } + + @Test + public void testInvalidTransitionMetric() throws Exception { + writeLocal(HAGroupStoreRecord.HAGroupState.ACTIVE_NOT_IN_SYNC); + HAGroupStoreClient client = localClient(); + HAGroupStoreMetricValues before = metrics(); + + assertThrows(InvalidClusterRoleTransitionException.class, + () -> client.setHAGroupStatusIfNeeded(HAGroupStoreRecord.HAGroupState.STANDBY)); + assertEquals(before.getInvalidTransitionRejectedCount() + 1, + metrics().getInvalidTransitionRejectedCount()); + } + + @Test + public void testSystemTableSyncFailureMetric() throws Exception { + writeLocal(HAGroupStoreRecord.HAGroupState.STANDBY); + HAGroupStoreClient client = localClient(); + awaitMetrics(values -> values.getCurrentLocalState() + == HAGroupStoreRecord.HAGroupState.STANDBY.getMetricCode()); + HAGroupStoreMetricValues before = metrics(); + Admin admin = CLUSTERS.getHBaseCluster1().getAdmin(); + + admin.disableTable(PhoenixDatabaseMetaData.SYSTEM_HA_GROUP_HBASE_TABLE_NAME); + try { + assertEquals(0L, + client.setHAGroupStatusIfNeeded(HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE)); + assertEquals(before.getSystemTableSyncFailedCount() + 1, + metrics().getSystemTableSyncFailedCount()); + Pair updated = localAdmin.getHAGroupStoreRecordInZooKeeper(group()); + assertEquals(HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE, + updated.getLeft().getHAGroupState()); + } finally { + admin.enableTable(PhoenixDatabaseMetaData.SYSTEM_HA_GROUP_HBASE_TABLE_NAME); + } + } + + @Test + public void testListenerMetrics() throws Exception { + writeLocal(HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC); + HAGroupStoreClient client = localClient(); + client.subscribeToTargetState(HAGroupStoreRecord.HAGroupState.ACTIVE_NOT_IN_SYNC, + ClusterType.LOCAL, (group, from, to, mtime, cluster, lastSync) -> { + throw new IllegalStateException("expected test failure"); + }); + AtomicInteger successfulListeners = new AtomicInteger(); + client.subscribeToTargetState(HAGroupStoreRecord.HAGroupState.ACTIVE_NOT_IN_SYNC, + ClusterType.LOCAL, + (group, from, to, mtime, cluster, lastSync) -> successfulListeners.incrementAndGet()); + HAGroupStoreMetricValues before = metrics(); + + writeLocal(HAGroupStoreRecord.HAGroupState.ACTIVE_NOT_IN_SYNC); + awaitMetrics(values -> values.getNotificationListenerErrorCount() + == before.getNotificationListenerErrorCount() + 1 + && values.getSubscriberNotifyTimeCount() == before.getSubscriberNotifyTimeCount() + 1 + && successfulListeners.get() == 1); + } + + private String group() { + return testName.getMethodName(); + } + + private HAGroupStoreClient localClient() { + HAGroupStoreClient client = HAGroupStoreClient + .getInstanceForZkUrl(CLUSTERS.getHBaseCluster1().getConfiguration(), group(), zkUrl); + if (client != null) { + clientToClose = client; + } + return client; + } + + private HAGroupStoreMetricValues metrics() { + return HAGroupStoreMetricsSourceFactory.getInstanceForHAGroup(group()).getCurrentMetricValues(); + } + + private void awaitMetrics(Predicate condition) throws Exception { + long deadline = System.currentTimeMillis() + EVENT_TIMEOUT_MS; + while (!condition.test(metrics()) && System.currentTimeMillis() < deadline) { + Thread.sleep(100L); + } + assertTrue("Metric condition not met for " + group(), condition.test(metrics())); + } + + private HAGroupStoreRecord localRecord(HAGroupStoreRecord.HAGroupState state, String peerUrl) { + return new HAGroupStoreRecord(HAGroupStoreRecord.DEFAULT_PROTOCOL_VERSION, group(), state, 0L, + HighAvailabilityPolicy.FAILOVER.toString(), peerUrl, masterUrl, peerMasterUrl, + CLUSTERS.getHdfsUrl1(), CLUSTERS.getHdfsUrl2(), 0L); + } + + private HAGroupStoreRecord peerRecord(HAGroupStoreRecord.HAGroupState state) { + return new HAGroupStoreRecord(HAGroupStoreRecord.DEFAULT_PROTOCOL_VERSION, group(), state, 0L, + HighAvailabilityPolicy.FAILOVER.toString(), zkUrl, peerMasterUrl, masterUrl, + CLUSTERS.getHdfsUrl2(), CLUSTERS.getHdfsUrl1(), 0L); + } + + private void writeLocal(HAGroupStoreRecord.HAGroupState state) throws Exception { + writeLocal(state, peerZkUrl); + } + + private void writeLocal(HAGroupStoreRecord.HAGroupState state, String peerUrl) throws Exception { + createOrUpdate(localAdmin, localRecord(state, peerUrl)); + } + + private void writePeer(HAGroupStoreRecord.HAGroupState state) throws Exception { + createOrUpdate(peerAdmin, peerRecord(state)); + } + + private void createOrUpdate(PhoenixHAAdmin admin, HAGroupStoreRecord record) throws Exception { + String path = toPath(group()); + if (admin.getCurator().checkExists().forPath(path) == null) { + admin.createHAGroupStoreRecordInZooKeeper(record); + return; + } + Pair current = admin.getHAGroupStoreRecordInZooKeeper(group()); + admin.updateHAGroupStoreRecordInZooKeeper(group(), record, current.getRight().getVersion()); + } + + private void deleteZNodes() throws Exception { + localAdmin.getCurator().delete().quietly().forPath(toPath(group())); + peerAdmin.getCurator().delete().quietly().forPath(toPath(group())); + } +} diff --git a/phoenix-core/src/test/java/org/apache/phoenix/jdbc/HAGroupStoreRecordTest.java b/phoenix-core/src/test/java/org/apache/phoenix/jdbc/HAGroupStoreRecordTest.java index 3fb77d4bd4d..ee451d1207d 100644 --- a/phoenix-core/src/test/java/org/apache/phoenix/jdbc/HAGroupStoreRecordTest.java +++ b/phoenix-core/src/test/java/org/apache/phoenix/jdbc/HAGroupStoreRecordTest.java @@ -252,6 +252,25 @@ public void testConstructorWithNullHAGroupState() { } // Tests for HAGroupState enum + @Test + public void testHAGroupStateMetricContract() { + assertEquals(1, HAGroupStoreRecord.HAGroupState.ABORT_TO_ACTIVE_IN_SYNC.getMetricCode()); + assertEquals(2, HAGroupStoreRecord.HAGroupState.ABORT_TO_ACTIVE_NOT_IN_SYNC.getMetricCode()); + assertEquals(3, HAGroupStoreRecord.HAGroupState.ABORT_TO_STANDBY.getMetricCode()); + assertEquals(4, HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC.getMetricCode()); + assertEquals(5, HAGroupStoreRecord.HAGroupState.ACTIVE_NOT_IN_SYNC.getMetricCode()); + assertEquals(6, HAGroupStoreRecord.HAGroupState.ACTIVE_NOT_IN_SYNC_TO_STANDBY.getMetricCode()); + assertEquals(7, + HAGroupStoreRecord.HAGroupState.ACTIVE_NOT_IN_SYNC_WITH_OFFLINE_PEER.getMetricCode()); + assertEquals(8, HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC_TO_STANDBY.getMetricCode()); + assertEquals(9, HAGroupStoreRecord.HAGroupState.ACTIVE_WITH_OFFLINE_PEER.getMetricCode()); + assertEquals(10, HAGroupStoreRecord.HAGroupState.DEGRADED_STANDBY.getMetricCode()); + assertEquals(11, HAGroupStoreRecord.HAGroupState.OFFLINE.getMetricCode()); + assertEquals(12, HAGroupStoreRecord.HAGroupState.STANDBY.getMetricCode()); + assertEquals(13, HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE.getMetricCode()); + assertEquals(0, HAGroupStoreRecord.HAGroupState.UNKNOWN.getMetricCode()); + } + @Test public void testHAGroupStateGetClusterRole() { // Test that each HAGroupState maps to the correct ClusterRole diff --git a/phoenix-core/src/test/java/org/apache/phoenix/jdbc/metrics/HAGroupStoreMetricsSourceImplTest.java b/phoenix-core/src/test/java/org/apache/phoenix/jdbc/metrics/HAGroupStoreMetricsSourceImplTest.java new file mode 100644 index 00000000000..9eabdd97b9c --- /dev/null +++ b/phoenix-core/src/test/java/org/apache/phoenix/jdbc/metrics/HAGroupStoreMetricsSourceImplTest.java @@ -0,0 +1,197 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.phoenix.jdbc.metrics; + +import static org.apache.phoenix.jdbc.metrics.HAGroupStoreMetricsSource.CURRENT_LOCAL_STATE; +import static org.apache.phoenix.jdbc.metrics.HAGroupStoreMetricsSource.CURRENT_PEER_STATE; +import static org.apache.phoenix.jdbc.metrics.HAGroupStoreMetricsSource.DEGRADED_STANDBY_ACTIVE; +import static org.apache.phoenix.jdbc.metrics.HAGroupStoreMetricsSource.DEGRADED_STANDBY_PRESENTED_COUNT; +import static org.apache.phoenix.jdbc.metrics.HAGroupStoreMetricsSource.HA_GROUP_TAG_NAME; +import static org.apache.phoenix.jdbc.metrics.HAGroupStoreMetricsSource.INVALID_TRANSITION_REJECTED_COUNT; +import static org.apache.phoenix.jdbc.metrics.HAGroupStoreMetricsSource.LOCAL_CACHE_HEALTH_STATUS; +import static org.apache.phoenix.jdbc.metrics.HAGroupStoreMetricsSource.LOCAL_ZK_CONNECTION_LOST_COUNT; +import static org.apache.phoenix.jdbc.metrics.HAGroupStoreMetricsSource.NOTIFICATION_LISTENER_ERROR_COUNT; +import static org.apache.phoenix.jdbc.metrics.HAGroupStoreMetricsSource.PEER_BLIND_COUNT; +import static org.apache.phoenix.jdbc.metrics.HAGroupStoreMetricsSource.PEER_VISIBILITY_STATUS; +import static org.apache.phoenix.jdbc.metrics.HAGroupStoreMetricsSource.SUBSCRIBER_NOTIFY_TIME_MS; +import static org.apache.phoenix.jdbc.metrics.HAGroupStoreMetricsSource.SUBSCRIBER_NOTIFY_TIME_MS_EXPORTED_BASE; +import static org.apache.phoenix.jdbc.metrics.HAGroupStoreMetricsSource.SYSTEM_TABLE_SYNC_FAILED_COUNT; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotEquals; +import static org.junit.Assert.assertNotSame; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; + +import java.util.Arrays; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; +import javax.management.ObjectName; +import org.apache.hadoop.metrics2.AbstractMetric; +import org.apache.hadoop.metrics2.impl.MetricsCollectorImpl; +import org.apache.hadoop.metrics2.impl.MetricsRecordImpl; +import org.apache.hadoop.metrics2.lib.MutableFastCounter; +import org.apache.hadoop.metrics2.lib.MutableGaugeLong; +import org.apache.hadoop.metrics2.lib.MutableMetric; +import org.apache.hadoop.metrics2.lib.MutableTimeHistogram; +import org.apache.phoenix.jdbc.HAGroupStoreRecord.HAGroupState; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +public class HAGroupStoreMetricsSourceImplTest { + + private HAGroupStoreMetricsSourceImpl source; + + @Before + public void setUp() { + source = new HAGroupStoreMetricsSourceImpl("metrics-test-" + System.nanoTime()); + } + + @After + public void tearDown() { + source.unregisterForTesting(); + } + + @Test + public void testMetricNamesAreUniqueAndPrefixed() { + Set metricNames = new HashSet<>(Arrays.asList(LOCAL_CACHE_HEALTH_STATUS, + PEER_VISIBILITY_STATUS, DEGRADED_STANDBY_ACTIVE, CURRENT_LOCAL_STATE, CURRENT_PEER_STATE, + LOCAL_ZK_CONNECTION_LOST_COUNT, PEER_BLIND_COUNT, DEGRADED_STANDBY_PRESENTED_COUNT, + INVALID_TRANSITION_REJECTED_COUNT, SYSTEM_TABLE_SYNC_FAILED_COUNT, + NOTIFICATION_LISTENER_ERROR_COUNT, SUBSCRIBER_NOTIFY_TIME_MS)); + + assertEquals(12, metricNames.size()); + for (String metricName : metricNames) { + assertTrue(metricName, metricName.startsWith("haGroupStore")); + } + } + + @Test + public void testMetricRegistryContract() { + Map metrics = source.getMetricsRegistry().getMetricsMap(); + assertEquals(12, metrics.size()); + + for (String gauge : Arrays.asList(LOCAL_CACHE_HEALTH_STATUS, PEER_VISIBILITY_STATUS, + DEGRADED_STANDBY_ACTIVE, CURRENT_LOCAL_STATE, CURRENT_PEER_STATE)) { + assertTrue(gauge, metrics.get(gauge) instanceof MutableGaugeLong); + } + for (String counter : Arrays.asList(LOCAL_ZK_CONNECTION_LOST_COUNT, PEER_BLIND_COUNT, + DEGRADED_STANDBY_PRESENTED_COUNT, INVALID_TRANSITION_REJECTED_COUNT, + SYSTEM_TABLE_SYNC_FAILED_COUNT, NOTIFICATION_LISTENER_ERROR_COUNT)) { + assertTrue(counter, metrics.get(counter) instanceof MutableFastCounter); + } + assertTrue(metrics.get(SUBSCRIBER_NOTIFY_TIME_MS) instanceof MutableTimeHistogram); + } + + @Test + public void testMetricsSnapshot() { + source.setLocalCacheHealthy(true); + source.setPeerVisible(true); + source.setDegradedStandbyActive(true); + source.setCurrentLocalState(HAGroupState.STANDBY_TO_ACTIVE); + source.setCurrentPeerState(HAGroupState.ACTIVE_IN_SYNC); + source.incrementLocalZkConnectionLostCount(); + source.incrementPeerBlindCount(); + source.incrementDegradedStandbyPresentedCount(); + source.incrementInvalidTransitionRejectedCount(); + source.incrementSystemTableSyncFailedCount(); + source.incrementNotificationListenerErrorCount(); + source.updateSubscriberNotifyTime(5_000_000L); + + HAGroupStoreMetricValues values = source.getCurrentMetricValues(); + assertEquals(0L, values.getLocalCacheHealthStatus()); + assertEquals(0L, values.getPeerVisibilityStatus()); + assertEquals(1L, values.getDegradedStandbyActive()); + assertEquals(HAGroupState.STANDBY_TO_ACTIVE.getMetricCode(), values.getCurrentLocalState()); + assertEquals(HAGroupState.ACTIVE_IN_SYNC.getMetricCode(), values.getCurrentPeerState()); + assertEquals(1L, values.getLocalZkConnectionLostCount()); + assertEquals(1L, values.getPeerBlindCount()); + assertEquals(1L, values.getDegradedStandbyPresentedCount()); + assertEquals(1L, values.getInvalidTransitionRejectedCount()); + assertEquals(1L, values.getSystemTableSyncFailedCount()); + assertEquals(1L, values.getNotificationListenerErrorCount()); + assertEquals(1L, values.getSubscriberNotifyTimeCount()); + assertEquals(5L, values.getSubscriberNotifyTimeMaxMs()); + } + + @Test + public void testSubMillisecondNotifyTimeRoundsDownToZero() { + source.updateSubscriberNotifyTime(999_999L); + + HAGroupStoreMetricValues values = source.getCurrentMetricValues(); + assertEquals(1L, values.getSubscriberNotifyTimeCount()); + assertEquals(0L, values.getSubscriberNotifyTimeMaxMs()); + } + + @Test + public void testExportedHistogramNames() { + source.updateSubscriberNotifyTime(5_000_000L); + MetricsCollectorImpl collector = new MetricsCollectorImpl(); + source.getMetrics(collector, true); + + Set exportedNames = new HashSet<>(); + for (MetricsRecordImpl record : collector.getRecords()) { + for (AbstractMetric metric : record.metrics()) { + exportedNames.add(metric.name()); + } + } + + assertTrue(exportedNames.contains(SUBSCRIBER_NOTIFY_TIME_MS_EXPORTED_BASE + "_num_ops")); + assertTrue(exportedNames.contains(SUBSCRIBER_NOTIFY_TIME_MS_EXPORTED_BASE + "_max")); + assertTrue( + exportedNames.contains(SUBSCRIBER_NOTIFY_TIME_MS_EXPORTED_BASE + "_99th_percentile")); + assertFalse(exportedNames.contains(SUBSCRIBER_NOTIFY_TIME_MS + "_num_ops")); + } + + @Test + public void testFactoryCachesSourceByGroup() { + String group = "factory-test-" + System.nanoTime(); + assertSame(HAGroupStoreMetricsSourceFactory.getInstanceForHAGroup(group), + HAGroupStoreMetricsSourceFactory.getInstanceForHAGroup(group)); + } + + @Test + public void testFactoryIsolatesGroups() { + String prefix = "factory-isolation-" + System.nanoTime(); + HAGroupStoreMetricsSource first = + HAGroupStoreMetricsSourceFactory.getInstanceForHAGroup(prefix + "-1"); + HAGroupStoreMetricsSource second = + HAGroupStoreMetricsSourceFactory.getInstanceForHAGroup(prefix + "-2"); + + assertNotSame(first, second); + assertNotEquals(first.getMetricsJmxContext(), second.getMetricsJmxContext()); + first.incrementInvalidTransitionRejectedCount(); + assertEquals(1L, first.getCurrentMetricValues().getInvalidTransitionRejectedCount()); + assertEquals(0L, second.getCurrentMetricValues().getInvalidTransitionRejectedCount()); + } + + @Test + public void testGroupIdentityIsTaggedAndQuoted() { + String group = "group,one=" + System.nanoTime(); + HAGroupStoreMetricsSourceImpl taggedSource = new HAGroupStoreMetricsSourceImpl(group); + try { + assertEquals(group, taggedSource.getMetricsRegistry().getTag(HA_GROUP_TAG_NAME).value()); + assertTrue( + taggedSource.getMetricsJmxContext().endsWith(",haGroup=" + ObjectName.quote(group))); + } finally { + taggedSource.unregisterForTesting(); + } + } +}