You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@ozone.apache.org by GitBox <gi...@apache.org> on 2021/03/02 06:56:41 UTC
[GitHub] [ozone] bharatviswa504 opened a new pull request #1978: HDDS-4876. [SCM HA Security] Add failover proxy to SCM Security Server Protocol
bharatviswa504 opened a new pull request #1978:
URL: https://github.com/apache/ozone/pull/1978
## What changes were proposed in this pull request?
Implement failover proxy provider for SCM Security Server Protocol.
## What is the link to the Apache JIRA
https://issues.apache.org/jira/browse/HDDS-4876
## How was this patch tested?
Updated few tests and also existing tests should cover. (For HA, later tests will be added end to end)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
[GitHub] [ozone] bharatviswa504 commented on a change in pull request #1978: HDDS-4876. [SCM HA Security] Add failover proxy to SCM Security Server Protocol
Posted by GitBox <gi...@apache.org>.
bharatviswa504 commented on a change in pull request #1978:
URL: https://github.com/apache/ozone/pull/1978#discussion_r590062773
##########
File path: hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/proxy/SCMSecurityProtocolFailoverProxyProvider.java
##########
@@ -0,0 +1,279 @@
+/*
+ * 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.scm.proxy;
+
+import com.google.common.base.Preconditions;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdds.conf.ConfigurationException;
+import org.apache.hadoop.hdds.conf.ConfigurationSource;
+import org.apache.hadoop.hdds.protocolPB.SCMSecurityProtocolPB;
+import org.apache.hadoop.hdds.scm.ha.SCMNodeInfo;
+import org.apache.hadoop.hdds.utils.HAUtils;
+import org.apache.hadoop.hdds.utils.LegacyHadoopConfigurationSource;
+import org.apache.hadoop.io.retry.FailoverProxyProvider;
+import org.apache.hadoop.io.retry.RetryPolicies;
+import org.apache.hadoop.io.retry.RetryPolicy;
+import org.apache.hadoop.io.retry.RetryPolicy.RetryAction.RetryDecision;
+import org.apache.hadoop.ipc.ProtobufRpcEngine;
+import org.apache.hadoop.ipc.RPC;
+import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.hadoop.io.retry.RetryPolicy.RetryAction.RetryDecision.FAILOVER_AND_RETRY;
+
+/**
+ * Failover proxy provider for SCMSecurityProtocol server.
+ */
+public class SCMSecurityProtocolFailoverProxyProvider implements
+ FailoverProxyProvider<SCMSecurityProtocolPB>, Closeable {
+
+ public static final Logger LOG =
+ LoggerFactory.getLogger(SCMSecurityProtocolFailoverProxyProvider.class);
+
+ // scmNodeId -> ProxyInfo<rpcProxy>
+ private final Map<String,
+ ProxyInfo<SCMSecurityProtocolPB>> scmProxies;
+
+ // scmNodeId -> SCMProxyInfo
+ private final Map<String, SCMProxyInfo> scmProxyInfoMap;
+
+ private List<String> scmNodeIds;
+
+ private String currentProxySCMNodeId;
+ private int currentProxyIndex;
+
+ private final ConfigurationSource conf;
+ private final SCMClientConfig scmClientConfig;
+ private final long scmVersion;
+
+ private String scmServiceId;
+
+ private final int maxRetryCount;
+ private final long retryInterval;
+
+ private final UserGroupInformation ugi;
+
+ /**
+ * Construct fail-over proxy provider for SCMSecurityProtocol Server.
+ * @param conf
+ * @param userGroupInformation
+ */
+ public SCMSecurityProtocolFailoverProxyProvider(ConfigurationSource conf,
+ UserGroupInformation userGroupInformation) {
+ Preconditions.checkNotNull(userGroupInformation);
+ this.ugi = userGroupInformation;
+ this.conf = conf;
+ this.scmVersion = RPC.getProtocolVersion(SCMSecurityProtocolPB.class);
+
+ this.scmProxies = new HashMap<>();
+ this.scmProxyInfoMap = new HashMap<>();
+ loadConfigs();
+
+ this.currentProxyIndex = 0;
+ currentProxySCMNodeId = scmNodeIds.get(currentProxyIndex);
+ scmClientConfig = conf.getObject(SCMClientConfig.class);
+ this.maxRetryCount = scmClientConfig.getRetryCount();
+ this.retryInterval = scmClientConfig.getRetryInterval();
+ }
+
+ protected void loadConfigs() {
+ List<SCMNodeInfo> scmNodeInfoList = SCMNodeInfo.buildNodeInfo(conf);
+ scmNodeIds = new ArrayList<>();
+
+ for (SCMNodeInfo scmNodeInfo : scmNodeInfoList) {
+ if (scmNodeInfo.getScmSecurityAddress() == null) {
+ throw new ConfigurationException("SCM Client Address could not " +
+ "be obtained from config. Config is not properly defined");
+ } else {
+ InetSocketAddress scmSecurityAddress =
+ NetUtils.createSocketAddr(scmNodeInfo.getScmSecurityAddress());
+
+ scmServiceId = scmNodeInfo.getServiceId();
+ String scmNodeId = scmNodeInfo.getNodeId();
+
+ scmNodeIds.add(scmNodeId);
+ SCMProxyInfo scmProxyInfo = new SCMProxyInfo(scmServiceId, scmNodeId,
+ scmSecurityAddress);
+ scmProxyInfoMap.put(scmNodeId, scmProxyInfo);
+ }
+ }
+ }
+
+ @Override
+ public ProxyInfo<SCMSecurityProtocolPB> getProxy() {
+ ProxyInfo currentProxyInfo = scmProxies.get(getCurrentProxySCMNodeId());
+ if (currentProxyInfo == null) {
+ currentProxyInfo = createSCMProxy(getCurrentProxySCMNodeId());
+ }
+ return currentProxyInfo;
+ }
+
+ /**
+ * Creates proxy object.
+ */
+ private ProxyInfo createSCMProxy(String nodeId) {
+ ProxyInfo proxyInfo;
+ SCMProxyInfo scmProxyInfo = scmProxyInfoMap.get(nodeId);
+ InetSocketAddress address = scmProxyInfo.getAddress();
+ try {
+ SCMSecurityProtocolPB scmProxy = createSCMProxy(address);
+ // Create proxyInfo here, to make it work with all Hadoop versions.
+ proxyInfo = new ProxyInfo<>(scmProxy, scmProxyInfo.toString());
+ scmProxies.put(nodeId, proxyInfo);
+ return proxyInfo;
+ } catch (IOException ioe) {
+ LOG.error("{} Failed to create RPC proxy to SCM at {}",
+ this.getClass().getSimpleName(), address, ioe);
+ throw new RuntimeException(ioe);
+ }
+ }
+
+ private SCMSecurityProtocolPB createSCMProxy(InetSocketAddress scmAddress)
+ throws IOException {
+ Configuration hadoopConf =
+ LegacyHadoopConfigurationSource.asHadoopConfiguration(conf);
+ RPC.setProtocolEngine(hadoopConf, SCMSecurityProtocolPB.class,
+ ProtobufRpcEngine.class);
+
+ // FailoverOnNetworkException ensures that the IPC layer does not attempt
+ // retries on the same SCM in case of connection exception. This retry
+ // policy essentially results in TRY_ONCE_THEN_FAIL.
+
+ RetryPolicy connectionRetryPolicy = RetryPolicies
+ .failoverOnNetworkException(0);
+
+ return RPC.getProtocolProxy(SCMSecurityProtocolPB.class,
+ scmVersion, scmAddress, ugi,
+ hadoopConf, NetUtils.getDefaultSocketFactory(hadoopConf),
+ (int)scmClientConfig.getRpcTimeOut(), connectionRetryPolicy).getProxy();
+ }
+
+
+ @Override
+ public void performFailover(SCMSecurityProtocolPB currentProxy) {
+ if (LOG.isDebugEnabled()) {
+ int currentIndex = getCurrentProxyIndex();
+ LOG.debug("Failing over SCM Security proxy to index: {}, nodeId: {}",
+ currentIndex, scmNodeIds.get(currentIndex));
+ }
+ }
+
+ /**
+ * Performs fail-over to the next proxy.
+ */
+ public void performFailoverToNextProxy() {
+ int newProxyIndex = incrementProxyIndex();
Review comment:
Currently in SCM we don't have leader hints, Added a TODO for this.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
[GitHub] [ozone] bharatviswa504 commented on a change in pull request #1978: HDDS-4876. [SCM HA Security] Add failover proxy to SCM Security Server Protocol
Posted by GitBox <gi...@apache.org>.
bharatviswa504 commented on a change in pull request #1978:
URL: https://github.com/apache/ozone/pull/1978#discussion_r590061539
##########
File path: hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/proxy/SCMSecurityProtocolFailoverProxyProvider.java
##########
@@ -0,0 +1,279 @@
+/*
+ * 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.scm.proxy;
+
+import com.google.common.base.Preconditions;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdds.conf.ConfigurationException;
+import org.apache.hadoop.hdds.conf.ConfigurationSource;
+import org.apache.hadoop.hdds.protocolPB.SCMSecurityProtocolPB;
+import org.apache.hadoop.hdds.scm.ha.SCMNodeInfo;
+import org.apache.hadoop.hdds.utils.HAUtils;
+import org.apache.hadoop.hdds.utils.LegacyHadoopConfigurationSource;
+import org.apache.hadoop.io.retry.FailoverProxyProvider;
+import org.apache.hadoop.io.retry.RetryPolicies;
+import org.apache.hadoop.io.retry.RetryPolicy;
+import org.apache.hadoop.io.retry.RetryPolicy.RetryAction.RetryDecision;
+import org.apache.hadoop.ipc.ProtobufRpcEngine;
+import org.apache.hadoop.ipc.RPC;
+import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.hadoop.io.retry.RetryPolicy.RetryAction.RetryDecision.FAILOVER_AND_RETRY;
+
+/**
+ * Failover proxy provider for SCMSecurityProtocol server.
+ */
+public class SCMSecurityProtocolFailoverProxyProvider implements
+ FailoverProxyProvider<SCMSecurityProtocolPB>, Closeable {
+
+ public static final Logger LOG =
+ LoggerFactory.getLogger(SCMSecurityProtocolFailoverProxyProvider.class);
+
+ // scmNodeId -> ProxyInfo<rpcProxy>
+ private final Map<String,
+ ProxyInfo<SCMSecurityProtocolPB>> scmProxies;
+
+ // scmNodeId -> SCMProxyInfo
+ private final Map<String, SCMProxyInfo> scmProxyInfoMap;
+
+ private List<String> scmNodeIds;
+
+ private String currentProxySCMNodeId;
+ private int currentProxyIndex;
+
+ private final ConfigurationSource conf;
+ private final SCMClientConfig scmClientConfig;
+ private final long scmVersion;
+
+ private String scmServiceId;
+
+ private final int maxRetryCount;
+ private final long retryInterval;
+
+ private final UserGroupInformation ugi;
+
+ /**
+ * Construct fail-over proxy provider for SCMSecurityProtocol Server.
+ * @param conf
+ * @param userGroupInformation
+ */
+ public SCMSecurityProtocolFailoverProxyProvider(ConfigurationSource conf,
+ UserGroupInformation userGroupInformation) {
+ Preconditions.checkNotNull(userGroupInformation);
+ this.ugi = userGroupInformation;
+ this.conf = conf;
+ this.scmVersion = RPC.getProtocolVersion(SCMSecurityProtocolPB.class);
+
+ this.scmProxies = new HashMap<>();
+ this.scmProxyInfoMap = new HashMap<>();
+ loadConfigs();
+
+ this.currentProxyIndex = 0;
+ currentProxySCMNodeId = scmNodeIds.get(currentProxyIndex);
+ scmClientConfig = conf.getObject(SCMClientConfig.class);
+ this.maxRetryCount = scmClientConfig.getRetryCount();
+ this.retryInterval = scmClientConfig.getRetryInterval();
+ }
+
+ protected void loadConfigs() {
+ List<SCMNodeInfo> scmNodeInfoList = SCMNodeInfo.buildNodeInfo(conf);
+ scmNodeIds = new ArrayList<>();
+
+ for (SCMNodeInfo scmNodeInfo : scmNodeInfoList) {
+ if (scmNodeInfo.getScmSecurityAddress() == null) {
+ throw new ConfigurationException("SCM Client Address could not " +
+ "be obtained from config. Config is not properly defined");
+ } else {
+ InetSocketAddress scmSecurityAddress =
+ NetUtils.createSocketAddr(scmNodeInfo.getScmSecurityAddress());
+
+ scmServiceId = scmNodeInfo.getServiceId();
Review comment:
Yes, we have a separate config for service id.
BuildNodeInfo got the list of ScmNodeInfo for a single scm serviceId, so all scmNodeInfo will have same scmServiceID.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
[GitHub] [ozone] bshashikant commented on a change in pull request #1978: HDDS-4876. [SCM HA Security] Add failover proxy to SCM Security Server Protocol
Posted by GitBox <gi...@apache.org>.
bshashikant commented on a change in pull request #1978:
URL: https://github.com/apache/ozone/pull/1978#discussion_r587430024
##########
File path: hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/proxy/SCMSecurityProtocolFailoverProxyProvider.java
##########
@@ -0,0 +1,279 @@
+/*
+ * 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.scm.proxy;
+
+import com.google.common.base.Preconditions;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdds.conf.ConfigurationException;
+import org.apache.hadoop.hdds.conf.ConfigurationSource;
+import org.apache.hadoop.hdds.protocolPB.SCMSecurityProtocolPB;
+import org.apache.hadoop.hdds.scm.ha.SCMNodeInfo;
+import org.apache.hadoop.hdds.utils.HAUtils;
+import org.apache.hadoop.hdds.utils.LegacyHadoopConfigurationSource;
+import org.apache.hadoop.io.retry.FailoverProxyProvider;
+import org.apache.hadoop.io.retry.RetryPolicies;
+import org.apache.hadoop.io.retry.RetryPolicy;
+import org.apache.hadoop.io.retry.RetryPolicy.RetryAction.RetryDecision;
+import org.apache.hadoop.ipc.ProtobufRpcEngine;
+import org.apache.hadoop.ipc.RPC;
+import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.hadoop.io.retry.RetryPolicy.RetryAction.RetryDecision.FAILOVER_AND_RETRY;
+
+/**
+ * Failover proxy provider for SCMSecurityProtocol server.
+ */
+public class SCMSecurityProtocolFailoverProxyProvider implements
+ FailoverProxyProvider<SCMSecurityProtocolPB>, Closeable {
+
+ public static final Logger LOG =
+ LoggerFactory.getLogger(SCMSecurityProtocolFailoverProxyProvider.class);
+
+ // scmNodeId -> ProxyInfo<rpcProxy>
+ private final Map<String,
+ ProxyInfo<SCMSecurityProtocolPB>> scmProxies;
+
+ // scmNodeId -> SCMProxyInfo
+ private final Map<String, SCMProxyInfo> scmProxyInfoMap;
+
+ private List<String> scmNodeIds;
+
+ private String currentProxySCMNodeId;
+ private int currentProxyIndex;
+
+ private final ConfigurationSource conf;
+ private final SCMClientConfig scmClientConfig;
+ private final long scmVersion;
+
+ private String scmServiceId;
+
+ private final int maxRetryCount;
+ private final long retryInterval;
+
+ private final UserGroupInformation ugi;
+
+ /**
+ * Construct fail-over proxy provider for SCMSecurityProtocol Server.
+ * @param conf
+ * @param userGroupInformation
+ */
+ public SCMSecurityProtocolFailoverProxyProvider(ConfigurationSource conf,
+ UserGroupInformation userGroupInformation) {
+ Preconditions.checkNotNull(userGroupInformation);
+ this.ugi = userGroupInformation;
+ this.conf = conf;
+ this.scmVersion = RPC.getProtocolVersion(SCMSecurityProtocolPB.class);
+
+ this.scmProxies = new HashMap<>();
+ this.scmProxyInfoMap = new HashMap<>();
+ loadConfigs();
+
+ this.currentProxyIndex = 0;
+ currentProxySCMNodeId = scmNodeIds.get(currentProxyIndex);
+ scmClientConfig = conf.getObject(SCMClientConfig.class);
+ this.maxRetryCount = scmClientConfig.getRetryCount();
+ this.retryInterval = scmClientConfig.getRetryInterval();
+ }
+
+ protected void loadConfigs() {
+ List<SCMNodeInfo> scmNodeInfoList = SCMNodeInfo.buildNodeInfo(conf);
+ scmNodeIds = new ArrayList<>();
+
+ for (SCMNodeInfo scmNodeInfo : scmNodeInfoList) {
+ if (scmNodeInfo.getScmSecurityAddress() == null) {
Review comment:
@bharatviswa504 , what will happen if SCM HA mode is disabled and no serviced id or node ids are defined.?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
[GitHub] [ozone] bharatviswa504 commented on a change in pull request #1978: HDDS-4876. [SCM HA Security] Add failover proxy to SCM Security Server Protocol
Posted by GitBox <gi...@apache.org>.
bharatviswa504 commented on a change in pull request #1978:
URL: https://github.com/apache/ozone/pull/1978#discussion_r587539431
##########
File path: hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/proxy/SCMSecurityProtocolFailoverProxyProvider.java
##########
@@ -0,0 +1,279 @@
+/*
+ * 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.scm.proxy;
+
+import com.google.common.base.Preconditions;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdds.conf.ConfigurationException;
+import org.apache.hadoop.hdds.conf.ConfigurationSource;
+import org.apache.hadoop.hdds.protocolPB.SCMSecurityProtocolPB;
+import org.apache.hadoop.hdds.scm.ha.SCMNodeInfo;
+import org.apache.hadoop.hdds.utils.HAUtils;
+import org.apache.hadoop.hdds.utils.LegacyHadoopConfigurationSource;
+import org.apache.hadoop.io.retry.FailoverProxyProvider;
+import org.apache.hadoop.io.retry.RetryPolicies;
+import org.apache.hadoop.io.retry.RetryPolicy;
+import org.apache.hadoop.io.retry.RetryPolicy.RetryAction.RetryDecision;
+import org.apache.hadoop.ipc.ProtobufRpcEngine;
+import org.apache.hadoop.ipc.RPC;
+import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.hadoop.io.retry.RetryPolicy.RetryAction.RetryDecision.FAILOVER_AND_RETRY;
+
+/**
+ * Failover proxy provider for SCMSecurityProtocol server.
+ */
+public class SCMSecurityProtocolFailoverProxyProvider implements
+ FailoverProxyProvider<SCMSecurityProtocolPB>, Closeable {
+
+ public static final Logger LOG =
+ LoggerFactory.getLogger(SCMSecurityProtocolFailoverProxyProvider.class);
+
+ // scmNodeId -> ProxyInfo<rpcProxy>
+ private final Map<String,
+ ProxyInfo<SCMSecurityProtocolPB>> scmProxies;
+
+ // scmNodeId -> SCMProxyInfo
+ private final Map<String, SCMProxyInfo> scmProxyInfoMap;
+
+ private List<String> scmNodeIds;
+
+ private String currentProxySCMNodeId;
+ private int currentProxyIndex;
+
+ private final ConfigurationSource conf;
+ private final SCMClientConfig scmClientConfig;
+ private final long scmVersion;
+
+ private String scmServiceId;
+
+ private final int maxRetryCount;
+ private final long retryInterval;
+
+ private final UserGroupInformation ugi;
+
+ /**
+ * Construct fail-over proxy provider for SCMSecurityProtocol Server.
+ * @param conf
+ * @param userGroupInformation
+ */
+ public SCMSecurityProtocolFailoverProxyProvider(ConfigurationSource conf,
+ UserGroupInformation userGroupInformation) {
+ Preconditions.checkNotNull(userGroupInformation);
+ this.ugi = userGroupInformation;
+ this.conf = conf;
+ this.scmVersion = RPC.getProtocolVersion(SCMSecurityProtocolPB.class);
+
+ this.scmProxies = new HashMap<>();
+ this.scmProxyInfoMap = new HashMap<>();
+ loadConfigs();
+
+ this.currentProxyIndex = 0;
+ currentProxySCMNodeId = scmNodeIds.get(currentProxyIndex);
+ scmClientConfig = conf.getObject(SCMClientConfig.class);
+ this.maxRetryCount = scmClientConfig.getRetryCount();
+ this.retryInterval = scmClientConfig.getRetryInterval();
+ }
+
+ protected void loadConfigs() {
+ List<SCMNodeInfo> scmNodeInfoList = SCMNodeInfo.buildNodeInfo(conf);
+ scmNodeIds = new ArrayList<>();
+
+ for (SCMNodeInfo scmNodeInfo : scmNodeInfoList) {
+ if (scmNodeInfo.getScmSecurityAddress() == null) {
Review comment:
buildNodeInfo() takes care of HA/non-HA config, and returns scmNodeInfo.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
[GitHub] [ozone] bharatviswa504 commented on a change in pull request #1978: HDDS-4876. [SCM HA Security] Add failover proxy to SCM Security Server Protocol
Posted by GitBox <gi...@apache.org>.
bharatviswa504 commented on a change in pull request #1978:
URL: https://github.com/apache/ozone/pull/1978#discussion_r589258722
##########
File path: hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/HAUtils.java
##########
@@ -285,4 +288,27 @@ public static File getMetaDir(DBDefinition definition,
}
return metadataDir;
}
+
+ /**
+ * Unwrap exception to check if it is some kind of access control problem
+=======
Review comment:
done
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
[GitHub] [ozone] bharatviswa504 commented on a change in pull request #1978: HDDS-4876. [SCM HA Security] Add failover proxy to SCM Security Server Protocol
Posted by GitBox <gi...@apache.org>.
bharatviswa504 commented on a change in pull request #1978:
URL: https://github.com/apache/ozone/pull/1978#discussion_r590121801
##########
File path: hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/protocol/SCMSecurityProtocolServerSideTranslatorPB.java
##########
@@ -70,62 +75,75 @@ public SCMSecurityProtocolServerSideTranslatorPB(SCMSecurityProtocol impl,
@Override
public SCMSecurityResponse submitRequest(RpcController controller,
SCMSecurityRequest request) throws ServiceException {
+ if (!scm.checkLeader()) {
+ throw new ServiceException(scm.getScmHAManager()
+ .getRatisServer()
+ .triggerNotLeaderException());
+ }
return dispatcher.processRequest(request, this::processRequest,
request.getCmdType(), request.getTraceID());
}
- public SCMSecurityResponse processRequest(SCMSecurityRequest request)
- throws ServiceException {
+ public SCMSecurityResponse processRequest(SCMSecurityRequest request) {
+ SCMSecurityResponse.Builder scmSecurityResponse =
+ SCMSecurityResponse.newBuilder().setCmdType(request.getCmdType())
+ .setStatus(Status.OK);
try {
switch (request.getCmdType()) {
case GetCertificate:
- return SCMSecurityResponse.newBuilder()
- .setCmdType(request.getCmdType())
- .setStatus(Status.OK)
- .setGetCertResponseProto(
- getCertificate(request.getGetCertificateRequest()))
- .build();
+ return scmSecurityResponse.setGetCertResponseProto(
+ getCertificate(request.getGetCertificateRequest())).build();
case GetCACertificate:
- return SCMSecurityResponse.newBuilder()
- .setCmdType(request.getCmdType())
- .setStatus(Status.OK)
- .setGetCertResponseProto(
- getCACertificate(request.getGetCACertificateRequest()))
- .build();
+ return scmSecurityResponse.setGetCertResponseProto(
+ getCACertificate(request.getGetCACertificateRequest())).build();
case GetOMCertificate:
- return SCMSecurityResponse.newBuilder()
- .setCmdType(request.getCmdType())
- .setStatus(Status.OK)
- .setGetCertResponseProto(
- getOMCertificate(request.getGetOMCertRequest()))
+ return scmSecurityResponse.setGetCertResponseProto(
+ getOMCertificate(request.getGetOMCertRequest()))
.build();
case GetDataNodeCertificate:
- return SCMSecurityResponse.newBuilder()
- .setCmdType(request.getCmdType())
- .setStatus(Status.OK)
- .setGetCertResponseProto(
- getDataNodeCertificate(request.getGetDataNodeCertRequest()))
+ return scmSecurityResponse.setGetCertResponseProto(
+ getDataNodeCertificate(request.getGetDataNodeCertRequest()))
.build();
case ListCertificate:
- return SCMSecurityResponse.newBuilder()
- .setCmdType(request.getCmdType())
- .setStatus(Status.OK)
- .setListCertificateResponseProto(
- listCertificate(request.getListCertificateRequest()))
+ return scmSecurityResponse.setListCertificateResponseProto(
+ listCertificate(request.getListCertificateRequest()))
.build();
case GetSCMCertificate:
- return SCMSecurityResponse.newBuilder()
- .setCmdType(request.getCmdType())
- .setStatus(Status.OK)
- .setGetCertResponseProto(getSCMCertificate(
- request.getGetSCMCertificateRequest()))
- .build();
+ return scmSecurityResponse.setGetCertResponseProto(getSCMCertificate(
+ request.getGetSCMCertificateRequest())).build();
default:
throw new IllegalArgumentException(
"Unknown request type: " + request.getCmdType());
}
} catch (IOException e) {
- throw new ServiceException(e);
+ scmSecurityResponse.setSuccess(false);
+ scmSecurityResponse.setStatus(exceptionToResponseStatus(e));
+ // If actual cause is set in SCMSecurityException, set message with
+ // actuall cause message.
+ if (e instanceof SCMSecurityException) {
+ if (e.getCause() != null && e.getCause().getMessage() != null) {
+ scmSecurityResponse.setMessage(e.getCause().getMessage());
+ return scmSecurityResponse.build();
+ }
+ }
+ if (e.getMessage() != null) {
Review comment:
Done
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
[GitHub] [ozone] bshashikant commented on a change in pull request #1978: HDDS-4876. [SCM HA Security] Add failover proxy to SCM Security Server Protocol
Posted by GitBox <gi...@apache.org>.
bshashikant commented on a change in pull request #1978:
URL: https://github.com/apache/ozone/pull/1978#discussion_r589254792
##########
File path: hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/HAUtils.java
##########
@@ -285,4 +288,27 @@ public static File getMetaDir(DBDefinition definition,
}
return metadataDir;
}
+
+ /**
+ * Unwrap exception to check if it is some kind of access control problem
+=======
Review comment:
The comments need to be fixed.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
[GitHub] [ozone] bharatviswa504 merged pull request #1978: HDDS-4876. [SCM HA Security] Add failover proxy to SCM Security Server Protocol
Posted by GitBox <gi...@apache.org>.
bharatviswa504 merged pull request #1978:
URL: https://github.com/apache/ozone/pull/1978
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
[GitHub] [ozone] xiaoyuyao commented on a change in pull request #1978: HDDS-4876. [SCM HA Security] Add failover proxy to SCM Security Server Protocol
Posted by GitBox <gi...@apache.org>.
xiaoyuyao commented on a change in pull request #1978:
URL: https://github.com/apache/ozone/pull/1978#discussion_r587898849
##########
File path: hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/security/exception/SCMSecurityException.java
##########
@@ -45,12 +55,24 @@ public SCMSecurityException(String message, Throwable cause) {
this.errorCode = ErrorCode.DEFAULT;
}
+ /**
+ * Ctor.
+ * @param message - Error Message
+ * @param cause - Actual cause.
+ * @param errorCode - Error code.
+ */
+ public SCMSecurityException(String message, Throwable cause,
+ ErrorCode errorCode) {
+ super(message, cause);
+ this.errorCode = errorCode;
+ }
+
/**
* Ctor.
* @param message - Message.
* @param error - error code.
*/
- public SCMSecurityException(String message, ErrorCode error) {
+ public SCMSecurityException(Exception message, ErrorCode error) {
Review comment:
NIT: can you change message to cause and the type to Throwable?
##########
File path: hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/proxy/SCMSecurityProtocolFailoverProxyProvider.java
##########
@@ -0,0 +1,279 @@
+/*
+ * 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.scm.proxy;
+
+import com.google.common.base.Preconditions;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdds.conf.ConfigurationException;
+import org.apache.hadoop.hdds.conf.ConfigurationSource;
+import org.apache.hadoop.hdds.protocolPB.SCMSecurityProtocolPB;
+import org.apache.hadoop.hdds.scm.ha.SCMNodeInfo;
+import org.apache.hadoop.hdds.utils.HAUtils;
+import org.apache.hadoop.hdds.utils.LegacyHadoopConfigurationSource;
+import org.apache.hadoop.io.retry.FailoverProxyProvider;
+import org.apache.hadoop.io.retry.RetryPolicies;
+import org.apache.hadoop.io.retry.RetryPolicy;
+import org.apache.hadoop.io.retry.RetryPolicy.RetryAction.RetryDecision;
+import org.apache.hadoop.ipc.ProtobufRpcEngine;
+import org.apache.hadoop.ipc.RPC;
+import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.hadoop.io.retry.RetryPolicy.RetryAction.RetryDecision.FAILOVER_AND_RETRY;
+
+/**
+ * Failover proxy provider for SCMSecurityProtocol server.
+ */
+public class SCMSecurityProtocolFailoverProxyProvider implements
+ FailoverProxyProvider<SCMSecurityProtocolPB>, Closeable {
+
+ public static final Logger LOG =
+ LoggerFactory.getLogger(SCMSecurityProtocolFailoverProxyProvider.class);
+
+ // scmNodeId -> ProxyInfo<rpcProxy>
+ private final Map<String,
+ ProxyInfo<SCMSecurityProtocolPB>> scmProxies;
+
+ // scmNodeId -> SCMProxyInfo
+ private final Map<String, SCMProxyInfo> scmProxyInfoMap;
+
+ private List<String> scmNodeIds;
+
+ private String currentProxySCMNodeId;
+ private int currentProxyIndex;
+
+ private final ConfigurationSource conf;
+ private final SCMClientConfig scmClientConfig;
+ private final long scmVersion;
+
+ private String scmServiceId;
+
+ private final int maxRetryCount;
+ private final long retryInterval;
+
+ private final UserGroupInformation ugi;
+
+ /**
+ * Construct fail-over proxy provider for SCMSecurityProtocol Server.
+ * @param conf
+ * @param userGroupInformation
+ */
+ public SCMSecurityProtocolFailoverProxyProvider(ConfigurationSource conf,
+ UserGroupInformation userGroupInformation) {
+ Preconditions.checkNotNull(userGroupInformation);
+ this.ugi = userGroupInformation;
+ this.conf = conf;
+ this.scmVersion = RPC.getProtocolVersion(SCMSecurityProtocolPB.class);
+
+ this.scmProxies = new HashMap<>();
+ this.scmProxyInfoMap = new HashMap<>();
+ loadConfigs();
+
+ this.currentProxyIndex = 0;
+ currentProxySCMNodeId = scmNodeIds.get(currentProxyIndex);
+ scmClientConfig = conf.getObject(SCMClientConfig.class);
+ this.maxRetryCount = scmClientConfig.getRetryCount();
+ this.retryInterval = scmClientConfig.getRetryInterval();
+ }
+
+ protected void loadConfigs() {
+ List<SCMNodeInfo> scmNodeInfoList = SCMNodeInfo.buildNodeInfo(conf);
+ scmNodeIds = new ArrayList<>();
+
+ for (SCMNodeInfo scmNodeInfo : scmNodeInfoList) {
+ if (scmNodeInfo.getScmSecurityAddress() == null) {
+ throw new ConfigurationException("SCM Client Address could not " +
+ "be obtained from config. Config is not properly defined");
+ } else {
+ InetSocketAddress scmSecurityAddress =
+ NetUtils.createSocketAddr(scmNodeInfo.getScmSecurityAddress());
+
+ scmServiceId = scmNodeInfo.getServiceId();
+ String scmNodeId = scmNodeInfo.getNodeId();
+
+ scmNodeIds.add(scmNodeId);
+ SCMProxyInfo scmProxyInfo = new SCMProxyInfo(scmServiceId, scmNodeId,
+ scmSecurityAddress);
+ scmProxyInfoMap.put(scmNodeId, scmProxyInfo);
+ }
+ }
+ }
+
+ @Override
+ public ProxyInfo<SCMSecurityProtocolPB> getProxy() {
Review comment:
Synchronized on the method for thread safe proxy creation.
##########
File path: hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/HAUtils.java
##########
@@ -285,4 +288,24 @@ public static File getMetaDir(DBDefinition definition,
}
return metadataDir;
}
+
+ /**
+ * Unwrap exception to check if it is some kind of access control problem.
+ * {@link AccessControlException}
+ */
+ public static boolean isAccessControlException(Exception ex) {
Review comment:
The same util exists in OmUtils, can we dedup the code with one copy?
##########
File path: hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/proxy/SCMSecurityProtocolFailoverProxyProvider.java
##########
@@ -0,0 +1,279 @@
+/*
+ * 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.scm.proxy;
+
+import com.google.common.base.Preconditions;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdds.conf.ConfigurationException;
+import org.apache.hadoop.hdds.conf.ConfigurationSource;
+import org.apache.hadoop.hdds.protocolPB.SCMSecurityProtocolPB;
+import org.apache.hadoop.hdds.scm.ha.SCMNodeInfo;
+import org.apache.hadoop.hdds.utils.HAUtils;
+import org.apache.hadoop.hdds.utils.LegacyHadoopConfigurationSource;
+import org.apache.hadoop.io.retry.FailoverProxyProvider;
+import org.apache.hadoop.io.retry.RetryPolicies;
+import org.apache.hadoop.io.retry.RetryPolicy;
+import org.apache.hadoop.io.retry.RetryPolicy.RetryAction.RetryDecision;
+import org.apache.hadoop.ipc.ProtobufRpcEngine;
+import org.apache.hadoop.ipc.RPC;
+import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.hadoop.io.retry.RetryPolicy.RetryAction.RetryDecision.FAILOVER_AND_RETRY;
+
+/**
+ * Failover proxy provider for SCMSecurityProtocol server.
+ */
+public class SCMSecurityProtocolFailoverProxyProvider implements
+ FailoverProxyProvider<SCMSecurityProtocolPB>, Closeable {
+
+ public static final Logger LOG =
+ LoggerFactory.getLogger(SCMSecurityProtocolFailoverProxyProvider.class);
+
+ // scmNodeId -> ProxyInfo<rpcProxy>
+ private final Map<String,
+ ProxyInfo<SCMSecurityProtocolPB>> scmProxies;
+
+ // scmNodeId -> SCMProxyInfo
+ private final Map<String, SCMProxyInfo> scmProxyInfoMap;
+
+ private List<String> scmNodeIds;
+
+ private String currentProxySCMNodeId;
+ private int currentProxyIndex;
+
+ private final ConfigurationSource conf;
+ private final SCMClientConfig scmClientConfig;
+ private final long scmVersion;
+
+ private String scmServiceId;
+
+ private final int maxRetryCount;
+ private final long retryInterval;
+
+ private final UserGroupInformation ugi;
+
+ /**
+ * Construct fail-over proxy provider for SCMSecurityProtocol Server.
+ * @param conf
+ * @param userGroupInformation
+ */
+ public SCMSecurityProtocolFailoverProxyProvider(ConfigurationSource conf,
+ UserGroupInformation userGroupInformation) {
+ Preconditions.checkNotNull(userGroupInformation);
+ this.ugi = userGroupInformation;
+ this.conf = conf;
+ this.scmVersion = RPC.getProtocolVersion(SCMSecurityProtocolPB.class);
+
+ this.scmProxies = new HashMap<>();
+ this.scmProxyInfoMap = new HashMap<>();
+ loadConfigs();
+
+ this.currentProxyIndex = 0;
+ currentProxySCMNodeId = scmNodeIds.get(currentProxyIndex);
+ scmClientConfig = conf.getObject(SCMClientConfig.class);
+ this.maxRetryCount = scmClientConfig.getRetryCount();
+ this.retryInterval = scmClientConfig.getRetryInterval();
+ }
+
+ protected void loadConfigs() {
+ List<SCMNodeInfo> scmNodeInfoList = SCMNodeInfo.buildNodeInfo(conf);
+ scmNodeIds = new ArrayList<>();
+
+ for (SCMNodeInfo scmNodeInfo : scmNodeInfoList) {
+ if (scmNodeInfo.getScmSecurityAddress() == null) {
+ throw new ConfigurationException("SCM Client Address could not " +
+ "be obtained from config. Config is not properly defined");
+ } else {
+ InetSocketAddress scmSecurityAddress =
+ NetUtils.createSocketAddr(scmNodeInfo.getScmSecurityAddress());
+
+ scmServiceId = scmNodeInfo.getServiceId();
+ String scmNodeId = scmNodeInfo.getNodeId();
+
+ scmNodeIds.add(scmNodeId);
+ SCMProxyInfo scmProxyInfo = new SCMProxyInfo(scmServiceId, scmNodeId,
+ scmSecurityAddress);
+ scmProxyInfoMap.put(scmNodeId, scmProxyInfo);
+ }
+ }
+ }
+
+ @Override
+ public ProxyInfo<SCMSecurityProtocolPB> getProxy() {
+ ProxyInfo currentProxyInfo = scmProxies.get(getCurrentProxySCMNodeId());
+ if (currentProxyInfo == null) {
+ currentProxyInfo = createSCMProxy(getCurrentProxySCMNodeId());
+ }
+ return currentProxyInfo;
+ }
+
+ /**
+ * Creates proxy object.
+ */
+ private ProxyInfo createSCMProxy(String nodeId) {
+ ProxyInfo proxyInfo;
+ SCMProxyInfo scmProxyInfo = scmProxyInfoMap.get(nodeId);
+ InetSocketAddress address = scmProxyInfo.getAddress();
+ try {
+ SCMSecurityProtocolPB scmProxy = createSCMProxy(address);
+ // Create proxyInfo here, to make it work with all Hadoop versions.
+ proxyInfo = new ProxyInfo<>(scmProxy, scmProxyInfo.toString());
+ scmProxies.put(nodeId, proxyInfo);
+ return proxyInfo;
+ } catch (IOException ioe) {
+ LOG.error("{} Failed to create RPC proxy to SCM at {}",
+ this.getClass().getSimpleName(), address, ioe);
+ throw new RuntimeException(ioe);
+ }
+ }
+
+ private SCMSecurityProtocolPB createSCMProxy(InetSocketAddress scmAddress)
+ throws IOException {
+ Configuration hadoopConf =
+ LegacyHadoopConfigurationSource.asHadoopConfiguration(conf);
+ RPC.setProtocolEngine(hadoopConf, SCMSecurityProtocolPB.class,
+ ProtobufRpcEngine.class);
+
+ // FailoverOnNetworkException ensures that the IPC layer does not attempt
+ // retries on the same SCM in case of connection exception. This retry
+ // policy essentially results in TRY_ONCE_THEN_FAIL.
+
+ RetryPolicy connectionRetryPolicy = RetryPolicies
+ .failoverOnNetworkException(0);
+
+ return RPC.getProtocolProxy(SCMSecurityProtocolPB.class,
+ scmVersion, scmAddress, ugi,
+ hadoopConf, NetUtils.getDefaultSocketFactory(hadoopConf),
+ (int)scmClientConfig.getRpcTimeOut(), connectionRetryPolicy).getProxy();
+ }
+
+
+ @Override
+ public void performFailover(SCMSecurityProtocolPB currentProxy) {
+ if (LOG.isDebugEnabled()) {
+ int currentIndex = getCurrentProxyIndex();
+ LOG.debug("Failing over SCM Security proxy to index: {}, nodeId: {}",
+ currentIndex, scmNodeIds.get(currentIndex));
+ }
+ }
+
+ /**
+ * Performs fail-over to the next proxy.
+ */
+ public void performFailoverToNextProxy() {
+ int newProxyIndex = incrementProxyIndex();
Review comment:
Do we handle failover with leader hints in addition to RR failover?
##########
File path: hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/proxy/SCMSecurityProtocolFailoverProxyProvider.java
##########
@@ -0,0 +1,279 @@
+/*
+ * 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.scm.proxy;
+
+import com.google.common.base.Preconditions;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdds.conf.ConfigurationException;
+import org.apache.hadoop.hdds.conf.ConfigurationSource;
+import org.apache.hadoop.hdds.protocolPB.SCMSecurityProtocolPB;
+import org.apache.hadoop.hdds.scm.ha.SCMNodeInfo;
+import org.apache.hadoop.hdds.utils.HAUtils;
+import org.apache.hadoop.hdds.utils.LegacyHadoopConfigurationSource;
+import org.apache.hadoop.io.retry.FailoverProxyProvider;
+import org.apache.hadoop.io.retry.RetryPolicies;
+import org.apache.hadoop.io.retry.RetryPolicy;
+import org.apache.hadoop.io.retry.RetryPolicy.RetryAction.RetryDecision;
+import org.apache.hadoop.ipc.ProtobufRpcEngine;
+import org.apache.hadoop.ipc.RPC;
+import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.hadoop.io.retry.RetryPolicy.RetryAction.RetryDecision.FAILOVER_AND_RETRY;
+
+/**
+ * Failover proxy provider for SCMSecurityProtocol server.
+ */
+public class SCMSecurityProtocolFailoverProxyProvider implements
+ FailoverProxyProvider<SCMSecurityProtocolPB>, Closeable {
+
+ public static final Logger LOG =
+ LoggerFactory.getLogger(SCMSecurityProtocolFailoverProxyProvider.class);
+
+ // scmNodeId -> ProxyInfo<rpcProxy>
+ private final Map<String,
+ ProxyInfo<SCMSecurityProtocolPB>> scmProxies;
+
+ // scmNodeId -> SCMProxyInfo
+ private final Map<String, SCMProxyInfo> scmProxyInfoMap;
+
+ private List<String> scmNodeIds;
+
+ private String currentProxySCMNodeId;
+ private int currentProxyIndex;
+
+ private final ConfigurationSource conf;
+ private final SCMClientConfig scmClientConfig;
+ private final long scmVersion;
+
+ private String scmServiceId;
+
+ private final int maxRetryCount;
+ private final long retryInterval;
+
+ private final UserGroupInformation ugi;
+
+ /**
+ * Construct fail-over proxy provider for SCMSecurityProtocol Server.
+ * @param conf
+ * @param userGroupInformation
+ */
+ public SCMSecurityProtocolFailoverProxyProvider(ConfigurationSource conf,
+ UserGroupInformation userGroupInformation) {
+ Preconditions.checkNotNull(userGroupInformation);
+ this.ugi = userGroupInformation;
+ this.conf = conf;
+ this.scmVersion = RPC.getProtocolVersion(SCMSecurityProtocolPB.class);
+
+ this.scmProxies = new HashMap<>();
+ this.scmProxyInfoMap = new HashMap<>();
+ loadConfigs();
+
+ this.currentProxyIndex = 0;
+ currentProxySCMNodeId = scmNodeIds.get(currentProxyIndex);
+ scmClientConfig = conf.getObject(SCMClientConfig.class);
+ this.maxRetryCount = scmClientConfig.getRetryCount();
+ this.retryInterval = scmClientConfig.getRetryInterval();
+ }
+
+ protected void loadConfigs() {
+ List<SCMNodeInfo> scmNodeInfoList = SCMNodeInfo.buildNodeInfo(conf);
+ scmNodeIds = new ArrayList<>();
+
+ for (SCMNodeInfo scmNodeInfo : scmNodeInfoList) {
+ if (scmNodeInfo.getScmSecurityAddress() == null) {
+ throw new ConfigurationException("SCM Client Address could not " +
+ "be obtained from config. Config is not properly defined");
+ } else {
+ InetSocketAddress scmSecurityAddress =
+ NetUtils.createSocketAddr(scmNodeInfo.getScmSecurityAddress());
+
+ scmServiceId = scmNodeInfo.getServiceId();
+ String scmNodeId = scmNodeInfo.getNodeId();
+
+ scmNodeIds.add(scmNodeId);
+ SCMProxyInfo scmProxyInfo = new SCMProxyInfo(scmServiceId, scmNodeId,
+ scmSecurityAddress);
+ scmProxyInfoMap.put(scmNodeId, scmProxyInfo);
+ }
+ }
+ }
+
+ @Override
+ public ProxyInfo<SCMSecurityProtocolPB> getProxy() {
+ ProxyInfo currentProxyInfo = scmProxies.get(getCurrentProxySCMNodeId());
+ if (currentProxyInfo == null) {
+ currentProxyInfo = createSCMProxy(getCurrentProxySCMNodeId());
+ }
+ return currentProxyInfo;
+ }
+
+ /**
+ * Creates proxy object.
+ */
+ private ProxyInfo createSCMProxy(String nodeId) {
+ ProxyInfo proxyInfo;
+ SCMProxyInfo scmProxyInfo = scmProxyInfoMap.get(nodeId);
+ InetSocketAddress address = scmProxyInfo.getAddress();
+ try {
+ SCMSecurityProtocolPB scmProxy = createSCMProxy(address);
+ // Create proxyInfo here, to make it work with all Hadoop versions.
+ proxyInfo = new ProxyInfo<>(scmProxy, scmProxyInfo.toString());
+ scmProxies.put(nodeId, proxyInfo);
+ return proxyInfo;
+ } catch (IOException ioe) {
+ LOG.error("{} Failed to create RPC proxy to SCM at {}",
+ this.getClass().getSimpleName(), address, ioe);
+ throw new RuntimeException(ioe);
+ }
+ }
+
+ private SCMSecurityProtocolPB createSCMProxy(InetSocketAddress scmAddress)
+ throws IOException {
+ Configuration hadoopConf =
+ LegacyHadoopConfigurationSource.asHadoopConfiguration(conf);
+ RPC.setProtocolEngine(hadoopConf, SCMSecurityProtocolPB.class,
+ ProtobufRpcEngine.class);
+
+ // FailoverOnNetworkException ensures that the IPC layer does not attempt
+ // retries on the same SCM in case of connection exception. This retry
+ // policy essentially results in TRY_ONCE_THEN_FAIL.
+
+ RetryPolicy connectionRetryPolicy = RetryPolicies
+ .failoverOnNetworkException(0);
+
+ return RPC.getProtocolProxy(SCMSecurityProtocolPB.class,
+ scmVersion, scmAddress, ugi,
+ hadoopConf, NetUtils.getDefaultSocketFactory(hadoopConf),
+ (int)scmClientConfig.getRpcTimeOut(), connectionRetryPolicy).getProxy();
+ }
+
+
+ @Override
+ public void performFailover(SCMSecurityProtocolPB currentProxy) {
+ if (LOG.isDebugEnabled()) {
+ int currentIndex = getCurrentProxyIndex();
+ LOG.debug("Failing over SCM Security proxy to index: {}, nodeId: {}",
+ currentIndex, scmNodeIds.get(currentIndex));
+ }
+ }
+
+ /**
+ * Performs fail-over to the next proxy.
+ */
+ public void performFailoverToNextProxy() {
+ int newProxyIndex = incrementProxyIndex();
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Incrementing SCM Security proxy index to {}, nodeId: {}",
+ newProxyIndex, scmNodeIds.get(newProxyIndex));
+ }
+ }
+
+ /**
+ * Update the proxy index to the next proxy in the list.
+ * @return the new proxy index
+ */
+ private synchronized int incrementProxyIndex() {
+ currentProxyIndex = (currentProxyIndex + 1) % scmProxies.size();
+ currentProxySCMNodeId = scmNodeIds.get(currentProxyIndex);
+ return currentProxyIndex;
+ }
+
+ public RetryPolicy getRetryPolicy() {
+ // Client will attempt up to maxFailovers number of failovers between
+ // available SCMs before throwing exception.
+ RetryPolicy retryPolicy = new RetryPolicy() {
+ @Override
+ public RetryAction shouldRetry(Exception exception, int retries,
+ int failovers, boolean isIdempotentOrAtMostOnce)
+ throws Exception {
+
+ if (LOG.isDebugEnabled()) {
+ if (exception.getCause() != null) {
+ LOG.debug("RetryProxy: SCM Security Server {}: {}: {}",
+ getCurrentProxySCMNodeId(),
+ exception.getCause().getClass().getSimpleName(),
+ exception.getCause().getMessage());
+ } else {
+ LOG.debug("RetryProxy: OM {}: {}", getCurrentProxySCMNodeId(),
+ exception.getMessage());
+ }
+ }
+
+ // For AccessControl Exception where Client is not authentica
+ if (HAUtils.isAccessControlException(exception)) {
+ return RetryAction.FAIL;
+ }
+
+ // Perform fail over to next proxy, as right now we don't have any
+ // suggested leader ID from server, we fail over to next one.
+ // TODO: Act based on server response if leader id is passed.
+ performFailoverToNextProxy();
+ return getRetryAction(FAILOVER_AND_RETRY, failovers);
+ }
+
+ private RetryAction getRetryAction(RetryDecision fallbackAction,
+ int failovers) {
+ if (failovers < maxRetryCount) {
+ return new RetryAction(fallbackAction, getRetryInterval());
+ } else {
+ return RetryAction.FAIL;
+ }
+ }
+ };
+
+ return retryPolicy;
+ }
+
+
+ @Override
+ public Class< SCMSecurityProtocolPB > getInterface() {
+ return SCMSecurityProtocolPB.class;
+ }
+
+ @Override
+ public void close() throws IOException {
+ for (ProxyInfo<SCMSecurityProtocolPB> proxyInfo : scmProxies.values()) {
+ if (proxyInfo != null) {
+ RPC.stopProxy(proxyInfo.proxy);
Review comment:
can we go through the proxy map, call stopProxy and null the value in the map?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
[GitHub] [ozone] bharatviswa504 commented on pull request #1978: HDDS-4876. [SCM HA Security] Add failover proxy to SCM Security Server Protocol
Posted by GitBox <gi...@apache.org>.
bharatviswa504 commented on pull request #1978:
URL: https://github.com/apache/ozone/pull/1978#issuecomment-793607385
Thank You @xiaoyuyao for the review addressed/replied to review comments.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
[GitHub] [ozone] bharatviswa504 commented on a change in pull request #1978: HDDS-4876. [SCM HA Security] Add failover proxy to SCM Security Server Protocol
Posted by GitBox <gi...@apache.org>.
bharatviswa504 commented on a change in pull request #1978:
URL: https://github.com/apache/ozone/pull/1978#discussion_r590065940
##########
File path: hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/HAUtils.java
##########
@@ -285,4 +288,24 @@ public static File getMetaDir(DBDefinition definition,
}
return metadataDir;
}
+
+ /**
+ * Unwrap exception to check if it is some kind of access control problem.
+ * {@link AccessControlException}
+ */
+ public static boolean isAccessControlException(Exception ex) {
Review comment:
It has additional InvalidToken, for that OM has special logic, where it retries all 3 OMs atleast once. So, added a new method.
When SCM also has similar logic for TokenInvalid we can have a single copy of this method.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
[GitHub] [ozone] bharatviswa504 commented on pull request #1978: HDDS-4876. [SCM HA Security] Add failover proxy to SCM Security Server Protocol
Posted by GitBox <gi...@apache.org>.
bharatviswa504 commented on pull request #1978:
URL: https://github.com/apache/ozone/pull/1978#issuecomment-794230102
Thank You @xiaoyuyao and @bshashikant for the review.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
[GitHub] [ozone] bharatviswa504 commented on a change in pull request #1978: HDDS-4876. [SCM HA Security] Add failover proxy to SCM Security Server Protocol
Posted by GitBox <gi...@apache.org>.
bharatviswa504 commented on a change in pull request #1978:
URL: https://github.com/apache/ozone/pull/1978#discussion_r587540276
##########
File path: hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/proxy/SCMSecurityProtocolFailoverProxyProvider.java
##########
@@ -0,0 +1,279 @@
+/*
+ * 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.scm.proxy;
+
+import com.google.common.base.Preconditions;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdds.conf.ConfigurationException;
+import org.apache.hadoop.hdds.conf.ConfigurationSource;
+import org.apache.hadoop.hdds.protocolPB.SCMSecurityProtocolPB;
+import org.apache.hadoop.hdds.scm.ha.SCMNodeInfo;
+import org.apache.hadoop.hdds.utils.HAUtils;
+import org.apache.hadoop.hdds.utils.LegacyHadoopConfigurationSource;
+import org.apache.hadoop.io.retry.FailoverProxyProvider;
+import org.apache.hadoop.io.retry.RetryPolicies;
+import org.apache.hadoop.io.retry.RetryPolicy;
+import org.apache.hadoop.io.retry.RetryPolicy.RetryAction.RetryDecision;
+import org.apache.hadoop.ipc.ProtobufRpcEngine;
+import org.apache.hadoop.ipc.RPC;
+import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.hadoop.io.retry.RetryPolicy.RetryAction.RetryDecision.FAILOVER_AND_RETRY;
+
+/**
+ * Failover proxy provider for SCMSecurityProtocol server.
+ */
+public class SCMSecurityProtocolFailoverProxyProvider implements
+ FailoverProxyProvider<SCMSecurityProtocolPB>, Closeable {
+
+ public static final Logger LOG =
+ LoggerFactory.getLogger(SCMSecurityProtocolFailoverProxyProvider.class);
+
+ // scmNodeId -> ProxyInfo<rpcProxy>
+ private final Map<String,
+ ProxyInfo<SCMSecurityProtocolPB>> scmProxies;
+
+ // scmNodeId -> SCMProxyInfo
+ private final Map<String, SCMProxyInfo> scmProxyInfoMap;
+
+ private List<String> scmNodeIds;
+
+ private String currentProxySCMNodeId;
+ private int currentProxyIndex;
+
+ private final ConfigurationSource conf;
+ private final SCMClientConfig scmClientConfig;
+ private final long scmVersion;
+
+ private String scmServiceId;
+
+ private final int maxRetryCount;
+ private final long retryInterval;
+
+ private final UserGroupInformation ugi;
+
+ /**
+ * Construct fail-over proxy provider for SCMSecurityProtocol Server.
+ * @param conf
+ * @param userGroupInformation
+ */
+ public SCMSecurityProtocolFailoverProxyProvider(ConfigurationSource conf,
+ UserGroupInformation userGroupInformation) {
+ Preconditions.checkNotNull(userGroupInformation);
+ this.ugi = userGroupInformation;
+ this.conf = conf;
+ this.scmVersion = RPC.getProtocolVersion(SCMSecurityProtocolPB.class);
+
+ this.scmProxies = new HashMap<>();
+ this.scmProxyInfoMap = new HashMap<>();
+ loadConfigs();
+
+ this.currentProxyIndex = 0;
+ currentProxySCMNodeId = scmNodeIds.get(currentProxyIndex);
+ scmClientConfig = conf.getObject(SCMClientConfig.class);
+ this.maxRetryCount = scmClientConfig.getRetryCount();
+ this.retryInterval = scmClientConfig.getRetryInterval();
+ }
+
+ protected void loadConfigs() {
+ List<SCMNodeInfo> scmNodeInfoList = SCMNodeInfo.buildNodeInfo(conf);
+ scmNodeIds = new ArrayList<>();
+
+ for (SCMNodeInfo scmNodeInfo : scmNodeInfoList) {
+ if (scmNodeInfo.getScmSecurityAddress() == null) {
Review comment:
When serviceId and nodeId are not defined, we use some const values and make the proxy failover work for HA/non-HA.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
[GitHub] [ozone] xiaoyuyao commented on a change in pull request #1978: HDDS-4876. [SCM HA Security] Add failover proxy to SCM Security Server Protocol
Posted by GitBox <gi...@apache.org>.
xiaoyuyao commented on a change in pull request #1978:
URL: https://github.com/apache/ozone/pull/1978#discussion_r589703002
##########
File path: hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/proxy/SCMSecurityProtocolFailoverProxyProvider.java
##########
@@ -0,0 +1,279 @@
+/*
+ * 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.scm.proxy;
+
+import com.google.common.base.Preconditions;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdds.conf.ConfigurationException;
+import org.apache.hadoop.hdds.conf.ConfigurationSource;
+import org.apache.hadoop.hdds.protocolPB.SCMSecurityProtocolPB;
+import org.apache.hadoop.hdds.scm.ha.SCMNodeInfo;
+import org.apache.hadoop.hdds.utils.HAUtils;
+import org.apache.hadoop.hdds.utils.LegacyHadoopConfigurationSource;
+import org.apache.hadoop.io.retry.FailoverProxyProvider;
+import org.apache.hadoop.io.retry.RetryPolicies;
+import org.apache.hadoop.io.retry.RetryPolicy;
+import org.apache.hadoop.io.retry.RetryPolicy.RetryAction.RetryDecision;
+import org.apache.hadoop.ipc.ProtobufRpcEngine;
+import org.apache.hadoop.ipc.RPC;
+import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.hadoop.io.retry.RetryPolicy.RetryAction.RetryDecision.FAILOVER_AND_RETRY;
+
+/**
+ * Failover proxy provider for SCMSecurityProtocol server.
+ */
+public class SCMSecurityProtocolFailoverProxyProvider implements
+ FailoverProxyProvider<SCMSecurityProtocolPB>, Closeable {
+
+ public static final Logger LOG =
+ LoggerFactory.getLogger(SCMSecurityProtocolFailoverProxyProvider.class);
+
+ // scmNodeId -> ProxyInfo<rpcProxy>
+ private final Map<String,
+ ProxyInfo<SCMSecurityProtocolPB>> scmProxies;
+
+ // scmNodeId -> SCMProxyInfo
+ private final Map<String, SCMProxyInfo> scmProxyInfoMap;
+
+ private List<String> scmNodeIds;
+
+ private String currentProxySCMNodeId;
+ private int currentProxyIndex;
+
+ private final ConfigurationSource conf;
+ private final SCMClientConfig scmClientConfig;
+ private final long scmVersion;
+
+ private String scmServiceId;
+
+ private final int maxRetryCount;
+ private final long retryInterval;
+
+ private final UserGroupInformation ugi;
+
+ /**
+ * Construct fail-over proxy provider for SCMSecurityProtocol Server.
+ * @param conf
+ * @param userGroupInformation
+ */
+ public SCMSecurityProtocolFailoverProxyProvider(ConfigurationSource conf,
+ UserGroupInformation userGroupInformation) {
+ Preconditions.checkNotNull(userGroupInformation);
+ this.ugi = userGroupInformation;
+ this.conf = conf;
+ this.scmVersion = RPC.getProtocolVersion(SCMSecurityProtocolPB.class);
+
+ this.scmProxies = new HashMap<>();
+ this.scmProxyInfoMap = new HashMap<>();
+ loadConfigs();
+
+ this.currentProxyIndex = 0;
+ currentProxySCMNodeId = scmNodeIds.get(currentProxyIndex);
+ scmClientConfig = conf.getObject(SCMClientConfig.class);
+ this.maxRetryCount = scmClientConfig.getRetryCount();
+ this.retryInterval = scmClientConfig.getRetryInterval();
+ }
+
+ protected void loadConfigs() {
+ List<SCMNodeInfo> scmNodeInfoList = SCMNodeInfo.buildNodeInfo(conf);
+ scmNodeIds = new ArrayList<>();
+
+ for (SCMNodeInfo scmNodeInfo : scmNodeInfoList) {
+ if (scmNodeInfo.getScmSecurityAddress() == null) {
+ throw new ConfigurationException("SCM Client Address could not " +
+ "be obtained from config. Config is not properly defined");
+ } else {
+ InetSocketAddress scmSecurityAddress =
+ NetUtils.createSocketAddr(scmNodeInfo.getScmSecurityAddress());
+
+ scmServiceId = scmNodeInfo.getServiceId();
Review comment:
Do we have a separate configuration for serviceId? If yes, we should assert the one from nodeinfo match with the smcServiceId from config instead of rewriting it each time a new node is loaded.
##########
File path: hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/protocol/SCMSecurityProtocolServerSideTranslatorPB.java
##########
@@ -70,62 +75,75 @@ public SCMSecurityProtocolServerSideTranslatorPB(SCMSecurityProtocol impl,
@Override
public SCMSecurityResponse submitRequest(RpcController controller,
SCMSecurityRequest request) throws ServiceException {
+ if (!scm.checkLeader()) {
+ throw new ServiceException(scm.getScmHAManager()
+ .getRatisServer()
+ .triggerNotLeaderException());
+ }
return dispatcher.processRequest(request, this::processRequest,
request.getCmdType(), request.getTraceID());
}
- public SCMSecurityResponse processRequest(SCMSecurityRequest request)
- throws ServiceException {
+ public SCMSecurityResponse processRequest(SCMSecurityRequest request) {
+ SCMSecurityResponse.Builder scmSecurityResponse =
+ SCMSecurityResponse.newBuilder().setCmdType(request.getCmdType())
+ .setStatus(Status.OK);
try {
switch (request.getCmdType()) {
case GetCertificate:
- return SCMSecurityResponse.newBuilder()
- .setCmdType(request.getCmdType())
- .setStatus(Status.OK)
- .setGetCertResponseProto(
- getCertificate(request.getGetCertificateRequest()))
- .build();
+ return scmSecurityResponse.setGetCertResponseProto(
+ getCertificate(request.getGetCertificateRequest())).build();
case GetCACertificate:
- return SCMSecurityResponse.newBuilder()
- .setCmdType(request.getCmdType())
- .setStatus(Status.OK)
- .setGetCertResponseProto(
- getCACertificate(request.getGetCACertificateRequest()))
- .build();
+ return scmSecurityResponse.setGetCertResponseProto(
+ getCACertificate(request.getGetCACertificateRequest())).build();
case GetOMCertificate:
- return SCMSecurityResponse.newBuilder()
- .setCmdType(request.getCmdType())
- .setStatus(Status.OK)
- .setGetCertResponseProto(
- getOMCertificate(request.getGetOMCertRequest()))
+ return scmSecurityResponse.setGetCertResponseProto(
+ getOMCertificate(request.getGetOMCertRequest()))
.build();
case GetDataNodeCertificate:
- return SCMSecurityResponse.newBuilder()
- .setCmdType(request.getCmdType())
- .setStatus(Status.OK)
- .setGetCertResponseProto(
- getDataNodeCertificate(request.getGetDataNodeCertRequest()))
+ return scmSecurityResponse.setGetCertResponseProto(
+ getDataNodeCertificate(request.getGetDataNodeCertRequest()))
.build();
case ListCertificate:
- return SCMSecurityResponse.newBuilder()
- .setCmdType(request.getCmdType())
- .setStatus(Status.OK)
- .setListCertificateResponseProto(
- listCertificate(request.getListCertificateRequest()))
+ return scmSecurityResponse.setListCertificateResponseProto(
+ listCertificate(request.getListCertificateRequest()))
.build();
case GetSCMCertificate:
- return SCMSecurityResponse.newBuilder()
- .setCmdType(request.getCmdType())
- .setStatus(Status.OK)
- .setGetCertResponseProto(getSCMCertificate(
- request.getGetSCMCertificateRequest()))
- .build();
+ return scmSecurityResponse.setGetCertResponseProto(getSCMCertificate(
+ request.getGetSCMCertificateRequest())).build();
default:
throw new IllegalArgumentException(
"Unknown request type: " + request.getCmdType());
}
} catch (IOException e) {
- throw new ServiceException(e);
+ scmSecurityResponse.setSuccess(false);
+ scmSecurityResponse.setStatus(exceptionToResponseStatus(e));
+ // If actual cause is set in SCMSecurityException, set message with
+ // actuall cause message.
+ if (e instanceof SCMSecurityException) {
+ if (e.getCause() != null && e.getCause().getMessage() != null) {
+ scmSecurityResponse.setMessage(e.getCause().getMessage());
+ return scmSecurityResponse.build();
+ }
+ }
+ if (e.getMessage() != null) {
Review comment:
should we switch the order of 129-130, and 123-128 so that the exception message from the cause can be preserved in scmSecurityResponse.message?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
[GitHub] [ozone] bharatviswa504 commented on a change in pull request #1978: HDDS-4876. [SCM HA Security] Add failover proxy to SCM Security Server Protocol
Posted by GitBox <gi...@apache.org>.
bharatviswa504 commented on a change in pull request #1978:
URL: https://github.com/apache/ozone/pull/1978#discussion_r590065940
##########
File path: hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/HAUtils.java
##########
@@ -285,4 +288,24 @@ public static File getMetaDir(DBDefinition definition,
}
return metadataDir;
}
+
+ /**
+ * Unwrap exception to check if it is some kind of access control problem.
+ * {@link AccessControlException}
+ */
+ public static boolean isAccessControlException(Exception ex) {
Review comment:
It has additional TokenValidation, for that OM has special logic. So, added a new method.
When SCM also has similar logic for TokenInvalid we can have a single copy of this method.
##########
File path: hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/HAUtils.java
##########
@@ -285,4 +288,24 @@ public static File getMetaDir(DBDefinition definition,
}
return metadataDir;
}
+
+ /**
+ * Unwrap exception to check if it is some kind of access control problem.
+ * {@link AccessControlException}
+ */
+ public static boolean isAccessControlException(Exception ex) {
Review comment:
It has additional InvalidToken, for that OM has special logic. So, added a new method.
When SCM also has similar logic for TokenInvalid we can have a single copy of this method.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
[GitHub] [ozone] xiaoyuyao commented on pull request #1978: HDDS-4876. [SCM HA Security] Add failover proxy to SCM Security Server Protocol
Posted by GitBox <gi...@apache.org>.
xiaoyuyao commented on pull request #1978:
URL: https://github.com/apache/ozone/pull/1978#issuecomment-794131119
Thanks @bharatviswa504 for the update. LGTM, +1.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
[GitHub] [ozone] bharatviswa504 commented on a change in pull request #1978: HDDS-4876. [SCM HA Security] Add failover proxy to SCM Security Server Protocol
Posted by GitBox <gi...@apache.org>.
bharatviswa504 commented on a change in pull request #1978:
URL: https://github.com/apache/ozone/pull/1978#discussion_r590118326
##########
File path: hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/proxy/SCMSecurityProtocolFailoverProxyProvider.java
##########
@@ -0,0 +1,279 @@
+/*
+ * 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.scm.proxy;
+
+import com.google.common.base.Preconditions;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdds.conf.ConfigurationException;
+import org.apache.hadoop.hdds.conf.ConfigurationSource;
+import org.apache.hadoop.hdds.protocolPB.SCMSecurityProtocolPB;
+import org.apache.hadoop.hdds.scm.ha.SCMNodeInfo;
+import org.apache.hadoop.hdds.utils.HAUtils;
+import org.apache.hadoop.hdds.utils.LegacyHadoopConfigurationSource;
+import org.apache.hadoop.io.retry.FailoverProxyProvider;
+import org.apache.hadoop.io.retry.RetryPolicies;
+import org.apache.hadoop.io.retry.RetryPolicy;
+import org.apache.hadoop.io.retry.RetryPolicy.RetryAction.RetryDecision;
+import org.apache.hadoop.ipc.ProtobufRpcEngine;
+import org.apache.hadoop.ipc.RPC;
+import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.hadoop.io.retry.RetryPolicy.RetryAction.RetryDecision.FAILOVER_AND_RETRY;
+
+/**
+ * Failover proxy provider for SCMSecurityProtocol server.
+ */
+public class SCMSecurityProtocolFailoverProxyProvider implements
+ FailoverProxyProvider<SCMSecurityProtocolPB>, Closeable {
+
+ public static final Logger LOG =
+ LoggerFactory.getLogger(SCMSecurityProtocolFailoverProxyProvider.class);
+
+ // scmNodeId -> ProxyInfo<rpcProxy>
+ private final Map<String,
+ ProxyInfo<SCMSecurityProtocolPB>> scmProxies;
+
+ // scmNodeId -> SCMProxyInfo
+ private final Map<String, SCMProxyInfo> scmProxyInfoMap;
+
+ private List<String> scmNodeIds;
+
+ private String currentProxySCMNodeId;
+ private int currentProxyIndex;
+
+ private final ConfigurationSource conf;
+ private final SCMClientConfig scmClientConfig;
+ private final long scmVersion;
+
+ private String scmServiceId;
+
+ private final int maxRetryCount;
+ private final long retryInterval;
+
+ private final UserGroupInformation ugi;
+
+ /**
+ * Construct fail-over proxy provider for SCMSecurityProtocol Server.
+ * @param conf
+ * @param userGroupInformation
+ */
+ public SCMSecurityProtocolFailoverProxyProvider(ConfigurationSource conf,
+ UserGroupInformation userGroupInformation) {
+ Preconditions.checkNotNull(userGroupInformation);
+ this.ugi = userGroupInformation;
+ this.conf = conf;
+ this.scmVersion = RPC.getProtocolVersion(SCMSecurityProtocolPB.class);
+
+ this.scmProxies = new HashMap<>();
+ this.scmProxyInfoMap = new HashMap<>();
+ loadConfigs();
+
+ this.currentProxyIndex = 0;
+ currentProxySCMNodeId = scmNodeIds.get(currentProxyIndex);
+ scmClientConfig = conf.getObject(SCMClientConfig.class);
+ this.maxRetryCount = scmClientConfig.getRetryCount();
+ this.retryInterval = scmClientConfig.getRetryInterval();
+ }
+
+ protected void loadConfigs() {
+ List<SCMNodeInfo> scmNodeInfoList = SCMNodeInfo.buildNodeInfo(conf);
+ scmNodeIds = new ArrayList<>();
+
+ for (SCMNodeInfo scmNodeInfo : scmNodeInfoList) {
+ if (scmNodeInfo.getScmSecurityAddress() == null) {
+ throw new ConfigurationException("SCM Client Address could not " +
+ "be obtained from config. Config is not properly defined");
+ } else {
+ InetSocketAddress scmSecurityAddress =
+ NetUtils.createSocketAddr(scmNodeInfo.getScmSecurityAddress());
+
+ scmServiceId = scmNodeInfo.getServiceId();
+ String scmNodeId = scmNodeInfo.getNodeId();
+
+ scmNodeIds.add(scmNodeId);
+ SCMProxyInfo scmProxyInfo = new SCMProxyInfo(scmServiceId, scmNodeId,
+ scmSecurityAddress);
+ scmProxyInfoMap.put(scmNodeId, scmProxyInfo);
+ }
+ }
+ }
+
+ @Override
+ public ProxyInfo<SCMSecurityProtocolPB> getProxy() {
+ ProxyInfo currentProxyInfo = scmProxies.get(getCurrentProxySCMNodeId());
+ if (currentProxyInfo == null) {
+ currentProxyInfo = createSCMProxy(getCurrentProxySCMNodeId());
+ }
+ return currentProxyInfo;
+ }
+
+ /**
+ * Creates proxy object.
+ */
+ private ProxyInfo createSCMProxy(String nodeId) {
+ ProxyInfo proxyInfo;
+ SCMProxyInfo scmProxyInfo = scmProxyInfoMap.get(nodeId);
+ InetSocketAddress address = scmProxyInfo.getAddress();
+ try {
+ SCMSecurityProtocolPB scmProxy = createSCMProxy(address);
+ // Create proxyInfo here, to make it work with all Hadoop versions.
+ proxyInfo = new ProxyInfo<>(scmProxy, scmProxyInfo.toString());
+ scmProxies.put(nodeId, proxyInfo);
+ return proxyInfo;
+ } catch (IOException ioe) {
+ LOG.error("{} Failed to create RPC proxy to SCM at {}",
+ this.getClass().getSimpleName(), address, ioe);
+ throw new RuntimeException(ioe);
+ }
+ }
+
+ private SCMSecurityProtocolPB createSCMProxy(InetSocketAddress scmAddress)
+ throws IOException {
+ Configuration hadoopConf =
+ LegacyHadoopConfigurationSource.asHadoopConfiguration(conf);
+ RPC.setProtocolEngine(hadoopConf, SCMSecurityProtocolPB.class,
+ ProtobufRpcEngine.class);
+
+ // FailoverOnNetworkException ensures that the IPC layer does not attempt
+ // retries on the same SCM in case of connection exception. This retry
+ // policy essentially results in TRY_ONCE_THEN_FAIL.
+
+ RetryPolicy connectionRetryPolicy = RetryPolicies
+ .failoverOnNetworkException(0);
+
+ return RPC.getProtocolProxy(SCMSecurityProtocolPB.class,
+ scmVersion, scmAddress, ugi,
+ hadoopConf, NetUtils.getDefaultSocketFactory(hadoopConf),
+ (int)scmClientConfig.getRpcTimeOut(), connectionRetryPolicy).getProxy();
+ }
+
+
+ @Override
+ public void performFailover(SCMSecurityProtocolPB currentProxy) {
+ if (LOG.isDebugEnabled()) {
+ int currentIndex = getCurrentProxyIndex();
+ LOG.debug("Failing over SCM Security proxy to index: {}, nodeId: {}",
+ currentIndex, scmNodeIds.get(currentIndex));
+ }
+ }
+
+ /**
+ * Performs fail-over to the next proxy.
+ */
+ public void performFailoverToNextProxy() {
+ int newProxyIndex = incrementProxyIndex();
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Incrementing SCM Security proxy index to {}, nodeId: {}",
+ newProxyIndex, scmNodeIds.get(newProxyIndex));
+ }
+ }
+
+ /**
+ * Update the proxy index to the next proxy in the list.
+ * @return the new proxy index
+ */
+ private synchronized int incrementProxyIndex() {
+ currentProxyIndex = (currentProxyIndex + 1) % scmProxies.size();
+ currentProxySCMNodeId = scmNodeIds.get(currentProxyIndex);
+ return currentProxyIndex;
+ }
+
+ public RetryPolicy getRetryPolicy() {
+ // Client will attempt up to maxFailovers number of failovers between
+ // available SCMs before throwing exception.
+ RetryPolicy retryPolicy = new RetryPolicy() {
+ @Override
+ public RetryAction shouldRetry(Exception exception, int retries,
+ int failovers, boolean isIdempotentOrAtMostOnce)
+ throws Exception {
+
+ if (LOG.isDebugEnabled()) {
+ if (exception.getCause() != null) {
+ LOG.debug("RetryProxy: SCM Security Server {}: {}: {}",
+ getCurrentProxySCMNodeId(),
+ exception.getCause().getClass().getSimpleName(),
+ exception.getCause().getMessage());
+ } else {
+ LOG.debug("RetryProxy: OM {}: {}", getCurrentProxySCMNodeId(),
+ exception.getMessage());
+ }
+ }
+
+ // For AccessControl Exception where Client is not authentica
+ if (HAUtils.isAccessControlException(exception)) {
+ return RetryAction.FAIL;
+ }
+
+ // Perform fail over to next proxy, as right now we don't have any
+ // suggested leader ID from server, we fail over to next one.
+ // TODO: Act based on server response if leader id is passed.
+ performFailoverToNextProxy();
+ return getRetryAction(FAILOVER_AND_RETRY, failovers);
+ }
+
+ private RetryAction getRetryAction(RetryDecision fallbackAction,
+ int failovers) {
+ if (failovers < maxRetryCount) {
+ return new RetryAction(fallbackAction, getRetryInterval());
+ } else {
+ return RetryAction.FAIL;
+ }
+ }
+ };
+
+ return retryPolicy;
+ }
+
+
+ @Override
+ public Class< SCMSecurityProtocolPB > getInterface() {
+ return SCMSecurityProtocolPB.class;
+ }
+
+ @Override
+ public void close() throws IOException {
+ for (ProxyInfo<SCMSecurityProtocolPB> proxyInfo : scmProxies.values()) {
+ if (proxyInfo != null) {
+ RPC.stopProxy(proxyInfo.proxy);
Review comment:
Done
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
[GitHub] [ozone] bharatviswa504 commented on a change in pull request #1978: HDDS-4876. [SCM HA Security] Add failover proxy to SCM Security Server Protocol
Posted by GitBox <gi...@apache.org>.
bharatviswa504 commented on a change in pull request #1978:
URL: https://github.com/apache/ozone/pull/1978#discussion_r590108348
##########
File path: hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/HAUtils.java
##########
@@ -285,4 +288,24 @@ public static File getMetaDir(DBDefinition definition,
}
return metadataDir;
}
+
+ /**
+ * Unwrap exception to check if it is some kind of access control problem.
+ * {@link AccessControlException}
+ */
+ public static boolean isAccessControlException(Exception ex) {
Review comment:
Opened a new Jira for this
https://issues.apache.org/jira/browse/HDDS-4934
##########
File path: hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/proxy/SCMSecurityProtocolFailoverProxyProvider.java
##########
@@ -0,0 +1,279 @@
+/*
+ * 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.scm.proxy;
+
+import com.google.common.base.Preconditions;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdds.conf.ConfigurationException;
+import org.apache.hadoop.hdds.conf.ConfigurationSource;
+import org.apache.hadoop.hdds.protocolPB.SCMSecurityProtocolPB;
+import org.apache.hadoop.hdds.scm.ha.SCMNodeInfo;
+import org.apache.hadoop.hdds.utils.HAUtils;
+import org.apache.hadoop.hdds.utils.LegacyHadoopConfigurationSource;
+import org.apache.hadoop.io.retry.FailoverProxyProvider;
+import org.apache.hadoop.io.retry.RetryPolicies;
+import org.apache.hadoop.io.retry.RetryPolicy;
+import org.apache.hadoop.io.retry.RetryPolicy.RetryAction.RetryDecision;
+import org.apache.hadoop.ipc.ProtobufRpcEngine;
+import org.apache.hadoop.ipc.RPC;
+import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.hadoop.io.retry.RetryPolicy.RetryAction.RetryDecision.FAILOVER_AND_RETRY;
+
+/**
+ * Failover proxy provider for SCMSecurityProtocol server.
+ */
+public class SCMSecurityProtocolFailoverProxyProvider implements
+ FailoverProxyProvider<SCMSecurityProtocolPB>, Closeable {
+
+ public static final Logger LOG =
+ LoggerFactory.getLogger(SCMSecurityProtocolFailoverProxyProvider.class);
+
+ // scmNodeId -> ProxyInfo<rpcProxy>
+ private final Map<String,
+ ProxyInfo<SCMSecurityProtocolPB>> scmProxies;
+
+ // scmNodeId -> SCMProxyInfo
+ private final Map<String, SCMProxyInfo> scmProxyInfoMap;
+
+ private List<String> scmNodeIds;
+
+ private String currentProxySCMNodeId;
+ private int currentProxyIndex;
+
+ private final ConfigurationSource conf;
+ private final SCMClientConfig scmClientConfig;
+ private final long scmVersion;
+
+ private String scmServiceId;
+
+ private final int maxRetryCount;
+ private final long retryInterval;
+
+ private final UserGroupInformation ugi;
+
+ /**
+ * Construct fail-over proxy provider for SCMSecurityProtocol Server.
+ * @param conf
+ * @param userGroupInformation
+ */
+ public SCMSecurityProtocolFailoverProxyProvider(ConfigurationSource conf,
+ UserGroupInformation userGroupInformation) {
+ Preconditions.checkNotNull(userGroupInformation);
+ this.ugi = userGroupInformation;
+ this.conf = conf;
+ this.scmVersion = RPC.getProtocolVersion(SCMSecurityProtocolPB.class);
+
+ this.scmProxies = new HashMap<>();
+ this.scmProxyInfoMap = new HashMap<>();
+ loadConfigs();
+
+ this.currentProxyIndex = 0;
+ currentProxySCMNodeId = scmNodeIds.get(currentProxyIndex);
+ scmClientConfig = conf.getObject(SCMClientConfig.class);
+ this.maxRetryCount = scmClientConfig.getRetryCount();
+ this.retryInterval = scmClientConfig.getRetryInterval();
+ }
+
+ protected void loadConfigs() {
+ List<SCMNodeInfo> scmNodeInfoList = SCMNodeInfo.buildNodeInfo(conf);
+ scmNodeIds = new ArrayList<>();
+
+ for (SCMNodeInfo scmNodeInfo : scmNodeInfoList) {
+ if (scmNodeInfo.getScmSecurityAddress() == null) {
+ throw new ConfigurationException("SCM Client Address could not " +
+ "be obtained from config. Config is not properly defined");
+ } else {
+ InetSocketAddress scmSecurityAddress =
+ NetUtils.createSocketAddr(scmNodeInfo.getScmSecurityAddress());
+
+ scmServiceId = scmNodeInfo.getServiceId();
+ String scmNodeId = scmNodeInfo.getNodeId();
+
+ scmNodeIds.add(scmNodeId);
+ SCMProxyInfo scmProxyInfo = new SCMProxyInfo(scmServiceId, scmNodeId,
+ scmSecurityAddress);
+ scmProxyInfoMap.put(scmNodeId, scmProxyInfo);
+ }
+ }
+ }
+
+ @Override
+ public ProxyInfo<SCMSecurityProtocolPB> getProxy() {
Review comment:
Done
##########
File path: hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/security/exception/SCMSecurityException.java
##########
@@ -45,12 +55,24 @@ public SCMSecurityException(String message, Throwable cause) {
this.errorCode = ErrorCode.DEFAULT;
}
+ /**
+ * Ctor.
+ * @param message - Error Message
+ * @param cause - Actual cause.
+ * @param errorCode - Error code.
+ */
+ public SCMSecurityException(String message, Throwable cause,
+ ErrorCode errorCode) {
+ super(message, cause);
+ this.errorCode = errorCode;
+ }
+
/**
* Ctor.
* @param message - Message.
* @param error - error code.
*/
- public SCMSecurityException(String message, ErrorCode error) {
+ public SCMSecurityException(Exception message, ErrorCode error) {
Review comment:
Done
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org