View Javadoc

1   /**
2    * Licensed to the Apache Software Foundation (ASF) under one
3    * or more contributor license agreements.  See the NOTICE file
4    * distributed with this work for additional information
5    * regarding copyright ownership.  The ASF licenses this file
6    * to you under the Apache License, Version 2.0 (the
7    * "License"); you may not use this file except in compliance
8    * with the License.  You may obtain a copy of the License at
9    *
10   *     http://www.apache.org/licenses/LICENSE-2.0
11   *
12   * Unless required by applicable law or agreed to in writing, software
13   * distributed under the License is distributed on an "AS IS" BASIS,
14   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15   * See the License for the specific language governing permissions and
16   * limitations under the License.
17   */
18  
19  package org.apache.hadoop.hbase.replication.regionserver;
20  
21  import org.apache.hadoop.hbase.metrics.MetricRegistryInfo;
22  import org.apache.hadoop.metrics2.lib.MutableFastCounter;
23  import org.apache.hadoop.metrics2.lib.MutableGaugeLong;
24  import org.apache.hadoop.metrics2.lib.MutableHistogram;
25  
26  public class MetricsReplicationGlobalSourceSource implements MetricsReplicationSourceSource{
27    private final MetricsReplicationSourceImpl rms;
28  
29    private final MutableHistogram ageOfLastShippedOpHist;
30    private final MutableGaugeLong sizeOfLogQueueGauge;
31    private final MutableFastCounter logReadInEditsCounter;
32    private final MutableFastCounter logEditsFilteredCounter;
33    private final MutableFastCounter shippedBatchesCounter;
34    private final MutableFastCounter shippedOpsCounter;
35    private final MutableFastCounter shippedBytesCounter;
36    @Deprecated
37    private final MutableFastCounter shippedKBsCounter;
38    private final MutableFastCounter logReadInBytesCounter;
39    private final MutableFastCounter shippedHFilesCounter;
40    private final MutableGaugeLong sizeOfHFileRefsQueueGauge;
41    private final MutableFastCounter unknownFileLengthForClosedWAL;
42    private final MutableFastCounter uncleanlyClosedWAL;
43    private final MutableFastCounter uncleanlyClosedSkippedBytes;
44    private final MutableFastCounter restartWALReading;
45    private final MutableFastCounter repeatedFileBytes;
46    private final MutableFastCounter completedWAL;
47    private final MutableFastCounter completedRecoveryQueue;
48    private final MutableFastCounter failedRecoveryQueue;
49  
50    public MetricsReplicationGlobalSourceSource(MetricsReplicationSourceImpl rms) {
51      this.rms = rms;
52  
53      ageOfLastShippedOpHist = rms.getMetricsRegistry().getHistogram(SOURCE_AGE_OF_LAST_SHIPPED_OP);
54  
55      sizeOfLogQueueGauge = rms.getMetricsRegistry().getGauge(SOURCE_SIZE_OF_LOG_QUEUE, 0L);
56  
57      shippedBatchesCounter = rms.getMetricsRegistry().getCounter(SOURCE_SHIPPED_BATCHES, 0L);
58  
59      shippedOpsCounter = rms.getMetricsRegistry().getCounter(SOURCE_SHIPPED_OPS, 0L);
60  
61      shippedKBsCounter = rms.getMetricsRegistry().getCounter(SOURCE_SHIPPED_KBS, 0L);
62  
63      shippedBytesCounter = rms.getMetricsRegistry().getCounter(SOURCE_SHIPPED_BYTES, 0L);
64  
65      logReadInBytesCounter = rms.getMetricsRegistry().getCounter(SOURCE_LOG_READ_IN_BYTES, 0L);
66  
67      logReadInEditsCounter = rms.getMetricsRegistry().getCounter(SOURCE_LOG_READ_IN_EDITS, 0L);
68  
69      logEditsFilteredCounter = rms.getMetricsRegistry().getCounter(SOURCE_LOG_EDITS_FILTERED, 0L);
70  
71      shippedHFilesCounter = rms.getMetricsRegistry().getCounter(SOURCE_SHIPPED_HFILES, 0L);
72  
73      sizeOfHFileRefsQueueGauge =
74          rms.getMetricsRegistry().getGauge(SOURCE_SIZE_OF_HFILE_REFS_QUEUE, 0L);
75  
76      unknownFileLengthForClosedWAL = rms.getMetricsRegistry().getCounter(SOURCE_CLOSED_LOGS_WITH_UNKNOWN_LENGTH, 0L);
77      uncleanlyClosedWAL = rms.getMetricsRegistry().getCounter(SOURCE_UNCLEANLY_CLOSED_LOGS, 0L);
78      uncleanlyClosedSkippedBytes = rms.getMetricsRegistry().getCounter(SOURCE_UNCLEANLY_CLOSED_IGNORED_IN_BYTES, 0L);
79      restartWALReading = rms.getMetricsRegistry().getCounter(SOURCE_RESTARTED_LOG_READING, 0L);
80      repeatedFileBytes = rms.getMetricsRegistry().getCounter(SOURCE_REPEATED_LOG_FILE_BYTES, 0L);
81      completedWAL = rms.getMetricsRegistry().getCounter(SOURCE_COMPLETED_LOGS, 0L);
82      completedRecoveryQueue = rms.getMetricsRegistry().getCounter(SOURCE_COMPLETED_RECOVERY_QUEUES, 0L);
83      failedRecoveryQueue = rms.getMetricsRegistry()
84          .getCounter(SOURCE_FAILED_RECOVERY_QUEUES, 0L);
85    }
86  
87    @Override public void setLastShippedAge(long age) {
88      ageOfLastShippedOpHist.add(age);
89    }
90  
91    @Override public void incrSizeOfLogQueue(int size) {
92      sizeOfLogQueueGauge.incr(size);
93    }
94  
95    @Override public void decrSizeOfLogQueue(int size) {
96      sizeOfLogQueueGauge.decr(size);
97    }
98  
99    @Override public void incrLogReadInEdits(long size) {
100     logReadInEditsCounter.incr(size);
101   }
102 
103   @Override public void incrLogEditsFiltered(long size) {
104     logEditsFilteredCounter.incr(size);
105   }
106 
107   @Override public void incrBatchesShipped(int batches) {
108     shippedBatchesCounter.incr(batches);
109   }
110 
111   @Override public void incrOpsShipped(long ops) {
112     shippedOpsCounter.incr(ops);
113   }
114 
115   @Override public void incrShippedBytes(long size) {
116     shippedBytesCounter.incr(size);
117     // obtained value maybe smaller than 1024. We should make sure that KB count
118     // eventually picks up even from multiple smaller updates.
119     incrementKBsCounter(shippedBytesCounter, shippedKBsCounter);
120   }
121 
122   static void incrementKBsCounter(MutableFastCounter bytesCounter, MutableFastCounter kbsCounter) {
123     // Following code should be thread-safe.
124     long delta = 0;
125     while(true) {
126       long bytes = bytesCounter.value();
127       delta = (bytes / 1024) - kbsCounter.value();
128       if (delta > 0) {
129         kbsCounter.incr(delta);
130       } else {
131         break;
132       }
133     }
134   }
135 
136   @Override public void incrLogReadInBytes(long size) {
137     logReadInBytesCounter.incr(size);
138   }
139 
140   @Override public void clear() {
141   }
142 
143   @Override
144   public long getLastShippedAge() {
145     return ageOfLastShippedOpHist.getMax();
146   }
147 
148   @Override public void incrHFilesShipped(long hfiles) {
149     shippedHFilesCounter.incr(hfiles);
150   }
151 
152   @Override
153   public void incrSizeOfHFileRefsQueue(long size) {
154     sizeOfHFileRefsQueueGauge.incr(size);
155   }
156 
157   @Override
158   public void decrSizeOfHFileRefsQueue(long size) {
159     sizeOfHFileRefsQueueGauge.decr(size);
160   }
161 
162   @Override
163   public int getSizeOfLogQueue() {
164     return (int)sizeOfLogQueueGauge.value();
165   }
166 
167   @Override
168   public void incrUnknownFileLengthForClosedWAL() {
169     unknownFileLengthForClosedWAL.incr(1L);
170   }
171   @Override
172   public void incrUncleanlyClosedWALs() {
173     uncleanlyClosedWAL.incr(1L);
174   }
175   @Override
176   public void incrBytesSkippedInUncleanlyClosedWALs(final long bytes) {
177     uncleanlyClosedSkippedBytes.incr(bytes);
178   }
179   @Override
180   public void incrRestartedWALReading() {
181     restartWALReading.incr(1L);
182   }
183   @Override
184   public void incrRepeatedFileBytes(final long bytes) {
185     repeatedFileBytes.incr(bytes);
186   }
187   @Override
188   public void incrCompletedWAL() {
189     completedWAL.incr(1L);
190   }
191   @Override
192   public void incrCompletedRecoveryQueue() {
193     completedRecoveryQueue.incr(1L);
194   }
195   @Override
196   public void incrFailedRecoveryQueue() {
197     failedRecoveryQueue.incr(1L);
198   }
199   @Override
200   public void init() {
201     rms.init();
202   }
203 
204   @Override
205   public void setGauge(String gaugeName, long value) {
206     rms.setGauge(gaugeName, value);
207   }
208 
209   @Override
210   public void incGauge(String gaugeName, long delta) {
211     rms.incGauge(gaugeName, delta);
212   }
213 
214   @Override
215   public void decGauge(String gaugeName, long delta) {
216     rms.decGauge(gaugeName, delta);
217   }
218 
219   @Override
220   public void removeMetric(String key) {
221     rms.removeMetric(key);
222   }
223 
224   @Override
225   public void incCounters(String counterName, long delta) {
226     rms.incCounters(counterName, delta);
227   }
228 
229   @Override
230   public void updateHistogram(String name, long value) {
231     rms.updateHistogram(name, value);
232   }
233 
234   @Override
235   public String getMetricsContext() {
236     return rms.getMetricsContext();
237   }
238 
239   @Override
240   public String getMetricsDescription() {
241     return rms.getMetricsDescription();
242   }
243 
244   @Override
245   public String getMetricsJmxContext() {
246     return rms.getMetricsJmxContext();
247   }
248 
249   @Override
250   public String getMetricsName() {
251     return rms.getMetricsName();
252   }
253 
254   @Override
255   public MetricRegistryInfo getMetricRegistryInfo() {
256     return new MetricRegistryInfo(getMetricsName(), getMetricsDescription(),
257       getMetricsContext(), getMetricsJmxContext(), true);
258   }
259 }