1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
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
59
60
61
62 public class HFileSystem extends FilterFileSystem {
63 public static final Log LOG = LogFactory.getLog(HFileSystem.class);
64
65
66 public static final String HBASE_WAL_DIR = "hbase.wal.dir";
67
68 private final FileSystem noChecksumFs;
69 private final boolean useHBaseChecksum;
70
71
72
73
74
75
76
77
78 public HFileSystem(Configuration conf, boolean useHBaseChecksum)
79 throws IOException {
80
81
82
83
84 this.fs = FileSystem.get(conf);
85 this.useHBaseChecksum = useHBaseChecksum;
86
87 fs.initialize(getDefaultUri(conf), conf);
88
89
90 if (fs instanceof LocalFileSystem) {
91 fs.setWriteChecksum(false);
92 fs.setVerifyChecksum(false);
93 }
94
95 addLocationsOrderInterceptor(conf);
96
97
98
99
100
101
102
103
104
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
119
120
121
122
123 public HFileSystem(FileSystem fs) {
124 this.fs = fs;
125 this.noChecksumFs = fs;
126 this.useHBaseChecksum = false;
127 }
128
129
130
131
132
133
134
135
136 public FileSystem getNoChecksumFs() {
137 return noChecksumFs;
138 }
139
140
141
142
143
144 public FileSystem getBackingFs() throws IOException {
145 return fs;
146 }
147
148
149
150
151
152
153 public boolean useHBaseChecksum() {
154 return useHBaseChecksum;
155 }
156
157
158
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
170
171
172
173
174
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
183 fs = (FileSystem)ReflectionUtils.newInstance(clazz, conf);
184 fs.initialize(uri, conf);
185 } else {
186
187
188
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
202
203
204
205
206
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
227
228
229
230
231
232
233 static boolean addLocationsOrderInterceptor(Configuration conf, final ReorderBlocks lrb) {
234 if (!conf.getBoolean("hbase.filesystem.reorder.blocks", true)) {
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 @Override
299 public Object invoke(Object proxy, Method method,
300 Object[] args) throws Throwable {
301 try {
302 if ((args == null || args.length == 0)
303 && "close".equals(method.getName())) {
304 RPC.stopProxy(cp);
305 return null;
306 } else {
307 Object res = method.invoke(cp, args);
308 if (res != null && args != null && args.length == 3
309 && "getBlockLocations".equals(method.getName())
310 && res instanceof LocatedBlocks
311 && args[0] instanceof String
312 && args[0] != null) {
313 lrb.reorderBlocks(conf, (LocatedBlocks) res, (String) args[0]);
314 }
315 return res;
316 }
317 } catch (InvocationTargetException ite) {
318
319
320 Throwable cause = ite.getCause();
321 if (cause == null){
322 throw new RuntimeException(
323 "Proxy invocation failed and getCause is null", ite);
324 }
325 if (cause instanceof UndeclaredThrowableException) {
326 Throwable causeCause = cause.getCause();
327 if (causeCause == null) {
328 throw new RuntimeException("UndeclaredThrowableException had null cause!");
329 }
330 cause = cause.getCause();
331 }
332 throw cause;
333 }
334 }
335 });
336 }
337
338
339
340
341 interface ReorderBlocks {
342
343
344
345
346
347
348
349 void reorderBlocks(Configuration conf, LocatedBlocks lbs, String src) throws IOException;
350 }
351
352
353
354
355
356
357 static class ReorderWALBlocks implements ReorderBlocks {
358 @Override
359 public void reorderBlocks(Configuration conf, LocatedBlocks lbs, String src)
360 throws IOException {
361
362 ServerName sn = DefaultWALProvider.getServerNameFromWALDirectoryName(conf, src);
363 if (sn == null) {
364
365 return;
366 }
367
368
369 String hostName = sn.getHostname();
370 if (LOG.isTraceEnabled()) {
371 LOG.trace(src +
372 " is an WAL file, so reordering blocks, last hostname will be:" + hostName);
373 }
374
375
376 for (LocatedBlock lb : lbs.getLocatedBlocks()) {
377 DatanodeInfo[] dnis = lb.getLocations();
378 if (dnis != null && dnis.length > 1) {
379 boolean found = false;
380 for (int i = 0; i < dnis.length - 1 && !found; i++) {
381 if (hostName.equals(dnis[i].getHostName())) {
382
383 DatanodeInfo toLast = dnis[i];
384 System.arraycopy(dnis, i + 1, dnis, i, dnis.length - i - 1);
385 dnis[dnis.length - 1] = toLast;
386 found = true;
387 }
388 }
389 }
390 }
391 }
392 }
393
394
395
396
397
398
399
400 static public FileSystem get(Configuration conf) throws IOException {
401 return new HFileSystem(conf, true);
402 }
403
404
405
406
407 static public FileSystem getLocalFs(Configuration conf) throws IOException {
408 return new HFileSystem(FileSystem.getLocal(conf));
409 }
410
411
412
413
414
415
416 @SuppressWarnings("deprecation")
417 public FSDataOutputStream createNonRecursive(Path f,
418 boolean overwrite,
419 int bufferSize, short replication, long blockSize,
420 Progressable progress) throws IOException {
421 return fs.createNonRecursive(f, overwrite, bufferSize, replication,
422 blockSize, progress);
423 }
424 }