You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@phoenix.apache.org by ya...@apache.org on 2021/05/13 17:19:15 UTC
[phoenix] branch 4.x updated: PHOENIX-6397 Implement
TableMetricsManager class and its associated functions for select. upsert
and Delete Queries
This is an automated email from the ASF dual-hosted git repository.
yanxinyi pushed a commit to branch 4.x
in repository https://gitbox.apache.org/repos/asf/phoenix.git
The following commit(s) were added to refs/heads/4.x by this push:
new 67f1f1c PHOENIX-6397 Implement TableMetricsManager class and its associated functions for select. upsert and Delete Queries
67f1f1c is described below
commit 67f1f1ce57487c87d655b2a22cfcbf79b272655a
Author: vikas meka <vm...@salesforce.com>
AuthorDate: Thu May 13 10:17:49 2021 -0700
PHOENIX-6397 Implement TableMetricsManager class and its associated functions for select. upsert and Delete Queries
---
.../phoenix/monitoring/BasePhoenixMetricsIT.java | 109 +-
.../monitoring/PhoenixLoggingMetricsIT.java | 2 +-
.../phoenix/monitoring/PhoenixMetricsIT.java | 194 ++-
.../monitoring/PhoenixTableLevelMetricsIT.java | 1237 ++++++++++++++++++++
.../phoenix/util/DelayedOrFailingRegionServer.java | 154 +++
.../apache/phoenix/util/DelayedRegionServer.java | 18 +
.../org/apache/phoenix/execute/MutationState.java | 249 +++-
.../phoenix/iterate/BaseResultIterators.java | 31 +-
.../phoenix/iterate/RoundRobinResultIterator.java | 8 +
.../phoenix/iterate/ScanningResultIterator.java | 2 +
.../org/apache/phoenix/jdbc/PhoenixConnection.java | 2 +
.../org/apache/phoenix/jdbc/PhoenixResultSet.java | 45 +-
.../org/apache/phoenix/jdbc/PhoenixStatement.java | 294 +++--
.../phoenix/monitoring/JmxMetricProvider.java | 96 ++
.../monitoring/MetricPublisherSupplierFactory.java | 34 +
.../phoenix/monitoring/MetricServiceResolver.java | 65 +
.../org/apache/phoenix/monitoring/MetricType.java | 76 +-
.../apache/phoenix/monitoring/MetricsRegistry.java | 34 +
.../phoenix/monitoring/MutationMetricQueue.java | 111 +-
.../monitoring/NoOpTableMetricsManager.java | 53 +
.../phoenix/monitoring/OverAllQueryMetrics.java | 42 +
.../phoenix/monitoring/ScanMetricsHolder.java | 7 +
.../phoenix/monitoring/TableClientMetrics.java | 182 +++
.../phoenix/monitoring/TableMetricsManager.java | 290 +++++
.../org/apache/phoenix/query/QueryServices.java | 6 +
.../apache/phoenix/query/QueryServicesOptions.java | 36 +
.../org/apache/phoenix/util/PhoenixRuntime.java | 20 +
...oenix.monitoring.MetricPublisherSupplierFactory | 16 +
.../phoenix/monitoring/TableClientMetricsTest.java | 214 ++++
.../monitoring/TableLevelMetricsTestData.java | 165 +++
.../monitoring/TableMetricsManagerTest.java | 207 ++++
31 files changed, 3829 insertions(+), 170 deletions(-)
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/monitoring/BasePhoenixMetricsIT.java b/phoenix-core/src/it/java/org/apache/phoenix/monitoring/BasePhoenixMetricsIT.java
index 03b61f4..7c58945 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/monitoring/BasePhoenixMetricsIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/monitoring/BasePhoenixMetricsIT.java
@@ -33,8 +33,13 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
+import static org.apache.phoenix.monitoring.MetricType.DELETE_BATCH_FAILED_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.DELETE_BATCH_FAILED_SIZE;
+import static org.apache.phoenix.monitoring.MetricType.MUTATION_COMMIT_TIME;
import static org.apache.phoenix.monitoring.MetricType.SCAN_BYTES;
import static org.apache.phoenix.monitoring.MetricType.TASK_EXECUTED_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.UPSERT_BATCH_FAILED_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.UPSERT_BATCH_FAILED_SIZE;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
@@ -102,30 +107,118 @@ public class BasePhoenixMetricsIT extends BaseUniqueNamesOwnClusterIT {
assertEquals("There should have been read metrics only for one table: " + tableName, 1, numTables);
}
- void assertMutationMetrics(String tableName, int numRows, Map<String, Map<MetricType, Long>> mutationMetrics) {
- assertTrue("No mutation metrics present when there should have been", mutationMetrics.size() > 0);
+ static void assertMutationMetrics(String tableName, int numRows, boolean isUpsert,
+ Map<String, Map<MetricType, Long>> mutationMetrics) {
for (Map.Entry<String, Map<MetricType, Long>> entry : mutationMetrics.entrySet()) {
String t = entry.getKey();
- assertEquals("Table name didn't match for mutation metrics", tableName, t);
+ assertEquals("Table names didn't match!", tableName, t);
Map<MetricType, Long> p = entry.getValue();
- assertEquals("There should have been five metrics", 5, p.size());
+ assertEquals("There should have been fifteen metrics", 15, p.size());
+ boolean mutationBatchSizePresent = false;
+ boolean mutationCommitTimePresent = false;
+ boolean mutationBytesPresent = false;
+ boolean mutationBatchFailedPresent = false;
+ boolean upsertBatchFailedSizePresent = false;
+ boolean deleteBatchFailedSizePresent = false;
+ boolean upsertBatchFailedCounterPresent = false;
+ boolean deleteBatchFailedCounterPresent = false;
+ boolean deleteMutationBytesPresent = false;
+ boolean upsertMutationBytesPresent = false;
+ boolean indexCommitFailureSizePresent = false;
+ boolean deleteMutationSqlCounterPresent = false;
+ boolean upsertMutationSqlCounterPresent = false;
+ boolean upsertCommitTimeCounterPresent = false;
+ boolean deleteCommitTimeCounterPresent = false;
for (Map.Entry<MetricType, Long> metric : p.entrySet()) {
MetricType metricType = metric.getKey();
long metricValue = metric.getValue();
if (metricType.equals(MetricType.MUTATION_BATCH_SIZE)) {
assertEquals("Mutation batch sizes didn't match!", numRows, metricValue);
- } else if (metricType.equals(MetricType.MUTATION_COMMIT_TIME)) {
+ mutationBatchSizePresent = true;
+ } else if (metricType.equals(MUTATION_COMMIT_TIME)) {
assertTrue("Mutation commit time should be greater than zero", metricValue > 0);
+ mutationCommitTimePresent = true;
} else if (metricType.equals(MetricType.MUTATION_BYTES)) {
assertTrue("Mutation bytes size should be greater than zero", metricValue > 0);
+ mutationBytesPresent = true;
} else if (metricType.equals(MetricType.MUTATION_BATCH_FAILED_SIZE)) {
assertEquals("Zero failed mutations expected", 0, metricValue);
+ mutationBatchFailedPresent = true;
+ } else if (metricType.equals(UPSERT_BATCH_FAILED_SIZE)) {
+ assertEquals("Zero failed upsert mutations size expected", 0, metricValue);
+ upsertBatchFailedSizePresent = true;
+ } else if (metricType.equals(DELETE_BATCH_FAILED_SIZE)) {
+ assertEquals("Zero failed delete mutations size expected", 0, metricValue);
+ deleteBatchFailedSizePresent = true;
+ } else if (metricType.equals(UPSERT_BATCH_FAILED_COUNTER)) {
+ assertEquals("Zero failed upsert mutations counter expected", 0, metricValue);
+ upsertBatchFailedCounterPresent = true;
+ } else if (metricType.equals(DELETE_BATCH_FAILED_COUNTER)) {
+ assertEquals("Zero failed delete mutations counter expected", 0, metricValue);
+ deleteBatchFailedCounterPresent = true;
+ } else if (metricType.equals(MetricType.DELETE_MUTATION_BYTES)) {
+ if (isUpsert) {
+ assertEquals("Zero delete mutation bytes size expected", 0, metricValue);
+ } else {
+ assertTrue("Delete mutation bytes size should be greater than zero", metricValue > 0);
+ }
+ deleteMutationBytesPresent = true;
+ } else if (metricType.equals(MetricType.UPSERT_MUTATION_BYTES)) {
+ if (isUpsert) {
+ assertTrue("Upsert mutation bytes size should be greater than zero", metricValue > 0);
+ } else {
+ assertEquals("Zero Upsert mutation bytes size expected", 0, metricValue);
+ }
+ upsertMutationBytesPresent = true;
} else if (metricType.equals(MetricType.INDEX_COMMIT_FAILURE_SIZE)) {
- assertEquals("Zero failed phase 3 mutations expected", 0, metricValue);
+ assertEquals("Zero index commit failures expected", 0, metricValue);
+ indexCommitFailureSizePresent = true;
+ } else if (metricType.equals(MetricType.DELETE_MUTATION_SQL_COUNTER)) {
+ if (isUpsert) {
+ assertEquals("Zero delete mutations sql counter expected", 0, metricValue);
+ } else {
+ assertTrue("Delete mutations sql counter should be greater than zero", metricValue > 0);
+ }
+ deleteMutationSqlCounterPresent = true;
+ } else if (metricType.equals(MetricType.UPSERT_MUTATION_SQL_COUNTER)) {
+ if (isUpsert) {
+ assertTrue("Upsert mutation sql counter should be greater than zero", metricValue > 0);
+ } else {
+ assertEquals("Zero upsert mutations sql counter expected", 0, metricValue);
+ }
+ upsertMutationSqlCounterPresent = true;
+ } else if (metricType.equals(MetricType.UPSERT_COMMIT_TIME)) {
+ if (isUpsert) {
+ assertTrue("Upsert commit time counter should be greater than zero", metricValue > 0);
+ } else {
+ assertEquals("Zero upsert commit time counter expected", 0, metricValue);
+ }
+ upsertCommitTimeCounterPresent = true;
+ } else if (metricType.equals(MetricType.DELETE_COMMIT_TIME)) {
+ if (!isUpsert) {
+ assertTrue("delete commit time counter should be greater than zero",
+ metricValue > 0);
+ } else {
+ assertEquals("Zero delete commit time counter expected", 0, metricValue);
+ }
+ deleteCommitTimeCounterPresent = true;
}
}
+ assertTrue(mutationBatchSizePresent);
+ assertTrue(mutationCommitTimePresent);
+ assertTrue(mutationBytesPresent);
+ assertTrue(mutationBatchFailedPresent);
+ assertTrue(deleteMutationBytesPresent);
+ assertTrue(upsertMutationBytesPresent);
+ assertTrue(indexCommitFailureSizePresent);
+ assertTrue(deleteMutationSqlCounterPresent);
+ assertTrue(upsertMutationSqlCounterPresent);
+ assertTrue(upsertBatchFailedSizePresent);
+ assertTrue(deleteBatchFailedSizePresent);
+ assertTrue(upsertBatchFailedCounterPresent);
+ assertTrue(deleteBatchFailedCounterPresent);
+ assertTrue(upsertCommitTimeCounterPresent);
+ assertTrue(deleteCommitTimeCounterPresent);
}
}
-
-
}
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/monitoring/PhoenixLoggingMetricsIT.java b/phoenix-core/src/it/java/org/apache/phoenix/monitoring/PhoenixLoggingMetricsIT.java
index b2413ea..33cd469 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/monitoring/PhoenixLoggingMetricsIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/monitoring/PhoenixLoggingMetricsIT.java
@@ -114,7 +114,7 @@ public class PhoenixLoggingMetricsIT extends BasePhoenixMetricsIT {
loggedConn.commit();
assertTrue("Mutation write metrics for not found for " + tableName2,
mutationWriteMetricsMap.get(tableName2).size() > 0);
- assertMutationMetrics(tableName2, NUM_ROWS, mutationWriteMetricsMap);
+ assertMutationMetrics(tableName2, NUM_ROWS, true, mutationWriteMetricsMap);
assertTrue("Mutation read metrics for not found for " + tableName1,
mutationReadMetricsMap.get(tableName1).size() > 0);
assertReadMetricsForMutatingSql(tableName1, 1, mutationReadMetricsMap);
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/monitoring/PhoenixMetricsIT.java b/phoenix-core/src/it/java/org/apache/phoenix/monitoring/PhoenixMetricsIT.java
index e8d9e40..dc27dee 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/monitoring/PhoenixMetricsIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/monitoring/PhoenixMetricsIT.java
@@ -35,10 +35,16 @@ import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_SELECT_SQ
import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_SPOOL_FILE_COUNTER;
import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_TASK_END_TO_END_TIME;
import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_TASK_EXECUTION_TIME;
+import static org.apache.phoenix.monitoring.MetricType.COUNT_MILLS_BETWEEN_NEXTS;
+import static org.apache.phoenix.monitoring.MetricType.DELETE_COMMIT_TIME;
import static org.apache.phoenix.monitoring.MetricType.MEMORY_CHUNK_BYTES;
+import static org.apache.phoenix.monitoring.MetricType.MUTATION_COMMIT_TIME;
import static org.apache.phoenix.monitoring.MetricType.QUERY_TIMEOUT_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.TASK_END_TO_END_TIME;
import static org.apache.phoenix.monitoring.MetricType.TASK_EXECUTED_COUNTER;
import static org.apache.phoenix.monitoring.MetricType.TASK_EXECUTION_TIME;
+import static org.apache.phoenix.monitoring.MetricType.TASK_QUEUE_WAIT_TIME;
+import static org.apache.phoenix.monitoring.MetricType.UPSERT_COMMIT_TIME;
import static org.apache.phoenix.util.PhoenixRuntime.TENANT_ID_ATTRIB;
import static org.apache.phoenix.util.PhoenixRuntime.UPSERT_BATCH_SIZE_ATTRIB;
import static org.junit.Assert.assertEquals;
@@ -47,11 +53,13 @@ import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import java.sql.Connection;
+import java.sql.Date;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
+import java.sql.Types;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
@@ -63,6 +71,7 @@ import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.hadoop.metrics2.AbstractMetric;
import org.apache.phoenix.compat.hbase.HbaseCompatCapabilities;
@@ -97,6 +106,40 @@ public class PhoenixMetricsIT extends BasePhoenixMetricsIT {
private static final Logger LOGGER = LoggerFactory.getLogger(PhoenixMetricsIT.class);
+ private static final String DDL = "CREATE TABLE IF NOT EXISTS %s ("
+ + " A CHAR(18) NOT NULL, "
+ + " B VARCHAR(30) NOT NULL, "
+ + " C VARCHAR(255) NOT NULL, "
+ + " D VARCHAR(255) NOT NULL, "
+ + " E VARBINARY, "
+ + " F INTEGER, "
+ + " G BOOLEAN, "
+ + " H VARCHAR(30), "
+ + " I DATE, "
+ + " J DATE, "
+ + " CONSTRAINT PK PRIMARY KEY (A, B, C, D))"
+ + " IMMUTABLE_ROWS=true, VERSIONS=1, MULTI_TENANT=true,"
+ + " DISABLE_TABLE_SOR=true, TTL=864000";
+ private static final String UPSERT_VALUES_DML = "UPSERT INTO %s (A, B, C, D, F, G, I)"
+ + " VALUES (?, ?, ?, ?, ?, ?, ?)";
+ private static final String UPSERT_SELECT_DML = "UPSERT INTO %s (A, B, C, D)"
+ + " SELECT A, B, C, D FROM %s";
+ private static final String POINT_DELETE_DML =
+ "DELETE FROM %s WHERE A=? AND B=? AND C=? AND D=? AND G=FALSE";
+ private static final String DELETE_ALL_DML = "DELETE FROM %s";
+
+ private static final List<MetricType> mutationMetricsToSkip =
+ Lists.newArrayList(MUTATION_COMMIT_TIME, UPSERT_COMMIT_TIME, DELETE_COMMIT_TIME);
+ private static final List<MetricType> readMetricsToSkip =
+ Lists.newArrayList(TASK_QUEUE_WAIT_TIME, TASK_EXECUTION_TIME, TASK_END_TO_END_TIME,
+ COUNT_MILLS_BETWEEN_NEXTS);
+ private static final String CUSTOM_URL_STRING = "SESSION";
+ private static final AtomicInteger numConnections = new AtomicInteger(0);
+ static final String POINT_LOOKUP_SELECT_QUERY = "SELECT J, G, E, (NOW() - I)*24*60*60*1000 FROM"
+ + " %s WHERE A='keyA1' AND B='keyB1' AND C='keyC1' AND D='keyD1'";
+ static final String RANGE_SCAN_SELECT_QUERY = "SELECT A, B, C FROM"
+ + " %s WHERE A='keyA1' AND B='keyB1' AND C > 'keyC0'";
+
private static class MyClock extends EnvironmentEdge {
private long time;
private final long delay;
@@ -128,7 +171,9 @@ public class PhoenixMetricsIT extends BasePhoenixMetricsIT {
@Test
public void testGlobalPhoenixMetricsForQueries() throws Exception {
String tableName = generateUniqueName();
- createTableAndInsertValues(tableName, true);
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ createTableAndInsertValues(tableName, true, false, 10, true, conn, false);
+ }
resetGlobalMetrics(); // we want to count metrics related only to the below query
Connection conn = DriverManager.getConnection(getUrl());
String query = "SELECT * FROM " + tableName;
@@ -164,7 +209,9 @@ public class PhoenixMetricsIT extends BasePhoenixMetricsIT {
@Test
public void testGlobalPhoenixMetricsForMutations() throws Exception {
String tableName = generateUniqueName();
- createTableAndInsertValues(tableName, true);
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ createTableAndInsertValues(tableName, true, false, 10, true, conn, false);
+ }
assertEquals(10, GLOBAL_MUTATION_BATCH_SIZE.getMetric().getValue());
assertEquals(10, GLOBAL_MUTATION_SQL_COUNTER.getMetric().getValue());
assertTrue(GLOBAL_MUTATION_BYTES.getMetric().getValue() > 0);
@@ -185,15 +232,18 @@ public class PhoenixMetricsIT extends BasePhoenixMetricsIT {
public void testGlobalPhoenixMetricsForUpsertSelect() throws Exception {
String tableFrom = generateUniqueName();
String tableTo = generateUniqueName();
- createTableAndInsertValues(tableFrom, true);
- resetGlobalMetrics();
- String ddl = "CREATE TABLE " + tableTo + " (K VARCHAR NOT NULL PRIMARY KEY, V VARCHAR)";
- Connection conn = DriverManager.getConnection(getUrl());
- conn.createStatement().execute(ddl);
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ createTableAndInsertValues(tableFrom, true, false, 10, true, conn, false);
+ }
resetGlobalMetrics();
- String dml = "UPSERT INTO " + tableTo + " (K, V) SELECT K, V FROM " + tableFrom;
- conn.createStatement().executeUpdate(dml);
- conn.commit();
+ String ddl = String.format(DDL, tableTo);
+ try (Connection conn = DriverManager.getConnection(getUrl());
+ Statement stmt = conn.createStatement()) {
+ stmt.execute(ddl);
+ resetGlobalMetrics();
+ stmt.executeUpdate(String.format(UPSERT_SELECT_DML, tableTo, tableFrom));
+ conn.commit();
+ }
assertEquals(10, GLOBAL_MUTATION_BATCH_SIZE.getMetric().getValue());
assertEquals(1, GLOBAL_MUTATION_SQL_COUNTER.getMetric().getValue());
assertEquals(1, GLOBAL_NUM_PARALLEL_SCANS.getMetric().getValue());
@@ -217,7 +267,7 @@ public class PhoenixMetricsIT extends BasePhoenixMetricsIT {
assertTrue(verifyMetricsFromSink());
}
- private static void resetGlobalMetrics() {
+ static void resetGlobalMetrics() {
for (GlobalMetric m : PhoenixRuntime.getGlobalPhoenixClientMetrics()) {
m.reset();
}
@@ -275,26 +325,72 @@ public class PhoenixMetricsIT extends BasePhoenixMetricsIT {
return true;
}
- private static void createTableAndInsertValues(String tableName,
- boolean resetGlobalMetricsAfterTableCreate) throws SQLException {
- String ddl = String.format("CREATE TABLE %s (K VARCHAR NOT NULL PRIMARY KEY, V VARCHAR)",
- tableName);
- try (Connection conn = DriverManager.getConnection(getUrl());
- Statement stmt = conn.createStatement()) {
- stmt.execute(ddl);
- if (resetGlobalMetricsAfterTableCreate) {
- resetGlobalMetrics();
- }
- // executing 10 upserts/mutations.
- String dml = String.format("UPSERT INTO %s VALUES (?, ?)", tableName);
- try(PreparedStatement prepStmt = conn.prepareStatement(dml)) {
- for (int i = 1; i <= 10; i++) {
- prepStmt.setString(1, "key" + i);
- prepStmt.setString(2, "value" + i);
- prepStmt.executeUpdate();
+ static void createTableAndInsertValues(String tableName,
+ boolean resetGlobalMetricsAfterTableCreate, boolean resetTableMetricsAfterTableCreate,
+ int numRows, boolean commit, Connection conn, boolean batchUpserts) throws SQLException {
+
+ try (Statement stmt = conn.createStatement()) {
+ stmt.execute(String.format(DDL, tableName));
+ }
+ conn.commit();
+ if (resetGlobalMetricsAfterTableCreate) {
+ resetGlobalMetrics();
+ }
+
+ if (resetTableMetricsAfterTableCreate) {
+ PhoenixRuntime.clearTableLevelMetrics();
+ }
+ // executing upserts/mutations.
+ try (PreparedStatement stmt = conn.prepareStatement(
+ String.format(UPSERT_VALUES_DML, tableName))) {
+ for (int i = 1; i <= numRows; i++) {
+ stmt.setString(1, "keyA" + i);
+ stmt.setString(2, "keyB" + i);
+ stmt.setString(3, "keyC" + i);
+ stmt.setString(4, "keyD" + i);
+ stmt.setBoolean(6, false);
+ // Set non-null for just half of the rows to ensure there's no problem with a null
+ // date when we query using the SELECT_QUERY
+ if (i % 2 == 0) {
+ stmt.setInt(5, i);
+ stmt.setDate(7, new Date(System.currentTimeMillis()));
+ } else {
+ stmt.setNull(5, Types.INTEGER);
+ stmt.setNull(7, Types.DATE);
+ }
+ if (batchUpserts) {
+ stmt.addBatch();
+ if (i % 3 == 0) {
+ stmt.executeBatch();
+ }
+ } else {
+ stmt.executeUpdate();
}
}
- conn.commit();
+ // Explicitly execute the batch to make sure we haven't missed any batches above
+ if (batchUpserts) {
+ stmt.executeBatch();
+ }
+ if (commit) {
+ conn.commit();
+ }
+ }
+ }
+
+ static void doPointDeleteFromTable(String tableName, Connection conn) throws SQLException {
+ try (PreparedStatement stmt = conn.prepareStatement(
+ String.format(POINT_DELETE_DML, tableName))) {
+ stmt.setString(1, "keyA1");
+ stmt.setString(2, "keyB1");
+ stmt.setString(3, "keyC1");
+ stmt.setString(4, "keyD1");
+ stmt.executeUpdate();
+ }
+ }
+
+ static void doDeleteAllFromTable(String tableName, Connection conn) throws SQLException {
+ try (Statement stmt = conn.createStatement()) {
+ stmt.executeUpdate(String.format(DELETE_ALL_DML, tableName));
}
}
@@ -326,9 +422,10 @@ public class PhoenixMetricsIT extends BasePhoenixMetricsIT {
public void testMetricsForSelectFetchResultsTimeout() throws SQLException {
String tableName = generateUniqueName();
final int queryTimeout = 10; //seconds
- createTableAndInsertValues(tableName, true);
+
try (Connection conn = DriverManager.getConnection(getUrl());
Statement stmt = conn.createStatement()) {
+ createTableAndInsertValues(tableName, true, true, 10, true, conn, false);
stmt.setQueryTimeout(queryTimeout);
ResultSet rs = stmt.executeQuery(String.format("SELECT * FROM %s", tableName));
// Make the query time out with a longer delay than the set query timeout value (in ms)
@@ -390,7 +487,7 @@ public class PhoenixMetricsIT extends BasePhoenixMetricsIT {
String t = entry.getKey();
assertEquals("Table names didn't match!", tableName, t);
Map<MetricType, Long> p = entry.getValue();
- assertEquals("There should have been five metrics", 5, p.size());
+ assertEquals("There should have been five metrics", 15, p.size());
boolean mutationBatchSizePresent = false;
boolean mutationCommitTimePresent = false;
boolean mutationBytesPresent = false;
@@ -446,7 +543,7 @@ public class PhoenixMetricsIT extends BasePhoenixMetricsIT {
PhoenixConnection pConn = conn.unwrap(PhoenixConnection.class);
Map<String, Map<MetricType, Long>> mutationMetrics = PhoenixRuntime.getWriteMetricInfoForMutationsSinceLastReset(pConn);
- assertMutationMetrics(tableName2, numRows, mutationMetrics);
+ assertMutationMetrics(tableName2, numRows, true, mutationMetrics);
Map<String, Map<MetricType, Long>> readMetrics = PhoenixRuntime.getReadMetricInfoForMutationsSinceLastReset(pConn);
assertReadMetricsForMutatingSql(tableName1, table1SaltBuckets, readMetrics);
}
@@ -457,21 +554,20 @@ public class PhoenixMetricsIT extends BasePhoenixMetricsIT {
long tableSaltBuckets = 6;
String ddl = "CREATE TABLE " + tableName + " (K VARCHAR NOT NULL PRIMARY KEY, V VARCHAR)" + " SALT_BUCKETS = "
+ tableSaltBuckets;
- Connection ddlConn = DriverManager.getConnection(getUrl());
- ddlConn.createStatement().execute(ddl);
- ddlConn.close();
+ try(Connection ddlConn = DriverManager.getConnection(getUrl())) {
+ ddlConn.createStatement().execute(ddl);
+ }
int numRows = 10;
insertRowsInTable(tableName, numRows);
- Connection conn = DriverManager.getConnection(getUrl());
- String delete = "DELETE FROM " + tableName;
- conn.createStatement().execute(delete);
- conn.commit();
- PhoenixConnection pConn = conn.unwrap(PhoenixConnection.class);
- Map<String, Map<MetricType, Long>> mutationMetrics = PhoenixRuntime.getWriteMetricInfoForMutationsSinceLastReset(pConn);
- assertMutationMetrics(tableName, numRows, mutationMetrics);
-
- Map<String, Map<MetricType, Long>> readMetrics = PhoenixRuntime.getReadMetricInfoForMutationsSinceLastReset(pConn);
- assertReadMetricsForMutatingSql(tableName, tableSaltBuckets, readMetrics);
+ try(Connection conn = DriverManager.getConnection(getUrl())) {
+ String delete = "DELETE FROM " + tableName;
+ conn.createStatement().execute(delete);
+ conn.commit();
+ Map<String, Map<MetricType, Long>> mutationMetrics = PhoenixRuntime.getWriteMetricInfoForMutationsSinceLastReset(conn);
+ assertMutationMetrics(tableName, numRows, false, mutationMetrics);
+ Map<String, Map<MetricType, Long>> readMetrics = PhoenixRuntime.getReadMetricInfoForMutationsSinceLastReset(conn);
+ assertReadMetricsForMutatingSql(tableName, tableSaltBuckets, readMetrics);
+ }
}
@Test
@@ -690,11 +786,11 @@ public class PhoenixMetricsIT extends BasePhoenixMetricsIT {
public void testMutationMetricsWhenUpsertingToMultipleTables() throws Exception {
try (Connection conn = DriverManager.getConnection(getUrl())) {
String table1 = generateUniqueName();
- createTableAndInsertValues(true, 10, conn, table1);
+ createTableAndInsertValues(table1, false, false, 10, true, conn, false);
String table2 = generateUniqueName();
- createTableAndInsertValues(true, 10, conn, table2);
+ createTableAndInsertValues(table2, false, false, 10, true, conn, false);
String table3 = generateUniqueName();
- createTableAndInsertValues(true, 10, conn, table3);
+ createTableAndInsertValues(table3, false, false, 10, true, conn, false);
Map<String, Map<MetricType, Long>> mutationMetrics = PhoenixRuntime.getWriteMetricInfoForMutationsSinceLastReset(conn);
assertTrue("Mutation metrics not present for " + table1, mutationMetrics.get(table1) != null);
assertTrue("Mutation metrics not present for " + table2, mutationMetrics.get(table2) != null);
@@ -724,7 +820,7 @@ public class PhoenixMetricsIT extends BasePhoenixMetricsIT {
Connection conn = null;
try {
conn = DriverManager.getConnection(getUrl());
- createTableAndInsertValues(true, 10, conn, generateUniqueName());
+ createTableAndInsertValues(generateUniqueName(), false, false, 10, true, conn, false);
assertTrue("Mutation metrics not present", PhoenixRuntime.getWriteMetricInfoForMutationsSinceLastReset(conn).size() > 0);
} finally {
if (conn != null) {
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/monitoring/PhoenixTableLevelMetricsIT.java b/phoenix-core/src/it/java/org/apache/phoenix/monitoring/PhoenixTableLevelMetricsIT.java
new file mode 100644
index 0000000..b1215cd
--- /dev/null
+++ b/phoenix-core/src/it/java/org/apache/phoenix/monitoring/PhoenixTableLevelMetricsIT.java
@@ -0,0 +1,1237 @@
+/*
+ * 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 com.google.common.collect.Maps;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.DoNotRetryIOException;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+
+import org.apache.hadoop.hbase.HBaseTestingUtility;
+import org.apache.hadoop.hbase.HRegionLocation;
+import org.apache.hadoop.hbase.client.RetriesExhaustedWithDetailsException;
+import org.apache.phoenix.end2end.BaseUniqueNamesOwnClusterIT;
+import org.apache.phoenix.end2end.NeedsOwnMiniClusterTest;
+import org.apache.phoenix.exception.PhoenixIOException;
+import org.apache.phoenix.exception.SQLExceptionInfo;
+import org.apache.phoenix.execute.CommitException;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.jdbc.PhoenixEmbeddedDriver;
+import org.apache.phoenix.jdbc.PhoenixTestDriver;
+import org.apache.phoenix.query.BaseTest;
+import org.apache.phoenix.query.ConfigurationFactory;
+import org.apache.phoenix.query.ConnectionQueryServices;
+import org.apache.phoenix.query.ConnectionQueryServicesImpl;
+import org.apache.phoenix.query.QueryServices;
+import org.apache.phoenix.query.QueryServicesTestImpl;
+import org.apache.phoenix.util.DelayedOrFailingRegionServer;
+import org.apache.phoenix.util.EnvironmentEdge;
+import org.apache.phoenix.util.EnvironmentEdgeManager;
+import org.apache.phoenix.util.InstanceResolver;
+import org.apache.phoenix.util.PhoenixRuntime;
+import org.apache.phoenix.util.ReadOnlyProps;
+
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import static org.apache.phoenix.exception.SQLExceptionCode.DATA_EXCEEDS_MAX_CAPACITY;
+import static org.apache.phoenix.exception.SQLExceptionCode.GET_TABLE_REGIONS_FAIL;
+import static org.apache.phoenix.exception.SQLExceptionCode.OPERATION_TIMED_OUT;
+import static org.apache.phoenix.monitoring.MetricType.DELETE_AGGREGATE_FAILURE_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.DELETE_AGGREGATE_SUCCESS_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.DELETE_BATCH_FAILED_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.DELETE_BATCH_FAILED_SIZE;
+import static org.apache.phoenix.monitoring.MetricType.DELETE_COMMIT_TIME;
+import static org.apache.phoenix.monitoring.MetricType.DELETE_FAILED_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.DELETE_MUTATION_BYTES;
+import static org.apache.phoenix.monitoring.MetricType.DELETE_MUTATION_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.DELETE_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.DELETE_SQL_QUERY_TIME;
+import static org.apache.phoenix.monitoring.MetricType.DELETE_SUCCESS_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.MUTATION_BATCH_FAILED_SIZE;
+import static org.apache.phoenix.monitoring.MetricType.MUTATION_BATCH_SIZE;
+import static org.apache.phoenix.monitoring.MetricType.MUTATION_BYTES;
+import static org.apache.phoenix.monitoring.MetricType.QUERY_FAILED_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.QUERY_POINTLOOKUP_FAILED_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.QUERY_POINTLOOKUP_TIMEOUT_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.QUERY_SCAN_FAILED_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.QUERY_SCAN_TIMEOUT_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.QUERY_TIMEOUT_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.RESULT_SET_TIME_MS;
+import static org.apache.phoenix.monitoring.MetricType.SCAN_BYTES;
+import static org.apache.phoenix.monitoring.MetricType.SELECT_AGGREGATE_FAILURE_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.SELECT_AGGREGATE_SUCCESS_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.SELECT_FAILED_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.SELECT_POINTLOOKUP_FAILED_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.SELECT_POINTLOOKUP_SUCCESS_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.SELECT_SCAN_FAILED_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.SELECT_SCAN_SUCCESS_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.SELECT_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.SELECT_SQL_QUERY_TIME;
+import static org.apache.phoenix.monitoring.MetricType.SELECT_SUCCESS_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.UPSERT_AGGREGATE_FAILURE_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.UPSERT_AGGREGATE_SUCCESS_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.UPSERT_BATCH_FAILED_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.UPSERT_BATCH_FAILED_SIZE;
+import static org.apache.phoenix.monitoring.MetricType.UPSERT_COMMIT_TIME;
+import static org.apache.phoenix.monitoring.MetricType.UPSERT_FAILED_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.UPSERT_MUTATION_BYTES;
+import static org.apache.phoenix.monitoring.MetricType.UPSERT_MUTATION_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.UPSERT_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.UPSERT_SQL_QUERY_TIME;
+import static org.apache.phoenix.monitoring.MetricType.UPSERT_SUCCESS_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.PhoenixMetricsIT.POINT_LOOKUP_SELECT_QUERY;
+import static org.apache.phoenix.monitoring.PhoenixMetricsIT.RANGE_SCAN_SELECT_QUERY;
+import static org.apache.phoenix.monitoring.PhoenixMetricsIT.createTableAndInsertValues;
+import static org.apache.phoenix.monitoring.PhoenixMetricsIT.doPointDeleteFromTable;
+import static org.apache.phoenix.monitoring.PhoenixMetricsIT.doDeleteAllFromTable;
+import static org.apache.phoenix.util.DelayedOrFailingRegionServer.INJECTED_EXCEPTION_STRING;
+import static org.apache.phoenix.util.PhoenixRuntime.JDBC_PROTOCOL_SEPARATOR;
+import static org.apache.phoenix.util.PhoenixRuntime.clearTableLevelMetrics;
+import static org.apache.phoenix.util.PhoenixRuntime.getOverAllReadRequestMetricInfo;
+import static org.apache.phoenix.util.PhoenixRuntime.getPhoenixTableClientMetrics;
+import static org.apache.phoenix.util.PhoenixRuntime.getRequestReadMetricInfo;
+import static org.apache.phoenix.util.PhoenixRuntime.getWriteMetricInfoForMutationsSinceLastReset;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.atomic.AtomicLong;
+
+@Category(NeedsOwnMiniClusterTest.class)
+public class PhoenixTableLevelMetricsIT extends BaseUniqueNamesOwnClusterIT {
+
+ private static final String
+ CREATE_TABLE_DDL =
+ "CREATE TABLE %s (K VARCHAR(%d) NOT NULL" + " PRIMARY KEY, V VARCHAR)";
+ private static final String UPSERT_DML = "UPSERT INTO %s VALUES (?, ?)";
+ private static final String KEY = "key";
+ private static final String VALUE = "value";
+ private static boolean failExecuteQueryAndClientSideDeletes;
+ private static long injectDelay;
+ private static HBaseTestingUtility hbaseTestUtil;
+
+ @BeforeClass public static void doSetup() throws Exception {
+ final Configuration conf = HBaseConfiguration.create();
+ conf.set(QueryServices.TABLE_LEVEL_METRICS_ENABLED, String.valueOf(true));
+ conf.set(QueryServices.METRIC_PUBLISHER_ENABLED, String.valueOf(true));
+ conf.set(QueryServices.COLLECT_REQUEST_LEVEL_METRICS, String.valueOf(true));
+
+ InstanceResolver.clearSingletons();
+ // Override to get required config for static fields loaded that require HBase config
+ InstanceResolver.getSingleton(ConfigurationFactory.class, new ConfigurationFactory() {
+
+ @Override public Configuration getConfiguration() {
+ return conf;
+ }
+
+ @Override public Configuration getConfiguration(Configuration confToClone) {
+ Configuration copy = new Configuration(conf);
+ copy.addResource(confToClone);
+ return copy;
+ }
+ });
+ hbaseTestUtil = new HBaseTestingUtility();
+ hbaseTestUtil.startMiniCluster(1, 1, null, null, DelayedOrFailingRegionServer.class);
+ // establish url and quorum. Need to use PhoenixDriver and not PhoenixTestDriver
+ String zkQuorum = "localhost:" + hbaseTestUtil.getZkCluster().getClientPort();
+ url = PhoenixRuntime.JDBC_PROTOCOL + JDBC_PROTOCOL_SEPARATOR + zkQuorum;
+
+ // Add our own driver
+ Map<String, String> props = Maps.newHashMapWithExpectedSize(1);
+ props.put(BaseTest.DRIVER_CLASS_NAME_ATTRIB, PhoenixMetricsTestingDriver.class.getName());
+ initAndRegisterTestDriver(url, new ReadOnlyProps(props.entrySet().iterator()));
+ }
+
+ @AfterClass public static void tearDownMiniCluster() {
+ try {
+ if (hbaseTestUtil != null) {
+ hbaseTestUtil.shutdownMiniCluster();
+ }
+ } catch (Exception e) {
+ // ignore
+ }
+ }
+
+ /**
+ * Assert table-level metrics related to SELECT queries
+ *
+ * @param tableName table name
+ * @param isPointLookup true if it is a point lookup
+ * @param expectedSqlSuccessCt expected number of successes related to query execution
+ * @param expectedSqlFailureCt expected number of failures related to query execution
+ * @param expectedMinTimeElapsed minimum expected time elapsed during query execution
+ * @param hasResultSetIterationStarted true if we have actually started issuing the scan(s) and
+ * iterating over results via ResultSet.next() calls
+ * @param expectedResultSetIterFailedCounter expected number of failures related to rs.next()
+ * @param expectedResultSetIterTimeoutCounter expected number of timeouts related to rs.next()
+ * @param rs current ResultSet which we can use to check table-level metric values against
+ * the ReadMetricQueue and OverallQueryMetrics. Null indicates that rs iteration
+ * has not started yet due to an exception in the executeMutation step itself
+ */
+ static void assertSelectQueryTableMetrics(final String tableName, final boolean isPointLookup,
+ final long expectedSelectAggregateSuccessCt,
+ final long expectedSelectAggregateFailureCt, final long expectedSqlSuccessCt,
+ final long expectedSqlFailureCt, final long expectedMinTimeElapsed,
+ final boolean hasResultSetIterationStarted,
+ final long expectedResultSetIterFailedCounter,
+ final long expectedResultSetIterTimeoutCounter, final ResultSet rs)
+ throws SQLException {
+ // The resultSet must be closed since we modify certain timing related metrics when calling rs.close()
+ if (hasResultSetIterationStarted) {
+ assertTrue(rs != null && rs.isClosed());
+ } else {
+ assertTrue(rs == null || rs.isBeforeFirst());
+ }
+ assertFalse(getPhoenixTableClientMetrics().isEmpty());
+ assertFalse(getPhoenixTableClientMetrics().get(tableName).isEmpty());
+ final long expectedTimeToFetchAllRecordsRsNext =
+ rs == null ? 0 : getOverAllReadRequestMetricInfo(rs).get(RESULT_SET_TIME_MS);
+ final long expectedScanBytes =
+ rs == null || rs.isBeforeFirst() ? 0 :
+ getRequestReadMetricInfo(rs).get(tableName).get(SCAN_BYTES);
+ final long expectedSqlCounter = expectedSqlSuccessCt + expectedSqlFailureCt;
+
+ for (PhoenixTableMetric metric : getPhoenixTableClientMetrics().get(tableName)) {
+ assertMetricValue(metric, SELECT_SQL_COUNTER, expectedSqlCounter, CompareOp.EQ);
+ assertMetricValue(metric, SELECT_AGGREGATE_FAILURE_SQL_COUNTER,
+ expectedSelectAggregateFailureCt, CompareOp.EQ);
+ assertMetricValue(metric, SELECT_AGGREGATE_SUCCESS_SQL_COUNTER,
+ expectedSelectAggregateSuccessCt, CompareOp.EQ);
+ assertMetricValue(metric, SELECT_SUCCESS_SQL_COUNTER, expectedSqlSuccessCt,
+ CompareOp.EQ);
+ assertMetricValue(metric, isPointLookup ?
+ SELECT_POINTLOOKUP_SUCCESS_SQL_COUNTER :
+ SELECT_SCAN_SUCCESS_SQL_COUNTER, expectedSqlSuccessCt, CompareOp.EQ);
+ assertMetricValue(metric, SELECT_FAILED_SQL_COUNTER, expectedSqlFailureCt,
+ CompareOp.EQ);
+ assertMetricValue(metric, isPointLookup ?
+ SELECT_POINTLOOKUP_FAILED_SQL_COUNTER :
+ SELECT_SCAN_FAILED_SQL_COUNTER, expectedSqlFailureCt, CompareOp.EQ);
+ assertMetricValue(metric, SELECT_SQL_QUERY_TIME, expectedMinTimeElapsed, CompareOp.GT);
+
+ if (hasResultSetIterationStarted) {
+ if (expectedResultSetIterFailedCounter == 0) {
+ assertMetricValue(metric, SCAN_BYTES, 0, CompareOp.GT);
+ assertMetricValue(metric, RESULT_SET_TIME_MS, 0, CompareOp.GT);
+ } else {
+ assertMetricValue(metric, SCAN_BYTES, 0, CompareOp.EQ);
+ }
+ assertMetricValue(metric, SCAN_BYTES, expectedScanBytes, CompareOp.EQ);
+ assertMetricValue(metric, QUERY_FAILED_COUNTER, expectedResultSetIterFailedCounter,
+ CompareOp.EQ);
+ assertMetricValue(metric, isPointLookup ?
+ QUERY_POINTLOOKUP_FAILED_COUNTER :
+ QUERY_SCAN_FAILED_COUNTER, expectedResultSetIterFailedCounter,
+ CompareOp.EQ);
+ assertMetricValue(metric, QUERY_TIMEOUT_COUNTER,
+ expectedResultSetIterTimeoutCounter, CompareOp.EQ);
+ assertMetricValue(metric, isPointLookup ?
+ QUERY_POINTLOOKUP_TIMEOUT_COUNTER :
+ QUERY_SCAN_TIMEOUT_COUNTER, expectedResultSetIterTimeoutCounter,
+ CompareOp.EQ);
+ assertMetricValue(metric, RESULT_SET_TIME_MS, expectedTimeToFetchAllRecordsRsNext,
+ CompareOp.EQ);
+ } else {
+ assertMetricValue(metric, SCAN_BYTES, 0, CompareOp.EQ);
+ assertMetricValue(metric, QUERY_FAILED_COUNTER, 0, CompareOp.EQ);
+ assertMetricValue(metric, isPointLookup ?
+ QUERY_POINTLOOKUP_FAILED_COUNTER :
+ QUERY_SCAN_FAILED_COUNTER, 0, CompareOp.EQ);
+ assertMetricValue(metric, QUERY_TIMEOUT_COUNTER, 0, CompareOp.EQ);
+ assertMetricValue(metric, isPointLookup ?
+ QUERY_POINTLOOKUP_TIMEOUT_COUNTER :
+ QUERY_SCAN_TIMEOUT_COUNTER, 0, CompareOp.EQ);
+ assertMetricValue(metric, RESULT_SET_TIME_MS, 0, CompareOp.EQ);
+ }
+ }
+ }
+
+ /**
+ * Assert table-level metrics related to UPSERT and DELETE queries
+ *
+ * @param tableName table name
+ * @param expectedUpsertOrDeleteSuccessSqlCt expected number of successes for upsert or delete
+ * mutation execution
+ * @param expectedUpsertOrDeleteFailedSqlCt expected number of failures for upsert or delete
+ * mutation execution
+ * @param expectedMinUpsertOrDeleteSqlQueryTime minimum expected time elapsed during upsert or
+ * delete mutation execution
+ * @param hasMutationBeenExplicitlyCommitted true if conn.commit() was explicitly called,
+ * false if connection was autoCommit = true or if
+ * conn.commit() was not called
+ * @param expectedMutBatchSize total number of mutations to be committed
+ * @param expectedMinUpsertOrDeleteCommitTime minimum expected time taken to commit upsert or
+ * delete mutations
+ * @param expectedUpsertOrDeleteBatchFailedSize expected total size of upsert or delete mutation
+ * batches that failed to commit
+ * @param writeMutMetrics write mutation metrics object
+ * @param conn connection object. Note: this method must be called after connection close
+ * since that's where we populate table-level write metrics
+ */
+ private static void assertMutationTableMetrics(final boolean isUpsert, final String tableName,
+ final long expectedUpsertOrDeleteSuccessSqlCt,
+ final long expectedUpsertOrDeleteFailedSqlCt,
+ final long expectedMinUpsertOrDeleteSqlQueryTime,
+ final boolean hasMutationBeenExplicitlyCommitted, final long expectedMutBatchSize,
+ final long expectedMinUpsertOrDeleteCommitTime,
+ final long expectedUpsertOrDeleteBatchFailedSize,
+ final long expectedUpsertOrDeleteAggregateSuccessCt,
+ final long expectedUpsertOrDeleteAggregateFailureCt,
+ final Map<MetricType, Long> writeMutMetrics, final Connection conn)
+ throws SQLException {
+ assertTrue(conn != null && conn.isClosed());
+ assertFalse(hasMutationBeenExplicitlyCommitted && writeMutMetrics == null);
+ assertFalse(getPhoenixTableClientMetrics().isEmpty());
+ assertFalse(getPhoenixTableClientMetrics().get(tableName).isEmpty());
+
+ final long
+ expectedUpsertOrDeleteSqlCt =
+ expectedUpsertOrDeleteSuccessSqlCt + expectedUpsertOrDeleteFailedSqlCt;
+
+ for (PhoenixTableMetric metric : getPhoenixTableClientMetrics().get(tableName)) {
+ // executeMutation() related metrics:
+ assertMetricValue(metric, isUpsert ? UPSERT_SQL_COUNTER : DELETE_SQL_COUNTER,
+ expectedUpsertOrDeleteSqlCt, CompareOp.EQ);
+ assertMetricValue(metric,
+ isUpsert ? UPSERT_SUCCESS_SQL_COUNTER : DELETE_SUCCESS_SQL_COUNTER,
+ expectedUpsertOrDeleteSuccessSqlCt, CompareOp.EQ);
+ assertMetricValue(metric,
+ isUpsert ? UPSERT_FAILED_SQL_COUNTER : DELETE_FAILED_SQL_COUNTER,
+ expectedUpsertOrDeleteFailedSqlCt, CompareOp.EQ);
+ assertMetricValue(metric, isUpsert ? UPSERT_SQL_QUERY_TIME : DELETE_SQL_QUERY_TIME,
+ expectedMinUpsertOrDeleteSqlQueryTime, CompareOp.GTEQ);
+
+ if (hasMutationBeenExplicitlyCommitted) {
+ // conn.commit() related metrics
+ assertMetricValue(metric, MUTATION_BATCH_SIZE,
+ writeMutMetrics.get(MUTATION_BATCH_SIZE), CompareOp.EQ);
+ assertMetricValue(metric, MUTATION_BATCH_SIZE, expectedMutBatchSize, CompareOp.EQ);
+ assertMetricValue(metric, MUTATION_BYTES, writeMutMetrics.get(MUTATION_BYTES),
+ CompareOp.EQ);
+ assertMetricValue(metric, MUTATION_BATCH_FAILED_SIZE,
+ writeMutMetrics.get(MUTATION_BATCH_FAILED_SIZE), CompareOp.EQ);
+ assertMetricValue(metric, MUTATION_BATCH_FAILED_SIZE,
+ expectedUpsertOrDeleteBatchFailedSize, CompareOp.EQ);
+
+ assertMetricValue(metric, isUpsert ? UPSERT_COMMIT_TIME : DELETE_COMMIT_TIME,
+ writeMutMetrics.get(isUpsert ? UPSERT_COMMIT_TIME : DELETE_COMMIT_TIME),
+ CompareOp.EQ);
+ if (expectedUpsertOrDeleteAggregateSuccessCt > 0) {
+ assertMetricValue(metric, isUpsert ?
+ UPSERT_AGGREGATE_SUCCESS_SQL_COUNTER :
+ DELETE_AGGREGATE_SUCCESS_SQL_COUNTER,
+ expectedUpsertOrDeleteAggregateSuccessCt, CompareOp.EQ);
+ }
+ if (expectedUpsertOrDeleteAggregateFailureCt > 0) {
+ assertMetricValue(metric, isUpsert ?
+ UPSERT_AGGREGATE_FAILURE_SQL_COUNTER :
+ DELETE_AGGREGATE_FAILURE_SQL_COUNTER,
+ expectedUpsertOrDeleteAggregateFailureCt, CompareOp.EQ);
+ }
+ if (expectedUpsertOrDeleteBatchFailedSize > 0) {
+ assertMetricValue(metric, isUpsert ? UPSERT_COMMIT_TIME : DELETE_COMMIT_TIME, 0,
+ CompareOp.EQ);
+ assertMetricValue(metric,
+ isUpsert ? UPSERT_MUTATION_SQL_COUNTER : DELETE_MUTATION_SQL_COUNTER, 0,
+ CompareOp.EQ);
+ } else {
+ assertMetricValue(metric, isUpsert ? UPSERT_COMMIT_TIME : DELETE_COMMIT_TIME,
+ expectedMinUpsertOrDeleteCommitTime, CompareOp.GTEQ);
+ assertMetricValue(metric,
+ isUpsert ? UPSERT_MUTATION_SQL_COUNTER : DELETE_MUTATION_SQL_COUNTER,
+ expectedMutBatchSize, CompareOp.EQ);
+ }
+ assertMetricValue(metric, isUpsert ? UPSERT_MUTATION_BYTES : DELETE_MUTATION_BYTES,
+ writeMutMetrics
+ .get(isUpsert ? UPSERT_MUTATION_BYTES : DELETE_MUTATION_BYTES),
+ CompareOp.EQ);
+ assertMetricValue(metric,
+ isUpsert ? UPSERT_MUTATION_SQL_COUNTER : DELETE_MUTATION_SQL_COUNTER,
+ writeMutMetrics.get(isUpsert ?
+ UPSERT_MUTATION_SQL_COUNTER :
+ DELETE_MUTATION_SQL_COUNTER), CompareOp.EQ);
+ assertMetricValue(metric,
+ isUpsert ? UPSERT_BATCH_FAILED_SIZE : DELETE_BATCH_FAILED_SIZE,
+ writeMutMetrics.get(isUpsert ?
+ UPSERT_BATCH_FAILED_SIZE :
+ DELETE_BATCH_FAILED_SIZE), CompareOp.EQ);
+ assertMetricValue(metric,
+ isUpsert ? UPSERT_BATCH_FAILED_SIZE : DELETE_BATCH_FAILED_SIZE,
+ expectedUpsertOrDeleteBatchFailedSize, CompareOp.EQ);
+ assertMetricValue(metric,
+ isUpsert ? UPSERT_BATCH_FAILED_COUNTER : DELETE_BATCH_FAILED_COUNTER,
+ writeMutMetrics.get(isUpsert ?
+ UPSERT_BATCH_FAILED_COUNTER :
+ DELETE_BATCH_FAILED_COUNTER), CompareOp.EQ);
+ }
+ }
+ }
+
+ /**
+ * Checks that if the metric is of the passed in type, it has the expected value
+ * (based on the CompareOp). If the metric type is different than checkType, ignore
+ *
+ * @param m metric to check
+ * @param checkType type to check for
+ * @param compareValue value to compare against
+ * @param op CompareOp
+ */
+ private static void assertMetricValue(Metric m, MetricType checkType, long compareValue,
+ CompareOp op) {
+ if (m.getMetricType().equals(checkType)) {
+ switch (op) {
+ case EQ:
+ assertEquals(compareValue, m.getValue());
+ break;
+ case LT:
+ assertTrue(m.getValue() < compareValue);
+ break;
+ case LTEQ:
+ assertTrue(m.getValue() <= compareValue);
+ break;
+ case GT:
+ assertTrue(m.getValue() > compareValue);
+ break;
+ case GTEQ:
+ assertTrue(m.getValue() >= compareValue);
+ break;
+ }
+ }
+ }
+
+ @Before public void resetTableLevelMetrics() {
+ clearTableLevelMetrics();
+ failExecuteQueryAndClientSideDeletes = false;
+ injectDelay = 0L;
+ // Need to reset otherwise tests that inject their own clock may cause tests run after them
+ // to flake
+ EnvironmentEdgeManager.reset();
+ DelayedOrFailingRegionServer.setDelayEnabled(false);
+ DelayedOrFailingRegionServer.injectFailureForRegionOfTable(null);
+
+ }
+
+ @Test public void testTableLevelMetricsforSuccessfulPointLookupQuery() throws Exception {
+ String tableName = generateUniqueName();
+ ResultSet rs;
+ try (Connection conn = getConnFromTestDriver()) {
+ createTableAndInsertValues(tableName, false, false, 20, true, conn, false);
+ }
+ try (Connection conn = getConnFromTestDriver(); Statement stmt = conn.createStatement()) {
+ rs = stmt.executeQuery(String.format(POINT_LOOKUP_SELECT_QUERY, tableName));
+ assertSelectQueryTableMetrics(tableName, true, 0, 0, 1, 0, 0, false, 0, 0, rs);
+ rs.next();
+ } // Note that connection close will close the underlying rs too
+ assertSelectQueryTableMetrics(tableName, true, 1, 0, 1, 0, 0, true, 0, 0, rs);
+ }
+
+ @Test public void testTableLevelMetricsforSuccessfulScanQuery() throws Exception {
+ String tableName = generateUniqueName();
+ ResultSet rs;
+ try (Connection conn = getConnFromTestDriver()) {
+ createTableAndInsertValues(tableName, false, false, 20, true, conn, false);
+ }
+ try (Connection conn = getConnFromTestDriver(); Statement stmt = conn.createStatement()) {
+ rs = stmt.executeQuery(String.format(RANGE_SCAN_SELECT_QUERY, tableName));
+ assertSelectQueryTableMetrics(tableName, false, 0, 0, 1, 0, 0, false, 0, 0, rs);
+ while (rs.next()) {
+ // do nothing
+ }
+ }
+ assertSelectQueryTableMetrics(tableName, false, 1, 0, 1, 0, 0, true, 0, 0, rs);
+ }
+
+ @Test public void testTableLevelMetricsforFailingSelectQuery() throws Exception {
+ String tableName = generateUniqueName();
+ try (Connection conn = getConnFromTestDriver()) {
+ createTableAndInsertValues(tableName, false, false, 10, true, conn, false);
+ }
+ try (Connection conn = getConnFromTestDriver(); Statement stmt = conn.createStatement()) {
+ failExecuteQueryAndClientSideDeletes = true;
+ try {
+ stmt.executeQuery(String.format(POINT_LOOKUP_SELECT_QUERY, tableName));
+ fail();
+ } catch (SQLException sqlE) {
+ assertEquals(GET_TABLE_REGIONS_FAIL.getErrorCode(), sqlE.getErrorCode());
+ assertSelectQueryTableMetrics(tableName, true, 0, 1, 0, 1, 0, false, 0, 0, null);
+ }
+ }
+ }
+
+ @Test public void testTableLevelMetricsforDelayedSelectQuery() throws Exception {
+ String tableName = generateUniqueName();
+ ResultSet rs;
+ try (Connection conn = getConnFromTestDriver()) {
+ createTableAndInsertValues(tableName, false, false, 10, true, conn, false);
+ }
+ try (Connection conn = getConnFromTestDriver(); Statement stmt = conn.createStatement()) {
+ injectDelay = 1000;
+ rs = stmt.executeQuery(String.format(POINT_LOOKUP_SELECT_QUERY, tableName));
+ assertSelectQueryTableMetrics(tableName, true, 0, 0, 1, 0, injectDelay, false, 0, 0,
+ rs);
+ rs.next();
+ }
+ assertSelectQueryTableMetrics(tableName, true, 1, 0, 1, 0, injectDelay, true, 0, 0, rs);
+ }
+
+ @Test public void testTableLevelMetricsForSelectFetchResultsTimeout() throws SQLException {
+ String tableName = generateUniqueName();
+ final int queryTimeout = 10; //seconds
+ ResultSet rs;
+ try (Connection conn = getConnFromTestDriver()) {
+ createTableAndInsertValues(tableName, false, false, 2, true, conn, false);
+ }
+ try (Connection conn = getConnFromTestDriver(); Statement stmt = conn.createStatement()) {
+ stmt.setQueryTimeout(queryTimeout);
+ rs = stmt.executeQuery(String.format(POINT_LOOKUP_SELECT_QUERY, tableName));
+ assertSelectQueryTableMetrics(tableName, true, 0, 0, 1, 0, 0, false, 0, 0, rs);
+ // Make the query time out with a longer delay than the set query timeout value (in ms)
+ MyClock clock = new MyClock(10, queryTimeout * 2 * 1000);
+ EnvironmentEdgeManager.injectEdge(clock);
+ try {
+ rs.next();
+ fail();
+ } catch (SQLException e) {
+ assertEquals(OPERATION_TIMED_OUT.getErrorCode(), e.getErrorCode());
+ }
+ }
+ assertSelectQueryTableMetrics(tableName, true, 0, 1, 1, 0, 0, true, 1, 1, rs);
+ }
+
+ @Test public void testTableLevelMetricsForSelectFetchResultsTimeoutSlowScanner()
+ throws SQLException {
+ String tableName = generateUniqueName();
+ final int queryTimeout = 10; //seconds
+ ResultSet rs;
+ try (Connection conn = getConnFromTestDriver()) {
+ createTableAndInsertValues(tableName, false, false, 10, true, conn, false);
+ }
+ try (Connection conn = getConnFromTestDriver(); Statement stmt = conn.createStatement()) {
+ stmt.setQueryTimeout(queryTimeout);
+ rs = stmt.executeQuery(String.format(POINT_LOOKUP_SELECT_QUERY, tableName));
+ assertSelectQueryTableMetrics(tableName, true, 0, 0, 1, 0, 0, false, 0, 0, rs);
+
+ // Delay the RS scanner by a value greater than the query timeout
+ DelayedOrFailingRegionServer.setDelayEnabled(true);
+ DelayedOrFailingRegionServer.setDelayScan(queryTimeout * 1000 + 1); // ms
+ try {
+ rs.next();
+ fail();
+ } catch (SQLException e) {
+ assertEquals(OPERATION_TIMED_OUT.getErrorCode(), e.getErrorCode());
+ assertTrue(getOverAllReadRequestMetricInfo(rs).get(RESULT_SET_TIME_MS)
+ >= queryTimeout * 1000);
+ }
+ }
+ assertSelectQueryTableMetrics(tableName, true, 0, 1, 1, 0, 0, true, 1, 1, rs);
+ }
+
+ @Test public void testTableLevelMetricsForSelectFetchResultsServerSideFailure()
+ throws SQLException {
+ String tableName = generateUniqueName();
+ ResultSet rs;
+ try (Connection conn = getConnFromTestDriver()) {
+ createTableAndInsertValues(tableName, false, false, 10, true, conn, false);
+ }
+ try (Connection conn = getConnFromTestDriver(); Statement stmt = conn.createStatement()) {
+ rs = stmt.executeQuery(String.format(POINT_LOOKUP_SELECT_QUERY, tableName));
+ assertSelectQueryTableMetrics(tableName, true, 0, 0, 1, 0, 0, false, 0, 0, rs);
+
+ // Inject a failure during the scan operation on the server-side
+ DelayedOrFailingRegionServer.injectFailureForRegionOfTable(tableName);
+ try {
+ while (rs.next()) {
+ // do nothing
+ }
+ fail();
+ } catch (PhoenixIOException e) {
+ Throwable doNotRetryIOException = null;
+ for (Throwable t = e.getCause(); t != null; t = t.getCause()) {
+ if (t instanceof DoNotRetryIOException) {
+ doNotRetryIOException = t;
+ break;
+ }
+ }
+ assertNotNull(doNotRetryIOException);
+ assertTrue(doNotRetryIOException.getMessage().contains(INJECTED_EXCEPTION_STRING));
+ }
+ }
+ assertSelectQueryTableMetrics(tableName, true, 0, 1, 1, 0, 0, true, 1, 0, rs);
+ }
+
+ @Test public void testTableLevelMetricsForUpsert() throws Throwable {
+ String tableName = generateUniqueName();
+ int numRows = 10000;
+ Connection conn = null;
+ Throwable exception = null;
+ try {
+ conn = getConnFromTestDriver();
+ createTableAndInsertValues(tableName, true, false, numRows, true, conn, false);
+ } catch (Throwable t) {
+ exception = t;
+ } finally {
+ // Otherwise the test fails with an error from assertions below instead of the real exception
+ if (exception != null) {
+ throw exception;
+ }
+ assertNotNull("Failed to get a connection!", conn);
+ // Get write metrics before closing the connection since that clears those metrics
+ Map<MetricType, Long>
+ writeMutMetrics =
+ getWriteMetricInfoForMutationsSinceLastReset(conn).get(tableName);
+ conn.close();
+ // Must be asserted after connection close since that's where
+ // we populate table-level metrics
+ assertMutationTableMetrics(true, tableName, numRows, 0, 0, true, numRows, 0, 0, 1, 0,
+ writeMutMetrics, conn);
+ }
+ }
+
+ @Test public void testTableLevelMetricsForBatchUpserts() throws Throwable {
+ String tableName = generateUniqueName();
+ int numRows = 20;
+ Connection conn = null;
+ Throwable exception = null;
+ try {
+ conn = getConnFromTestDriver();
+ createTableAndInsertValues(tableName, true, false, numRows, true, conn, true);
+ } catch (Throwable t) {
+ exception = t;
+ } finally {
+ // Otherwise the test fails with an error from assertions below instead of the real exception
+ if (exception != null) {
+ throw exception;
+ }
+ assertNotNull("Failed to get a connection!", conn);
+ // Get write metrics before closing the connection since that clears those metrics
+ Map<MetricType, Long>
+ writeMutMetrics =
+ getWriteMetricInfoForMutationsSinceLastReset(conn).get(tableName);
+ conn.close();
+ assertMutationTableMetrics(true, tableName, numRows, 0, 0, true, numRows, 0, 0, 1, 0,
+ writeMutMetrics, conn);
+ }
+ }
+
+ @Test public void testTableLevelMetricsAutoCommitTrueUpsert() throws Throwable {
+ String tableName = generateUniqueName();
+ String ddl = String.format(CREATE_TABLE_DDL, tableName, 20);
+ int numRows = 10;
+ try (Connection conn = getConnFromTestDriver(); Statement stmt = conn.createStatement()) {
+ stmt.execute(ddl);
+ }
+ String dml = String.format(UPSERT_DML, tableName);
+ Connection conn = null;
+ Throwable exception = null;
+ try {
+ conn = getConnFromTestDriver();
+ try (PreparedStatement prepStmt = conn.prepareStatement(dml)) {
+ conn.setAutoCommit(true);
+ for (int i = 0; i < numRows; i++) {
+ prepStmt.setString(1, KEY + i);
+ prepStmt.setString(2, VALUE + i);
+ prepStmt.executeUpdate();
+ }
+ }
+ } catch (Throwable t) {
+ exception = t;
+ } finally {
+ // Otherwise the test fails with an error from assertions below instead of the real exception
+ if (exception != null) {
+ throw exception;
+ }
+ assertNotNull("Failed to get a connection!", conn);
+ Map<MetricType, Long>
+ writeMutMetrics =
+ getWriteMetricInfoForMutationsSinceLastReset(conn).get(tableName);
+ conn.close();
+ // Time taken during executeMutation should be longer than the actual
+ // mutation commit time since autoCommit was on
+ assertMutationTableMetrics(true, tableName, numRows, 0,
+ writeMutMetrics.get(UPSERT_COMMIT_TIME), true, numRows, 0, 0, numRows, 0,
+ writeMutMetrics, conn);
+ }
+ }
+
+ @Test public void testTableLevelMetricsforFailingUpsert() throws Throwable {
+ String tableName = generateUniqueName();
+ // Restrict the key to just 2 characters so that we fail later
+ String ddl = String.format(CREATE_TABLE_DDL, tableName, 2);
+ try (Connection conn = getConnFromTestDriver(); Statement stmt = conn.createStatement()) {
+ stmt.execute(ddl);
+ }
+ String dml = String.format(UPSERT_DML, tableName);
+ Connection conn = null;
+ Throwable exception = null;
+ try {
+ conn = getConnFromTestDriver();
+ try (PreparedStatement prepStmt = conn.prepareStatement(dml)) {
+ prepStmt.setString(1, KEY);
+ prepStmt.setString(2, VALUE);
+ try {
+ prepStmt.executeUpdate();
+ fail();
+ } catch (SQLException sqlE) {
+ assertEquals(DATA_EXCEEDS_MAX_CAPACITY.getErrorCode(), sqlE.getErrorCode());
+ }
+ }
+ } catch (Throwable t) {
+ exception = t;
+ } finally {
+ // Otherwise the test fails with an error from assertions below instead of the real exception
+ if (exception != null) {
+ throw exception;
+ }
+ assertNotNull("Failed to get a connection!", conn);
+ conn.close();
+ assertMutationTableMetrics(true, tableName, 0, 1, 0, false, 0, 0, 0, 1, 0, null, conn);
+ }
+ }
+
+ @Test public void testTableLevelMetricsforUpsertSqlTime() throws Throwable {
+ String tableName = generateUniqueName();
+ String ddl = String.format(CREATE_TABLE_DDL, tableName, 10);
+ int numRows = 10;
+ long delay = 300;
+ try (Connection conn = getConnFromTestDriver(); Statement stmt = conn.createStatement()) {
+ stmt.execute(ddl);
+ }
+ String dml = String.format(UPSERT_DML, tableName);
+ Connection conn = null;
+ Throwable exception = null;
+ try {
+ conn = getConnFromTestDriver();
+ try (PreparedStatement prepStmt = conn.prepareStatement(dml)) {
+ MyClock clock = new MyClock(10, delay);
+ EnvironmentEdgeManager.injectEdge(clock);
+ for (int i = 0; i < numRows; i++) {
+ prepStmt.setString(1, KEY + i);
+ prepStmt.setString(2, VALUE + i);
+ prepStmt.executeUpdate();
+ }
+ }
+ conn.commit();
+ } catch (Throwable t) {
+ exception = t;
+ } finally {
+ // Otherwise the test fails with an error from assertions below instead of the real exception
+ if (exception != null) {
+ throw exception;
+ }
+ assertNotNull("Failed to get a connection!", conn);
+ Map<MetricType, Long>
+ writeMutMetrics =
+ getWriteMetricInfoForMutationsSinceLastReset(conn).get(tableName);
+ conn.close();
+ assertMutationTableMetrics(true, tableName, numRows, 0, delay, true, numRows, 0, 0, 1,
+ 0, writeMutMetrics, conn);
+ }
+ }
+
+ @Test public void testTableLevelMetricsUpsertCommitFailedWithAutoCommitTrue() throws Throwable {
+ String tableName = generateUniqueName();
+ String ddl = String.format(CREATE_TABLE_DDL, tableName, 10);
+ int numRows = 10;
+ try (Connection conn = getConnFromTestDriver(); Statement stmt = conn.createStatement()) {
+ stmt.execute(ddl);
+ }
+ String dml = String.format(UPSERT_DML, tableName);
+ Connection conn = null;
+ Throwable exception = null;
+ try {
+ conn = getConnFromTestDriver();
+ conn.setAutoCommit(true);
+ DelayedOrFailingRegionServer.injectFailureForRegionOfTable(tableName);
+ try (PreparedStatement prepStmt = conn.prepareStatement(dml)) {
+ for (int i = 0; i < numRows; i++) {
+ prepStmt.setString(1, KEY + i);
+ prepStmt.setString(2, VALUE + i);
+ prepStmt.executeUpdate();
+ }
+ }
+ } catch (CommitException e) {
+ Throwable retriesExhaustedEx = null;
+ for (Throwable t = e.getCause(); t != null; t = t.getCause()) {
+ if (t instanceof RetriesExhaustedWithDetailsException) {
+ retriesExhaustedEx = t;
+ break;
+ }
+ }
+ assertNotNull(retriesExhaustedEx);
+ assertTrue(retriesExhaustedEx.getMessage().contains(INJECTED_EXCEPTION_STRING));
+ } catch (Throwable t) {
+ exception = t;
+ } finally {
+ // Otherwise the test fails with an error from assertions below instead of the real exception
+ if (exception != null) {
+ throw exception;
+ }
+ assertNotNull("Failed to get a connection!", conn);
+ Map<MetricType, Long>
+ writeMutMetrics =
+ getWriteMetricInfoForMutationsSinceLastReset(conn).get(tableName);
+ conn.close();
+ assertMutationTableMetrics(true, tableName, 0, 1, 0, true, 1, 0, 1, 0, 1,
+ writeMutMetrics, conn);
+ }
+ }
+
+ @Test public void testTableLevelMetricsUpsertCommitFailed() throws Throwable {
+ String tableName = generateUniqueName();
+ String ddl = String.format(CREATE_TABLE_DDL, tableName, 10);
+ int numRows = 10;
+ try (Connection conn = getConnFromTestDriver(); Statement stmt = conn.createStatement()) {
+ stmt.execute(ddl);
+ }
+ String dml = String.format(UPSERT_DML, tableName);
+ Connection conn = null;
+ Throwable exception = null;
+ try {
+ conn = getConnFromTestDriver();
+ try (PreparedStatement prepStmt = conn.prepareStatement(dml)) {
+ for (int i = 0; i < numRows; i++) {
+ prepStmt.setString(1, KEY + i);
+ prepStmt.setString(2, VALUE + i);
+ prepStmt.executeUpdate();
+ }
+ }
+ DelayedOrFailingRegionServer.injectFailureForRegionOfTable(tableName);
+ try {
+ conn.commit();
+ fail();
+ } catch (CommitException e) {
+ Throwable retriesExhaustedEx = null;
+ for (Throwable t = e.getCause(); t != null; t = t.getCause()) {
+ if (t instanceof RetriesExhaustedWithDetailsException) {
+ retriesExhaustedEx = t;
+ break;
+ }
+ }
+ assertNotNull(retriesExhaustedEx);
+ assertTrue(retriesExhaustedEx.getMessage().contains(INJECTED_EXCEPTION_STRING));
+ }
+ } catch (Throwable t) {
+ exception = t;
+ } finally {
+ // Otherwise the test fails with an error from assertions below instead of the real exception
+ if (exception != null) {
+ throw exception;
+ }
+ assertNotNull("Failed to get a connection!", conn);
+ Map<MetricType, Long>
+ writeMutMetrics =
+ getWriteMetricInfoForMutationsSinceLastReset(conn).get(tableName);
+ conn.close();
+ assertMutationTableMetrics(true, tableName, numRows, 0, 0, true, numRows, 0, numRows, 0,
+ 1, writeMutMetrics, conn);
+ }
+ }
+
+ @Test public void testUpsertCommitTimeSlowRS() throws Throwable {
+ String tableName = generateUniqueName();
+ String ddl = String.format(CREATE_TABLE_DDL, tableName, 10);
+ int numRows = 10;
+ final int delayRs = 5000; // ms
+ try (Connection conn = getConnFromTestDriver(); Statement stmt = conn.createStatement()) {
+ stmt.execute(ddl);
+ }
+ String dml = String.format(UPSERT_DML, tableName);
+ Connection conn = null;
+ Throwable exception = null;
+ try {
+ conn = getConnFromTestDriver();
+ try (PreparedStatement prepStmt = conn.prepareStatement(dml)) {
+ for (int i = 0; i < numRows; i++) {
+ prepStmt.setString(1, KEY + i);
+ prepStmt.setString(2, VALUE + i);
+ prepStmt.executeUpdate();
+ }
+ }
+ DelayedOrFailingRegionServer.setDelayEnabled(true);
+ DelayedOrFailingRegionServer.setDelayMultiOp(delayRs);
+ conn.commit();
+ } catch (Throwable t) {
+ exception = t;
+ } finally {
+ // Otherwise the test fails with an error from assertions below instead of the real exception
+ if (exception != null) {
+ throw exception;
+ }
+ assertNotNull("Failed to get a connection!", conn);
+ Map<MetricType, Long>
+ writeMutMetrics =
+ getWriteMetricInfoForMutationsSinceLastReset(conn).get(tableName);
+ conn.close();
+ assertMutationTableMetrics(true, tableName, numRows, 0, 0, true, numRows, delayRs, 0, 1,
+ 0, writeMutMetrics, conn);
+ }
+ }
+
+ @Test public void testTableLevelMetricsForPointDelete() throws Throwable {
+ String tableName = generateUniqueName();
+ int numRows = 15;
+ Connection conn = null;
+ Throwable exception = null;
+ try {
+ conn = getConnFromTestDriver();
+ createTableAndInsertValues(tableName, true, true, numRows, true, conn, false);
+ // Reset metrics from the upserts
+ PhoenixRuntime.resetMetrics(conn);
+ clearTableLevelMetrics();
+ doPointDeleteFromTable(tableName, conn);
+ conn.commit();
+ } catch (Throwable t) {
+ exception = t;
+ } finally {
+ // Otherwise the test fails with an error from assertions below instead of the real exception
+ if (exception != null) {
+ throw exception;
+ }
+ assertNotNull("Failed to get a connection!", conn);
+ // Get write metrics before closing the connection since that clears those metrics
+ Map<MetricType, Long>
+ writeMutMetrics =
+ getWriteMetricInfoForMutationsSinceLastReset(conn).get(tableName);
+ conn.close();
+ assertMutationTableMetrics(false, tableName, 1, 0, 0, true, 1, 0, 0, 1, 0,
+ writeMutMetrics, conn);
+ }
+ }
+
+ @Test public void testTableLevelMetricsForDeleteAll() throws Throwable {
+ String tableName = generateUniqueName();
+ int numRows = 15;
+ Connection conn = null;
+ Throwable exception = null;
+ try {
+ conn = getConnFromTestDriver();
+ createTableAndInsertValues(tableName, true, true, numRows, true, conn, false);
+ // Reset metrics from the upserts
+ PhoenixRuntime.resetMetrics(conn);
+ clearTableLevelMetrics();
+ doDeleteAllFromTable(tableName, conn);
+ conn.commit();
+ } catch (Throwable t) {
+ exception = t;
+ } finally {
+ // Otherwise the test fails with an error from assertions below instead of the real exception
+ if (exception != null) {
+ throw exception;
+ }
+ assertNotNull("Failed to get a connection!", conn);
+ // Get write metrics before closing the connection since that clears those metrics
+ Map<MetricType, Long>
+ writeMutMetrics =
+ getWriteMetricInfoForMutationsSinceLastReset(conn).get(tableName);
+ conn.close();
+ assertMutationTableMetrics(false, tableName, 1, 0, 0, true, numRows, 0, 0, 1, 0,
+ writeMutMetrics, conn);
+ }
+ }
+
+ @Test public void testTableLevelMetricsAutoCommitTrueDelete() throws Throwable {
+ String tableName = generateUniqueName();
+ int numRows = 15;
+ Connection conn = null;
+ Throwable exception = null;
+ try (Connection ddlAndUpsertConn = getConnFromTestDriver()) {
+ createTableAndInsertValues(tableName, true, true, numRows, true, ddlAndUpsertConn,
+ false);
+ // Reset metrics from the upserts
+ PhoenixRuntime.resetMetrics(ddlAndUpsertConn);
+ clearTableLevelMetrics();
+ }
+ try {
+ conn = getConnFromTestDriver();
+ conn.setAutoCommit(true);
+ doPointDeleteFromTable(tableName, conn);
+ } catch (Throwable t) {
+ exception = t;
+ } finally {
+ // Otherwise the test fails with an error from assertions below instead of the real exception
+ if (exception != null) {
+ throw exception;
+ }
+ assertNotNull("Failed to get a connection!", conn);
+ // Get write metrics before closing the connection since that clears those metrics
+ Map<MetricType, Long>
+ writeMutMetrics =
+ getWriteMetricInfoForMutationsSinceLastReset(conn).get(tableName);
+ // When autoCommit = true, deletes happen on the server and so mutation metrics are not
+ // accumulated for those mutations
+ assertNull(writeMutMetrics);
+ conn.close();
+ assertMutationTableMetrics(false, tableName, 1, 0, 0, false, 0, 0, 0, 0, 0,
+ writeMutMetrics, conn);
+ }
+ }
+
+ @Test public void testTableLevelMetricsforFailingDelete() throws Throwable {
+ String tableName = generateUniqueName();
+ int numRows = 15;
+ Connection conn = null;
+ Throwable exception = null;
+ try {
+ conn = getConnFromTestDriver();
+ createTableAndInsertValues(tableName, true, true, numRows, true, conn, false);
+ // Reset metrics from the upserts
+ PhoenixRuntime.resetMetrics(conn);
+ clearTableLevelMetrics();
+
+ failExecuteQueryAndClientSideDeletes = true;
+ try {
+ doPointDeleteFromTable(tableName, conn);
+ fail();
+ } catch (SQLException sqlE) {
+ assertEquals(GET_TABLE_REGIONS_FAIL.getErrorCode(), sqlE.getErrorCode());
+ }
+ } catch (Throwable t) {
+ exception = t;
+ } finally {
+ // Otherwise the test fails with an error from assertions below instead of the real exception
+ if (exception != null) {
+ throw exception;
+ }
+ assertNotNull("Failed to get a connection!", conn);
+ // Get write metrics before closing the connection since that clears those metrics
+ Map<MetricType, Long>
+ writeMutMetrics =
+ getWriteMetricInfoForMutationsSinceLastReset(conn).get(tableName);
+ assertNull(writeMutMetrics);
+ conn.close();
+ assertMutationTableMetrics(false, tableName, 0, 1, 0, false, 0, 0, 0, 0, 1, null, conn);
+ }
+ }
+
+ @Test public void testTableLevelMetricsforDelayedDeleteQuery() throws Throwable {
+ String tableName = generateUniqueName();
+ Connection conn = null;
+ Throwable exception = null;
+ try {
+ conn = getConnFromTestDriver();
+ createTableAndInsertValues(tableName, false, true, 10, true, conn, false);
+ // Reset metrics from the upserts
+ PhoenixRuntime.resetMetrics(conn);
+ clearTableLevelMetrics();
+
+ injectDelay = 3000;
+ doPointDeleteFromTable(tableName, conn);
+ conn.commit();
+ } catch (Throwable t) {
+ exception = t;
+ } finally {
+ // Otherwise the test fails with an error from assertions below instead of the real exception
+ if (exception != null) {
+ throw exception;
+ }
+ assertNotNull("Failed to get a connection!", conn);
+ // Get write metrics before closing the connection since that clears those metrics
+ Map<MetricType, Long>
+ writeMutMetrics =
+ getWriteMetricInfoForMutationsSinceLastReset(conn).get(tableName);
+ conn.close();
+ assertMutationTableMetrics(false, tableName, 1, 0, injectDelay, true, 1, 0, 0, 1, 0,
+ writeMutMetrics, conn);
+ }
+ }
+
+ @Test public void testTableLevelMetricsDeleteCommitFailed() throws Throwable {
+ String tableName = generateUniqueName();
+ int numRows = 15;
+ Connection conn = null;
+ Throwable exception = null;
+ try {
+ conn = getConnFromTestDriver();
+ createTableAndInsertValues(tableName, true, true, numRows, true, conn, false);
+ // Reset metrics from the upserts
+ PhoenixRuntime.resetMetrics(conn);
+ clearTableLevelMetrics();
+ doDeleteAllFromTable(tableName, conn);
+
+ DelayedOrFailingRegionServer.injectFailureForRegionOfTable(tableName);
+ try {
+ conn.commit();
+ fail();
+ } catch (CommitException e) {
+ Throwable retriesExhaustedEx = null;
+ for (Throwable t = e.getCause(); t != null; t = t.getCause()) {
+ if (t instanceof RetriesExhaustedWithDetailsException) {
+ retriesExhaustedEx = t;
+ break;
+ }
+ }
+ assertNotNull(retriesExhaustedEx);
+ assertTrue(retriesExhaustedEx.getMessage().contains(INJECTED_EXCEPTION_STRING));
+ }
+ } catch (Throwable t) {
+ exception = t;
+ } finally {
+ // Otherwise the test fails with an error from assertions below instead of the real exception
+ if (exception != null) {
+ throw exception;
+ }
+ assertNotNull("Failed to get a connection!", conn);
+ // Get write metrics before closing the connection since that clears those metrics
+ Map<MetricType, Long>
+ writeMutMetrics =
+ getWriteMetricInfoForMutationsSinceLastReset(conn).get(tableName);
+ conn.close();
+ assertMutationTableMetrics(false, tableName, 1, 0, 0, true, numRows, 0, numRows, 0, 1,
+ writeMutMetrics, conn);
+ }
+ }
+
+ @Test public void testDeleteCommitTimeSlowRS() throws Throwable {
+ String tableName = generateUniqueName();
+ int numRows = 15;
+ Connection conn = null;
+ Throwable exception = null;
+ final int delayRs = 5000; //ms
+ try {
+ conn = getConnFromTestDriver();
+ createTableAndInsertValues(tableName, true, true, numRows, true, conn, false);
+ // Reset metrics from the upserts
+ PhoenixRuntime.resetMetrics(conn);
+ clearTableLevelMetrics();
+ doDeleteAllFromTable(tableName, conn);
+
+ DelayedOrFailingRegionServer.setDelayEnabled(true);
+ DelayedOrFailingRegionServer.setDelayMultiOp(delayRs);
+ conn.commit();
+ } catch (Throwable t) {
+ exception = t;
+ } finally {
+ // Otherwise the test fails with an error from assertions below instead of the real exception
+ if (exception != null) {
+ throw exception;
+ }
+ assertNotNull("Failed to get a connection!", conn);
+ // Get write metrics before closing the connection since that clears those metrics
+ Map<MetricType, Long>
+ writeMutMetrics =
+ getWriteMetricInfoForMutationsSinceLastReset(conn).get(tableName);
+ conn.close();
+ assertMutationTableMetrics(false, tableName, 1, 0, 0, true, numRows, delayRs, 0, 1, 0,
+ writeMutMetrics, conn);
+ }
+ }
+
+ private Connection getConnFromTestDriver() throws SQLException {
+ Connection conn = DriverManager.getConnection(url);
+ assertTrue(conn.unwrap(PhoenixConnection.class)
+ .getQueryServices() instanceof PhoenixMetricsTestingQueryServices);
+ return conn;
+ }
+
+ private long getMetricFromTableMetrics(String tableName, MetricType type) {
+ Long value = TableMetricsManager.getMetricValue(tableName, type);
+ Assert.assertNotNull(value);
+ return value;
+ }
+
+ private enum CompareOp {
+ LT, EQ, GT, LTEQ, GTEQ
+ }
+
+ private static class MyClock extends EnvironmentEdge {
+ private final long delay;
+ private AtomicLong time;
+
+ public MyClock(long time, long delay) {
+ this.time = new AtomicLong(time);
+ this.delay = delay;
+ }
+
+ @Override public long currentTime() {
+ long currentTime = this.time.get();
+ this.time.addAndGet(this.delay);
+ return currentTime;
+ }
+ }
+
+ /**
+ * Custom QueryServices object which we can use to inject failures and delays
+ */
+ private static class PhoenixMetricsTestingQueryServices extends ConnectionQueryServicesImpl {
+
+ PhoenixMetricsTestingQueryServices(QueryServices services,
+ PhoenixEmbeddedDriver.ConnectionInfo connectionInfo, Properties info) {
+ super(services, connectionInfo, info);
+ }
+
+ // Make plan.iterator() fail (ultimately calls CQSI.getAllTableRegions())
+ @Override public List<HRegionLocation> getAllTableRegions(byte[] tableName)
+ throws SQLException {
+ if (failExecuteQueryAndClientSideDeletes) {
+ throw new SQLExceptionInfo.Builder(GET_TABLE_REGIONS_FAIL).build().buildException();
+ }
+ try {
+ Thread.sleep(injectDelay);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ return super.getAllTableRegions(tableName);
+ }
+
+ }
+
+ /**
+ * Custom driver to return a custom QueryServices object
+ */
+ public static class PhoenixMetricsTestingDriver extends PhoenixTestDriver {
+ private ConnectionQueryServices cqs;
+ private ReadOnlyProps overrideProps;
+
+ public PhoenixMetricsTestingDriver(ReadOnlyProps props) {
+ overrideProps = props;
+ }
+
+ @Override public boolean acceptsURL(String url) {
+ return true;
+ }
+
+ @Override public synchronized ConnectionQueryServices getConnectionQueryServices(String url,
+ Properties info) throws SQLException {
+ if (cqs == null) {
+ cqs =
+ new PhoenixMetricsTestingQueryServices(
+ new QueryServicesTestImpl(getDefaultProps(), overrideProps),
+ ConnectionInfo.create(url), info);
+ cqs.init(url, info);
+ }
+ return cqs;
+ }
+ }
+}
\ No newline at end of file
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/util/DelayedOrFailingRegionServer.java b/phoenix-core/src/it/java/org/apache/phoenix/util/DelayedOrFailingRegionServer.java
new file mode 100644
index 0000000..7307072
--- /dev/null
+++ b/phoenix-core/src/it/java/org/apache/phoenix/util/DelayedOrFailingRegionServer.java
@@ -0,0 +1,154 @@
+/*
+ * 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.util;
+
+import com.google.protobuf.RpcController;
+import com.google.protobuf.ServiceException;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.CoordinatedStateManager;
+import org.apache.hadoop.hbase.DoNotRetryIOException;
+import org.apache.hadoop.hbase.protobuf.generated.ClientProtos;
+import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos;
+import org.apache.hadoop.hbase.regionserver.HRegionServer;
+import org.apache.hadoop.hbase.regionserver.RSRpcServices;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+
+/**
+ * This is an extended DelayedRegionServer which also allows injecting failure for specific
+ * server-side operations for testing purposes.
+ */
+public class DelayedOrFailingRegionServer extends DelayedRegionServer {
+
+ private static String injectFailureForRegionOfTable = null;
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(DelayedOrFailingRegionServer.class);
+
+ public static final String INJECTED_EXCEPTION_STRING = "Injected exception message";
+
+ public DelayedOrFailingRegionServer(Configuration conf, CoordinatedStateManager cp)
+ throws IOException, InterruptedException {
+ super(conf, cp);
+ }
+
+ @Override
+ protected RSRpcServices createRpcServices() throws IOException {
+ return new DelayedOrFailingRSRpcServices(this);
+ }
+
+ public static void injectFailureForRegionOfTable(String tableName) {
+ injectFailureForRegionOfTable = tableName;
+ }
+
+ /**
+ * This class injects failure for RPC calls or executes super methods if failure injection is
+ * disabled
+ */
+ public static class DelayedOrFailingRSRpcServices extends DelayedRSRpcServices {
+
+ DelayedOrFailingRSRpcServices(HRegionServer rs) throws IOException {
+ super(rs);
+ }
+
+ @Override
+ public ClientProtos.GetResponse get(RpcController controller,
+ ClientProtos.GetRequest request) throws ServiceException {
+ optionallyInjectFailureIfRegionBelongsToTable(request.getRegion(), "get");
+ return super.get(controller, request);
+ }
+
+ @Override
+ public ClientProtos.MutateResponse mutate(RpcController rpcc,
+ ClientProtos.MutateRequest request) throws ServiceException {
+ optionallyInjectFailureIfRegionBelongsToTable(request.getRegion(), "mutate");
+ return super.mutate(rpcc, request);
+ }
+
+ @Override
+ public ClientProtos.MultiResponse multi(RpcController rpcc,
+ ClientProtos.MultiRequest request) throws ServiceException {
+ for (ClientProtos.RegionAction req : request.getRegionActionList()) {
+ try {
+ if (doesRegionBelongToTable(req.getRegion())) {
+ boolean isBatchOfOnlyPuts = true;
+ boolean isBatchOfOnlyDeletes = true;
+ for (ClientProtos.Action action : req.getActionList()) {
+ ClientProtos.MutationProto.MutationType type =
+ action.getMutation().getMutateType();
+ if (type.equals(ClientProtos.MutationProto.MutationType.PUT)) {
+ isBatchOfOnlyDeletes = false;
+ } else if (type.equals(ClientProtos.MutationProto.MutationType.DELETE)) {
+ isBatchOfOnlyPuts = false;
+ } else {
+ isBatchOfOnlyPuts = false;
+ isBatchOfOnlyDeletes = false;
+ }
+ }
+ if (isBatchOfOnlyPuts) {
+ optionallyInjectFailureIfRegionBelongsToTable(req.getRegion(), "put");
+ } else if (isBatchOfOnlyDeletes) {
+ optionallyInjectFailureIfRegionBelongsToTable(req.getRegion(), "delete");
+ }
+ }
+ } catch (IOException e) {
+ LOGGER.error("Mutate failed with exception: ", e);
+ }
+ }
+ return super.multi(rpcc, request);
+ }
+
+ @Override
+ public ClientProtos.ScanResponse scan(RpcController controller,
+ ClientProtos.ScanRequest request) throws ServiceException {
+ optionallyInjectFailureIfRegionBelongsToTable(request.getRegion(), "scan");
+ return super.scan(controller, request);
+ }
+
+ /**
+ * If {@link DelayedOrFailingRegionServer#injectFailureForRegionOfTable} matches the name
+ * of the table hosted by this RS, then we throw an exception which causes immediate
+ * server-side failure
+ * @param specifier region specifier
+ * @param op scan/get/mutate operation
+ * @throws ServiceException exception to be thrown
+ */
+ private void optionallyInjectFailureIfRegionBelongsToTable(
+ HBaseProtos.RegionSpecifier specifier, String op) throws ServiceException {
+ try {
+ if (doesRegionBelongToTable(specifier)) {
+ // We have to throw a DoNotRetryIOException, otherwise retries will lead
+ // to timeouts rather than immediate failure
+ throw new ServiceException(new DoNotRetryIOException(INJECTED_EXCEPTION_STRING));
+ }
+ } catch (IOException e) {
+ LOGGER.error("Not injecting failure since getting region during " + op
+ + " failed with: ", e);
+ }
+ }
+
+ private boolean doesRegionBelongToTable(HBaseProtos.RegionSpecifier specifier)
+ throws IOException {
+ return injectFailureForRegionOfTable != null && this.getRegion(specifier)
+ .getTableDesc().getTableName().getNameAsString()
+ .equals(injectFailureForRegionOfTable);
+ }
+ }
+
+}
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/util/DelayedRegionServer.java b/phoenix-core/src/it/java/org/apache/phoenix/util/DelayedRegionServer.java
index bf02621..ba7ce3d 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/util/DelayedRegionServer.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/util/DelayedRegionServer.java
@@ -45,6 +45,7 @@ public class DelayedRegionServer extends MiniHBaseCluster.MiniHBaseClusterRegion
private static int DELAY_GET = 0;
private static int DELAY_SCAN = 30000;
private static int DELAY_MUTATE = 0;
+ private static int DELAY_MULTI_OP = 0;
public static void setDelayEnabled(boolean delay) {
doDelay = delay;
@@ -62,6 +63,10 @@ public class DelayedRegionServer extends MiniHBaseCluster.MiniHBaseClusterRegion
DELAY_MUTATE = delayMutate;
}
+ public static void setDelayMultiOp(int delayMultiOp) {
+ DELAY_MULTI_OP = delayMultiOp;
+ }
+
public DelayedRegionServer(Configuration conf, CoordinatedStateManager cp)
throws IOException, InterruptedException {
super(conf, cp);
@@ -104,6 +109,19 @@ public class DelayedRegionServer extends MiniHBaseCluster.MiniHBaseClusterRegion
return super.mutate(rpcc, request);
}
+ @Override
+ public ClientProtos.MultiResponse multi(RpcController rpcc,
+ ClientProtos.MultiRequest request) throws ServiceException {
+ try {
+ if (doDelay) {
+ Thread.sleep(DELAY_MULTI_OP);
+ }
+ } catch (InterruptedException e) {
+ LOGGER.error("Sleep interrupted during multi operation", e);
+ }
+ return super.multi(rpcc, request);
+ }
+
@Override public ClientProtos.ScanResponse scan(RpcController controller,
ClientProtos.ScanRequest request) throws ServiceException {
try {
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/execute/MutationState.java b/phoenix-core/src/main/java/org/apache/phoenix/execute/MutationState.java
index b3b2fee..1efbcf0 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/execute/MutationState.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/execute/MutationState.java
@@ -18,6 +18,10 @@
package org.apache.phoenix.execute;
import static com.google.common.base.Preconditions.checkNotNull;
+import static org.apache.phoenix.monitoring.MetricType.DELETE_AGGREGATE_FAILURE_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.DELETE_AGGREGATE_SUCCESS_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.UPSERT_AGGREGATE_FAILURE_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.UPSERT_AGGREGATE_SUCCESS_SQL_COUNTER;
import static org.apache.phoenix.query.QueryServices.SOURCE_OPERATION_ATTRIB;
import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_MUTATION_BATCH_FAILED_COUNT;
import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_MUTATION_BATCH_SIZE;
@@ -30,6 +34,7 @@ import static org.apache.phoenix.query.QueryServicesOptions.DEFAULT_WILDCARD_QUE
import java.io.IOException;
import java.sql.SQLException;
import java.sql.Timestamp;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
@@ -43,6 +48,8 @@ import java.util.Set;
import javax.annotation.Nonnull;
import javax.annotation.concurrent.Immutable;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Strings;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Mutation;
@@ -76,6 +83,7 @@ import org.apache.phoenix.monitoring.MutationMetricQueue;
import org.apache.phoenix.monitoring.MutationMetricQueue.MutationMetric;
import org.apache.phoenix.monitoring.MutationMetricQueue.NoOpMutationMetricsQueue;
import org.apache.phoenix.monitoring.ReadMetricQueue;
+import org.apache.phoenix.monitoring.TableMetricsManager;
import org.apache.phoenix.query.QueryConstants;
import org.apache.phoenix.query.QueryServices;
import org.apache.phoenix.query.QueryServicesOptions;
@@ -156,6 +164,15 @@ public class MutationState implements SQLCloseable {
private final MutationMetricQueue mutationMetricQueue;
private ReadMetricQueue readMetricQueue;
+ private static boolean allUpsertsMutations = true;
+ private static boolean allDeletesMutations = true;
+
+ @VisibleForTesting
+ public static void resetAllMutationState(){
+ allDeletesMutations = true;
+ allUpsertsMutations = true;
+ }
+
public MutationState(int maxSize, long maxSizeBytes, PhoenixConnection connection) {
this(maxSize, maxSizeBytes, connection, false, null);
}
@@ -933,15 +950,33 @@ public class MutationState implements SQLCloseable {
return serverTimeStamp == QueryConstants.UNSET_TIMESTAMP ? HConstants.LATEST_TIMESTAMP : serverTimeStamp;
}
- private static long calculateMutationSize(List<Mutation> mutations) {
+ static MutationBytes calculateMutationSize(List<Mutation> mutations, boolean updateGlobalClientMetrics ) {
long byteSize = 0;
+ long tempSize;
+ long deleteSize = 0, deleteCounter = 0;
+ long upsertsize = 0, upsertCounter = 0;
if (GlobalClientMetrics.isMetricsEnabled()) {
for (Mutation mutation : mutations) {
- byteSize += KeyValueUtil.calculateMutationDiskSize(mutation);
+ tempSize = KeyValueUtil.calculateMutationDiskSize(mutation);
+ byteSize += tempSize;
+ if(mutation instanceof Delete ) {
+ deleteSize += tempSize;
+ deleteCounter++;
+ allUpsertsMutations = false;
+ }else if(mutation instanceof Put) {
+ upsertsize += tempSize;
+ upsertCounter++;
+ allDeletesMutations = false;
+ } else {
+ allUpsertsMutations = false;
+ allDeletesMutations = false;
+ }
}
}
- GLOBAL_MUTATION_BYTES.update(byteSize);
- return byteSize;
+ if (updateGlobalClientMetrics) {
+ GLOBAL_MUTATION_BYTES.update(byteSize);
+ }
+ return new MutationBytes(deleteCounter, deleteSize, byteSize, upsertCounter, upsertsize);
}
public long getBatchSizeBytes() {
@@ -952,6 +987,45 @@ public class MutationState implements SQLCloseable {
return batchCount;
}
+ public static final class MutationBytes {
+
+ private long deleteMutationCounter;
+ private long deleteMutationBytes;
+ private long totalMutationBytes;
+ private long upsertMutationCounter;
+ private long upsertMutationBytes;
+
+ public MutationBytes(long deleteMutationCounter, long deleteMutationBytes, long totalMutationBytes, long
+ upsertMutationCounter, long upsertMutationBytes) {
+ this.deleteMutationCounter = deleteMutationCounter;
+ this.deleteMutationBytes = deleteMutationBytes;
+ this.totalMutationBytes = totalMutationBytes;
+ this.upsertMutationCounter = upsertMutationCounter;
+ this.upsertMutationBytes = upsertMutationBytes;
+ }
+
+
+ public long getDeleteMutationCounter() {
+ return deleteMutationCounter;
+ }
+
+ public long getDeleteMutationBytes() {
+ return deleteMutationBytes;
+ }
+
+ public long getTotalMutationBytes() {
+ return totalMutationBytes;
+ }
+
+ public long getUpsertMutationCounter() {
+ return upsertMutationCounter;
+ }
+
+ public long getUpsertMutationBytes() {
+ return upsertMutationBytes;
+ }
+ }
+
public enum MutationMetadataType {
TENANT_ID,
SCHEMA_NAME,
@@ -1183,6 +1257,7 @@ public class MutationState implements SQLCloseable {
Entry<TableInfo, List<Mutation>> pair = mutationsIterator.next();
TableInfo tableInfo = pair.getKey();
byte[] htableName = tableInfo.getHTableName().getBytes();
+ String htableNameStr = tableInfo.getHTableName().getString();
List<Mutation> mutationList = pair.getValue();
List<List<Mutation>> mutationBatchList =
getMutationBatchList(batchSize, batchSizeBytes, mutationList);
@@ -1194,12 +1269,13 @@ public class MutationState implements SQLCloseable {
int retryCount = 0;
boolean shouldRetry = false;
long numMutations = 0;
- long mutationSizeBytes = 0;
long mutationCommitTime = 0;
long numFailedMutations = 0;
long numFailedPhase3Mutations = 0;
- long startTime = 0;
+
+ long startTime = EnvironmentEdgeManager.currentTimeMillis();
+ MutationBytes totalMutationBytesObject = null;
boolean shouldRetryIndexedMutation = false;
IndexWriteException iwe = null;
do {
@@ -1215,6 +1291,8 @@ public class MutationState implements SQLCloseable {
shouldRetry = cache != null;
SQLException sqlE = null;
Table hTable = connection.getQueryServices().getTable(htableName);
+ List<Mutation> currentMutationBatch = null;
+ boolean areAllBatchesSuccessful = false;
try {
if (table.isTransactional()) {
// Track tables to which we've sent uncommitted data
@@ -1231,13 +1309,14 @@ public class MutationState implements SQLCloseable {
}
numMutations = mutationList.size();
GLOBAL_MUTATION_BATCH_SIZE.update(numMutations);
- mutationSizeBytes = calculateMutationSize(mutationList);
+ totalMutationBytesObject = calculateMutationSize(mutationList, true);
startTime = EnvironmentEdgeManager.currentTimeMillis();
child.addTimelineAnnotation("Attempt " + retryCount);
Iterator<List<Mutation>> itrListMutation = mutationBatchList.iterator();
while (itrListMutation.hasNext()) {
final List<Mutation> mutationBatch = itrListMutation.next();
+ currentMutationBatch = mutationBatch;
if (shouldRetryIndexedMutation) {
// if there was an index write failure, retry the mutation in a loop
final Table finalHTable = hTable;
@@ -1316,8 +1395,8 @@ public class MutationState implements SQLCloseable {
// recalculate the estimated size
estimatedSize = KeyValueUtil.getEstimatedRowMutationSizeWithBatch(this.mutationsMap);
}
+ areAllBatchesSuccessful = true;
} catch (Exception e) {
- mutationCommitTime = EnvironmentEdgeManager.currentTimeMillis() - startTime;
long serverTimestamp = ServerUtil.parseServerTimestamp(e);
SQLException inferredE = ServerUtil.parseServerExceptionOrNull(e);
if (inferredE != null) {
@@ -1374,9 +1453,45 @@ public class MutationState implements SQLCloseable {
GLOBAL_MUTATION_INDEX_COMMIT_FAILURE_COUNT.update(numFailedPhase3Mutations);
}
} finally {
- MutationMetric mutationsMetric = new MutationMetric(numMutations, mutationSizeBytes,
- mutationCommitTime, numFailedMutations, numFailedPhase3Mutations);
- mutationMetricQueue.addMetricsForTable(Bytes.toString(htableName), mutationsMetric);
+ mutationCommitTime = EnvironmentEdgeManager.currentTimeMillis() - startTime;
+ GLOBAL_MUTATION_COMMIT_TIME.update(mutationCommitTime);
+ MutationMetric failureMutationMetrics = MutationMetric.EMPTY_METRIC;
+ if (!areAllBatchesSuccessful) {
+ failureMutationMetrics =
+ updateMutationBatchFailureMetrics(currentMutationBatch,
+ htableNameStr, numFailedMutations,
+ table.isTransactional());
+ }
+
+ MutationMetric committedMutationsMetric =
+ getCommittedMutationsMetric(
+ totalMutationBytesObject,
+ mutationBatchList,
+ numMutations,
+ numFailedMutations,
+ numFailedPhase3Mutations,
+ mutationCommitTime);
+ // Combine failure mutation metrics with committed ones for the final picture
+ committedMutationsMetric.combineMetric(failureMutationMetrics);
+ mutationMetricQueue.addMetricsForTable(htableNameStr, committedMutationsMetric);
+
+ // updated only for all upsertMutations or all DeleteMutations
+ if (allUpsertsMutations ^ allDeletesMutations) {
+ //success cases are updated for both cases autoCommit=true and conn.commit explicit
+ if(areAllBatchesSuccessful){
+ TableMetricsManager
+ .updateMetricsMethod(htableNameStr, allUpsertsMutations ? UPSERT_AGGREGATE_SUCCESS_SQL_COUNTER :
+ DELETE_AGGREGATE_SUCCESS_SQL_COUNTER, 1);
+ }
+ //Failures cases are updated only for conn.commit explicit case.
+ if(!areAllBatchesSuccessful && !connection.getAutoCommit()){
+ TableMetricsManager.updateMetricsMethod(htableNameStr, allUpsertsMutations ? UPSERT_AGGREGATE_FAILURE_SQL_COUNTER :
+ DELETE_AGGREGATE_FAILURE_SQL_COUNTER, 1);
+ }
+
+ }
+ resetAllMutationState();
+
try {
if (cache != null) cache.close();
} finally {
@@ -1396,6 +1511,118 @@ public class MutationState implements SQLCloseable {
}
}
+ /**
+ * Update metrics related to failed mutations
+ * @param failedMutationBatch the batch of mutations that failed
+ * @param tableName table that was to be mutated
+ * @param numFailedMutations total number of failed mutations
+ * @param isTransactional true if the table is transactional
+ */
+ @VisibleForTesting
+ public static MutationMetricQueue.MutationMetric updateMutationBatchFailureMetrics(
+ List<Mutation> failedMutationBatch,
+ String tableName,
+ long numFailedMutations,
+ boolean isTransactional) {
+ if (failedMutationBatch == null || failedMutationBatch.isEmpty() ||
+ Strings.isNullOrEmpty(tableName)) {
+ return MutationMetricQueue.MutationMetric.EMPTY_METRIC;
+ }
+ long numUpsertMutationsInBatch = 0L;
+ long numDeleteMutationsInBatch = 0L;
+
+ for (Mutation m : failedMutationBatch) {
+ if (m instanceof Put) {
+ numUpsertMutationsInBatch++;
+ } else if (m instanceof Delete) {
+ numDeleteMutationsInBatch++;
+ }
+ }
+ // Update the MUTATION_BATCH_FAILED_SIZE counter with the number of failed delete mutations
+ // in case we are dealing with all deletes for a non-transactional table, since there is a
+ // bug in sendMutations where we don't get the correct value for numFailedMutations when
+ // we don't use transactions
+ return new MutationMetricQueue.MutationMetric(0, 0, 0, 0, 0,
+ allDeletesMutations && !isTransactional ? numDeleteMutationsInBatch : numFailedMutations,
+ 0, 0, 0, 0,
+ numUpsertMutationsInBatch,
+ allUpsertsMutations ? 1 : 0,
+ numDeleteMutationsInBatch,
+ allDeletesMutations ? 1 : 0);
+ }
+
+ /**
+ * Get mutation metrics that correspond to committed mutations only
+ * @param totalMutationBytesObject MutationBytes object corresponding to all the mutations we
+ * attempted to commit including those that failed, those that
+ * were already sent and those that were unsent
+ * @param unsentMutationBatchList list of mutation batches that are unsent
+ * @param numMutations total number of mutations
+ * @param numFailedMutations number of failed mutations in the most recent failed batch
+ * @param numFailedPhase3Mutations number of mutations failed in phase 3 of index commits
+ * @param mutationCommitTime time taken for committing all mutations
+ * @return mutation metric object just accounting for mutations that are already
+ * successfully committed
+ */
+ @VisibleForTesting
+ static MutationMetric getCommittedMutationsMetric(
+ MutationBytes totalMutationBytesObject, List<List<Mutation>> unsentMutationBatchList,
+ long numMutations, long numFailedMutations,
+ long numFailedPhase3Mutations, long mutationCommitTime) {
+ long committedUpsertMutationBytes = totalMutationBytesObject == null ? 0 :
+ totalMutationBytesObject.getUpsertMutationBytes();
+ long committedDeleteMutationBytes = totalMutationBytesObject == null ? 0 :
+ totalMutationBytesObject.getDeleteMutationBytes();
+ long committedUpsertMutationCounter = totalMutationBytesObject == null ? 0 :
+ totalMutationBytesObject.getUpsertMutationCounter();
+ long committedDeleteMutationCounter = totalMutationBytesObject == null ? 0 :
+ totalMutationBytesObject.getDeleteMutationCounter();
+ long committedTotalMutationBytes = totalMutationBytesObject == null ? 0 :
+ totalMutationBytesObject.getTotalMutationBytes();
+ long upsertMutationCommitTime = 0L;
+ long deleteMutationCommitTime = 0L;
+
+ if (totalMutationBytesObject != null && numFailedMutations != 0) {
+ List<Mutation> uncommittedMutationsList = new ArrayList<>();
+ for (List<Mutation> mutationBatch : unsentMutationBatchList) {
+ uncommittedMutationsList.addAll(mutationBatch);
+ }
+ // Calculate the uncommitted mutations
+ MutationBytes uncommittedMutationBytesObject =
+ calculateMutationSize(uncommittedMutationsList, false);
+ committedUpsertMutationBytes -=
+ uncommittedMutationBytesObject.getUpsertMutationBytes();
+ committedDeleteMutationBytes -=
+ uncommittedMutationBytesObject.getDeleteMutationBytes();
+ committedUpsertMutationCounter -=
+ uncommittedMutationBytesObject.getUpsertMutationCounter();
+ committedDeleteMutationCounter -=
+ uncommittedMutationBytesObject.getDeleteMutationCounter();
+ committedTotalMutationBytes -=
+ uncommittedMutationBytesObject.getTotalMutationBytes();
+ }
+
+ if (committedTotalMutationBytes > 0) {
+ upsertMutationCommitTime =
+ (long)Math.floor((double)(committedUpsertMutationBytes * mutationCommitTime)/
+ committedTotalMutationBytes);
+ deleteMutationCommitTime =
+ (long)Math.ceil((double)(committedDeleteMutationBytes * mutationCommitTime)/
+ committedTotalMutationBytes);
+ }
+ return new MutationMetric(numMutations,
+ committedUpsertMutationBytes,
+ committedDeleteMutationBytes,
+ upsertMutationCommitTime,
+ deleteMutationCommitTime,
+ 0, // num failed mutations have been counted already in updateMutationBatchFailureMetrics()
+ committedUpsertMutationCounter,
+ committedDeleteMutationCounter,
+ committedTotalMutationBytes,
+ numFailedPhase3Mutations,
+ 0, 0, 0, 0 );
+ }
+
private void filterIndexCheckerMutations(Map<TableInfo, List<Mutation>> mutationMap,
Map<TableInfo, List<Mutation>> unverifiedIndexMutations,
Map<TableInfo, List<Mutation>> verifiedOrDeletedIndexMutations) throws SQLException {
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/iterate/BaseResultIterators.java b/phoenix-core/src/main/java/org/apache/phoenix/iterate/BaseResultIterators.java
index a7abfa3..508ced4 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/iterate/BaseResultIterators.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/iterate/BaseResultIterators.java
@@ -93,6 +93,7 @@ import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
import org.apache.phoenix.hbase.index.util.VersionUtil;
import org.apache.phoenix.join.HashCacheClient;
import org.apache.phoenix.mapreduce.util.PhoenixConfigurationUtil;
+import org.apache.phoenix.monitoring.OverAllQueryMetrics;
import org.apache.phoenix.parse.FilterableStatement;
import org.apache.phoenix.parse.HintNode;
import org.apache.phoenix.parse.HintNode.Hint;
@@ -1394,15 +1395,27 @@ public abstract class BaseResultIterators extends ExplainTable implements Result
success = true;
return iterators;
} catch (TimeoutException e) {
- context.getOverallQueryMetrics().queryTimedOut();
+ OverAllQueryMetrics overAllQueryMetrics = context.getOverallQueryMetrics();
+ overAllQueryMetrics.queryTimedOut();
+ if (context.getScanRanges().isPointLookup()) {
+ overAllQueryMetrics.queryPointLookupTimedOut();
+ } else {
+ overAllQueryMetrics.queryScanTimedOut();
+ }
GLOBAL_QUERY_TIMEOUT_COUNTER.increment();
// thrown when a thread times out waiting for the future.get() call to return
- toThrow = new SQLExceptionInfo.Builder(OPERATION_TIMED_OUT)
- .setMessage(". Query couldn't be completed in the allotted time: "
- + queryTimeOut + " ms").setRootCause(e).build().buildException();
+ toThrow = new SQLExceptionInfo.Builder(OPERATION_TIMED_OUT).setMessage(
+ ". Query couldn't be completed in the allotted time: " + queryTimeOut + " ms")
+ .setRootCause(e).build().buildException();
} catch (SQLException e) {
if (e.getErrorCode() == OPERATION_TIMED_OUT.getErrorCode()) {
- context.getOverallQueryMetrics().queryTimedOut();
+ OverAllQueryMetrics overAllQueryMetrics = context.getOverallQueryMetrics();
+ overAllQueryMetrics.queryTimedOut();
+ if (context.getScanRanges().isPointLookup()) {
+ overAllQueryMetrics.queryPointLookupTimedOut();
+ } else {
+ overAllQueryMetrics.queryScanTimedOut();
+ }
GLOBAL_QUERY_TIMEOUT_COUNTER.increment();
}
toThrow = e;
@@ -1434,7 +1447,13 @@ public abstract class BaseResultIterators extends ExplainTable implements Result
} finally {
if (toThrow != null) {
GLOBAL_FAILED_QUERY_COUNTER.increment();
- context.getOverallQueryMetrics().queryFailed();
+ OverAllQueryMetrics overAllQueryMetrics = context.getOverallQueryMetrics();
+ overAllQueryMetrics.queryFailed();
+ if (context.getScanRanges().isPointLookup()) {
+ overAllQueryMetrics.queryPointLookupFailed();
+ } else {
+ overAllQueryMetrics.queryScanFailed();
+ }
throw toThrow;
}
}
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/iterate/RoundRobinResultIterator.java b/phoenix-core/src/main/java/org/apache/phoenix/iterate/RoundRobinResultIterator.java
index 3c6f6fc..d05f8f3 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/iterate/RoundRobinResultIterator.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/iterate/RoundRobinResultIterator.java
@@ -32,6 +32,7 @@ import org.apache.phoenix.compile.ExplainPlanAttributes
.ExplainPlanAttributesBuilder;
import org.apache.phoenix.compile.QueryPlan;
import org.apache.phoenix.compile.StatementContext;
+import org.apache.phoenix.monitoring.OverAllQueryMetrics;
import org.apache.phoenix.query.ConnectionQueryServices;
import org.apache.phoenix.schema.tuple.Tuple;
import org.apache.phoenix.util.ServerUtil;
@@ -280,6 +281,13 @@ public class RoundRobinResultIterator implements ResultIterator {
} finally {
if (toThrow != null) {
GLOBAL_FAILED_QUERY_COUNTER.increment();
+ OverAllQueryMetrics overAllQueryMetrics = plan.getContext().getOverallQueryMetrics();
+ overAllQueryMetrics.queryFailed();
+ if (plan.getContext().getScanRanges().isPointLookup()) {
+ overAllQueryMetrics.queryPointLookupFailed();
+ } else {
+ overAllQueryMetrics.queryScanFailed();
+ }
throw toThrow;
}
}
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/iterate/ScanningResultIterator.java b/phoenix-core/src/main/java/org/apache/phoenix/iterate/ScanningResultIterator.java
index 46cbf96..4db8cfe 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/iterate/ScanningResultIterator.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/iterate/ScanningResultIterator.java
@@ -125,6 +125,8 @@ public class ScanningResultIterator implements ResultIterator {
scanMetricsMap.get(COUNT_OF_ROWS_SCANNED_KEY_METRIC_NAME));
changeMetric(scanMetricsHolder.getCountOfRowsFiltered(),
scanMetricsMap.get(COUNT_OF_ROWS_FILTERED_KEY_METRIC_NAME));
+ changeMetric(scanMetricsHolder.getCountOfBytesScanned(),
+ scanMetricsMap.get(BYTES_IN_RESULTS_METRIC_NAME));
changeMetric(GLOBAL_SCAN_BYTES,
scanMetricsMap.get(BYTES_IN_RESULTS_METRIC_NAME));
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixConnection.java b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixConnection.java
index f64c6b3..7ec162f 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixConnection.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixConnection.java
@@ -79,6 +79,7 @@ import org.apache.phoenix.iterate.TableResultIteratorFactory;
import org.apache.phoenix.jdbc.PhoenixStatement.PhoenixStatementParser;
import org.apache.phoenix.log.LogLevel;
import org.apache.phoenix.monitoring.MetricType;
+import org.apache.phoenix.monitoring.TableMetricsManager;
import org.apache.phoenix.parse.PFunction;
import org.apache.phoenix.parse.PSchema;
import org.apache.phoenix.query.ConnectionQueryServices;
@@ -713,6 +714,7 @@ public class PhoenixConnection implements Connection, MetaDataMutated, SQLClosea
return;
}
try {
+ TableMetricsManager.pushMetricsFromConnInstanceMethod(getMutationMetrics());
clearMetrics();
try {
if (traceScope != null) {
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixResultSet.java b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixResultSet.java
index 2a9d227..99eeeb9 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixResultSet.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixResultSet.java
@@ -45,9 +45,11 @@ import java.text.Format;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Calendar;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import com.google.common.base.Strings;
import com.google.common.primitives.Bytes;
import com.google.protobuf.InvalidProtocolBufferException;
import org.apache.commons.lang.ArrayUtils;
@@ -73,6 +75,7 @@ import org.apache.phoenix.log.QueryStatus;
import org.apache.phoenix.monitoring.MetricType;
import org.apache.phoenix.monitoring.OverAllQueryMetrics;
import org.apache.phoenix.monitoring.ReadMetricQueue;
+import org.apache.phoenix.monitoring.TableMetricsManager;
import org.apache.phoenix.schema.PColumn;
import org.apache.phoenix.schema.PColumnImpl;
import org.apache.phoenix.schema.tuple.ResultTuple;
@@ -156,6 +159,7 @@ public class PhoenixResultSet implements ResultSet, SQLCloseable {
private Long count = 0L;
private Object exception;
+ private long queryTime;
public PhoenixResultSet(ResultIterator resultIterator, RowProjector rowProjector,
StatementContext ctx) throws SQLException {
@@ -213,6 +217,14 @@ public class PhoenixResultSet implements ResultSet, SQLCloseable {
statement.getResultSets().remove(this);
overAllQueryMetrics.endQuery();
overAllQueryMetrics.stopResultSetWatch();
+ if (context.getCurrentTable() != null && context.getCurrentTable().getTable() != null
+ && !Strings.isNullOrEmpty(
+ context.getCurrentTable().getTable().getPhysicalName().getString())) {
+ boolean isPointLookup = context.getScanRanges().isPointLookup();
+ String tableName =
+ context.getCurrentTable().getTable().getPhysicalName().toString();
+ updateTableLevelReadMetrics(tableName, isPointLookup);
+ }
if (!queryLogger.isSynced()) {
if(this.exception==null){
queryLogger.log(QueryLogInfo.QUERY_STATUS_I,QueryStatus.COMPLETED.toString());
@@ -853,7 +865,7 @@ public class PhoenixResultSet implements ResultSet, SQLCloseable {
if (rowProjectorWithDynamicCols != null) {
rowProjectorWithDynamicCols.reset();
}
- } catch (RuntimeException e) {
+ } catch (RuntimeException | SQLException e) {
// FIXME: Expression.evaluate does not throw SQLException
// so this will unwrap throws from that.
queryLogger.log(QueryLogInfo.QUERY_STATUS_I, QueryStatus.FAILED.toString());
@@ -865,8 +877,13 @@ public class PhoenixResultSet implements ResultSet, SQLCloseable {
throw (SQLException) e.getCause();
}
throw e;
- }finally{
- if (this.exception!=null) {
+ } finally{
+ // If an exception occurs during rs.next(), or if we're on the last row, update metrics
+ if (this.exception != null || currentRow == null) {
+ overAllQueryMetrics.endQuery();
+ overAllQueryMetrics.stopResultSetWatch();
+ }
+ if (this.exception != null) {
queryLogger.log(QueryLogInfo.NO_OF_RESULTS_ITERATED_I, count);
if (queryLogger.isDebugEnabled()) {
queryLogger.log(QueryLogInfo.SCAN_METRICS_JSON_I,
@@ -877,14 +894,24 @@ public class PhoenixResultSet implements ResultSet, SQLCloseable {
queryLogger.sync(getReadMetrics(), getOverAllRequestReadMetrics());
}
}
- if (currentRow == null) {
- overAllQueryMetrics.endQuery();
- overAllQueryMetrics.stopResultSetWatch();
- }
}
return currentRow != null;
}
+ private void updateTableLevelReadMetrics(String tableName, boolean isPointLookup) {
+ Map<String, Map<MetricType, Long>> readMetrics = getReadMetrics();
+ TableMetricsManager.pushMetricsFromConnInstanceMethod(readMetrics);
+ Map<String, Map<MetricType, Long>> metricsFromOverallQuery = new HashMap<>();
+ Map<MetricType, Long> overAllReadMetrics = getOverAllRequestReadMetrics();
+ metricsFromOverallQuery.put(tableName, overAllReadMetrics);
+ TableMetricsManager.pushMetricsFromConnInstanceMethod(metricsFromOverallQuery);
+ if (readMetrics.get(tableName) != null) {
+ TableMetricsManager.updateMetricsMethod(tableName, this.exception == null ?
+ MetricType.SELECT_AGGREGATE_SUCCESS_SQL_COUNTER :
+ MetricType.SELECT_AGGREGATE_FAILURE_SQL_COUNTER, 1);
+ }
+ }
+
@Override
public boolean previous() throws SQLException {
throw new SQLFeatureNotSupportedException();
@@ -1398,6 +1425,10 @@ public class PhoenixResultSet implements ResultSet, SQLCloseable {
return context;
}
+ public void setQueryTime(long queryTime) {
+ this.queryTime = queryTime;
+ }
+
/**
* Return the row projector to use
* @return the row projector including dynamic column projectors in case we are including
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java
index 3d748e8..9d7d5f0 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java
@@ -20,6 +20,26 @@ package org.apache.phoenix.jdbc;
import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_MUTATION_SQL_COUNTER;
import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_QUERY_TIME;
import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_SELECT_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.DELETE_AGGREGATE_FAILURE_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.DELETE_FAILED_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.DELETE_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.DELETE_SQL_QUERY_TIME;
+import static org.apache.phoenix.monitoring.MetricType.DELETE_SUCCESS_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.MUTATION_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.SELECT_AGGREGATE_FAILURE_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.SELECT_FAILED_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.SELECT_POINTLOOKUP_FAILED_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.SELECT_POINTLOOKUP_SUCCESS_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.SELECT_SCAN_FAILED_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.SELECT_SCAN_SUCCESS_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.SELECT_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.SELECT_SQL_QUERY_TIME;
+import static org.apache.phoenix.monitoring.MetricType.SELECT_SUCCESS_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.UPSERT_AGGREGATE_FAILURE_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.UPSERT_FAILED_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.UPSERT_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.UPSERT_SQL_QUERY_TIME;
+import static org.apache.phoenix.monitoring.MetricType.UPSERT_SUCCESS_SQL_COUNTER;
import java.io.File;
import java.io.IOException;
@@ -41,6 +61,7 @@ import java.util.Map;
import java.util.Properties;
import java.util.Set;
+import com.google.common.base.Strings;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
@@ -95,6 +116,7 @@ import org.apache.phoenix.log.QueryLogInfo;
import org.apache.phoenix.log.QueryLogger;
import org.apache.phoenix.log.QueryLoggerUtil;
import org.apache.phoenix.log.QueryStatus;
+import org.apache.phoenix.monitoring.TableMetricsManager;
import org.apache.phoenix.optimize.Cost;
import org.apache.phoenix.parse.AddColumnStatement;
import org.apache.phoenix.parse.AddJarsStatement;
@@ -291,88 +313,150 @@ public class PhoenixStatement implements Statement, SQLCloseable {
private PhoenixResultSet executeQuery(final CompilableStatement stmt,
final boolean doRetryOnMetaNotFoundError, final QueryLogger queryLogger) throws SQLException {
GLOBAL_SELECT_SQL_COUNTER.increment();
-
+
try {
- return CallRunner.run(
- new CallRunner.CallableThrowable<PhoenixResultSet, SQLException>() {
- @Override
- public PhoenixResultSet call() throws SQLException {
- final long startTime = EnvironmentEdgeManager.currentTimeMillis();
- try {
- PhoenixConnection conn = getConnection();
-
- if (conn.getQueryServices().isUpgradeRequired() && !conn.isRunningUpgrade()
- && stmt.getOperation() != Operation.UPGRADE) {
- throw new UpgradeRequiredException();
- }
- QueryPlan plan = stmt.compilePlan(PhoenixStatement.this, Sequence.ValueOp.VALIDATE_SEQUENCE);
- // Send mutations to hbase, so they are visible to subsequent reads.
- // Use original plan for data table so that data and immutable indexes will be sent
- // TODO: for joins, we need to iterate through all tables, but we need the original table,
- // not the projected table, so plan.getContext().getResolver().getTables() won't work.
- Iterator<TableRef> tableRefs = plan.getSourceRefs().iterator();
- connection.getMutationState().sendUncommitted(tableRefs);
- plan = connection.getQueryServices().getOptimizer().optimize(PhoenixStatement.this, plan);
- // this will create its own trace internally, so we don't wrap this
- // whole thing in tracing
- ResultIterator resultIterator = plan.iterator();
- if (LOGGER.isDebugEnabled()) {
- String explainPlan = QueryUtil.getExplainPlan(resultIterator);
- LOGGER.debug(LogUtil.addCustomAnnotations(
- "Explain plan: " + explainPlan, connection));
- }
- StatementContext context = plan.getContext();
- context.setQueryLogger(queryLogger);
- if(queryLogger.isDebugEnabled()){
- queryLogger.log(QueryLogInfo.EXPLAIN_PLAN_I, QueryUtil.getExplainPlan(resultIterator));
- queryLogger.log(QueryLogInfo.GLOBAL_SCAN_DETAILS_I, context.getScan()!=null?context.getScan().toString():null);
- }
- context.getOverallQueryMetrics().startQuery();
- PhoenixResultSet rs = newResultSet(resultIterator, plan.getProjector(), plan.getContext());
- resultSets.add(rs);
- setLastQueryPlan(plan);
- setLastResultSet(rs);
- setLastUpdateCount(NO_UPDATE);
- setLastUpdateOperation(stmt.getOperation());
- // If transactional, this will move the read pointer forward
- if (connection.getAutoCommit()) {
- connection.commit();
- }
- connection.incrementStatementExecutionCounter();
- return rs;
- }
- //Force update cache and retry if meta not found error occurs
- catch (MetaDataEntityNotFoundException e) {
- if(doRetryOnMetaNotFoundError && e.getTableName()!=null){
- if(LOGGER.isDebugEnabled())
- LOGGER.debug("Reloading table "
- + e.getTableName()+" data from server");
- if(new MetaDataClient(connection).updateCache(connection.getTenantId(),
- e.getSchemaName(), e.getTableName(), true).wasUpdated()){
- //TODO we can log retry count and error for debugging in LOG table
- return executeQuery(stmt, false, queryLogger);
+ return CallRunner
+ .run(new CallRunner.CallableThrowable<PhoenixResultSet, SQLException>() {
+ @Override public PhoenixResultSet call() throws SQLException {
+ final long startTime = EnvironmentEdgeManager.currentTimeMillis();
+ boolean success = false;
+ boolean pointLookup = false;
+ String tableName = null;
+ PhoenixResultSet rs = null;
+ try {
+ PhoenixConnection conn = getConnection();
+
+ if (conn.getQueryServices().isUpgradeRequired() && !conn
+ .isRunningUpgrade()
+ && stmt.getOperation() != Operation.UPGRADE) {
+ throw new UpgradeRequiredException();
+ }
+ QueryPlan
+ plan =
+ stmt.compilePlan(PhoenixStatement.this,
+ Sequence.ValueOp.VALIDATE_SEQUENCE);
+ // Send mutations to hbase, so they are visible to subsequent reads.
+ // Use original plan for data table so that data and immutable indexes will be sent
+ // TODO: for joins, we need to iterate through all tables, but we need the original table,
+ // not the projected table, so plan.getContext().getResolver().getTables() won't work.
+ if (plan.getTableRef() != null
+ && plan.getTableRef().getTable() != null && !Strings
+ .isNullOrEmpty(
+ plan.getTableRef().getTable().getPhysicalName()
+ .toString())) {
+ tableName = plan.getTableRef().getTable().getPhysicalName()
+ .toString();
+ }
+ if (plan.getContext().getScanRanges().isPointLookup()) {
+ pointLookup = true;
+ }
+ Iterator<TableRef> tableRefs = plan.getSourceRefs().iterator();
+ connection.getMutationState().sendUncommitted(tableRefs);
+ plan =
+ connection.getQueryServices().getOptimizer()
+ .optimize(PhoenixStatement.this, plan);
+ // this will create its own trace internally, so we don't wrap this
+ // whole thing in tracing
+ ResultIterator resultIterator = plan.iterator();
+ if (LOGGER.isDebugEnabled()) {
+ String explainPlan = QueryUtil.getExplainPlan(resultIterator);
+ LOGGER.debug(LogUtil.addCustomAnnotations(
+ "Explain plan: " + explainPlan, connection));
+ }
+ StatementContext context = plan.getContext();
+ context.setQueryLogger(queryLogger);
+ if (queryLogger.isDebugEnabled()) {
+ queryLogger.log(QueryLogInfo.EXPLAIN_PLAN_I,
+ QueryUtil.getExplainPlan(resultIterator));
+ queryLogger.log(QueryLogInfo.GLOBAL_SCAN_DETAILS_I,
+ context.getScan() != null ?
+ context.getScan().toString() :
+ null);
+ }
+ context.getOverallQueryMetrics().startQuery();
+ rs =
+ newResultSet(resultIterator, plan.getProjector(),
+ plan.getContext());
+ resultSets.add(rs);
+ setLastQueryPlan(plan);
+ setLastResultSet(rs);
+ setLastUpdateCount(NO_UPDATE);
+ setLastUpdateOperation(stmt.getOperation());
+ // If transactional, this will move the read pointer forward
+ if (connection.getAutoCommit()) {
+ connection.commit();
+ }
+ connection.incrementStatementExecutionCounter();
+ success = true;
}
+ //Force update cache and retry if meta not found error occurs
+ catch (MetaDataEntityNotFoundException e) {
+ if (doRetryOnMetaNotFoundError && e.getTableName() != null) {
+ if (LOGGER.isDebugEnabled()) LOGGER.debug(
+ "Reloading table " + e.getTableName()
+ + " data from server");
+ if (new MetaDataClient(connection)
+ .updateCache(connection.getTenantId(),
+ e.getSchemaName(), e.getTableName(), true)
+ .wasUpdated()) {
+ //TODO we can log retry count and error for debugging in LOG table
+ return executeQuery(stmt, false, queryLogger);
+ }
+ }
+ throw e;
+ } catch (RuntimeException e) {
+
+ // FIXME: Expression.evaluate does not throw SQLException
+ // so this will unwrap throws from that.
+ if (e.getCause() instanceof SQLException) {
+ throw (SQLException) e.getCause();
+ }
+ throw e;
+ } finally {
+ // Regardless of whether the query was successfully handled or not,
+ // update the time spent so far. If needed, we can separate out the
+ // success times and failure times.
+ GLOBAL_QUERY_TIME.update(EnvironmentEdgeManager.currentTimeMillis()
+ - startTime);
+ long
+ executeQueryTimeSpent =
+ EnvironmentEdgeManager.currentTimeMillis() - startTime;
+ if (tableName != null) {
+
+ TableMetricsManager
+ .updateMetricsMethod(tableName, SELECT_SQL_COUNTER, 1);
+ TableMetricsManager
+ .updateMetricsMethod(tableName, SELECT_SQL_QUERY_TIME,
+ executeQueryTimeSpent);
+ if (success) {
+ TableMetricsManager.updateMetricsMethod(tableName,
+ SELECT_SUCCESS_SQL_COUNTER, 1);
+ TableMetricsManager.updateMetricsMethod(tableName,
+ pointLookup ?
+ SELECT_POINTLOOKUP_SUCCESS_SQL_COUNTER :
+ SELECT_SCAN_SUCCESS_SQL_COUNTER, 1);
+ } else {
+ TableMetricsManager.updateMetricsMethod(tableName,
+ SELECT_FAILED_SQL_COUNTER, 1);
+ TableMetricsManager.updateMetricsMethod(tableName,
+ SELECT_AGGREGATE_FAILURE_SQL_COUNTER, 1);
+ TableMetricsManager.updateMetricsMethod(tableName,
+ pointLookup ?
+ SELECT_POINTLOOKUP_FAILED_SQL_COUNTER :
+ SELECT_SCAN_FAILED_SQL_COUNTER, 1);
+ }
+ }
+ if (rs != null) {
+ rs.setQueryTime(executeQueryTimeSpent);
+ }
+ }
+ return rs;
}
- throw e;
- }catch (RuntimeException e) {
-
- // FIXME: Expression.evaluate does not throw SQLException
- // so this will unwrap throws from that.
- if (e.getCause() instanceof SQLException) {
- throw (SQLException) e.getCause();
- }
- throw e;
- } finally {
- // Regardless of whether the query was successfully handled or not,
- // update the time spent so far. If needed, we can separate out the
- // success times and failure times.
- GLOBAL_QUERY_TIME.update(EnvironmentEdgeManager.currentTimeMillis() - startTime);
- }
- }
- }, PhoenixContextExecutor.inContext());
- }catch (Exception e) {
+ }, PhoenixContextExecutor.inContext());
+ } catch (Exception e) {
if (queryLogger.isDebugEnabled()) {
- queryLogger.log(QueryLogInfo.EXCEPTION_TRACE_I, Throwables.getStackTraceAsString(e));
+ queryLogger
+ .log(QueryLogInfo.EXCEPTION_TRACE_I, Throwables.getStackTraceAsString(e));
queryLogger.log(QueryLogInfo.QUERY_STATUS_I, QueryStatus.FAILED.toString());
queryLogger.sync(null, null);
}
@@ -424,16 +508,30 @@ public class PhoenixStatement implements Statement, SQLCloseable {
new CallRunner.CallableThrowable<Integer, SQLException>() {
@Override
public Integer call() throws SQLException {
+ boolean success = false;
+ String tableName = null;
+ boolean isUpsert = false;
+ boolean isDelete = false;
+ MutationState state = null;
+ MutationPlan plan = null;
+ final long startExecuteMutationTime = EnvironmentEdgeManager.currentTimeMillis();
try {
PhoenixConnection conn = getConnection();
if (conn.getQueryServices().isUpgradeRequired() && !conn.isRunningUpgrade()
&& stmt.getOperation() != Operation.UPGRADE) {
throw new UpgradeRequiredException();
}
- MutationState state = connection.getMutationState();
- MutationPlan plan = stmt.compilePlan(PhoenixStatement.this, Sequence.ValueOp.VALIDATE_SEQUENCE);
- if (plan.getTargetRef() != null && plan.getTargetRef().getTable() != null && plan.getTargetRef().getTable().isTransactional()) {
- state.startTransaction(plan.getTargetRef().getTable().getTransactionProvider());
+ state = connection.getMutationState();
+ plan = stmt.compilePlan(PhoenixStatement.this, Sequence.ValueOp.VALIDATE_SEQUENCE);
+ isUpsert = stmt instanceof ExecutableUpsertStatement;
+ isDelete = stmt instanceof ExecutableDeleteStatement;
+ if (plan.getTargetRef() != null && plan.getTargetRef().getTable() != null) {
+ if(!Strings.isNullOrEmpty(plan.getTargetRef().getTable().getPhysicalName().toString())) {
+ tableName = plan.getTargetRef().getTable().getPhysicalName().toString();
+ }
+ if (plan.getTargetRef().getTable().isTransactional()) {
+ state.startTransaction(plan.getTargetRef().getTable().getTransactionProvider());
+ }
}
Iterator<TableRef> tableRefs = plan.getSourceRefs().iterator();
state.sendUncommitted(tableRefs);
@@ -459,6 +557,7 @@ public class PhoenixStatement implements Statement, SQLCloseable {
queryLogger.syncAudit();
}
+ success = true;
return lastUpdateCount;
}
//Force update cache and retry if meta not found error occurs
@@ -480,6 +579,37 @@ public class PhoenixStatement implements Statement, SQLCloseable {
throw (SQLException) e.getCause();
}
throw e;
+ } finally {
+ // Regardless of whether the mutation was successfully handled or not,
+ // update the time spent so far. If needed, we can separate out the
+ // success times and failure times.
+ if (tableName != null) {
+ // Counts for both ddl and dml
+ TableMetricsManager.updateMetricsMethod(tableName,
+ MUTATION_SQL_COUNTER, 1);
+ // Only count dml operations
+ if (isUpsert || isDelete) {
+ long executeMutationTimeSpent =
+ EnvironmentEdgeManager.currentTimeMillis() - startExecuteMutationTime;
+
+ TableMetricsManager.updateMetricsMethod(tableName, isUpsert ?
+ UPSERT_SQL_COUNTER : DELETE_SQL_COUNTER, 1);
+ TableMetricsManager.updateMetricsMethod(tableName, isUpsert ?
+ UPSERT_SQL_QUERY_TIME : DELETE_SQL_QUERY_TIME, executeMutationTimeSpent);
+
+ if (success) {
+ TableMetricsManager.updateMetricsMethod(tableName, isUpsert ?
+ UPSERT_SUCCESS_SQL_COUNTER : DELETE_SUCCESS_SQL_COUNTER, 1);
+ } else {
+ TableMetricsManager.updateMetricsMethod(tableName, isUpsert ?
+ UPSERT_FAILED_SQL_COUNTER : DELETE_FAILED_SQL_COUNTER, 1);
+ //Failures are updated for executeMutation phase and for autocommit=true case here.
+ TableMetricsManager.updateMetricsMethod(tableName, isUpsert ? UPSERT_AGGREGATE_FAILURE_SQL_COUNTER:
+ DELETE_AGGREGATE_FAILURE_SQL_COUNTER, 1);
+ }
+ }
+ }
+
}
}
}, PhoenixContextExecutor.inContext(),
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/JmxMetricProvider.java b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/JmxMetricProvider.java
new file mode 100644
index 0000000..bc3a17b
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/JmxMetricProvider.java
@@ -0,0 +1,96 @@
+/*
+ * 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 org.apache.hadoop.hbase.metrics.Gauge;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.apache.hadoop.hbase.metrics.MetricRegistries;
+import org.apache.hadoop.hbase.metrics.MetricRegistry;
+import org.apache.hadoop.hbase.metrics.MetricRegistryInfo;
+import org.apache.phoenix.query.QueryServicesOptions;
+
+import java.util.Map;
+
+/**
+ * This class implements the JMX based default Metric publishing
+ * of Metrics to JMX end point.
+ * This class is defined in phoenix/phoenix-core/src/main/resources/META-INF/services/org.apache.phoenix.monitoring.MetricPublisherSupplierFactory
+ */
+public class JmxMetricProvider implements MetricPublisherSupplierFactory {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(JmxMetricProvider.class);
+ private static final String metricsName = "PHOENIX-TableLevel";
+ private static final String metricsDesc = "Phoenix Client Metrics";
+ private static final String metricsjmxContext = "phoenixTableLevel";
+ private static final String metricsContext = "Phoenix,sub=CLIENT";
+
+ private MetricRegistry metricRegistry;
+
+ @Override public void registerMetricProvider() {
+ metricRegistry = createMetricRegistry();
+ GlobalMetricRegistriesAdapter.getInstance().registerMetricRegistry(metricRegistry,
+ QueryServicesOptions.withDefaults().getClientMetricTag());
+ }
+
+ @Override public void unregisterMetricProvider() {
+
+ }
+
+ private MetricRegistry createMetricRegistry() {
+ LOGGER.info("Creating Metric Registry for Phoenix Table Level Metrics");
+ MetricRegistryInfo registryInfo =
+ new MetricRegistryInfo(metricsName, metricsDesc,
+ metricsjmxContext, metricsContext, true);
+ return MetricRegistries.global().create(registryInfo);
+ }
+
+ private static class PhoenixMetricGauge implements Gauge<Long> {
+ private final PhoenixTableMetric metric;
+
+ public PhoenixMetricGauge(PhoenixTableMetric metric) {
+ this.metric = metric;
+ }
+
+ @Override public Long getValue() {
+ return metric.getValue();
+ }
+ }
+
+ private String getMetricNameFromMetricType(MetricType type, String tableName) {
+ return tableName + "_table_" + type;
+ }
+
+ @Override public void registerMetrics(TableClientMetrics tInstance) {
+ for (Map.Entry<MetricType, PhoenixTableMetric> entry : tInstance.getMetricRegistry()
+ .entrySet()) {
+ metricRegistry
+ .register(getMetricNameFromMetricType(entry.getKey(), tInstance.getTableName()),
+ new PhoenixMetricGauge(entry.getValue()));
+ }
+ }
+
+ @Override public void unRegisterMetrics(TableClientMetrics tInstance) {
+ for (Map.Entry<MetricType, PhoenixTableMetric> entry : tInstance.getMetricRegistry()
+ .entrySet()) {
+ metricRegistry
+ .remove(getMetricNameFromMetricType(entry.getKey(), tInstance.getTableName()));
+ }
+ }
+
+}
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/MetricPublisherSupplierFactory.java b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/MetricPublisherSupplierFactory.java
new file mode 100644
index 0000000..85c2ee9
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/MetricPublisherSupplierFactory.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.monitoring;
+
+/**
+ * Interface for configurable MetricPublisher interface construction
+ *
+ */
+public interface MetricPublisherSupplierFactory extends MetricsRegistry {
+
+ /**
+ * Interface for Registering Publisher Method
+ */
+ void registerMetricProvider();
+
+ /**
+ * Interface for UnRegistering Publisher Method
+ */
+ void unregisterMetricProvider();
+}
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/MetricServiceResolver.java b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/MetricServiceResolver.java
new file mode 100644
index 0000000..48b42b7
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/MetricServiceResolver.java
@@ -0,0 +1,65 @@
+/**
+ * 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 com.google.common.base.Preconditions;
+import org.apache.phoenix.util.InstanceResolver;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+
+/**
+ * This class helps resolve the metricpublisher supplier class at the run time.
+ * Based on the classString name passed, it will return the appropriate class Instance.
+ */
+public class MetricServiceResolver {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(MetricServiceResolver.class);
+
+ private MetricPublisherSupplierFactory metricSupplier = null;
+
+ public MetricPublisherSupplierFactory instantiate(String classString) {
+ Preconditions.checkNotNull(classString);
+ if (metricSupplier == null) {
+ try {
+ Class clazz = Class.forName(classString);
+ List<MetricPublisherSupplierFactory>
+ factoryList =
+ InstanceResolver.get(MetricPublisherSupplierFactory.class, null);
+ for (MetricPublisherSupplierFactory factory : factoryList) {
+ if (clazz.isInstance(factory)) {
+ metricSupplier = factory;
+ LOGGER.info(String.format(
+ "Sucessfully loaded class for MetricPublishFactory of type: %s",
+ classString));
+ break;
+ }
+ }
+ if (metricSupplier == null) {
+ String msg = String.format("Could not load/instantiate class %s", classString);
+ LOGGER.error(msg);
+ }
+ } catch (ClassNotFoundException e) {
+ LOGGER.error(String.format("Could not load/instantiate class %s", classString), e);
+ }
+ }
+ return metricSupplier;
+ }
+}
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/MetricType.java b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/MetricType.java
index 40fcad0..dced4ca 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/MetricType.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/MetricType.java
@@ -28,13 +28,87 @@ import org.apache.phoenix.schema.types.PLong;
public enum MetricType {
NO_OP_METRIC("no", "No op metric",LogLevel.OFF, PLong.INSTANCE),
- // mutation (write) related metrics
MUTATION_BATCH_SIZE("ms", "Number of mutations in the batch",LogLevel.OFF, PLong.INSTANCE),
MUTATION_BYTES("mb", "Size of mutations in bytes",LogLevel.OFF, PLong.INSTANCE),
MUTATION_COMMIT_TIME("mt", "Time it took to commit a batch of mutations",LogLevel.OFF, PLong.INSTANCE),
MUTATION_BATCH_FAILED_SIZE("mfs", "Number of mutations that failed to be committed",LogLevel.OFF, PLong.INSTANCE),
MUTATION_SQL_COUNTER("msc", "Counter for number of mutation sql statements",LogLevel.OFF, PLong.INSTANCE),
+ UPSERT_SQL_COUNTER("uc", "Counter for number of upsert sql queries", LogLevel.OFF, PLong.INSTANCE),
+ UPSERT_COMMIT_TIME("ut", "Time it took to commit a batch of upserts", LogLevel.OFF, PLong.INSTANCE),
+ UPSERT_MUTATION_BYTES("umb", "Size of mutations in upsert statement in bytes",LogLevel.OFF, PLong.INSTANCE),
+ UPSERT_MUTATION_SQL_COUNTER("umsc", "Counter for number of upsert mutations committed",LogLevel.OFF, PLong.INSTANCE),
+ UPSERT_BATCH_FAILED_SIZE("ubfs", "Number of upsert mutations in a batch that failed to be committed",
+ LogLevel.OFF, PLong.INSTANCE),
+ UPSERT_BATCH_FAILED_COUNTER("ubfc", "Number of upsert mutation batches that failed to be committed",
+ LogLevel.OFF, PLong.INSTANCE),
+
+ UPSERT_AGGREGATE_SUCCESS_SQL_COUNTER("uassc", "Counter which indicates the total number of upsert Mutations which passed executeUpdate phase "
+ + "(since last commit called) and subsequent conn.commit() are successful.", LogLevel.OFF, PLong.INSTANCE),
+ UPSERT_AGGREGATE_FAILURE_SQL_COUNTER("uafsc", "Counter which indicates the total number of upsert Mutations for all statements which failed either in executeUpdate phase "
+ + "(since last commit called) or subsequent conn.commit() fails", LogLevel.OFF, PLong.INSTANCE),
+ UPSERT_SUCCESS_SQL_COUNTER("ussc", "Counter for number of upsert sql queries that successfully"
+ + " passed the executeMutation phase, or if autoCommit is true, the total"
+ + " number of successful upserts", LogLevel.OFF, PLong.INSTANCE),
+ UPSERT_FAILED_SQL_COUNTER("ufsc", "Counter for number of upsert sql queries that"
+ + " failed the executeMutation phase, or if autoCommit is true, the total"
+ + " number of upsert failures", LogLevel.OFF, PLong.INSTANCE),
+ UPSERT_SQL_QUERY_TIME("uqt", "Time taken by upsert sql queries inside executeMutation or if"
+ + " autoCommit is true, the total time taken for executeMutation + conn.commit",
+ LogLevel.OFF, PLong.INSTANCE),
+
+ // delete-specific metrics updated during executeMutation
+ DELETE_SQL_COUNTER("dc", "Counter for number of delete sql queries", LogLevel.OFF, PLong.INSTANCE),
+ DELETE_SUCCESS_SQL_COUNTER("dssc", "Counter for number of delete sql queries that successfully"
+ + " passed the executeMutation phase, or if autoCommit is true, the total"
+ + " number of successful deletes", LogLevel.OFF, PLong.INSTANCE),
+ DELETE_AGGREGATE_SUCCESS_SQL_COUNTER("dassc", "Counter which indicates if everything in the executeUpdate phase for all "
+ + "statements (since last commit called) and subsequent conn.commit() is successful.", LogLevel.OFF, PLong.INSTANCE),
+ DELETE_AGGREGATE_FAILURE_SQL_COUNTER("dafsc", "Counter which indicates if anything in the executeUpdate phase for any "
+ + "statements (since last commit called) or subsequent conn.commit() fails.", LogLevel.OFF, PLong.INSTANCE),
+ DELETE_FAILED_SQL_COUNTER("dfsc", "Counter for number of delete sql queries that"
+ + " failed the executeMutation phase, or if autoCommit is true, the total"
+ + " number of delete failures", LogLevel.OFF, PLong.INSTANCE),
+ DELETE_SQL_QUERY_TIME("dqt", "Time taken by delete sql queries inside executeMutation or if"
+ + " autoCommit is true, the total time taken for executeMutation + conn.commit",
+ LogLevel.OFF, PLong.INSTANCE),
+
+ DELETE_COMMIT_TIME("dt", "Time it took to commit a batch of deletes", LogLevel.OFF, PLong.INSTANCE),
+ DELETE_MUTATION_BYTES("dmb", "Size of mutations in delete statement in bytes",LogLevel.OFF, PLong.INSTANCE),
+ DELETE_MUTATION_SQL_COUNTER("dmsc", "Counter for number of delete mutations committed",LogLevel.OFF, PLong.INSTANCE),
+ DELETE_BATCH_FAILED_SIZE("dbfs", "Number of delete mutations in a batch that failed to be committed",
+ LogLevel.OFF, PLong.INSTANCE),
+ DELETE_BATCH_FAILED_COUNTER("dbfc", "Number of delete mutation batches that failed to be committed",
+ LogLevel.OFF, PLong.INSTANCE),
+
+ // select-specific query (read) metrics updated during executeQuery
+ SELECT_SUCCESS_SQL_COUNTER("sss", "Counter for number of select sql queries that successfully"
+ + " passed the executeQuery phase", LogLevel.OFF, PLong.INSTANCE),
+ SELECT_AGGREGATE_SUCCESS_SQL_COUNTER("sassc","Counter which indicates if everything in executeQuery"
+ + " phase and all rs.next() are successful",LogLevel.OFF, PLong.INSTANCE),
+ SELECT_AGGREGATE_FAILURE_SQL_COUNTER("safsc","Counter which indicates if anything in "
+ + "executeQuery phase or any of the rs.next() fail",LogLevel.OFF, PLong.INSTANCE),
+ SELECT_POINTLOOKUP_SUCCESS_SQL_COUNTER("spls", "Counter for number of point lookup select sql "
+ + "queries that succeeded the executeQuery phase", LogLevel.OFF, PLong.INSTANCE),
+ SELECT_SCAN_SUCCESS_SQL_COUNTER("sscs", "Counter for number of scan select sql queries "
+ + "that succeed the executeQuery phase", LogLevel.OFF, PLong.INSTANCE),
+ SELECT_FAILED_SQL_COUNTER("sfsc", "Counter for number of select sql queries that"
+ + " failed the executeQuery phase", LogLevel.OFF, PLong.INSTANCE),
+ SELECT_POINTLOOKUP_FAILED_SQL_COUNTER("splf", "Counter for number of point lookup select sql "
+ + "queries that failed the executeQuery phase", LogLevel.OFF, PLong.INSTANCE),
+ SELECT_SCAN_FAILED_SQL_COUNTER("sscf", "Counter for number of scan select sql queries "
+ + "that failed the executeQuery phase", LogLevel.OFF, PLong.INSTANCE),
+ SELECT_SQL_QUERY_TIME("sqt", "Time taken by select sql queries inside executeQuery",
+ LogLevel.OFF, PLong.INSTANCE),
+
INDEX_COMMIT_FAILURE_SIZE("p3s", "Number of mutations that failed in phase 3", LogLevel.OFF, PLong.INSTANCE),
+ QUERY_POINTLOOKUP_TIMEOUT_COUNTER("qplo", "Number of times the point lookup select query timed out"
+ + " when fetching results", LogLevel.DEBUG, PLong.INSTANCE),
+ QUERY_SCAN_TIMEOUT_COUNTER("qso", "Number of times the scan select query timed out"
+ + " when fetching results", LogLevel.DEBUG, PLong.INSTANCE),
+ QUERY_POINTLOOKUP_FAILED_COUNTER("qplf", "Number of times the point lookup select query failed"
+ + " when fetching results", LogLevel.DEBUG, PLong.INSTANCE),
+ QUERY_SCAN_FAILED_COUNTER("qsf", "Number of times the scan select query failed when fetching"
+ + " results", LogLevel.DEBUG, PLong.INSTANCE),
// query (read) related metrics
QUERY_TIME("qt", "Query times",LogLevel.OFF, PLong.INSTANCE),
QUERY_TIMEOUT_COUNTER("qo", "Number of times query timed out",LogLevel.DEBUG, PLong.INSTANCE),
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/MetricsRegistry.java b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/MetricsRegistry.java
new file mode 100644
index 0000000..05a53ee
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/MetricsRegistry.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.monitoring;
+
+/**
+ * Interface for configurable registering Metrics interface construction
+ *
+ */
+
+public interface MetricsRegistry {
+
+ /**
+ * Interface for Registering Metrics
+ */
+ void registerMetrics(TableClientMetrics tInstance);
+
+ /**
+ * Interface for unRegistering Metrics
+ */
+ void unRegisterMetrics(TableClientMetrics tInstance);
+}
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/MutationMetricQueue.java b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/MutationMetricQueue.java
index 9efb706..b8444c3 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/MutationMetricQueue.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/MutationMetricQueue.java
@@ -17,11 +17,21 @@
*/
package org.apache.phoenix.monitoring;
+import static org.apache.phoenix.monitoring.MetricType.DELETE_BATCH_FAILED_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.DELETE_BATCH_FAILED_SIZE;
+import static org.apache.phoenix.monitoring.MetricType.DELETE_COMMIT_TIME;
+import static org.apache.phoenix.monitoring.MetricType.DELETE_MUTATION_BYTES;
+import static org.apache.phoenix.monitoring.MetricType.DELETE_MUTATION_SQL_COUNTER;
import static org.apache.phoenix.monitoring.MetricType.MUTATION_BATCH_FAILED_SIZE;
import static org.apache.phoenix.monitoring.MetricType.MUTATION_BATCH_SIZE;
import static org.apache.phoenix.monitoring.MetricType.MUTATION_BYTES;
import static org.apache.phoenix.monitoring.MetricType.MUTATION_COMMIT_TIME;
import static org.apache.phoenix.monitoring.MetricType.INDEX_COMMIT_FAILURE_SIZE;
+import static org.apache.phoenix.monitoring.MetricType.UPSERT_BATCH_FAILED_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.UPSERT_BATCH_FAILED_SIZE;
+import static org.apache.phoenix.monitoring.MetricType.UPSERT_COMMIT_TIME;
+import static org.apache.phoenix.monitoring.MetricType.UPSERT_MUTATION_BYTES;
+import static org.apache.phoenix.monitoring.MetricType.UPSERT_MUTATION_SQL_COUNTER;
import java.util.Collections;
import java.util.HashMap;
@@ -68,10 +78,21 @@ public class MutationMetricQueue {
publishedMetrics.put(tableName, publishedMetricsForTable);
}
publishedMetricsForTable.put(metric.getNumMutations().getMetricType(), metric.getNumMutations().getValue());
- publishedMetricsForTable.put(metric.getMutationsSizeBytes().getMetricType(), metric.getMutationsSizeBytes().getValue());
+ publishedMetricsForTable.put(metric.getUpsertMutationsSizeBytes().getMetricType(), metric.getUpsertMutationsSizeBytes().getValue());
+ publishedMetricsForTable.put(metric.getDeleteMutationsSizeBytes().getMetricType(), metric.getDeleteMutationsSizeBytes().getValue());
publishedMetricsForTable.put(metric.getCommitTimeForMutations().getMetricType(), metric.getCommitTimeForMutations().getValue());
+ publishedMetricsForTable.put(metric.getTotalCommitTimeForUpserts().getMetricType(), metric.getTotalCommitTimeForUpserts().getValue());
+ publishedMetricsForTable.put(metric.getTotalCommitTimeForDeletes().getMetricType(), metric.getTotalCommitTimeForDeletes().getValue());
publishedMetricsForTable.put(metric.getNumFailedMutations().getMetricType(), metric.getNumFailedMutations().getValue());
publishedMetricsForTable.put(metric.getNumOfIndexCommitFailedMutations().getMetricType(), metric.getNumOfIndexCommitFailedMutations().getValue());
+ publishedMetricsForTable.put(metric.getUpsertMutationSqlCounterSuccess().getMetricType(), metric.getUpsertMutationSqlCounterSuccess().getValue());
+ publishedMetricsForTable.put(metric.getDeleteMutationSqlCounterSuccess().getMetricType(), metric.getDeleteMutationSqlCounterSuccess().getValue());
+ publishedMetricsForTable.put(metric.getMutationsSizeBytes().getMetricType(), metric.getMutationsSizeBytes().getValue());
+ publishedMetricsForTable.put(metric.getUpsertBatchFailedSize().getMetricType(), metric.getUpsertBatchFailedSize().getValue());
+ publishedMetricsForTable.put(metric.getUpsertBatchFailedCounter().getMetricType(), metric.getUpsertBatchFailedCounter().getValue());
+ publishedMetricsForTable.put(metric.getDeleteBatchFailedSize().getMetricType(), metric.getDeleteBatchFailedSize().getValue());
+ publishedMetricsForTable.put(metric.getDeleteBatchFailedCounter().getMetricType(), metric.getDeleteBatchFailedCounter().getValue());
+
}
return publishedMetrics;
}
@@ -88,15 +109,53 @@ public class MutationMetricQueue {
private final CombinableMetric mutationsSizeBytes = new CombinableMetricImpl(MUTATION_BYTES);
private final CombinableMetric totalCommitTimeForMutations = new CombinableMetricImpl(MUTATION_COMMIT_TIME);
private final CombinableMetric numFailedMutations = new CombinableMetricImpl(MUTATION_BATCH_FAILED_SIZE);
+ private final CombinableMetric totalCommitTimeForUpserts = new CombinableMetricImpl(UPSERT_COMMIT_TIME);
+ private final CombinableMetric totalCommitTimeForDeletes = new CombinableMetricImpl(DELETE_COMMIT_TIME);
+ private final CombinableMetric upsertMutationsSizeBytes = new CombinableMetricImpl(UPSERT_MUTATION_BYTES);
+ private final CombinableMetric deleteMutationsSizeBytes = new CombinableMetricImpl(DELETE_MUTATION_BYTES);
+ private final CombinableMetric upsertMutationSqlCounterSuccess = new CombinableMetricImpl(UPSERT_MUTATION_SQL_COUNTER);
+ private final CombinableMetric deleteMutationSqlCounterSuccess = new CombinableMetricImpl(DELETE_MUTATION_SQL_COUNTER);
+ private final CombinableMetric upsertBatchFailedSize = new CombinableMetricImpl(UPSERT_BATCH_FAILED_SIZE);
+ private final CombinableMetric upsertBatchFailedCounter = new CombinableMetricImpl(UPSERT_BATCH_FAILED_COUNTER);
+ private final CombinableMetric deleteBatchFailedSize = new CombinableMetricImpl(DELETE_BATCH_FAILED_SIZE);
+ private final CombinableMetric deleteBatchFailedCounter = new CombinableMetricImpl(DELETE_BATCH_FAILED_COUNTER);
+
private final CombinableMetric numOfIndexCommitFailMutations = new CombinableMetricImpl(
INDEX_COMMIT_FAILURE_SIZE);
- public MutationMetric(long numMutations, long mutationsSizeBytes, long commitTimeForMutations, long numFailedMutations, long numOfPhase3Failed) {
+ public static final MutationMetric EMPTY_METRIC =
+ new MutationMetric(0,0,0,0,0,0,0,0,0,0,0,0,0,0);
+
+ public MutationMetric(long numMutations, long upsertMutationsSizeBytes,
+ long deleteMutationsSizeBytes, long commitTimeForUpserts, long commitTimeForDeletes,
+ long numFailedMutations, long upsertMutationSqlCounterSuccess,
+ long deleteMutationSqlCounterSuccess, long totalMutationBytes,
+ long numOfPhase3Failed, long upsertBatchFailedSize,
+ long upsertBatchFailedCounter, long deleteBatchFailedSize,
+ long deleteBatchFailedCounter) {
this.numMutations.change(numMutations);
- this.mutationsSizeBytes.change(mutationsSizeBytes);
- this.totalCommitTimeForMutations.change(commitTimeForMutations);
+ this.totalCommitTimeForUpserts.change(commitTimeForUpserts);
+ this.totalCommitTimeForDeletes.change(commitTimeForDeletes);
+ this.totalCommitTimeForMutations.change(commitTimeForUpserts + commitTimeForDeletes);
this.numFailedMutations.change(numFailedMutations);
this.numOfIndexCommitFailMutations.change(numOfPhase3Failed);
+ this.upsertMutationsSizeBytes.change(upsertMutationsSizeBytes);
+ this.deleteMutationsSizeBytes.change(deleteMutationsSizeBytes);
+ this.mutationsSizeBytes.change(totalMutationBytes);
+ this.upsertMutationSqlCounterSuccess.change(upsertMutationSqlCounterSuccess);
+ this.deleteMutationSqlCounterSuccess.change(deleteMutationSqlCounterSuccess);
+ this.upsertBatchFailedSize.change(upsertBatchFailedSize);
+ this.upsertBatchFailedCounter.change(upsertBatchFailedCounter);
+ this.deleteBatchFailedSize.change(deleteBatchFailedSize);
+ this.deleteBatchFailedCounter.change(deleteBatchFailedCounter);
+ }
+
+ public CombinableMetric getTotalCommitTimeForUpserts() {
+ return totalCommitTimeForUpserts;
+ }
+
+ public CombinableMetric getTotalCommitTimeForDeletes() {
+ return totalCommitTimeForDeletes;
}
public CombinableMetric getCommitTimeForMutations() {
@@ -119,12 +178,54 @@ public class MutationMetricQueue {
return numOfIndexCommitFailMutations;
}
+ public CombinableMetric getUpsertMutationsSizeBytes() {
+ return upsertMutationsSizeBytes;
+ }
+
+ public CombinableMetric getDeleteMutationsSizeBytes() {
+ return deleteMutationsSizeBytes;
+ }
+
+ public CombinableMetric getUpsertMutationSqlCounterSuccess() {
+ return upsertMutationSqlCounterSuccess;
+ }
+
+ public CombinableMetric getDeleteMutationSqlCounterSuccess() {
+ return deleteMutationSqlCounterSuccess;
+ }
+
+ public CombinableMetric getUpsertBatchFailedSize() {
+ return upsertBatchFailedSize;
+ }
+
+ public CombinableMetric getUpsertBatchFailedCounter() {
+ return upsertBatchFailedCounter;
+ }
+
+ public CombinableMetric getDeleteBatchFailedSize() {
+ return deleteBatchFailedSize;
+ }
+
+ public CombinableMetric getDeleteBatchFailedCounter() {
+ return deleteBatchFailedCounter;
+ }
+
public void combineMetric(MutationMetric other) {
this.numMutations.combine(other.numMutations);
- this.mutationsSizeBytes.combine(other.mutationsSizeBytes);
+ this.totalCommitTimeForUpserts.combine(other.totalCommitTimeForUpserts);
+ this.totalCommitTimeForDeletes.combine(other.totalCommitTimeForDeletes);
this.totalCommitTimeForMutations.combine(other.totalCommitTimeForMutations);
this.numFailedMutations.combine(other.numFailedMutations);
this.numOfIndexCommitFailMutations.combine(other.numOfIndexCommitFailMutations);
+ this.upsertMutationsSizeBytes.combine(other.upsertMutationsSizeBytes);
+ this.deleteMutationsSizeBytes.combine(other.deleteMutationsSizeBytes);
+ this.mutationsSizeBytes.combine(other.mutationsSizeBytes);
+ this.upsertMutationSqlCounterSuccess.combine(other.upsertMutationSqlCounterSuccess);
+ this.deleteMutationSqlCounterSuccess.combine(other.deleteMutationSqlCounterSuccess);
+ this.upsertBatchFailedSize.combine(other.upsertBatchFailedSize);
+ this.upsertBatchFailedCounter.combine(other.upsertBatchFailedCounter);
+ this.deleteBatchFailedSize.combine(other.deleteBatchFailedSize);
+ this.deleteBatchFailedCounter.combine(other.deleteBatchFailedCounter);
}
}
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/NoOpTableMetricsManager.java b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/NoOpTableMetricsManager.java
new file mode 100644
index 0000000..23248d0
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/NoOpTableMetricsManager.java
@@ -0,0 +1,53 @@
+/*
+ * 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.Collections;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * TableMetricsManager will be replaced by this case
+ * incase of tableMetrics flag is set to false.
+ */
+
+
+public class NoOpTableMetricsManager extends TableMetricsManager {
+
+ public static final NoOpTableMetricsManager noOpsTableMetricManager = new NoOpTableMetricsManager();
+
+ private NoOpTableMetricsManager() {
+ super();
+ }
+
+ @Override public void updateMetrics(String tableName, MetricType type, long value) {
+
+ }
+
+ @Override public void pushMetricsFromConnInstance(Map<String, Map<MetricType, Long>> map) {
+
+ }
+
+ @Override public void clearTableLevelMetrics() {
+
+ }
+
+ @Override public Map<String, List<PhoenixTableMetric>> getTableLevelMetrics() {
+ return Collections.emptyMap();
+ }
+}
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/OverAllQueryMetrics.java b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/OverAllQueryMetrics.java
index ae3edab..3d81c2a 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/OverAllQueryMetrics.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/OverAllQueryMetrics.java
@@ -20,6 +20,10 @@ package org.apache.phoenix.monitoring;
import static org.apache.phoenix.monitoring.MetricType.CACHE_REFRESH_SPLITS_COUNTER;
import static org.apache.phoenix.monitoring.MetricType.NUM_PARALLEL_SCANS;
import static org.apache.phoenix.monitoring.MetricType.QUERY_FAILED_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.QUERY_POINTLOOKUP_FAILED_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.QUERY_POINTLOOKUP_TIMEOUT_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.QUERY_SCAN_FAILED_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.QUERY_SCAN_TIMEOUT_COUNTER;
import static org.apache.phoenix.monitoring.MetricType.QUERY_TIMEOUT_COUNTER;
import static org.apache.phoenix.monitoring.MetricType.RESULT_SET_TIME_MS;
import static org.apache.phoenix.monitoring.MetricType.WALL_CLOCK_TIME_MS;
@@ -40,7 +44,11 @@ public class OverAllQueryMetrics {
private final CombinableMetric wallClockTimeMS;
private final CombinableMetric resultSetTimeMS;
private final CombinableMetric queryTimedOut;
+ private final CombinableMetric queryPointLookupTimedOut;
+ private final CombinableMetric queryScanTimedOut;
private final CombinableMetric queryFailed;
+ private final CombinableMetric queryPointLookupFailed;
+ private final CombinableMetric queryScanFailed;
private final CombinableMetric cacheRefreshedDueToSplits;
public OverAllQueryMetrics(boolean isRequestMetricsEnabled, LogLevel connectionLogLevel) {
@@ -56,8 +64,12 @@ public class OverAllQueryMetrics {
connectionLogLevel, RESULT_SET_TIME_MS);
queryTimedOut = MetricUtil.getCombinableMetric(isRequestMetricsEnabled,
connectionLogLevel, QUERY_TIMEOUT_COUNTER);
+ queryPointLookupTimedOut = MetricUtil.getCombinableMetric(isRequestMetricsEnabled,connectionLogLevel, QUERY_POINTLOOKUP_TIMEOUT_COUNTER);
+ queryScanTimedOut = MetricUtil.getCombinableMetric(isRequestMetricsEnabled,connectionLogLevel, QUERY_SCAN_TIMEOUT_COUNTER);
queryFailed = MetricUtil.getCombinableMetric(isRequestMetricsEnabled,
connectionLogLevel, QUERY_FAILED_COUNTER);
+ queryPointLookupFailed = MetricUtil.getCombinableMetric(isRequestMetricsEnabled,connectionLogLevel, QUERY_POINTLOOKUP_FAILED_COUNTER);
+ queryScanFailed = MetricUtil.getCombinableMetric(isRequestMetricsEnabled,connectionLogLevel, QUERY_SCAN_FAILED_COUNTER);
cacheRefreshedDueToSplits = MetricUtil.getCombinableMetric(isRequestMetricsEnabled,
connectionLogLevel, CACHE_REFRESH_SPLITS_COUNTER);
}
@@ -70,10 +82,26 @@ public class OverAllQueryMetrics {
queryTimedOut.increment();
}
+ public void queryPointLookupTimedOut() {
+ queryPointLookupTimedOut.increment();
+ }
+
+ public void queryScanTimedOut() {
+ queryScanTimedOut.increment();
+ }
+
public void queryFailed() {
queryFailed.increment();
}
+ public void queryPointLookupFailed() {
+ queryPointLookupFailed.increment();
+ }
+
+ public void queryScanFailed() {
+ queryScanFailed.increment();
+ }
+
public void cacheRefreshedDueToSplits() {
cacheRefreshedDueToSplits.increment();
}
@@ -118,7 +146,11 @@ public class OverAllQueryMetrics {
metricsForPublish.put(wallClockTimeMS.getMetricType(), wallClockTimeMS.getValue());
metricsForPublish.put(resultSetTimeMS.getMetricType(), resultSetTimeMS.getValue());
metricsForPublish.put(queryTimedOut.getMetricType(), queryTimedOut.getValue());
+ metricsForPublish.put(queryPointLookupTimedOut.getMetricType(), queryPointLookupTimedOut.getValue());
+ metricsForPublish.put(queryScanTimedOut.getMetricType(), queryScanTimedOut.getValue());
metricsForPublish.put(queryFailed.getMetricType(), queryFailed.getValue());
+ metricsForPublish.put(queryPointLookupFailed.getMetricType(), queryPointLookupFailed.getValue());
+ metricsForPublish.put(queryScanFailed.getMetricType(), queryScanFailed.getValue());
metricsForPublish.put(cacheRefreshedDueToSplits.getMetricType(), cacheRefreshedDueToSplits.getValue());
return metricsForPublish;
}
@@ -128,7 +160,11 @@ public class OverAllQueryMetrics {
wallClockTimeMS.reset();
resultSetTimeMS.reset();
queryTimedOut.reset();
+ queryPointLookupTimedOut.reset();
+ queryScanTimedOut.reset();
queryFailed.reset();
+ queryPointLookupFailed.reset();
+ queryScanFailed.reset();
cacheRefreshedDueToSplits.reset();
queryWatch.stop();
resultSetWatch.stop();
@@ -137,8 +173,14 @@ public class OverAllQueryMetrics {
public OverAllQueryMetrics combine(OverAllQueryMetrics metric) {
cacheRefreshedDueToSplits.combine(metric.cacheRefreshedDueToSplits);
queryFailed.combine(metric.queryFailed);
+ queryPointLookupFailed.combine(metric.queryPointLookupFailed);
+ queryScanFailed.combine(metric.queryScanFailed);
queryTimedOut.combine(metric.queryTimedOut);
+ queryPointLookupTimedOut.combine(metric.queryPointLookupTimedOut);
+ queryScanTimedOut.combine(metric.queryScanTimedOut);
numParallelScans.combine(metric.numParallelScans);
+ wallClockTimeMS.combine(metric.wallClockTimeMS);
+ resultSetTimeMS.combine(metric.resultSetTimeMS);
return this;
}
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/ScanMetricsHolder.java b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/ScanMetricsHolder.java
index dd6603e..08f2fb8 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/ScanMetricsHolder.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/ScanMetricsHolder.java
@@ -28,6 +28,7 @@ import static org.apache.phoenix.monitoring.MetricType.COUNT_ROWS_SCANNED;
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.SCAN_BYTES;
import java.io.IOException;
import java.util.Map;
@@ -49,6 +50,7 @@ public class ScanMetricsHolder {
private final CombinableMetric countOfRemoteRPCRetries;
private final CombinableMetric countOfRowsScanned;
private final CombinableMetric countOfRowsFiltered;
+ private final CombinableMetric countOfBytesScanned;
private Map<String, Long> scanMetricMap;
private Object scan;
@@ -78,6 +80,7 @@ public class ScanMetricsHolder {
countOfRemoteRPCRetries = readMetrics.allotMetric(COUNT_REMOTE_RPC_RETRIES, tableName);
countOfRowsScanned = readMetrics.allotMetric(COUNT_ROWS_SCANNED, tableName);
countOfRowsFiltered = readMetrics.allotMetric(COUNT_ROWS_FILTERED, tableName);
+ countOfBytesScanned = readMetrics.allotMetric(SCAN_BYTES,tableName);
}
public CombinableMetric getCountOfRemoteRPCcalls() {
@@ -128,6 +131,10 @@ public class ScanMetricsHolder {
return scanMetricMap;
}
+ public CombinableMetric getCountOfBytesScanned() {
+ return countOfBytesScanned;
+ }
+
public void setScanMetricMap(Map<String, Long> scanMetricMap) {
this.scanMetricMap = scanMetricMap;
}
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/TableClientMetrics.java b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/TableClientMetrics.java
new file mode 100644
index 0000000..b640f29
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/TableClientMetrics.java
@@ -0,0 +1,182 @@
+/**
+ * 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.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.phoenix.monitoring.MetricType.MUTATION_BATCH_SIZE;
+import static org.apache.phoenix.monitoring.MetricType.MUTATION_BATCH_FAILED_SIZE;
+import static org.apache.phoenix.monitoring.MetricType.MUTATION_BYTES;
+import static org.apache.phoenix.monitoring.MetricType.QUERY_POINTLOOKUP_FAILED_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.QUERY_POINTLOOKUP_TIMEOUT_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.QUERY_SCAN_FAILED_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.QUERY_SCAN_TIMEOUT_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.RESULT_SET_TIME_MS;
+import static org.apache.phoenix.monitoring.MetricType.SELECT_FAILED_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.SELECT_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.SELECT_SUCCESS_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.UPSERT_MUTATION_BYTES;
+import static org.apache.phoenix.monitoring.MetricType.UPSERT_MUTATION_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.DELETE_BATCH_FAILED_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.DELETE_BATCH_FAILED_SIZE;
+import static org.apache.phoenix.monitoring.MetricType.DELETE_COMMIT_TIME;
+import static org.apache.phoenix.monitoring.MetricType.DELETE_MUTATION_BYTES;
+import static org.apache.phoenix.monitoring.MetricType.DELETE_MUTATION_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.MUTATION_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.MUTATION_COMMIT_TIME;
+import static org.apache.phoenix.monitoring.MetricType.TASK_END_TO_END_TIME;
+import static org.apache.phoenix.monitoring.MetricType.COUNT_ROWS_SCANNED;
+import static org.apache.phoenix.monitoring.MetricType.QUERY_FAILED_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.QUERY_TIMEOUT_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.SCAN_BYTES;
+import static org.apache.phoenix.monitoring.MetricType.SELECT_POINTLOOKUP_SUCCESS_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.SELECT_POINTLOOKUP_FAILED_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.SELECT_SQL_QUERY_TIME;
+import static org.apache.phoenix.monitoring.MetricType.SELECT_SCAN_SUCCESS_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.SELECT_SCAN_FAILED_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.DELETE_FAILED_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.DELETE_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.DELETE_SQL_QUERY_TIME;
+import static org.apache.phoenix.monitoring.MetricType.DELETE_SUCCESS_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.UPSERT_BATCH_FAILED_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.UPSERT_BATCH_FAILED_SIZE;
+import static org.apache.phoenix.monitoring.MetricType.UPSERT_COMMIT_TIME;
+import static org.apache.phoenix.monitoring.MetricType.UPSERT_FAILED_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.UPSERT_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.UPSERT_SQL_QUERY_TIME;
+import static org.apache.phoenix.monitoring.MetricType.UPSERT_SUCCESS_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.UPSERT_AGGREGATE_SUCCESS_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.UPSERT_AGGREGATE_FAILURE_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.DELETE_AGGREGATE_SUCCESS_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.DELETE_AGGREGATE_FAILURE_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.SELECT_AGGREGATE_SUCCESS_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.SELECT_AGGREGATE_FAILURE_SQL_COUNTER;
+
+/**
+ * This is used by TableMetricsManager class to store instance of
+ * object associated with a tableName.
+ */
+public class TableClientMetrics {
+
+ public enum TableMetrics {
+ TABLE_MUTATION_BATCH_FAILED_SIZE(MUTATION_BATCH_FAILED_SIZE), TABLE_MUTATION_BATCH_SIZE(
+ MUTATION_BATCH_SIZE), TABLE_MUTATION_BYTES(
+ MUTATION_BYTES), TABLE_UPSERT_MUTATION_BYTES(
+ UPSERT_MUTATION_BYTES), TABLE_UPSERT_MUTATION_SQL_COUNTER(
+ UPSERT_MUTATION_SQL_COUNTER), TABLE_DELETE_MUTATION_BYTES(
+ DELETE_MUTATION_BYTES), TABLE_DELETE_MUTATION_SQL_COUNTER(
+ DELETE_MUTATION_SQL_COUNTER), TABLE_MUTATION_SQL_COUNTER(
+ MUTATION_SQL_COUNTER), TABLE_MUTATION_COMMIT_TIME(
+ MUTATION_COMMIT_TIME), TABLE_UPSERT_SQL_COUNTER(
+ UPSERT_SQL_COUNTER), TABLE_UPSERT_SQL_QUERY_TIME(
+ UPSERT_SQL_QUERY_TIME), TABLE_SUCCESS_UPSERT_SQL_COUNTER(
+ UPSERT_SUCCESS_SQL_COUNTER), TABLE_FAILED_UPSERT_SQL_COUNTER(
+ UPSERT_FAILED_SQL_COUNTER), TABLE_UPSERT_BATCH_FAILED_SIZE(
+ UPSERT_BATCH_FAILED_SIZE), TABLE_UPSERT_BATCH_FAILED_COUNTER(
+ UPSERT_BATCH_FAILED_COUNTER), TABLE_DELETE_SQL_COUNTER(
+ DELETE_SQL_COUNTER), TABLE_DELETE_SQL_QUERY_TIME(
+ DELETE_SQL_QUERY_TIME), TABLE_SUCCESS_DELETE_SQL_COUNTER(
+ DELETE_SUCCESS_SQL_COUNTER), TABLE_FAILED_DELETE_SQL_COUNTER(
+ DELETE_FAILED_SQL_COUNTER), TABLE_DELETE_BATCH_FAILED_SIZE(
+ DELETE_BATCH_FAILED_SIZE), TABLE_DELETE_BATCH_FAILED_COUNTER(
+ DELETE_BATCH_FAILED_COUNTER), TABLE_UPSERT_COMMIT_TIME(
+ UPSERT_COMMIT_TIME), TABLE_DELETE_COMMIT_TIME(
+ DELETE_COMMIT_TIME), TABLE_TASK_END_TO_END_TIME(
+ TASK_END_TO_END_TIME), TABLE_COUNT_ROWS_SCANNED(
+ COUNT_ROWS_SCANNED), TABLE_QUERY_FAILED_COUNTER(
+ QUERY_FAILED_COUNTER), TABLE_QUERY_POINTLOOKUP_FAILED_COUNTER(
+ QUERY_POINTLOOKUP_FAILED_COUNTER), TABLE_QUERY_SCAN_FAILED_COUNTER(
+ QUERY_SCAN_FAILED_COUNTER), TABLE_QUERY_TIMEOUT_COUNTER(
+ QUERY_TIMEOUT_COUNTER), TABLE_QUERY_POINTLOOKUP_TIMEOUT_COUNTER(
+ QUERY_POINTLOOKUP_TIMEOUT_COUNTER), TABLE_QUERY_SCAN_TIMEOUT_COUNTER(
+ QUERY_SCAN_TIMEOUT_COUNTER), TABLE_SELECT_QUERY_RESULT_SET_MS(
+ RESULT_SET_TIME_MS), TABLE_SCANBYTES(SCAN_BYTES), TABLE_SELECT_SQL_COUNTER(
+ SELECT_SQL_COUNTER), TABLE_SELECT_SQL_QUERY_TIME(
+ SELECT_SQL_QUERY_TIME), TABLE_SUCCESS_SELECT_SQL_COUNTER(
+ SELECT_SUCCESS_SQL_COUNTER), TABLE_FAILED_SELECT_SQL_COUNTER(
+ SELECT_FAILED_SQL_COUNTER), TABLE_SELECT_POINTLOOKUP_COUNTER_SUCCESS(
+ SELECT_POINTLOOKUP_SUCCESS_SQL_COUNTER), TABLE_SELECT_POINTLOOKUP_COUNTER_FAILED(
+ SELECT_POINTLOOKUP_FAILED_SQL_COUNTER), TABLE_SELECT_SCAN_COUNTER_SUCCESS(
+ SELECT_SCAN_SUCCESS_SQL_COUNTER), TABLE_SELECT_SCAN_COUNTER_FAILED(
+ SELECT_SCAN_FAILED_SQL_COUNTER), TABLE_UPSERT_AGGREGATE_SUCCESS_SQL_COUNTER(
+ UPSERT_AGGREGATE_SUCCESS_SQL_COUNTER), TABLE_UPSERT_AGGREGATE_FAILURE_SQL_COUNTER(
+ UPSERT_AGGREGATE_FAILURE_SQL_COUNTER), TABLE_DELETE_AGGREGATE_SUCCESS_SQL_COUNTER(
+ DELETE_AGGREGATE_SUCCESS_SQL_COUNTER), TABLE_DELETE_AGGREGATE_FAILURE_SQL_COUNTER(
+ DELETE_AGGREGATE_FAILURE_SQL_COUNTER), TABLE_SELECT_AGGREGATE_SUCCESS_SQL_COUNTER(
+ SELECT_AGGREGATE_SUCCESS_SQL_COUNTER), TABLE_SELECT_AGGREGATE_FAILURE_SQL_COUNTER(
+ SELECT_AGGREGATE_FAILURE_SQL_COUNTER);
+
+ private final MetricType metricType;
+ private PhoenixTableMetric metric;
+
+ TableMetrics(MetricType metricType) {
+ this.metricType = metricType;
+ }
+ }
+
+ private final String tableName;
+ private final Map<MetricType, PhoenixTableMetric> metricRegister;
+
+ public TableClientMetrics(final String tableName) {
+ this.tableName = tableName;
+ metricRegister = new HashMap<>();
+ for (TableMetrics tableMetric : TableMetrics.values()) {
+ tableMetric.metric = new PhoenixTableMetricImpl(tableMetric.metricType);
+ metricRegister.put(tableMetric.metricType, tableMetric.metric);
+ }
+ }
+
+ /**
+ * This function is used to update the value of Metric
+ * Incase of counter val will passed as 1.
+ *
+ * @param type metric type
+ * @param val update value. In case of counters, this will be 1
+ */
+ public void changeMetricValue(MetricType type, long val) {
+ if (!metricRegister.containsKey(type)) {
+ return;
+ }
+ PhoenixTableMetric metric = metricRegister.get(type);
+ metric.change(val);
+ }
+
+ public String getTableName() {
+ return tableName;
+ }
+
+ /**
+ * This method is called to aggregate all the Metrics across all Tables in Phoenix.
+ *
+ * @return map of table name -> list of TableMetric.
+ */
+ public List<PhoenixTableMetric> getMetricMap() {
+ List<PhoenixTableMetric> metricsList = new ArrayList<>();
+ for (PhoenixTableMetric value : metricRegister.values()) {
+ metricsList.add(value);
+ }
+ return metricsList;
+ }
+
+ public Map<MetricType, PhoenixTableMetric> getMetricRegistry() {
+ return metricRegister;
+ }
+
+}
\ No newline at end of file
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/TableMetricsManager.java b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/TableMetricsManager.java
new file mode 100644
index 0000000..b9ffcc9
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/TableMetricsManager.java
@@ -0,0 +1,290 @@
+/**
+ * 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.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Strings;
+import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
+import org.apache.phoenix.query.QueryServicesOptions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Central place where we keep track of all the Table Level metrics. Register each tableMetrics and
+ * store the instance of it associated with TableName in a map
+ * This class exposes following functions as static methods to help catch all execptions
+ * 1.clearTableLevelMetricsMethod
+ * 2.getTableMetricsMethod
+ * 3.pushMetricsFromConnInstanceMethod
+ * 4.updateMetricsMethod
+ */
+
+public class TableMetricsManager {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(TableMetricsManager.class);
+ private static final Set<String> allowedListOfTableNames = new HashSet<>();
+ private static volatile boolean isTableLevelMetricsEnabled;
+ private static volatile boolean isMetricPublisherEnabled;
+ private static volatile ConcurrentMap<String, TableClientMetrics> tableClientMetricsMapping = null;
+ // Singleton object
+ private static volatile TableMetricsManager tableMetricsManager = null;
+ private static volatile MetricPublisherSupplierFactory mPublisher = null;
+ private static volatile QueryServicesOptions options = null;
+
+ public TableMetricsManager(QueryServicesOptions ops) {
+ options = ops;
+ isTableLevelMetricsEnabled = options.isTableLevelMetricsEnabled();
+ LOGGER.info(String.format("Phoenix Table metrics enabled status: %s",
+ isTableLevelMetricsEnabled));
+ tableClientMetricsMapping = new ConcurrentHashMap<>();
+
+ String tableNamesList = options.getAllowedListTableNames();
+ if (tableNamesList != null && !tableNamesList.isEmpty()) {
+ for (String tableName : tableNamesList.split(",")) {
+ allowedListOfTableNames.add(tableName);
+ }
+ }
+ isMetricPublisherEnabled = options.isMetricPublisherEnabled();
+ LOGGER.info(String.format("Phoenix table level metrics publisher enabled status %s",
+ isMetricPublisherEnabled));
+ }
+
+ public TableMetricsManager() {
+
+ }
+
+ /**
+ * Method to provide instance of TableMetricsManager(Create if needed in thread safe manner)
+ *
+ * @return
+ */
+ private static TableMetricsManager getInstance() {
+
+ TableMetricsManager localRef = tableMetricsManager;
+ if (localRef == null) {
+ synchronized (TableMetricsManager.class) {
+ if (localRef == null) {
+ QueryServicesOptions options = QueryServicesOptions.withDefaults();
+ if (!options.isTableLevelMetricsEnabled()) {
+ localRef = tableMetricsManager =
+ NoOpTableMetricsManager.noOpsTableMetricManager;
+ return localRef;
+ }
+ localRef = tableMetricsManager = new TableMetricsManager(options);
+ LOGGER.info("Phoenix Table metrics created object for metrics manager");
+ if (isMetricPublisherEnabled) {
+ String className = options.getMetricPublisherClass();
+ if (className != null) {
+ MetricServiceResolver mResolver = new MetricServiceResolver();
+ LOGGER.info(String.format(
+ "Phoenix table level metrics publisher className %s",
+ className));
+ try {
+ mPublisher = mResolver.instantiate(className);
+ mPublisher.registerMetricProvider();
+ } catch (Throwable e) {
+ LOGGER.error("The exception from metric publish Function", e);
+ }
+
+ } else {
+ LOGGER.error(
+ "Phoenix table level metrics publisher className cannot be null");
+ }
+
+ }
+ }
+ }
+ }
+ return localRef;
+ }
+
+ @VisibleForTesting public static void setInstance(TableMetricsManager metricsManager) {
+ tableMetricsManager = metricsManager;
+ }
+
+ public static void updateMetricsMethod(String tableName, MetricType type, long value) {
+ try {
+ TableMetricsManager.getInstance().updateMetrics(tableName, type, value);
+ } catch (Exception e) {
+ LOGGER.error("Failed updating Phoenix table level metrics", e);
+ }
+ }
+
+ public static void pushMetricsFromConnInstanceMethod(Map<String, Map<MetricType, Long>> map) {
+ try {
+ TableMetricsManager.getInstance().pushMetricsFromConnInstance(map);
+ } catch (Exception e) {
+ LOGGER.error("Failed pushing Phoenix table level metrics", e);
+ }
+ }
+
+ public static Map<String, List<PhoenixTableMetric>> getTableMetricsMethod() {
+ try {
+ return TableMetricsManager.getInstance().getTableLevelMetrics();
+ } catch (Exception e) {
+ LOGGER.error("Failed retrieving table level Metrics", e);
+ }
+ return null;
+ }
+
+ public static void clearTableLevelMetricsMethod() {
+ try {
+ TableMetricsManager.getInstance().clearTableLevelMetrics();
+ } catch (Exception e) {
+ LOGGER.error("Failed resetting table level Metrics", e);
+ }
+ }
+
+ @VisibleForTesting public static Long getMetricValue(String tableName, MetricType type) {
+ TableClientMetrics tableMetrics = getInstance().getTableClientMetrics(tableName);
+ if (tableMetrics == null) {
+ return null;
+ }
+ for (PhoenixTableMetric metric : tableMetrics.getMetricMap()) {
+ if (metric.getMetricType() == type) {
+ return metric.getValue();
+ }
+ }
+ return null;
+ }
+
+ // static methods to push, update or retrieve TableLevel Metrics.
+
+ /**
+ * This function is provided as hook to publish the tableLevel Metrics to
+ * LocalStore(tablePhoenixMapping).
+ *
+ * @param map of tableName to pair of (MetricType, Metric Value)
+ */
+ public void pushMetricsFromConnInstance(Map<String, Map<MetricType, Long>> map) {
+
+ if (map == null) {
+ LOGGER.debug("Phoenix table level metrics input map cannott be null");
+ return;
+ }
+
+ long startTime = EnvironmentEdgeManager.currentTime();
+ for (Map.Entry<String, Map<MetricType, Long>> tableEntry : map.entrySet()) {
+ for (Map.Entry<MetricType, Long> metricEntry : tableEntry.getValue().entrySet()) {
+ updateMetrics(tableEntry.getKey(), metricEntry.getKey(), metricEntry.getValue());
+ }
+ }
+
+ LOGGER.debug(String.format(
+ "Phoenix table level metrics completed updating metrics from conn instance, timetaken:\t%d",
+ +EnvironmentEdgeManager.currentTime() - startTime));
+ }
+
+ /**
+ * This function will be used to add individual MetricType to LocalStore.
+ *
+ * @param tableName
+ * @param type
+ * @param value
+ */
+ public void updateMetrics(String tableName, MetricType type, long value) {
+
+ long startTime = EnvironmentEdgeManager.currentTime();
+
+ TableClientMetrics tInstance = getTableClientMetrics(tableName);
+ if (tInstance == null) {
+ LOGGER.debug("Table level client metrics are disabled for table: " + tableName);
+ return;
+ }
+ tInstance.changeMetricValue(type, value);
+
+ LOGGER.debug(String.format("Phoenix table level metrics completed updating metric"
+ + " %s to value %s, timetaken = %s", type, value,
+ EnvironmentEdgeManager.currentTime() - startTime));
+ }
+
+ /**
+ * Get Table specific metrics object and create if not initialized(thread safe)
+ *
+ * @param tableName
+ * @return TableClientMetrics object
+ */
+ private TableClientMetrics getTableClientMetrics(String tableName) {
+
+ if (Strings.isNullOrEmpty(tableName)) {
+ LOGGER.debug("Phoenix Table metrics TableName cannot be null or empty");
+ return null;
+ }
+
+ if (!allowedListOfTableNames.isEmpty() && !allowedListOfTableNames.contains(tableName)) {
+ return null;
+ }
+
+ TableClientMetrics tInstance;
+ tInstance = tableClientMetricsMapping.get(tableName);
+ if (tInstance == null) {
+ synchronized (TableMetricsManager.class) {
+ tInstance = tableClientMetricsMapping.get(tableName);
+ if (tInstance == null) {
+ LOGGER.info(String.format("Phoenix Table metrics creating object for table: %s",
+ tableName));
+ tInstance = new TableClientMetrics(tableName);
+ if (isMetricPublisherEnabled && mPublisher != null) {
+ mPublisher.registerMetrics(tInstance);
+ }
+ tableClientMetricsMapping.put(tableName, tInstance);
+ }
+ }
+ }
+ return tInstance;
+ }
+
+ /**
+ * Publish the metrics to wherever you want them published.
+ *
+ * @return map of table name ->TableMetric
+ */
+ public Map<String, List<PhoenixTableMetric>> getTableLevelMetrics() {
+
+ long startTime = EnvironmentEdgeManager.currentTime();
+ Map<String, List<PhoenixTableMetric>> map = new HashMap<>();
+ for (Map.Entry<String, TableClientMetrics> entry : tableClientMetricsMapping.entrySet()) {
+ map.put(entry.getKey(), entry.getValue().getMetricMap());
+ }
+ long timeTakenForMetricConversion = EnvironmentEdgeManager.currentTime() - startTime;
+ LOGGER.info(String.format("Phoenix Table metrics fetching complete, timeTaken: \t%d",
+ +timeTakenForMetricConversion));
+ return map;
+ }
+
+ /**
+ * Helps reset the localstore(tableClientMetricsMapping)
+ */
+ public void clearTableLevelMetrics() {
+ if (tableClientMetricsMapping != null) {
+ tableClientMetricsMapping.clear();
+ }
+ LOGGER.info("Phoenix Table metrics clearing complete");
+ }
+
+ public void clear() {
+ TableMetricsManager.clearTableLevelMetricsMethod();
+ }
+}
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServices.java b/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServices.java
index 6b85e79..1c2ba13 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServices.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServices.java
@@ -224,6 +224,12 @@ public interface QueryServices extends SQLCloseable {
public static final String DEFAULT_TRANSACTION_PROVIDER_ATTRIB = "phoenix.table.transaction.provider.default";
public static final String GLOBAL_METRICS_ENABLED = "phoenix.query.global.metrics.enabled";
+ public static final String TABLE_LEVEL_METRICS_ENABLED = "phoenix.monitoring.tableMetrics.enabled";
+ public static final String METRIC_PUBLISHER_ENABLED = "phoenix.monitoring.metricsPublisher.enabled";
+ public static final String METRIC_PUBLISHER_CLASS_NAME = "phoenix.monitoring.metricProvider.className";
+ public static final String ALLOWED_LIST_FOR_TABLE_LEVEL_METRICS = "phoenix.monitoring.allowedTableNames.list";
+
+
// Tag Name to determine the Phoenix Client Type
public static final String CLIENT_METRICS_TAG = "phoenix.client.metrics.tag";
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java b/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java
index 58284a7..dcd937e 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java
@@ -18,6 +18,7 @@
package org.apache.phoenix.query;
import static org.apache.hadoop.hbase.HConstants.DEFAULT_HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD;
+import static org.apache.phoenix.query.QueryServices.ALLOWED_LIST_FOR_TABLE_LEVEL_METRICS;
import static org.apache.phoenix.query.QueryServices.ALLOW_ONLINE_TABLE_SCHEMA_UPDATE;
import static org.apache.phoenix.query.QueryServices.ALLOW_VIEWS_ADD_NEW_CF_BASE_TABLE;
import static org.apache.phoenix.query.QueryServices.AUTO_UPGRADE_ENABLED;
@@ -60,6 +61,8 @@ import static org.apache.phoenix.query.QueryServices.MAX_SERVER_CACHE_TIME_TO_LI
import static org.apache.phoenix.query.QueryServices.MAX_SERVER_METADATA_CACHE_SIZE_ATTRIB;
import static org.apache.phoenix.query.QueryServices.MAX_SPOOL_TO_DISK_BYTES_ATTRIB;
import static org.apache.phoenix.query.QueryServices.MAX_TENANT_MEMORY_PERC_ATTRIB;
+import static org.apache.phoenix.query.QueryServices.METRIC_PUBLISHER_CLASS_NAME;
+import static org.apache.phoenix.query.QueryServices.METRIC_PUBLISHER_ENABLED;
import static org.apache.phoenix.query.QueryServices.MIN_STATS_UPDATE_FREQ_MS_ATTRIB;
import static org.apache.phoenix.query.QueryServices.MUTATE_BATCH_SIZE_ATTRIB;
import static org.apache.phoenix.query.QueryServices.NUM_RETRIES_FOR_SCHEMA_UPDATE_CHECK;
@@ -84,6 +87,7 @@ import static org.apache.phoenix.query.QueryServices.STATS_COLLECTION_ENABLED;
import static org.apache.phoenix.query.QueryServices.STATS_GUIDEPOST_WIDTH_BYTES_ATTRIB;
import static org.apache.phoenix.query.QueryServices.STATS_UPDATE_FREQ_MS_ATTRIB;
import static org.apache.phoenix.query.QueryServices.STATS_USE_CURRENT_TIME_ATTRIB;
+import static org.apache.phoenix.query.QueryServices.TABLE_LEVEL_METRICS_ENABLED;
import static org.apache.phoenix.query.QueryServices.THREAD_POOL_SIZE_ATTRIB;
import static org.apache.phoenix.query.QueryServices.THREAD_TIMEOUT_MS_ATTRIB;
import static org.apache.phoenix.query.QueryServices.TRACING_BATCH_SIZE;
@@ -101,6 +105,7 @@ import static org.apache.phoenix.query.QueryServices.PHOENIX_TTL_SERVER_SIDE_MAS
import java.util.Map.Entry;
+import com.google.common.annotations.VisibleForTesting;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Coprocessor;
import org.apache.hadoop.hbase.client.Consistency;
@@ -279,6 +284,10 @@ public class QueryServicesOptions {
public static final String DEFAULT_TRANSACTION_PROVIDER = TransactionFactory.Provider.getDefault().name();
public static final boolean DEFAULT_TRANSACTIONS_ENABLED = false;
public static final boolean DEFAULT_IS_GLOBAL_METRICS_ENABLED = true;
+ public static final boolean DEFAULT_IS_TABLE_LEVEL_METRICS_ENABLED = false;
+ public static final boolean DEFAULT_IS_METRIC_PUBLISHER_ENABLED = false;
+ public static final String DEFAULT_ALLOWED_LIST_FOR_TABLE_LEVEL_METRICS = null; //All the tables metrics will be allowed.
+ public static final String DEFAULT_METRIC_PUBLISHER_CLASS_NAME = "org.apache.phoenix.monitoring.JmxMetricProvider";
public static final String DEFAULT_CLIENT_METRICS_TAG = "FAT_CLIENT";
public static final boolean DEFAULT_TRANSACTIONAL = false;
@@ -684,10 +693,37 @@ public class QueryServicesOptions {
return config.getBoolean(GLOBAL_METRICS_ENABLED, DEFAULT_IS_GLOBAL_METRICS_ENABLED);
}
+ public String getMetricPublisherClass() {
+ return config.get(METRIC_PUBLISHER_CLASS_NAME, DEFAULT_METRIC_PUBLISHER_CLASS_NAME);
+ }
+
public boolean isUseByteBasedRegex() {
return config.getBoolean(USE_BYTE_BASED_REGEX_ATTRIB, DEFAULT_USE_BYTE_BASED_REGEX);
}
+ public String getAllowedListTableNames() {
+ return config.get(ALLOWED_LIST_FOR_TABLE_LEVEL_METRICS,
+ DEFAULT_ALLOWED_LIST_FOR_TABLE_LEVEL_METRICS);
+ }
+
+ public boolean isTableLevelMetricsEnabled() {
+ return config
+ .getBoolean(TABLE_LEVEL_METRICS_ENABLED, DEFAULT_IS_TABLE_LEVEL_METRICS_ENABLED);
+ }
+
+ public void setTableLevelMetricsEnabled() {
+ set(TABLE_LEVEL_METRICS_ENABLED, true);
+ }
+
+ public boolean isMetricPublisherEnabled() {
+ return config.getBoolean(METRIC_PUBLISHER_ENABLED, DEFAULT_IS_METRIC_PUBLISHER_ENABLED);
+ }
+
+ @VisibleForTesting
+ public void setAllowedListForTableLevelMetrics(String tableNameList){
+ set(ALLOWED_LIST_FOR_TABLE_LEVEL_METRICS,tableNameList);
+ }
+
public int getScanCacheSize() {
return config.getInt(SCAN_CACHE_SIZE_ATTRIB, DEFAULT_SCAN_CACHE_SIZE);
}
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/PhoenixRuntime.java b/phoenix-core/src/main/java/org/apache/phoenix/util/PhoenixRuntime.java
index f07dd29..e852506 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/util/PhoenixRuntime.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/PhoenixRuntime.java
@@ -44,6 +44,9 @@ import java.util.TreeSet;
import javax.annotation.Nullable;
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.phoenix.monitoring.PhoenixTableMetric;
+import org.apache.phoenix.monitoring.TableMetricsManager;
import org.apache.phoenix.thirdparty.org.apache.commons.cli.CommandLine;
import org.apache.phoenix.thirdparty.org.apache.commons.cli.CommandLineParser;
import org.apache.phoenix.thirdparty.org.apache.commons.cli.DefaultParser;
@@ -1368,6 +1371,23 @@ public class PhoenixRuntime {
public static Collection<GlobalMetric> getGlobalPhoenixClientMetrics() {
return GlobalClientMetrics.getMetrics();
}
+
+ /**
+ * This function will be called mainly in Metric Publisher methods.
+ * Its the only way Metric publisher will be able to access the metrics in phoenix system.
+ * @return map of TableName to List of GlobalMetric's.
+ */
+ public static Map<String,List<PhoenixTableMetric>> getPhoenixTableClientMetrics() {
+ return TableMetricsManager.getTableMetricsMethod();
+ }
+
+ /**
+ * This is only used in testcases to reset the tableLevel Metrics data
+ */
+ @VisibleForTesting
+ public static void clearTableLevelMetrics(){
+ TableMetricsManager.clearTableLevelMetricsMethod();
+ }
/**
*
diff --git a/phoenix-core/src/main/resources/META-INF/services/org.apache.phoenix.monitoring.MetricPublisherSupplierFactory b/phoenix-core/src/main/resources/META-INF/services/org.apache.phoenix.monitoring.MetricPublisherSupplierFactory
new file mode 100644
index 0000000..6113544
--- /dev/null
+++ b/phoenix-core/src/main/resources/META-INF/services/org.apache.phoenix.monitoring.MetricPublisherSupplierFactory
@@ -0,0 +1,16 @@
+# 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.
+org.apache.phoenix.monitoring.JmxMetricProvider
\ No newline at end of file
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/monitoring/TableClientMetricsTest.java b/phoenix-core/src/test/java/org/apache/phoenix/monitoring/TableClientMetricsTest.java
new file mode 100644
index 0000000..20ceccf
--- /dev/null
+++ b/phoenix-core/src/test/java/org/apache/phoenix/monitoring/TableClientMetricsTest.java
@@ -0,0 +1,214 @@
+/**
+ * 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.List;
+import org.apache.hadoop.conf.Configuration;
+import org.junit.Test;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.apache.phoenix.monitoring.MetricType.COUNT_ROWS_SCANNED;
+import static org.apache.phoenix.monitoring.MetricType.DELETE_MUTATION_BYTES;
+import static org.apache.phoenix.monitoring.MetricType.DELETE_MUTATION_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.MUTATION_BATCH_SIZE;
+import static org.apache.phoenix.monitoring.MetricType.MUTATION_COMMIT_TIME;
+import static org.apache.phoenix.monitoring.MetricType.MUTATION_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.QUERY_FAILED_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.QUERY_TIMEOUT_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.SCAN_BYTES;
+import static org.apache.phoenix.monitoring.MetricType.SELECT_POINTLOOKUP_FAILED_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.SELECT_POINTLOOKUP_SUCCESS_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.SELECT_SCAN_FAILED_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.SELECT_SCAN_SUCCESS_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.SELECT_SQL_QUERY_TIME;
+import static org.apache.phoenix.monitoring.MetricType.TASK_END_TO_END_TIME;
+import static org.apache.phoenix.monitoring.MetricType.UPSERT_MUTATION_BYTES;
+import static org.apache.phoenix.monitoring.MetricType.UPSERT_MUTATION_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.TableLevelMetricsTestData.countRowsScannedCounter;
+import static org.apache.phoenix.monitoring.TableLevelMetricsTestData.deleteMutationByesCounter;
+import static org.apache.phoenix.monitoring.TableLevelMetricsTestData.deleteMutationSqlCounter;
+import static org.apache.phoenix.monitoring.TableLevelMetricsTestData.mutationBatchSizeCounter;
+import static org.apache.phoenix.monitoring.TableLevelMetricsTestData.mutationSqlCommitTimeCounter;
+import static org.apache.phoenix.monitoring.TableLevelMetricsTestData.mutationSqlCounter;
+import static org.apache.phoenix.monitoring.TableLevelMetricsTestData.queryFailedCounter;
+import static org.apache.phoenix.monitoring.TableLevelMetricsTestData.queryTimeOutCounter;
+import static org.apache.phoenix.monitoring.TableLevelMetricsTestData.scanBytesCounter;
+import static org.apache.phoenix.monitoring.TableLevelMetricsTestData.selectPointLookUpFailedCounter;
+import static org.apache.phoenix.monitoring.TableLevelMetricsTestData.selectPointLookUpSuccessCounter;
+import static org.apache.phoenix.monitoring.TableLevelMetricsTestData.selectScanFailedCounter;
+import static org.apache.phoenix.monitoring.TableLevelMetricsTestData.selectScanSuccessCounter;
+import static org.apache.phoenix.monitoring.TableLevelMetricsTestData.selectSqlQueryTimeCounter;
+import static org.apache.phoenix.monitoring.TableLevelMetricsTestData.taskEndToEndTimeCounter;
+import static org.apache.phoenix.monitoring.TableLevelMetricsTestData.upsertMutationBytesCounter;
+import static org.apache.phoenix.monitoring.TableLevelMetricsTestData.upsertMutationSqlCounter;
+import static org.apache.phoenix.monitoring.TableLevelMetricsTestData.tableNames;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * This test does UT for TableClientMetrics class
+ * This class has following API
+ * 1. changeMetricValue
+ * 2.getTableName
+ * 3.getMetricMap
+ */
+public class TableClientMetricsTest {
+
+ private static Map<String, TableClientMetrics> tableMetricsSet = new HashMap<>();
+
+ public void verifyMetricsFromTableClientMetrics() {
+ assertFalse(tableMetricsSet.isEmpty());
+ for (int i = 0; i < tableNames.length; i++) {
+ TableClientMetrics instance = tableMetricsSet.get(tableNames[i]);
+ assertEquals(instance.getTableName(), tableNames[i]);
+ List<PhoenixTableMetric> metricList = instance.getMetricMap();
+ for (PhoenixTableMetric metric : metricList) {
+
+ if (metric.getMetricType().equals(MUTATION_BATCH_SIZE)) {
+ assertEquals(mutationBatchSizeCounter[i], metric.getValue());
+ }
+ if (metric.getMetricType().equals(UPSERT_MUTATION_BYTES)) {
+ assertEquals(upsertMutationBytesCounter[i], metric.getValue());
+ }
+ if (metric.getMetricType().equals(UPSERT_MUTATION_SQL_COUNTER)) {
+ assertEquals(upsertMutationSqlCounter[i], metric.getValue());
+ }
+ if (metric.getMetricType().equals(DELETE_MUTATION_BYTES)) {
+ assertEquals(deleteMutationByesCounter[i], metric.getValue());
+ }
+ if (metric.getMetricType().equals(DELETE_MUTATION_SQL_COUNTER)) {
+ assertEquals(deleteMutationSqlCounter[i], metric.getValue());
+ }
+ if (metric.getMetricType().equals(MUTATION_SQL_COUNTER)) {
+ assertEquals(mutationSqlCounter[i], metric.getValue());
+ }
+ if (metric.getMetricType().equals(MUTATION_COMMIT_TIME)) {
+ assertEquals(mutationSqlCommitTimeCounter[i], metric.getValue());
+ }
+ if (metric.getMetricType().equals(TASK_END_TO_END_TIME)) {
+ assertEquals(taskEndToEndTimeCounter[i], metric.getValue());
+ }
+ if (metric.getMetricType().equals(COUNT_ROWS_SCANNED)) {
+ assertEquals(countRowsScannedCounter[i], metric.getValue());
+ }
+ if (metric.getMetricType().equals(QUERY_FAILED_COUNTER)) {
+ assertEquals(queryFailedCounter[i], metric.getValue());
+ }
+ if (metric.getMetricType().equals(QUERY_TIMEOUT_COUNTER)) {
+ assertEquals(queryTimeOutCounter[i], metric.getValue());
+ }
+ if (metric.getMetricType().equals(SCAN_BYTES)) {
+ assertEquals(scanBytesCounter[i], metric.getValue());
+ }
+ if (metric.getMetricType().equals(SELECT_POINTLOOKUP_SUCCESS_SQL_COUNTER)) {
+ assertEquals(selectPointLookUpSuccessCounter[i], metric.getValue());
+ }
+ if (metric.getMetricType().equals(SELECT_POINTLOOKUP_FAILED_SQL_COUNTER)) {
+ assertEquals(selectPointLookUpFailedCounter[i], metric.getValue());
+ }
+ if (metric.getMetricType().equals(SELECT_SQL_QUERY_TIME)) {
+ assertEquals(selectSqlQueryTimeCounter[i], metric.getValue());
+ }
+ if (metric.getMetricType().equals(SELECT_SCAN_FAILED_SQL_COUNTER)) {
+ assertEquals(selectScanFailedCounter[i], metric.getValue());
+ }
+ if (metric.getMetricType().equals(SELECT_SCAN_SUCCESS_SQL_COUNTER)) {
+ assertEquals(selectScanSuccessCounter[i], metric.getValue());
+ }
+ }
+ }
+ }
+
+ public boolean verifyTableName() {
+
+ if (tableMetricsSet.isEmpty()) {
+ return false;
+ }
+ for (String tableName : tableNames) {
+ TableClientMetrics instance = tableMetricsSet.get(tableName);
+ if (!instance.getTableName().equals(tableName)) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+
+ /**
+ * This test is for changeMetricValue() Method and getMetricMap()
+ */
+ @Test
+ public void testTableClientMetrics() {
+ for (int i = 0; i < tableNames.length; i++) {
+ TableClientMetrics tableClientMetrics = new TableClientMetrics(tableNames[i]);
+ tableMetricsSet.put(tableNames[i], tableClientMetrics);
+
+ tableClientMetrics.changeMetricValue(MUTATION_BATCH_SIZE,
+ mutationBatchSizeCounter[i]);
+ tableClientMetrics.changeMetricValue(UPSERT_MUTATION_BYTES,
+ upsertMutationBytesCounter[i]);
+ tableClientMetrics.changeMetricValue(UPSERT_MUTATION_SQL_COUNTER,
+ upsertMutationSqlCounter[i]);
+ tableClientMetrics.changeMetricValue(DELETE_MUTATION_BYTES,
+ deleteMutationByesCounter[i]);
+ tableClientMetrics.changeMetricValue(DELETE_MUTATION_SQL_COUNTER,
+ deleteMutationSqlCounter[i]);
+ tableClientMetrics.changeMetricValue(MUTATION_SQL_COUNTER,
+ mutationSqlCounter[i]);
+ tableClientMetrics.changeMetricValue(MUTATION_COMMIT_TIME,
+ mutationSqlCommitTimeCounter[i]);
+ tableClientMetrics.changeMetricValue(TASK_END_TO_END_TIME,
+ taskEndToEndTimeCounter[i]);
+ tableClientMetrics.changeMetricValue(COUNT_ROWS_SCANNED,
+ countRowsScannedCounter[i]);
+ tableClientMetrics.changeMetricValue(QUERY_FAILED_COUNTER,
+ queryFailedCounter[i]);
+ tableClientMetrics.changeMetricValue(QUERY_TIMEOUT_COUNTER,
+ queryTimeOutCounter[i]);
+ tableClientMetrics.changeMetricValue(SCAN_BYTES, scanBytesCounter[i]);
+ tableClientMetrics.changeMetricValue(MetricType.SELECT_POINTLOOKUP_SUCCESS_SQL_COUNTER,
+ selectPointLookUpSuccessCounter[i]);
+ tableClientMetrics.changeMetricValue(SELECT_POINTLOOKUP_FAILED_SQL_COUNTER,
+ selectPointLookUpFailedCounter[i]);
+ tableClientMetrics.changeMetricValue(SELECT_SQL_QUERY_TIME,
+ selectSqlQueryTimeCounter[i]);
+ tableClientMetrics.changeMetricValue(SELECT_SCAN_SUCCESS_SQL_COUNTER,
+ selectScanSuccessCounter[i]);
+ tableClientMetrics.changeMetricValue(SELECT_SCAN_FAILED_SQL_COUNTER,
+ selectScanFailedCounter[i]);
+ }
+ verifyMetricsFromTableClientMetrics();
+ tableMetricsSet.clear();
+ }
+
+ /**
+ * This test is for getTableName()
+ */
+ @Test
+ public void testTableClientMetricsforTableName() {
+ Configuration conf = new Configuration();
+ for (int i = 0; i < tableNames.length; i++) {
+ TableClientMetrics tableClientMetrics = new TableClientMetrics(tableNames[i]);
+ tableMetricsSet.put(tableNames[i], tableClientMetrics);
+ }
+ assertTrue(verifyTableName());
+ }
+}
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/monitoring/TableLevelMetricsTestData.java b/phoenix-core/src/test/java/org/apache/phoenix/monitoring/TableLevelMetricsTestData.java
new file mode 100644
index 0000000..4310f6d
--- /dev/null
+++ b/phoenix-core/src/test/java/org/apache/phoenix/monitoring/TableLevelMetricsTestData.java
@@ -0,0 +1,165 @@
+/**
+ * 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.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.phoenix.monitoring.MetricType.COUNT_ROWS_SCANNED;
+import static org.apache.phoenix.monitoring.MetricType.DELETE_MUTATION_BYTES;
+import static org.apache.phoenix.monitoring.MetricType.DELETE_MUTATION_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.MUTATION_BATCH_SIZE;
+import static org.apache.phoenix.monitoring.MetricType.MUTATION_COMMIT_TIME;
+import static org.apache.phoenix.monitoring.MetricType.MUTATION_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.QUERY_FAILED_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.QUERY_TIMEOUT_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.SCAN_BYTES;
+import static org.apache.phoenix.monitoring.MetricType.SELECT_POINTLOOKUP_FAILED_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.SELECT_POINTLOOKUP_SUCCESS_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.SELECT_SCAN_FAILED_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.SELECT_SCAN_SUCCESS_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.SELECT_SQL_QUERY_TIME;
+import static org.apache.phoenix.monitoring.MetricType.TASK_END_TO_END_TIME;
+import static org.apache.phoenix.monitoring.MetricType.UPSERT_MUTATION_BYTES;
+import static org.apache.phoenix.monitoring.MetricType.UPSERT_MUTATION_SQL_COUNTER;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * This class is used primarily to populate data and
+ * verification methods
+ */
+
+public class TableLevelMetricsTestData {
+
+ public static final String[] tableNames = { "T0001", "T0002", "T0003" };
+ public static Map<String, Map<MetricType, Long>>[] tableMetricsMap = new Map[tableNames.length];
+ public static final long[] mutationBatchSizeCounter = { 90, 100, 150 };
+ public static final long[] upsertMutationBytesCounter = { 100, 200, 250 };
+ public static final long[] upsertMutationSqlCounter = { 100, 200, 300 };
+ public static final long[] deleteMutationByesCounter = { 100, 200, 150 };
+ public static final long[] deleteMutationSqlCounter = { 100, 200, 140 };
+ public static final long[] mutationSqlCounter = { 200, 400, 600 };
+ public static final long[] mutationSqlCommitTimeCounter = { 150, 300, 100 };
+ public static final long[] taskEndToEndTimeCounter = { 10, 20, 30 };
+ public static final long[] countRowsScannedCounter = { 500, 600, 400 };
+ public static final long[] queryFailedCounter = { 10, 20, 30 };
+ public static final long[] queryTimeOutCounter = { 30, 40, 40 };
+ public static final long[] scanBytesCounter = { 500, 600, 400 };
+ public static final long[] selectPointLookUpFailedCounter = { 10, 29, 49 };
+ public static final long[] selectSqlQueryTimeCounter = { 30, 40, 55 };
+ public static final long[] selectPointLookUpSuccessCounter = { 10, 20, 55 };
+ public static final long[] selectScanSuccessCounter = { 200000, 300000, 4444 };
+ public static final long[] selectScanFailedCounter = { 1000000, 20000000, 3455 };
+
+ public static void populateMetrics() {
+ for (int i = 0; i < tableMetricsMap.length; i++) {
+ tableMetricsMap[i] = new HashMap<>();
+ }
+ for (int i = 0; i < tableNames.length; i++) {
+ Map<MetricType, Long> metrics = new HashMap<>();
+ metrics.put(MUTATION_BATCH_SIZE, mutationBatchSizeCounter[i]);
+ metrics.put(UPSERT_MUTATION_BYTES, upsertMutationBytesCounter[i]);
+ metrics.put(UPSERT_MUTATION_SQL_COUNTER, upsertMutationSqlCounter[i]);
+ metrics.put(DELETE_MUTATION_BYTES, deleteMutationByesCounter[i]);
+ metrics.put(DELETE_MUTATION_SQL_COUNTER, deleteMutationSqlCounter[i]);
+ metrics.put(MUTATION_SQL_COUNTER, mutationSqlCounter[i]);
+ metrics.put(MUTATION_COMMIT_TIME, mutationSqlCommitTimeCounter[i]);
+ metrics.put(TASK_END_TO_END_TIME, taskEndToEndTimeCounter[i]);
+ metrics.put(COUNT_ROWS_SCANNED, countRowsScannedCounter[i]);
+ metrics.put(QUERY_FAILED_COUNTER, queryFailedCounter[i]);
+ metrics.put(QUERY_TIMEOUT_COUNTER, queryTimeOutCounter[i]);
+ metrics.put(SCAN_BYTES, scanBytesCounter[i]);
+ metrics.put(MetricType.SELECT_POINTLOOKUP_SUCCESS_SQL_COUNTER,
+ selectPointLookUpSuccessCounter[i]);
+ metrics.put(SELECT_POINTLOOKUP_FAILED_SQL_COUNTER,
+ selectPointLookUpFailedCounter[i]);
+ metrics.put(MetricType.SELECT_SQL_QUERY_TIME, selectSqlQueryTimeCounter[i]);
+ metrics.put(SELECT_SCAN_SUCCESS_SQL_COUNTER, selectScanSuccessCounter[i]);
+ metrics.put(MetricType.SELECT_SCAN_FAILED_SQL_COUNTER, selectScanFailedCounter[i]);
+ tableMetricsMap[i].put(tableNames[i], metrics);
+ }
+ }
+
+ public void verifyMetricsInjection(int noOfTables) {
+ Map<String, List<PhoenixTableMetric>> map = TableMetricsManager.getTableMetricsMethod();
+ assertFalse(map == null || map.isEmpty());
+ for (int i = 0; i < noOfTables; i++) {
+ System.out.println("CURRENTLY ON: " + tableNames[i]);
+ assertTrue(map.containsKey(tableNames[i]));
+ List<PhoenixTableMetric> tableMetric = map.get(tableNames[i]);
+ for (PhoenixTableMetric metric : tableMetric) {
+ if (metric.getMetricType().equals(MUTATION_BATCH_SIZE)) {
+ assertEquals(mutationBatchSizeCounter[i], metric.getValue());
+ }
+ if (metric.getMetricType().equals(UPSERT_MUTATION_BYTES)) {
+ assertEquals(upsertMutationBytesCounter[i], metric.getValue());
+ }
+ if (metric.getMetricType().equals(UPSERT_MUTATION_SQL_COUNTER)) {
+ assertEquals(upsertMutationSqlCounter[i], metric.getValue());
+ }
+ if (metric.getMetricType().equals(DELETE_MUTATION_BYTES)) {
+ assertEquals(deleteMutationByesCounter[i], metric.getValue());
+ }
+ if (metric.getMetricType().equals(DELETE_MUTATION_SQL_COUNTER)) {
+ assertEquals(deleteMutationSqlCounter[i], metric.getValue());
+ }
+ if (metric.getMetricType().equals(MUTATION_SQL_COUNTER)) {
+ assertEquals(mutationSqlCounter[i], metric.getValue());
+ }
+ if (metric.getMetricType().equals(MUTATION_COMMIT_TIME)) {
+ assertEquals(mutationSqlCommitTimeCounter[i], metric.getValue());
+ }
+ if (metric.getMetricType().equals(TASK_END_TO_END_TIME)) {
+ assertEquals(taskEndToEndTimeCounter[i], metric.getValue());
+ }
+ if (metric.getMetricType().equals(COUNT_ROWS_SCANNED)) {
+ assertEquals(countRowsScannedCounter[i], metric.getValue());
+ }
+ if (metric.getMetricType().equals(QUERY_FAILED_COUNTER)) {
+ assertEquals(queryFailedCounter[i], metric.getValue());
+ }
+ if (metric.getMetricType().equals(QUERY_TIMEOUT_COUNTER)) {
+ assertEquals(queryTimeOutCounter[i], metric.getValue());
+ }
+ if (metric.getMetricType().equals(SCAN_BYTES)) {
+ assertEquals(scanBytesCounter[i], metric.getValue());
+ }
+ if (metric.getMetricType().equals(SELECT_POINTLOOKUP_SUCCESS_SQL_COUNTER)) {
+ assertEquals(selectPointLookUpSuccessCounter[i], metric.getValue());
+ }
+ if (metric.getMetricType().equals(SELECT_POINTLOOKUP_FAILED_SQL_COUNTER)) {
+ assertEquals(selectPointLookUpFailedCounter[i], metric.getValue());
+ }
+ if (metric.getMetricType().equals(SELECT_SQL_QUERY_TIME)) {
+ assertEquals(selectSqlQueryTimeCounter[i], metric.getValue());
+ }
+ if (metric.getMetricType().equals(SELECT_SCAN_FAILED_SQL_COUNTER)) {
+ assertEquals(selectScanFailedCounter[i], metric.getValue());
+ }
+ if (metric.getMetricType().equals(SELECT_SCAN_SUCCESS_SQL_COUNTER)) {
+ assertEquals(selectScanSuccessCounter[i], metric.getValue());
+ }
+ }
+ }
+ }
+
+}
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/monitoring/TableMetricsManagerTest.java b/phoenix-core/src/test/java/org/apache/phoenix/monitoring/TableMetricsManagerTest.java
new file mode 100644
index 0000000..d85c805
--- /dev/null
+++ b/phoenix-core/src/test/java/org/apache/phoenix/monitoring/TableMetricsManagerTest.java
@@ -0,0 +1,207 @@
+/**
+ * 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 com.google.common.collect.Lists;
+import org.apache.phoenix.query.QueryServicesOptions;
+import org.junit.Test;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+
+import static org.apache.phoenix.monitoring.TableLevelMetricsTestData.countRowsScannedCounter;
+import static org.apache.phoenix.monitoring.TableLevelMetricsTestData.deleteMutationByesCounter;
+import static org.apache.phoenix.monitoring.TableLevelMetricsTestData.deleteMutationSqlCounter;
+import static org.apache.phoenix.monitoring.TableLevelMetricsTestData.mutationBatchSizeCounter;
+import static org.apache.phoenix.monitoring.TableLevelMetricsTestData.mutationSqlCommitTimeCounter;
+import static org.apache.phoenix.monitoring.TableLevelMetricsTestData.mutationSqlCounter;
+import static org.apache.phoenix.monitoring.TableLevelMetricsTestData.queryFailedCounter;
+import static org.apache.phoenix.monitoring.TableLevelMetricsTestData.queryTimeOutCounter;
+import static org.apache.phoenix.monitoring.TableLevelMetricsTestData.scanBytesCounter;
+import static org.apache.phoenix.monitoring.TableLevelMetricsTestData.selectPointLookUpFailedCounter;
+import static org.apache.phoenix.monitoring.TableLevelMetricsTestData.selectPointLookUpSuccessCounter;
+import static org.apache.phoenix.monitoring.TableLevelMetricsTestData.selectScanFailedCounter;
+import static org.apache.phoenix.monitoring.TableLevelMetricsTestData.selectScanSuccessCounter;
+import static org.apache.phoenix.monitoring.TableLevelMetricsTestData.selectSqlQueryTimeCounter;
+import static org.apache.phoenix.monitoring.TableLevelMetricsTestData.tableNames;
+import static org.apache.phoenix.monitoring.TableLevelMetricsTestData.tableMetricsMap;
+import static org.apache.phoenix.monitoring.TableLevelMetricsTestData.populateMetrics;
+import static org.apache.phoenix.monitoring.TableLevelMetricsTestData.taskEndToEndTimeCounter;
+import static org.apache.phoenix.monitoring.TableLevelMetricsTestData.upsertMutationBytesCounter;
+import static org.apache.phoenix.monitoring.TableLevelMetricsTestData.upsertMutationSqlCounter;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * This method is for UT TableMetricManager class.
+ * This class exposes 4 static functions
+ * 1.pushMetricsFromConnInstanceMethod
+ * 2.getTableMetricsMethod
+ * 3.clearTableLevelMetricsMethod
+ * 4.updateMetricsMethod
+ */
+public class TableMetricsManagerTest {
+
+ public boolean verifyMetricsReset(){
+ Map<String, List<PhoenixTableMetric>>map = TableMetricsManager.getTableMetricsMethod();
+ return map != null && map.isEmpty();
+ }
+
+ private static class PushMetrics implements Runnable {
+ private final Map<String, Map<MetricType, Long>> map;
+
+ public PushMetrics(Map<String, Map<MetricType, Long>> map) {
+ this.map = map;
+ }
+
+ @Override public void run() {
+ TableMetricsManager.pushMetricsFromConnInstanceMethod(map);
+ }
+ }
+
+ public boolean verifyTableNamesExists(String tableName){
+ Map<String,List<PhoenixTableMetric>>map = TableMetricsManager.getTableMetricsMethod();
+ return map != null && map.containsKey(tableName);
+ }
+
+ /**
+ * Injecting data parallely to TableMetricsManager using pushMetricsFromConnInstanceMethod()
+ */
+ @Test
+ public void testVerifyTableLevelMetricsMutilpleThreads() throws Exception {
+ QueryServicesOptions options = QueryServicesOptions.withDefaults();
+ options.setTableLevelMetricsEnabled();
+ String tableNamesList = tableNames[0] + "," + tableNames[1] + "," + tableNames[2];
+ options.setAllowedListForTableLevelMetrics(tableNamesList);
+ TableMetricsManager tableMetricsManager = new TableMetricsManager(options);
+ TableMetricsManager.setInstance(tableMetricsManager);
+ TableLevelMetricsTestData testData = new TableLevelMetricsTestData();
+ populateMetrics();
+
+ ExecutorService executorService =
+ Executors.newFixedThreadPool(tableNames.length, new ThreadFactory() {
+ @Override public Thread newThread(Runnable r) {
+ Thread t = Executors.defaultThreadFactory().newThread(r);
+ t.setDaemon(true);
+ t.setPriority(Thread.MIN_PRIORITY);
+ return t;
+ }
+ });
+ List<Future<?>> futureList = Lists.newArrayListWithExpectedSize(tableNames.length);
+ for (int i = 0; i < tableNames.length; ++i) {
+ futureList.add(executorService.submit(new PushMetrics(tableMetricsMap[i])));
+ }
+ executorService.shutdown();
+ executorService.awaitTermination(10, TimeUnit.SECONDS);
+ testData.verifyMetricsInjection(tableNames.length);
+ TableMetricsManager.clearTableLevelMetricsMethod();
+ assertTrue(verifyMetricsReset());
+ }
+
+ /**
+ * test for pushMetricsFromConnInstanceMethod() , getTableMetricsMethod()
+ * and clearTableLevelMetrics();
+ */
+ @Test
+ public void testTableMetricsForPushMetricsFromConnInstanceMethod() {
+ QueryServicesOptions options = QueryServicesOptions.withDefaults();
+ options.setTableLevelMetricsEnabled();
+ String tableNamesList = tableNames[0] + "," + tableNames[1] + "," + tableNames[2];
+ options.setAllowedListForTableLevelMetrics(tableNamesList);
+ TableMetricsManager tableMetricsManager = new TableMetricsManager(options);
+ TableMetricsManager.setInstance(tableMetricsManager);
+
+ TableLevelMetricsTestData testData = new TableLevelMetricsTestData();
+ populateMetrics();
+ for(int i = 0; i < tableNames.length ; i++){
+ TableMetricsManager.pushMetricsFromConnInstanceMethod(tableMetricsMap[i]);
+ }
+ testData.verifyMetricsInjection(3);
+ TableMetricsManager.clearTableLevelMetricsMethod();
+ assertTrue(verifyMetricsReset());
+ }
+
+ /**
+ * test for updateMetricsMethod() , getTableMetricsMethod()
+ * and clearTableLevelMetrics();
+ */
+ @Test
+ public void testTableMetricsForUpdateMetricsMethod() {
+
+ QueryServicesOptions options = QueryServicesOptions.withDefaults();
+ options.setTableLevelMetricsEnabled();
+ TableMetricsManager tableMetricsManager = new TableMetricsManager(options);
+ TableMetricsManager.setInstance(tableMetricsManager);
+
+ TableLevelMetricsTestData testData = new TableLevelMetricsTestData();
+ for(int i = 0; i < tableNames.length; i++) {
+ TableMetricsManager.updateMetricsMethod(tableNames[i],MetricType.MUTATION_BATCH_SIZE, mutationBatchSizeCounter[i]);
+ TableMetricsManager.updateMetricsMethod(tableNames[i],MetricType.UPSERT_MUTATION_SQL_COUNTER, upsertMutationSqlCounter[i]);
+ TableMetricsManager.updateMetricsMethod(tableNames[i],MetricType.UPSERT_MUTATION_BYTES, upsertMutationBytesCounter[i]);
+ TableMetricsManager.updateMetricsMethod(tableNames[i],MetricType.DELETE_MUTATION_SQL_COUNTER, deleteMutationSqlCounter[i]);
+ TableMetricsManager.updateMetricsMethod(tableNames[i],MetricType.DELETE_MUTATION_BYTES, deleteMutationByesCounter[i]);
+ TableMetricsManager.updateMetricsMethod(tableNames[i],MetricType.MUTATION_SQL_COUNTER, mutationSqlCounter[i]);
+ TableMetricsManager.updateMetricsMethod(tableNames[i],MetricType.MUTATION_COMMIT_TIME, mutationSqlCommitTimeCounter[i]);
+ TableMetricsManager.updateMetricsMethod(tableNames[i],MetricType.TASK_END_TO_END_TIME, taskEndToEndTimeCounter[i]);
+ TableMetricsManager.updateMetricsMethod(tableNames[i],MetricType.COUNT_ROWS_SCANNED, countRowsScannedCounter[i]);
+ TableMetricsManager.updateMetricsMethod(tableNames[i],MetricType.QUERY_FAILED_COUNTER, queryFailedCounter[i]);
+ TableMetricsManager.updateMetricsMethod(tableNames[i],MetricType.QUERY_TIMEOUT_COUNTER, queryTimeOutCounter[i]);
+ TableMetricsManager.updateMetricsMethod(tableNames[i],MetricType.SCAN_BYTES, scanBytesCounter[i]);
+ TableMetricsManager.updateMetricsMethod(tableNames[i],MetricType.SELECT_POINTLOOKUP_SUCCESS_SQL_COUNTER, selectPointLookUpSuccessCounter[i]);
+ TableMetricsManager.updateMetricsMethod(tableNames[i],MetricType.SELECT_POINTLOOKUP_FAILED_SQL_COUNTER, selectPointLookUpFailedCounter[i]);
+ TableMetricsManager.updateMetricsMethod(tableNames[i],MetricType.SELECT_SQL_QUERY_TIME, selectSqlQueryTimeCounter[i]);
+ TableMetricsManager.updateMetricsMethod(tableNames[i],MetricType.SELECT_SCAN_FAILED_SQL_COUNTER, selectScanFailedCounter[i]);
+ TableMetricsManager.updateMetricsMethod(tableNames[i],MetricType.SELECT_SCAN_SUCCESS_SQL_COUNTER, selectScanSuccessCounter[i]);
+
+ }
+ testData.verifyMetricsInjection(3);
+ TableMetricsManager.clearTableLevelMetricsMethod();
+ assertTrue(verifyMetricsReset());
+ }
+
+
+
+ //This test puts T0001, T0002 in allowed list and verifies the existence of metrics
+ //and blocking the T0003.
+ @Test
+ public void testTableMetricsForPushMetricsFromConnInstanceMethodWithAllowedTables() {
+ QueryServicesOptions options = QueryServicesOptions.withDefaults();
+ options.setTableLevelMetricsEnabled();
+ String tableNamesList = tableNames[0] + "," + tableNames[1];
+ options.setAllowedListForTableLevelMetrics(tableNamesList);
+
+ TableMetricsManager tableMetricsManager = new TableMetricsManager(options);
+ TableMetricsManager.setInstance(tableMetricsManager);
+
+ TableLevelMetricsTestData testData = new TableLevelMetricsTestData();
+ populateMetrics();
+ for(int i = 0; i < tableNames.length ; i++){
+ TableMetricsManager.pushMetricsFromConnInstanceMethod(tableMetricsMap[i]);
+ }
+
+ testData.verifyMetricsInjection(2);
+ assertFalse(verifyTableNamesExists(tableNames[2]));
+ }
+
+}
\ No newline at end of file