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  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   * This will find where data for a region is located in HDFS. It ranks
59   * {@link ServerName}'s by the size of the store files they are holding for a
60   * given region.
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    // Do not scheduleFullRefresh at master startup
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    // The cache for where regions are located.
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    * Create a cache for region to list of servers
110    * @param time time to cache the locations
111    * @return A new Cache.
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       // Only count the refresh if it includes user tables ( eg more than meta and namespace ).
136       lastFullRefresh = scheduleFullRefresh()?currentTime:lastFullRefresh;
137     }
138 
139   }
140 
141   /**
142    * Refresh all the region locations.
143    *
144    * @return true if user created regions got refreshed.
145    */
146   private boolean scheduleFullRefresh() {
147     // Protect from anything being null while starting up.
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    * Returns an ordered list of hosts which have better locality for this region
177    * than the current host.
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    * Returns an ordered list of hosts that are hosting the blocks for this
193    * region. The weight of each host is the sum of the block lengths of all
194    * files on that host, so the first host in the list is the server which holds
195    * the most bytes of the given region's HFiles.
196    *
197    * @param region region
198    * @return ordered list of hosts holding blocks of the specified region
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    * return HTableDescriptor for a given tableName
218    *
219    * @param tableName the table name
220    * @return HTableDescriptor
221    * @throws IOException
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    * Map hostname to ServerName, The output ServerName list will have the same
239    * order as input hosts.
240    *
241    * @param hosts the list of hosts
242    * @return ServerName list
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     // create a mapping from hostname to ServerName for fast lookup
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         // it is possible that HDFS is up ( thus host is valid ),
271         // but RS is down ( thus sn is null )
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   // For test
333   LoadingCache<HRegionInfo, HDFSBlocksDistribution> getCache() {
334     return cache;
335   }
336 }