1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18 package org.apache.hadoop.hbase.master.balancer;
19
20 import com.google.common.cache.CacheBuilder;
21 import com.google.common.cache.CacheLoader;
22 import com.google.common.cache.LoadingCache;
23 import com.google.common.collect.Lists;
24 import com.google.common.util.concurrent.Futures;
25 import com.google.common.util.concurrent.ListenableFuture;
26 import com.google.common.util.concurrent.ListeningExecutorService;
27 import com.google.common.util.concurrent.MoreExecutors;
28 import com.google.common.util.concurrent.ThreadFactoryBuilder;
29 import org.apache.commons.logging.Log;
30 import org.apache.commons.logging.LogFactory;
31 import org.apache.hadoop.conf.Configuration;
32 import org.apache.hadoop.hbase.ClusterStatus;
33 import org.apache.hadoop.hbase.HDFSBlocksDistribution;
34 import org.apache.hadoop.hbase.HRegionInfo;
35 import org.apache.hadoop.hbase.HTableDescriptor;
36 import org.apache.hadoop.hbase.ServerName;
37 import org.apache.hadoop.hbase.TableName;
38 import org.apache.hadoop.hbase.classification.InterfaceAudience;
39 import org.apache.hadoop.hbase.master.AssignmentManager;
40 import org.apache.hadoop.hbase.master.MasterServices;
41 import org.apache.hadoop.hbase.master.RegionStates;
42 import org.apache.hadoop.hbase.regionserver.HRegion;
43 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
44
45 import java.io.FileNotFoundException;
46 import java.io.IOException;
47 import java.util.ArrayList;
48 import java.util.Collection;
49 import java.util.HashMap;
50 import java.util.List;
51 import java.util.Set;
52 import java.util.concurrent.Callable;
53 import java.util.concurrent.ExecutionException;
54 import java.util.concurrent.Executors;
55 import java.util.concurrent.TimeUnit;
56
57
58
59
60
61
62
63 @InterfaceAudience.Private
64 class RegionLocationFinder {
65 private static final Log LOG = LogFactory.getLog(RegionLocationFinder.class);
66 private static final long CACHE_TIME = 240 * 60 * 1000;
67 private static final HDFSBlocksDistribution EMPTY_BLOCK_DISTRIBUTION = new HDFSBlocksDistribution();
68 private Configuration conf;
69 private volatile ClusterStatus status;
70 private MasterServices services;
71 private final ListeningExecutorService executor;
72
73 private long lastFullRefresh = EnvironmentEdgeManager.currentTime();
74
75 private CacheLoader<HRegionInfo, HDFSBlocksDistribution> loader =
76 new CacheLoader<HRegionInfo, HDFSBlocksDistribution>() {
77
78 public ListenableFuture<HDFSBlocksDistribution> reload(final HRegionInfo hri,
79 HDFSBlocksDistribution oldValue) throws Exception {
80 return executor.submit(new Callable<HDFSBlocksDistribution>() {
81 @Override
82 public HDFSBlocksDistribution call() throws Exception {
83 return internalGetTopBlockLocation(hri);
84 }
85 });
86 }
87
88 @Override
89 public HDFSBlocksDistribution load(HRegionInfo key) throws Exception {
90 return internalGetTopBlockLocation(key);
91 }
92 };
93
94
95 private LoadingCache<HRegionInfo, HDFSBlocksDistribution> cache = null;
96
97 RegionLocationFinder() {
98 this.cache = createCache();
99 executor = MoreExecutors.listeningDecorator(
100 Executors.newScheduledThreadPool(
101 5,
102 new ThreadFactoryBuilder().
103 setDaemon(true)
104 .setNameFormat("region-location-%d")
105 .build()));
106 }
107
108
109
110
111
112
113 private LoadingCache<HRegionInfo, HDFSBlocksDistribution> createCache() {
114 return CacheBuilder.newBuilder()
115 .expireAfterWrite(CACHE_TIME, TimeUnit.MILLISECONDS)
116 .build(loader);
117 }
118
119 public Configuration getConf() {
120 return conf;
121 }
122
123 public void setConf(Configuration conf) {
124 this.conf = conf;
125 }
126
127 public void setServices(MasterServices services) {
128 this.services = services;
129 }
130
131 public void setClusterStatus(ClusterStatus status) {
132 long currentTime = EnvironmentEdgeManager.currentTime();
133 this.status = status;
134 if (currentTime > lastFullRefresh + (CACHE_TIME / 2)) {
135
136 lastFullRefresh = scheduleFullRefresh()?currentTime:lastFullRefresh;
137 }
138
139 }
140
141
142
143
144
145
146 private boolean scheduleFullRefresh() {
147
148 if (services == null) {
149 return false;
150 }
151 AssignmentManager am = services.getAssignmentManager();
152
153 if (am == null) {
154 return false;
155 }
156 RegionStates regionStates = am.getRegionStates();
157 if (regionStates == null) {
158 return false;
159 }
160
161 Set<HRegionInfo> regions = regionStates.getRegionAssignments().keySet();
162 boolean includesUserTables = false;
163 for (final HRegionInfo hri : regions) {
164 cache.refresh(hri);
165 includesUserTables = includesUserTables || !hri.isSystemTable();
166 }
167 return includesUserTables;
168 }
169
170 protected List<ServerName> getTopBlockLocations(HRegionInfo region) {
171 List<String> topHosts = getBlockDistribution(region).getTopHosts();
172 return mapHostNameToServerName(topHosts);
173 }
174
175
176
177
178
179 protected List<ServerName> getTopBlockLocations(HRegionInfo region, String currentHost) {
180 HDFSBlocksDistribution blocksDistribution = getBlockDistribution(region);
181 List<String> topHosts = new ArrayList<String>();
182 for (String host : blocksDistribution.getTopHosts()) {
183 if (host.equals(currentHost)) {
184 break;
185 }
186 topHosts.add(host);
187 }
188 return mapHostNameToServerName(topHosts);
189 }
190
191
192
193
194
195
196
197
198
199
200 protected HDFSBlocksDistribution internalGetTopBlockLocation(HRegionInfo region) {
201 try {
202 HTableDescriptor tableDescriptor = getTableDescriptor(region.getTable());
203 if (tableDescriptor != null) {
204 HDFSBlocksDistribution blocksDistribution =
205 HRegion.computeHDFSBlocksDistribution(getConf(), tableDescriptor, region);
206 return blocksDistribution;
207 }
208 } catch (IOException ioe) {
209 LOG.warn("IOException during HDFSBlocksDistribution computation. for " + "region = "
210 + region.getEncodedName(), ioe);
211 }
212
213 return EMPTY_BLOCK_DISTRIBUTION;
214 }
215
216
217
218
219
220
221
222
223 protected HTableDescriptor getTableDescriptor(TableName tableName) throws IOException {
224 HTableDescriptor tableDescriptor = null;
225 try {
226 if (this.services != null && this.services.getTableDescriptors() != null) {
227 tableDescriptor = this.services.getTableDescriptors().get(tableName);
228 }
229 } catch (FileNotFoundException fnfe) {
230 LOG.debug("FileNotFoundException during getTableDescriptors." + " Current table name = "
231 + tableName, fnfe);
232 }
233
234 return tableDescriptor;
235 }
236
237
238
239
240
241
242
243
244 protected List<ServerName> mapHostNameToServerName(List<String> hosts) {
245 if (hosts == null || status == null) {
246 if (hosts == null) {
247 LOG.warn("RegionLocationFinder top hosts is null");
248 }
249 return Lists.newArrayList();
250 }
251
252 List<ServerName> topServerNames = new ArrayList<ServerName>();
253 Collection<ServerName> regionServers = status.getServers();
254
255
256 HashMap<String, List<ServerName>> hostToServerName = new HashMap<String, List<ServerName>>();
257 for (ServerName sn : regionServers) {
258 String host = sn.getHostname();
259 if (!hostToServerName.containsKey(host)) {
260 hostToServerName.put(host, new ArrayList<ServerName>());
261 }
262 hostToServerName.get(host).add(sn);
263 }
264
265 for (String host : hosts) {
266 if (!hostToServerName.containsKey(host)) {
267 continue;
268 }
269 for (ServerName sn : hostToServerName.get(host)) {
270
271
272 if (sn != null) {
273 topServerNames.add(sn);
274 }
275 }
276 }
277 return topServerNames;
278 }
279
280 public HDFSBlocksDistribution getBlockDistribution(HRegionInfo hri) {
281 HDFSBlocksDistribution blockDistbn = null;
282 try {
283 if (cache.asMap().containsKey(hri)) {
284 blockDistbn = cache.get(hri);
285 return blockDistbn;
286 } else {
287 LOG.trace("HDFSBlocksDistribution not found in cache for region " + hri.getRegionNameAsString());
288 blockDistbn = internalGetTopBlockLocation(hri);
289 cache.put(hri, blockDistbn);
290 return blockDistbn;
291 }
292 } catch (ExecutionException e) {
293 LOG.warn("Error while fetching cache entry ", e);
294 blockDistbn = internalGetTopBlockLocation(hri);
295 cache.put(hri, blockDistbn);
296 return blockDistbn;
297 }
298 }
299
300 private ListenableFuture<HDFSBlocksDistribution> asyncGetBlockDistribution(
301 HRegionInfo hri) {
302 try {
303 return loader.reload(hri, EMPTY_BLOCK_DISTRIBUTION);
304 } catch (Exception e) {
305 return Futures.immediateFuture(EMPTY_BLOCK_DISTRIBUTION);
306 }
307 }
308
309 public void refreshAndWait(Collection<HRegionInfo> hris) {
310 ArrayList<ListenableFuture<HDFSBlocksDistribution>> regionLocationFutures =
311 new ArrayList<ListenableFuture<HDFSBlocksDistribution>>(hris.size());
312 for (HRegionInfo hregionInfo : hris) {
313 regionLocationFutures.add(asyncGetBlockDistribution(hregionInfo));
314 }
315 int index = 0;
316 for (HRegionInfo hregionInfo : hris) {
317 ListenableFuture<HDFSBlocksDistribution> future = regionLocationFutures
318 .get(index);
319 try {
320 cache.put(hregionInfo, future.get());
321 } catch (InterruptedException ite) {
322 Thread.currentThread().interrupt();
323 } catch (ExecutionException ee) {
324 LOG.debug(
325 "ExecutionException during HDFSBlocksDistribution computation. for region = "
326 + hregionInfo.getEncodedName(), ee);
327 }
328 index++;
329 }
330 }
331
332
333 LoadingCache<HRegionInfo, HDFSBlocksDistribution> getCache() {
334 return cache;
335 }
336 }