>From Murtadha Hubail <[email protected]>:
Murtadha Hubail has uploaded this change for review. (
https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/19371 )
Change subject: WIP: IExternalCredentialsCache API
......................................................................
WIP: IExternalCredentialsCache API
Change-Id: Ib7aee286315067065706e8567783cd1fa2eac07a
---
M
asterixdb/asterix-app/src/main/java/org/apache/asterix/app/external/ExternalCredentialsCacheUpdater.java
M
asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/NCAppRuntimeContext.java
A
asterixdb/asterix-common/src/main/java/org/apache/asterix/common/external/IExternalEntityNameResolver.java
M
asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
M
asterixdb/asterix-common/src/main/java/org/apache/asterix/common/external/IExternalCredentialsCache.java
M
asterixdb/asterix-app/src/main/java/org/apache/asterix/app/external/ExternalCredentialsCache.java
M
asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/aws/s3/S3AuthUtils.java
M
asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/UpdateAwsCredentialsCacheRequest.java
M
asterixdb/asterix-common/src/main/java/org/apache/asterix/common/config/ExternalProperties.java
M
asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
A
asterixdb/asterix-common/src/main/java/org/apache/asterix/common/external/IExternalEntityNameResolverProvider.java
M
asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/MetadataProvider.java
12 files changed, 176 insertions(+), 142 deletions(-)
git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb
refs/changes/71/19371/1
diff --git
a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/external/ExternalCredentialsCache.java
b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/external/ExternalCredentialsCache.java
index 8cb3a47..9609568 100644
---
a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/external/ExternalCredentialsCache.java
+++
b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/external/ExternalCredentialsCache.java
@@ -24,14 +24,8 @@
import java.util.concurrent.TimeUnit;
import org.apache.asterix.common.api.IApplicationContext;
-import org.apache.asterix.common.exceptions.AsterixException;
import org.apache.asterix.common.exceptions.CompilationException;
-import org.apache.asterix.common.exceptions.ErrorCode;
import org.apache.asterix.common.external.IExternalCredentialsCache;
-import org.apache.asterix.common.metadata.DatasetFullyQualifiedName;
-import org.apache.asterix.common.metadata.DataverseName;
-import org.apache.asterix.common.metadata.IFullyQualifiedName;
-import org.apache.asterix.common.metadata.MetadataConstants;
import org.apache.asterix.external.util.ExternalDataConstants;
import org.apache.asterix.external.util.aws.s3.S3Constants;
import org.apache.commons.lang3.tuple.Pair;
@@ -46,41 +40,36 @@
private static final Logger LOGGER = LogManager.getLogger();
private final ConcurrentMap<String, Pair<Span, Object>> cache = new
ConcurrentHashMap<>();
private final int awsAssumeRoleDuration;
- private final double refreshAwsAssumeRoleThreshold;
+ private final int refreshAwsAssumeRoleThresholdPercentage;
public ExternalCredentialsCache(IApplicationContext appCtx) {
this.awsAssumeRoleDuration =
appCtx.getExternalProperties().getAwsAssumeRoleDuration();
- this.refreshAwsAssumeRoleThreshold =
appCtx.getExternalProperties().getAwsRefreshAssumeRoleThreshold();
+ this.refreshAwsAssumeRoleThresholdPercentage =
+
appCtx.getExternalProperties().getAwsRefreshAssumeRoleThresholdPercentage();
}
@Override
- public synchronized Object getCredentials(Map<String, String>
configuration) throws CompilationException {
- IFullyQualifiedName fqn =
getFullyQualifiedNameFromConfiguration(configuration);
- return getCredentials(fqn);
- }
-
- @Override
- public synchronized Object getCredentials(IFullyQualifiedName fqn) {
- String name = getName(fqn);
- if (cache.containsKey(name) &&
!needsRefresh(cache.get(name).getLeft())) {
- return cache.get(name).getRight();
+ public synchronized Object get(String key) throws CompilationException {
+ if (cache.containsKey(key) && !needsRefresh(cache.get(key).getLeft()))
{
+ return cache.get(key).getRight();
}
return null;
}
@Override
- public synchronized void updateCache(Map<String, String> configuration,
Map<String, String> credentials)
- throws CompilationException {
- IFullyQualifiedName fqn =
getFullyQualifiedNameFromConfiguration(configuration);
- String type =
configuration.get(ExternalDataConstants.KEY_EXTERNAL_SOURCE_TYPE);
- updateCache(fqn, type, credentials);
+ public synchronized void delete(String key) throws CompilationException {
+ Object removed = cache.remove(key);
+ if (removed != null) {
+ LOGGER.info("Removed cached credentials for {}", key);
+ } else {
+ LOGGER.info("No cached credentials found for {}, nothing to
remove", key);
+ }
}
@Override
- public synchronized void updateCache(IFullyQualifiedName fqn, String type,
Map<String, String> credentials) {
- String name = getName(fqn);
+ public synchronized void put(String key, String type, Map<String, String>
credentials) throws CompilationException {
if
(ExternalDataConstants.KEY_ADAPTER_NAME_AWS_S3.equalsIgnoreCase(type)) {
- updateAwsCache(name, credentials);
+ updateAwsCache(key, credentials);
}
}
@@ -88,36 +77,12 @@
String accessKeyId =
credentials.get(S3Constants.ACCESS_KEY_ID_FIELD_NAME);
String secretAccessKey =
credentials.get(S3Constants.SECRET_ACCESS_KEY_FIELD_NAME);
String sessionToken =
credentials.get(S3Constants.SESSION_TOKEN_FIELD_NAME);
- doUpdateAwsCache(name, AwsSessionCredentials.create(accessKeyId,
secretAccessKey, sessionToken));
- }
- private void doUpdateAwsCache(String name, AwsSessionCredentials
credentials) {
- cache.put(name, Pair.of(Span.start(awsAssumeRoleDuration,
TimeUnit.SECONDS), credentials));
+ AwsSessionCredentials sessionCreds =
AwsSessionCredentials.create(accessKeyId, secretAccessKey, sessionToken);
+ cache.put(name, Pair.of(Span.start(awsAssumeRoleDuration,
TimeUnit.SECONDS), sessionCreds));
LOGGER.info("Received and cached new credentials for {}", name);
}
- @Override
- public void deleteCredentials(IFullyQualifiedName fqn) {
- String name = getName(fqn);
- Object removed = cache.remove(name);
- if (removed != null) {
- LOGGER.info("Removed cached credentials for {}", name);
- } else {
- LOGGER.info("No cached credentials found for {}, nothing to
remove", name);
- }
- }
-
- @Override
- public String getName(Map<String, String> configuration) throws
CompilationException {
- IFullyQualifiedName fqn =
getFullyQualifiedNameFromConfiguration(configuration);
- return getName(fqn);
- }
-
- @Override
- public String getName(IFullyQualifiedName fqn) {
- return fqn.toString();
- }
-
/**
* Refresh if the remaining time is less than the configured refresh
percentage
*
@@ -125,27 +90,9 @@
* @return true if the remaining time is less than the configured refresh
percentage, false otherwise
*/
private boolean needsRefresh(Span span) {
- return (double) span.remaining(TimeUnit.SECONDS)
- / span.getSpan(TimeUnit.SECONDS) <
refreshAwsAssumeRoleThreshold;
- }
-
- protected IFullyQualifiedName
getFullyQualifiedNameFromConfiguration(Map<String, String> configuration)
- throws CompilationException {
- String database =
configuration.get(ExternalDataConstants.KEY_DATASET_DATABASE);
- if (database == null) {
- database = MetadataConstants.DEFAULT_DATABASE;
- }
- String stringDataverse =
configuration.get(ExternalDataConstants.KEY_DATASET_DATAVERSE);
- DataverseName dataverse = getDataverseName(stringDataverse);
- String dataset = configuration.get(ExternalDataConstants.KEY_DATASET);
- return new DatasetFullyQualifiedName(database, dataverse, dataset);
- }
-
- protected DataverseName getDataverseName(String dataverse) throws
CompilationException {
- try {
- return DataverseName.createSinglePartName(dataverse);
- } catch (AsterixException ex) {
- throw new
CompilationException(ErrorCode.INVALID_DATABASE_OBJECT_NAME, dataverse);
- }
+ double remaining = (double) span.remaining(TimeUnit.SECONDS) /
span.getSpan(TimeUnit.SECONDS);
+ double passed = 1 - remaining;
+ int passedPercentage = (int) (passed * 100);
+ return passedPercentage > refreshAwsAssumeRoleThresholdPercentage;
}
}
diff --git
a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/external/ExternalCredentialsCacheUpdater.java
b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/external/ExternalCredentialsCacheUpdater.java
index f07caaa..29e3239 100644
---
a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/external/ExternalCredentialsCacheUpdater.java
+++
b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/external/ExternalCredentialsCacheUpdater.java
@@ -40,6 +40,7 @@
import org.apache.asterix.common.external.IExternalCredentialsCache;
import org.apache.asterix.common.external.IExternalCredentialsCacheUpdater;
import org.apache.asterix.common.messaging.api.MessageFuture;
+import org.apache.asterix.external.util.ExternalDataConstants;
import org.apache.asterix.external.util.aws.s3.S3AuthUtils;
import org.apache.asterix.external.util.aws.s3.S3Constants;
import org.apache.asterix.messaging.CCMessageBroker;
@@ -65,7 +66,8 @@
public synchronized Object generateAndCacheCredentials(Map<String, String>
configuration)
throws HyracksDataException, CompilationException {
IExternalCredentialsCache cache = appCtx.getExternalCredentialsCache();
- Object credentials = cache.getCredentials(configuration);
+ String name = configuration.get(ExternalDataConstants.KEY_ENTITY_ID);
+ Object credentials = cache.get(name);
if (credentials != null) {
return credentials;
}
@@ -74,9 +76,7 @@
* if we are the CC, generate new creds and ask all NCs to update
their cache
* if we are the NC, send a message to the CC to generate new creds
and ask all NCs to update their cache
*/
- String name = cache.getName(configuration);
- if (appCtx instanceof ICcApplicationContext) {
- ICcApplicationContext ccAppCtx = (ICcApplicationContext) appCtx;
+ if (appCtx instanceof ICcApplicationContext ccAppCtx) {
IClusterManagementWork.ClusterState state =
ccAppCtx.getClusterStateManager().getState();
if (!(state == ACTIVE || state == REBALANCE_REQUIRED)) {
throw new RuntimeDataException(REJECT_BAD_CLUSTER_STATE,
state);
@@ -106,7 +106,8 @@
// request all NCs to update their credentials cache with the
latest creds
updateNcsCredentialsCache(ccAppCtx, name, credentialsMap,
configuration);
- cache.updateCache(configuration, credentialsMap);
+ String type =
configuration.get(ExternalDataConstants.KEY_EXTERNAL_SOURCE_TYPE);
+ cache.put(name, type, credentialsMap);
credentials = AwsSessionCredentials.create(accessKeyId,
secretAccessKey, sessionToken);
} else {
NCMessageBroker broker = (NCMessageBroker)
appCtx.getServiceContext().getMessageBroker();
diff --git
a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/UpdateAwsCredentialsCacheRequest.java
b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/UpdateAwsCredentialsCacheRequest.java
index d1a9ebf..8284369 100644
---
a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/UpdateAwsCredentialsCacheRequest.java
+++
b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/UpdateAwsCredentialsCacheRequest.java
@@ -23,6 +23,7 @@
import org.apache.asterix.common.api.INcApplicationContext;
import org.apache.asterix.common.exceptions.CompilationException;
import org.apache.asterix.common.messaging.api.INcAddressedMessage;
+import org.apache.asterix.external.util.ExternalDataConstants;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@@ -42,7 +43,9 @@
@Override
public void handle(INcApplicationContext appCtx) throws
HyracksDataException {
try {
- appCtx.getExternalCredentialsCache().updateCache(configuration,
credentials);
+ String name =
configuration.get(ExternalDataConstants.KEY_ENTITY_ID);
+ String type =
configuration.get(ExternalDataConstants.KEY_EXTERNAL_SOURCE_TYPE);
+ appCtx.getExternalCredentialsCache().put(name, type, credentials);
} catch (CompilationException ex) {
LOGGER.info("Failed to process request", ex);
throw HyracksDataException.create(ex);
diff --git
a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/NCAppRuntimeContext.java
b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/NCAppRuntimeContext.java
index 343baf0..829e854 100644
---
a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/NCAppRuntimeContext.java
+++
b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/NCAppRuntimeContext.java
@@ -67,6 +67,7 @@
import org.apache.asterix.common.context.IStorageComponentProvider;
import org.apache.asterix.common.external.IExternalCredentialsCache;
import org.apache.asterix.common.external.IExternalCredentialsCacheUpdater;
+import org.apache.asterix.common.external.IExternalEntityNameResolverProvider;
import org.apache.asterix.common.library.ILibraryManager;
import org.apache.asterix.common.replication.IReplicationChannel;
import org.apache.asterix.common.replication.IReplicationManager;
@@ -190,6 +191,7 @@
private final INamespacePathResolver namespacePathResolver;
private final INamespaceResolver namespaceResolver;
private IDiskCacheMonitoringService diskCacheService;
+ protected IExternalEntityNameResolverProvider
externalEntityNameResolverProvider;
protected IExternalCredentialsCache externalCredentialsCache;
protected IExternalCredentialsCacheUpdater externalCredentialsCacheUpdater;
diff --git
a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
index 51371da..58fc320 100644
---
a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
+++
b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
@@ -2448,9 +2448,7 @@
sourceLoc, EnumSet.of(DropOption.IF_EXISTS),
requestParameters.isForceDropDataset());
MetadataManager.INSTANCE.commitTransaction(mdTxnCtx.getValue());
- if (ds.getDatasetType().equals(DatasetType.EXTERNAL)) {
-
appCtx.getExternalCredentialsCache().deleteCredentials(ds.getDatasetFullyQualifiedName());
- }
+ deleteDatasetCachedCredentials(ds);
return true;
} catch (Exception e) {
LOGGER.error("failed to drop dataset; executing compensating
operations", e);
@@ -2498,6 +2496,12 @@
}
}
+ protected void deleteDatasetCachedCredentials(Dataset dataset) throws
CompilationException {
+ if (dataset.getDatasetType().equals(DatasetType.EXTERNAL)) {
+
appCtx.getExternalCredentialsCache().delete(dataset.getDatasetFullyQualifiedName().toString());
+ }
+ }
+
protected void handleIndexDropStatement(MetadataProvider metadataProvider,
Statement stmt,
IHyracksClientConnection hcc, IRequestParameters
requestParameters) throws Exception {
IndexDropStatement stmtIndexDrop = (IndexDropStatement) stmt;
diff --git
a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/config/ExternalProperties.java
b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/config/ExternalProperties.java
index f6559f0..62437b2 100644
---
a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/config/ExternalProperties.java
+++
b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/config/ExternalProperties.java
@@ -18,12 +18,12 @@
*/
package org.apache.asterix.common.config;
-import static org.apache.hyracks.control.common.config.OptionTypes.DOUBLE;
import static org.apache.hyracks.control.common.config.OptionTypes.LEVEL;
import static
org.apache.hyracks.control.common.config.OptionTypes.NONNEGATIVE_INTEGER;
import static
org.apache.hyracks.control.common.config.OptionTypes.POSITIVE_INTEGER;
import static
org.apache.hyracks.control.common.config.OptionTypes.POSITIVE_INTEGER_BYTE_UNIT;
import static org.apache.hyracks.control.common.config.OptionTypes.STRING;
+import static
org.apache.hyracks.control.common.config.OptionTypes.getRangedIntegerType;
import org.apache.hyracks.api.config.IOption;
import org.apache.hyracks.api.config.IOptionType;
@@ -55,14 +55,16 @@
LIBRARY_DEPLOY_TIMEOUT(POSITIVE_INTEGER, 1800, "Timeout to upload a
UDF in seconds"),
AZURE_REQUEST_TIMEOUT(POSITIVE_INTEGER, 120, "Timeout for Azure client
requests in seconds"),
AWS_ASSUME_ROLE_DURATION(
- POSITIVE_INTEGER,
+ getRangedIntegerType(900, 43200),
900,
"AWS assuming role duration in seconds. "
+ "Range from 900 seconds (15 mins) to 43200 seconds
(12 hours)"),
- AWS_REFRESH_ASSUME_ROLE_THRESHOLD(
- DOUBLE,
- .5,
- "Percentage of left duration before assume role credentials "
+ "needs to be refreshed");
+ AWS_REFRESH_ASSUME_ROLE_THRESHOLD_PERCENTAGE(
+ getRangedIntegerType(25, 90),
+ 75,
+ "Percentage of duration passed before assume role credentials
need to be refreshed, the value ranges "
+ + "from 25 to 90, default is 75. For example, if the
value is set to 65, this means the "
+ + "credentials need to be refreshed if 65% of the
total expiration duration is already passed");
private final IOptionType type;
private final Object defaultValue;
@@ -91,7 +93,7 @@
case LIBRARY_DEPLOY_TIMEOUT:
case AZURE_REQUEST_TIMEOUT:
case AWS_ASSUME_ROLE_DURATION:
- case AWS_REFRESH_ASSUME_ROLE_THRESHOLD:
+ case AWS_REFRESH_ASSUME_ROLE_THRESHOLD_PERCENTAGE:
return Section.COMMON;
case CC_JAVA_OPTS:
case NC_JAVA_OPTS:
@@ -177,7 +179,7 @@
return accessor.getInt(Option.AWS_ASSUME_ROLE_DURATION);
}
- public double getAwsRefreshAssumeRoleThreshold() {
- return accessor.getDouble(Option.AWS_REFRESH_ASSUME_ROLE_THRESHOLD);
+ public int getAwsRefreshAssumeRoleThresholdPercentage() {
+ return
accessor.getInt(Option.AWS_REFRESH_ASSUME_ROLE_THRESHOLD_PERCENTAGE);
}
}
diff --git
a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/external/IExternalCredentialsCache.java
b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/external/IExternalCredentialsCache.java
index 3ff444d..7399fce 100644
---
a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/external/IExternalCredentialsCache.java
+++
b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/external/IExternalCredentialsCache.java
@@ -21,63 +21,23 @@
import java.util.Map;
import org.apache.asterix.common.exceptions.CompilationException;
-import org.apache.asterix.common.metadata.IFullyQualifiedName;
public interface IExternalCredentialsCache {
/**
- * Returns the cached credentials. Can be of any supported external
credentials type
+ * Returns the cached credentials.
*
- * @param configuration configuration containing external collection
details
- * @return credentials if present, null otherwise
+ * @return credentials if present and not expired/need refreshing, null
otherwise
*/
- Object getCredentials(Map<String, String> configuration) throws
CompilationException;
+ Object get(String key) throws CompilationException;
/**
- * Returns the cached credentials. Can be of any supported external
credentials type
- *
- * @param fqn fully qualified name of the credentials entity
- * @return credentials if present, null otherwise
+ * Deletes the cache for the provided entity
*/
- Object getCredentials(IFullyQualifiedName fqn) throws CompilationException;
+ void delete(String key) throws CompilationException;
/**
* Updates the credentials cache with the provided credentials for the
specified name
- *
- * @param configuration configuration containing external collection
details
- * @param credentials credentials to cache
*/
- void updateCache(Map<String, String> configuration, Map<String, String>
credentials) throws CompilationException;
-
- /**
- * Updates the credentials cache with the provided credentials for the
specified name
- *
- * @param fqn fully qualified name for the credentials entity
- * @param type type of the entity
- * @param credentials credentials to cache
- */
- void updateCache(IFullyQualifiedName fqn, String type, Map<String, String>
credentials);
-
- /**
- * Deletes the cache for the provided enitty
- *
- * @param fqn fully qualified name of entity for which the credentials are
to be deleted
- */
- void deleteCredentials(IFullyQualifiedName fqn);
-
- /**
- * Returns the name of the entity which the cached credentials belong to
- *
- * @param configuration configuration containing external collection
details
- * @return name of entity which credentials belong to
- */
- String getName(Map<String, String> configuration) throws
CompilationException;
-
- /**
- * Returns the name of the entity which the cached credentials belong to
- *
- * @param fqn fully qualified name for the credentials entity
- * @return name of entity which credentials belong to
- */
- String getName(IFullyQualifiedName fqn) throws CompilationException;
+ void put(String key, String type, Map<String, String> credentials) throws
CompilationException;
}
diff --git
a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/external/IExternalEntityNameResolver.java
b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/external/IExternalEntityNameResolver.java
new file mode 100644
index 0000000..1d9caf7
--- /dev/null
+++
b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/external/IExternalEntityNameResolver.java
@@ -0,0 +1,27 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.asterix.common.external;
+
+import org.apache.asterix.common.exceptions.CompilationException;
+
+@FunctionalInterface
+public interface IExternalEntityNameResolver {
+
+ String resolveName() throws CompilationException;
+}
diff --git
a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/external/IExternalEntityNameResolverProvider.java
b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/external/IExternalEntityNameResolverProvider.java
new file mode 100644
index 0000000..a6a5a36
--- /dev/null
+++
b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/external/IExternalEntityNameResolverProvider.java
@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.asterix.common.external;
+
+import java.util.Map;
+
+public interface IExternalEntityNameResolverProvider {
+
+ /**
+ * Creates an external entity name resolver based on the provided object
entity
+ * @param entity object entity
+ * @return external entity name resolver
+ */
+ IExternalEntityNameResolver create(Object entity);
+
+ /**
+ * Creates an external entity name resolver based on the provided
configuration
+ * @param configuration configuration
+ * @return external entity name resolver
+ */
+ IExternalEntityNameResolver create(Map<String, String> configuration);
+}
diff --git
a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
index 4cc1656..00c4475 100644
---
a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
+++
b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
@@ -106,6 +106,7 @@
public static final String KEY_FEED_NAME = "feed";
// a string representing external bucket name
public static final String KEY_EXTERNAL_SOURCE_TYPE = "type";
+ public static final String KEY_ENTITY_ID = "entity-id";
// a comma delimited list of nodes
public static final String KEY_NODES = "nodes";
// a string representing the password used to authenticate with the
external data source
diff --git
a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/aws/s3/S3AuthUtils.java
b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/aws/s3/S3AuthUtils.java
index 57889ea..334c78d 100644
---
a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/aws/s3/S3AuthUtils.java
+++
b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/aws/s3/S3AuthUtils.java
@@ -241,7 +241,7 @@
public static AwsCredentialsProvider
getTrustAccountCredentials(IApplicationContext appCtx,
Map<String, String> configuration) throws CompilationException {
IExternalCredentialsCache cache = appCtx.getExternalCredentialsCache();
- Object credentialsObject = cache.getCredentials(configuration);
+ Object credentialsObject =
cache.get(configuration.get(ExternalDataConstants.KEY_ENTITY_ID));
if (credentialsObject != null) {
return () -> (AwsSessionCredentials) credentialsObject;
}
@@ -389,7 +389,7 @@
*/
public static void configureAwsS3HdfsJobConf(IApplicationContext appCtx,
JobConf jobConf,
Map<String, String> configuration, int numberOfPartitions) throws
CompilationException {
- setHadoopCredentials(jobConf, configuration);
+ setHadoopCredentials(appCtx, jobConf, configuration);
String serviceEndpoint =
configuration.get(SERVICE_END_POINT_FIELD_NAME);
Region region =
validateAndGetRegion(configuration.get(REGION_FIELD_NAME));
jobConf.set(HADOOP_REGION, region.toString());
@@ -421,7 +421,8 @@
* @param jobConf hadoop job config
* @param configuration external details configuration
*/
- private static void setHadoopCredentials(JobConf jobConf, Map<String,
String> configuration) {
+ private static void setHadoopCredentials(IApplicationContext appCtx,
JobConf jobConf,
+ Map<String, String> configuration) {
AuthenticationType authenticationType =
getAuthenticationType(configuration);
switch (authenticationType) {
case ANONYMOUS:
@@ -432,7 +433,11 @@
jobConf.set(HADOOP_ASSUME_ROLE_ARN,
configuration.get(ROLE_ARN_FIELD_NAME));
jobConf.set(HADOOP_ASSUME_ROLE_EXTERNAL_ID,
configuration.get(EXTERNAL_ID_FIELD_NAME));
jobConf.set(HADOOP_ASSUME_ROLE_SESSION_NAME, "parquet-" +
UUID.randomUUID());
- jobConf.set(HADOOP_ASSUME_ROLE_SESSION_DURATION, "15m");
+
+ // hadoop accepts time 15m to 1h, we will base it on the
provided configuration
+ int durationInSeconds =
appCtx.getExternalProperties().getAwsAssumeRoleDuration();
+ String hadoopDuration = getHadoopDuration(durationInSeconds);
+ jobConf.set(HADOOP_ASSUME_ROLE_SESSION_DURATION,
hadoopDuration);
// TODO: this assumes basic keys always, also support if we
use InstanceProfile to assume a role
jobConf.set(HADOOP_CREDENTIALS_TO_ASSUME_ROLE_KEY,
HADOOP_SIMPLE);
@@ -456,6 +461,36 @@
}
/**
+ * Hadoop accepts duration values from 15m to 1h (in this format). We will
base this on the configured
+ * duration in seconds. If the time exceeds 1 hour, we will return 1h
+ *
+ * @param seconds configured duration in seconds
+ * @return hadoop updated duration
+ */
+ private static String getHadoopDuration(int seconds) {
+ // constants for time thresholds
+ final int FIFTEEN_MINUTES_IN_SECONDS = 15 * 60;
+ final int ONE_HOUR_IN_SECONDS = 60 * 60;
+
+ // Adjust seconds to fit within bounds
+ if (seconds < FIFTEEN_MINUTES_IN_SECONDS) {
+ seconds = FIFTEEN_MINUTES_IN_SECONDS;
+ } else if (seconds > ONE_HOUR_IN_SECONDS) {
+ seconds = ONE_HOUR_IN_SECONDS;
+ }
+
+ // Convert seconds to minutes
+ int minutes = seconds / 60;
+
+ // Format the result
+ if (minutes == 60) {
+ return "1h";
+ } else {
+ return minutes + "m";
+ }
+ }
+
+ /**
* Validate external dataset properties
*
* @param configuration properties
diff --git
a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/MetadataProvider.java
b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/MetadataProvider.java
index ef056c9..95c4ee5 100644
---
a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/MetadataProvider.java
+++
b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/MetadataProvider.java
@@ -998,6 +998,7 @@
configuration.put(ExternalDataConstants.KEY_DATASET_DATABASE,
dataset.getDatabaseName());
configuration.put(ExternalDataConstants.KEY_DATASET_DATAVERSE,
dataset.getDataverseName().getCanonicalForm());
+ setExternalEntityID(configuration, dataset);
return
AdapterFactoryProvider.getAdapterFactory(getApplicationContext().getServiceContext(),
adapterName,
configuration, itemType, null, warningCollector,
filterEvaluatorFactory);
} catch (Exception e) {
@@ -1005,6 +1006,10 @@
}
}
+ protected void setExternalEntityID(Map<String, String> configuration,
Dataset dataset) throws AlgebricksException {
+ configuration.put(ExternalDataConstants.KEY_ENTITY_ID,
dataset.getDatasetFullyQualifiedName().toString());
+ }
+
public TxnId getTxnId() {
return txnId;
}
--
To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/19371
To unsubscribe, or for help writing mail filters, visit
https://asterix-gerrit.ics.uci.edu/settings
Gerrit-Project: asterixdb
Gerrit-Branch: master
Gerrit-Change-Id: Ib7aee286315067065706e8567783cd1fa2eac07a
Gerrit-Change-Number: 19371
Gerrit-PatchSet: 1
Gerrit-Owner: Murtadha Hubail <[email protected]>
Gerrit-MessageType: newchange
