honeymoose
/
OpenSearch
mirror of https://github.com/honeymoose/OpenSearch.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' into immutable_map_be_gone 

Original commit: elastic/x-pack-elasticsearch@62358ec345
This commit is contained in:
Nik Everett 2015-10-09 12:25:28 -04:00
parent 0ff0cb7cd5 2f1c88a633
commit 2bde3de3f0
37 changed files with 621 additions and 621 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
marvel/pom.xml
View File

@ -42,6 +42,13 @@
<scope>provided</scope>
</dependency>
<dependency>
<groupId>com.squareup.okhttp</groupId>
<artifactId>mockwebserver</artifactId>
<version>2.3.0</version>
<scope>test</scope>
</dependency>
</dependencies>
<build>

3
marvel/src/main/java/org/elasticsearch/marvel/agent/collector/cluster/ClusterInfoCollector.java
View File

@ -60,7 +60,8 @@ public class ClusterInfoCollector extends AbstractCollector<ClusterInfoMarvelDoc
List<MarvelDoc> results = new ArrayList<>(1);
// Retrieves all licenses
List<License> licenses = licenseService.licenses();
// TODO: we should only work with one license
List<License> licenses = Collections.singletonList(licenseService.license());
// Retrieves additional cluster stats
ClusterStatsResponse clusterStats = client.admin().cluster().prepareClusterStats().get(marvelSettings.clusterStatsTimeout());

5
marvel/src/main/java/org/elasticsearch/marvel/agent/exporter/http/HttpExporter.java
View File

