1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18 package org.apache.hadoop.hbase.client;
19
20 import com.google.protobuf.ByteString;
21 import com.yammer.metrics.util.RatioGauge;
22 import org.apache.hadoop.hbase.client.ConnectionManager.HConnectionImplementation;
23 import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
24 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ClientService;
25 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.GetRequest;
26 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ScanRequest;
27 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MultiRequest;
28 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MutateRequest;
29 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MutationProto.MutationType;
30 import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.RegionSpecifier;
31 import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.RegionSpecifier.RegionSpecifierType;
32 import org.apache.hadoop.hbase.testclassification.MetricsTests;
33 import org.apache.hadoop.hbase.testclassification.SmallTests;
34 import org.apache.hadoop.hbase.util.Bytes;
35 import org.junit.AfterClass;
36 import org.junit.Assert;
37 import org.junit.BeforeClass;
38 import org.junit.Test;
39 import org.junit.experimental.categories.Category;
40 import org.mockito.Mockito;
41 import java.io.IOException;
42 import java.util.concurrent.ExecutorService;
43 import java.util.concurrent.Executors;
44 import java.util.concurrent.TimeUnit;
45 import java.util.concurrent.atomic.AtomicBoolean;
46 import static org.junit.Assert.assertEquals;
47
48 @Category({MetricsTests.class, SmallTests.class})
49 public class TestMetricsConnection {
50 private static final ExecutorService BATCH_POOL = Executors.newFixedThreadPool(2);
51 private static MetricsConnection METRICS;
52 private static final AtomicBoolean closed = new AtomicBoolean(false);
53 private static final Runnable RUNNER = new Runnable() {
54 @Override
55 public void run() {
56 try {
57 while (!closed.get() && !Thread.interrupted()) {
58 TimeUnit.MILLISECONDS.sleep(10);
59 }
60 } catch (InterruptedException e) {
61 }
62 }
63 };
64 @BeforeClass
65 public static void beforeClass() {
66 HConnectionImplementation mocked = Mockito.mock(HConnectionImplementation.class);
67 Mockito.when(mocked.toString()).thenReturn("mocked-connection");
68 Mockito.when(mocked.getCurrentBatchPool()).thenReturn(BATCH_POOL);
69 BATCH_POOL.submit(RUNNER);
70 METRICS = new MetricsConnection(mocked);
71 }
72
73 @AfterClass
74 public static void afterClass() throws InterruptedException {
75 METRICS.shutdown();
76 BATCH_POOL.shutdownNow();
77 BATCH_POOL.awaitTermination(Long.MAX_VALUE, TimeUnit.DAYS);
78 }
79
80 @Test
81 public void testStaticMetrics() throws IOException {
82 final byte[] foo = Bytes.toBytes("foo");
83 final RegionSpecifier region = RegionSpecifier.newBuilder()
84 .setValue(ByteString.EMPTY)
85 .setType(RegionSpecifierType.REGION_NAME)
86 .build();
87 final int loop = 5;
88
89 for (int i = 0; i < loop; i++) {
90 METRICS.updateRpc(
91 ClientService.getDescriptor().findMethodByName("Get"),
92 GetRequest.getDefaultInstance(),
93 MetricsConnection.newCallStats());
94 METRICS.updateRpc(
95 ClientService.getDescriptor().findMethodByName("Scan"),
96 ScanRequest.getDefaultInstance(),
97 MetricsConnection.newCallStats());
98 METRICS.updateRpc(
99 ClientService.getDescriptor().findMethodByName("Multi"),
100 MultiRequest.getDefaultInstance(),
101 MetricsConnection.newCallStats());
102 METRICS.updateRpc(
103 ClientService.getDescriptor().findMethodByName("Mutate"),
104 MutateRequest.newBuilder()
105 .setMutation(ProtobufUtil.toMutation(MutationType.APPEND, new Append(foo)))
106 .setRegion(region)
107 .build(),
108 MetricsConnection.newCallStats());
109 METRICS.updateRpc(
110 ClientService.getDescriptor().findMethodByName("Mutate"),
111 MutateRequest.newBuilder()
112 .setMutation(ProtobufUtil.toMutation(MutationType.DELETE, new Delete(foo)))
113 .setRegion(region)
114 .build(),
115 MetricsConnection.newCallStats());
116 METRICS.updateRpc(
117 ClientService.getDescriptor().findMethodByName("Mutate"),
118 MutateRequest.newBuilder()
119 .setMutation(ProtobufUtil.toMutation(MutationType.INCREMENT, new Increment(foo)))
120 .setRegion(region)
121 .build(),
122 MetricsConnection.newCallStats());
123 METRICS.updateRpc(
124 ClientService.getDescriptor().findMethodByName("Mutate"),
125 MutateRequest.newBuilder()
126 .setMutation(ProtobufUtil.toMutation(MutationType.PUT, new Put(foo)))
127 .setRegion(region)
128 .build(),
129 MetricsConnection.newCallStats());
130 }
131 for (MetricsConnection.CallTracker t : new MetricsConnection.CallTracker[] {
132 METRICS.getTracker, METRICS.scanTracker, METRICS.multiTracker, METRICS.appendTracker,
133 METRICS.deleteTracker, METRICS.incrementTracker, METRICS.putTracker
134 }) {
135 Assert.assertEquals("Failed to invoke callTimer on " + t, loop, t.callTimer.count());
136 Assert.assertEquals("Failed to invoke reqHist on " + t, loop, t.reqHist.count());
137 Assert.assertEquals("Failed to invoke respHist on " + t, loop, t.respHist.count());
138 }
139 RatioGauge executorMetrics = (RatioGauge) METRICS.getMetricsRegistry()
140 .allMetrics().get(METRICS.getExecutorPoolName());
141 RatioGauge metaMetrics = (RatioGauge) METRICS.getMetricsRegistry()
142 .allMetrics().get(METRICS.getMetaPoolName());
143 assertEquals((double) 0.5, executorMetrics.value(), 0);
144 assertEquals(Double.NaN, metaMetrics.value(), 0);
145 }
146 }