| /** |
| * 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 |
| * <p> |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * <p> |
| * 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.hdds.utils; |
| |
| import org.apache.hadoop.hdds.HddsConfigKeys; |
| import com.google.protobuf.ServiceException; |
| import org.apache.hadoop.hdds.conf.ConfigurationSource; |
| import org.apache.hadoop.hdds.conf.OzoneConfiguration; |
| import org.apache.hadoop.hdds.function.SupplierWithIOException; |
| import org.apache.hadoop.hdds.protocol.proto.SCMSecurityProtocolProtos.SCMGetCertResponseProto; |
| import org.apache.hadoop.hdds.protocolPB.SCMSecurityProtocolClientSideTranslatorPB; |
| import org.apache.hadoop.hdds.scm.AddSCMRequest; |
| import org.apache.hadoop.hdds.scm.ScmConfigKeys; |
| import org.apache.hadoop.hdds.scm.ScmInfo; |
| import org.apache.hadoop.hdds.scm.ha.SCMHAUtils; |
| import org.apache.hadoop.hdds.scm.ha.SCMNodeInfo; |
| import org.apache.hadoop.hdds.scm.protocol.ScmBlockLocationProtocol; |
| import org.apache.hadoop.hdds.scm.protocol.StorageContainerLocationProtocol; |
| import org.apache.hadoop.hdds.scm.protocolPB.ScmBlockLocationProtocolClientSideTranslatorPB; |
| import org.apache.hadoop.hdds.scm.protocolPB.StorageContainerLocationProtocolClientSideTranslatorPB; |
| import org.apache.hadoop.hdds.scm.proxy.SCMBlockLocationFailoverProxyProvider; |
| import org.apache.hadoop.hdds.scm.proxy.SCMClientConfig; |
| import org.apache.hadoop.hdds.scm.proxy.SCMContainerLocationFailoverProxyProvider; |
| import org.apache.hadoop.hdds.security.exception.SCMSecurityException; |
| import org.apache.hadoop.hdds.security.x509.certificate.client.CertificateClient; |
| import org.apache.hadoop.hdds.security.x509.certificate.utils.CertificateCodec; |
| import org.apache.hadoop.hdds.tracing.TracingUtil; |
| import org.apache.hadoop.hdds.utils.db.DBDefinition; |
| import org.apache.hadoop.hdds.utils.db.DBColumnFamilyDefinition; |
| import org.apache.hadoop.hdds.utils.db.DBStore; |
| import org.apache.hadoop.hdds.utils.db.RocksDBConfiguration; |
| import org.apache.hadoop.hdds.utils.db.Table; |
| import org.apache.hadoop.hdds.utils.db.DBStoreBuilder; |
| import org.apache.hadoop.ozone.OzoneSecurityUtil; |
| import org.apache.hadoop.util.Time; |
| import org.apache.ratis.util.ExitUtils; |
| import org.apache.ratis.util.FileUtils; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| import org.apache.hadoop.ipc.RemoteException; |
| import org.apache.hadoop.security.AccessControlException; |
| |
| import java.io.File; |
| import java.io.IOException; |
| import java.nio.file.Files; |
| import java.nio.file.Path; |
| import java.nio.file.Paths; |
| import java.security.cert.X509Certificate; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Collection; |
| import java.util.List; |
| import java.util.concurrent.TimeUnit; |
| |
| import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_INFO_WAIT_DURATION; |
| import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_INFO_WAIT_DURATION_DEFAULT; |
| import static org.apache.hadoop.hdds.server.ServerUtils.getOzoneMetaDirPath; |
| import static org.apache.hadoop.ozone.OzoneConsts.DB_TRANSIENT_MARKER; |
| import static org.apache.hadoop.ozone.OzoneConsts.TRANSACTION_INFO_KEY; |
| |
| /** |
| * utility class used by SCM and OM for HA. |
| */ |
| public final class HAUtils { |
| public static final Logger LOG = LoggerFactory.getLogger(HAUtils.class); |
| |
| private HAUtils() { |
| } |
| |
| public static ScmInfo getScmInfo(OzoneConfiguration conf) |
| throws IOException { |
| OzoneConfiguration configuration = new OzoneConfiguration(conf); |
| try { |
| long duration = conf.getTimeDuration(OZONE_SCM_INFO_WAIT_DURATION, |
| OZONE_SCM_INFO_WAIT_DURATION_DEFAULT, TimeUnit.SECONDS); |
| SCMClientConfig scmClientConfig = |
| configuration.getObject(SCMClientConfig.class); |
| int retryCount = |
| (int) (duration / (scmClientConfig.getRetryInterval()/1000)); |
| |
| // If duration is set to lesser value, fall back to actual default |
| // retry count. |
| if (retryCount > scmClientConfig.getRetryCount()) { |
| scmClientConfig.setRetryCount(retryCount); |
| configuration.setFromObject(scmClientConfig); |
| } |
| |
| return getScmBlockClient(configuration).getScmInfo(); |
| } catch (IOException e) { |
| throw e; |
| } catch (Exception e) { |
| throw new IOException("Failed to get SCM info", e); |
| } |
| } |
| |
| /** |
| * Add SCM to the cluster. |
| * @param conf - OzoneConfiguration |
| * @param request - AddSCMRequest which has details of SCM to be added. |
| * @param selfId - Node Id of the SCM which is submitting the request to |
| * add SCM. |
| * @return true - if SCM node is added successfully, else false. |
| * @throws IOException |
| */ |
| public static boolean addSCM(OzoneConfiguration conf, AddSCMRequest request, |
| String selfId) throws IOException { |
| OzoneConfiguration config = SCMHAUtils.removeSelfId(conf, selfId); |
| try { |
| return getScmBlockClient(config).addSCM(request); |
| } catch (IOException e) { |
| throw e; |
| } catch (Exception e) { |
| throw new IOException("Failed to add SCM", e); |
| } |
| } |
| |
| /** |
| * Create a scm block client. |
| * |
| * @return {@link ScmBlockLocationProtocol} |
| * @throws IOException |
| */ |
| public static ScmBlockLocationProtocol getScmBlockClient( |
| OzoneConfiguration conf) { |
| ScmBlockLocationProtocolClientSideTranslatorPB scmBlockLocationClient = |
| new ScmBlockLocationProtocolClientSideTranslatorPB( |
| new SCMBlockLocationFailoverProxyProvider(conf)); |
| return TracingUtil |
| .createProxy(scmBlockLocationClient, ScmBlockLocationProtocol.class, |
| conf); |
| } |
| |
| public static StorageContainerLocationProtocol getScmContainerClient( |
| ConfigurationSource conf) { |
| SCMContainerLocationFailoverProxyProvider proxyProvider = |
| new SCMContainerLocationFailoverProxyProvider(conf); |
| StorageContainerLocationProtocol scmContainerClient = |
| TracingUtil.createProxy( |
| new StorageContainerLocationProtocolClientSideTranslatorPB( |
| proxyProvider), StorageContainerLocationProtocol.class, conf); |
| return scmContainerClient; |
| } |
| |
| /** |
| * Replace the current DB with the new DB checkpoint. |
| * |
| * @param lastAppliedIndex the last applied index in the current SCM DB. |
| * @param checkpointPath path to the new DB checkpoint |
| * @return location of backup of the original DB |
| * @throws Exception |
| */ |
| public static File replaceDBWithCheckpoint(long lastAppliedIndex, |
| File oldDB, Path checkpointPath, String dbPrefix) throws IOException { |
| |
| // Take a backup of the current DB |
| String dbBackupName = |
| dbPrefix + lastAppliedIndex + "_" + System |
| .currentTimeMillis(); |
| File dbDir = oldDB.getParentFile(); |
| File dbBackup = new File(dbDir, dbBackupName); |
| |
| try { |
| Files.move(oldDB.toPath(), dbBackup.toPath()); |
| } catch (IOException e) { |
| LOG.error("Failed to create a backup of the current DB. Aborting " |
| + "snapshot installation."); |
| throw e; |
| } |
| |
| // Move the new DB checkpoint into the metadata dir |
| Path markerFile = new File(dbDir, DB_TRANSIENT_MARKER).toPath(); |
| try { |
| // Create a Transient Marker file. This file will be deleted if the |
| // checkpoint DB is successfully moved to the old DB location or if the |
| // old DB backup is reset to its location. If not, then the DB is in |
| // an inconsistent state and this marker file will fail it from |
| // starting up. |
| Files.createFile(markerFile); |
| Files.move(checkpointPath, oldDB.toPath()); |
| Files.deleteIfExists(markerFile); |
| } catch (IOException e) { |
| LOG.error("Failed to move downloaded DB checkpoint {} to metadata " |
| + "directory {}. Resetting to original DB.", checkpointPath, |
| oldDB.toPath()); |
| try { |
| Files.move(dbBackup.toPath(), oldDB.toPath()); |
| Files.deleteIfExists(markerFile); |
| } catch (IOException ex) { |
| String errorMsg = "Failed to reset to original DB. SCM is in an " |
| + "inconsistent state."; |
| ExitUtils.terminate(1, errorMsg, ex, LOG); |
| } |
| throw e; |
| } |
| return dbBackup; |
| } |
| |
| /** |
| * Obtain SCMTransactionInfo from Checkpoint. |
| */ |
| public static TransactionInfo getTrxnInfoFromCheckpoint( |
| OzoneConfiguration conf, Path dbPath, DBDefinition definition) |
| throws Exception { |
| |
| if (dbPath != null) { |
| Path dbDir = dbPath.getParent(); |
| Path dbFile = dbPath.getFileName(); |
| if (dbDir != null && dbFile != null) { |
| return getTransactionInfoFromDB(conf, dbDir, dbFile.toString(), |
| definition); |
| } |
| } |
| |
| throw new IOException("Checkpoint " + dbPath + " does not have proper " + |
| "DB location"); |
| } |
| |
| /** |
| * Obtain Transaction info from DB. |
| * @param tempConfig |
| * @param dbDir path to DB |
| * @return TransactionInfo |
| * @throws Exception |
| */ |
| private static TransactionInfo getTransactionInfoFromDB( |
| OzoneConfiguration tempConfig, Path dbDir, String dbName, |
| DBDefinition definition) |
| throws Exception { |
| DBStore dbStore = loadDB(tempConfig, dbDir.toFile(), |
| dbName, definition); |
| |
| // Get the table name with TransactionInfo as the value. The transaction |
| // info table name are different in SCM and SCM. |
| |
| // In case, a new table gets added where the value is TransactionInfo, this |
| // logic may not work. |
| |
| |
| Table<String, TransactionInfo> transactionInfoTable = |
| getTransactionInfoTable(dbStore, definition); |
| |
| TransactionInfo transactionInfo = |
| transactionInfoTable.get(TRANSACTION_INFO_KEY); |
| dbStore.close(); |
| |
| if (transactionInfo == null) { |
| throw new IOException("Failed to read TransactionInfo from DB " + |
| definition.getName() + " at " + dbDir); |
| } |
| return transactionInfo; |
| } |
| |
| public static Table<String, TransactionInfo> getTransactionInfoTable( |
| DBStore dbStore, DBDefinition definition) throws IOException { |
| return Arrays.stream(definition.getColumnFamilies()) |
| .filter(t -> t.getValueType() == TransactionInfo.class).findFirst() |
| .get().getTable(dbStore); |
| } |
| |
| /** |
| * Verify transaction info with provided lastAppliedIndex. |
| * |
| * If transaction info transaction Index is less than or equal to |
| * lastAppliedIndex, return false, else return true. |
| * @param transactionInfo |
| * @param lastAppliedIndex |
| * @param leaderId |
| * @param newDBlocation |
| * @return boolean |
| */ |
| public static boolean verifyTransactionInfo(TransactionInfo transactionInfo, |
| long lastAppliedIndex, String leaderId, Path newDBlocation, |
| Logger logger) { |
| if (transactionInfo.getTransactionIndex() <= lastAppliedIndex) { |
| logger.error("Failed to install checkpoint from SCM leader: {}" |
| + ". The last applied index: {} is greater than or equal to the " |
| + "checkpoint's applied index: {}. Deleting the downloaded " |
| + "checkpoint {}", leaderId, lastAppliedIndex, |
| transactionInfo.getTransactionIndex(), newDBlocation); |
| try { |
| FileUtils.deleteFully(newDBlocation); |
| } catch (IOException e) { |
| logger.error("Failed to fully delete the downloaded DB " |
| + "checkpoint {} from SCM leader {}.", newDBlocation, leaderId, e); |
| } |
| return false; |
| } |
| return true; |
| } |
| |
| public static DBStore loadDB(OzoneConfiguration configuration, File metaDir, |
| String dbName, DBDefinition definition) throws IOException { |
| RocksDBConfiguration rocksDBConfiguration = |
| configuration.getObject(RocksDBConfiguration.class); |
| DBStoreBuilder dbStoreBuilder = |
| DBStoreBuilder.newBuilder(configuration, rocksDBConfiguration) |
| .setName(dbName).setPath(Paths.get(metaDir.getPath())); |
| // Add column family names and codecs. |
| for (DBColumnFamilyDefinition columnFamily : definition |
| .getColumnFamilies()) { |
| |
| dbStoreBuilder.addTable(columnFamily.getName()); |
| dbStoreBuilder |
| .addCodec(columnFamily.getKeyType(), columnFamily.getKeyCodec()); |
| dbStoreBuilder |
| .addCodec(columnFamily.getValueType(), columnFamily.getValueCodec()); |
| } |
| return dbStoreBuilder.build(); |
| } |
| |
| public static File getMetaDir(DBDefinition definition, |
| OzoneConfiguration configuration) { |
| // Set metadata dirs. |
| File metadataDir = definition.getDBLocation(configuration); |
| |
| if (metadataDir == null) { |
| LOG.warn("{} is not configured. We recommend adding this setting. " |
| + "Falling back to {} instead.", |
| definition.getLocationConfigKey(), HddsConfigKeys. |
| OZONE_METADATA_DIRS); |
| metadataDir = getOzoneMetaDirPath(configuration); |
| } |
| return metadataDir; |
| } |
| |
| /** |
| * Unwrap exception to check if it is some kind of access control problem. |
| * {@link AccessControlException} |
| */ |
| public static boolean isAccessControlException(Exception ex) { |
| if (ex instanceof ServiceException) { |
| Throwable t = ex.getCause(); |
| if (t instanceof RemoteException) { |
| t = ((RemoteException) t).unwrapRemoteException(); |
| } |
| while (t != null) { |
| if (t instanceof AccessControlException) { |
| return true; |
| } |
| t = t.getCause(); |
| } |
| } |
| return false; |
| } |
| |
| public static void checkSecurityAndSCMHAEnabled(OzoneConfiguration conf) { |
| boolean enable = |
| conf.getBoolean(ScmConfigKeys.OZONE_SCM_HA_SECURITY_SUPPORTED, |
| ScmConfigKeys.OZONE_SCM_HA_SECURITY_SUPPORTED_DEFAULT); |
| if (OzoneSecurityUtil.isSecurityEnabled(conf) && !enable) { |
| List<SCMNodeInfo> scmNodeInfo = SCMNodeInfo.buildNodeInfo(conf); |
| if (scmNodeInfo.size() > 1) { |
| System.err.println("Ozone Services cannot be started on a secure SCM " + |
| "HA enabled cluster"); |
| System.exit(1); |
| } |
| } |
| } |
| |
| /** |
| * Build CA list which need to be passed to client. |
| * |
| * If certificate client is null, obtain the list of CA using SCM security |
| * client, else it uses certificate client. |
| * @param certClient |
| * @param configuration |
| * @return list of CA |
| * @throws IOException |
| */ |
| public static List<String> buildCAList(CertificateClient certClient, |
| ConfigurationSource configuration) throws IOException { |
| //TODO: make it configurable. |
| List<String> caCertPemList; |
| if (certClient != null) { |
| caCertPemList = new ArrayList<>(); |
| if (!SCMHAUtils.isSCMHAEnabled(configuration)) { |
| if (certClient.getRootCACertificate() != null) { |
| caCertPemList.add(CertificateCodec.getPEMEncodedString( |
| certClient.getRootCACertificate())); |
| } |
| caCertPemList.add(CertificateCodec.getPEMEncodedString( |
| certClient.getCACertificate())); |
| } else { |
| Collection<String> scmNodes = SCMHAUtils.getSCMNodeIds(configuration); |
| int expectedCount = scmNodes.size() + 1; |
| if (scmNodes.size() > 1) { |
| // First check if cert client has ca list initialized. |
| // This is being done, when this method is called multiple times we |
| // don't make call to SCM, we return from in-memory. |
| caCertPemList = certClient.getCAList(); |
| if (caCertPemList != null && caCertPemList.size() == expectedCount) { |
| return caCertPemList; |
| } |
| caCertPemList = waitForCACerts(() -> certClient.updateCAList(), |
| expectedCount); |
| checkCertCount(caCertPemList.size(), expectedCount); |
| } else { |
| caCertPemList = certClient.listCA(); |
| } |
| } |
| } else { |
| SCMSecurityProtocolClientSideTranslatorPB scmSecurityProtocolClient = |
| HddsServerUtil.getScmSecurityClient(configuration); |
| if (!SCMHAUtils.isSCMHAEnabled(configuration)) { |
| caCertPemList = new ArrayList<>(); |
| SCMGetCertResponseProto scmGetCertResponseProto = |
| scmSecurityProtocolClient.getCACert(); |
| if (scmGetCertResponseProto.hasX509Certificate()) { |
| caCertPemList.add(scmGetCertResponseProto.getX509Certificate()); |
| } |
| if (scmGetCertResponseProto.hasX509RootCACertificate()) { |
| caCertPemList.add(scmGetCertResponseProto.getX509RootCACertificate()); |
| } |
| } else { |
| Collection<String> scmNodes = SCMHAUtils.getSCMNodeIds(configuration); |
| int expectedCount = scmNodes.size() + 1; |
| if (scmNodes.size() > 1) { |
| caCertPemList = waitForCACerts( |
| () -> scmSecurityProtocolClient.listCACertificate(), |
| expectedCount); |
| checkCertCount(caCertPemList.size(), expectedCount); |
| } else{ |
| caCertPemList = scmSecurityProtocolClient.listCACertificate(); |
| } |
| } |
| } |
| return caCertPemList; |
| } |
| |
| private static List<String> waitForCACerts( |
| final SupplierWithIOException<List<String>> applyFunction, |
| int expectedCount) throws IOException { |
| //TODO: make wait time and sleep time configurable if needed. |
| // TODO: If SCMs are bootstrapped later, then listCA need to be |
| // refetched if listCA size is less than scm ha config node list size. |
| // For now when Client of SCM's are started we compare their node list |
| // size and ca list size if it is as expected, we return the ca list. |
| boolean caListUpToDate; |
| long waitTime = 5 * 60 * 1000L; |
| long retryTime = 10 * 1000L; |
| long currentTime = Time.monotonicNow(); |
| List<String> caCertPemList; |
| do { |
| caCertPemList = applyFunction.get(); |
| caListUpToDate = |
| caCertPemList.size() == expectedCount ? true : false; |
| if (!caListUpToDate) { |
| LOG.info("Expected CA list size {}, where as received CA List size " + |
| "{}. Retry to fetch CA List after {} seconds", expectedCount, |
| caCertPemList.size(), waitTime/1000); |
| try { |
| Thread.sleep(retryTime); |
| } catch (InterruptedException ex) { |
| Thread.currentThread().interrupt(); |
| } |
| } |
| } while (!caListUpToDate && |
| Time.monotonicNow() - currentTime < waitTime); |
| return caCertPemList; |
| } |
| |
| |
| private static void checkCertCount(int certCount, int expectedCount) |
| throws SCMSecurityException{ |
| if (certCount != expectedCount) { |
| LOG.error("Unable to obtain CA list for SCM cluster, obtained CA list " + |
| "size is {}, where as expected list size is {}", |
| certCount, expectedCount); |
| throw new SCMSecurityException("Unable to obtain complete CA list"); |
| } |
| } |
| |
| /** |
| * Build CA List in the format of X509Certificate. |
| * If certificate client is null, obtain the list of CA using SCM |
| * security client, else it uses certificate client. |
| * @param certClient |
| * @param conf |
| * @return list of CA X509Certificates. |
| * @throws IOException |
| */ |
| public static List<X509Certificate> buildCAX509List( |
| CertificateClient certClient, |
| ConfigurationSource conf) throws IOException { |
| if (certClient != null) { |
| // Do this here to avoid extra conversion of X509 to pem and again to |
| // X509 by buildCAList. |
| if (!SCMHAUtils.isSCMHAEnabled(conf)) { |
| List<X509Certificate> x509Certificates = new ArrayList<>(); |
| if (certClient.getRootCACertificate() != null) { |
| x509Certificates.add(certClient.getRootCACertificate()); |
| } |
| x509Certificates.add(certClient.getCACertificate()); |
| return x509Certificates; |
| } |
| } |
| List<String> pemEncodedCerts = HAUtils.buildCAList(certClient, conf); |
| return OzoneSecurityUtil.convertToX509(pemEncodedCerts); |
| } |
| |
| } |
