1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18 package org.apache.hadoop.hbase.metrics.impl;
19
20
21 import java.lang.reflect.InvocationTargetException;
22 import java.lang.reflect.Method;
23 import java.util.Collection;
24 import java.util.HashMap;
25 import java.util.Iterator;
26 import java.util.Map;
27 import java.util.Map.Entry;
28 import java.util.concurrent.TimeUnit;
29 import java.util.concurrent.atomic.AtomicBoolean;
30
31 import org.apache.commons.lang.reflect.FieldUtils;
32 import org.apache.commons.logging.Log;
33 import org.apache.commons.logging.LogFactory;
34 import org.apache.hadoop.hbase.metrics.MetricRegistries;
35 import org.apache.hadoop.hbase.metrics.MetricRegistry;
36 import org.apache.hadoop.hbase.metrics.MetricRegistryInfo;
37 import org.apache.hadoop.metrics2.MetricsCollector;
38 import org.apache.hadoop.metrics2.MetricsExecutor;
39 import org.apache.hadoop.metrics2.MetricsSource;
40 import org.apache.hadoop.metrics2.MetricsSystem;
41 import org.apache.hadoop.metrics2.impl.JmxCacheBuster;
42 import org.apache.hadoop.metrics2.impl.MetricsSystemImpl;
43 import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem;
44 import org.apache.hadoop.metrics2.lib.DefaultMetricsSystemHelper;
45 import org.apache.hadoop.metrics2.lib.MetricsExecutorImpl;
46
47 import com.google.common.annotations.VisibleForTesting;
48 import com.google.common.base.Optional;
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72 public class GlobalMetricRegistriesAdapter {
73
74 private static final Log LOG = LogFactory.getLog(GlobalMetricRegistriesAdapter.class);
75
76 private class MetricsSourceAdapter implements MetricsSource {
77 private final MetricRegistry registry;
78 MetricsSourceAdapter(MetricRegistry registry) {
79 this.registry = registry;
80 }
81
82 @Override
83 public void getMetrics(MetricsCollector collector, boolean all) {
84 metricsAdapter.snapshotAllMetrics(registry, collector);
85 }
86 }
87
88 private final MetricsExecutor executor;
89 private final AtomicBoolean stopped;
90 private final DefaultMetricsSystemHelper helper;
91 private final HBaseMetrics2HadoopMetricsAdapter metricsAdapter;
92 private final HashMap<MetricRegistryInfo, MetricsSourceAdapter> registeredSources;
93
94 private GlobalMetricRegistriesAdapter() {
95 this.executor = new MetricsExecutorImpl();
96 this.stopped = new AtomicBoolean(false);
97 this.metricsAdapter = new HBaseMetrics2HadoopMetricsAdapter();
98 this.registeredSources = new HashMap<>();
99 this.helper = new DefaultMetricsSystemHelper();
100 executor.getExecutor().scheduleAtFixedRate(new Runnable(){
101 @Override
102 public void run() {
103 doRun();
104
105 }}, 10, 10, TimeUnit.SECONDS);
106 }
107
108
109
110
111
112 public static GlobalMetricRegistriesAdapter init() {
113 return new GlobalMetricRegistriesAdapter();
114 }
115
116 @VisibleForTesting
117 public void stop() {
118 stopped.set(true);
119 }
120
121 private void doRun() {
122 if (stopped.get()) {
123 executor.stop();
124 return;
125 }
126 if (LOG.isTraceEnabled()) {
127 LOG.trace("doRun called: " + registeredSources);
128 }
129
130 Collection<MetricRegistry> registries = MetricRegistries.global().getMetricRegistries();
131 for (MetricRegistry registry : registries) {
132 MetricRegistryInfo info = registry.getMetricRegistryInfo();
133
134 if (info.isExistingSource()) {
135
136
137
138
139
140
141 continue;
142 }
143
144 if (!registeredSources.containsKey(info)) {
145 if (LOG.isDebugEnabled()) {
146 LOG.debug("Registering adapter for the MetricRegistry: " + info.getMetricsJmxContext());
147 }
148
149 MetricsSourceAdapter adapter = new MetricsSourceAdapter(registry);
150 LOG.info("Registering " + info.getMetricsJmxContext() + " " + info.getMetricsDescription());
151 DefaultMetricsSystem.instance().register(info.getMetricsJmxContext(),
152 info.getMetricsDescription(), adapter);
153 registeredSources.put(info, adapter);
154
155
156 }
157 }
158
159 boolean removed = false;
160
161 for (Iterator<Entry<MetricRegistryInfo, MetricsSourceAdapter>> it =
162 registeredSources.entrySet().iterator(); it.hasNext();) {
163 Entry<MetricRegistryInfo, MetricsSourceAdapter> entry = it.next();
164 MetricRegistryInfo info = entry.getKey();
165 Optional<MetricRegistry> found = MetricRegistries.global().get(info);
166 if (!found.isPresent()) {
167 if (LOG.isDebugEnabled()) {
168 LOG.debug("Removing adapter for the MetricRegistry: " + info.getMetricsJmxContext());
169 }
170 synchronized(DefaultMetricsSystem.instance()) {
171 unregisterSource(info);
172 helper.removeSourceName(info.getMetricsJmxContext());
173 helper.removeObjectName(info.getMetricsJmxContext());
174 it.remove();
175 removed = true;
176 }
177 }
178 }
179 if (removed) {
180 JmxCacheBuster.clearJmxCache();
181 }
182 }
183
184
185
186
187
188 @VisibleForTesting
189 protected void unregisterSource(MetricRegistryInfo info) {
190
191 MetricsSystem metricsSystem = DefaultMetricsSystem.instance();
192 if (metricsSystem instanceof MetricsSystemImpl) {
193 try {
194
195
196 @SuppressWarnings("unchecked")
197 Map<String, Object> sources =
198 (Map<String, Object>) FieldUtils.readField(metricsSystem, "sources", true);
199 String sourceName = info.getMetricsJmxContext();
200 if (sources.containsKey(sourceName)) {
201 Object sourceAdapter = sources.get(sourceName);
202 Method method = null;
203 try {
204 method = sourceAdapter.getClass().getDeclaredMethod("stop");
205 } catch (NoSuchMethodException e) {
206 LOG.info("Stop method not found on MetricsSourceAdapter");
207 } catch (SecurityException e) {
208 LOG.info("Don't have access to call stop method not found on MetricsSourceAdapter", e);
209 }
210 if (method != null) {
211 method.setAccessible(true);
212 try {
213 method.invoke(sourceAdapter);
214 } catch (IllegalArgumentException | InvocationTargetException e) {
215 LOG.warn("Couldn't invoke stop on metrics source adapter: " + sourceName);
216 e.printStackTrace();
217 }
218 }
219 sources.remove(sourceName);
220
221 }
222 @SuppressWarnings("unchecked")
223 Map<String, MetricsSource> allSources =
224 (Map<String, MetricsSource>) FieldUtils.readField(metricsSystem, "allSources", true);
225 if (allSources.containsKey(sourceName)) {
226 allSources.remove(sourceName);
227 }
228 } catch (IllegalAccessException e) {
229 LOG.warn("Error unregistering metric source " + info.getMetricsJmxContext());
230 }
231 }
232 }
233 }