@ -54,6 +54,7 @@ public class HttpExporter extends Exporter {
public static final String HOST_SETTING = "host";
public static final String CONNECTION_TIMEOUT_SETTING = "connection.timeout";
public static final String CONNECTION_READ_TIMEOUT_SETTING = "connection.read_timeout";
public static final String CONNECTION_KEEP_ALIVE_SETTING = "connection.keep_alive";
public static final String AUTH_USERNAME_SETTING = "auth.username";
public static final String AUTH_PASSWORD_SETTING = "auth.password";
@ -91,6 +92,7 @@ public class HttpExporter extends Exporter {
/** Version of the built-in template **/
final Version templateVersion;
boolean keepAlive;
final ConnectionKeepAliveWorker keepAliveWorker;
Thread keepAliveThread;
@ -117,6 +119,7 @@ public class HttpExporter extends Exporter {
String templateCheckTimeoutValue = config.settings().get(TEMPLATE_CHECK_TIMEOUT_SETTING, null);
templateCheckTimeout = TimeValue.parseTimeValue(templateCheckTimeoutValue, null, settingFQN(TEMPLATE_CHECK_TIMEOUT_SETTING));
keepAlive = config.settings().getAsBoolean(CONNECTION_KEEP_ALIVE_SETTING, true);
keepAliveWorker = new ConnectionKeepAliveWorker();
sslSocketFactory = createSSLSocketFactory(config.settings().getAsSettings(SSL_SETTING));
@ -511,10 +514,12 @@ public class HttpExporter extends Exporter {
}
protected void initKeepAliveThread() {
if (keepAlive) {
keepAliveThread = new Thread(keepAliveWorker, "marvel-exporter[" + config.name() + "][keep_alive]");
keepAliveThread.setDaemon(true);
keepAliveThread.start();
}
}
static private void validateHosts(String[] hosts) {

2
marvel/src/main/java/org/elasticsearch/marvel/agent/renderer/cluster/ClusterInfoRenderer.java
View File

@ -37,7 +37,6 @@ public class ClusterInfoRenderer extends AbstractRenderer<ClusterInfoMarvelDoc>
builder.field(Fields.UID, license.uid());
builder.field(Fields.TYPE, license.type());
builder.dateValueField(Fields.ISSUE_DATE_IN_MILLIS, Fields.ISSUE_DATE, license.issueDate());
builder.field(Fields.FEATURE, license.feature());
builder.dateValueField(Fields.EXPIRY_DATE_IN_MILLIS, Fields.EXPIRY_DATE, license.expiryDate());
builder.field(Fields.MAX_NODES, license.maxNodes());
builder.field(Fields.ISSUED_TO, license.issuedTo());
@ -77,7 +76,6 @@ public class ClusterInfoRenderer extends AbstractRenderer<ClusterInfoMarvelDoc>
static final XContentBuilderString STATUS = new XContentBuilderString("status");
static final XContentBuilderString UID = new XContentBuilderString("uid");
static final XContentBuilderString TYPE = new XContentBuilderString("type");
static final XContentBuilderString FEATURE = new XContentBuilderString("feature");
static final XContentBuilderString ISSUE_DATE_IN_MILLIS = new XContentBuilderString("issue_date_in_millis");
static final XContentBuilderString ISSUE_DATE = new XContentBuilderString("issue_date");
static final XContentBuilderString EXPIRY_DATE_IN_MILLIS = new XContentBuilderString("expiry_date_in_millis");

1
marvel/src/main/java/org/elasticsearch/marvel/agent/renderer/cluster/ClusterStateRenderer.java
View File

@ -20,6 +20,7 @@ public class ClusterStateRenderer extends AbstractRenderer<ClusterStateMarvelDoc
public static final String[] FILTERS = {
"cluster_state.version",
"cluster_state.master_node",
"cluster_state.state_uuid",
"cluster_state.status",
"cluster_state.nodes",
};

160
marvel/src/main/java/org/elasticsearch/marvel/license/LicenseService.java
View File

@ -6,110 +6,45 @@
package org.elasticsearch.marvel.license;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.common.Strings;
import org.elasticsearch.common.component.AbstractLifecycleComponent;
import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.joda.FormatDateTimeFormatter;
import org.elasticsearch.common.joda.Joda;
import org.elasticsearch.common.logging.support.LoggerMessageFormat;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.license.core.License;
import org.elasticsearch.license.plugin.core.LicensesClientService;
import org.elasticsearch.license.plugin.core.LicenseState;
import org.elasticsearch.license.plugin.core.Licensee;
import org.elasticsearch.license.plugin.core.LicenseeRegistry;
import org.elasticsearch.license.plugin.core.LicensesManagerService;
import org.elasticsearch.license.plugin.core.LicensesService;
import org.elasticsearch.marvel.MarvelPlugin;
import org.elasticsearch.marvel.agent.settings.MarvelSettings;
import org.elasticsearch.marvel.mode.Mode;
import java.util.*;
public class LicenseService extends AbstractLifecycleComponent<LicenseService> {
public class LicenseService extends AbstractLifecycleComponent<LicenseService> implements Licensee {
public static final String FEATURE_NAME = MarvelPlugin.NAME;
private static final LicensesService.TrialLicenseOptions TRIAL_LICENSE_OPTIONS =
new LicensesService.TrialLicenseOptions(TimeValue.timeValueHours(30 * 24), 1000);
private static final FormatDateTimeFormatter DATE_FORMATTER = Joda.forPattern("EEEE, MMMMM dd, yyyy", Locale.ROOT);
static final TimeValue GRACE_PERIOD = days(7);
private final LicensesManagerService managerService;
private final LicensesClientService clientService;
private final LicenseeRegistry clientService;
private final MarvelSettings marvelSettings;
private final Collection<LicensesService.ExpirationCallback> expirationLoggers;
private final LicensesClientService.AcknowledgementCallback acknowledgementCallback;
private volatile Mode mode;
private volatile boolean enabled;
private volatile LicenseState state;
private volatile long expiryDate;
@Inject
public LicenseService(Settings settings, LicensesClientService clientService, LicensesManagerService managerService, MarvelSettings marvelSettings) {
public LicenseService(Settings settings, LicenseeRegistry clientService, LicensesManagerService managerService, MarvelSettings marvelSettings) {
super(settings);
this.managerService = managerService;
this.clientService = clientService;
this.marvelSettings = marvelSettings;
this.mode = Mode.LITE;
this.expirationLoggers = Arrays.asList(
new LicensesService.ExpirationCallback.Pre(days(7), days(30), days(1)) {
@Override
public void on(License license, LicensesService.ExpirationStatus status) {
logger.error("\n" +
"#\n" +
"# Marvel license will expire on [{}].\n" +
"# Have a new license? please update it. Otherwise, please reach out to your support contact.\n" +
"#", DATE_FORMATTER.printer().print(license.expiryDate()));
}
},
new LicensesService.ExpirationCallback.Pre(days(0), days(7), minutes(10)) {
@Override
public void on(License license, LicensesService.ExpirationStatus status) {
logger.error("\n" +
"#\n" +
"# Marvel license will expire on [{}].\n" +
"# Have a new license? please update it. Otherwise, please reach out to your support contact.\n" +
"#", DATE_FORMATTER.printer().print(license.expiryDate()));
}
},
new LicensesService.ExpirationCallback.Post(days(0), GRACE_PERIOD, minutes(10)) {
@Override
public void on(License license, LicensesService.ExpirationStatus status) {
long endOfGracePeriod = license.expiryDate() + GRACE_PERIOD.getMillis();
logger.error("\n" +
"#\n" +
"# MARVEL LICENSE HAS EXPIRED ON [{}].\n" +
"# MARVEL WILL STOP COLLECTING DATA ON [{}].\n" +
"# HAVE A NEW LICENSE? PLEASE UPDATE IT. OTHERWISE, PLEASE REACH OUT TO YOUR SUPPORT CONTACT.\n" +
"#", DATE_FORMATTER.printer().print(endOfGracePeriod), DATE_FORMATTER.printer().print(license.expiryDate()));
}
}
);
this.acknowledgementCallback = new LicensesClientService.AcknowledgementCallback() {
@Override
public List<String> acknowledge(License currentLicense, License newLicense) {
switch (newLicense.type()) {
case "trial":
case "gold":
case "platinum":
return Collections.emptyList();
default: // "basic" - we also fall back to basic for an unknown type
return Collections.singletonList(LoggerMessageFormat.format(
"Marvel: Multi-cluster support is disabled for clusters with [{}] licenses.\n" +
"If you are running multiple customers, users won't be able to access this\n" +
"all the clusters with [{}] licenses from a single Marvel instance. To access them\n" +
"a dedicated and separated marvel instance will be required for each cluster",
newLicense.type(), newLicense.type()));
}
}
};
}
@Override
protected void doStart() throws ElasticsearchException {
clientService.register(FEATURE_NAME, TRIAL_LICENSE_OPTIONS, expirationLoggers, acknowledgementCallback, new InternalListener(this));
clientService.register(this);
}
@Override
@ -120,14 +55,6 @@ public class LicenseService extends AbstractLifecycleComponent<LicenseService> {
protected void doClose() throws ElasticsearchException {
}
static TimeValue days(int days) {
return TimeValue.timeValueHours(days * 24);
}
static TimeValue minutes(int minutes) {
return TimeValue.timeValueMinutes(minutes);
}
/**
* @return the current marvel's operating mode
*/
@ -138,18 +65,20 @@ public class LicenseService extends AbstractLifecycleComponent<LicenseService> {
/**
* @return all registered licenses
*/
public List<License> licenses() {
return managerService.getLicenses();
public License license() {
return managerService.getLicense();
}
/**
* @return true if the marvel license is enabled
*/
public boolean enabled() {
return enabled;
return state == LicenseState.ENABLED || state == LicenseState.GRACE_PERIOD;
}
/**
* TODO: remove licensing grace period, just check for state == LicensesClientService.LicenseState.GRACE_PERIOD instead
*
* @return true if marvel is running within the "grace period", ie when the license
* is expired but a given extra delay is not yet elapsed
*/
@ -164,30 +93,57 @@ public class LicenseService extends AbstractLifecycleComponent<LicenseService> {
return expiryDate;
}
class InternalListener implements LicensesClientService.Listener {
private final LicenseService service;
public InternalListener(LicenseService service) {
this.service = service;
@Override
public String id() {
return FEATURE_NAME;
}
@Override
public void onEnabled(License license) {
public String[] expirationMessages() {
// TODO add messages to be logged around license expiry
return Strings.EMPTY_ARRAY;
}
@Override
public String[] acknowledgmentMessages(License currentLicense, License newLicense) {
switch (newLicense.operationMode()) {
case BASIC:
if (currentLicense != null) {
switch (currentLicense.operationMode()) {
case TRIAL:
case GOLD:
case PLATINUM:
return new String[] {
LoggerMessageFormat.format(
"Multi-cluster support is disabled for clusters with [{}] licenses.\n" +
"If you are running multiple customers, users won't be able to access this\n" +
"all the clusters with [{}] licenses from a single Marvel instance. To access them\n" +
"a dedicated and separated marvel instance will be required for each cluster",
newLicense.type(), newLicense.type())
};
}
}
}
return Strings.EMPTY_ARRAY;
}
@Override
public void onChange(License license, LicenseState state) {
synchronized (this) {
this.state = state;
if (license != null) {
try {
service.enabled = true;
service.expiryDate = license.expiryDate();
service.mode = Mode.fromName(license.type());
mode = Mode.fromName(license.type());
} catch (IllegalArgumentException e) {
service.mode = Mode.LITE;
mode = Mode.LITE;
}
expiryDate = license.expiryDate();
} else {
mode = Mode.LITE;
}
if (state == LicenseState.DISABLED) {
mode = Mode.LITE;
}
@Override
public void onDisabled(License license) {
service.enabled = false;
service.expiryDate = license.expiryDate();
service.mode = Mode.LITE;
}
}
}

12
marvel/src/main/java/org/elasticsearch/marvel/mode/Mode.java
View File

@ -16,6 +16,8 @@ public enum Mode {
/**
* Marvel runs in downgraded mode
*
* TODO: do we really need mode?
*/
TRIAL(0),
@ -55,9 +57,13 @@ public enum Mode {
public static Mode fromName(String name) {
switch (name.toLowerCase(Locale.ROOT)) {
case "trial": return TRIAL;
case "lite": return LITE;
case "standard" : return STANDARD;
case "trial":
return LITE;
case "basic":
case "gold" :
case "silver":
case "platinum":
return STANDARD;
default:
throw new ElasticsearchException("unknown marvel mode name [" + name + "]");
}

2
marvel/src/main/java/org/elasticsearch/marvel/shield/MarvelInternalUserHolder.java
View File

@ -33,7 +33,7 @@ public class MarvelInternalUserHolder {
// and full access to .marvel-* and .marvel-data indices
.add(Privilege.Index.ALL, MarvelSettings.MARVEL_INDICES_PREFIX + "*")
// note, we don't need _licenses permission as we're taking the licenses
// note, we don't need _license permission as we're taking the licenses
// directly form the license service.
.build();

32
marvel/src/main/java/org/elasticsearch/marvel/shield/SecuredClient.java
View File

@ -82,6 +82,10 @@ import org.elasticsearch.action.admin.cluster.tasks.PendingClusterTasksAction;
import org.elasticsearch.action.admin.cluster.tasks.PendingClusterTasksRequest;
import org.elasticsearch.action.admin.cluster.tasks.PendingClusterTasksRequestBuilder;
import org.elasticsearch.action.admin.cluster.tasks.PendingClusterTasksResponse;
import org.elasticsearch.action.admin.cluster.validate.template.RenderSearchTemplateAction;
import org.elasticsearch.action.admin.cluster.validate.template.RenderSearchTemplateRequest;
import org.elasticsearch.action.admin.cluster.validate.template.RenderSearchTemplateRequestBuilder;
import org.elasticsearch.action.admin.cluster.validate.template.RenderSearchTemplateResponse;
import org.elasticsearch.action.admin.indices.alias.IndicesAliasesAction;
import org.elasticsearch.action.admin.indices.alias.IndicesAliasesRequest;
import org.elasticsearch.action.admin.indices.alias.IndicesAliasesRequestBuilder;
@ -194,10 +198,6 @@ import org.elasticsearch.action.admin.indices.validate.query.ValidateQueryAction
import org.elasticsearch.action.admin.indices.validate.query.ValidateQueryRequest;
import org.elasticsearch.action.admin.indices.validate.query.ValidateQueryRequestBuilder;
import org.elasticsearch.action.admin.indices.validate.query.ValidateQueryResponse;
import org.elasticsearch.action.admin.indices.validate.template.RenderSearchTemplateAction;
import org.elasticsearch.action.admin.indices.validate.template.RenderSearchTemplateRequest;
import org.elasticsearch.action.admin.indices.validate.template.RenderSearchTemplateRequestBuilder;
import org.elasticsearch.action.admin.indices.validate.template.RenderSearchTemplateResponse;
import org.elasticsearch.action.admin.indices.warmer.delete.DeleteWarmerAction;
import org.elasticsearch.action.admin.indices.warmer.delete.DeleteWarmerRequest;
import org.elasticsearch.action.admin.indices.warmer.delete.DeleteWarmerRequestBuilder;
@ -1047,18 +1047,6 @@ public class SecuredClient implements Client {
return new ValidateQueryRequestBuilder(this, ValidateQueryAction.INSTANCE).setIndices(indices);
}
public ActionFuture<RenderSearchTemplateResponse> renderSearchTemplate(RenderSearchTemplateRequest request) {
return this.execute(RenderSearchTemplateAction.INSTANCE, request);
}
public void renderSearchTemplate(RenderSearchTemplateRequest request, ActionListener<RenderSearchTemplateResponse> listener) {
this.execute(RenderSearchTemplateAction.INSTANCE, request, listener);
}
public RenderSearchTemplateRequestBuilder prepareRenderSearchTemplate() {
return new RenderSearchTemplateRequestBuilder(this, RenderSearchTemplateAction.INSTANCE);
}
public ActionFuture<PutWarmerResponse> putWarmer(PutWarmerRequest request) {
return this.execute(PutWarmerAction.INSTANCE, request);
}
@ -1370,6 +1358,18 @@ public class SecuredClient implements Client {
public SnapshotsStatusRequestBuilder prepareSnapshotStatus() {
return new SnapshotsStatusRequestBuilder(this, SnapshotsStatusAction.INSTANCE);
}
public ActionFuture<RenderSearchTemplateResponse> renderSearchTemplate(RenderSearchTemplateRequest request) {
return this.execute(RenderSearchTemplateAction.INSTANCE, request);
}
public void renderSearchTemplate(RenderSearchTemplateRequest request, ActionListener<RenderSearchTemplateResponse> listener) {
this.execute(RenderSearchTemplateAction.INSTANCE, request, listener);
}
public RenderSearchTemplateRequestBuilder prepareRenderSearchTemplate() {
return new RenderSearchTemplateRequestBuilder(this, RenderSearchTemplateAction.INSTANCE);
}
}
static class Admin implements AdminClient {

29
marvel/src/main/resources/marvel_index_template.json
View File

@ -178,6 +178,10 @@
"type": "string",
"index": "not_analyzed"
},
"state_uuid": {
"type": "string",
"index": "not_analyzed"
},
"status": {
"type": "string",
"index": "not_analyzed"
@ -248,7 +252,30 @@
"index": "not_analyzed"
},
"shard": {
"type": "object"
"properties": {
"state": {
"type": "string",
"index": "not_analyzed"
},
"primary": {
"type": "boolean"
},
"index": {
"type": "string",
"index": "not_analyzed"
},
"relocating_node": {
"type": "string",
"index": "not_analyzed"
},
"shard": {
"type": "long"
},
"node": {
"type": "string",
"index": "not_analyzed"
}
}
}
}
}

54
marvel/src/test/java/org/elasticsearch/marvel/agent/collector/AbstractCollectorTestCase.java
View File

@ -17,12 +17,11 @@ import org.elasticsearch.common.inject.Module;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.license.core.License;
import org.elasticsearch.license.plugin.core.LicensesClientService;
import org.elasticsearch.license.plugin.core.LicensesManagerService;
import org.elasticsearch.license.plugin.core.LicensesService;
import org.elasticsearch.license.plugin.action.delete.DeleteLicenseRequest;
import org.elasticsearch.license.plugin.action.put.PutLicenseRequest;
import org.elasticsearch.license.plugin.core.*;
import org.elasticsearch.marvel.MarvelPlugin;
import org.elasticsearch.marvel.agent.settings.MarvelSettings;
import org.elasticsearch.marvel.license.LicenseService;
import org.elasticsearch.marvel.shield.MarvelShieldIntegration;
import org.elasticsearch.marvel.shield.SecuredClient;
import org.elasticsearch.marvel.test.MarvelIntegTestCase;
@ -88,15 +87,13 @@ public class AbstractCollectorTestCase extends MarvelIntegTestCase {
private static License createTestingLicense(long issueDate, long expiryDate) {
return License.builder()
.feature(LicenseService.FEATURE_NAME)
.expiryDate(expiryDate)
.issueDate(issueDate)
.issuedTo("AbstractCollectorTestCase")
.issuer("test")
.maxNodes(Integer.MAX_VALUE)
.signature("_signature")
.type("standard")
.subscriptionType("all_is_good")
.type("basic")
.uid(String.valueOf(RandomizedTest.systemPropertyAsInt(SysGlobals.CHILDVM_SYSPROP_JVM_ID, 0)) + System.identityHashCode(AbstractCollectorTestCase.class))
.build();
}
@ -107,7 +104,7 @@ public class AbstractCollectorTestCase extends MarvelIntegTestCase {
final License license = createTestingLicense(issueDate, expiryDate);
for (LicenseServiceForCollectors service : internalCluster().getInstances(LicenseServiceForCollectors.class)) {
service.enable(license);
service.onChange(license, LicenseState.ENABLED);
}
for (LicensesManagerServiceForCollectors service : internalCluster().getInstances(LicensesManagerServiceForCollectors.class)) {
service.update(license);
@ -120,7 +117,7 @@ public class AbstractCollectorTestCase extends MarvelIntegTestCase {
final License license = createTestingLicense(issueDate, expiryDate);
for (LicenseServiceForCollectors service : internalCluster().getInstances(LicenseServiceForCollectors.class)) {
service.disable(license);
service.onChange(license, LicenseState.GRACE_PERIOD);
}
for (LicensesManagerServiceForCollectors service : internalCluster().getInstances(LicensesManagerServiceForCollectors.class)) {
service.update(license);
@ -133,7 +130,7 @@ public class AbstractCollectorTestCase extends MarvelIntegTestCase {
final License license = createTestingLicense(issueDate, expiryDate);
for (LicenseServiceForCollectors service : internalCluster().getInstances(LicenseServiceForCollectors.class)) {
service.disable(license);
service.onChange(license, LicenseState.DISABLED);
}
for (LicensesManagerServiceForCollectors service : internalCluster().getInstances(LicensesManagerServiceForCollectors.class)) {
service.update(license);
@ -146,7 +143,7 @@ public class AbstractCollectorTestCase extends MarvelIntegTestCase {
final License license = createTestingLicense(issueDate, expiryDate);
for (LicenseServiceForCollectors service : internalCluster().getInstances(LicenseServiceForCollectors.class)) {
service.disable(license);
service.onChange(license, LicenseState.DISABLED);
}
for (LicensesManagerServiceForCollectors service : internalCluster().getInstances(LicensesManagerServiceForCollectors.class)) {
service.update(license);
@ -205,7 +202,7 @@ public class AbstractCollectorTestCase extends MarvelIntegTestCase {
@Override
protected void configure() {
bind(LicenseServiceForCollectors.class).asEagerSingleton();
bind(LicensesClientService.class).to(LicenseServiceForCollectors.class);
bind(LicenseeRegistry.class).to(LicenseServiceForCollectors.class);
bind(LicensesManagerServiceForCollectors.class).asEagerSingleton();
bind(LicensesManagerService.class).to(LicensesManagerServiceForCollectors.class);
}
@ -213,9 +210,9 @@ public class AbstractCollectorTestCase extends MarvelIntegTestCase {
}
}
public static class LicenseServiceForCollectors extends AbstractComponent implements LicensesClientService {
public static class LicenseServiceForCollectors extends AbstractComponent implements LicenseeRegistry {
private final List<Listener> listeners = new ArrayList<>();
private final List<Licensee> licensees = new ArrayList<>();
@Inject
public LicenseServiceForCollectors(Settings settings) {
@ -223,19 +220,13 @@ public class AbstractCollectorTestCase extends MarvelIntegTestCase {
}
@Override
public void register(String feature, TrialLicenseOptions trialLicenseOptions, Collection<ExpirationCallback> expirationCallbacks, AcknowledgementCallback acknowledgementCallback, Listener listener) {
listeners.add(listener);
public void register(Licensee licensee) {
licensees.add(licensee);
}
public void enable(License license) {
for (Listener listener : listeners) {
listener.onEnabled(license);
}
}
public void disable(License license) {
for (Listener listener : listeners) {
listener.onDisabled(license);
public void onChange(License license, LicenseState state) {
for (Licensee licensee : licensees) {
licensee.onChange(license, state);
}
}
}
@ -245,21 +236,24 @@ public class AbstractCollectorTestCase extends MarvelIntegTestCase {
private final Map<String, License> licenses = Collections.synchronizedMap(new HashMap<String, License>());
@Override
public void registerLicenses(LicensesService.PutLicenseRequestHolder requestHolder, ActionListener<LicensesService.LicensesUpdateResponse> listener) {
public void registerLicense(PutLicenseRequest request, ActionListener<LicensesService.LicensesUpdateResponse> listener) {
}
@Override
public void removeLicenses(LicensesService.DeleteLicenseRequestHolder requestHolder, ActionListener<ClusterStateUpdateResponse> listener) {
public void removeLicense(DeleteLicenseRequest request, ActionListener<ClusterStateUpdateResponse> listener) {
}
@Override
public Set<String> enabledFeatures() {
public List<String> licenseesWithState(LicenseState state) {
return null;
}
@Override
public List<License> getLicenses() {
return new ArrayList<>(licenses.values());
public License getLicense() {
// TODO: we only take the first of the licenses that are updated
// FIXME
Iterator<License> iterator = licenses.values().iterator();
return iterator.hasNext() ? iterator.next() : null;
}
public void update(License license) {

403
marvel/src/test/java/org/elasticsearch/marvel/agent/exporter/http/HttpExporterTests.java
View File

@ -5,134 +5,114 @@
*/
package org.elasticsearch.marvel.agent.exporter.http;
import com.squareup.okhttp.mockwebserver.MockResponse;
import com.squareup.okhttp.mockwebserver.MockWebServer;
import com.squareup.okhttp.mockwebserver.QueueDispatcher;
import com.squareup.okhttp.mockwebserver.RecordedRequest;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.Version;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthStatus;
import org.elasticsearch.action.admin.indices.recovery.RecoveryResponse;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.common.bytes.BytesArray;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.TransportAddress;
import org.elasticsearch.http.HttpServerTransport;
import org.elasticsearch.common.xcontent.XContentHelper;
import org.elasticsearch.marvel.agent.collector.cluster.ClusterStateCollector;
import org.elasticsearch.marvel.agent.collector.cluster.ClusterStateMarvelDoc;
import org.elasticsearch.marvel.agent.collector.indices.IndexRecoveryCollector;
import org.elasticsearch.marvel.agent.collector.indices.IndexRecoveryMarvelDoc;
import org.elasticsearch.marvel.agent.exporter.Exporters;
import org.elasticsearch.marvel.agent.exporter.MarvelDoc;
import org.elasticsearch.marvel.agent.exporter.MarvelTemplateUtils;
import org.elasticsearch.marvel.agent.settings.MarvelSettings;
import org.elasticsearch.marvel.test.MarvelIntegTestCase;
import org.elasticsearch.node.Node;
import org.elasticsearch.test.ESIntegTestCase.ClusterScope;
import org.elasticsearch.test.ESIntegTestCase.SuppressLocalMode;
import org.elasticsearch.test.InternalTestCluster;
import org.elasticsearch.test.ESIntegTestCase;
import org.elasticsearch.test.ESIntegTestCase.Scope;
import org.hamcrest.Matchers;
import org.joda.time.format.DateTimeFormat;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import java.io.IOException;
import java.net.BindException;
import java.util.Collections;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.Map;
import static org.elasticsearch.test.ESIntegTestCase.Scope.TEST;
import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.notNullValue;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.greaterThanOrEqualTo;
import org.apache.lucene.util.LuceneTestCase.AwaitsFix;
// Transport Client instantiation also calls the marvel plugin, which then fails to find modules
@SuppressLocalMode
@ClusterScope(scope = TEST, transportClientRatio = 0.0, numDataNodes = 0, numClientNodes = 0)
@AwaitsFix(bugUrl = "https://github.com/elastic/x-plugins/issues/729")
@ESIntegTestCase.ClusterScope(scope = Scope.TEST, numDataNodes = 0, numClientNodes = 0, transportClientRatio = 0.0)
public class HttpExporterTests extends MarvelIntegTestCase {
final static AtomicLong timeStampGenerator = new AtomicLong();
@Override
protected boolean enableShield() {
return false;
}
private int webPort;
private MockWebServer webServer;
@Before
public void init() throws Exception {
startCollection();
public void startWebservice() throws Exception {
for (webPort = 9250; webPort < 9300; webPort++) {
try {
webServer = new MockWebServer();
QueueDispatcher dispatcher = new QueueDispatcher();
dispatcher.setFailFast(true);
webServer.setDispatcher(dispatcher);
webServer.start(webPort);
return;
} catch (BindException be) {
logger.warn("port [{}] was already in use trying next port", webPort);
}
}
throw new ElasticsearchException("unable to find open port between 9200 and 9300");
}
@After
public void cleanup() throws Exception {
stopCollection();
}
@Override
protected Settings nodeSettings(int nodeOrdinal) {
return Settings.builder()
.put(super.nodeSettings(nodeOrdinal))
.put(Node.HTTP_ENABLED, true)
.put("shield.enabled", false)
.build();
webServer.shutdown();
}
@Test
public void testSimpleExport() throws Exception {
TargetNode target = TargetNode.start(internalCluster());
public void testExport() throws Exception {
enqueueGetClusterVersionResponse(Version.CURRENT);
enqueueResponse(404, "marvel template does not exist");
enqueueResponse(201, "marvel template created");
enqueueResponse(200, "successful bulk request ");
Settings.Builder builder = Settings.builder()
.put(MarvelSettings.INTERVAL, "-1")
.put("marvel.agent.exporters._http.type", "http")
.put("marvel.agent.exporters._http.host", target.httpAddress);
.put("marvel.agent.exporters._http.host", webServer.getHostName() + ":" + webServer.getPort())
.put("marvel.agent.exporters._http.connection.keep_alive", false);
String agentNode = internalCluster().startNode(builder);
ensureGreen();
HttpExporter exporter = getExporter(agentNode);
MarvelDoc doc = newRandomMarvelDoc();
exporter.export(Collections.singletonList(doc));
flush();
refresh();
assertThat(webServer.getRequestCount(), greaterThanOrEqualTo(4));
SearchResponse response = client().prepareSearch(".marvel-es-*").setTypes(doc.type()).get();
assertThat(response, notNullValue());
assertThat(response.getHits().totalHits(), is(1L));
}
RecordedRequest recordedRequest = webServer.takeRequest();
assertThat(recordedRequest.getMethod(), equalTo("GET"));
assertThat(recordedRequest.getPath(), equalTo("/"));
@Test
public void testTemplateAdditionDespiteOfLateClusterForming() throws Exception {
recordedRequest = webServer.takeRequest();
assertThat(recordedRequest.getMethod(), equalTo("GET"));
assertThat(recordedRequest.getPath(), equalTo("/_template/marvel"));
TargetNode target = TargetNode.start(internalCluster());
recordedRequest = webServer.takeRequest();
assertThat(recordedRequest.getMethod(), equalTo("PUT"));
assertThat(recordedRequest.getPath(), equalTo("/_template/marvel"));
assertThat(recordedRequest.getBody().readByteArray(), equalTo(MarvelTemplateUtils.loadDefaultTemplate()));
Settings.Builder builder = Settings.builder()
.put(MarvelSettings.STARTUP_DELAY, "200m")
.put(Node.HTTP_ENABLED, true)
.put("discovery.type", "zen")
.put("discovery.zen.ping_timeout", "1s")
.put("discovery.initial_state_timeout", "100ms")
.put("discovery.zen.minimum_master_nodes", 2)
.put("marvel.agent.exporters._http.type", "http")
.put("marvel.agent.exporters._http.host", target.httpAddress)
.put("marvel.agent.exporters._http." + HttpExporter.BULK_TIMEOUT_SETTING, "1s")
.put("marvel.agent.exporters._http." + HttpExporter.TEMPLATE_CHECK_TIMEOUT_SETTING, "1s");
String nodeName = internalCluster().startNode(builder);
HttpExporter exporter = getExporter(nodeName);
logger.info("exporting events while there is no cluster");
exporter.export(Collections.singletonList(newRandomMarvelDoc()));
logger.info("bringing up a second node");
internalCluster().startNode(builder);
ensureGreen();
logger.info("exporting a second event");
exporter.export(Collections.singletonList(newRandomMarvelDoc()));
logger.info("verifying that template has been created");
assertMarvelTemplateInstalled();
recordedRequest = webServer.takeRequest();
assertThat(recordedRequest.getMethod(), equalTo("POST"));
assertThat(recordedRequest.getPath(), equalTo("/_bulk"));
}
@Test
public void testDynamicHostChange() {
// disable exporting to be able to use non valid hosts
Settings.Builder builder = Settings.builder()
.put(MarvelSettings.INTERVAL, "-1")
@ -159,144 +139,222 @@ public class HttpExporterTests extends MarvelIntegTestCase {
@Test
public void testHostChangeReChecksTemplate() throws Exception {
TargetNode targetNode = TargetNode.start(internalCluster());
Settings.Builder builder = Settings.builder()
.put(MarvelSettings.STARTUP_DELAY, "200m")
.put(MarvelSettings.INTERVAL, "-1")
.put("marvel.agent.exporters._http.type", "http")
.put("marvel.agent.exporters._http.host", targetNode.httpAddress);
.put("marvel.agent.exporters._http.host", webServer.getHostName() + ":" + webServer.getPort())
.put("marvel.agent.exporters._http.connection.keep_alive", false);
logger.info("--> starting node");
enqueueGetClusterVersionResponse(Version.CURRENT);
enqueueResponse(404, "marvel template does not exist");
enqueueResponse(201, "marvel template created");
enqueueResponse(200, "successful bulk request ");
String agentNode = internalCluster().startNode(builder);
logger.info("--> exporting data");
HttpExporter exporter = getExporter(agentNode);
logger.info("exporting an event");
exporter.export(Collections.singletonList(newRandomMarvelDoc()));
logger.info("removing the marvel template");
assertAcked(client().admin().indices().prepareDeleteTemplate("marvel").get());
assertMarvelTemplateMissing();
assertThat(webServer.getRequestCount(), greaterThanOrEqualTo(4));
RecordedRequest recordedRequest = webServer.takeRequest();
assertThat(recordedRequest.getMethod(), equalTo("GET"));
assertThat(recordedRequest.getPath(), equalTo("/"));
recordedRequest = webServer.takeRequest();
assertThat(recordedRequest.getMethod(), equalTo("GET"));
assertThat(recordedRequest.getPath(), equalTo("/_template/marvel"));
recordedRequest = webServer.takeRequest();
assertThat(recordedRequest.getMethod(), equalTo("PUT"));
assertThat(recordedRequest.getPath(), equalTo("/_template/marvel"));
assertThat(recordedRequest.getBody().readByteArray(), equalTo(MarvelTemplateUtils.loadDefaultTemplate()));
recordedRequest = webServer.takeRequest();
assertThat(recordedRequest.getMethod(), equalTo("POST"));
assertThat(recordedRequest.getPath(), equalTo("/_bulk"));
logger.info("--> setting up another web server");
MockWebServer secondWebServer = null;
int secondWebPort;
try {
for (secondWebPort = 9250; secondWebPort < 9300; secondWebPort++) {
try {
secondWebServer = new MockWebServer();
QueueDispatcher dispatcher = new QueueDispatcher();
dispatcher.setFailFast(true);
secondWebServer.setDispatcher(dispatcher);
secondWebServer.start(secondWebPort);
break;
} catch (BindException be) {
logger.warn("port [{}] was already in use trying next port", secondWebPort);
}
}
assertNotNull("Unable to start the second mock web server", secondWebServer);
assertAcked(client().admin().cluster().prepareUpdateSettings().setTransientSettings(
Settings.builder().putArray("marvel.agent.exporters._http.host", exporter.hosts)).get());
Settings.builder().putArray("marvel.agent.exporters._http.host", secondWebServer.getHostName() + ":" + secondWebServer.getPort())).get());
// a new exporter is created on update, so we need to re-fetch it
exporter = getExporter(agentNode);
logger.info("exporting a second event");
enqueueGetClusterVersionResponse(secondWebServer, Version.CURRENT);
enqueueResponse(secondWebServer, 404, "marvel template does not exist");
enqueueResponse(secondWebServer, 201, "marvel template created");
enqueueResponse(secondWebServer, 200, "successful bulk request ");
logger.info("--> exporting a second event");
exporter.export(Collections.singletonList(newRandomMarvelDoc()));
logger.info("verifying that template has been created");
assertMarvelTemplateInstalled();
assertThat(secondWebServer.getRequestCount(), greaterThanOrEqualTo(4));
recordedRequest = secondWebServer.takeRequest();
assertThat(recordedRequest.getMethod(), equalTo("GET"));
assertThat(recordedRequest.getPath(), equalTo("/"));
recordedRequest = secondWebServer.takeRequest();
assertThat(recordedRequest.getMethod(), equalTo("GET"));
assertThat(recordedRequest.getPath(), equalTo("/_template/marvel"));
recordedRequest = secondWebServer.takeRequest();
assertThat(recordedRequest.getMethod(), equalTo("PUT"));
assertThat(recordedRequest.getPath(), equalTo("/_template/marvel"));
assertThat(recordedRequest.getBody().readByteArray(), equalTo(MarvelTemplateUtils.loadDefaultTemplate()));
recordedRequest = secondWebServer.takeRequest();
assertThat(recordedRequest.getMethod(), equalTo("POST"));
assertThat(recordedRequest.getPath(), equalTo("/_bulk"));
} finally {
if (secondWebServer != null) {
secondWebServer.shutdown();
}
@Test
public void testHostFailureChecksTemplate() throws Exception {
TargetNode target0 = TargetNode.start(internalCluster());
assertThat(target0.name, is(internalCluster().getMasterName()));
TargetNode target1 = TargetNode.start(internalCluster());
// lets start node0 & node1 first, such that node0 will be the master (it's first to start)
final String node0 = internalCluster().startNode(Settings.builder()
.put(MarvelSettings.STARTUP_DELAY, "200m")
.put("marvel.agent.exporters._http.type", "http")
.putArray("marvel.agent.exporters._http.host", target0.httpAddress, target1.httpAddress));
HttpExporter exporter = getExporter(node0);
logger.info("--> exporting events to have new settings take effect");
exporter.export(Collections.singletonList(newRandomMarvelDoc()));
logger.info("verifying that template has been created");
assertMarvelTemplateInstalled();
logger.info("--> removing the marvel template");
assertAcked(client().admin().indices().prepareDeleteTemplate("marvel").get());
assertMarvelTemplateMissing();
logger.info("--> shutting down target0");
assertThat(target0.name, is(internalCluster().getMasterName())); // just to be sure it's still the master
internalCluster().stopCurrentMasterNode();
// we use assert busy node because url caching may cause the node failure to be only detected while sending the event
assertBusy(new Runnable() {
@Override
public void run() {
try {
logger.info("--> exporting events from node0");
getExporter(node0).export(Collections.singletonList(newRandomMarvelDoc()));
} catch (Exception e) {
e.printStackTrace();
fail("failed to export event from node0");
}
logger.debug("--> checking for template");
assertMarvelTemplateInstalled();
logger.debug("--> template exists");
}
}, 30, TimeUnit.SECONDS);
}
@Test
public void testDynamicIndexFormatChange() throws Exception {
TargetNode targetNode = TargetNode.start(internalCluster());
Settings.Builder builder = Settings.builder()
.put(MarvelSettings.STARTUP_DELAY, "200m")
.put(MarvelSettings.INTERVAL, "-1")
.put("marvel.agent.exporters._http.type", "http")
.put("marvel.agent.exporters._http.host", targetNode.httpAddress);
.put("marvel.agent.exporters._http.host", webServer.getHostName() + ":" + webServer.getPort())
.put("marvel.agent.exporters._http.connection.keep_alive", false);
String agentNode = internalCluster().startNode(builder);
logger.info("exporting a first event");
logger.info("--> exporting a first event");
enqueueGetClusterVersionResponse(Version.CURRENT);
enqueueResponse(404, "marvel template does not exist");
enqueueResponse(201, "marvel template created");
enqueueResponse(200, "successful bulk request ");
HttpExporter exporter = getExporter(agentNode);
MarvelDoc doc = newRandomMarvelDoc();
exporter.export(Collections.singletonList(doc));
assertThat(webServer.getRequestCount(), greaterThanOrEqualTo(4));
RecordedRequest recordedRequest = webServer.takeRequest();
assertThat(recordedRequest.getMethod(), equalTo("GET"));
assertThat(recordedRequest.getPath(), equalTo("/"));
recordedRequest = webServer.takeRequest();
assertThat(recordedRequest.getMethod(), equalTo("GET"));
assertThat(recordedRequest.getPath(), equalTo("/_template/marvel"));
recordedRequest = webServer.takeRequest();
assertThat(recordedRequest.getMethod(), equalTo("PUT"));
assertThat(recordedRequest.getPath(), equalTo("/_template/marvel"));
assertThat(recordedRequest.getBody().readByteArray(), equalTo(MarvelTemplateUtils.loadDefaultTemplate()));
recordedRequest = webServer.takeRequest();
assertThat(recordedRequest.getMethod(), equalTo("POST"));
assertThat(recordedRequest.getPath(), equalTo("/_bulk"));
String indexName = exporter.indexNameResolver().resolve(doc);
logger.info("checks that the index [{}] is created", indexName);
assertTrue(client().admin().indices().prepareExists(indexName).get().isExists());
logger.info("--> checks that the document in the bulk request is indexed in [{}]", indexName);
byte[] bytes = recordedRequest.getBody().readByteArray();
Map<String, Object> data = XContentHelper.convertToMap(new BytesArray(bytes), false).v2();
Map<String, Object> index = (Map<String, Object>) data.get("index");
assertThat(index.get("_index"), equalTo(indexName));
String newTimeFormat = randomFrom("YY", "YYYY", "YYYY.MM", "YYYY-MM", "MM.YYYY", "MM");
logger.info("updating index time format setting to {}", newTimeFormat);
logger.info("--> updating index time format setting to {}", newTimeFormat);
assertAcked(client().admin().cluster().prepareUpdateSettings().setTransientSettings(Settings.builder()
.put("marvel.agent.exporters._http.index.name.time_format", newTimeFormat)));
exporter = getExporter(agentNode);
logger.info("exporting a second event");
logger.info("--> exporting a second event");
enqueueGetClusterVersionResponse(Version.CURRENT);
enqueueResponse(404, "marvel template does not exist");
enqueueResponse(201, "marvel template created");
enqueueResponse(200, "successful bulk request ");
doc = newRandomMarvelDoc();
exporter = getExporter(agentNode);
exporter.export(Collections.singletonList(doc));
String expectedMarvelIndex = MarvelSettings.MARVEL_INDICES_PREFIX
+ DateTimeFormat.forPattern(newTimeFormat).withZoneUTC().print(doc.timestamp());
logger.info("checks that the index [{}] is created", expectedMarvelIndex);
assertTrue(client().admin().indices().prepareExists(expectedMarvelIndex).get().isExists());
assertThat(webServer.getRequestCount(), greaterThanOrEqualTo(4));
logger.info("verifying that template has been created");
assertMarvelTemplateInstalled();
recordedRequest = webServer.takeRequest();
assertThat(recordedRequest.getMethod(), equalTo("GET"));
assertThat(recordedRequest.getPath(), equalTo("/"));
recordedRequest = webServer.takeRequest();
assertThat(recordedRequest.getMethod(), equalTo("GET"));
assertThat(recordedRequest.getPath(), equalTo("/_template/marvel"));
recordedRequest = webServer.takeRequest();
assertThat(recordedRequest.getMethod(), equalTo("PUT"));
assertThat(recordedRequest.getPath(), equalTo("/_template/marvel"));
assertThat(recordedRequest.getBody().readByteArray(), equalTo(MarvelTemplateUtils.loadDefaultTemplate()));
recordedRequest = webServer.takeRequest();
assertThat(recordedRequest.getMethod(), equalTo("POST"));
assertThat(recordedRequest.getPath(), equalTo("/_bulk"));
logger.info("--> checks that the document in the bulk request is indexed in [{}]", expectedMarvelIndex);
bytes = recordedRequest.getBody().readByteArray();
data = XContentHelper.convertToMap(new BytesArray(bytes), false).v2();
index = (Map<String, Object>) data.get("index");
assertThat(index.get("_index"), equalTo(expectedMarvelIndex));
}
@Test
public void testLoadRemoteClusterVersion() {
TargetNode targetNode = TargetNode.start(internalCluster());
public void testLoadRemoteClusterVersion() throws IOException {
final String host = webServer.getHostName() + ":" + webServer.getPort();
Settings.Builder builder = Settings.builder()
.put(MarvelSettings.STARTUP_DELAY, "200m")
.put(MarvelSettings.INTERVAL, "-1")
.put("marvel.agent.exporters._http.type", "http")
.put("marvel.agent.exporters._http.host", targetNode.httpAddress);
.put("marvel.agent.exporters._http.host", host)
.put("marvel.agent.exporters._http.connection.keep_alive", false);
String agentNode = internalCluster().startNode(builder);
HttpExporter exporter = getExporter(agentNode);
logger.info("--> loading remote cluster version");
Version resolved = exporter.loadRemoteClusterVersion(targetNode.httpAddress);
enqueueGetClusterVersionResponse(Version.CURRENT);
Version resolved = exporter.loadRemoteClusterVersion(host);
assertTrue(resolved.equals(Version.CURRENT));
final Version expected = randomFrom(Version.CURRENT, Version.V_0_18_0, Version.V_1_1_0, Version.V_1_2_5, Version.V_1_4_5, Version.V_1_6_0);
enqueueGetClusterVersionResponse(expected);
resolved = exporter.loadRemoteClusterVersion(host);
assertTrue(resolved.equals(expected));
}
private HttpExporter getExporter(String nodeName) {
@ -307,29 +365,26 @@ public class HttpExporterTests extends MarvelIntegTestCase {
private MarvelDoc newRandomMarvelDoc() {
if (randomBoolean()) {
return new IndexRecoveryMarvelDoc(internalCluster().getClusterName(),
IndexRecoveryCollector.TYPE, timeStampGenerator.incrementAndGet(), new RecoveryResponse());
IndexRecoveryCollector.TYPE, System.currentTimeMillis(), new RecoveryResponse());
} else {
return new ClusterStateMarvelDoc(internalCluster().getClusterName(),
ClusterStateCollector.TYPE, timeStampGenerator.incrementAndGet(), ClusterState.PROTO, ClusterHealthStatus.GREEN);
ClusterStateCollector.TYPE, System.currentTimeMillis(), ClusterState.PROTO, ClusterHealthStatus.GREEN);
}
}
static class TargetNode {
private final String name;
private final TransportAddress address;
private final String httpAddress;
private final Client client;
private TargetNode(InternalTestCluster cluster) {
name = cluster.startNode(Settings.builder().put(Node.HTTP_ENABLED, true));
address = cluster.getInstance(HttpServerTransport.class, name).boundAddress().publishAddress();
httpAddress = address.getHost() + ":" + address.getPort();
this.client = cluster.client(name);
private void enqueueGetClusterVersionResponse(Version v) throws IOException {
enqueueGetClusterVersionResponse(webServer, v);
}
static TargetNode start(InternalTestCluster cluster) {
return new TargetNode(cluster);
private void enqueueGetClusterVersionResponse(MockWebServer mockWebServer, Version v) throws IOException {
mockWebServer.enqueue(new MockResponse().setResponseCode(200).setBody(jsonBuilder().startObject().startObject("version").field("number", v.number()).endObject().endObject().bytes().toUtf8()));
}
private void enqueueResponse(int responseCode, String body) throws IOException {
enqueueResponse(webServer, responseCode, body);
}
private void enqueueResponse(MockWebServer mockWebServer, int responseCode, String body) throws IOException {
mockWebServer.enqueue(new MockResponse().setResponseCode(responseCode).setBody(body));
}
}

1
marvel/src/test/java/org/elasticsearch/marvel/agent/renderer/cluster/ClusterInfoIT.java
View File

@ -93,7 +93,6 @@ public class ClusterInfoIT extends MarvelIntegTestCase {
String recalculated = ClusterInfoRenderer.hash(status, uid, type, String.valueOf(expiryDate), clusterUUID);
assertThat(hkey, equalTo(recalculated));
assertThat((String) license.get(ClusterInfoRenderer.Fields.FEATURE.underscore().toString()), not(isEmptyOrNullString()));
assertThat((String) license.get(ClusterInfoRenderer.Fields.ISSUER.underscore().toString()), not(isEmptyOrNullString()));
assertThat((String) license.get(ClusterInfoRenderer.Fields.ISSUED_TO.underscore().toString()), not(isEmptyOrNullString()));
assertThat((Long) license.get(ClusterInfoRenderer.Fields.ISSUE_DATE_IN_MILLIS.underscore().toString()), greaterThan(0L));

44
marvel/src/test/java/org/elasticsearch/marvel/agent/renderer/shards/ShardsIT.java
View File

@ -6,27 +6,37 @@
package org.elasticsearch.marvel.agent.renderer.shards;
import org.apache.lucene.util.LuceneTestCase.AwaitsFix;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.marvel.agent.collector.shards.ShardsCollector;
import org.elasticsearch.marvel.agent.settings.MarvelSettings;
import org.elasticsearch.marvel.test.MarvelIntegTestCase;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.aggregations.Aggregation;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.bucket.terms.StringTerms;
import org.junit.Test;
import java.util.Map;
import static org.hamcrest.Matchers.greaterThan;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
import static org.hamcrest.Matchers.*;
@AwaitsFix(bugUrl="https://github.com/elastic/x-plugins/issues/729")
public class ShardsIT extends MarvelIntegTestCase {
private static final String INDEX_PREFIX = "test-shards-";
@Override
protected Settings nodeSettings(int nodeOrdinal) {
return Settings.builder()
.put(super.nodeSettings(nodeOrdinal))
.put(MarvelSettings.INTERVAL, "3s")
.put(MarvelSettings.COLLECTORS, ShardsCollector.NAME)
.put(MarvelSettings.INDICES, INDEX_PREFIX + "*")
.build();
}
@ -34,7 +44,7 @@ public class ShardsIT extends MarvelIntegTestCase {
public void testShards() throws Exception {
logger.debug("--> creating some indices so that shards collector reports data");
for (int i = 0; i < randomIntBetween(1, 5); i++) {
client().prepareIndex("test-" + i, "foo").setRefresh(true).setSource("field1", "value1").get();
client().prepareIndex(INDEX_PREFIX + i, "foo").setRefresh(true).setSource("field1", "value1").get();
}
awaitMarvelDocsCount(greaterThan(0L), ShardsCollector.TYPE);
@ -55,4 +65,34 @@ public class ShardsIT extends MarvelIntegTestCase {
logger.debug("--> shards successfully collected");
}
/**
* This test uses a terms aggregation to check that the "not_analyzed"
* fields of the "shards" document type are indeed not analyzed
*/
@Test
public void testNotAnalyzedFields() throws Exception {
final String indexName = INDEX_PREFIX + randomInt();
assertAcked(prepareCreate(indexName).setSettings(IndexMetaData.SETTING_NUMBER_OF_SHARDS, 1, IndexMetaData.SETTING_NUMBER_OF_REPLICAS, 0));
awaitMarvelDocsCount(greaterThan(0L), ShardsCollector.TYPE);
SearchRequestBuilder searchRequestBuilder = client()
.prepareSearch()
.setTypes(ShardsCollector.TYPE)
.setQuery(QueryBuilders.termQuery("shard.index", indexName));
String[] notAnalyzedFields = {"state_uuid", "shard.state", "shard.index", "shard.node"};
for (String field : notAnalyzedFields) {
searchRequestBuilder.addAggregation(AggregationBuilders.terms("agg_" + field.replace('.', '_')).field(field));
}
SearchResponse response = searchRequestBuilder.get();
assertThat(response.getHits().getTotalHits(), greaterThanOrEqualTo(1L));
for (Aggregation aggregation : response.getAggregations()) {
assertThat(aggregation, instanceOf(StringTerms.class));
assertThat(((StringTerms) aggregation).getBuckets().size(), equalTo(1));
}
}
}

29
marvel/src/test/java/org/elasticsearch/marvel/license/LicenseIntegrationTests.java
View File

@ -13,8 +13,9 @@ import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.inject.Module;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.license.core.License;
import org.elasticsearch.license.plugin.core.LicensesClientService;
import org.elasticsearch.license.plugin.core.LicensesService;
import org.elasticsearch.license.plugin.core.LicenseState;
import org.elasticsearch.license.plugin.core.Licensee;
import org.elasticsearch.license.plugin.core.LicenseeRegistry;
import org.elasticsearch.marvel.MarvelPlugin;
import org.elasticsearch.marvel.mode.Mode;
import org.elasticsearch.marvel.test.MarvelIntegTestCase;
@ -106,26 +107,24 @@ public class LicenseIntegrationTests extends MarvelIntegTestCase {
@Override
protected void configure() {
bind(MockLicenseService.class).asEagerSingleton();
bind(LicensesClientService.class).to(MockLicenseService.class);
bind(LicenseeRegistry.class).to(MockLicenseService.class);
}
}
public static class MockLicenseService extends AbstractComponent implements LicensesClientService {
public static class MockLicenseService extends AbstractComponent implements LicenseeRegistry {
static final License DUMMY_LICENSE = License.builder()
.feature(LicenseService.FEATURE_NAME)
.expiryDate(System.currentTimeMillis())
.issueDate(System.currentTimeMillis())
.issuedTo("LicensingTests")
.issuer("test")
.maxNodes(Integer.MAX_VALUE)
.signature("_signature")
.type("standard")
.subscriptionType("all_is_good")
.type("basic")
.uid(String.valueOf(RandomizedTest.systemPropertyAsInt(SysGlobals.CHILDVM_SYSPROP_JVM_ID, 0)) + System.identityHashCode(LicenseIntegrationTests.class))
.build();
private final List<Listener> listeners = new ArrayList<>();
private final List<Licensee> licensees = new ArrayList<>();
@Inject
public MockLicenseService(Settings settings) {
@ -134,22 +133,20 @@ public class LicenseIntegrationTests extends MarvelIntegTestCase {
}
@Override
public void register(String s, LicensesService.TrialLicenseOptions trialLicenseOptions, Collection<LicensesService.ExpirationCallback> collection, AcknowledgementCallback acknowledgementCallback, Listener listener) {
listeners.add(listener);
public void register(Licensee licensee) {
licensees.add(licensee);
enable();
}
public void enable() {
// enabled all listeners (incl. shield)
for (Listener listener : listeners) {
listener.onEnabled(DUMMY_LICENSE);
for (Licensee licensee : licensees) {
licensee.onChange(DUMMY_LICENSE, randomBoolean() ? LicenseState.GRACE_PERIOD : LicenseState.ENABLED);
}
}
public void disable() {
// only disable watcher listener (we need shield to work)
for (Listener listener : listeners) {
listener.onDisabled(DUMMY_LICENSE);
for (Licensee licensee : licensees) {
licensee.onChange(DUMMY_LICENSE, LicenseState.DISABLED);
}
}
}

1
marvel/src/test/resources/samples/cluster_state.json
View File

@ -4,6 +4,7 @@
"cluster_state": {
"status": "yellow",
"version": 14,
"state_uuid": "lj0hNoO9QaeNa1eR2ukktQ",
"master_node": "__node_id__",
"nodes": {
"__node_id__": {

2
qa/shield-core-rest-tests/pom.xml
View File

@ -23,7 +23,7 @@
<skip.unit.tests>true</skip.unit.tests>
<elasticsearch.integ.antfile>${project.basedir}/integration-tests.xml</elasticsearch.integ.antfile>
<tests.rest.load_packaged>true</tests.rest.load_packaged>
<tests.rest.blacklist>indices.get/10_basic/*allow_no_indices*,cat.count/10_basic/Test cat count output,cat.aliases/10_basic/Empty cluster,indices.segments/10_basic/no segments test,indices.clear_cache/10_basic/clear_cache test,indices.status/10_basic/Indices status test,cat.indices/10_basic/Test cat indices output,cat.recovery/10_basic/Test cat recovery output,cat.shards/10_basic/Test cat shards output,termvector/20_issue7121/*,index/10_with_id/Index with ID,indices.get_alias/20_emtpy/*,cat.segments/10_basic/Test cat segments output,indices.put_settings/10_basic/Test indices settings allow_no_indices,indices.put_settings/10_basic/Test indices settings ignore_unavailable,indices.refresh/10_basic/Indices refresh test no-match wildcard,indices.stats/10_index/Index - star*,indices.recovery/10_basic/Indices recovery test*,template/30_render_search_template/*,indices.shard_stores/10_basic/no indices test,cat.nodeattrs/10_basic/Test cat nodes attrs output</tests.rest.blacklist>
<tests.rest.blacklist>indices.get/10_basic/*allow_no_indices*,cat.count/10_basic/Test cat count output,cat.aliases/10_basic/Empty cluster,indices.segments/10_basic/no segments test,indices.clear_cache/10_basic/clear_cache test,indices.status/10_basic/Indices status test,cat.indices/10_basic/Test cat indices output,cat.recovery/10_basic/Test cat recovery output,cat.shards/10_basic/Test cat shards output,termvector/20_issue7121/*,index/10_with_id/Index with ID,indices.get_alias/20_emtpy/*,cat.segments/10_basic/Test cat segments output,indices.put_settings/10_basic/Test indices settings allow_no_indices,indices.put_settings/10_basic/Test indices settings ignore_unavailable,indices.refresh/10_basic/Indices refresh test no-match wildcard,indices.stats/10_index/Index - star*,indices.recovery/10_basic/Indices recovery test*,indices.shard_stores/10_basic/no indices test,cat.nodeattrs/10_basic/Test cat nodes attrs output</tests.rest.blacklist>
<xplugins.list>license,shield</xplugins.list>
</properties>

11
shield/docs/public/getting-started.asciidoc
View File

@ -33,14 +33,8 @@ NOTE: If you are using a <<deb-rpm-install, DEB/RPM distribution>> of Elasticsea
bin/elasticsearch
----------------------------------------------------------
. To verify that Shield is up and running, use the `_shield` API to get the Shield version:
+
[source,shell]
----------------------------------------------------------
curl -u es_admin -XGET 'http://localhost:9200/_shield'
----------------------------------------------------------
+
You can also check the startup log entries. When Shield is operating normally, the log indicates that the network transports are using Shield:
. To verify that Shield is up and running, check the startup log entries. When Shield is operating
normally, the log indicates that the network transports are using Shield:
+
[source,shell]
----------------
@ -49,7 +43,6 @@ You can also check the startup log entries. When Shield is operating normally, t
[2014-10-09 13:47:38,842][INFO ][http ] [Ezekiel Stane] Using [org.elasticsearch.shield.transport.netty.ShieldNettyHttpServerTransport] as http transport, overridden by [shield]
----------------
Now you're ready to secure your cluster! Here are a few things
you might want to do to start with:

4
shield/docs/public/managing-shield-licenses.asciidoc
View File

@ -22,7 +22,7 @@ To install or update the license use the following REST API:
[source,shell]
-----------------------------------------------------------------------
curl -XPUT -u admin 'http://<host>:<port>/_licenses' -d @license.json
curl -XPUT -u admin 'http://<host>:<port>/_license' -d @license.json
-----------------------------------------------------------------------
Where:
@ -42,7 +42,7 @@ You can list all currently installed licenses by executing the following REST AP
[source,shell]
-----------------------------------------------------
curl -XGET -u admin:password 'http://<host>:<port>/_licenses'
curl -XGET -u admin:password 'http://<host>:<port>/_license'
-----------------------------------------------------
The response of this command will be a JSON listing all available licenses. In the case of Shield, the following

2
shield/docs/public/reference.asciidoc
View File

@ -67,6 +67,7 @@ Elasticsearch. This only applies to publicly available indices and cluster actio
[float]
===== Cluster actions privileges
* `cluster:admin/render/template/search`
* `cluster:admin/repository/delete`
* `cluster:admin/repository/get`
* `cluster:admin/repository/put`
@ -118,7 +119,6 @@ NOTE: While indices template actions typically relate to indices, they are categ
* `indices:admin/open`
* `indices:admin/optimize`
* `indices:admin/refresh`
* `indices:admin/render/template/search`
* `indices:admin/settings/update`
* `indices:admin/shards/search_shards`
* `indices:admin/template/delete`

12
shield/docs/public/release-notes.asciidoc
View File

@ -42,6 +42,18 @@ version of Shield. We recommend copying the changes listed below to your `roles.
[[changelist]]
=== Change List
[float]
==== 2.0.0-rc1
.enhancements
* Added a caching interface that can be used by <<custom-realms, custom authentication realms>> to integrate with the <<cache-eviction-api, cache eviction api>>.
.bug fixes
* <<configuring-auditing, Auditing>> now captures requests from nodes using a different system key as tampered requests.
* The <<audit-index, index output for auditing>> stores the type of request when available.
* <<ip-filtering, IP filtering>> could have allowed a user to block all access to their node if the system was incorrectly configured, but now explicitly
allows connections from all addresses that the node is bound to so that connections coming from the node's host will not be blocked.
[float]
==== 2.0.0-beta2

10
shield/src/main/java/org/elasticsearch/shield/action/ShieldActionFilter.java
View File

@ -16,6 +16,7 @@ import org.elasticsearch.action.support.ActionFilterChain;
import org.elasticsearch.common.component.AbstractComponent;
import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.license.plugin.core.LicenseState;
import org.elasticsearch.license.plugin.core.LicenseUtils;
import org.elasticsearch.shield.User;
import org.elasticsearch.shield.action.interceptor.RequestInterceptor;
@ -62,13 +63,8 @@ public class ShieldActionFilter extends AbstractComponent implements ActionFilte
this.actionMapper = actionMapper;
licenseEventsNotifier.register(new LicenseEventsNotifier.Listener() {
@Override
public void enabled() {
licenseEnabled = true;
}
@Override
public void disabled() {
licenseEnabled = false;
public void notify(LicenseState state) {
licenseEnabled = state != LicenseState.DISABLED;
}
});
this.requestInterceptors = requestInterceptors;

16
shield/src/main/java/org/elasticsearch/shield/license/LicenseEventsNotifier.java
View File

@ -5,6 +5,8 @@
*/
package org.elasticsearch.shield.license;
import org.elasticsearch.license.plugin.core.LicenseState;
import java.util.HashSet;
import java.util.Set;
@ -26,22 +28,14 @@ public class LicenseEventsNotifier {
listeners.add(listener);
}
protected void notifyEnabled() {
protected void notify(LicenseState state) {
for (Listener listener : listeners) {
listener.enabled();
}
}
protected void notifyDisabled() {
for (Listener listener : listeners) {
listener.disabled();
listener.notify(state);
}
}
public static interface Listener {
void enabled();
void disabled();
void notify(LicenseState state);
}
}

150
shield/src/main/java/org/elasticsearch/shield/license/LicenseService.java
View File

@ -6,99 +6,95 @@
package org.elasticsearch.shield.license;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.common.Strings;
import org.elasticsearch.common.component.AbstractLifecycleComponent;
import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.joda.FormatDateTimeFormatter;
import org.elasticsearch.common.joda.Joda;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.license.core.License;
import org.elasticsearch.license.plugin.core.LicensesClientService;
import org.elasticsearch.license.plugin.core.LicenseState;
import org.elasticsearch.license.plugin.core.Licensee;
import org.elasticsearch.license.plugin.core.LicenseeRegistry;
import org.elasticsearch.shield.ShieldPlugin;
import java.util.*;
/**
*
*/
public class LicenseService extends AbstractLifecycleComponent<LicenseService> {
public class LicenseService extends AbstractLifecycleComponent<LicenseService> implements Licensee {
public static final String FEATURE_NAME = ShieldPlugin.NAME;
private static final LicensesClientService.TrialLicenseOptions TRIAL_LICENSE_OPTIONS =
new LicensesClientService.TrialLicenseOptions(TimeValue.timeValueHours(30 * 24), 1000);
private static final FormatDateTimeFormatter DATE_FORMATTER = Joda.forPattern("EEEE, MMMMM dd, yyyy", Locale.ROOT);
private final LicensesClientService licensesClientService;
private final LicenseeRegistry licenseeRegistry;
private final LicenseEventsNotifier notifier;
private final Collection<LicensesClientService.ExpirationCallback> expirationLoggers;
private final LicensesClientService.AcknowledgementCallback acknowledgementCallback;
private boolean enabled = false;
private volatile LicenseState state = LicenseState.DISABLED;
@Inject
public LicenseService(Settings settings, LicensesClientService licensesClientService, LicenseEventsNotifier notifier) {
public LicenseService(Settings settings, LicenseeRegistry licenseeRegistry, LicenseEventsNotifier notifier) {
super(settings);
this.licensesClientService = licensesClientService;
this.licenseeRegistry = licenseeRegistry;
this.notifier = notifier;
this.expirationLoggers = Arrays.asList(
new LicensesClientService.ExpirationCallback.Pre(days(7), days(30), days(1)) {
}
@Override
public void on(License license, LicensesClientService.ExpirationStatus status) {
logger.error("\n" +
"#\n" +
"# Shield license will expire on [{}]. Cluster health, cluster stats and indices stats operations are\n" +
"# blocked on Shield license expiration. All data operations (read and write) continue to work. If you\n" +
"# have a new license, please update it. Otherwise, please reach out to your support contact.\n" +
"#", DATE_FORMATTER.printer().print(license.expiryDate()));
public String id() {
return FEATURE_NAME;
}
},
new LicensesClientService.ExpirationCallback.Pre(days(0), days(7), minutes(10)) {
@Override
public void on(License license, LicensesClientService.ExpirationStatus status) {
logger.error("\n" +
"#\n" +
"# Shield license will expire on [{}]. Cluster health, cluster stats and indices stats operations are\n" +
"# blocked on Shield license expiration. All data operations (read and write) continue to work. If you\n" +
"# have a new license, please update it. Otherwise, please reach out to your support contact.\n" +
"#", DATE_FORMATTER.printer().print(license.expiryDate()));
}
},
new LicensesClientService.ExpirationCallback.Post(days(0), null, minutes(10)) {
@Override
public void on(License license, LicensesClientService.ExpirationStatus status) {
logger.error("\n" +
"#\n" +
"# SHIELD LICENSE EXPIRED ON [{}]! CLUSTER HEALTH, CLUSTER STATS AND INDICES STATS OPERATIONS ARE\n" +
"# NOW BLOCKED. ALL DATA OPERATIONS (READ AND WRITE) CONTINUE TO WORK. IF YOU HAVE A NEW LICENSE, PLEASE\n" +
"# UPDATE IT. OTHERWISE, PLEASE REACH OUT TO YOUR SUPPORT CONTACT.\n" +
"#", DATE_FORMATTER.printer().print(license.expiryDate()));
}
}
);
this.acknowledgementCallback = new LicensesClientService.AcknowledgementCallback() {
@Override
public List<String> acknowledge(License currentLicense, License newLicense) {
// TODO: add messages to be acknowledged when installing newLicense from currentLicense
// NOTE: currentLicense can be null, as a license registration can happen before
// a trial license could be generated
return Collections.emptyList();
}
public String[] expirationMessages() {
return new String[] {
"Cluster health, cluster stats and indices stats operations are blocked",
"All data operations (read and write) continue to work"
};
}
public synchronized boolean enabled() {
return enabled;
@Override
public String[] acknowledgmentMessages(License currentLicense, License newLicense) {
switch (newLicense.operationMode()) {
case BASIC:
if (currentLicense != null) {
switch (currentLicense.operationMode()) {
case TRIAL:
case GOLD:
case PLATINUM:
return new String[] { "The following Shield functionality will be disabled: authentication, authorization, ip filtering, auditing, SSL will be disabled on node restart. Please restart your node after applying the license." };
}
}
break;
case GOLD:
if (currentLicense != null) {
switch (currentLicense.operationMode()) {
case TRIAL:
case BASIC:
case PLATINUM:
return new String[] {
"Field and document level access control will be disabled"
};
}
}
break;
}
return Strings.EMPTY_ARRAY;
}
@Override
public void onChange(License license, LicenseState state) {
synchronized (this) {
this.state = state;
notifier.notify(state);
}
}
public LicenseState state() {
return state;
}
@Override
protected void doStart() throws ElasticsearchException {
if (settings.getGroups("tribe", true).isEmpty()) {
licensesClientService.register(FEATURE_NAME, TRIAL_LICENSE_OPTIONS, expirationLoggers, acknowledgementCallback, new InternalListener());
licenseeRegistry.register(this);
} else {
//TODO currently we disable licensing on tribe node. remove this once es core supports merging cluster
new InternalListener().onEnabled(null);
onChange(null, LicenseState.ENABLED);
}
}
@ -109,34 +105,4 @@ public class LicenseService extends AbstractLifecycleComponent<LicenseService> {
@Override
protected void doClose() throws ElasticsearchException {
}
static TimeValue days(int days) {
return TimeValue.timeValueHours(days * 24);
}
static TimeValue minutes(int minutes) {
return TimeValue.timeValueMinutes(minutes);
}
class InternalListener implements LicensesClientService.Listener {
@Override
public void onEnabled(License license) {
synchronized (LicenseService.this) {
logger.info("enabling license for [{}]", FEATURE_NAME);
enabled = true;
notifier.notifyEnabled();
}
}
@Override
public void onDisabled(License license) {
synchronized (LicenseService.this) {
logger.info("DISABLING LICENSE FOR [{}]", FEATURE_NAME);
enabled = false;
notifier.notifyDisabled();
}
}
}
}

3
shield/src/main/java/org/elasticsearch/shield/rest/action/RestShieldInfoAction.java
View File

@ -12,6 +12,7 @@ import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.inject.internal.Nullable;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.license.plugin.core.LicenseState;
import org.elasticsearch.rest.*;
import org.elasticsearch.shield.ShieldBuild;
import org.elasticsearch.shield.ShieldPlugin;
@ -71,7 +72,7 @@ public class RestShieldInfoAction extends BaseRestHandler {
private Status resolveStatus() {
if (shieldEnabled) {
if (licenseService.enabled()) {
if (licenseService.state() != LicenseState.DISABLED) {
return Status.ENABLED;
}
return Status.UNLICENSED;

2
shield/src/main/java/org/elasticsearch/shield/transport/filter/ShieldIpFilterRule.java
View File

@ -5,8 +5,8 @@
*/
package org.elasticsearch.shield.transport.filter;
import com.google.common.net.InetAddresses;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.common.network.InetAddresses;
import org.elasticsearch.common.network.NetworkAddress;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.transport.TransportAddress;

34
shield/src/test/java/org/elasticsearch/integration/LicensingTests.java
View File

@ -19,7 +19,9 @@ import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.inject.Module;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.license.core.License;
import org.elasticsearch.license.plugin.core.LicensesClientService;
import org.elasticsearch.license.plugin.core.LicenseState;
import org.elasticsearch.license.plugin.core.Licensee;
import org.elasticsearch.license.plugin.core.LicenseeRegistry;
import org.elasticsearch.plugins.Plugin;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.shield.license.LicenseService;
@ -169,13 +171,13 @@ public class LicensingTests extends ShieldIntegTestCase {
}
public static void disableLicensing() {
for (InternalLicensesClientService service : internalCluster().getInstances(InternalLicensesClientService.class)) {
for (InternalLicenseeRegistry service : internalCluster().getInstances(InternalLicenseeRegistry.class)) {
service.disable();
}
}
public static void enableLicensing() {
for (InternalLicensesClientService service : internalCluster().getInstances(InternalLicensesClientService.class)) {
for (InternalLicenseeRegistry service : internalCluster().getInstances(InternalLicenseeRegistry.class)) {
service.enable();
}
}
@ -203,49 +205,47 @@ public class LicensingTests extends ShieldIntegTestCase {
public static class InternalLicenseModule extends AbstractModule {
@Override
protected void configure() {
bind(InternalLicensesClientService.class).asEagerSingleton();
bind(LicensesClientService.class).to(InternalLicensesClientService.class);
bind(InternalLicenseeRegistry.class).asEagerSingleton();
bind(LicenseeRegistry.class).to(InternalLicenseeRegistry.class);
}
}
public static class InternalLicensesClientService extends AbstractComponent implements LicensesClientService {
public static class InternalLicenseeRegistry extends AbstractComponent implements LicenseeRegistry {
private final List<Listener> listeners = new ArrayList<>();
private final List<Licensee> licensees = new ArrayList<>();
static final License DUMMY_LICENSE = License.builder()
.feature(LicenseService.FEATURE_NAME)
.expiryDate(System.currentTimeMillis())
.issueDate(System.currentTimeMillis())
.issuedTo("LicensingTests")
.issuer("test")
.maxNodes(Integer.MAX_VALUE)
.signature("_signature")
.type("test_license_for_shield")
.subscriptionType("all_is_good")
.type("basic")
.uid(String.valueOf(randomLong()) + System.identityHashCode(LicensingTests.class))
.build();
@Inject
public InternalLicensesClientService(Settings settings) {
public InternalLicenseeRegistry(Settings settings) {
super(settings);
enable();
}
@Override
public void register(String s, LicensesClientService.TrialLicenseOptions trialLicenseOptions, Collection<LicensesClientService.ExpirationCallback> collection, AcknowledgementCallback acknowledgementCallback, Listener listener) {
listeners.add(listener);
public void register(Licensee licensee) {
licensees.add(licensee);
enable();
}
void enable() {
for (Listener listener : listeners) {
listener.onEnabled(DUMMY_LICENSE);
for (Licensee licensee : licensees) {
licensee.onChange(DUMMY_LICENSE, LicenseState.ENABLED);
}
}
void disable() {
for (Listener listener : listeners) {
listener.onDisabled(DUMMY_LICENSE);
for (Licensee licensee : licensees) {
licensee.onChange(DUMMY_LICENSE, LicenseState.DISABLED);
}
}
}

3
shield/src/test/java/org/elasticsearch/shield/action/ShieldActionFilterTests.java
View File

@ -11,6 +11,7 @@ import org.elasticsearch.action.ActionRequest;
import org.elasticsearch.action.search.SearchScrollRequest;
import org.elasticsearch.action.support.ActionFilterChain;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.license.plugin.core.LicenseState;
import org.elasticsearch.shield.User;
import org.elasticsearch.shield.action.interceptor.RequestInterceptor;
import org.elasticsearch.shield.audit.AuditTrail;
@ -112,7 +113,7 @@ public class ShieldActionFilterTests extends ESTestCase {
private class MockLicenseEventsNotifier extends LicenseEventsNotifier {
@Override
public void register(MockLicenseEventsNotifier.Listener listener) {
listener.enabled();
listener.notify(LicenseState.ENABLED);
}
}
}

7
shield/src/test/java/org/elasticsearch/shield/authz/accesscontrol/ShieldIndexSearcherWrapperUnitTests.java
View File

@ -43,8 +43,7 @@ import org.elasticsearch.index.mapper.MapperService;
import org.elasticsearch.index.mapper.internal.ParentFieldMapper;
import org.elasticsearch.index.shard.IndexShard;
import org.elasticsearch.index.shard.ShardId;
import org.elasticsearch.index.similarity.SimilarityLookupService;
import org.elasticsearch.indices.InternalIndicesLifecycle;
import org.elasticsearch.index.similarity.SimilarityService;
import org.elasticsearch.script.ScriptService;
import org.elasticsearch.search.aggregations.LeafBucketCollector;
import org.elasticsearch.shield.authz.InternalAuthorizationService;
@ -80,9 +79,9 @@ public class ShieldIndexSearcherWrapperUnitTests extends ESTestCase {
Index index = new Index("_index");
Settings settings = Settings.builder().put(IndexMetaData.SETTING_VERSION_CREATED, Version.CURRENT).build();
AnalysisService analysisService = new AnalysisService(index, settings);
SimilarityLookupService similarityLookupService = new SimilarityLookupService(index, settings);
SimilarityService similarityService = new SimilarityService(index, settings);
ScriptService scriptService = mock(ScriptService.class);
mapperService = new MapperService(index, settings, analysisService, similarityLookupService, scriptService);
mapperService = new MapperService(index, settings, analysisService, similarityService, scriptService);
shardId = new ShardId(index, 0);
shieldIndexSearcherWrapper = new ShieldIndexSearcherWrapper(settings, null, mapperService, null);

2
shield/src/test/java/org/elasticsearch/shield/transport/filter/IPFilterTests.java
View File

@ -5,8 +5,8 @@
*/
package org.elasticsearch.shield.transport.filter;
import com.google.common.net.InetAddresses;
import org.elasticsearch.common.component.Lifecycle;
import org.elasticsearch.common.network.InetAddresses;
import org.elasticsearch.common.network.NetworkAddress;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.BoundTransportAddress;

2
shield/src/test/java/org/elasticsearch/shield/transport/netty/IPFilterNettyUpstreamHandlerTests.java
View File

@ -5,8 +5,8 @@
*/
package org.elasticsearch.shield.transport.netty;
import com.google.common.net.InetAddresses;
import org.elasticsearch.common.component.Lifecycle;
import org.elasticsearch.common.network.InetAddresses;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.BoundTransportAddress;
import org.elasticsearch.common.transport.InetSocketTransportAddress;

2
shield/src/test/resources/org/elasticsearch/transport/actions
View File

@ -1,3 +1,4 @@
cluster:admin/render/template/search
cluster:admin/repository/delete
cluster:admin/repository/get
cluster:admin/repository/put
@ -33,7 +34,6 @@ indices:admin/mappings/get
indices:admin/open
indices:admin/optimize
indices:admin/refresh
indices:admin/render/template/search
indices:admin/settings/update
indices:admin/shards/search_shards
indices:admin/template/delete

2
shield/src/test/resources/org/elasticsearch/transport/handlers
View File

@ -1,3 +1,4 @@
cluster:admin/render/template/search
cluster:admin/snapshot/status[nodes]
cluster:admin/snapshot/status[nodes][n]
cluster:monitor/nodes/hot_threads[n]
@ -16,7 +17,6 @@ indices:admin/mappings/fields/get[index][s]
indices:admin/optimize[n]
indices:admin/refresh[s]
indices:admin/refresh[s][r]
indices:admin/render/template/search
indices:admin/upgrade
indices:admin/upgrade[n]
indices:admin/validate/query[s]

6
watcher/docs/release-notes.asciidoc
View File

@ -36,6 +36,12 @@ bin/plugin remove watcher
[[change-list]]
=== Change List
[float]
==== 2.0.0
.Bug fixes
* Fixed an issue where the scheduler may get stuck during Watcher startup. This caused no watches to ever fire.
[float]
==== 2.0.0-rc1

122
watcher/src/main/java/org/elasticsearch/watcher/license/LicenseService.java
View File

@ -6,93 +6,70 @@
package org.elasticsearch.watcher.license;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.common.Strings;
import org.elasticsearch.common.component.AbstractLifecycleComponent;
import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.joda.FormatDateTimeFormatter;
import org.elasticsearch.common.joda.Joda;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.license.core.License;
import org.elasticsearch.license.plugin.core.LicensesClientService;
import org.elasticsearch.license.plugin.core.LicensesService;
import org.elasticsearch.license.plugin.core.LicenseState;
import org.elasticsearch.license.plugin.core.Licensee;
import org.elasticsearch.license.plugin.core.LicenseeRegistry;
import org.elasticsearch.watcher.WatcherPlugin;
import java.util.*;
/**
*
*/
public class LicenseService extends AbstractLifecycleComponent<LicenseService> {
public class LicenseService extends AbstractLifecycleComponent<LicenseService> implements Licensee {
public static final String FEATURE_NAME = WatcherPlugin.NAME;
private static final LicensesClientService.TrialLicenseOptions TRIAL_LICENSE_OPTIONS =
new LicensesClientService.TrialLicenseOptions(TimeValue.timeValueHours(30 * 24), 1000);
private static final FormatDateTimeFormatter DATE_FORMATTER = Joda.forPattern("EEEE, MMMMM dd, yyyy", Locale.ROOT);
private final LicensesClientService clientService;
private final Collection<LicensesService.ExpirationCallback> expirationLoggers;
private final LicensesClientService.AcknowledgementCallback acknowledgementCallback;
private volatile boolean enabled;
private final LicenseeRegistry clientService;
private volatile LicenseState state;
@Inject
public LicenseService(Settings settings, LicensesClientService clientService) {
public LicenseService(Settings settings, LicenseeRegistry clientService) {
super(settings);
this.clientService = clientService;
this.expirationLoggers = Arrays.asList(
new LicensesService.ExpirationCallback.Pre(days(7), days(30), days(1)) {
}
@Override
public void on(License license, LicensesService.ExpirationStatus status) {
logger.error("\n" +
"#\n" +
"# Watcher license will expire on [{}]. All configured actions on\n" +
"# all registered watches are throttled (not executed) on Watcher license expiration. \n" +
"# Watches will continue be evaluated and watch history will continue being recorded.\n" +
"# Have a new license? please update it. Otherwise, please reach out to your support contact.\n" +
"#", DATE_FORMATTER.printer().print(license.expiryDate()));
public String id() {
return FEATURE_NAME;
}
},
new LicensesService.ExpirationCallback.Pre(days(0), days(7), minutes(10)) {
@Override
public void on(License license, LicensesService.ExpirationStatus status) {
logger.error("\n" +
"#\n" +
"# Watcher license will expire on [{}]. All configured actions on\n" +
"# all registered watches are throttled (not executed) on Watcher license expiration. \n" +
"# Watches will continue be evaluated and watch history will continue being recorded.\n" +
"# Have a new license? please update it. Otherwise, please reach out to your support contact.\n" +
"#", DATE_FORMATTER.printer().print(license.expiryDate()));
public String[] expirationMessages() {
// TODO add messages to be logged around license expiry
return new String[0];
}
},
new LicensesService.ExpirationCallback.Post(days(0), null, minutes(10)) {
@Override
public void on(License license, LicensesService.ExpirationStatus status) {
logger.error("\n" +
"#\n" +
"# WATCHER LICENSE WAS EXPIRED ON [{}]. ALL CONFIGURED ACTIONS ON\n" +
"# ALL REGISTERED WATCHES ARE THROTTLED (NOT EXECUTED) ON WATCHER LICENSE EXPIRATION. \n" +
"# WATCHES WILL CONTINUE BE EVALUATED AND WATCH HISTORY WILL CONTINUE BEING RECORDED.\n" +
"# HAVE A NEW LICENSE? PLEASE UPDATE IT. OTHERWISE, PLEASE REACH OUT TO YOUR SUPPORT CONTACT.\n" +
"#", DATE_FORMATTER.printer().print(license.expiryDate()));
public String[] acknowledgmentMessages(License currentLicense, License newLicense) {
switch (newLicense.operationMode()) {
case BASIC:
if (currentLicense != null) {
switch (currentLicense.operationMode()) {
case TRIAL:
case GOLD:
case PLATINUM:
return new String[] { "Watcher will be disabled" };
}
}
);
this.acknowledgementCallback = new LicensesClientService.AcknowledgementCallback() {
break;
}
return Strings.EMPTY_ARRAY;
}
@Override
public List<String> acknowledge(License currentLicense, License newLicense) {
// TODO: add messages to be acknowledged when installing newLicense from currentLicense
// NOTE: currentLicense can be null, as a license registration can happen before
// a trial license could be generated
return Collections.emptyList();
public void onChange(License license, LicenseState state) {
synchronized (this) {
this.state = state;
}
};
}
@Override
protected void doStart() throws ElasticsearchException {
clientService.register(FEATURE_NAME, TRIAL_LICENSE_OPTIONS, expirationLoggers, acknowledgementCallback, new InternalListener(this));
clientService.register(this);
}
@Override
@ -104,33 +81,6 @@ public class LicenseService extends AbstractLifecycleComponent<LicenseService> {
}
public boolean enabled() {
return enabled;
}
static TimeValue days(int days) {
return TimeValue.timeValueHours(days * 24);
}
static TimeValue minutes(int minutes) {
return TimeValue.timeValueMinutes(minutes);
}
class InternalListener implements LicensesClientService.Listener {
private final LicenseService service;
public InternalListener(LicenseService service) {
this.service = service;
}
@Override
public void onEnabled(License license) {
service.enabled = true;
}
@Override
public void onDisabled(License license) {
service.enabled = false;
}
return state != LicenseState.DISABLED;
}
}

4
watcher/src/main/java/org/elasticsearch/watcher/trigger/schedule/engine/TickerScheduleTriggerEngine.java
View File

@ -136,10 +136,6 @@ public class TickerScheduleTriggerEngine extends ScheduleTriggerEngine {
@Override
public void run() {
// calibrate with round clock
while (clock.millis() % 1000 > 15) {
}
while (active) {
logger.trace("checking jobs [{}]", clock.nowUTC());
checkJobs();

29
watcher/src/test/java/org/elasticsearch/watcher/license/LicenseIntegrationTests.java
View File

@ -14,8 +14,9 @@ import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.inject.Module;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.license.core.License;
import org.elasticsearch.license.plugin.core.LicensesClientService;
import org.elasticsearch.license.plugin.core.LicensesService;
import org.elasticsearch.license.plugin.core.LicenseState;
import org.elasticsearch.license.plugin.core.Licensee;
import org.elasticsearch.license.plugin.core.LicenseeRegistry;
import org.elasticsearch.plugins.Plugin;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.watcher.actions.ActionStatus;
@ -46,15 +47,13 @@ import static org.hamcrest.Matchers.*;
public class LicenseIntegrationTests extends AbstractWatcherIntegrationTestCase {
static final License DUMMY_LICENSE = License.builder()
.feature(LicenseService.FEATURE_NAME)
.expiryDate(System.currentTimeMillis())
.issueDate(System.currentTimeMillis())
.issuedTo("LicensingTests")
.issuer("test")
.maxNodes(Integer.MAX_VALUE)
.signature("_signature")
.type("test_license_for_watcher")
.subscriptionType("all_is_good")
.type("basic")
.uid(String.valueOf(RandomizedTest.systemPropertyAsInt(SysGlobals.CHILDVM_SYSPROP_JVM_ID, 0)) + System.identityHashCode(LicenseIntegrationTests.class))
.build();
@ -301,13 +300,13 @@ public class LicenseIntegrationTests extends AbstractWatcherIntegrationTestCase
@Override
protected void configure() {
bind(MockLicenseService.class).asEagerSingleton();
bind(LicensesClientService.class).to(MockLicenseService.class);
bind(LicenseeRegistry.class).to(MockLicenseService.class);
}
}
public static class MockLicenseService extends AbstractComponent implements LicensesClientService {
public static class MockLicenseService extends AbstractComponent implements LicenseeRegistry {
private final List<Listener> listeners = new ArrayList<>();
private final List<Licensee> licensees = new ArrayList<>();
@Inject
public MockLicenseService(Settings settings) {
@ -316,23 +315,23 @@ public class LicenseIntegrationTests extends AbstractWatcherIntegrationTestCase
}
@Override
public void register(String s, LicensesService.TrialLicenseOptions trialLicenseOptions, Collection<LicensesService.ExpirationCallback> collection, AcknowledgementCallback acknowledgementCallback, Listener listener) {
listeners.add(listener);
public void register(Licensee licensee) {
licensees.add(licensee);
enable();
}
public void enable() {
// enabled all listeners (incl. shield)
for (Listener listener : listeners) {
listener.onEnabled(DUMMY_LICENSE);
for (Licensee licensee : licensees) {
licensee.onChange(DUMMY_LICENSE, LicenseState.ENABLED);
}
}
public void disable() {
// only disable watcher listener (we need shield to work)
for (Listener listener : listeners) {
if (listener instanceof LicenseService.InternalListener) {
listener.onDisabled(DUMMY_LICENSE);
for (Licensee licensee : licensees) {
if (licensee instanceof LicenseService) {
licensee.onChange(DUMMY_LICENSE, LicenseState.DISABLED);
}
}
}