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.commons.logging.Log;
22  import org.apache.commons.logging.LogFactory;
23  import org.apache.hadoop.hbase.classification.InterfaceAudience;
24  import org.apache.hadoop.hbase.CompatibilitySingletonFactory;
25  import org.apache.hadoop.hbase.HBaseInterfaceAudience;
26  import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
27  
28  /**
29   * This class is for maintaining the various replication statistics for a source and publishing them
30   * through the metrics interfaces.
31   */
32  @InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.REPLICATION)
33  public class MetricsSource {
34  
35    public static final Log LOG = LogFactory.getLog(MetricsSource.class);
36  
37    private long lastTimestamp = 0;
38    private int lastQueueSize = 0;
39  
40    private final MetricsReplicationSourceSource singleSourceSource;
41    private final MetricsReplicationSourceSource globalSourceSource;
42  
43    /**
44     * Constructor used to register the metrics
45     *
46     * @param id Name of the source this class is monitoring
47     */
48    public MetricsSource(String id) {
49      singleSourceSource =
50          CompatibilitySingletonFactory.getInstance(MetricsReplicationSourceFactory.class)
51              .getSource(id);
52      globalSourceSource = CompatibilitySingletonFactory.getInstance(MetricsReplicationSourceFactory.class).getGlobalSource();
53    }
54  
55    /**
56     * Set the age of the last edit that was shipped
57     *
58     * @param timestamp write time of the edit
59     */
60    public void setAgeOfLastShippedOp(long timestamp) {
61      long age = EnvironmentEdgeManager.currentTime() - timestamp;
62      singleSourceSource.setLastShippedAge(age);
63      globalSourceSource.setLastShippedAge(age);
64      this.lastTimestamp = timestamp;
65    }
66  
67    /**
68     * Convenience method to use the last given timestamp to refresh the age of the last edit. Used
69     * when replication fails and need to keep that metric accurate.
70     */
71    public void refreshAgeOfLastShippedOp() {
72      if (this.lastTimestamp > 0) {
73        setAgeOfLastShippedOp(this.lastTimestamp);
74      }
75    }
76  
77    /**
78     * Set the size of the log queue
79     *
80     * @param size the size.
81     */
82    public void setSizeOfLogQueue(int size) {
83      singleSourceSource.setSizeOfLogQueue(size);
84      globalSourceSource.incrSizeOfLogQueue(size - lastQueueSize);
85      lastQueueSize = size;
86    }
87  
88    /**
89     * Add on the the number of log edits read
90     *
91     * @param delta the number of log edits read.
92     */
93    private void incrLogEditsRead(long delta) {
94      singleSourceSource.incrLogReadInEdits(delta);
95      globalSourceSource.incrLogReadInEdits(delta);
96    }
97  
98    /** Increment the number of log edits read by one. */
99    public void incrLogEditsRead() {
100     incrLogEditsRead(1);
101   }
102 
103   /**
104    * Add on the number of log edits filtered
105    *
106    * @param delta the number filtered.
107    */
108   private void incrLogEditsFiltered(long delta) {
109     singleSourceSource.incrLogEditsFiltered(delta);
110     globalSourceSource.incrLogEditsFiltered(delta);
111   }
112 
113   /** The number of log edits filtered out. */
114   public void incrLogEditsFiltered() {
115     incrLogEditsFiltered(1);
116   }
117 
118   /**
119    * Convience method to apply changes to metrics do to shipping a batch of logs.
120    *
121    * @param batchSize the size of the batch that was shipped to sinks.
122    */
123   public void shipBatch(long batchSize, int sizeInKB) {
124     singleSourceSource.incrBatchesShipped(1);
125     globalSourceSource.incrBatchesShipped(1);
126 
127     singleSourceSource.incrOpsShipped(batchSize);
128     globalSourceSource.incrOpsShipped(batchSize);
129 
130     singleSourceSource.incrShippedKBs(sizeInKB);
131     globalSourceSource.incrShippedKBs(sizeInKB);
132   }
133 
134   /** increase the byte number read by source from log file */
135   public void incrLogReadInBytes(long readInBytes) {
136     singleSourceSource.incrLogReadInBytes(readInBytes);
137     globalSourceSource.incrLogReadInBytes(readInBytes);
138   }
139 
140   /** Removes all metrics about this Source. */
141   public void clear() {
142     singleSourceSource.clear();
143     globalSourceSource.decrSizeOfLogQueue(lastQueueSize);
144     lastQueueSize = 0;
145   }
146 }