Source code

001/*
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018package org.apache.hadoop.hbase.client;
019
020import static org.junit.jupiter.api.Assertions.assertEquals;
021import static org.junit.jupiter.api.Assertions.assertNotNull;
022import static org.junit.jupiter.api.Assertions.assertNull;
023
024import java.util.ArrayList;
025import java.util.Arrays;
026import java.util.List;
027import java.util.concurrent.CompletableFuture;
028import org.apache.hadoop.hbase.HBaseTestingUtil;
029import org.apache.hadoop.hbase.TableName;
030import org.apache.hadoop.hbase.regionserver.MetricsRegionServer;
031import org.apache.hadoop.hbase.regionserver.MetricsRegionServerSource;
032import org.apache.hadoop.hbase.regionserver.MetricsRegionServerSourceImpl;
033import org.apache.hadoop.hbase.testclassification.ClientTests;
034import org.apache.hadoop.hbase.testclassification.MediumTests;
035import org.apache.hadoop.hbase.util.Bytes;
036import org.apache.hadoop.hbase.util.JVMClusterUtil;
037import org.junit.jupiter.api.AfterAll;
038import org.junit.jupiter.api.BeforeAll;
039import org.junit.jupiter.api.Tag;
040import org.junit.jupiter.api.Test;
041
042import org.apache.hbase.thirdparty.com.google.common.io.Closeables;
043
044@Tag(MediumTests.TAG)
045@Tag(ClientTests.TAG)
046public class TestAsyncTableQueryMetrics {
047
048  private static final HBaseTestingUtil UTIL = new HBaseTestingUtil();
049
050  private static final TableName TABLE_NAME = TableName.valueOf("ResultMetrics");
051
052  private static final byte[] CF = Bytes.toBytes("cf");
053
054  private static final byte[] CQ = Bytes.toBytes("cq");
055
056  private static final byte[] VALUE = Bytes.toBytes("value");
057
058  private static final byte[] ROW_1 = Bytes.toBytes("zzz1");
059  private static final byte[] ROW_2 = Bytes.toBytes("zzz2");
060  private static final byte[] ROW_3 = Bytes.toBytes("zzz3");
061
062  private static AsyncConnection CONN;
063
064  @BeforeAll
065  public static void setUp() throws Exception {
066    UTIL.startMiniCluster(3);
067    // Create 3 rows in the table, with rowkeys starting with "zzz*" so that
068    // scan are forced to hit all the regions.
069    try (Table table = UTIL.createMultiRegionTable(TABLE_NAME, CF)) {
070      table.put(Arrays.asList(new Put(ROW_1).addColumn(CF, CQ, VALUE),
071        new Put(ROW_2).addColumn(CF, CQ, VALUE), new Put(ROW_3).addColumn(CF, CQ, VALUE)));
072    }
073    CONN = ConnectionFactory.createAsyncConnection(UTIL.getConfiguration()).get();
074    CONN.getAdmin().flush(TABLE_NAME).join();
075  }
076
077  @AfterAll
078  public static void tearDown() throws Exception {
079    Closeables.close(CONN, true);
080    UTIL.shutdownMiniCluster();
081  }
082
083  @Test
084  public void itTestsGets() throws Exception {
085    // Test a single Get
086    Get g1 = new Get(ROW_1);
087    g1.setQueryMetricsEnabled(true);
088
089    long bbs = getClusterBlockBytesScanned();
090    Result result = CONN.getTable(TABLE_NAME).get(g1).get();
091    bbs += result.getMetrics().getBlockBytesScanned();
092    assertNotNull(result.getMetrics());
093    assertEquals(getClusterBlockBytesScanned(), bbs);
094
095    // Test multigets
096    Get g2 = new Get(ROW_2);
097    g2.setQueryMetricsEnabled(true);
098
099    Get g3 = new Get(ROW_3);
100    g3.setQueryMetricsEnabled(true);
101
102    List<CompletableFuture<Result>> futures = CONN.getTable(TABLE_NAME).get(List.of(g1, g2, g3));
103
104    for (CompletableFuture<Result> future : futures) {
105      result = future.join();
106      assertNotNull(result.getMetrics());
107      bbs += result.getMetrics().getBlockBytesScanned();
108    }
109
110    assertEquals(getClusterBlockBytesScanned(), bbs);
111
112    g1.setCheckExistenceOnly(true);
113    g2.setCheckExistenceOnly(true);
114    g3.setCheckExistenceOnly(true);
115
116    result = CONN.getTable(TABLE_NAME).get(g1).get();
117    assertEquals(Boolean.TRUE, result.getExists());
118    assertNotNull(result.getMetrics());
119    bbs += result.getMetrics().getBlockBytesScanned();
120    assertEquals(getClusterBlockBytesScanned(), bbs);
121
122    futures = CONN.getTable(TABLE_NAME).get(List.of(g1, g2, g3));
123    for (CompletableFuture<Result> future : futures) {
124      result = future.join();
125      assertEquals(Boolean.TRUE, result.getExists());
126      assertNotNull(result.getMetrics());
127      bbs += result.getMetrics().getBlockBytesScanned();
128    }
129    assertEquals(getClusterBlockBytesScanned(), bbs);
130  }
131
132  @Test
133  public void itTestsDefaultGetNoMetrics() throws Exception {
134    // Test a single Get
135    Get g1 = new Get(ROW_1);
136
137    Result result = CONN.getTable(TABLE_NAME).get(g1).get();
138    assertNull(result.getMetrics());
139
140    // Test multigets
141    Get g2 = new Get(ROW_2);
142    Get g3 = new Get(ROW_3);
143    List<CompletableFuture<Result>> futures = CONN.getTable(TABLE_NAME).get(List.of(g1, g2, g3));
144    futures.forEach(f -> assertNull(f.join().getMetrics()));
145
146  }
147
148  @Test
149  public void itTestsScans() {
150    Scan scan = new Scan();
151    scan.setQueryMetricsEnabled(true);
152
153    long bbs = getClusterBlockBytesScanned();
154    try (ResultScanner scanner = CONN.getTable(TABLE_NAME).getScanner(scan)) {
155      for (Result result : scanner) {
156        assertNotNull(result.getMetrics());
157        bbs += result.getMetrics().getBlockBytesScanned();
158        assertEquals(getClusterBlockBytesScanned(), bbs);
159      }
160    }
161  }
162
163  @Test
164  public void itTestsDefaultScanNoMetrics() {
165    Scan scan = new Scan();
166
167    try (ResultScanner scanner = CONN.getTable(TABLE_NAME).getScanner(scan)) {
168      for (Result result : scanner) {
169        assertNull(result.getMetrics());
170      }
171    }
172  }
173
174  @Test
175  public void itTestsAtomicOperations() {
176    CheckAndMutate cam = CheckAndMutate.newBuilder(ROW_1).ifEquals(CF, CQ, VALUE)
177      .queryMetricsEnabled(true).build(new Put(ROW_1).addColumn(CF, CQ, VALUE));
178
179    long bbs = getClusterBlockBytesScanned();
180    CheckAndMutateResult result = CONN.getTable(TABLE_NAME).checkAndMutate(cam).join();
181    QueryMetrics metrics = result.getMetrics();
182
183    assertNotNull(metrics);
184    assertEquals(getClusterBlockBytesScanned(), bbs + metrics.getBlockBytesScanned());
185
186    bbs = getClusterBlockBytesScanned();
187    List<CheckAndMutate> batch = new ArrayList<>();
188    batch.add(cam);
189    batch.add(CheckAndMutate.newBuilder(ROW_2).queryMetricsEnabled(true).ifEquals(CF, CQ, VALUE)
190      .build(new Put(ROW_2).addColumn(CF, CQ, VALUE)));
191    batch.add(CheckAndMutate.newBuilder(ROW_3).queryMetricsEnabled(true).ifEquals(CF, CQ, VALUE)
192      .build(new Put(ROW_3).addColumn(CF, CQ, VALUE)));
193
194    List<Object> res = CONN.getTable(TABLE_NAME).batchAll(batch).join();
195    long totalBbs = res.stream()
196      .mapToLong(r -> ((CheckAndMutateResult) r).getMetrics().getBlockBytesScanned()).sum();
197    assertEquals(getClusterBlockBytesScanned(), bbs + totalBbs);
198
199    bbs = getClusterBlockBytesScanned();
200
201    // flush to force fetch from disk
202    CONN.getAdmin().flush(TABLE_NAME).join();
203    List<CompletableFuture<Object>> futures = CONN.getTable(TABLE_NAME).batch(batch);
204
205    totalBbs = futures.stream().map(CompletableFuture::join)
206      .mapToLong(r -> ((CheckAndMutateResult) r).getMetrics().getBlockBytesScanned()).sum();
207    assertEquals(getClusterBlockBytesScanned(), bbs + totalBbs);
208  }
209
210  @Test
211  public void itTestsDefaultAtomicOperations() {
212    CheckAndMutate cam = CheckAndMutate.newBuilder(ROW_1).ifEquals(CF, CQ, VALUE)
213      .build(new Put(ROW_1).addColumn(CF, CQ, VALUE));
214
215    CheckAndMutateResult result = CONN.getTable(TABLE_NAME).checkAndMutate(cam).join();
216    QueryMetrics metrics = result.getMetrics();
217
218    assertNull(metrics);
219
220    List<CheckAndMutate> batch = new ArrayList<>();
221    batch.add(cam);
222    batch.add(CheckAndMutate.newBuilder(ROW_2).ifEquals(CF, CQ, VALUE)
223      .build(new Put(ROW_2).addColumn(CF, CQ, VALUE)));
224    batch.add(CheckAndMutate.newBuilder(ROW_3).ifEquals(CF, CQ, VALUE)
225      .build(new Put(ROW_3).addColumn(CF, CQ, VALUE)));
226
227    List<Object> res = CONN.getTable(TABLE_NAME).batchAll(batch).join();
228    for (Object r : res) {
229      assertNull(((CheckAndMutateResult) r).getMetrics());
230    }
231
232    // flush to force fetch from disk
233    CONN.getAdmin().flush(TABLE_NAME).join();
234    List<CompletableFuture<Object>> futures = CONN.getTable(TABLE_NAME).batch(batch);
235
236    for (CompletableFuture<Object> future : futures) {
237      Object r = future.join();
238      assertNull(((CheckAndMutateResult) r).getMetrics());
239    }
240  }
241
242  private static long getClusterBlockBytesScanned() {
243    long bbs = 0L;
244
245    for (JVMClusterUtil.RegionServerThread rs : UTIL.getHBaseCluster().getRegionServerThreads()) {
246      MetricsRegionServer metrics = rs.getRegionServer().getMetrics();
247      MetricsRegionServerSourceImpl source =
248        (MetricsRegionServerSourceImpl) metrics.getMetricsSource();
249
250      bbs += source.getMetricsRegistry()
251        .getCounter(MetricsRegionServerSource.BLOCK_BYTES_SCANNED_KEY, 0L).value();
252    }
253
254    return bbs;
255  }
256}