http://git-wip-us.apache.org/repos/asf/hadoop/blob/87700d45/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreBaseImpl.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreBaseImpl.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreBaseImpl.java new file mode 100644 index 0000000..1bd35f2 --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreBaseImpl.java @@ -0,0 +1,86 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.hadoop.hdfs.server.federation.store.driver.impl; + +import static org.apache.hadoop.hdfs.server.federation.store.StateStoreUtils.filterMultiple; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; + +import org.apache.hadoop.hdfs.server.federation.store.StateStoreUtils; +import org.apache.hadoop.hdfs.server.federation.store.driver.StateStoreDriver; +import org.apache.hadoop.hdfs.server.federation.store.records.BaseRecord; +import org.apache.hadoop.hdfs.server.federation.store.records.Query; +import org.apache.hadoop.hdfs.server.federation.store.records.QueryResult; + +/** + * Base implementation of a State Store driver. It contains default + * implementations for the optional functions. These implementations use an + * uncached read/write all algorithm for all changes. In most cases it is + * recommended to override the optional functions. + * <p> + * Drivers may optionally override additional routines for performance + * optimization, such as custom get/put/remove queries, depending on the + * capabilities of the data store. + * <p> + */ +public abstract class StateStoreBaseImpl extends StateStoreDriver { + + @Override + public <T extends BaseRecord> T get( + Class<T> clazz, Query<T> query) throws IOException { + List<T> records = getMultiple(clazz, query); + if (records.size() > 1) { + throw new IOException("Found more than one object in collection"); + } else if (records.size() == 1) { + return records.get(0); + } else { + return null; + } + } + + @Override + public <T extends BaseRecord> List<T> getMultiple( + Class<T> clazz, Query<T> query) throws IOException { + QueryResult<T> result = get(clazz); + List<T> records = result.getRecords(); + List<T> ret = filterMultiple(query, records); + if (ret == null) { + throw new IOException("Cannot fetch records from the store"); + } + return ret; + } + + @Override + public <T extends BaseRecord> boolean put( + T record, boolean allowUpdate, boolean errorIfExists) throws IOException { + List<T> singletonList = new ArrayList<>(); + singletonList.add(record); + return putAll(singletonList, allowUpdate, errorIfExists); + } + + @Override + public <T extends BaseRecord> boolean remove(T record) throws IOException { + final Query<T> query = new Query<T>(record); + Class<? extends BaseRecord> clazz = record.getClass(); + @SuppressWarnings("unchecked") + Class<T> recordClass = (Class<T>)StateStoreUtils.getRecordClass(clazz); + return remove(recordClass, query) == 1; + } +} \ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/87700d45/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreFileBaseImpl.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreFileBaseImpl.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreFileBaseImpl.java new file mode 100644 index 0000000..6638d1c --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreFileBaseImpl.java @@ -0,0 +1,462 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.hadoop.hdfs.server.federation.store.driver.impl; + +import static org.apache.hadoop.hdfs.server.federation.store.StateStoreUtils.filterMultiple; +import static org.apache.hadoop.util.Time.monotonicNow; +import static org.apache.hadoop.util.Time.now; + +import java.io.BufferedReader; +import java.io.BufferedWriter; +import java.io.IOException; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Map.Entry; +import java.util.concurrent.TimeUnit; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +import org.apache.hadoop.hdfs.server.federation.metrics.StateStoreMetrics; +import org.apache.hadoop.hdfs.server.federation.store.StateStoreUnavailableException; +import org.apache.hadoop.hdfs.server.federation.store.StateStoreUtils; +import org.apache.hadoop.hdfs.server.federation.store.driver.StateStoreDriver; +import org.apache.hadoop.hdfs.server.federation.store.records.BaseRecord; +import org.apache.hadoop.hdfs.server.federation.store.records.Query; +import org.apache.hadoop.hdfs.server.federation.store.records.QueryResult; +import org.apache.hadoop.util.Time; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.google.common.annotations.VisibleForTesting; + +/** + * {@link StateStoreDriver} implementation based on files. In this approach, we + * use temporary files for the writes and renaming "atomically" to the final + * value. Instead of writing to the final location, it will go to a temporary + * one and then rename to the final destination. + */ +public abstract class StateStoreFileBaseImpl + extends StateStoreSerializableImpl { + + private static final Logger LOG = + LoggerFactory.getLogger(StateStoreFileBaseImpl.class); + + /** File extension for temporary files. */ + private static final String TMP_MARK = ".tmp"; + /** We remove temporary files older than 10 seconds. */ + private static final long OLD_TMP_RECORD_MS = TimeUnit.SECONDS.toMillis(10); + /** File pattern for temporary records: file.XYZ.tmp. */ + private static final Pattern OLD_TMP_RECORD_PATTERN = + Pattern.compile(".+\\.(\\d+)\\.tmp"); + + /** If it is initialized. */ + private boolean initialized = false; + + + /** + * Get the reader of a record for the file system. + * + * @param path Path of the record to read. + * @return Reader for the record. + */ + protected abstract <T extends BaseRecord> BufferedReader getReader( + String path); + + /** + * Get the writer of a record for the file system. + * + * @param path Path of the record to write. + * @return Writer for the record. + */ + protected abstract <T extends BaseRecord> BufferedWriter getWriter( + String path); + + /** + * Check if a path exists. + * + * @param path Path to check. + * @return If the path exists. + */ + protected abstract boolean exists(String path); + + /** + * Make a directory. + * + * @param path Path of the directory to create. + * @return If the directory was created. + */ + protected abstract boolean mkdir(String path); + + /** + * Rename a file. This should be atomic. + * + * @param src Source name. + * @param dst Destination name. + * @return If the rename was successful. + */ + protected abstract boolean rename(String src, String dst); + + /** + * Remove a file. + * + * @param path Path for the file to remove + * @return If the file was removed. + */ + protected abstract boolean remove(String path); + + /** + * Get the children for a path. + * + * @param path Path to check. + * @return List of children. + */ + protected abstract List<String> getChildren(String path); + + /** + * Get root directory. + * + * @return Root directory. + */ + protected abstract String getRootDir(); + + /** + * Set the driver as initialized. + * + * @param ini If the driver is initialized. + */ + public void setInitialized(boolean ini) { + this.initialized = ini; + } + + @Override + public boolean initDriver() { + String rootDir = getRootDir(); + try { + if (rootDir == null) { + LOG.error("Invalid root directory, unable to initialize driver."); + return false; + } + + // Check root path + if (!exists(rootDir)) { + if (!mkdir(rootDir)) { + LOG.error("Cannot create State Store root directory {}", rootDir); + return false; + } + } + } catch (Exception ex) { + LOG.error( + "Cannot initialize filesystem using root directory {}", rootDir, ex); + return false; + } + setInitialized(true); + return true; + } + + @Override + public <T extends BaseRecord> boolean initRecordStorage( + String className, Class<T> recordClass) { + + String dataDirPath = getRootDir() + "/" + className; + try { + // Create data directories for files + if (!exists(dataDirPath)) { + LOG.info("{} data directory doesn't exist, creating it", dataDirPath); + if (!mkdir(dataDirPath)) { + LOG.error("Cannot create data directory {}", dataDirPath); + return false; + } + } + } catch (Exception ex) { + LOG.error("Cannot create data directory {}", dataDirPath, ex); + return false; + } + return true; + } + + @Override + public <T extends BaseRecord> QueryResult<T> get(Class<T> clazz) + throws IOException { + verifyDriverReady(); + long start = monotonicNow(); + StateStoreMetrics metrics = getMetrics(); + List<T> ret = new ArrayList<>(); + try { + String path = getPathForClass(clazz); + List<String> children = getChildren(path); + for (String child : children) { + String pathRecord = path + "/" + child; + if (child.endsWith(TMP_MARK)) { + LOG.debug("There is a temporary file {} in {}", child, path); + if (isOldTempRecord(child)) { + LOG.warn("Removing {} as it's an old temporary record", child); + remove(pathRecord); + } + } else { + T record = getRecord(pathRecord, clazz); + ret.add(record); + } + } + } catch (Exception e) { + if (metrics != null) { + metrics.addFailure(monotonicNow() - start); + } + String msg = "Cannot fetch records for " + clazz.getSimpleName(); + LOG.error(msg, e); + throw new IOException(msg, e); + } + + if (metrics != null) { + metrics.addRead(monotonicNow() - start); + } + return new QueryResult<T>(ret, getTime()); + } + + /** + * Check if a record is temporary and old. + * + * @param pathRecord Path for the record to check. + * @return If the record is temporary and old. + */ + @VisibleForTesting + public static boolean isOldTempRecord(final String pathRecord) { + if (!pathRecord.endsWith(TMP_MARK)) { + return false; + } + // Extract temporary record creation time + Matcher m = OLD_TMP_RECORD_PATTERN.matcher(pathRecord); + if (m.find()) { + long time = Long.parseLong(m.group(1)); + return now() - time > OLD_TMP_RECORD_MS; + } + return false; + } + + /** + * Read a record from a file. + * + * @param path Path to the file containing the record. + * @param clazz Class of the record. + * @return Record read from the file. + * @throws IOException If the file cannot be read. + */ + private <T extends BaseRecord> T getRecord( + final String path, final Class<T> clazz) throws IOException { + BufferedReader reader = getReader(path); + try { + String line; + while ((line = reader.readLine()) != null) { + if (!line.startsWith("#") && line.length() > 0) { + try { + T record = newRecord(line, clazz, false); + return record; + } catch (Exception ex) { + LOG.error("Cannot parse line {} in file {}", line, path, ex); + } + } + } + } finally { + if (reader != null) { + reader.close(); + } + } + throw new IOException("Cannot read " + path + " for record " + + clazz.getSimpleName()); + } + + /** + * Get the path for a record class. + * @param clazz Class of the record. + * @return Path for this record class. + */ + private <T extends BaseRecord> String getPathForClass(final Class<T> clazz) { + String className = StateStoreUtils.getRecordName(clazz); + StringBuilder sb = new StringBuilder(); + sb.append(getRootDir()); + if (sb.charAt(sb.length() - 1) != '/') { + sb.append("/"); + } + sb.append(className); + return sb.toString(); + } + + @Override + public boolean isDriverReady() { + return this.initialized; + } + + @Override + public <T extends BaseRecord> boolean putAll( + List<T> records, boolean allowUpdate, boolean errorIfExists) + throws StateStoreUnavailableException { + verifyDriverReady(); + if (records.isEmpty()) { + return true; + } + + long start = monotonicNow(); + StateStoreMetrics metrics = getMetrics(); + + // Check if any record exists + Map<String, T> toWrite = new HashMap<>(); + for (T record : records) { + Class<? extends BaseRecord> recordClass = record.getClass(); + String path = getPathForClass(recordClass); + String primaryKey = getPrimaryKey(record); + String recordPath = path + "/" + primaryKey; + + if (exists(recordPath)) { + if (allowUpdate) { + // Update the mod time stamp. Many backends will use their + // own timestamp for the mod time. + record.setDateModified(this.getTime()); + toWrite.put(recordPath, record); + } else if (errorIfExists) { + LOG.error("Attempt to insert record {} that already exists", + recordPath); + if (metrics != null) { + metrics.addFailure(monotonicNow() - start); + } + return false; + } else { + LOG.debug("Not updating {}", record); + } + } else { + toWrite.put(recordPath, record); + } + } + + // Write the records + boolean success = true; + for (Entry<String, T> entry : toWrite.entrySet()) { + String recordPath = entry.getKey(); + String recordPathTemp = recordPath + "." + now() + TMP_MARK; + BufferedWriter writer = getWriter(recordPathTemp); + try { + T record = entry.getValue(); + String line = serializeString(record); + writer.write(line); + } catch (IOException e) { + LOG.error("Cannot write {}", recordPathTemp, e); + success = false; + } finally { + if (writer != null) { + try { + writer.close(); + } catch (IOException e) { + LOG.error("Cannot close the writer for {}", recordPathTemp); + } + } + } + // Commit + if (!rename(recordPathTemp, recordPath)) { + LOG.error("Failed committing record into {}", recordPath); + success = false; + } + } + + long end = monotonicNow(); + if (metrics != null) { + if (success) { + metrics.addWrite(end - start); + } else { + metrics.addFailure(end - start); + } + } + return success; + } + + @Override + public <T extends BaseRecord> int remove(Class<T> clazz, Query<T> query) + throws StateStoreUnavailableException { + verifyDriverReady(); + + if (query == null) { + return 0; + } + + long start = Time.monotonicNow(); + StateStoreMetrics metrics = getMetrics(); + int removed = 0; + // Get the current records + try { + final QueryResult<T> result = get(clazz); + final List<T> existingRecords = result.getRecords(); + // Write all of the existing records except those to be removed + final List<T> recordsToRemove = filterMultiple(query, existingRecords); + boolean success = true; + for (T recordToRemove : recordsToRemove) { + String path = getPathForClass(clazz); + String primaryKey = getPrimaryKey(recordToRemove); + String recordToRemovePath = path + "/" + primaryKey; + if (remove(recordToRemovePath)) { + removed++; + } else { + LOG.error("Cannot remove record {}", recordToRemovePath); + success = false; + } + } + if (!success) { + LOG.error("Cannot remove records {} query {}", clazz, query); + if (metrics != null) { + metrics.addFailure(monotonicNow() - start); + } + } + } catch (IOException e) { + LOG.error("Cannot remove records {} query {}", clazz, query, e); + if (metrics != null) { + metrics.addFailure(monotonicNow() - start); + } + } + + if (removed > 0 && metrics != null) { + metrics.addRemove(monotonicNow() - start); + } + return removed; + } + + @Override + public <T extends BaseRecord> boolean removeAll(Class<T> clazz) + throws StateStoreUnavailableException { + verifyDriverReady(); + long start = Time.monotonicNow(); + StateStoreMetrics metrics = getMetrics(); + + boolean success = true; + String path = getPathForClass(clazz); + List<String> children = getChildren(path); + for (String child : children) { + String pathRecord = path + "/" + child; + if (!remove(pathRecord)) { + success = false; + } + } + + if (metrics != null) { + long time = Time.monotonicNow() - start; + if (success) { + metrics.addRemove(time); + } else { + metrics.addFailure(time); + } + } + return success; + } +} http://git-wip-us.apache.org/repos/asf/hadoop/blob/87700d45/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreFileImpl.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreFileImpl.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreFileImpl.java new file mode 100644 index 0000000..7d9ddc6 --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreFileImpl.java @@ -0,0 +1,148 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.hadoop.hdfs.server.federation.store.driver.impl; + +import java.io.BufferedReader; +import java.io.BufferedWriter; +import java.io.File; +import java.io.FileInputStream; +import java.io.FileOutputStream; +import java.io.IOException; +import java.io.InputStreamReader; +import java.io.OutputStreamWriter; +import java.nio.charset.StandardCharsets; +import java.util.LinkedList; +import java.util.List; + +import org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys; +import org.apache.hadoop.hdfs.server.federation.store.records.BaseRecord; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.google.common.io.Files; + +/** + * StateStoreDriver implementation based on a local file. + */ +public class StateStoreFileImpl extends StateStoreFileBaseImpl { + + private static final Logger LOG = + LoggerFactory.getLogger(StateStoreFileImpl.class); + + /** Configuration keys. */ + public static final String FEDERATION_STORE_FILE_DIRECTORY = + RBFConfigKeys.FEDERATION_STORE_PREFIX + "driver.file.directory"; + + /** Root directory for the state store. */ + private String rootDirectory; + + + @Override + protected boolean exists(String path) { + File test = new File(path); + return test.exists(); + } + + @Override + protected boolean mkdir(String path) { + File dir = new File(path); + return dir.mkdirs(); + } + + @Override + protected boolean rename(String src, String dst) { + try { + Files.move(new File(src), new File(dst)); + return true; + } catch (IOException e) { + LOG.error("Cannot rename {} to {}", src, dst, e); + return false; + } + } + + @Override + protected boolean remove(String path) { + File file = new File(path); + return file.delete(); + } + + @Override + protected String getRootDir() { + if (this.rootDirectory == null) { + String dir = getConf().get(FEDERATION_STORE_FILE_DIRECTORY); + if (dir == null) { + File tempDir = Files.createTempDir(); + dir = tempDir.getAbsolutePath(); + LOG.warn("The root directory is not available, using {}", dir); + } + this.rootDirectory = dir; + } + return this.rootDirectory; + } + + @Override + protected <T extends BaseRecord> BufferedReader getReader(String filename) { + BufferedReader reader = null; + try { + LOG.debug("Loading file: {}", filename); + File file = new File(filename); + FileInputStream fis = new FileInputStream(file); + InputStreamReader isr = + new InputStreamReader(fis, StandardCharsets.UTF_8); + reader = new BufferedReader(isr); + } catch (Exception ex) { + LOG.error("Cannot open read stream for record {}", filename, ex); + } + return reader; + } + + @Override + protected <T extends BaseRecord> BufferedWriter getWriter(String filename) { + BufferedWriter writer = null; + try { + LOG.debug("Writing file: {}", filename); + File file = new File(filename); + FileOutputStream fos = new FileOutputStream(file, false); + OutputStreamWriter osw = + new OutputStreamWriter(fos, StandardCharsets.UTF_8); + writer = new BufferedWriter(osw); + } catch (IOException e) { + LOG.error("Cannot open write stream for record {}", filename, e); + } + return writer; + } + + @Override + public void close() throws Exception { + setInitialized(false); + } + + @Override + protected List<String> getChildren(String path) { + List<String> ret = new LinkedList<>(); + File dir = new File(path); + File[] files = dir.listFiles(); + if (files != null) { + for (File file : files) { + String filename = file.getName(); + ret.add(filename); + } + } + return ret; + } +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/hadoop/blob/87700d45/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreFileSystemImpl.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreFileSystemImpl.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreFileSystemImpl.java new file mode 100644 index 0000000..ad822fb --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreFileSystemImpl.java @@ -0,0 +1,178 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.hadoop.hdfs.server.federation.store.driver.impl; + +import java.io.BufferedReader; +import java.io.BufferedWriter; +import java.io.IOException; +import java.io.InputStreamReader; +import java.io.OutputStreamWriter; +import java.net.URI; +import java.nio.charset.StandardCharsets; +import java.util.LinkedList; +import java.util.List; + +import org.apache.hadoop.fs.FSDataInputStream; +import org.apache.hadoop.fs.FSDataOutputStream; +import org.apache.hadoop.fs.FileStatus; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Options; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys; +import org.apache.hadoop.hdfs.DistributedFileSystem; +import org.apache.hadoop.hdfs.server.federation.store.records.BaseRecord; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * StateStoreDriver} implementation based on a filesystem. The most common uses + * HDFS as a backend. + */ +public class StateStoreFileSystemImpl extends StateStoreFileBaseImpl { + + private static final Logger LOG = + LoggerFactory.getLogger(StateStoreFileSystemImpl.class); + + + /** Configuration keys. */ + public static final String FEDERATION_STORE_FS_PATH = + RBFConfigKeys.FEDERATION_STORE_PREFIX + "driver.fs.path"; + + /** File system to back the State Store. */ + private FileSystem fs; + /** Working path in the filesystem. */ + private String workPath; + + @Override + protected boolean exists(String path) { + try { + return fs.exists(new Path(path)); + } catch (IOException e) { + return false; + } + } + + @Override + protected boolean mkdir(String path) { + try { + return fs.mkdirs(new Path(path)); + } catch (IOException e) { + return false; + } + } + + @Override + protected boolean rename(String src, String dst) { + try { + if (fs instanceof DistributedFileSystem) { + DistributedFileSystem dfs = (DistributedFileSystem)fs; + dfs.rename(new Path(src), new Path(dst), Options.Rename.OVERWRITE); + return true; + } else { + // Replace should be atomic but not available + if (fs.exists(new Path(dst))) { + fs.delete(new Path(dst), true); + } + return fs.rename(new Path(src), new Path(dst)); + } + } catch (Exception e) { + LOG.error("Cannot rename {} to {}", src, dst, e); + return false; + } + } + + @Override + protected boolean remove(String path) { + try { + return fs.delete(new Path(path), true); + } catch (Exception e) { + LOG.error("Cannot remove {}", path, e); + return false; + } + } + + @Override + protected String getRootDir() { + if (this.workPath == null) { + String rootPath = getConf().get(FEDERATION_STORE_FS_PATH); + URI workUri; + try { + workUri = new URI(rootPath); + fs = FileSystem.get(workUri, getConf()); + } catch (Exception ex) { + return null; + } + this.workPath = rootPath; + } + return this.workPath; + } + + @Override + public void close() throws Exception { + if (fs != null) { + fs.close(); + } + } + + @Override + protected <T extends BaseRecord> BufferedReader getReader(String pathName) { + BufferedReader reader = null; + Path path = new Path(pathName); + try { + FSDataInputStream fdis = fs.open(path); + InputStreamReader isr = + new InputStreamReader(fdis, StandardCharsets.UTF_8); + reader = new BufferedReader(isr); + } catch (IOException ex) { + LOG.error("Cannot open read stream for {}", path); + } + return reader; + } + + @Override + protected <T extends BaseRecord> BufferedWriter getWriter(String pathName) { + BufferedWriter writer = null; + Path path = new Path(pathName); + try { + FSDataOutputStream fdos = fs.create(path, true); + OutputStreamWriter osw = + new OutputStreamWriter(fdos, StandardCharsets.UTF_8); + writer = new BufferedWriter(osw); + } catch (IOException ex) { + LOG.error("Cannot open write stream for {}", path); + } + return writer; + } + + @Override + protected List<String> getChildren(String pathName) { + List<String> ret = new LinkedList<>(); + Path path = new Path(workPath, pathName); + try { + FileStatus[] files = fs.listStatus(path); + for (FileStatus file : files) { + Path filePath = file.getPath(); + String fileName = filePath.getName(); + ret.add(fileName); + } + } catch (Exception e) { + LOG.error("Cannot get children for {}", pathName, e); + } + return ret; + } +} http://git-wip-us.apache.org/repos/asf/hadoop/blob/87700d45/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreSerializableImpl.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreSerializableImpl.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreSerializableImpl.java new file mode 100644 index 0000000..7bc93de --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreSerializableImpl.java @@ -0,0 +1,98 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.hadoop.hdfs.server.federation.store.driver.impl; + +import java.io.IOException; +import java.util.Collection; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hdfs.server.federation.metrics.StateStoreMetrics; +import org.apache.hadoop.hdfs.server.federation.store.driver.StateStoreSerializer; +import org.apache.hadoop.hdfs.server.federation.store.records.BaseRecord; + +/** + * State Store driver that stores a serialization of the records. The serializer + * is pluggable. + */ +public abstract class StateStoreSerializableImpl extends StateStoreBaseImpl { + + /** Mark for slashes in path names. */ + protected static final String SLASH_MARK = "0SLASH0"; + /** Mark for colon in path names. */ + protected static final String COLON_MARK = "_"; + + /** Default serializer for this driver. */ + private StateStoreSerializer serializer; + + + @Override + public boolean init(final Configuration config, final String id, + final Collection<Class<? extends BaseRecord>> records, + final StateStoreMetrics metrics) { + boolean ret = super.init(config, id, records, metrics); + + this.serializer = StateStoreSerializer.getSerializer(config); + + return ret; + } + + /** + * Serialize a record using the serializer. + * @param record Record to serialize. + * @return Byte array with the serialization of the record. + */ + protected <T extends BaseRecord> byte[] serialize(T record) { + return serializer.serialize(record); + } + + /** + * Serialize a record using the serializer. + * @param record Record to serialize. + * @return String with the serialization of the record. + */ + protected <T extends BaseRecord> String serializeString(T record) { + return serializer.serializeString(record); + } + + /** + * Creates a record from an input data string. + * @param data Serialized text of the record. + * @param clazz Record class. + * @param includeDates If dateModified and dateCreated are serialized. + * @return The created record. + * @throws IOException + */ + protected <T extends BaseRecord> T newRecord( + String data, Class<T> clazz, boolean includeDates) throws IOException { + return serializer.deserialize(data, clazz); + } + + /** + * Get the primary key for a record. If we don't want to store in folders, we + * need to remove / from the name. + * + * @param record Record to get the primary key for. + * @return Primary key for the record. + */ + protected static String getPrimaryKey(BaseRecord record) { + String primaryKey = record.getPrimaryKey(); + primaryKey = primaryKey.replaceAll("/", SLASH_MARK); + primaryKey = primaryKey.replaceAll(":", COLON_MARK); + return primaryKey; + } +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/hadoop/blob/87700d45/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreSerializerPBImpl.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreSerializerPBImpl.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreSerializerPBImpl.java new file mode 100644 index 0000000..45c5dd6 --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreSerializerPBImpl.java @@ -0,0 +1,115 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.hadoop.hdfs.server.federation.store.driver.impl; + +import java.io.IOException; + +import org.apache.commons.codec.binary.Base64; +import org.apache.commons.codec.binary.StringUtils; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hdfs.server.federation.store.driver.StateStoreSerializer; +import org.apache.hadoop.hdfs.server.federation.store.records.BaseRecord; +import org.apache.hadoop.hdfs.server.federation.store.records.impl.pb.PBRecord; +import org.apache.hadoop.util.ReflectionUtils; + +import com.google.protobuf.Message; + +/** + * Protobuf implementation of the State Store serializer. + */ +public final class StateStoreSerializerPBImpl extends StateStoreSerializer { + + private static final String PB_IMPL_PACKAGE_SUFFIX = "impl.pb"; + private static final String PB_IMPL_CLASS_SUFFIX = "PBImpl"; + + private Configuration localConf = new Configuration(); + + + private StateStoreSerializerPBImpl() { + } + + @Override + @SuppressWarnings("unchecked") + public <T> T newRecordInstance(Class<T> clazz) { + try { + String clazzPBImpl = getPBImplClassName(clazz); + Class<?> pbClazz = localConf.getClassByName(clazzPBImpl); + Object retObject = ReflectionUtils.newInstance(pbClazz, localConf); + return (T)retObject; + } catch (ClassNotFoundException e) { + throw new RuntimeException(e); + } + } + + private String getPBImplClassName(Class<?> clazz) { + String srcPackagePart = getPackageName(clazz); + String srcClassName = getClassName(clazz); + String destPackagePart = srcPackagePart + "." + PB_IMPL_PACKAGE_SUFFIX; + String destClassPart = srcClassName + PB_IMPL_CLASS_SUFFIX; + return destPackagePart + "." + destClassPart; + } + + private String getClassName(Class<?> clazz) { + String fqName = clazz.getName(); + return (fqName.substring(fqName.lastIndexOf(".") + 1, fqName.length())); + } + + private String getPackageName(Class<?> clazz) { + return clazz.getPackage().getName(); + } + + @Override + public byte[] serialize(BaseRecord record) { + byte[] byteArray64 = null; + if (record instanceof PBRecord) { + PBRecord recordPB = (PBRecord) record; + Message msg = recordPB.getProto(); + byte[] byteArray = msg.toByteArray(); + byteArray64 = Base64.encodeBase64(byteArray, false); + } + return byteArray64; + } + + @Override + public String serializeString(BaseRecord record) { + byte[] byteArray64 = serialize(record); + String base64Encoded = StringUtils.newStringUtf8(byteArray64); + return base64Encoded; + } + + @Override + public <T extends BaseRecord> T deserialize( + byte[] byteArray, Class<T> clazz) throws IOException { + + T record = newRecord(clazz); + if (record instanceof PBRecord) { + PBRecord pbRecord = (PBRecord) record; + byte[] byteArray64 = Base64.encodeBase64(byteArray, false); + String base64Encoded = StringUtils.newStringUtf8(byteArray64); + pbRecord.readInstance(base64Encoded); + } + return record; + } + + @Override + public <T extends BaseRecord> T deserialize(String data, Class<T> clazz) + throws IOException { + byte[] byteArray64 = Base64.decodeBase64(data); + return deserialize(byteArray64, clazz); + } +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/hadoop/blob/87700d45/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreZooKeeperImpl.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreZooKeeperImpl.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreZooKeeperImpl.java new file mode 100644 index 0000000..cd5372d --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/StateStoreZooKeeperImpl.java @@ -0,0 +1,328 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.hadoop.hdfs.server.federation.store.driver.impl; + +import static org.apache.hadoop.hdfs.server.federation.store.StateStoreUtils.filterMultiple; +import static org.apache.hadoop.hdfs.server.federation.store.StateStoreUtils.getRecordName; +import static org.apache.hadoop.util.curator.ZKCuratorManager.getNodePath; +import static org.apache.hadoop.util.Time.monotonicNow; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; + +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.imps.CuratorFrameworkState; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys; +import org.apache.hadoop.hdfs.server.federation.store.driver.StateStoreDriver; +import org.apache.hadoop.hdfs.server.federation.store.records.BaseRecord; +import org.apache.hadoop.hdfs.server.federation.store.records.Query; +import org.apache.hadoop.hdfs.server.federation.store.records.QueryResult; +import org.apache.hadoop.util.curator.ZKCuratorManager; +import org.apache.zookeeper.data.Stat; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * {@link StateStoreDriver} driver implementation that uses ZooKeeper as a + * backend. + * <p> + * The structure of the znodes in the ensemble is: + * PARENT_PATH + * |--- MOUNT + * |--- MEMBERSHIP + * |--- REBALANCER + * |--- ROUTERS + */ +public class StateStoreZooKeeperImpl extends StateStoreSerializableImpl { + + private static final Logger LOG = + LoggerFactory.getLogger(StateStoreZooKeeperImpl.class); + + + /** Configuration keys. */ + public static final String FEDERATION_STORE_ZK_DRIVER_PREFIX = + RBFConfigKeys.FEDERATION_STORE_PREFIX + "driver.zk."; + public static final String FEDERATION_STORE_ZK_PARENT_PATH = + FEDERATION_STORE_ZK_DRIVER_PREFIX + "parent-path"; + public static final String FEDERATION_STORE_ZK_PARENT_PATH_DEFAULT = + "/hdfs-federation"; + + + /** Directory to store the state store data. */ + private String baseZNode; + + /** Interface to ZooKeeper. */ + private ZKCuratorManager zkManager; + + + @Override + public boolean initDriver() { + LOG.info("Initializing ZooKeeper connection"); + + Configuration conf = getConf(); + baseZNode = conf.get( + FEDERATION_STORE_ZK_PARENT_PATH, + FEDERATION_STORE_ZK_PARENT_PATH_DEFAULT); + try { + this.zkManager = new ZKCuratorManager(conf); + this.zkManager.start(); + } catch (IOException e) { + LOG.error("Cannot initialize the ZK connection", e); + return false; + } + return true; + } + + @Override + public <T extends BaseRecord> boolean initRecordStorage( + String className, Class<T> clazz) { + try { + String checkPath = getNodePath(baseZNode, className); + zkManager.createRootDirRecursively(checkPath); + return true; + } catch (Exception e) { + LOG.error("Cannot initialize ZK node for {}: {}", + className, e.getMessage()); + return false; + } + } + + @Override + public void close() throws Exception { + if (zkManager != null) { + zkManager.close(); + } + } + + @Override + public boolean isDriverReady() { + if (zkManager == null) { + return false; + } + CuratorFramework curator = zkManager.getCurator(); + if (curator == null) { + return false; + } + return curator.getState() == CuratorFrameworkState.STARTED; + } + + @Override + public <T extends BaseRecord> QueryResult<T> get(Class<T> clazz) + throws IOException { + verifyDriverReady(); + long start = monotonicNow(); + List<T> ret = new ArrayList<>(); + String znode = getZNodeForClass(clazz); + try { + List<String> children = zkManager.getChildren(znode); + for (String child : children) { + try { + String path = getNodePath(znode, child); + Stat stat = new Stat(); + String data = zkManager.getStringData(path, stat); + boolean corrupted = false; + if (data == null || data.equals("")) { + // All records should have data, otherwise this is corrupted + corrupted = true; + } else { + try { + T record = createRecord(data, stat, clazz); + ret.add(record); + } catch (IOException e) { + LOG.error("Cannot create record type \"{}\" from \"{}\": {}", + clazz.getSimpleName(), data, e.getMessage()); + corrupted = true; + } + } + + if (corrupted) { + LOG.error("Cannot get data for {} at {}, cleaning corrupted data", + child, path); + zkManager.delete(path); + } + } catch (Exception e) { + LOG.error("Cannot get data for {}: {}", child, e.getMessage()); + } + } + } catch (Exception e) { + getMetrics().addFailure(monotonicNow() - start); + String msg = "Cannot get children for \"" + znode + "\": " + + e.getMessage(); + LOG.error(msg); + throw new IOException(msg); + } + long end = monotonicNow(); + getMetrics().addRead(end - start); + return new QueryResult<T>(ret, getTime()); + } + + @Override + public <T extends BaseRecord> boolean putAll( + List<T> records, boolean update, boolean error) throws IOException { + verifyDriverReady(); + if (records.isEmpty()) { + return true; + } + + // All records should be the same + T record0 = records.get(0); + Class<? extends BaseRecord> recordClass = record0.getClass(); + String znode = getZNodeForClass(recordClass); + + long start = monotonicNow(); + boolean status = true; + for (T record : records) { + String primaryKey = getPrimaryKey(record); + String recordZNode = getNodePath(znode, primaryKey); + byte[] data = serialize(record); + if (!writeNode(recordZNode, data, update, error)){ + status = false; + } + } + long end = monotonicNow(); + if (status) { + getMetrics().addWrite(end - start); + } else { + getMetrics().addFailure(end - start); + } + return status; + } + + @Override + public <T extends BaseRecord> int remove( + Class<T> clazz, Query<T> query) throws IOException { + verifyDriverReady(); + if (query == null) { + return 0; + } + + // Read the current data + long start = monotonicNow(); + List<T> records = null; + try { + QueryResult<T> result = get(clazz); + records = result.getRecords(); + } catch (IOException ex) { + LOG.error("Cannot get existing records", ex); + getMetrics().addFailure(monotonicNow() - start); + return 0; + } + + // Check the records to remove + String znode = getZNodeForClass(clazz); + List<T> recordsToRemove = filterMultiple(query, records); + + // Remove the records + int removed = 0; + for (T existingRecord : recordsToRemove) { + LOG.info("Removing \"{}\"", existingRecord); + try { + String primaryKey = getPrimaryKey(existingRecord); + String path = getNodePath(znode, primaryKey); + if (zkManager.delete(path)) { + removed++; + } else { + LOG.error("Did not remove \"{}\"", existingRecord); + } + } catch (Exception e) { + LOG.error("Cannot remove \"{}\"", existingRecord, e); + getMetrics().addFailure(monotonicNow() - start); + } + } + long end = monotonicNow(); + if (removed > 0) { + getMetrics().addRemove(end - start); + } + return removed; + } + + @Override + public <T extends BaseRecord> boolean removeAll(Class<T> clazz) + throws IOException { + long start = monotonicNow(); + boolean status = true; + String znode = getZNodeForClass(clazz); + LOG.info("Deleting all children under {}", znode); + try { + List<String> children = zkManager.getChildren(znode); + for (String child : children) { + String path = getNodePath(znode, child); + LOG.info("Deleting {}", path); + zkManager.delete(path); + } + } catch (Exception e) { + LOG.error("Cannot remove {}: {}", znode, e.getMessage()); + status = false; + } + long time = monotonicNow() - start; + if (status) { + getMetrics().addRemove(time); + } else { + getMetrics().addFailure(time); + } + return status; + } + + private boolean writeNode( + String znode, byte[] bytes, boolean update, boolean error) { + try { + boolean created = zkManager.create(znode); + if (!update && !created && error) { + LOG.info("Cannot write record \"{}\", it already exists", znode); + return false; + } + + // Write data + zkManager.setData(znode, bytes, -1); + return true; + } catch (Exception e) { + LOG.error("Cannot write record \"{}\": {}", znode, e.getMessage()); + } + return false; + } + + /** + * Get the ZNode for a class. + * + * @param clazz Record class to evaluate. + * @return The ZNode for the class. + */ + private <T extends BaseRecord> String getZNodeForClass(Class<T> clazz) { + String className = getRecordName(clazz); + return getNodePath(baseZNode, className); + } + + /** + * Creates a record from a string returned by ZooKeeper. + * + * @param data The data to write. + * @param stat Stat of the data record to create. + * @param clazz The data record type to create. + * @return The created record. + * @throws IOException + */ + private <T extends BaseRecord> T createRecord( + String data, Stat stat, Class<T> clazz) throws IOException { + T record = newRecord(data, clazz, false); + record.setDateCreated(stat.getCtime()); + record.setDateModified(stat.getMtime()); + return record; + } +} http://git-wip-us.apache.org/repos/asf/hadoop/blob/87700d45/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/package-info.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/package-info.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/package-info.java new file mode 100644 index 0000000..a18433e --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/impl/package-info.java @@ -0,0 +1,39 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +/** + * Implementations of state store data providers/drivers. Each driver is + * responsible for maintaining, querying, updating and deleting persistent data + * records. Data records are defined as subclasses of + * {@link org.apache.hadoop.hdfs.server.federation.store.records.BaseRecord}. + * Each driver implements the + * {@link org.apache.hadoop.hdfs.server.federation.store.driver.StateStoreDriver + * StateStoreDriver} interface. + * <p> + * Currently supported drivers: + * <ul> + * <li>{@link StateStoreFileImpl} A file-based data storage backend. + * <li>{@link StateStoreZooKeeperImpl} A zookeeper based data storage backend. + * </ul> + */ +@InterfaceAudience.Private +@InterfaceStability.Evolving +package org.apache.hadoop.hdfs.server.federation.store.driver.impl; + +import org.apache.hadoop.classification.InterfaceAudience; +import org.apache.hadoop.classification.InterfaceStability; \ No newline at end of file http://git-wip-us.apache.org/repos/asf/hadoop/blob/87700d45/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/package-info.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/package-info.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/package-info.java new file mode 100644 index 0000000..da998b5 --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/driver/package-info.java @@ -0,0 +1,37 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +/** + * The state store uses modular data storage classes derived from + * StateStoreDriver to handle querying, updating and deleting data records. The + * data storage driver is initialized and maintained by the StateStoreService. + * The state store supports fetching all records of a type, filtering by column + * values or fetching a single record by its primary key. + * <p> + * Each data storage backend is required to implement the methods contained in + * the StateStoreDriver interface. These methods allow the querying, updating, + * inserting and deleting of data records into the state store. + */ + +@InterfaceAudience.Private +@InterfaceStability.Evolving + +package org.apache.hadoop.hdfs.server.federation.store.driver; + +import org.apache.hadoop.classification.InterfaceAudience; +import org.apache.hadoop.classification.InterfaceStability; \ No newline at end of file http://git-wip-us.apache.org/repos/asf/hadoop/blob/87700d45/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/impl/MembershipStoreImpl.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/impl/MembershipStoreImpl.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/impl/MembershipStoreImpl.java new file mode 100644 index 0000000..57b7b61 --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/impl/MembershipStoreImpl.java @@ -0,0 +1,310 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.hadoop.hdfs.server.federation.store.impl; + +import static org.apache.hadoop.hdfs.server.federation.store.StateStoreUtils.filterMultiple; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.LinkedList; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.TreeSet; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReadWriteLock; +import java.util.concurrent.locks.ReentrantReadWriteLock; + +import org.apache.hadoop.classification.InterfaceAudience; +import org.apache.hadoop.classification.InterfaceStability; +import org.apache.hadoop.hdfs.server.federation.resolver.FederationNamenodeServiceState; +import org.apache.hadoop.hdfs.server.federation.resolver.FederationNamespaceInfo; +import org.apache.hadoop.hdfs.server.federation.store.MembershipStore; +import org.apache.hadoop.hdfs.server.federation.store.StateStoreCache; +import org.apache.hadoop.hdfs.server.federation.store.driver.StateStoreDriver; +import org.apache.hadoop.hdfs.server.federation.store.protocol.GetNamenodeRegistrationsRequest; +import org.apache.hadoop.hdfs.server.federation.store.protocol.GetNamenodeRegistrationsResponse; +import org.apache.hadoop.hdfs.server.federation.store.protocol.GetNamespaceInfoRequest; +import org.apache.hadoop.hdfs.server.federation.store.protocol.GetNamespaceInfoResponse; +import org.apache.hadoop.hdfs.server.federation.store.protocol.NamenodeHeartbeatRequest; +import org.apache.hadoop.hdfs.server.federation.store.protocol.NamenodeHeartbeatResponse; +import org.apache.hadoop.hdfs.server.federation.store.protocol.UpdateNamenodeRegistrationRequest; +import org.apache.hadoop.hdfs.server.federation.store.protocol.UpdateNamenodeRegistrationResponse; +import org.apache.hadoop.hdfs.server.federation.store.records.MembershipState; +import org.apache.hadoop.hdfs.server.federation.store.records.Query; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Implementation of the {@link MembershipStore} State Store API. + */ +@InterfaceAudience.Private +@InterfaceStability.Evolving +public class MembershipStoreImpl + extends MembershipStore implements StateStoreCache { + + private static final Logger LOG = + LoggerFactory.getLogger(MembershipStoreImpl.class); + + + /** Reported namespaces that are not decommissioned. */ + private final Set<FederationNamespaceInfo> activeNamespaces; + + /** Namenodes (after evaluating the quorum) that are active in the cluster. */ + private final Map<String, MembershipState> activeRegistrations; + /** Namenode status reports (raw) that were discarded for being too old. */ + private final Map<String, MembershipState> expiredRegistrations; + + /** Lock to access the local memory cache. */ + private final ReadWriteLock cacheReadWriteLock = + new ReentrantReadWriteLock(); + private final Lock cacheReadLock = cacheReadWriteLock.readLock(); + private final Lock cacheWriteLock = cacheReadWriteLock.writeLock(); + + + public MembershipStoreImpl(StateStoreDriver driver) { + super(driver); + + this.activeRegistrations = new HashMap<>(); + this.expiredRegistrations = new HashMap<>(); + this.activeNamespaces = new TreeSet<>(); + } + + @Override + public GetNamenodeRegistrationsResponse getExpiredNamenodeRegistrations( + GetNamenodeRegistrationsRequest request) throws IOException { + + GetNamenodeRegistrationsResponse response = + GetNamenodeRegistrationsResponse.newInstance(); + cacheReadLock.lock(); + try { + Collection<MembershipState> vals = this.expiredRegistrations.values(); + List<MembershipState> copyVals = new ArrayList<>(vals); + response.setNamenodeMemberships(copyVals); + } finally { + cacheReadLock.unlock(); + } + return response; + } + + @Override + public GetNamespaceInfoResponse getNamespaceInfo( + GetNamespaceInfoRequest request) throws IOException { + + Set<FederationNamespaceInfo> namespaces = new HashSet<>(); + try { + cacheReadLock.lock(); + namespaces.addAll(activeNamespaces); + } finally { + cacheReadLock.unlock(); + } + + GetNamespaceInfoResponse response = + GetNamespaceInfoResponse.newInstance(namespaces); + return response; + } + + @Override + public GetNamenodeRegistrationsResponse getNamenodeRegistrations( + final GetNamenodeRegistrationsRequest request) throws IOException { + + // TODO Cache some common queries and sorts + List<MembershipState> ret = null; + + cacheReadLock.lock(); + try { + Collection<MembershipState> registrations = activeRegistrations.values(); + MembershipState partialMembership = request.getPartialMembership(); + if (partialMembership == null) { + ret = new ArrayList<>(registrations); + } else { + Query<MembershipState> query = new Query<>(partialMembership); + ret = filterMultiple(query, registrations); + } + } finally { + cacheReadLock.unlock(); + } + // Sort in ascending update date order + Collections.sort(ret); + + GetNamenodeRegistrationsResponse response = + GetNamenodeRegistrationsResponse.newInstance(ret); + return response; + } + + @Override + public NamenodeHeartbeatResponse namenodeHeartbeat( + NamenodeHeartbeatRequest request) throws IOException { + + MembershipState record = request.getNamenodeMembership(); + String nnId = record.getNamenodeKey(); + MembershipState existingEntry = null; + cacheReadLock.lock(); + try { + existingEntry = this.activeRegistrations.get(nnId); + } finally { + cacheReadLock.unlock(); + } + + if (existingEntry != null) { + if (existingEntry.getState() != record.getState()) { + LOG.info("NN registration state has changed: {} -> {}", + existingEntry, record); + } else { + LOG.debug("Updating NN registration: {} -> {}", existingEntry, record); + } + } else { + LOG.info("Inserting new NN registration: {}", record); + } + + boolean status = getDriver().put(record, true, false); + + NamenodeHeartbeatResponse response = + NamenodeHeartbeatResponse.newInstance(status); + return response; + } + + @Override + public boolean loadCache(boolean force) throws IOException { + super.loadCache(force); + + // Update local cache atomically + cacheWriteLock.lock(); + try { + this.activeRegistrations.clear(); + this.expiredRegistrations.clear(); + this.activeNamespaces.clear(); + + // Build list of NN registrations: nnId -> registration list + Map<String, List<MembershipState>> nnRegistrations = new HashMap<>(); + List<MembershipState> cachedRecords = getCachedRecords(); + for (MembershipState membership : cachedRecords) { + String nnId = membership.getNamenodeKey(); + if (membership.getState() == FederationNamenodeServiceState.EXPIRED) { + // Expired, RPC service does not use these + String key = membership.getPrimaryKey(); + this.expiredRegistrations.put(key, membership); + } else { + // This is a valid NN registration, build a list of all registrations + // using the NN id to use for the quorum calculation. + List<MembershipState> nnRegistration = + nnRegistrations.get(nnId); + if (nnRegistration == null) { + nnRegistration = new LinkedList<>(); + nnRegistrations.put(nnId, nnRegistration); + } + nnRegistration.add(membership); + String bpId = membership.getBlockPoolId(); + String cId = membership.getClusterId(); + String nsId = membership.getNameserviceId(); + FederationNamespaceInfo nsInfo = + new FederationNamespaceInfo(bpId, cId, nsId); + this.activeNamespaces.add(nsInfo); + } + } + + // Calculate most representative entry for each active NN id + for (List<MembershipState> nnRegistration : nnRegistrations.values()) { + // Run quorum based on NN state + MembershipState representativeRecord = + getRepresentativeQuorum(nnRegistration); + String nnKey = representativeRecord.getNamenodeKey(); + this.activeRegistrations.put(nnKey, representativeRecord); + } + LOG.debug("Refreshed {} NN registrations from State Store", + cachedRecords.size()); + } finally { + cacheWriteLock.unlock(); + } + return true; + } + + @Override + public UpdateNamenodeRegistrationResponse updateNamenodeRegistration( + UpdateNamenodeRegistrationRequest request) throws IOException { + + boolean status = false; + cacheWriteLock.lock(); + try { + String namenode = MembershipState.getNamenodeKey( + request.getNameserviceId(), request.getNamenodeId()); + MembershipState member = this.activeRegistrations.get(namenode); + if (member != null) { + member.setState(request.getState()); + status = true; + } + } finally { + cacheWriteLock.unlock(); + } + UpdateNamenodeRegistrationResponse response = + UpdateNamenodeRegistrationResponse.newInstance(status); + return response; + } + + /** + * Picks the most recent entry in the subset that is most agreeable on the + * specified field. 1) If a majority of the collection has the same value for + * the field, the first sorted entry within the subset the matches the + * majority value 2) Otherwise the first sorted entry in the set of all + * entries + * + * @param records - Collection of state store record objects of the same type + * @return record that is most representative of the field name + */ + private MembershipState getRepresentativeQuorum( + Collection<MembershipState> records) { + + // Collate objects by field value: field value -> order set of records + Map<FederationNamenodeServiceState, TreeSet<MembershipState>> occurenceMap = + new HashMap<>(); + for (MembershipState record : records) { + FederationNamenodeServiceState state = record.getState(); + TreeSet<MembershipState> matchingSet = occurenceMap.get(state); + if (matchingSet == null) { + // TreeSet orders elements by descending date via comparators + matchingSet = new TreeSet<>(); + occurenceMap.put(state, matchingSet); + } + matchingSet.add(record); + } + + // Select largest group + TreeSet<MembershipState> largestSet = new TreeSet<>(); + for (TreeSet<MembershipState> matchingSet : occurenceMap.values()) { + if (largestSet.size() < matchingSet.size()) { + largestSet = matchingSet; + } + } + + // If quorum, use the newest element here + if (largestSet.size() > records.size() / 2) { + return largestSet.first(); + // Otherwise, return most recent by class comparator + } else if (records.size() > 0) { + TreeSet<MembershipState> sortedList = new TreeSet<>(records); + LOG.debug("Quorum failed, using most recent: {}", sortedList.first()); + return sortedList.first(); + } else { + return null; + } + } +} http://git-wip-us.apache.org/repos/asf/hadoop/blob/87700d45/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/impl/MountTableStoreImpl.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/impl/MountTableStoreImpl.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/impl/MountTableStoreImpl.java new file mode 100644 index 0000000..eb117d6 --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/impl/MountTableStoreImpl.java @@ -0,0 +1,154 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.hadoop.hdfs.server.federation.store.impl; + +import java.io.IOException; +import java.util.Collections; +import java.util.Iterator; +import java.util.List; + +import org.apache.hadoop.classification.InterfaceAudience; +import org.apache.hadoop.classification.InterfaceStability; +import org.apache.hadoop.fs.permission.FsAction; +import org.apache.hadoop.hdfs.server.federation.router.RouterAdminServer; +import org.apache.hadoop.hdfs.server.federation.router.RouterPermissionChecker; +import org.apache.hadoop.hdfs.server.federation.store.MountTableStore; +import org.apache.hadoop.hdfs.server.federation.store.driver.StateStoreDriver; +import org.apache.hadoop.hdfs.server.federation.store.protocol.AddMountTableEntryRequest; +import org.apache.hadoop.hdfs.server.federation.store.protocol.AddMountTableEntryResponse; +import org.apache.hadoop.hdfs.server.federation.store.protocol.GetMountTableEntriesRequest; +import org.apache.hadoop.hdfs.server.federation.store.protocol.GetMountTableEntriesResponse; +import org.apache.hadoop.hdfs.server.federation.store.protocol.RemoveMountTableEntryRequest; +import org.apache.hadoop.hdfs.server.federation.store.protocol.RemoveMountTableEntryResponse; +import org.apache.hadoop.hdfs.server.federation.store.protocol.UpdateMountTableEntryRequest; +import org.apache.hadoop.hdfs.server.federation.store.protocol.UpdateMountTableEntryResponse; +import org.apache.hadoop.hdfs.server.federation.store.records.MountTable; +import org.apache.hadoop.hdfs.server.federation.store.records.Query; +import org.apache.hadoop.security.AccessControlException; +import org.apache.hadoop.util.Time; + +/** + * Implementation of the {@link MountTableStore} state store API. + */ +@InterfaceAudience.Private +@InterfaceStability.Evolving +public class MountTableStoreImpl extends MountTableStore { + + public MountTableStoreImpl(StateStoreDriver driver) { + super(driver); + } + + @Override + public AddMountTableEntryResponse addMountTableEntry( + AddMountTableEntryRequest request) throws IOException { + MountTable mountTable = request.getEntry(); + if (mountTable != null) { + RouterPermissionChecker pc = RouterAdminServer.getPermissionChecker(); + if (pc != null) { + pc.checkPermission(mountTable, FsAction.WRITE); + } + } + + boolean status = getDriver().put(mountTable, false, true); + AddMountTableEntryResponse response = + AddMountTableEntryResponse.newInstance(); + response.setStatus(status); + return response; + } + + @Override + public UpdateMountTableEntryResponse updateMountTableEntry( + UpdateMountTableEntryRequest request) throws IOException { + MountTable mountTable = request.getEntry(); + if (mountTable != null) { + RouterPermissionChecker pc = RouterAdminServer.getPermissionChecker(); + if (pc != null) { + pc.checkPermission(mountTable, FsAction.WRITE); + } + } + + boolean status = getDriver().put(mountTable, true, true); + UpdateMountTableEntryResponse response = + UpdateMountTableEntryResponse.newInstance(); + response.setStatus(status); + return response; + } + + @Override + public RemoveMountTableEntryResponse removeMountTableEntry( + RemoveMountTableEntryRequest request) throws IOException { + final String srcPath = request.getSrcPath(); + final MountTable partial = MountTable.newInstance(); + partial.setSourcePath(srcPath); + final Query<MountTable> query = new Query<>(partial); + final MountTable deleteEntry = getDriver().get(getRecordClass(), query); + + boolean status = false; + if (deleteEntry != null) { + RouterPermissionChecker pc = RouterAdminServer.getPermissionChecker(); + if (pc != null) { + pc.checkPermission(deleteEntry, FsAction.WRITE); + } + status = getDriver().remove(deleteEntry); + } + + RemoveMountTableEntryResponse response = + RemoveMountTableEntryResponse.newInstance(); + response.setStatus(status); + return response; + } + + @Override + public GetMountTableEntriesResponse getMountTableEntries( + GetMountTableEntriesRequest request) throws IOException { + RouterPermissionChecker pc = + RouterAdminServer.getPermissionChecker(); + // Get all values from the cache + List<MountTable> records = getCachedRecords(); + + // Sort and filter + Collections.sort(records, MountTable.SOURCE_COMPARATOR); + String reqSrcPath = request.getSrcPath(); + if (reqSrcPath != null && !reqSrcPath.isEmpty()) { + // Return only entries beneath this path + Iterator<MountTable> it = records.iterator(); + while (it.hasNext()) { + MountTable record = it.next(); + String srcPath = record.getSourcePath(); + if (!srcPath.startsWith(reqSrcPath)) { + it.remove(); + } else if (pc != null) { + // do the READ permission check + try { + pc.checkPermission(record, FsAction.READ); + } catch (AccessControlException ignored) { + // Remove this mount table entry if it cannot + // be accessed by current user. + it.remove(); + } + } + } + } + + GetMountTableEntriesResponse response = + GetMountTableEntriesResponse.newInstance(); + response.setEntries(records); + response.setTimestamp(Time.now()); + return response; + } +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/hadoop/blob/87700d45/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/impl/RouterStoreImpl.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/impl/RouterStoreImpl.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/impl/RouterStoreImpl.java new file mode 100644 index 0000000..d58c288 --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/impl/RouterStoreImpl.java @@ -0,0 +1,93 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.hadoop.hdfs.server.federation.store.impl; + +import java.io.IOException; +import java.util.List; + +import org.apache.hadoop.classification.InterfaceAudience; +import org.apache.hadoop.classification.InterfaceStability; +import org.apache.hadoop.hdfs.server.federation.store.RouterStore; +import org.apache.hadoop.hdfs.server.federation.store.driver.StateStoreDriver; +import org.apache.hadoop.hdfs.server.federation.store.protocol.GetRouterRegistrationRequest; +import org.apache.hadoop.hdfs.server.federation.store.protocol.GetRouterRegistrationResponse; +import org.apache.hadoop.hdfs.server.federation.store.protocol.GetRouterRegistrationsRequest; +import org.apache.hadoop.hdfs.server.federation.store.protocol.GetRouterRegistrationsResponse; +import org.apache.hadoop.hdfs.server.federation.store.protocol.RouterHeartbeatRequest; +import org.apache.hadoop.hdfs.server.federation.store.protocol.RouterHeartbeatResponse; +import org.apache.hadoop.hdfs.server.federation.store.records.Query; +import org.apache.hadoop.hdfs.server.federation.store.records.QueryResult; +import org.apache.hadoop.hdfs.server.federation.store.records.RouterState; + +/** + * Implementation of the {@link RouterStore} state store API. + */ +@InterfaceAudience.Private +@InterfaceStability.Evolving +public class RouterStoreImpl extends RouterStore { + + public RouterStoreImpl(StateStoreDriver driver) { + super(driver); + } + + @Override + public GetRouterRegistrationResponse getRouterRegistration( + GetRouterRegistrationRequest request) throws IOException { + + final RouterState partial = RouterState.newInstance(); + partial.setAddress(request.getRouterId()); + final Query<RouterState> query = new Query<RouterState>(partial); + RouterState record = getDriver().get(getRecordClass(), query); + if (record != null) { + overrideExpiredRecord(record); + } + GetRouterRegistrationResponse response = + GetRouterRegistrationResponse.newInstance(); + response.setRouter(record); + return response; + } + + @Override + public GetRouterRegistrationsResponse getRouterRegistrations( + GetRouterRegistrationsRequest request) throws IOException { + + // Get all values from the cache + QueryResult<RouterState> recordsAndTimeStamp = + getCachedRecordsAndTimeStamp(); + List<RouterState> records = recordsAndTimeStamp.getRecords(); + long timestamp = recordsAndTimeStamp.getTimestamp(); + + // Generate response + GetRouterRegistrationsResponse response = + GetRouterRegistrationsResponse.newInstance(); + response.setRouters(records); + response.setTimestamp(timestamp); + return response; + } + + @Override + public RouterHeartbeatResponse routerHeartbeat(RouterHeartbeatRequest request) + throws IOException { + + RouterState record = request.getRouter(); + boolean status = getDriver().put(record, true, false); + RouterHeartbeatResponse response = + RouterHeartbeatResponse.newInstance(status); + return response; + } +} http://git-wip-us.apache.org/repos/asf/hadoop/blob/87700d45/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/impl/package-info.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/impl/package-info.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/impl/package-info.java new file mode 100644 index 0000000..1a50d15 --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/impl/package-info.java @@ -0,0 +1,31 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +/** + * Contains implementations of the state store API interfaces. All classes + * derive from {@link + * org.apache.hadoop.hdfs.server.federation.store.RecordStore}. The API + * definitions are contained in the + * org.apache.hadoop.hdfs.server.federation.store package. + */ +@InterfaceAudience.Private +@InterfaceStability.Evolving +package org.apache.hadoop.hdfs.server.federation.store.impl; + +import org.apache.hadoop.classification.InterfaceAudience; +import org.apache.hadoop.classification.InterfaceStability; \ No newline at end of file http://git-wip-us.apache.org/repos/asf/hadoop/blob/87700d45/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/package-info.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/package-info.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/package-info.java new file mode 100644 index 0000000..949ec7c --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/package-info.java @@ -0,0 +1,62 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +/** + * The federation state store tracks persistent values that are shared between + * multiple routers. + * <p> + * Data is stored in data records that inherit from a common class. Data records + * are serialized when written to the data store using a modular serialization + * implementation. The default is profobuf serialization. Data is stored as rows + * of records of the same type with each data member in a record representing a + * column. + * <p> + * The state store uses a modular data storage + * {@link org.apache.hadoop.hdfs.server.federation.store.driver.StateStoreDriver + * StateStoreDriver} to handle querying, updating and deleting data records. The + * data storage driver is initialized and maintained by the + * {@link org.apache.hadoop.hdfs.server.federation.store. + * StateStoreService FederationStateStoreService}. The state store + * supports fetching all records of a type, filtering by column values or + * fetching a single record by its primary key. + * <p> + * The state store contains several API interfaces, one for each data records + * type. + * <p> + * <ul> + * <li>FederationMembershipStateStore: state of all Namenodes in the federation. + * Uses the MembershipState record. + * <li>FederationMountTableStore: Mount table mapping paths in the global + * namespace to individual subcluster paths. Uses the MountTable record. + * <li>RouterStateStore: State of all routers in the federation. Uses the + * RouterState record. + * </ul> + * <p> + * Each API is defined in a separate interface. The implementations of these + * interfaces are responsible for accessing the + * {@link org.apache.hadoop.hdfs.server.federation.store.driver.StateStoreDriver + * StateStoreDriver} to query, update and delete data records. + */ + +@InterfaceAudience.Private +@InterfaceStability.Evolving + +package org.apache.hadoop.hdfs.server.federation.store; + +import org.apache.hadoop.classification.InterfaceAudience; +import org.apache.hadoop.classification.InterfaceStability; http://git-wip-us.apache.org/repos/asf/hadoop/blob/87700d45/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/protocol/AddMountTableEntryRequest.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/protocol/AddMountTableEntryRequest.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/protocol/AddMountTableEntryRequest.java new file mode 100644 index 0000000..2d9f102 --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/protocol/AddMountTableEntryRequest.java @@ -0,0 +1,47 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.hadoop.hdfs.server.federation.store.protocol; + +import org.apache.hadoop.classification.InterfaceAudience.Public; +import org.apache.hadoop.classification.InterfaceStability.Unstable; +import org.apache.hadoop.hdfs.server.federation.store.driver.StateStoreSerializer; +import org.apache.hadoop.hdfs.server.federation.store.records.MountTable; + +/** + * API request for adding a mount table entry to the state store. + */ +public abstract class AddMountTableEntryRequest { + + public static AddMountTableEntryRequest newInstance() { + return StateStoreSerializer.newRecord(AddMountTableEntryRequest.class); + } + + public static AddMountTableEntryRequest newInstance(MountTable newEntry) { + AddMountTableEntryRequest request = newInstance(); + request.setEntry(newEntry); + return request; + } + + @Public + @Unstable + public abstract MountTable getEntry(); + + @Public + @Unstable + public abstract void setEntry(MountTable mount); +} \ No newline at end of file --------------------------------------------------------------------- To unsubscribe, e-mail: common-commits-unsubscr...@hadoop.apache.org For additional commands, e-mail: common-commits-h...@hadoop.apache.org
