View Javadoc

1   /*
2    * Copyright The Apache Software Foundation
3    *
4    * Licensed to the Apache Software Foundation (ASF) under one
5    * or more contributor license agreements.  See the NOTICE file
6    * distributed with this work for additional information
7    * regarding copyright ownership.  The ASF licenses this file
8    * to you under the Apache License, Version 2.0 (the
9    * "License"); you may not use this file except in compliance
10   * with the License.  You may obtain a copy of the License at
11   *
12   *     http://www.apache.org/licenses/LICENSE-2.0
13   *
14   * Unless required by applicable law or agreed to in writing, software
15   * distributed under the License is distributed on an "AS IS" BASIS,
16   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
17   * See the License for the specific language governing permissions and
18   * limitations under the License.
19   */
20  
21  package org.apache.hadoop.hbase.fs;
22  
23  import java.io.Closeable;
24  import java.io.IOException;
25  import java.lang.reflect.Field;
26  import java.lang.reflect.InvocationHandler;
27  import java.lang.reflect.InvocationTargetException;
28  import java.lang.reflect.Method;
29  import java.lang.reflect.Modifier;
30  import java.lang.reflect.Proxy;
31  import java.lang.reflect.UndeclaredThrowableException;
32  import java.net.URI;
33  
34  import com.google.common.annotations.VisibleForTesting;
35  import org.apache.commons.logging.Log;
36  import org.apache.commons.logging.LogFactory;
37  import org.apache.hadoop.conf.Configuration;
38  import org.apache.hadoop.fs.FSDataOutputStream;
39  import org.apache.hadoop.fs.FileSystem;
40  import org.apache.hadoop.fs.FilterFileSystem;
41  import org.apache.hadoop.fs.LocalFileSystem;
42  import org.apache.hadoop.fs.Path;
43  import org.apache.hadoop.hbase.ServerName;
44  import org.apache.hadoop.hbase.wal.DefaultWALProvider;
45  import org.apache.hadoop.hdfs.DFSClient;
46  import org.apache.hadoop.hdfs.DistributedFileSystem;
47  import org.apache.hadoop.hdfs.protocol.ClientProtocol;
48  import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
49  import org.apache.hadoop.hdfs.protocol.LocatedBlock;
50  import org.apache.hadoop.hdfs.protocol.LocatedBlocks;
51  import org.apache.hadoop.ipc.RPC;
52  import org.apache.hadoop.util.Progressable;
53  import org.apache.hadoop.util.ReflectionUtils;
54  
55  import static org.apache.hadoop.hbase.HConstants.HBASE_DIR;
56  
57  /**
58   * An encapsulation for the FileSystem object that hbase uses to access
59   * data. This class allows the flexibility of using
60   * separate filesystem objects for reading and writing hfiles and wals.
61   */
62  public class HFileSystem extends FilterFileSystem {
63    public static final Log LOG = LogFactory.getLog(HFileSystem.class);
64  
65    /** Parameter name for HBase WAL directory */
66    public static final String HBASE_WAL_DIR = "hbase.wal.dir";
67  
68    private final FileSystem noChecksumFs;   // read hfile data from storage
69    private final boolean useHBaseChecksum;
70  
71    /**
72     * Create a FileSystem object for HBase regionservers.
73     * @param conf The configuration to be used for the filesystem
74     * @param useHBaseChecksum if true, then use
75     *        checksum verfication in hbase, otherwise
76     *        delegate checksum verification to the FileSystem.
77     */
78    public HFileSystem(Configuration conf, boolean useHBaseChecksum)
79      throws IOException {
80  
81      // Create the default filesystem with checksum verification switched on.
82      // By default, any operation to this FilterFileSystem occurs on
83      // the underlying filesystem that has checksums switched on.
84      this.fs = FileSystem.get(conf);
85      this.useHBaseChecksum = useHBaseChecksum;
86  
87      fs.initialize(getDefaultUri(conf), conf);
88      
89      // disable checksum verification for local fileSystem, see HBASE-11218
90      if (fs instanceof LocalFileSystem) {
91        fs.setWriteChecksum(false);
92        fs.setVerifyChecksum(false);
93      }
94  
95      addLocationsOrderInterceptor(conf);
96  
97      // If hbase checksum verification is switched on, then create a new
98      // filesystem object that has cksum verification turned off.
99      // We will avoid verifying checksums in the fs client, instead do it
100     // inside of hbase.
101     // If this is the local file system hadoop has a bug where seeks
102     // do not go to the correct location if setVerifyChecksum(false) is called.
103     // This manifests itself in that incorrect data is read and HFileBlocks won't be able to read
104     // their header magic numbers. See HBASE-5885
105     if (useHBaseChecksum && !(fs instanceof LocalFileSystem)) {
106       conf = new Configuration(conf);
107       conf.setBoolean("dfs.client.read.shortcircuit.skip.checksum", true);
108       this.noChecksumFs = maybeWrapFileSystem(newInstanceFileSystem(conf), conf);
109       this.noChecksumFs.setVerifyChecksum(false);
110     } else {
111       this.noChecksumFs = maybeWrapFileSystem(fs, conf);
112     }
113 
114     this.fs = maybeWrapFileSystem(this.fs, conf);
115   }
116 
117   /**
118    * Wrap a FileSystem object within a HFileSystem. The noChecksumFs and
119    * writefs are both set to be the same specified fs. 
120    * Do not verify hbase-checksums while reading data from filesystem.
121    * @param fs Set the noChecksumFs and writeFs to this specified filesystem.
122    */
123   public HFileSystem(FileSystem fs) {
124     this.fs = fs;
125     this.noChecksumFs = fs;
126     this.useHBaseChecksum = false;
127   }
128 
129   /**
130    * Returns the filesystem that is specially setup for 
131    * doing reads from storage. This object avoids doing 
132    * checksum verifications for reads.
133    * @return The FileSystem object that can be used to read data
134    *         from files.
135    */
136   public FileSystem getNoChecksumFs() {
137     return noChecksumFs;
138   }
139 
140   /**
141    * Returns the underlying filesystem
142    * @return The underlying FileSystem for this FilterFileSystem object.
143    */
144   public FileSystem getBackingFs() throws IOException {
145     return fs;
146   }
147 
148   /**
149    * Are we verifying checksums in HBase?
150    * @return True, if hbase is configured to verify checksums,
151    *         otherwise false.
152    */
153   public boolean useHBaseChecksum() {
154     return useHBaseChecksum;
155   }
156 
157   /**
158    * Close this filesystem object
159    */
160   @Override
161   public void close() throws IOException {
162     super.close();
163     if (this.noChecksumFs != fs) {
164       this.noChecksumFs.close();
165     }
166   }
167 
168  /**
169    * Returns a brand new instance of the FileSystem. It does not use
170    * the FileSystem.Cache. In newer versions of HDFS, we can directly
171    * invoke FileSystem.newInstance(Configuration).
172    * 
173    * @param conf Configuration
174    * @return A new instance of the filesystem
175    */
176   private static FileSystem newInstanceFileSystem(Configuration conf)
177     throws IOException {
178     URI uri = FileSystem.getDefaultUri(conf);
179     FileSystem fs = null;
180     Class<?> clazz = conf.getClass("fs." + uri.getScheme() + ".impl", null);
181     if (clazz != null) {
182       // This will be true for Hadoop 1.0, or 0.20.
183       fs = (FileSystem)ReflectionUtils.newInstance(clazz, conf);
184       fs.initialize(uri, conf);
185     } else {
186       // For Hadoop 2.0, we have to go through FileSystem for the filesystem
187       // implementation to be loaded by the service loader in case it has not
188       // been loaded yet.
189       Configuration clone = new Configuration(conf);
190       clone.setBoolean("fs." + uri.getScheme() + ".impl.disable.cache", true);
191       fs = FileSystem.get(uri, clone);
192     }
193     if (fs == null) {
194       throw new IOException("No FileSystem for scheme: " + uri.getScheme());
195     }
196 
197     return fs;
198   }
199 
200   /**
201    * Returns an instance of Filesystem wrapped into the class specified in
202    * hbase.fs.wrapper property, if one is set in the configuration, returns
203    * unmodified FS instance passed in as an argument otherwise.
204    * @param base Filesystem instance to wrap
205    * @param conf Configuration
206    * @return wrapped instance of FS, or the same instance if no wrapping configured.
207    */
208   private FileSystem maybeWrapFileSystem(FileSystem base, Configuration conf) {
209     try {
210       Class<?> clazz = conf.getClass("hbase.fs.wrapper", null);
211       if (clazz != null) {
212         return (FileSystem) clazz.getConstructor(FileSystem.class, Configuration.class)
213           .newInstance(base, conf);
214       }
215     } catch (Exception e) {
216       LOG.error("Failed to wrap filesystem: " + e);
217     }
218     return base;
219   }
220 
221   public static boolean addLocationsOrderInterceptor(Configuration conf) throws IOException {
222     return addLocationsOrderInterceptor(conf, new ReorderWALBlocks());
223   }
224 
225   /**
226    * Add an interceptor on the calls to the namenode#getBlockLocations from the DFSClient
227    * linked to this FileSystem. See HBASE-6435 for the background.
228    * <p/>
229    * There should be no reason, except testing, to create a specific ReorderBlocks.
230    *
231    * @return true if the interceptor was added, false otherwise.
232    */
233   static boolean addLocationsOrderInterceptor(Configuration conf, final ReorderBlocks lrb) {
234     if (!conf.getBoolean("hbase.filesystem.reorder.blocks", true)) {  // activated by default
235       LOG.debug("addLocationsOrderInterceptor configured to false");
236       return false;
237     }
238 
239     FileSystem fs;
240     try {
241       fs = FileSystem.get(conf);
242     } catch (IOException e) {
243       LOG.warn("Can't get the file system from the conf.", e);
244       return false;
245     }
246 
247     if (!(fs instanceof DistributedFileSystem)) {
248       LOG.debug("The file system is not a DistributedFileSystem. " +
249           "Skipping on block location reordering");
250       return false;
251     }
252 
253     DistributedFileSystem dfs = (DistributedFileSystem) fs;
254     DFSClient dfsc = dfs.getClient();
255     if (dfsc == null) {
256       LOG.warn("The DistributedFileSystem does not contain a DFSClient. Can't add the location " +
257           "block reordering interceptor. Continuing, but this is unexpected."
258       );
259       return false;
260     }
261 
262     try {
263       Field nf = DFSClient.class.getDeclaredField("namenode");
264       nf.setAccessible(true);
265       Field modifiersField = Field.class.getDeclaredField("modifiers");
266       modifiersField.setAccessible(true);
267       modifiersField.setInt(nf, nf.getModifiers() & ~Modifier.FINAL);
268 
269       ClientProtocol namenode = (ClientProtocol) nf.get(dfsc);
270       if (namenode == null) {
271         LOG.warn("The DFSClient is not linked to a namenode. Can't add the location block" +
272             " reordering interceptor. Continuing, but this is unexpected."
273         );
274         return false;
275       }
276 
277       ClientProtocol cp1 = createReorderingProxy(namenode, lrb, conf);
278       nf.set(dfsc, cp1);
279       LOG.info("Added intercepting call to namenode#getBlockLocations so can do block reordering" +
280         " using class " + lrb.getClass().getName());
281     } catch (NoSuchFieldException e) {
282       LOG.warn("Can't modify the DFSClient#namenode field to add the location reorder.", e);
283       return false;
284     } catch (IllegalAccessException e) {
285       LOG.warn("Can't modify the DFSClient#namenode field to add the location reorder.", e);
286       return false;
287     }
288 
289     return true;
290   }
291 
292   private static ClientProtocol createReorderingProxy(final ClientProtocol cp,
293       final ReorderBlocks lrb, final Configuration conf) {
294     return (ClientProtocol) Proxy.newProxyInstance
295         (cp.getClass().getClassLoader(),
296             new Class[]{ClientProtocol.class, Closeable.class},
297             new InvocationHandler() {
298               public Object invoke(Object proxy, Method method,
299                                    Object[] args) throws Throwable {
300                 try {
301                   if ((args == null || args.length == 0)
302                       && "close".equals(method.getName())) {
303                     RPC.stopProxy(cp);
304                     return null;
305                   } else {
306                     Object res = method.invoke(cp, args);
307                     if (res != null && args != null && args.length == 3
308                         && "getBlockLocations".equals(method.getName())
309                         && res instanceof LocatedBlocks
310                         && args[0] instanceof String
311                         && args[0] != null) {
312                       lrb.reorderBlocks(conf, (LocatedBlocks) res, (String) args[0]);
313                     }
314                     return res;
315                   }
316                 } catch  (InvocationTargetException ite) {
317                   // We will have this for all the exception, checked on not, sent
318                   //  by any layer, including the functional exception
319                   Throwable cause = ite.getCause();
320                   if (cause == null){
321                     throw new RuntimeException(
322                       "Proxy invocation failed and getCause is null", ite);
323                   }
324                   if (cause instanceof UndeclaredThrowableException) {
325                     Throwable causeCause = cause.getCause();
326                     if (causeCause == null) {
327                       throw new RuntimeException("UndeclaredThrowableException had null cause!");
328                     }
329                     cause = cause.getCause();
330                   }
331                   throw cause;
332                 }
333               }
334             });
335   }
336 
337   /**
338    * Interface to implement to add a specific reordering logic in hdfs.
339    */
340   interface ReorderBlocks {
341     /**
342      *
343      * @param conf - the conf to use
344      * @param lbs - the LocatedBlocks to reorder
345      * @param src - the file name currently read
346      * @throws IOException - if something went wrong
347      */
348     void reorderBlocks(Configuration conf, LocatedBlocks lbs, String src) throws IOException;
349   }
350 
351   /**
352    * We're putting at lowest priority the wal files blocks that are on the same datanode
353    * as the original regionserver which created these files. This because we fear that the
354    * datanode is actually dead, so if we use it it will timeout.
355    */
356   static class ReorderWALBlocks implements ReorderBlocks {
357     public void reorderBlocks(Configuration conf, LocatedBlocks lbs, String src)
358         throws IOException {
359 
360       ServerName sn = DefaultWALProvider.getServerNameFromWALDirectoryName(conf, src);
361       if (sn == null) {
362         // It's not an WAL
363         return;
364       }
365 
366       // Ok, so it's an WAL
367       String hostName = sn.getHostname();
368       if (LOG.isTraceEnabled()) {
369         LOG.trace(src +
370             " is an WAL file, so reordering blocks, last hostname will be:" + hostName);
371       }
372 
373       // Just check for all blocks
374       for (LocatedBlock lb : lbs.getLocatedBlocks()) {
375         DatanodeInfo[] dnis = lb.getLocations();
376         if (dnis != null && dnis.length > 1) {
377           boolean found = false;
378           for (int i = 0; i < dnis.length - 1 && !found; i++) {
379             if (hostName.equals(dnis[i].getHostName())) {
380               // advance the other locations by one and put this one at the last place.
381               DatanodeInfo toLast = dnis[i];
382               System.arraycopy(dnis, i + 1, dnis, i, dnis.length - i - 1);
383               dnis[dnis.length - 1] = toLast;
384               found = true;
385             }
386           }
387         }
388       }
389     }
390   }
391 
392   /**
393    * Create a new HFileSystem object, similar to FileSystem.get().
394    * This returns a filesystem object that avoids checksum
395    * verification in the filesystem for hfileblock-reads.
396    * For these blocks, checksum verification is done by HBase.
397    */
398   static public FileSystem get(Configuration conf) throws IOException {
399     return new HFileSystem(conf, true);
400   }
401 
402   /**
403    * Wrap a LocalFileSystem within a HFileSystem.
404    */
405   static public FileSystem getLocalFs(Configuration conf) throws IOException {
406     return new HFileSystem(FileSystem.getLocal(conf));
407   }
408 
409   /**
410    * The org.apache.hadoop.fs.FilterFileSystem does not yet support 
411    * createNonRecursive. This is a hadoop bug and when it is fixed in Hadoop,
412    * this definition will go away.
413    */
414   @SuppressWarnings("deprecation")
415   public FSDataOutputStream createNonRecursive(Path f,
416       boolean overwrite,
417       int bufferSize, short replication, long blockSize,
418       Progressable progress) throws IOException {
419     return fs.createNonRecursive(f, overwrite, bufferSize, replication,
420                                  blockSize, progress);
421   }
422 }