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 com.codahale.metrics.MetricRegistry.name;
021import static org.apache.hadoop.hbase.util.ConcurrentMapUtils.computeIfAbsent;
022
023import com.codahale.metrics.Counter;
024import com.codahale.metrics.Histogram;
025import com.codahale.metrics.JmxReporter;
026import com.codahale.metrics.MetricRegistry;
027import com.codahale.metrics.RatioGauge;
028import com.codahale.metrics.Timer;
029import org.apache.hbase.thirdparty.com.google.common.annotations.VisibleForTesting;
030
031import java.util.concurrent.ConcurrentHashMap;
032import java.util.concurrent.ConcurrentMap;
033import java.util.concurrent.ConcurrentSkipListMap;
034import java.util.concurrent.ThreadPoolExecutor;
035import java.util.concurrent.TimeUnit;
036import java.util.function.Supplier;
037import org.apache.hadoop.hbase.ServerName;
038import org.apache.yetus.audience.InterfaceAudience;
039import org.apache.hbase.thirdparty.com.google.protobuf.Descriptors.MethodDescriptor;
040import org.apache.hbase.thirdparty.com.google.protobuf.Message;
041import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.ClientService;
042import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.MutateRequest;
043import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.MutationProto.MutationType;
044import org.apache.hadoop.hbase.util.Bytes;
045
046/**
047 * This class is for maintaining the various connection statistics and publishing them through
048 * the metrics interfaces.
049 *
050 * This class manages its own {@link MetricRegistry} and {@link JmxReporter} so as to not
051 * conflict with other uses of Yammer Metrics within the client application. Instantiating
052 * this class implicitly creates and "starts" instances of these classes; be sure to call
053 * {@link #shutdown()} to terminate the thread pools they allocate.
054 */
055@InterfaceAudience.Private
056public class MetricsConnection implements StatisticTrackable {
057
058  /** Set this key to {@code true} to enable metrics collection of client requests. */
059  public static final String CLIENT_SIDE_METRICS_ENABLED_KEY = "hbase.client.metrics.enable";
060
061  private static final String CNT_BASE = "rpcCount_";
062  private static final String DRTN_BASE = "rpcCallDurationMs_";
063  private static final String REQ_BASE = "rpcCallRequestSizeBytes_";
064  private static final String RESP_BASE = "rpcCallResponseSizeBytes_";
065  private static final String MEMLOAD_BASE = "memstoreLoad_";
066  private static final String HEAP_BASE = "heapOccupancy_";
067  private static final String CACHE_BASE = "cacheDroppingExceptions_";
068  private static final String UNKNOWN_EXCEPTION = "UnknownException";
069  private static final String CLIENT_SVC = ClientService.getDescriptor().getName();
070
071  /** A container class for collecting details about the RPC call as it percolates. */
072  public static class CallStats {
073    private long requestSizeBytes = 0;
074    private long responseSizeBytes = 0;
075    private long startTime = 0;
076    private long callTimeMs = 0;
077    private int concurrentCallsPerServer = 0;
078    private int numActionsPerServer = 0;
079
080    public long getRequestSizeBytes() {
081      return requestSizeBytes;
082    }
083
084    public void setRequestSizeBytes(long requestSizeBytes) {
085      this.requestSizeBytes = requestSizeBytes;
086    }
087
088    public long getResponseSizeBytes() {
089      return responseSizeBytes;
090    }
091
092    public void setResponseSizeBytes(long responseSizeBytes) {
093      this.responseSizeBytes = responseSizeBytes;
094    }
095
096    public long getStartTime() {
097      return startTime;
098    }
099
100    public void setStartTime(long startTime) {
101      this.startTime = startTime;
102    }
103
104    public long getCallTimeMs() {
105      return callTimeMs;
106    }
107
108    public void setCallTimeMs(long callTimeMs) {
109      this.callTimeMs = callTimeMs;
110    }
111
112    public int getConcurrentCallsPerServer() {
113      return concurrentCallsPerServer;
114    }
115
116    public void setConcurrentCallsPerServer(int callsPerServer) {
117      this.concurrentCallsPerServer = callsPerServer;
118    }
119
120    public int getNumActionsPerServer() {
121      return numActionsPerServer;
122    }
123
124    public void setNumActionsPerServer(int numActionsPerServer) {
125      this.numActionsPerServer = numActionsPerServer;
126    }
127  }
128
129  @VisibleForTesting
130  protected static final class CallTracker {
131    private final String name;
132    @VisibleForTesting final Timer callTimer;
133    @VisibleForTesting final Histogram reqHist;
134    @VisibleForTesting final Histogram respHist;
135
136    private CallTracker(MetricRegistry registry, String name, String subName, String scope) {
137      StringBuilder sb = new StringBuilder(CLIENT_SVC).append("_").append(name);
138      if (subName != null) {
139        sb.append("(").append(subName).append(")");
140      }
141      this.name = sb.toString();
142      this.callTimer = registry.timer(name(MetricsConnection.class,
143        DRTN_BASE + this.name, scope));
144      this.reqHist = registry.histogram(name(MetricsConnection.class,
145        REQ_BASE + this.name, scope));
146      this.respHist = registry.histogram(name(MetricsConnection.class,
147        RESP_BASE + this.name, scope));
148    }
149
150    private CallTracker(MetricRegistry registry, String name, String scope) {
151      this(registry, name, null, scope);
152    }
153
154    public void updateRpc(CallStats stats) {
155      this.callTimer.update(stats.getCallTimeMs(), TimeUnit.MILLISECONDS);
156      this.reqHist.update(stats.getRequestSizeBytes());
157      this.respHist.update(stats.getResponseSizeBytes());
158    }
159
160    @Override
161    public String toString() {
162      return "CallTracker:" + name;
163    }
164  }
165
166  protected static class RegionStats {
167    final String name;
168    final Histogram memstoreLoadHist;
169    final Histogram heapOccupancyHist;
170
171    public RegionStats(MetricRegistry registry, String name) {
172      this.name = name;
173      this.memstoreLoadHist = registry.histogram(name(MetricsConnection.class,
174          MEMLOAD_BASE + this.name));
175      this.heapOccupancyHist = registry.histogram(name(MetricsConnection.class,
176          HEAP_BASE + this.name));
177    }
178
179    public void update(RegionLoadStats regionStatistics) {
180      this.memstoreLoadHist.update(regionStatistics.getMemStoreLoad());
181      this.heapOccupancyHist.update(regionStatistics.getHeapOccupancy());
182    }
183  }
184
185  @VisibleForTesting
186  protected static class RunnerStats {
187    final Counter normalRunners;
188    final Counter delayRunners;
189    final Histogram delayIntevalHist;
190
191    public RunnerStats(MetricRegistry registry) {
192      this.normalRunners = registry.counter(
193        name(MetricsConnection.class, "normalRunnersCount"));
194      this.delayRunners = registry.counter(
195        name(MetricsConnection.class, "delayRunnersCount"));
196      this.delayIntevalHist = registry.histogram(
197        name(MetricsConnection.class, "delayIntervalHist"));
198    }
199
200    public void incrNormalRunners() {
201      this.normalRunners.inc();
202    }
203
204    public void incrDelayRunners() {
205      this.delayRunners.inc();
206    }
207
208    public void updateDelayInterval(long interval) {
209      this.delayIntevalHist.update(interval);
210    }
211  }
212
213  @VisibleForTesting
214  protected ConcurrentHashMap<ServerName, ConcurrentMap<byte[], RegionStats>> serverStats
215          = new ConcurrentHashMap<>();
216
217  public void updateServerStats(ServerName serverName, byte[] regionName,
218                                Object r) {
219    if (!(r instanceof Result)) {
220      return;
221    }
222    Result result = (Result) r;
223    RegionLoadStats stats = result.getStats();
224    if (stats == null) {
225      return;
226    }
227    updateRegionStats(serverName, regionName, stats);
228  }
229
230  @Override
231  public void updateRegionStats(ServerName serverName, byte[] regionName, RegionLoadStats stats) {
232    String name = serverName.getServerName() + "," + Bytes.toStringBinary(regionName);
233    ConcurrentMap<byte[], RegionStats> rsStats = computeIfAbsent(serverStats, serverName,
234      () -> new ConcurrentSkipListMap<>(Bytes.BYTES_COMPARATOR));
235    RegionStats regionStats =
236        computeIfAbsent(rsStats, regionName, () -> new RegionStats(this.registry, name));
237    regionStats.update(stats);
238  }
239
240  /** A lambda for dispatching to the appropriate metric factory method */
241  private static interface NewMetric<T> {
242    T newMetric(Class<?> clazz, String name, String scope);
243  }
244
245  /** Anticipated number of metric entries */
246  private static final int CAPACITY = 50;
247  /** Default load factor from {@link java.util.HashMap#DEFAULT_LOAD_FACTOR} */
248  private static final float LOAD_FACTOR = 0.75f;
249  /**
250   * Anticipated number of concurrent accessor threads
251   */
252  private static final int CONCURRENCY_LEVEL = 256;
253
254  private final MetricRegistry registry;
255  private final JmxReporter reporter;
256  private final String scope;
257
258  private final NewMetric<Timer> timerFactory = new NewMetric<Timer>() {
259    @Override public Timer newMetric(Class<?> clazz, String name, String scope) {
260      return registry.timer(name(clazz, name, scope));
261    }
262  };
263
264  private final NewMetric<Histogram> histogramFactory = new NewMetric<Histogram>() {
265    @Override public Histogram newMetric(Class<?> clazz, String name, String scope) {
266      return registry.histogram(name(clazz, name, scope));
267    }
268  };
269
270  private final NewMetric<Counter> counterFactory = new NewMetric<Counter>() {
271    @Override public Counter newMetric(Class<?> clazz, String name, String scope) {
272      return registry.counter(name(clazz, name, scope));
273    }
274  };
275
276  // static metrics
277
278  @VisibleForTesting protected final Counter metaCacheHits;
279  @VisibleForTesting protected final Counter metaCacheMisses;
280  @VisibleForTesting protected final CallTracker getTracker;
281  @VisibleForTesting protected final CallTracker scanTracker;
282  @VisibleForTesting protected final CallTracker appendTracker;
283  @VisibleForTesting protected final CallTracker deleteTracker;
284  @VisibleForTesting protected final CallTracker incrementTracker;
285  @VisibleForTesting protected final CallTracker putTracker;
286  @VisibleForTesting protected final CallTracker multiTracker;
287  @VisibleForTesting protected final RunnerStats runnerStats;
288  @VisibleForTesting protected final Counter metaCacheNumClearServer;
289  @VisibleForTesting protected final Counter metaCacheNumClearRegion;
290  @VisibleForTesting protected final Counter hedgedReadOps;
291  @VisibleForTesting protected final Counter hedgedReadWin;
292  @VisibleForTesting protected final Histogram concurrentCallsPerServerHist;
293  @VisibleForTesting protected final Histogram numActionsPerServerHist;
294
295  // dynamic metrics
296
297  // These maps are used to cache references to the metric instances that are managed by the
298  // registry. I don't think their use perfectly removes redundant allocations, but it's
299  // a big improvement over calling registry.newMetric each time.
300  @VisibleForTesting protected final ConcurrentMap<String, Timer> rpcTimers =
301      new ConcurrentHashMap<>(CAPACITY, LOAD_FACTOR, CONCURRENCY_LEVEL);
302  @VisibleForTesting protected final ConcurrentMap<String, Histogram> rpcHistograms =
303      new ConcurrentHashMap<>(CAPACITY * 2 /* tracking both request and response sizes */,
304          LOAD_FACTOR, CONCURRENCY_LEVEL);
305  private final ConcurrentMap<String, Counter> cacheDroppingExceptions =
306    new ConcurrentHashMap<>(CAPACITY, LOAD_FACTOR, CONCURRENCY_LEVEL);
307  @VisibleForTesting protected final ConcurrentMap<String, Counter>  rpcCounters =
308      new ConcurrentHashMap<>(CAPACITY, LOAD_FACTOR, CONCURRENCY_LEVEL);
309
310  MetricsConnection(String scope, Supplier<ThreadPoolExecutor> batchPool,
311      Supplier<ThreadPoolExecutor> metaPool) {
312    this.scope = scope;
313    this.registry = new MetricRegistry();
314    this.registry.register(getExecutorPoolName(),
315        new RatioGauge() {
316          @Override
317          protected Ratio getRatio() {
318            ThreadPoolExecutor pool = batchPool.get();
319            if (pool == null) {
320              return Ratio.of(0, 0);
321            }
322            return Ratio.of(pool.getActiveCount(), pool.getMaximumPoolSize());
323          }
324        });
325    this.registry.register(getMetaPoolName(),
326        new RatioGauge() {
327          @Override
328          protected Ratio getRatio() {
329            ThreadPoolExecutor pool = metaPool.get();
330            if (pool == null) {
331              return Ratio.of(0, 0);
332            }
333            return Ratio.of(pool.getActiveCount(), pool.getMaximumPoolSize());
334          }
335        });
336    this.metaCacheHits = registry.counter(name(this.getClass(), "metaCacheHits", scope));
337    this.metaCacheMisses = registry.counter(name(this.getClass(), "metaCacheMisses", scope));
338    this.metaCacheNumClearServer = registry.counter(name(this.getClass(),
339      "metaCacheNumClearServer", scope));
340    this.metaCacheNumClearRegion = registry.counter(name(this.getClass(),
341      "metaCacheNumClearRegion", scope));
342    this.hedgedReadOps = registry.counter(name(this.getClass(), "hedgedReadOps", scope));
343    this.hedgedReadWin = registry.counter(name(this.getClass(), "hedgedReadWin", scope));
344    this.getTracker = new CallTracker(this.registry, "Get", scope);
345    this.scanTracker = new CallTracker(this.registry, "Scan", scope);
346    this.appendTracker = new CallTracker(this.registry, "Mutate", "Append", scope);
347    this.deleteTracker = new CallTracker(this.registry, "Mutate", "Delete", scope);
348    this.incrementTracker = new CallTracker(this.registry, "Mutate", "Increment", scope);
349    this.putTracker = new CallTracker(this.registry, "Mutate", "Put", scope);
350    this.multiTracker = new CallTracker(this.registry, "Multi", scope);
351    this.runnerStats = new RunnerStats(this.registry);
352    this.concurrentCallsPerServerHist = registry.histogram(name(MetricsConnection.class,
353      "concurrentCallsPerServer", scope));
354    this.numActionsPerServerHist = registry.histogram(name(MetricsConnection.class,
355      "numActionsPerServer", scope));
356
357    this.reporter = JmxReporter.forRegistry(this.registry).build();
358    this.reporter.start();
359  }
360
361  @VisibleForTesting
362  final String getExecutorPoolName() {
363    return name(getClass(), "executorPoolActiveThreads", scope);
364  }
365
366  @VisibleForTesting
367  final String getMetaPoolName() {
368    return name(getClass(), "metaPoolActiveThreads", scope);
369  }
370
371  @VisibleForTesting
372  MetricRegistry getMetricRegistry() {
373    return registry;
374  }
375
376  public void shutdown() {
377    this.reporter.stop();
378  }
379
380  /** Produce an instance of {@link CallStats} for clients to attach to RPCs. */
381  public static CallStats newCallStats() {
382    // TODO: instance pool to reduce GC?
383    return new CallStats();
384  }
385
386  /** Increment the number of meta cache hits. */
387  public void incrMetaCacheHit() {
388    metaCacheHits.inc();
389  }
390
391  /** Increment the number of meta cache misses. */
392  public void incrMetaCacheMiss() {
393    metaCacheMisses.inc();
394  }
395
396  /** Increment the number of meta cache drops requested for entire RegionServer. */
397  public void incrMetaCacheNumClearServer() {
398    metaCacheNumClearServer.inc();
399  }
400
401  /** Increment the number of meta cache drops requested for individual region. */
402  public void incrMetaCacheNumClearRegion() {
403    metaCacheNumClearRegion.inc();
404  }
405
406  /** Increment the number of meta cache drops requested for individual region. */
407  public void incrMetaCacheNumClearRegion(int count) {
408    metaCacheNumClearRegion.inc(count);
409  }
410
411  /** Increment the number of hedged read that have occurred. */
412  public void incrHedgedReadOps() {
413    hedgedReadOps.inc();
414  }
415
416  /** Increment the number of hedged read returned faster than the original read. */
417  public void incrHedgedReadWin() {
418    hedgedReadWin.inc();
419  }
420
421  /** Increment the number of normal runner counts. */
422  public void incrNormalRunners() {
423    this.runnerStats.incrNormalRunners();
424  }
425
426  /** Increment the number of delay runner counts and update delay interval of delay runner. */
427  public void incrDelayRunnersAndUpdateDelayInterval(long interval) {
428    this.runnerStats.incrDelayRunners();
429    this.runnerStats.updateDelayInterval(interval);
430  }
431
432  /**
433   * Get a metric for {@code key} from {@code map}, or create it with {@code factory}.
434   */
435  private <T> T getMetric(String key, ConcurrentMap<String, T> map, NewMetric<T> factory) {
436    return computeIfAbsent(map, key, () -> factory.newMetric(getClass(), key, scope));
437  }
438
439  /** Update call stats for non-critical-path methods */
440  private void updateRpcGeneric(String methodName, CallStats stats) {
441    getMetric(DRTN_BASE + methodName, rpcTimers, timerFactory)
442        .update(stats.getCallTimeMs(), TimeUnit.MILLISECONDS);
443    getMetric(REQ_BASE + methodName, rpcHistograms, histogramFactory)
444        .update(stats.getRequestSizeBytes());
445    getMetric(RESP_BASE + methodName, rpcHistograms, histogramFactory)
446        .update(stats.getResponseSizeBytes());
447  }
448
449  /** Report RPC context to metrics system. */
450  public void updateRpc(MethodDescriptor method, Message param, CallStats stats) {
451    int callsPerServer = stats.getConcurrentCallsPerServer();
452    if (callsPerServer > 0) {
453      concurrentCallsPerServerHist.update(callsPerServer);
454    }
455    // Update the counter that tracks RPCs by type.
456    final String methodName = method.getService().getName() + "_" + method.getName();
457    getMetric(CNT_BASE + methodName, rpcCounters, counterFactory).inc();
458    // this implementation is tied directly to protobuf implementation details. would be better
459    // if we could dispatch based on something static, ie, request Message type.
460    if (method.getService() == ClientService.getDescriptor()) {
461      switch(method.getIndex()) {
462        case 0:
463          assert "Get".equals(method.getName());
464          getTracker.updateRpc(stats);
465          return;
466        case 1:
467          assert "Mutate".equals(method.getName());
468          final MutationType mutationType = ((MutateRequest) param).getMutation().getMutateType();
469          switch(mutationType) {
470            case APPEND:
471              appendTracker.updateRpc(stats);
472              return;
473            case DELETE:
474              deleteTracker.updateRpc(stats);
475              return;
476            case INCREMENT:
477              incrementTracker.updateRpc(stats);
478              return;
479            case PUT:
480              putTracker.updateRpc(stats);
481              return;
482            default:
483              throw new RuntimeException("Unrecognized mutation type " + mutationType);
484          }
485        case 2:
486          assert "Scan".equals(method.getName());
487          scanTracker.updateRpc(stats);
488          return;
489        case 3:
490          assert "BulkLoadHFile".equals(method.getName());
491          // use generic implementation
492          break;
493        case 4:
494          assert "PrepareBulkLoad".equals(method.getName());
495          // use generic implementation
496          break;
497        case 5:
498          assert "CleanupBulkLoad".equals(method.getName());
499          // use generic implementation
500          break;
501        case 6:
502          assert "ExecService".equals(method.getName());
503          // use generic implementation
504          break;
505        case 7:
506          assert "ExecRegionServerService".equals(method.getName());
507          // use generic implementation
508          break;
509        case 8:
510          assert "Multi".equals(method.getName());
511          numActionsPerServerHist.update(stats.getNumActionsPerServer());
512          multiTracker.updateRpc(stats);
513          return;
514        default:
515          throw new RuntimeException("Unrecognized ClientService RPC type " + method.getFullName());
516      }
517    }
518    // Fallback to dynamic registry lookup for DDL methods.
519    updateRpcGeneric(methodName, stats);
520  }
521
522  public void incrCacheDroppingExceptions(Object exception) {
523    getMetric(CACHE_BASE +
524      (exception == null? UNKNOWN_EXCEPTION : exception.getClass().getSimpleName()),
525      cacheDroppingExceptions, counterFactory).inc();
526  }
527}