Skip to content

Commit ac4a2e1

Browse files
HDFS-15306. Make mount-table to read from central place ( Let's say from HDFS). Contributed by Uma Maheswara Rao G.
1 parent d08b9e9 commit ac4a2e1

File tree

11 files changed

+797
-133
lines changed

11 files changed

+797
-133
lines changed

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/Constants.java

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,11 @@ public interface Constants {
3030
* Prefix for the config variable prefix for the ViewFs mount-table
3131
*/
3232
public static final String CONFIG_VIEWFS_PREFIX = "fs.viewfs.mounttable";
33+
34+
/**
35+
* Prefix for the config variable for the ViewFs mount-table path.
36+
*/
37+
String CONFIG_VIEWFS_MOUNTTABLE_PATH = CONFIG_VIEWFS_PREFIX + ".path";
3338

3439
/**
3540
* Prefix for the home dir for the mount table - if not specified
Lines changed: 122 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,122 @@
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.fs.viewfs;
19+
20+
import java.io.IOException;
21+
22+
import org.apache.hadoop.conf.Configuration;
23+
import org.apache.hadoop.fs.FSDataInputStream;
24+
import org.apache.hadoop.fs.FileSystem;
25+
import org.apache.hadoop.fs.LocatedFileStatus;
26+
import org.apache.hadoop.fs.Path;
27+
import org.apache.hadoop.fs.RemoteIterator;
28+
import org.slf4j.Logger;
29+
import org.slf4j.LoggerFactory;
30+
31+
/**
32+
* An implementation for Apache Hadoop compatible file system based mount-table
33+
* file loading.
34+
*/
35+
public class HCFSMountTableConfigLoader implements MountTableConfigLoader {
36+
private static final String REGEX_DOT = "[.]";
37+
private static final Logger LOGGER =
38+
LoggerFactory.getLogger(HCFSMountTableConfigLoader.class);
39+
private Path mountTable = null;
40+
41+
/**
42+
* Loads the mount-table configuration from hadoop compatible file system and
43+
* add the configuration items to given configuration. Mount-table
44+
* configuration format should be suffixed with version number.
45+
* Format: mount-table.<versionNumber>.xml
46+
* Example: mount-table.1.xml
47+
* When user wants to update mount-table, the expectation is to upload new
48+
* mount-table configuration file with monotonically increasing integer as
49+
* version number. This API loads the highest version number file. We can
50+
* also configure single file path directly.
51+
*
52+
* @param mountTableConfigPath : A directory path where mount-table files
53+
* stored or a mount-table file path. We recommend to configure
54+
* directory with the mount-table version files.
55+
* @param conf : to add the mount table as resource.
56+
*/
57+
@Override
58+
public void load(String mountTableConfigPath, Configuration conf)
59+
throws IOException {
60+
this.mountTable = new Path(mountTableConfigPath);
61+
String scheme = mountTable.toUri().getScheme();
62+
ViewFileSystem.FsGetter fsGetter =
63+
new ViewFileSystemOverloadScheme.ChildFsGetter(scheme);
64+
try (FileSystem fs = fsGetter.getNewInstance(mountTable.toUri(), conf)) {
65+
RemoteIterator<LocatedFileStatus> listFiles =
66+
fs.listFiles(mountTable, false);
67+
LocatedFileStatus lfs = null;
68+
int higherVersion = -1;
69+
while (listFiles.hasNext()) {
70+
LocatedFileStatus curLfs = listFiles.next();
71+
String cur = curLfs.getPath().getName();
72+
String[] nameParts = cur.split(REGEX_DOT);
73+
if (nameParts.length < 2) {
74+
logInvalidFileNameFormat(cur);
75+
continue; // invalid file name
76+
}
77+
int curVersion = higherVersion;
78+
try {
79+
curVersion = Integer.parseInt(nameParts[nameParts.length - 2]);
80+
} catch (NumberFormatException nfe) {
81+
logInvalidFileNameFormat(cur);
82+
continue;
83+
}
84+
85+
if (curVersion > higherVersion) {
86+
higherVersion = curVersion;
87+
lfs = curLfs;
88+
}
89+
}
90+
91+
if (lfs == null) {
92+
// No valid mount table file found.
93+
// TODO: Should we fail? Currently viewfs init will fail if no mount
94+
// links anyway.
95+
LOGGER.warn("No valid mount-table file exist at: {}. At least one "
96+
+ "mount-table file should present with the name format: "
97+
+ "mount-table.<versionNumber>.xml", mountTableConfigPath);
98+
return;
99+
}
100+
// Latest version file.
101+
Path latestVersionMountTable = lfs.getPath();
102+
if (LOGGER.isDebugEnabled()) {
103+
LOGGER.debug("Loading the mount-table {} into configuration.",
104+
latestVersionMountTable);
105+
}
106+
try (FSDataInputStream open = fs.open(latestVersionMountTable)) {
107+
Configuration newConf = new Configuration(false);
108+
newConf.addResource(open);
109+
// This will add configuration props as resource, instead of stream
110+
// itself. So, that stream can be closed now.
111+
conf.addResource(newConf);
112+
}
113+
}
114+
}
115+
116+
private void logInvalidFileNameFormat(String cur) {
117+
LOGGER.warn("Invalid file name format for mount-table version file: {}. "
118+
+ "The valid file name format is mount-table-name.<versionNumber>.xml",
119+
cur);
120+
}
121+
122+
}
Lines changed: 44 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,44 @@
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.fs.viewfs;
19+
20+
import java.io.IOException;
21+
22+
import org.apache.hadoop.classification.InterfaceAudience;
23+
import org.apache.hadoop.classification.InterfaceStability;
24+
import org.apache.hadoop.conf.Configuration;
25+
26+
/**
27+
* An interface for loading mount-table configuration. This class can have more
28+
* APIs like refreshing mount tables automatically etc.
29+
*/
30+
@InterfaceAudience.Private
31+
@InterfaceStability.Evolving
32+
public interface MountTableConfigLoader {
33+
34+
/**
35+
* Loads the mount-table configuration into given configuration.
36+
*
37+
* @param mountTableConfigPath - Path of the mount table. It can be a file or
38+
* a directory in the case of multiple versions of mount-table
39+
* files(Recommended option).
40+
* @param conf - Configuration object to add mount table.
41+
*/
42+
void load(String mountTableConfigPath, Configuration conf)
43+
throws IOException;
44+
}

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/ViewFileSystemOverloadScheme.java

Lines changed: 103 additions & 77 deletions
Original file line numberDiff line numberDiff line change
@@ -98,107 +98,133 @@ public String getScheme() {
9898

9999
@Override
100100
public void initialize(URI theUri, Configuration conf) throws IOException {
101+
this.myUri = theUri;
101102
if (LOG.isDebugEnabled()) {
102103
LOG.debug("Initializing the ViewFileSystemOverloadScheme with the uri: "
103104
+ theUri);
104105
}
105-
this.myUri = theUri;
106+
String mountTableConfigPath =
107+
conf.get(Constants.CONFIG_VIEWFS_MOUNTTABLE_PATH);
108+
if (null != mountTableConfigPath) {
109+
MountTableConfigLoader loader = new HCFSMountTableConfigLoader();
110+
loader.load(mountTableConfigPath, conf);
111+
} else {
112+
// TODO: Should we fail here.?
113+
if (LOG.isDebugEnabled()) {
114+
LOG.debug(
115+
"Missing configuration for fs.viewfs.mounttable.path. Proceeding"
116+
+ "with core-site.xml mount-table information if avaialable.");
117+
}
118+
}
106119
super.initialize(theUri, conf);
107120
}
108121

109122
/**
110123
* This method is overridden because in ViewFileSystemOverloadScheme if
111-
* overloaded cheme matches with mounted target fs scheme, file system should
112-
* be created without going into fs.<scheme>.impl based resolution. Otherwise
113-
* it will end up in an infinite loop as the target will be resolved again
114-
* to ViewFileSystemOverloadScheme as fs.<scheme>.impl points to
115-
* ViewFileSystemOverloadScheme. So, below method will initialize the
124+
* overloaded scheme matches with mounted target fs scheme, file system
125+
* should be created without going into fs.<scheme>.impl based resolution.
126+
* Otherwise it will end up in an infinite loop as the target will be
127+
* resolved again to ViewFileSystemOverloadScheme as fs.<scheme>.impl points
128+
* to ViewFileSystemOverloadScheme. So, below method will initialize the
116129
* fs.viewfs.overload.scheme.target.<scheme>.impl. Other schemes can
117130
* follow fs.newInstance
118131
*/
119132
@Override
120133
protected FsGetter fsGetter() {
134+
return new ChildFsGetter(getScheme());
135+
}
121136

122-
return new FsGetter() {
123-
@Override
124-
public FileSystem getNewInstance(URI uri, Configuration conf)
125-
throws IOException {
126-
if (uri.getScheme().equals(getScheme())) {
127-
if (LOG.isDebugEnabled()) {
128-
LOG.debug(
129-
"The file system initialized uri scheme is matching with the "
130-
+ "given target uri scheme. The target uri is: " + uri);
131-
}
132-
/*
133-
* Avoid looping when target fs scheme is matching to overloaded
134-
* scheme.
135-
*/
136-
return createFileSystem(uri, conf);
137-
} else {
138-
return FileSystem.newInstance(uri, conf);
137+
/**
138+
* This class checks whether the rooScheme is same as URI scheme. If both are
139+
* same, then it will initialize file systems by using the configured
140+
* fs.viewfs.overload.scheme.target.<scheme>.impl class.
141+
*/
142+
static class ChildFsGetter extends FsGetter {
143+
144+
private final String rootScheme;
145+
146+
ChildFsGetter(String rootScheme) {
147+
this.rootScheme = rootScheme;
148+
}
149+
150+
@Override
151+
public FileSystem getNewInstance(URI uri, Configuration conf)
152+
throws IOException {
153+
if (uri.getScheme().equals(this.rootScheme)) {
154+
if (LOG.isDebugEnabled()) {
155+
LOG.debug(
156+
"The file system initialized uri scheme is matching with the "
157+
+ "given target uri scheme. The target uri is: " + uri);
139158
}
159+
/*
160+
* Avoid looping when target fs scheme is matching to overloaded scheme.
161+
*/
162+
return createFileSystem(uri, conf);
163+
} else {
164+
return FileSystem.newInstance(uri, conf);
140165
}
166+
}
141167

142-
/**
143-
* When ViewFileSystemOverloadScheme scheme and target uri scheme are
144-
* matching, it will not take advantage of FileSystem cache as it will
145-
* create instance directly. For caching needs please set
146-
* "fs.viewfs.enable.inner.cache" to true.
147-
*/
148-
@Override
149-
public FileSystem get(URI uri, Configuration conf) throws IOException {
150-
if (uri.getScheme().equals(getScheme())) {
151-
// Avoid looping when target fs scheme is matching to overloaded
152-
// scheme.
153-
if (LOG.isDebugEnabled()) {
154-
LOG.debug(
155-
"The file system initialized uri scheme is matching with the "
156-
+ "given target uri scheme. So, the target file system "
157-
+ "instances will not be cached. To cache fs instances, "
158-
+ "please set fs.viewfs.enable.inner.cache to true. "
159-
+ "The target uri is: " + uri);
160-
}
161-
return createFileSystem(uri, conf);
162-
} else {
163-
return FileSystem.get(uri, conf);
168+
/**
169+
* When ViewFileSystemOverloadScheme scheme and target uri scheme are
170+
* matching, it will not take advantage of FileSystem cache as it will
171+
* create instance directly. For caching needs please set
172+
* "fs.viewfs.enable.inner.cache" to true.
173+
*/
174+
@Override
175+
public FileSystem get(URI uri, Configuration conf) throws IOException {
176+
if (uri.getScheme().equals(this.rootScheme)) {
177+
// Avoid looping when target fs scheme is matching to overloaded
178+
// scheme.
179+
if (LOG.isDebugEnabled()) {
180+
LOG.debug(
181+
"The file system initialized uri scheme is matching with the "
182+
+ "given target uri scheme. So, the target file system "
183+
+ "instances will not be cached. To cache fs instances, "
184+
+ "please set fs.viewfs.enable.inner.cache to true. "
185+
+ "The target uri is: " + uri);
164186
}
187+
return createFileSystem(uri, conf);
188+
} else {
189+
return FileSystem.get(uri, conf);
165190
}
191+
}
166192

167-
private FileSystem createFileSystem(URI uri, Configuration conf)
168-
throws IOException {
169-
final String fsImplConf = String.format(
170-
FsConstants.FS_VIEWFS_OVERLOAD_SCHEME_TARGET_FS_IMPL_PATTERN,
171-
uri.getScheme());
172-
Class<?> clazz = conf.getClass(fsImplConf, null);
173-
if (clazz == null) {
174-
throw new UnsupportedFileSystemException(
175-
String.format("%s=null: %s: %s", fsImplConf,
176-
"No overload scheme fs configured", uri.getScheme()));
177-
}
178-
FileSystem fs = (FileSystem) newInstance(clazz, uri, conf);
179-
fs.initialize(uri, conf);
180-
return fs;
193+
private FileSystem createFileSystem(URI uri, Configuration conf)
194+
throws IOException {
195+
final String fsImplConf = String.format(
196+
FsConstants.FS_VIEWFS_OVERLOAD_SCHEME_TARGET_FS_IMPL_PATTERN,
197+
uri.getScheme());
198+
Class<?> clazz = conf.getClass(fsImplConf, null);
199+
if (clazz == null) {
200+
throw new UnsupportedFileSystemException(
201+
String.format("%s=null: %s: %s", fsImplConf,
202+
"No overload scheme fs configured", uri.getScheme()));
181203
}
204+
FileSystem fs = (FileSystem) newInstance(clazz, uri, conf);
205+
fs.initialize(uri, conf);
206+
return fs;
207+
}
182208

183-
private <T> T newInstance(Class<T> theClass, URI uri,
184-
Configuration conf) {
185-
T result;
186-
try {
187-
Constructor<T> meth = theClass.getConstructor();
188-
meth.setAccessible(true);
189-
result = meth.newInstance();
190-
} catch (InvocationTargetException e) {
191-
Throwable cause = e.getCause();
192-
if (cause instanceof RuntimeException) {
193-
throw (RuntimeException) cause;
194-
} else {
195-
throw new RuntimeException(cause);
196-
}
197-
} catch (Exception e) {
198-
throw new RuntimeException(e);
209+
private <T> T newInstance(Class<T> theClass, URI uri, Configuration conf) {
210+
T result;
211+
try {
212+
Constructor<T> meth = theClass.getConstructor();
213+
meth.setAccessible(true);
214+
result = meth.newInstance();
215+
} catch (InvocationTargetException e) {
216+
Throwable cause = e.getCause();
217+
if (cause instanceof RuntimeException) {
218+
throw (RuntimeException) cause;
219+
} else {
220+
throw new RuntimeException(cause);
199221
}
200-
return result;
222+
} catch (Exception e) {
223+
throw new RuntimeException(e);
201224
}
202-
};
225+
return result;
226+
}
227+
203228
}
229+
204230
}

0 commit comments

Comments
 (0)