S3 Repository: Eagerly load static settings (#23910)
The S3 repostiory has many levels of settings it looks at to create a repository, and these settings were read at repository creation time. This meant secure settings like access and secret keys had to be available after node construction. This change makes setting loading for every except repository level settings eager, so that secure settings can be stashed, and the keystore can once again be closed after bootstrapping the node is complete.
This commit is contained in:
Ryan Ernst
GitHub
parent
b4c3bb5d21
commit
1207103b6d
|
|
@ -338,13 +338,12 @@ final class Bootstrap {
|
|||
|
||||
INSTANCE.setup(true, environment);
|
||||
|
||||
/* TODO: close this once s3 repository doesn't try to read during repository construction
|
||||
try {
|
||||
// any secure settings must be read during node construction
|
||||
IOUtils.close(keystore);
|
||||
} catch (IOException e) {
|
||||
throw new BootstrapException(e);
|
||||
}*/
|
||||
}
|
||||
|
||||
INSTANCE.start();
|
||||
|
||||
|
|
|
|||
|
|
@ -54,6 +54,11 @@ bundlePlugin {
|
|||
}
|
||||
}
|
||||
|
||||
integTestCluster {
|
||||
keystoreSetting 's3.client.default.access_key', 'myaccesskey'
|
||||
keystoreSetting 's3.client.default.secret_key', 'mysecretkey'
|
||||
}
|
||||
|
||||
thirdPartyAudit.excludes = [
|
||||
// classes are missing
|
||||
'javax.servlet.ServletContextEvent',
|
||||
|
|
|
|||
|
|
@ -38,13 +38,11 @@ import org.apache.logging.log4j.Logger;
|
|||
import org.elasticsearch.ElasticsearchException;
|
||||
import org.elasticsearch.cluster.metadata.RepositoryMetaData;
|
||||
import org.elasticsearch.common.Strings;
|
||||
import org.elasticsearch.common.collect.Tuple;
|
||||
import org.elasticsearch.common.component.AbstractLifecycleComponent;
|
||||
import org.elasticsearch.common.logging.DeprecationLogger;
|
||||
import org.elasticsearch.common.settings.SecureString;
|
||||
import org.elasticsearch.common.settings.Setting;
|
||||
import org.elasticsearch.common.settings.Settings;
|
||||
import org.elasticsearch.common.unit.TimeValue;
|
||||
|
||||
import static org.elasticsearch.repositories.s3.S3Repository.getValue;
|
||||
|
||||
|
|
@ -53,28 +51,31 @@ class InternalAwsS3Service extends AbstractLifecycleComponent implements AwsS3Se
|
|||
// pkg private for tests
|
||||
static final Setting<String> CLIENT_NAME = new Setting<>("client", "default", Function.identity());
|
||||
|
||||
/**
|
||||
* (acceskey, endpoint) -> client
|
||||
*/
|
||||
private Map<Tuple<String, String>, AmazonS3Client> clients = new HashMap<>();
|
||||
private final Map<String, S3ClientSettings> clientsSettings;
|
||||
|
||||
InternalAwsS3Service(Settings settings) {
|
||||
private final Map<String, AmazonS3Client> clientsCache = new HashMap<>();
|
||||
|
||||
InternalAwsS3Service(Settings settings, Map<String, S3ClientSettings> clientsSettings) {
|
||||
super(settings);
|
||||
this.clientsSettings = clientsSettings;
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized AmazonS3 client(Settings repositorySettings) {
|
||||
String clientName = CLIENT_NAME.get(repositorySettings);
|
||||
String foundEndpoint = findEndpoint(logger, repositorySettings, settings, clientName);
|
||||
|
||||
AWSCredentialsProvider credentials = buildCredentials(logger, deprecationLogger, settings, repositorySettings, clientName);
|
||||
|
||||
Tuple<String, String> clientDescriptor = new Tuple<>(foundEndpoint, credentials.getCredentials().getAWSAccessKeyId());
|
||||
AmazonS3Client client = clients.get(clientDescriptor);
|
||||
AmazonS3Client client = clientsCache.get(clientName);
|
||||
if (client != null) {
|
||||
return client;
|
||||
}
|
||||
|
||||
S3ClientSettings clientSettings = clientsSettings.get(clientName);
|
||||
if (clientSettings == null) {
|
||||
throw new IllegalArgumentException("Unknown s3 client name [" + clientName + "]. " +
|
||||
"Existing client configs: " +
|
||||
Strings.collectionToDelimitedString(clientsSettings.keySet(), ","));
|
||||
}
|
||||
|
||||
String endpoint = findEndpoint(logger, clientSettings, repositorySettings);
|
||||
Integer maxRetries = getValue(repositorySettings, settings,
|
||||
S3Repository.Repository.MAX_RETRIES_SETTING,
|
||||
S3Repository.Repositories.MAX_RETRIES_SETTING);
|
||||
|
|
@ -85,60 +86,49 @@ class InternalAwsS3Service extends AbstractLifecycleComponent implements AwsS3Se
|
|||
// otherwise we use the default value set by the SDK
|
||||
Boolean pathStyleAccess = null;
|
||||
if (S3Repository.Repository.PATH_STYLE_ACCESS_SETTING.exists(repositorySettings) ||
|
||||
S3Repository.Repositories.PATH_STYLE_ACCESS_SETTING.exists(settings)) {
|
||||
S3Repository.Repositories.PATH_STYLE_ACCESS_SETTING.exists(settings)) {
|
||||
pathStyleAccess = getValue(repositorySettings, settings,
|
||||
S3Repository.Repository.PATH_STYLE_ACCESS_SETTING,
|
||||
S3Repository.Repositories.PATH_STYLE_ACCESS_SETTING);
|
||||
}
|
||||
|
||||
logger.debug("creating S3 client with client_name [{}], endpoint [{}], max_retries [{}], " +
|
||||
"use_throttle_retries [{}], path_style_access [{}]",
|
||||
clientName, foundEndpoint, maxRetries, useThrottleRetries, pathStyleAccess);
|
||||
"use_throttle_retries [{}], path_style_access [{}]",
|
||||
clientName, endpoint, maxRetries, useThrottleRetries, pathStyleAccess);
|
||||
|
||||
client = new AmazonS3Client(
|
||||
credentials,
|
||||
buildConfiguration(logger, repositorySettings, settings, clientName, maxRetries, foundEndpoint, useThrottleRetries));
|
||||
AWSCredentialsProvider credentials = buildCredentials(logger, deprecationLogger, clientSettings, repositorySettings);
|
||||
ClientConfiguration configuration = buildConfiguration(logger, clientSettings, repositorySettings, maxRetries, endpoint, useThrottleRetries);
|
||||
|
||||
client = new AmazonS3Client(credentials, configuration);
|
||||
|
||||
if (pathStyleAccess != null) {
|
||||
client.setS3ClientOptions(new S3ClientOptions().withPathStyleAccess(pathStyleAccess));
|
||||
}
|
||||
|
||||
if (!foundEndpoint.isEmpty()) {
|
||||
client.setEndpoint(foundEndpoint);
|
||||
if (Strings.hasText(endpoint)) {
|
||||
client.setEndpoint(endpoint);
|
||||
}
|
||||
|
||||
clients.put(clientDescriptor, client);
|
||||
clientsCache.put(clientName, client);
|
||||
return client;
|
||||
}
|
||||
|
||||
// pkg private for tests
|
||||
static ClientConfiguration buildConfiguration(Logger logger, Settings repositorySettings, Settings settings,
|
||||
String clientName, Integer maxRetries, String endpoint,
|
||||
boolean useThrottleRetries) {
|
||||
static ClientConfiguration buildConfiguration(Logger logger, S3ClientSettings clientSettings, Settings repositorySettings,
|
||||
Integer maxRetries, String endpoint, boolean useThrottleRetries) {
|
||||
ClientConfiguration clientConfiguration = new ClientConfiguration();
|
||||
// the response metadata cache is only there for diagnostics purposes,
|
||||
// but can force objects from every response to the old generation.
|
||||
clientConfiguration.setResponseMetadataCacheSize(0);
|
||||
Protocol protocol = getConfigValue(repositorySettings, settings, clientName, S3Repository.PROTOCOL_SETTING,
|
||||
S3Repository.Repository.PROTOCOL_SETTING, S3Repository.Repositories.PROTOCOL_SETTING);
|
||||
Protocol protocol = getRepoValue(repositorySettings, S3Repository.Repository.PROTOCOL_SETTING, clientSettings.protocol);
|
||||
clientConfiguration.setProtocol(protocol);
|
||||
|
||||
String proxyHost = getConfigValue(null, settings, clientName,
|
||||
S3Repository.PROXY_HOST_SETTING, null, CLOUD_S3.PROXY_HOST_SETTING);
|
||||
if (Strings.hasText(proxyHost)) {
|
||||
Integer proxyPort = getConfigValue(null, settings, clientName,
|
||||
S3Repository.PROXY_PORT_SETTING, null, CLOUD_S3.PROXY_PORT_SETTING);
|
||||
try (SecureString proxyUsername = getConfigValue(null, settings, clientName,
|
||||
S3Repository.PROXY_USERNAME_SETTING, null, CLOUD_S3.PROXY_USERNAME_SETTING);
|
||||
SecureString proxyPassword = getConfigValue(null, settings, clientName,
|
||||
S3Repository.PROXY_PASSWORD_SETTING, null, CLOUD_S3.PROXY_PASSWORD_SETTING)) {
|
||||
|
||||
clientConfiguration
|
||||
.withProxyHost(proxyHost)
|
||||
.withProxyPort(proxyPort)
|
||||
.withProxyUsername(proxyUsername.toString())
|
||||
.withProxyPassword(proxyPassword.toString());
|
||||
}
|
||||
if (Strings.hasText(clientSettings.proxyHost)) {
|
||||
// TODO: remove this leniency, these settings should exist together and be validated
|
||||
clientConfiguration.setProxyHost(clientSettings.proxyHost);
|
||||
clientConfiguration.setProxyPort(clientSettings.proxyPort);
|
||||
clientConfiguration.setProxyUsername(clientSettings.proxyUsername);
|
||||
clientConfiguration.setProxyPassword(clientSettings.proxyPassword);
|
||||
}
|
||||
|
||||
if (maxRetries != null) {
|
||||
|
|
@ -146,64 +136,56 @@ class InternalAwsS3Service extends AbstractLifecycleComponent implements AwsS3Se
|
|||
clientConfiguration.setMaxErrorRetry(maxRetries);
|
||||
}
|
||||
clientConfiguration.setUseThrottleRetries(useThrottleRetries);
|
||||
|
||||
TimeValue readTimeout = getConfigValue(null, settings, clientName,
|
||||
S3Repository.READ_TIMEOUT_SETTING, null, CLOUD_S3.READ_TIMEOUT);
|
||||
clientConfiguration.setSocketTimeout((int)readTimeout.millis());
|
||||
clientConfiguration.setSocketTimeout(clientSettings.readTimeoutMillis);
|
||||
|
||||
return clientConfiguration;
|
||||
}
|
||||
|
||||
public static AWSCredentialsProvider buildCredentials(Logger logger, DeprecationLogger deprecationLogger,
|
||||
Settings settings, Settings repositorySettings, String clientName) {
|
||||
try (SecureString key = getConfigValue(repositorySettings, settings, clientName, S3Repository.ACCESS_KEY_SETTING,
|
||||
S3Repository.Repository.KEY_SETTING, S3Repository.Repositories.KEY_SETTING);
|
||||
SecureString secret = getConfigValue(repositorySettings, settings, clientName, S3Repository.SECRET_KEY_SETTING,
|
||||
S3Repository.Repository.SECRET_SETTING, S3Repository.Repositories.SECRET_SETTING)) {
|
||||
|
||||
if (key.length() == 0 && secret.length() == 0) {
|
||||
logger.debug("Using instance profile credentials");
|
||||
return new PrivilegedInstanceProfileCredentialsProvider();
|
||||
} else {
|
||||
logger.debug("Using basic key/secret credentials");
|
||||
return new StaticCredentialsProvider(new BasicAWSCredentials(key.toString(), secret.toString()));
|
||||
// pkg private for tests
|
||||
static AWSCredentialsProvider buildCredentials(Logger logger, DeprecationLogger deprecationLogger,
|
||||
S3ClientSettings clientSettings, Settings repositorySettings) {
|
||||
BasicAWSCredentials credentials = clientSettings.credentials;
|
||||
if (S3Repository.Repository.KEY_SETTING.exists(repositorySettings)) {
|
||||
if (S3Repository.Repository.SECRET_SETTING.exists(repositorySettings) == false) {
|
||||
throw new IllegalArgumentException("Repository setting [" + S3Repository.Repository.KEY_SETTING +
|
||||
" must be accompanied by setting [" + S3Repository.Repository.SECRET_SETTING + "]");
|
||||
}
|
||||
// backcompat for reading keys out of repository settings
|
||||
deprecationLogger.deprecated("Using s3 access/secret key from repository settings. Instead " +
|
||||
"store these in named clients and the elasticsearch keystore for secure settings.");
|
||||
try (SecureString key = S3Repository.Repository.KEY_SETTING.get(repositorySettings);
|
||||
SecureString secret = S3Repository.Repository.SECRET_SETTING.get(repositorySettings)) {
|
||||
credentials = new BasicAWSCredentials(key.toString(), secret.toString());
|
||||
}
|
||||
} else if (S3Repository.Repository.SECRET_SETTING.exists(repositorySettings)) {
|
||||
throw new IllegalArgumentException("Repository setting [" + S3Repository.Repository.SECRET_SETTING +
|
||||
" must be accompanied by setting [" + S3Repository.Repository.KEY_SETTING + "]");
|
||||
}
|
||||
if (credentials == null) {
|
||||
logger.debug("Using instance profile credentials");
|
||||
return new PrivilegedInstanceProfileCredentialsProvider();
|
||||
} else {
|
||||
logger.debug("Using basic key/secret credentials");
|
||||
return new StaticCredentialsProvider(credentials);
|
||||
}
|
||||
}
|
||||
|
||||
// pkg private for tests
|
||||
/** Returns the endpoint the client should use, based on the available endpoint settings found. */
|
||||
static String findEndpoint(Logger logger, Settings repositorySettings, Settings settings, String clientName) {
|
||||
String endpoint = getConfigValue(repositorySettings, settings, clientName, S3Repository.ENDPOINT_SETTING,
|
||||
S3Repository.Repository.ENDPOINT_SETTING, S3Repository.Repositories.ENDPOINT_SETTING);
|
||||
if (Strings.isNullOrEmpty(endpoint)) {
|
||||
// No region has been set so we will use the default endpoint
|
||||
if (CLOUD_S3.ENDPOINT_SETTING.exists(settings)) {
|
||||
endpoint = CLOUD_S3.ENDPOINT_SETTING.get(settings);
|
||||
logger.debug("using explicit s3 endpoint [{}]", endpoint);
|
||||
}
|
||||
} else {
|
||||
static String findEndpoint(Logger logger, S3ClientSettings clientSettings, Settings repositorySettings) {
|
||||
String endpoint = getRepoValue(repositorySettings, S3Repository.Repository.ENDPOINT_SETTING, clientSettings.endpoint);
|
||||
if (Strings.hasText(endpoint)) {
|
||||
logger.debug("using repository level endpoint [{}]", endpoint);
|
||||
}
|
||||
|
||||
return endpoint;
|
||||
}
|
||||
|
||||
/**
|
||||
* Find the setting value, trying first with named configs,
|
||||
* then falling back to repository and global repositories settings.
|
||||
*/
|
||||
private static <T> T getConfigValue(Settings repositorySettings, Settings globalSettings, String clientName,
|
||||
Setting.AffixSetting<T> configSetting, Setting<T> repositorySetting, Setting<T> globalSetting) {
|
||||
Setting<T> concreteSetting = configSetting.getConcreteSettingForNamespace(clientName);
|
||||
if (concreteSetting.exists(globalSettings)) {
|
||||
return concreteSetting.get(globalSettings);
|
||||
} else if (repositorySetting == null) {
|
||||
// no repository setting, just use global setting
|
||||
return globalSetting.get(globalSettings);
|
||||
} else {
|
||||
return getValue(repositorySettings, globalSettings, repositorySetting, globalSetting);
|
||||
/** Returns the value for a given setting from the repository, or returns the fallback value. */
|
||||
private static <T> T getRepoValue(Settings repositorySettings, Setting<T> repositorySetting, T fallback) {
|
||||
if (repositorySetting.exists(repositorySettings)) {
|
||||
return repositorySetting.get(repositorySettings);
|
||||
}
|
||||
return fallback;
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
@ -216,7 +198,7 @@ class InternalAwsS3Service extends AbstractLifecycleComponent implements AwsS3Se
|
|||
|
||||
@Override
|
||||
protected void doClose() throws ElasticsearchException {
|
||||
for (AmazonS3Client client : clients.values()) {
|
||||
for (AmazonS3Client client : clientsCache.values()) {
|
||||
client.shutdown();
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,183 @@
|
|||
/*
|
||||
* Licensed to Elasticsearch under one or more contributor
|
||||
* license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright
|
||||
* ownership. Elasticsearch 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.elasticsearch.repositories.s3;
|
||||
|
||||
import java.util.Collections;
|
||||
import java.util.HashMap;
|
||||
import java.util.Locale;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
|
||||
import com.amazonaws.ClientConfiguration;
|
||||
import com.amazonaws.Protocol;
|
||||
import com.amazonaws.auth.BasicAWSCredentials;
|
||||
import org.elasticsearch.common.settings.SecureSetting;
|
||||
import org.elasticsearch.common.settings.SecureString;
|
||||
import org.elasticsearch.common.settings.Setting;
|
||||
import org.elasticsearch.common.settings.Settings;
|
||||
import org.elasticsearch.common.unit.TimeValue;
|
||||
import org.elasticsearch.repositories.s3.AwsS3Service.CLOUD_S3;
|
||||
|
||||
/**
|
||||
* A container for settings used to create an S3 client.
|
||||
*/
|
||||
class S3ClientSettings {
|
||||
|
||||
// prefix for s3 client settings
|
||||
private static final String PREFIX = "s3.client.";
|
||||
|
||||
/** The access key (ie login id) for connecting to s3. */
|
||||
static final Setting.AffixSetting<SecureString> ACCESS_KEY_SETTING = Setting.affixKeySetting(PREFIX, "access_key",
|
||||
key -> SecureSetting.secureString(key, S3Repository.Repositories.KEY_SETTING));
|
||||
|
||||
/** The secret key (ie password) for connecting to s3. */
|
||||
static final Setting.AffixSetting<SecureString> SECRET_KEY_SETTING = Setting.affixKeySetting(PREFIX, "secret_key",
|
||||
key -> SecureSetting.secureString(key, S3Repository.Repositories.SECRET_SETTING));
|
||||
|
||||
/** An override for the s3 endpoint to connect to. */
|
||||
static final Setting.AffixSetting<String> ENDPOINT_SETTING = Setting.affixKeySetting(PREFIX, "endpoint",
|
||||
key -> new Setting<>(key, S3Repository.Repositories.ENDPOINT_SETTING, s -> s.toLowerCase(Locale.ROOT),
|
||||
Setting.Property.NodeScope));
|
||||
|
||||
/** The protocol to use to connect to s3. */
|
||||
static final Setting.AffixSetting<Protocol> PROTOCOL_SETTING = Setting.affixKeySetting(PREFIX, "protocol",
|
||||
key -> new Setting<>(key, "https", s -> Protocol.valueOf(s.toUpperCase(Locale.ROOT)), Setting.Property.NodeScope));
|
||||
|
||||
/** The host name of a proxy to connect to s3 through. */
|
||||
static final Setting.AffixSetting<String> PROXY_HOST_SETTING = Setting.affixKeySetting(PREFIX, "proxy.host",
|
||||
key -> Setting.simpleString(key, Setting.Property.NodeScope));
|
||||
|
||||
/** The port of a proxy to connect to s3 through. */
|
||||
static final Setting.AffixSetting<Integer> PROXY_PORT_SETTING = Setting.affixKeySetting(PREFIX, "proxy.port",
|
||||
key -> Setting.intSetting(key, 80, 0, 1<<16, Setting.Property.NodeScope));
|
||||
|
||||
/** The username of a proxy to connect to s3 through. */
|
||||
static final Setting.AffixSetting<SecureString> PROXY_USERNAME_SETTING = Setting.affixKeySetting(PREFIX, "proxy.username",
|
||||
key -> SecureSetting.secureString(key, AwsS3Service.PROXY_USERNAME_SETTING));
|
||||
|
||||
/** The password of a proxy to connect to s3 through. */
|
||||
static final Setting.AffixSetting<SecureString> PROXY_PASSWORD_SETTING = Setting.affixKeySetting(PREFIX, "proxy.password",
|
||||
key -> SecureSetting.secureString(key, AwsS3Service.PROXY_PASSWORD_SETTING));
|
||||
|
||||
/** The socket timeout for connecting to s3. */
|
||||
static final Setting.AffixSetting<TimeValue> READ_TIMEOUT_SETTING = Setting.affixKeySetting(PREFIX, "read_timeout",
|
||||
key -> Setting.timeSetting(key, TimeValue.timeValueMillis(ClientConfiguration.DEFAULT_SOCKET_TIMEOUT),
|
||||
Setting.Property.NodeScope));
|
||||
|
||||
/** Credentials to authenticate with s3. */
|
||||
final BasicAWSCredentials credentials;
|
||||
|
||||
/** The s3 endpoint the client should talk to, or empty string to use the default. */
|
||||
final String endpoint;
|
||||
|
||||
/** The protocol to use to talk to s3. Defaults to https. */
|
||||
final Protocol protocol;
|
||||
|
||||
/** An optional proxy host that requests to s3 should be made through. */
|
||||
final String proxyHost;
|
||||
|
||||
/** The port number the proxy host should be connected on. */
|
||||
final int proxyPort;
|
||||
|
||||
// these should be "secure" yet the api for the s3 client only takes String, so storing them
|
||||
// as SecureString here won't really help with anything
|
||||
/** An optional username for the proxy host, for basic authentication. */
|
||||
final String proxyUsername;
|
||||
|
||||
/** An optional password for the proxy host, for basic authentication. */
|
||||
final String proxyPassword;
|
||||
|
||||
/** The read timeout for the s3 client. */
|
||||
final int readTimeoutMillis;
|
||||
|
||||
private S3ClientSettings(BasicAWSCredentials credentials, String endpoint, Protocol protocol,
|
||||
String proxyHost, int proxyPort, String proxyUsername,
|
||||
String proxyPassword, int readTimeoutMillis) {
|
||||
this.credentials = credentials;
|
||||
this.endpoint = endpoint;
|
||||
this.protocol = protocol;
|
||||
this.proxyHost = proxyHost;
|
||||
this.proxyPort = proxyPort;
|
||||
this.proxyUsername = proxyUsername;
|
||||
this.proxyPassword = proxyPassword;
|
||||
this.readTimeoutMillis = readTimeoutMillis;
|
||||
}
|
||||
|
||||
/**
|
||||
* Load all client settings from the given settings.
|
||||
*
|
||||
* Note this will always at least return a client named "default".
|
||||
*/
|
||||
static Map<String, S3ClientSettings> load(Settings settings) {
|
||||
Set<String> clientNames = settings.getGroups(PREFIX).keySet();
|
||||
Map<String, S3ClientSettings> clients = new HashMap<>();
|
||||
for (String clientName : clientNames) {
|
||||
clients.put(clientName, getClientSettings(settings, clientName));
|
||||
}
|
||||
if (clients.containsKey("default") == false) {
|
||||
// this won't find any settings under the default client,
|
||||
// but it will pull all the fallback static settings
|
||||
clients.put("default", getClientSettings(settings, "default"));
|
||||
}
|
||||
return Collections.unmodifiableMap(clients);
|
||||
}
|
||||
|
||||
// pkg private for tests
|
||||
/** Parse settings for a single client. */
|
||||
static S3ClientSettings getClientSettings(Settings settings, String clientName) {
|
||||
try (SecureString accessKey = getConfigValue(settings, clientName, ACCESS_KEY_SETTING, S3Repository.Repositories.KEY_SETTING);
|
||||
SecureString secretKey = getConfigValue(settings, clientName, SECRET_KEY_SETTING, S3Repository.Repositories.SECRET_SETTING);
|
||||
SecureString proxyUsername = getConfigValue(settings, clientName, PROXY_USERNAME_SETTING, CLOUD_S3.PROXY_USERNAME_SETTING);
|
||||
SecureString proxyPassword = getConfigValue(settings, clientName, PROXY_PASSWORD_SETTING, CLOUD_S3.PROXY_PASSWORD_SETTING)) {
|
||||
BasicAWSCredentials credentials = null;
|
||||
if (accessKey.length() != 0) {
|
||||
if (secretKey.length() != 0) {
|
||||
credentials = new BasicAWSCredentials(accessKey.toString(), secretKey.toString());
|
||||
} else {
|
||||
throw new IllegalArgumentException("Missing secret key for s3 client [" + clientName + "]");
|
||||
}
|
||||
} else if (secretKey.length() != 0) {
|
||||
throw new IllegalArgumentException("Missing access key for s3 client [" + clientName + "]");
|
||||
}
|
||||
return new S3ClientSettings(
|
||||
credentials,
|
||||
getConfigValue(settings, clientName, ENDPOINT_SETTING, S3Repository.Repositories.ENDPOINT_SETTING),
|
||||
getConfigValue(settings, clientName, PROTOCOL_SETTING, S3Repository.Repositories.PROTOCOL_SETTING),
|
||||
getConfigValue(settings, clientName, PROXY_HOST_SETTING, AwsS3Service.CLOUD_S3.PROXY_HOST_SETTING),
|
||||
getConfigValue(settings, clientName, PROXY_PORT_SETTING, AwsS3Service.CLOUD_S3.PROXY_PORT_SETTING),
|
||||
proxyUsername.toString(),
|
||||
proxyPassword.toString(),
|
||||
(int)getConfigValue(settings, clientName, READ_TIMEOUT_SETTING, AwsS3Service.CLOUD_S3.READ_TIMEOUT).millis()
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
private static <T> T getConfigValue(Settings settings, String clientName,
|
||||
Setting.AffixSetting<T> clientSetting,
|
||||
Setting<T> globalSetting) {
|
||||
Setting<T> concreteSetting = clientSetting.getConcreteSettingForNamespace(clientName);
|
||||
if (concreteSetting.exists(settings)) {
|
||||
return concreteSetting.get(settings);
|
||||
} else {
|
||||
return globalSetting.get(settings);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -60,45 +60,6 @@ class S3Repository extends BlobStoreRepository {
|
|||
|
||||
public static final String TYPE = "s3";
|
||||
|
||||
// prefix for s3 client settings
|
||||
private static final String PREFIX = "s3.client.";
|
||||
|
||||
/** The access key (ie login id) for connecting to s3. */
|
||||
public static final AffixSetting<SecureString> ACCESS_KEY_SETTING = Setting.affixKeySetting(PREFIX, "access_key",
|
||||
key -> SecureSetting.secureString(key, Repositories.KEY_SETTING));
|
||||
|
||||
/** The secret key (ie password) for connecting to s3. */
|
||||
public static final AffixSetting<SecureString> SECRET_KEY_SETTING = Setting.affixKeySetting(PREFIX, "secret_key",
|
||||
key -> SecureSetting.secureString(key, Repositories.SECRET_SETTING));
|
||||
|
||||
/** An override for the s3 endpoint to connect to. */
|
||||
public static final AffixSetting<String> ENDPOINT_SETTING = Setting.affixKeySetting(PREFIX, "endpoint",
|
||||
key -> new Setting<>(key, Repositories.ENDPOINT_SETTING, s -> s.toLowerCase(Locale.ROOT), Property.NodeScope));
|
||||
|
||||
/** The protocol to use to connec to to s3. */
|
||||
public static final AffixSetting<Protocol> PROTOCOL_SETTING = Setting.affixKeySetting(PREFIX, "protocol",
|
||||
key -> new Setting<>(key, "https", s -> Protocol.valueOf(s.toUpperCase(Locale.ROOT)), Property.NodeScope));
|
||||
|
||||
/** The host name of a proxy to connect to s3 through. */
|
||||
public static final AffixSetting<String> PROXY_HOST_SETTING = Setting.affixKeySetting(PREFIX, "proxy.host",
|
||||
key -> Setting.simpleString(key, Property.NodeScope));
|
||||
|
||||
/** The port of a proxy to connect to s3 through. */
|
||||
public static final AffixSetting<Integer> PROXY_PORT_SETTING = Setting.affixKeySetting(PREFIX, "proxy.port",
|
||||
key -> Setting.intSetting(key, 80, 0, 1<<16, Property.NodeScope));
|
||||
|
||||
/** The username of a proxy to connect to s3 through. */
|
||||
public static final AffixSetting<SecureString> PROXY_USERNAME_SETTING = Setting.affixKeySetting(PREFIX, "proxy.username",
|
||||
key -> SecureSetting.secureString(key, AwsS3Service.PROXY_USERNAME_SETTING));
|
||||
|
||||
/** The password of a proxy to connect to s3 through. */
|
||||
public static final AffixSetting<SecureString> PROXY_PASSWORD_SETTING = Setting.affixKeySetting(PREFIX, "proxy.password",
|
||||
key -> SecureSetting.secureString(key, AwsS3Service.PROXY_PASSWORD_SETTING));
|
||||
|
||||
/** The socket timeout for connecting to s3. */
|
||||
public static final AffixSetting<TimeValue> READ_TIMEOUT_SETTING = Setting.affixKeySetting(PREFIX, "read_timeout",
|
||||
key -> Setting.timeSetting(key, TimeValue.timeValueMillis(ClientConfiguration.DEFAULT_SOCKET_TIMEOUT), Property.NodeScope));
|
||||
|
||||
/**
|
||||
* Global S3 repositories settings. Starting with: repositories.s3
|
||||
* NOTE: These are legacy settings. Use the named client config settings above.
|
||||
|
|
|
|||
|
|
@ -57,9 +57,17 @@ public class S3RepositoryPlugin extends Plugin implements RepositoryPlugin {
|
|||
});
|
||||
}
|
||||
|
||||
private final Map<String, S3ClientSettings> clientsSettings;
|
||||
|
||||
public S3RepositoryPlugin(Settings settings) {
|
||||
// eagerly load client settings so that secure settings are read
|
||||
clientsSettings = S3ClientSettings.load(settings);
|
||||
assert clientsSettings.isEmpty() == false : "always at least have 'default'";
|
||||
}
|
||||
|
||||
// overridable for tests
|
||||
protected AwsS3Service createStorageService(Settings settings) {
|
||||
return new InternalAwsS3Service(settings);
|
||||
return new InternalAwsS3Service(settings, clientsSettings);
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
@ -80,15 +88,15 @@ public class S3RepositoryPlugin extends Plugin implements RepositoryPlugin {
|
|||
return Arrays.asList(
|
||||
|
||||
// named s3 client configuration settings
|
||||
S3Repository.ACCESS_KEY_SETTING,
|
||||
S3Repository.SECRET_KEY_SETTING,
|
||||
S3Repository.ENDPOINT_SETTING,
|
||||
S3Repository.PROTOCOL_SETTING,
|
||||
S3Repository.PROXY_HOST_SETTING,
|
||||
S3Repository.PROXY_PORT_SETTING,
|
||||
S3Repository.PROXY_USERNAME_SETTING,
|
||||
S3Repository.PROXY_PASSWORD_SETTING,
|
||||
S3Repository.READ_TIMEOUT_SETTING,
|
||||
S3ClientSettings.ACCESS_KEY_SETTING,
|
||||
S3ClientSettings.SECRET_KEY_SETTING,
|
||||
S3ClientSettings.ENDPOINT_SETTING,
|
||||
S3ClientSettings.PROTOCOL_SETTING,
|
||||
S3ClientSettings.PROXY_HOST_SETTING,
|
||||
S3ClientSettings.PROXY_PORT_SETTING,
|
||||
S3ClientSettings.PROXY_USERNAME_SETTING,
|
||||
S3ClientSettings.PROXY_PASSWORD_SETTING,
|
||||
S3ClientSettings.READ_TIMEOUT_SETTING,
|
||||
|
||||
// Register global cloud aws settings: cloud.aws (might have been registered in ec2 plugin)
|
||||
AwsS3Service.KEY_SETTING,
|
||||
|
|
|
|||
|
|
@ -20,6 +20,7 @@
|
|||
package org.elasticsearch.repositories.s3;
|
||||
|
||||
import com.amazonaws.ClientConfiguration;
|
||||
import org.elasticsearch.common.settings.Settings;
|
||||
import org.elasticsearch.repositories.s3.AwsSigner;
|
||||
import org.elasticsearch.repositories.s3.S3RepositoryPlugin;
|
||||
import org.elasticsearch.test.ESTestCase;
|
||||
|
|
@ -35,7 +36,7 @@ public class AWSSignersTests extends ESTestCase {
|
|||
*/
|
||||
@BeforeClass
|
||||
public static void instantiatePlugin() {
|
||||
new S3RepositoryPlugin();
|
||||
new S3RepositoryPlugin(Settings.EMPTY);
|
||||
}
|
||||
|
||||
public void testSigners() {
|
||||
|
|
|
|||
|
|
@ -34,8 +34,8 @@ import static org.hamcrest.Matchers.is;
|
|||
public class AwsS3ServiceImplTests extends ESTestCase {
|
||||
|
||||
public void testAWSCredentialsWithSystemProviders() {
|
||||
AWSCredentialsProvider credentialsProvider =
|
||||
InternalAwsS3Service.buildCredentials(logger, deprecationLogger, Settings.EMPTY, Settings.EMPTY, "default");
|
||||
S3ClientSettings clientSettings = S3ClientSettings.getClientSettings(Settings.EMPTY, "default");
|
||||
AWSCredentialsProvider credentialsProvider = InternalAwsS3Service.buildCredentials(logger, deprecationLogger, clientSettings, Settings.EMPTY);
|
||||
assertThat(credentialsProvider, instanceOf(InternalAwsS3Service.PrivilegedInstanceProfileCredentialsProvider.class));
|
||||
}
|
||||
|
||||
|
|
@ -142,8 +142,12 @@ public class AwsS3ServiceImplTests extends ESTestCase {
|
|||
.put(S3Repository.Repositories.SECRET_SETTING.getKey(), "repositories_secret")
|
||||
.build();
|
||||
launchAWSCredentialsWithElasticsearchSettingsTest(repositorySettings, settings, "repository_key", "repository_secret");
|
||||
assertSettingDeprecationsAndWarnings(
|
||||
new Setting<?>[]{S3Repository.Repository.KEY_SETTING, S3Repository.Repository.SECRET_SETTING});
|
||||
assertSettingDeprecationsAndWarnings(new Setting<?>[]{
|
||||
S3Repository.Repositories.KEY_SETTING,
|
||||
S3Repository.Repositories.SECRET_SETTING,
|
||||
S3Repository.Repository.KEY_SETTING,
|
||||
S3Repository.Repository.SECRET_SETTING},
|
||||
"Using s3 access/secret key from repository settings. Instead store these in named clients and the elasticsearch keystore for secure settings.");
|
||||
}
|
||||
|
||||
public void testAWSCredentialsWithElasticsearchAwsAndRepositoriesSettingsAndRepositorySettingsBackcompat() {
|
||||
|
|
@ -155,8 +159,14 @@ public class AwsS3ServiceImplTests extends ESTestCase {
|
|||
.put(S3Repository.Repositories.SECRET_SETTING.getKey(), "repositories_secret")
|
||||
.build();
|
||||
launchAWSCredentialsWithElasticsearchSettingsTest(repositorySettings, settings, "repository_key", "repository_secret");
|
||||
assertSettingDeprecationsAndWarnings(
|
||||
new Setting<?>[]{S3Repository.Repository.KEY_SETTING, S3Repository.Repository.SECRET_SETTING});
|
||||
assertSettingDeprecationsAndWarnings(new Setting<?>[]{
|
||||
AwsS3Service.KEY_SETTING,
|
||||
AwsS3Service.SECRET_SETTING,
|
||||
S3Repository.Repositories.KEY_SETTING,
|
||||
S3Repository.Repositories.SECRET_SETTING,
|
||||
S3Repository.Repository.KEY_SETTING,
|
||||
S3Repository.Repository.SECRET_SETTING},
|
||||
"Using s3 access/secret key from repository settings. Instead store these in named clients and the elasticsearch keystore for secure settings.");
|
||||
}
|
||||
|
||||
public void testAWSCredentialsWithElasticsearchAwsAndS3AndRepositoriesSettingsAndRepositorySettingsBackcompat() {
|
||||
|
|
@ -170,15 +180,25 @@ public class AwsS3ServiceImplTests extends ESTestCase {
|
|||
.put(S3Repository.Repositories.SECRET_SETTING.getKey(), "repositories_secret")
|
||||
.build();
|
||||
launchAWSCredentialsWithElasticsearchSettingsTest(repositorySettings, settings, "repository_key", "repository_secret");
|
||||
assertSettingDeprecationsAndWarnings(
|
||||
new Setting<?>[]{S3Repository.Repository.KEY_SETTING, S3Repository.Repository.SECRET_SETTING});
|
||||
assertSettingDeprecationsAndWarnings(new Setting<?>[]{
|
||||
AwsS3Service.KEY_SETTING,
|
||||
AwsS3Service.SECRET_SETTING,
|
||||
AwsS3Service.CLOUD_S3.KEY_SETTING,
|
||||
AwsS3Service.CLOUD_S3.SECRET_SETTING,
|
||||
S3Repository.Repositories.KEY_SETTING,
|
||||
S3Repository.Repositories.SECRET_SETTING,
|
||||
S3Repository.Repository.KEY_SETTING,
|
||||
S3Repository.Repository.SECRET_SETTING},
|
||||
"Using s3 access/secret key from repository settings. Instead store these in named clients and the elasticsearch keystore for secure settings.");
|
||||
}
|
||||
|
||||
protected void launchAWSCredentialsWithElasticsearchSettingsTest(Settings singleRepositorySettings, Settings settings,
|
||||
String expectedKey, String expectedSecret) {
|
||||
String configName = InternalAwsS3Service.CLIENT_NAME.get(singleRepositorySettings);
|
||||
AWSCredentials credentials = InternalAwsS3Service.buildCredentials(logger, deprecationLogger, settings,
|
||||
singleRepositorySettings, configName).getCredentials();
|
||||
S3ClientSettings clientSettings = S3ClientSettings.getClientSettings(settings, configName);
|
||||
AWSCredentials credentials = InternalAwsS3Service
|
||||
.buildCredentials(logger, deprecationLogger, clientSettings, singleRepositorySettings)
|
||||
.getCredentials();
|
||||
assertThat(credentials.getAWSAccessKeyId(), is(expectedKey));
|
||||
assertThat(credentials.getAWSSecretKey(), is(expectedSecret));
|
||||
}
|
||||
|
|
@ -287,8 +307,9 @@ public class AwsS3ServiceImplTests extends ESTestCase {
|
|||
Boolean useThrottleRetries = S3Repository.getValue(singleRepositorySettings, settings,
|
||||
S3Repository.Repository.USE_THROTTLE_RETRIES_SETTING, S3Repository.Repositories.USE_THROTTLE_RETRIES_SETTING);
|
||||
|
||||
ClientConfiguration configuration = InternalAwsS3Service.buildConfiguration(logger, singleRepositorySettings, settings,
|
||||
"default", maxRetries, null, useThrottleRetries);
|
||||
S3ClientSettings clientSettings = S3ClientSettings.getClientSettings(settings, "default");
|
||||
ClientConfiguration configuration = InternalAwsS3Service.buildConfiguration(logger, clientSettings,
|
||||
singleRepositorySettings, maxRetries, null, useThrottleRetries);
|
||||
|
||||
assertThat(configuration.getResponseMetadataCacheSize(), is(0));
|
||||
assertThat(configuration.getProtocol(), is(expectedProtocol));
|
||||
|
|
@ -344,7 +365,8 @@ public class AwsS3ServiceImplTests extends ESTestCase {
|
|||
private void assertEndpoint(Settings repositorySettings, Settings settings,
|
||||
String expectedEndpoint) {
|
||||
String configName = InternalAwsS3Service.CLIENT_NAME.get(repositorySettings);
|
||||
String foundEndpoint = InternalAwsS3Service.findEndpoint(logger, repositorySettings, settings, configName);
|
||||
S3ClientSettings clientSettings = S3ClientSettings.getClientSettings(settings, configName);
|
||||
String foundEndpoint = InternalAwsS3Service.findEndpoint(logger, clientSettings, repositorySettings);
|
||||
assertThat(foundEndpoint, is(expectedEndpoint));
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -28,6 +28,9 @@ import org.elasticsearch.common.settings.Settings;
|
|||
|
||||
public class TestAwsS3Service extends InternalAwsS3Service {
|
||||
public static class TestPlugin extends S3RepositoryPlugin {
|
||||
public TestPlugin(Settings settings) {
|
||||
super(settings);
|
||||
}
|
||||
@Override
|
||||
protected AwsS3Service createStorageService(Settings settings) {
|
||||
return new TestAwsS3Service(settings);
|
||||
|
|
@ -37,10 +40,9 @@ public class TestAwsS3Service extends InternalAwsS3Service {
|
|||
IdentityHashMap<AmazonS3, TestAmazonS3> clients = new IdentityHashMap<>();
|
||||
|
||||
public TestAwsS3Service(Settings settings) {
|
||||
super(settings);
|
||||
super(settings, S3ClientSettings.load(settings));
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
public synchronized AmazonS3 client(Settings repositorySettings) {
|
||||
return cachedWrapper(super.client(repositorySettings));
|
||||
|
|
|
|||
|
|
@ -1,12 +1,7 @@
|
|||
# Integration tests for Repository S3 component
|
||||
#
|
||||
"S3 repository can be registered":
|
||||
- skip:
|
||||
features: warnings
|
||||
- do:
|
||||
warnings:
|
||||
- "[access_key] setting was deprecated in Elasticsearch and will be removed in a future release! See the breaking changes documentation for the next major version."
|
||||
- "[secret_key] setting was deprecated in Elasticsearch and will be removed in a future release! See the breaking changes documentation for the next major version."
|
||||
snapshot.create_repository:
|
||||
repository: test_repo_s3_1
|
||||
verify: false
|
||||
|
|
@ -14,8 +9,6 @@
|
|||
type: s3
|
||||
settings:
|
||||
bucket: "my_bucket_name"
|
||||
access_key: "AKVAIQBF2RECL7FJWGJQ"
|
||||
secret_key: "vExyMThREXeRMm/b/LRzEB8jWwvzQeXgjqMX+6br"
|
||||
canned_acl: "public-read"
|
||||
storage_class: "standard"
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue