simplify license service to use schedule engine for license notification
Original commit: elastic/x-pack-elasticsearch@81ccb56406
This commit is contained in:
Areek Zillur
parent
91c03c4423
commit
d350073f4b
|
|
@ -8,9 +8,14 @@ package org.elasticsearch.license.plugin.core;
|
|||
import org.elasticsearch.common.logging.LoggerMessageFormat;
|
||||
import org.elasticsearch.common.unit.TimeValue;
|
||||
import org.elasticsearch.license.core.License;
|
||||
import org.elasticsearch.xpack.scheduler.SchedulerEngine;
|
||||
|
||||
import java.util.UUID;
|
||||
|
||||
public abstract class ExpirationCallback {
|
||||
|
||||
final static String EXPIRATION_JOB_PREFIX = ".license_expiration_job_";
|
||||
|
||||
public enum Orientation {PRE, POST}
|
||||
|
||||
public static abstract class Pre extends ExpirationCallback {
|
||||
|
|
@ -38,8 +43,8 @@ public abstract class ExpirationCallback {
|
|||
}
|
||||
|
||||
@Override
|
||||
public TimeValue delay(long expiryDuration) {
|
||||
return TimeValue.timeValueMillis(expiryDuration - max.getMillis());
|
||||
public TimeValue delay(long expirationDate, long now) {
|
||||
return TimeValue.timeValueMillis((expirationDate - now) - max.getMillis());
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -68,7 +73,8 @@ public abstract class ExpirationCallback {
|
|||
}
|
||||
|
||||
@Override
|
||||
public TimeValue delay(long expiryDuration) {
|
||||
public TimeValue delay(long expirationDate, long now) {
|
||||
long expiryDuration = expirationDate - now;
|
||||
final long delay;
|
||||
if (expiryDuration >= 0L) {
|
||||
delay = expiryDuration + min.getMillis();
|
||||
|
|
@ -83,6 +89,7 @@ public abstract class ExpirationCallback {
|
|||
}
|
||||
}
|
||||
|
||||
private final String id;
|
||||
protected final Orientation orientation;
|
||||
protected final TimeValue min;
|
||||
protected final TimeValue max;
|
||||
|
|
@ -93,21 +100,57 @@ public abstract class ExpirationCallback {
|
|||
this.min = (min == null) ? TimeValue.timeValueMillis(0) : min;
|
||||
this.max = (max == null) ? TimeValue.timeValueMillis(Long.MAX_VALUE) : max;
|
||||
this.frequency = frequency;
|
||||
this.id = String.join("", EXPIRATION_JOB_PREFIX, UUID.randomUUID().toString());
|
||||
}
|
||||
|
||||
public String getId() {
|
||||
return id;
|
||||
}
|
||||
|
||||
public TimeValue frequency() {
|
||||
return frequency;
|
||||
}
|
||||
|
||||
public abstract TimeValue delay(long expiryDuration);
|
||||
public abstract TimeValue delay(long expirationDate, long now);
|
||||
|
||||
public abstract boolean matches(long expirationDate, long now);
|
||||
|
||||
public abstract void on(License license);
|
||||
|
||||
@Override
|
||||
public SchedulerEngine.Schedule schedule(long expiryDate) {
|
||||
return new ExpirySchedule(expiryDate);
|
||||
}
|
||||
|
||||
public String toString() {
|
||||
return LoggerMessageFormat.format(null, "ExpirationCallback:(orientation [{}], min [{}], max [{}], freq [{}])",
|
||||
orientation.name(), min, max, frequency);
|
||||
}
|
||||
}
|
||||
|
||||
private class ExpirySchedule implements SchedulerEngine.Schedule {
|
||||
|
||||
private final long expiryDate;
|
||||
|
||||
private ExpirySchedule(long expiryDate) {
|
||||
this.expiryDate = expiryDate;
|
||||
}
|
||||
|
||||
@Override
|
||||
public long nextScheduledTimeAfter(long startTime, long time) {
|
||||
if (matches(expiryDate, time)) {
|
||||
if (startTime == time) {
|
||||
return time;
|
||||
} else {
|
||||
return time + frequency().getMillis();
|
||||
}
|
||||
} else {
|
||||
if (startTime == time) {
|
||||
final TimeValue delay = delay(expiryDate, time);
|
||||
if (delay != null) {
|
||||
return time + delay.getMillis();
|
||||
}
|
||||
}
|
||||
return -1;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,40 @@
|
|||
/*
|
||||
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
|
||||
* or more contributor license agreements. Licensed under the Elastic License;
|
||||
* you may not use this file except in compliance with the Elastic License.
|
||||
*/
|
||||
package org.elasticsearch.license.plugin.core;
|
||||
|
||||
import org.elasticsearch.license.core.License;
|
||||
import org.elasticsearch.xpack.scheduler.SchedulerEngine;
|
||||
|
||||
import static org.elasticsearch.license.plugin.core.LicensesService.GRACE_PERIOD_DURATION;
|
||||
import static org.elasticsearch.license.plugin.core.LicensesService.getLicenseState;
|
||||
|
||||
public class LicenseSchedule implements SchedulerEngine.Schedule {
|
||||
|
||||
private final License license;
|
||||
|
||||
LicenseSchedule(License license) {
|
||||
this.license = license;
|
||||
}
|
||||
|
||||
@Override
|
||||
public long nextScheduledTimeAfter(long startTime, long time) {
|
||||
long nextScheduledTime = -1;
|
||||
switch (getLicenseState(license, time)) {
|
||||
case ENABLED:
|
||||
nextScheduledTime = license.expiryDate();
|
||||
break;
|
||||
case GRACE_PERIOD:
|
||||
nextScheduledTime = license.expiryDate() + GRACE_PERIOD_DURATION.getMillis();
|
||||
break;
|
||||
case DISABLED:
|
||||
if (license.issueDate() > time) {
|
||||
nextScheduledTime = license.issueDate();
|
||||
}
|
||||
break;
|
||||
}
|
||||
return nextScheduledTime;
|
||||
}
|
||||
}
|
||||
|
|
@ -27,8 +27,6 @@ import org.elasticsearch.common.joda.Joda;
|
|||
import org.elasticsearch.common.logging.LoggerMessageFormat;
|
||||
import org.elasticsearch.common.settings.Settings;
|
||||
import org.elasticsearch.common.unit.TimeValue;
|
||||
import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException;
|
||||
import org.elasticsearch.common.util.concurrent.FutureUtils;
|
||||
import org.elasticsearch.gateway.GatewayService;
|
||||
import org.elasticsearch.license.core.License;
|
||||
import org.elasticsearch.license.core.LicenseVerifier;
|
||||
|
|
@ -42,21 +40,20 @@ import org.elasticsearch.transport.TransportRequest;
|
|||
import org.elasticsearch.transport.TransportRequestHandler;
|
||||
import org.elasticsearch.transport.TransportResponse;
|
||||
import org.elasticsearch.transport.TransportService;
|
||||
import org.elasticsearch.xpack.scheduler.SchedulerEngine;
|
||||
import org.elasticsearch.xpack.support.clock.Clock;
|
||||
|
||||
import java.io.ByteArrayOutputStream;
|
||||
import java.io.IOException;
|
||||
import java.io.InputStream;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collections;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Locale;
|
||||
import java.util.Map;
|
||||
import java.util.Queue;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.ConcurrentLinkedQueue;
|
||||
import java.util.concurrent.CopyOnWriteArrayList;
|
||||
import java.util.concurrent.ScheduledFuture;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
|
||||
/**
|
||||
|
|
@ -77,18 +74,16 @@ import java.util.concurrent.atomic.AtomicReference;
|
|||
* <p>
|
||||
* All registered listeners are notified of the current license upon registration or when a new license is installed in the cluster state.
|
||||
* When a new license is notified as enabled to the registered listener, a notification is scheduled at the time of license expiry.
|
||||
* Registered listeners are notified using {@link #notifyAndSchedule(LicensesMetaData)}
|
||||
* Registered listeners are notified using {@link #onUpdate(LicensesMetaData)}
|
||||
*/
|
||||
@Singleton
|
||||
public class LicensesService extends AbstractLifecycleComponent<LicensesService> implements ClusterStateListener, LicensesManagerService,
|
||||
LicenseeRegistry {
|
||||
LicenseeRegistry, SchedulerEngine.Listener {
|
||||
|
||||
public static final String REGISTER_TRIAL_LICENSE_ACTION_NAME = "internal:plugin/license/cluster/register_trial_license";
|
||||
|
||||
private final ClusterService clusterService;
|
||||
|
||||
private final ThreadPool threadPool;
|
||||
|
||||
private final TransportService transportService;
|
||||
|
||||
/**
|
||||
|
|
@ -96,21 +91,11 @@ public class LicensesService extends AbstractLifecycleComponent<LicensesService>
|
|||
*/
|
||||
private final List<InternalLicensee> registeredLicensees = new CopyOnWriteArrayList<>();
|
||||
|
||||
/**
|
||||
* Currently active expiry notifications
|
||||
*/
|
||||
private final Queue<ScheduledFuture> expiryNotifications = new ConcurrentLinkedQueue<>();
|
||||
|
||||
/**
|
||||
* Currently active event notifications for every registered listener
|
||||
*/
|
||||
private final Queue<ScheduledFuture> eventNotifications = new ConcurrentLinkedQueue<>();
|
||||
|
||||
/**
|
||||
* Currently active license
|
||||
*/
|
||||
private final AtomicReference<License> currentLicense = new AtomicReference<>();
|
||||
|
||||
private SchedulerEngine scheduler;
|
||||
private final Clock clock;
|
||||
|
||||
/**
|
||||
|
|
@ -131,7 +116,9 @@ public class LicensesService extends AbstractLifecycleComponent<LicensesService>
|
|||
/**
|
||||
* Duration of grace period after a license has expired
|
||||
*/
|
||||
private TimeValue gracePeriodDuration = days(7);
|
||||
public static final TimeValue GRACE_PERIOD_DURATION = days(7);
|
||||
|
||||
private final static String LICENSE_JOB = "licenseJob";
|
||||
|
||||
private static final FormatDateTimeFormatter DATE_FORMATTER = Joda.forPattern("EEEE, MMMMM dd, yyyy", Locale.ROOT);
|
||||
|
||||
|
|
@ -139,10 +126,9 @@ public class LicensesService extends AbstractLifecycleComponent<LicensesService>
|
|||
"please read the following messages and update the license again, this time with the \"acknowledge=true\" parameter:";
|
||||
|
||||
@Inject
|
||||
public LicensesService(Settings settings, ClusterService clusterService, ThreadPool threadPool, TransportService transportService, Clock clock) {
|
||||
public LicensesService(Settings settings, ClusterService clusterService, TransportService transportService, Clock clock) {
|
||||
super(settings);
|
||||
this.clusterService = clusterService;
|
||||
this.threadPool = threadPool;
|
||||
this.transportService = transportService;
|
||||
if (DiscoveryNode.isMasterNode(settings)) {
|
||||
transportService.registerRequestHandler(REGISTER_TRIAL_LICENSE_ACTION_NAME, TransportRequest.Empty::new,
|
||||
|
|
@ -152,6 +138,10 @@ public class LicensesService extends AbstractLifecycleComponent<LicensesService>
|
|||
this.clock = clock;
|
||||
}
|
||||
|
||||
protected SchedulerEngine getScheduler() {
|
||||
return new SchedulerEngine(clock);
|
||||
}
|
||||
|
||||
private void populateExpirationCallbacks() {
|
||||
expirationCallbacks.add(new ExpirationCallback.Pre(days(7), days(25), days(1)) {
|
||||
@Override
|
||||
|
|
@ -264,9 +254,8 @@ public class LicensesService extends AbstractLifecycleComponent<LicensesService>
|
|||
listener.onResponse(new PutLicenseResponse(true, LicensesStatus.EXPIRED));
|
||||
} else {
|
||||
if (!request.acknowledged()) {
|
||||
final LicensesMetaData currentMetaData = clusterService.state().metaData().custom(LicensesMetaData.TYPE);
|
||||
final License currentLicense = getLicense(currentMetaData);
|
||||
if (currentLicense != null && currentLicense != LicensesMetaData.LICENSE_TOMBSTONE) {
|
||||
final License currentLicense = getLicense();
|
||||
if (currentLicense != null) {
|
||||
Map<String, String[]> acknowledgeMessages = new HashMap<>(registeredLicensees.size() + 1);
|
||||
if (!License.isAutoGeneratedLicense(currentLicense.signature()) // current license is not auto-generated
|
||||
&& currentLicense.issueDate() > newLicense.issueDate()) { // and has a later issue date
|
||||
|
|
@ -290,22 +279,22 @@ public class LicensesService extends AbstractLifecycleComponent<LicensesService>
|
|||
}
|
||||
clusterService.submitStateUpdateTask("register license [" + newLicense.uid() + "]", new
|
||||
AckedClusterStateUpdateTask<PutLicenseResponse>(request, listener) {
|
||||
@Override
|
||||
protected PutLicenseResponse newResponse(boolean acknowledged) {
|
||||
return new PutLicenseResponse(acknowledged, LicensesStatus.VALID);
|
||||
}
|
||||
@Override
|
||||
protected PutLicenseResponse newResponse(boolean acknowledged) {
|
||||
return new PutLicenseResponse(acknowledged, LicensesStatus.VALID);
|
||||
}
|
||||
|
||||
@Override
|
||||
public ClusterState execute(ClusterState currentState) throws Exception {
|
||||
MetaData.Builder mdBuilder = MetaData.builder(currentState.metaData());
|
||||
mdBuilder.putCustom(LicensesMetaData.TYPE, new LicensesMetaData(newLicense));
|
||||
return ClusterState.builder(currentState).metaData(mdBuilder).build();
|
||||
}
|
||||
});
|
||||
@Override
|
||||
public ClusterState execute(ClusterState currentState) throws Exception {
|
||||
MetaData.Builder mdBuilder = MetaData.builder(currentState.metaData());
|
||||
mdBuilder.putCustom(LicensesMetaData.TYPE, new LicensesMetaData(newLicense));
|
||||
return ClusterState.builder(currentState).metaData(mdBuilder).build();
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
private boolean verifyLicense(final License license) {
|
||||
static boolean verifyLicense(final License license) {
|
||||
final byte[] publicKeyBytes;
|
||||
try (InputStream is = LicensesService.class.getResourceAsStream("/public.key")) {
|
||||
ByteArrayOutputStream out = new ByteArrayOutputStream();
|
||||
|
|
@ -317,42 +306,59 @@ public class LicensesService extends AbstractLifecycleComponent<LicensesService>
|
|||
return LicenseVerifier.verifyLicense(license, publicKeyBytes);
|
||||
}
|
||||
|
||||
|
||||
static TimeValue days(int days) {
|
||||
return TimeValue.timeValueHours(days * 24);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void triggered(SchedulerEngine.Event event) {
|
||||
final LicensesMetaData licensesMetaData = clusterService.state().metaData().custom(LicensesMetaData.TYPE);
|
||||
if (licensesMetaData != null) {
|
||||
final License license = licensesMetaData.getLicense();
|
||||
if (event.getJobName().equals(LICENSE_JOB)) {
|
||||
notifyLicensees(license);
|
||||
} else if (event.getJobName().startsWith(ExpirationCallback.EXPIRATION_JOB_PREFIX)) {
|
||||
expirationCallbacks.stream()
|
||||
.filter(expirationCallback -> expirationCallback.getId().equals(event.getJobName()))
|
||||
.forEach(expirationCallback -> expirationCallback.on(license));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Remove license from the cluster state metadata
|
||||
*/
|
||||
public void removeLicense(final DeleteLicenseRequest request, final ActionListener<ClusterStateUpdateResponse> listener) {
|
||||
clusterService.submitStateUpdateTask("delete license",
|
||||
new AckedClusterStateUpdateTask<ClusterStateUpdateResponse>(request, listener) {
|
||||
@Override
|
||||
protected ClusterStateUpdateResponse newResponse(boolean acknowledged) {
|
||||
return new ClusterStateUpdateResponse(acknowledged);
|
||||
}
|
||||
@Override
|
||||
protected ClusterStateUpdateResponse newResponse(boolean acknowledged) {
|
||||
return new ClusterStateUpdateResponse(acknowledged);
|
||||
}
|
||||
|
||||
@Override
|
||||
public ClusterState execute(ClusterState currentState) throws Exception {
|
||||
MetaData metaData = currentState.metaData();
|
||||
final LicensesMetaData currentLicenses = metaData.custom(LicensesMetaData.TYPE);
|
||||
if (currentLicenses.getLicense() != LicensesMetaData.LICENSE_TOMBSTONE) {
|
||||
MetaData.Builder mdBuilder = MetaData.builder(currentState.metaData());
|
||||
mdBuilder.putCustom(LicensesMetaData.TYPE, new LicensesMetaData(LicensesMetaData.LICENSE_TOMBSTONE));
|
||||
return ClusterState.builder(currentState).metaData(mdBuilder).build();
|
||||
} else {
|
||||
return currentState;
|
||||
}
|
||||
}
|
||||
});
|
||||
@Override
|
||||
public ClusterState execute(ClusterState currentState) throws Exception {
|
||||
MetaData metaData = currentState.metaData();
|
||||
final LicensesMetaData currentLicenses = metaData.custom(LicensesMetaData.TYPE);
|
||||
if (currentLicenses.getLicense() != LicensesMetaData.LICENSE_TOMBSTONE) {
|
||||
MetaData.Builder mdBuilder = MetaData.builder(currentState.metaData());
|
||||
mdBuilder.putCustom(LicensesMetaData.TYPE, new LicensesMetaData(LicensesMetaData.LICENSE_TOMBSTONE));
|
||||
return ClusterState.builder(currentState).metaData(mdBuilder).build();
|
||||
} else {
|
||||
return currentState;
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public LicenseState licenseState() {
|
||||
if (registeredLicensees.size() > 0) {
|
||||
return registeredLicensees.get(0).currentLicenseState;
|
||||
} else {
|
||||
return getLicenseState(getLicense(), clock.millis());
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
@ -407,27 +413,25 @@ public class LicensesService extends AbstractLifecycleComponent<LicensesService>
|
|||
|
||||
@Override
|
||||
protected void doStart() throws ElasticsearchException {
|
||||
if (scheduler == null) {
|
||||
this.scheduler = getScheduler();
|
||||
this.scheduler.register(this);
|
||||
}
|
||||
clusterService.add(this);
|
||||
try {
|
||||
scheduler.start(Collections.emptyList());
|
||||
} catch (Exception e) {
|
||||
logger.warn("failed to start license trigger service", e);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void doStop() throws ElasticsearchException {
|
||||
clusterService.remove(this);
|
||||
|
||||
// cancel all notifications
|
||||
for (ScheduledFuture scheduledNotification : expiryNotifications) {
|
||||
FutureUtils.cancel(scheduledNotification);
|
||||
}
|
||||
for (ScheduledFuture eventNotification : eventNotifications) {
|
||||
FutureUtils.cancel(eventNotification);
|
||||
}
|
||||
|
||||
scheduler.stop();
|
||||
// clear all handlers
|
||||
registeredLicensees.clear();
|
||||
|
||||
// empty out notification queue
|
||||
expiryNotifications.clear();
|
||||
|
||||
// clear current license
|
||||
currentLicense.set(null);
|
||||
}
|
||||
|
|
@ -454,14 +458,14 @@ public class LicensesService extends AbstractLifecycleComponent<LicensesService>
|
|||
}
|
||||
// notify all interested plugins
|
||||
if (previousClusterState.blocks().hasGlobalBlock(GatewayService.STATE_NOT_RECOVERED_BLOCK)) {
|
||||
notifyAndSchedule(currentLicensesMetaData);
|
||||
onUpdate(currentLicensesMetaData);
|
||||
} else {
|
||||
if (prevLicensesMetaData == null) {
|
||||
if (currentLicensesMetaData != null) {
|
||||
notifyAndSchedule(currentLicensesMetaData);
|
||||
onUpdate(currentLicensesMetaData);
|
||||
}
|
||||
} else if (!prevLicensesMetaData.equals(currentLicensesMetaData)) {
|
||||
notifyAndSchedule(currentLicensesMetaData);
|
||||
onUpdate(currentLicensesMetaData);
|
||||
}
|
||||
}
|
||||
// auto-generate license if no licenses ever existed
|
||||
|
|
@ -475,140 +479,69 @@ public class LicensesService extends AbstractLifecycleComponent<LicensesService>
|
|||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Notifies registered licensees of license state change and/or new active license
|
||||
* based on the license in <code>currentLicensesMetaData</code>.
|
||||
* Additionally schedules license expiry notifications and event callbacks
|
||||
* relative to the current license's expiry
|
||||
*/
|
||||
private void notifyAndSchedule(final LicensesMetaData currentLicensesMetaData) {
|
||||
final License license = getLicense(currentLicensesMetaData);
|
||||
private void notifyLicensees(final License license) {
|
||||
if (license == LicensesMetaData.LICENSE_TOMBSTONE) {
|
||||
// implies license has been explicitly deleted
|
||||
// update licensee states
|
||||
registeredLicensees.forEach(InternalLicensee::onRemove);
|
||||
return;
|
||||
}
|
||||
if (license != null) {
|
||||
logger.debug("notifying [{}] listeners", registeredLicensees.size());
|
||||
switch (getLicenseState(license, clock.millis())) {
|
||||
case ENABLED:
|
||||
for (InternalLicensee licensee : registeredLicensees) {
|
||||
licensee.onChange(license, LicenseState.ENABLED);
|
||||
}
|
||||
logger.debug("license [{}] - valid", license.uid());
|
||||
break;
|
||||
case GRACE_PERIOD:
|
||||
for (InternalLicensee licensee : registeredLicensees) {
|
||||
licensee.onChange(license, LicenseState.GRACE_PERIOD);
|
||||
}
|
||||
logger.warn("license [{}] - grace", license.uid());
|
||||
break;
|
||||
case DISABLED:
|
||||
for (InternalLicensee licensee : registeredLicensees) {
|
||||
licensee.onChange(license, LicenseState.DISABLED);
|
||||
}
|
||||
logger.warn("license [{}] - expired", license.uid());
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
static LicenseState getLicenseState(final License license, long time) {
|
||||
if (license.issueDate() > time) {
|
||||
return LicenseState.DISABLED;
|
||||
}
|
||||
if (license.expiryDate() > time) {
|
||||
return LicenseState.ENABLED;
|
||||
}
|
||||
if ((license.expiryDate() + GRACE_PERIOD_DURATION.getMillis()) > time) {
|
||||
return LicenseState.GRACE_PERIOD;
|
||||
}
|
||||
return LicenseState.DISABLED;
|
||||
}
|
||||
|
||||
/**
|
||||
* Notifies registered licensees of license state change and/or new active license
|
||||
* based on the license in <code>currentLicensesMetaData</code>.
|
||||
* Additionally schedules license expiry notifications and event callbacks
|
||||
* relative to the current license's expiry
|
||||
*/
|
||||
void onUpdate(final LicensesMetaData currentLicensesMetaData) {
|
||||
final License license = getLicense(currentLicensesMetaData);
|
||||
// license can be null if the trial license is yet to be auto-generated
|
||||
// in this case, it is a no-op
|
||||
if (license != null) {
|
||||
logger.debug("notifying [{}] listeners", registeredLicensees.size());
|
||||
long now = clock.millis();
|
||||
if (license.issueDate() > now) {
|
||||
logger.warn("license [{}] - invalid", license.uid());
|
||||
return;
|
||||
}
|
||||
long expiryDuration = license.expiryDate() - now;
|
||||
if (license.expiryDate() > now) {
|
||||
for (InternalLicensee licensee : registeredLicensees) {
|
||||
licensee.onChange(license, LicenseState.ENABLED);
|
||||
}
|
||||
logger.debug("license [{}] - valid", license.uid());
|
||||
final TimeValue delay = TimeValue.timeValueMillis(expiryDuration);
|
||||
// cancel any previous notifications
|
||||
cancelNotifications(expiryNotifications);
|
||||
try {
|
||||
logger.debug("schedule grace notification after [{}] for license [{}]", delay.toString(), license.uid());
|
||||
expiryNotifications.add(threadPool.schedule(delay, executorName(), new LicensingClientNotificationJob()));
|
||||
} catch (EsRejectedExecutionException ex) {
|
||||
logger.debug("couldn't schedule grace notification", ex);
|
||||
}
|
||||
} else if ((license.expiryDate() + gracePeriodDuration.getMillis()) > now) {
|
||||
for (InternalLicensee licensee : registeredLicensees) {
|
||||
licensee.onChange(license, LicenseState.GRACE_PERIOD);
|
||||
}
|
||||
logger.warn("license [{}] - grace", license.uid());
|
||||
final TimeValue delay = TimeValue.timeValueMillis(expiryDuration + gracePeriodDuration.getMillis());
|
||||
// cancel any previous notifications
|
||||
cancelNotifications(expiryNotifications);
|
||||
try {
|
||||
logger.debug("schedule expiry notification after [{}] for license [{}]", delay.toString(), license.uid());
|
||||
expiryNotifications.add(threadPool.schedule(delay, executorName(), new LicensingClientNotificationJob()));
|
||||
} catch (EsRejectedExecutionException ex) {
|
||||
logger.debug("couldn't schedule expiry notification", ex);
|
||||
}
|
||||
} else {
|
||||
for (InternalLicensee licensee : registeredLicensees) {
|
||||
licensee.onChange(license, LicenseState.DISABLED);
|
||||
}
|
||||
logger.warn("license [{}] - expired", license.uid());
|
||||
}
|
||||
if (!license.equals(currentLicense.get())) {
|
||||
notifyLicensees(license);
|
||||
if (license.equals(currentLicense.get()) == false) {
|
||||
currentLicense.set(license);
|
||||
// cancel all scheduled event notifications
|
||||
cancelNotifications(eventNotifications);
|
||||
// schedule expiry callbacks
|
||||
for (ExpirationCallback expirationCallback : this.expirationCallbacks) {
|
||||
final TimeValue delay;
|
||||
if (expirationCallback.matches(license.expiryDate(), now)) {
|
||||
expirationCallback.on(license);
|
||||
TimeValue frequency = expirationCallback.frequency();
|
||||
delay = frequency != null ? frequency : expirationCallback.delay(expiryDuration);
|
||||
} else {
|
||||
delay = expirationCallback.delay(expiryDuration);
|
||||
}
|
||||
if (delay != null) {
|
||||
eventNotifications.add(threadPool.schedule(delay, executorName(), new EventNotificationJob(expirationCallback)));
|
||||
}
|
||||
if (logger.isDebugEnabled()) {
|
||||
logger.debug("schedule [{}] after [{}]", expirationCallback, delay);
|
||||
}
|
||||
}
|
||||
logger.debug("scheduled expiry callbacks for [{}] expiring after [{}]", license.uid(),
|
||||
TimeValue.timeValueMillis(expiryDuration));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private class LicensingClientNotificationJob implements Runnable {
|
||||
@Override
|
||||
public void run() {
|
||||
logger.debug("running expiry notification");
|
||||
final ClusterState currentClusterState = clusterService.state();
|
||||
if (!currentClusterState.blocks().hasGlobalBlock(GatewayService.STATE_NOT_RECOVERED_BLOCK)) {
|
||||
final LicensesMetaData currentLicensesMetaData = currentClusterState.metaData().custom(LicensesMetaData.TYPE);
|
||||
notifyAndSchedule(currentLicensesMetaData);
|
||||
} else if (logger.isDebugEnabled()) {
|
||||
// next clusterChanged event will deal with the missed notifications
|
||||
logger.debug("skip expiry notification [{}]", GatewayService.STATE_NOT_RECOVERED_BLOCK);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private class EventNotificationJob implements Runnable {
|
||||
private final ExpirationCallback expirationCallback;
|
||||
|
||||
EventNotificationJob(ExpirationCallback expirationCallback) {
|
||||
this.expirationCallback = expirationCallback;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void run() {
|
||||
logger.debug("running event notification for [{}]", expirationCallback);
|
||||
LicensesMetaData currentLicensesMetaData = clusterService.state().metaData().custom(LicensesMetaData.TYPE);
|
||||
License license = getLicense(currentLicensesMetaData);
|
||||
if (license != null) {
|
||||
long now = clock.millis();
|
||||
if (expirationCallback.matches(license.expiryDate(), now)) {
|
||||
expirationCallback.on(license);
|
||||
if (expirationCallback.frequency() != null) {
|
||||
// schedule next event
|
||||
eventNotifications.add(threadPool.schedule(expirationCallback.frequency(), executorName(), this));
|
||||
}
|
||||
} else if (logger.isDebugEnabled()) {
|
||||
logger.debug("skip scheduling notification for [{}] with license expiring after [{}]", expirationCallback,
|
||||
TimeValue.timeValueMillis(license.expiryDate() - now));
|
||||
}
|
||||
}
|
||||
// clear out any finished event notifications
|
||||
while (!eventNotifications.isEmpty()) {
|
||||
ScheduledFuture notification = eventNotifications.peek();
|
||||
if (notification != null && notification.isDone()) {
|
||||
// remove the notifications that are done
|
||||
eventNotifications.poll();
|
||||
} else {
|
||||
// stop emptying out the queue as soon as the first undone future hits
|
||||
break;
|
||||
scheduler.add(new SchedulerEngine.Job(LICENSE_JOB, new LicenseSchedule(license)));
|
||||
for (ExpirationCallback expirationCallback : expirationCallbacks) {
|
||||
scheduler.add(new SchedulerEngine.Job(expirationCallback.getId(),
|
||||
expirationCallback.schedule(license.expiryDate())));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -632,8 +565,8 @@ public class LicensesService extends AbstractLifecycleComponent<LicensesService>
|
|||
// triggers a cluster changed event
|
||||
// eventually notifying the current licensee
|
||||
requestTrialLicense(clusterState);
|
||||
} else {
|
||||
notifyAndSchedule(currentMetaData);
|
||||
} else if (lifecycleState() == Lifecycle.State.STARTED) {
|
||||
notifyLicensees(currentMetaData.getLicense());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -663,25 +596,6 @@ public class LicensesService extends AbstractLifecycleComponent<LicensesService>
|
|||
return null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Cancels out all notification futures
|
||||
*/
|
||||
private static void cancelNotifications(Queue<ScheduledFuture> scheduledNotifications) {
|
||||
// clear out notification queue
|
||||
while (!scheduledNotifications.isEmpty()) {
|
||||
ScheduledFuture notification = scheduledNotifications.peek();
|
||||
if (notification != null) {
|
||||
// cancel
|
||||
FutureUtils.cancel(notification);
|
||||
scheduledNotifications.poll();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private String executorName() {
|
||||
return ThreadPool.Names.GENERIC;
|
||||
}
|
||||
|
||||
/**
|
||||
* Stores acknowledgement, expiration and license notification callbacks
|
||||
* for a registered listener
|
||||
|
|
@ -747,19 +661,4 @@ public class LicensesService extends AbstractLifecycleComponent<LicensesService>
|
|||
channel.sendResponse(TransportResponse.Empty.INSTANCE);
|
||||
}
|
||||
}
|
||||
|
||||
// TODO - temporary hack for tests, should be removed once we introduce `ClockMock`
|
||||
public void setGracePeriodDuration(TimeValue gracePeriodDuration) {
|
||||
this.gracePeriodDuration = gracePeriodDuration;
|
||||
}
|
||||
|
||||
// only for adding expiration callbacks for tests
|
||||
public void setExpirationCallbacks(List<ExpirationCallback> expirationCallbacks) {
|
||||
this.expirationCallbacks = expirationCallbacks;
|
||||
}
|
||||
|
||||
// TODO - temporary hack for tests, should be removed once we introduce `ClockMock`
|
||||
public void setTrialLicenseDuration(TimeValue trialLicenseDuration) {
|
||||
this.trialLicenseDuration = trialLicenseDuration;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -36,11 +36,14 @@ public abstract class TestPluginServiceBase extends AbstractLifecycleComponent<T
|
|||
this.eagerLicenseRegistration = eagerLicenseRegistration;
|
||||
this.licensesClientService = licensesClientService;
|
||||
int trialDurationInSec = settings.getAsInt("_trial_license_duration_in_seconds", -1);
|
||||
// TODO: remove integ licensing tests for external plugins
|
||||
/*
|
||||
if (trialDurationInSec != -1) {
|
||||
licensesClientService.setTrialLicenseDuration(TimeValue.timeValueSeconds(trialDurationInSec));
|
||||
}
|
||||
int graceDurationInSec = settings.getAsInt("_grace_duration_in_seconds", 5);
|
||||
licensesClientService.setGracePeriodDuration(TimeValue.timeValueSeconds(graceDurationInSec));
|
||||
*/
|
||||
if (!eagerLicenseRegistration) {
|
||||
this.clusterService = clusterService;
|
||||
clusterService.add(this);
|
||||
|
|
|
|||
|
|
@ -6,80 +6,58 @@
|
|||
package org.elasticsearch.license.plugin.core;
|
||||
|
||||
import org.elasticsearch.action.ActionListener;
|
||||
import org.elasticsearch.cluster.metadata.MetaData;
|
||||
import org.elasticsearch.cluster.ClusterStateUpdateTask;
|
||||
import org.elasticsearch.common.unit.TimeValue;
|
||||
import org.elasticsearch.license.core.License;
|
||||
import org.elasticsearch.license.plugin.TestUtils;
|
||||
import org.elasticsearch.license.plugin.action.put.PutLicenseRequest;
|
||||
import org.elasticsearch.license.plugin.action.put.PutLicenseResponse;
|
||||
import org.elasticsearch.test.ESSingleNodeTestCase;
|
||||
|
||||
import java.util.Collections;
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import static org.elasticsearch.license.plugin.TestUtils.awaitNoBlock;
|
||||
import static org.elasticsearch.license.plugin.TestUtils.awaitNoPendingTasks;
|
||||
import static org.elasticsearch.license.plugin.TestUtils.generateSignedLicense;
|
||||
import static org.hamcrest.Matchers.equalTo;
|
||||
import static org.hamcrest.Matchers.not;
|
||||
import static org.mockito.Matchers.any;
|
||||
import static org.mockito.Mockito.times;
|
||||
import static org.mockito.Mockito.verify;
|
||||
|
||||
public class LicensesAcknowledgementTests extends ESSingleNodeTestCase {
|
||||
static {
|
||||
MetaData.registerPrototype(LicensesMetaData.TYPE, LicensesMetaData.PROTO);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected boolean resetNodeAfterTest() {
|
||||
return true;
|
||||
}
|
||||
public class LicensesAcknowledgementTests extends AbstractLicenseServiceTestCase {
|
||||
|
||||
public void testAcknowledgment() throws Exception {
|
||||
final LicensesService licensesService = getInstanceFromNode(LicensesService.class);
|
||||
setInitialState(TestUtils.generateSignedLicense("trial", TimeValue.timeValueHours(2)));
|
||||
licensesService.start();
|
||||
String id = "testAcknowledgment";
|
||||
String[] acknowledgeMessages = new String[] {"message"};
|
||||
TestUtils.AssertingLicensee licensee = new TestUtils.AssertingLicensee(id, logger);
|
||||
licensee.setAcknowledgementMessages(acknowledgeMessages);
|
||||
awaitNoBlock(client());
|
||||
licensesService.register(licensee);
|
||||
awaitNoPendingTasks(client());
|
||||
// try installing a signed license
|
||||
License signedLicense = generateSignedLicense(TimeValue.timeValueHours(10));
|
||||
PutLicenseRequest putLicenseRequest = new PutLicenseRequest().license(signedLicense);
|
||||
CountDownLatch latch = new CountDownLatch(1);
|
||||
// ensure acknowledgement message was part of the response
|
||||
licensesService.registerLicense(putLicenseRequest, new AssertingLicensesUpdateResponse(false, LicensesStatus.VALID,
|
||||
Collections.singletonMap(id, acknowledgeMessages), latch));
|
||||
if (!latch.await(5, TimeUnit.SECONDS)) {
|
||||
fail("waiting too long for a response to license registration");
|
||||
}
|
||||
awaitNoPendingTasks(client());
|
||||
Collections.singletonMap(id, acknowledgeMessages)));
|
||||
assertThat(licensee.acknowledgementRequested.size(), equalTo(1));
|
||||
assertThat(licensee.acknowledgementRequested.get(0).v2(), equalTo(signedLicense));
|
||||
assertThat(licensesService.getLicense(), not(signedLicense));
|
||||
|
||||
latch = new CountDownLatch(1);
|
||||
// try installing a signed license with acknowledgement
|
||||
putLicenseRequest = new PutLicenseRequest().license(signedLicense).acknowledge(true);
|
||||
// ensure license was installed and no acknowledgment message was returned
|
||||
licensee.setAcknowledgementMessages(new String[0]);
|
||||
licensesService.registerLicense(putLicenseRequest, new AssertingLicensesUpdateResponse(true, LicensesStatus.VALID,
|
||||
Collections.<String, String[]>emptyMap(), latch));
|
||||
if (!latch.await(5, TimeUnit.SECONDS)) {
|
||||
fail("waiting too long for a response to license registration");
|
||||
}
|
||||
awaitNoPendingTasks(client());
|
||||
Collections.<String, String[]>emptyMap()));
|
||||
verify(clusterService, times(1)).submitStateUpdateTask(any(String.class), any(ClusterStateUpdateTask.class));
|
||||
assertThat(licensee.acknowledgementRequested.size(), equalTo(1));
|
||||
assertThat(licensee.acknowledgementRequested.get(0).v2(), equalTo(signedLicense));
|
||||
assertThat(licensesService.getLicense(), equalTo(signedLicense));
|
||||
licensesService.stop();
|
||||
}
|
||||
|
||||
public void testAcknowledgementMultipleLicensee() throws Exception {
|
||||
final LicensesService licensesService = getInstanceFromNode(LicensesService.class);
|
||||
setInitialState(TestUtils.generateSignedLicense("trial", TimeValue.timeValueHours(2)));
|
||||
licensesService.start();
|
||||
String id1 = "testAcknowledgementMultipleLicensee_1";
|
||||
String[] acknowledgeMessages1 = new String[] {"testAcknowledgementMultipleLicensee_1"};
|
||||
|
|
@ -89,43 +67,32 @@ public class LicensesAcknowledgementTests extends ESSingleNodeTestCase {
|
|||
licensee1.setAcknowledgementMessages(acknowledgeMessages1);
|
||||
TestUtils.AssertingLicensee licensee2 = new TestUtils.AssertingLicensee(id2, logger);
|
||||
licensee2.setAcknowledgementMessages(acknowledgeMessages2);
|
||||
awaitNoBlock(client());
|
||||
licensesService.register(licensee1);
|
||||
licensesService.register(licensee2);
|
||||
awaitNoPendingTasks(client());
|
||||
// try installing a signed license
|
||||
License signedLicense = generateSignedLicense(TimeValue.timeValueHours(10));
|
||||
PutLicenseRequest putLicenseRequest = new PutLicenseRequest().license(signedLicense);
|
||||
CountDownLatch latch = new CountDownLatch(1);
|
||||
// ensure acknowledgement message was part of the response
|
||||
final HashMap<String, String[]> expectedMessages = new HashMap<>();
|
||||
expectedMessages.put(id1, acknowledgeMessages1);
|
||||
expectedMessages.put(id2, acknowledgeMessages2);
|
||||
licensesService.registerLicense(putLicenseRequest, new AssertingLicensesUpdateResponse(false, LicensesStatus.VALID,
|
||||
expectedMessages, latch));
|
||||
if (!latch.await(5, TimeUnit.SECONDS)) {
|
||||
fail("waiting too long for a response to license registration");
|
||||
}
|
||||
awaitNoPendingTasks(client());
|
||||
expectedMessages));
|
||||
verify(clusterService, times(0)).submitStateUpdateTask(any(String.class), any(ClusterStateUpdateTask.class));
|
||||
assertThat(licensee2.acknowledgementRequested.size(), equalTo(1));
|
||||
assertThat(licensee2.acknowledgementRequested.get(0).v2(), equalTo(signedLicense));
|
||||
assertThat(licensee1.acknowledgementRequested.size(), equalTo(1));
|
||||
assertThat(licensee1.acknowledgementRequested.get(0).v2(), equalTo(signedLicense));
|
||||
assertThat(licensesService.getLicense(), not(signedLicense));
|
||||
|
||||
latch = new CountDownLatch(1);
|
||||
// try installing a signed license with acknowledgement
|
||||
putLicenseRequest = new PutLicenseRequest().license(signedLicense).acknowledge(true);
|
||||
// ensure license was installed and no acknowledgment message was returned
|
||||
licensee1.setAcknowledgementMessages(new String[0]);
|
||||
licensee2.setAcknowledgementMessages(new String[0]);
|
||||
licensesService.registerLicense(putLicenseRequest, new AssertingLicensesUpdateResponse(true, LicensesStatus.VALID,
|
||||
Collections.<String, String[]>emptyMap(), latch));
|
||||
if (!latch.await(5, TimeUnit.SECONDS)) {
|
||||
fail("waiting too long for a response to license registration");
|
||||
}
|
||||
awaitNoPendingTasks(client());
|
||||
assertThat(licensesService.getLicense(), equalTo(signedLicense));
|
||||
Collections.<String, String[]>emptyMap()));
|
||||
verify(clusterService, times(1)).submitStateUpdateTask(any(String.class), any(ClusterStateUpdateTask.class));
|
||||
licensesService.stop();
|
||||
}
|
||||
|
||||
|
|
@ -133,14 +100,12 @@ public class LicensesAcknowledgementTests extends ESSingleNodeTestCase {
|
|||
private final boolean expectedAcknowledgement;
|
||||
private final LicensesStatus expectedStatus;
|
||||
private final Map<String, String[]> expectedAckMessages;
|
||||
private final CountDownLatch latch;
|
||||
|
||||
public AssertingLicensesUpdateResponse(boolean expectedAcknowledgement, LicensesStatus expectedStatus,
|
||||
Map<String, String[]> expectedAckMessages, CountDownLatch latch) {
|
||||
Map<String, String[]> expectedAckMessages) {
|
||||
this.expectedAcknowledgement = expectedAcknowledgement;
|
||||
this.expectedStatus = expectedStatus;
|
||||
this.expectedAckMessages = expectedAckMessages;
|
||||
this.latch = latch;
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
@ -154,12 +119,10 @@ public class LicensesAcknowledgementTests extends ESSingleNodeTestCase {
|
|||
String[] actualMessages = actual.get(expectedEntry.getKey());
|
||||
assertThat(actualMessages, equalTo(expectedEntry.getValue()));
|
||||
}
|
||||
latch.countDown();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onFailure(Throwable throwable) {
|
||||
latch.countDown();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1,167 +0,0 @@
|
|||
/*
|
||||
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
|
||||
* or more contributor license agreements. Licensed under the Elastic License;
|
||||
* you may not use this file except in compliance with the Elastic License.
|
||||
*/
|
||||
package org.elasticsearch.license.plugin.core;
|
||||
|
||||
import org.elasticsearch.cluster.metadata.MetaData;
|
||||
import org.elasticsearch.common.unit.TimeValue;
|
||||
import org.elasticsearch.license.core.License;
|
||||
import org.elasticsearch.license.plugin.TestUtils;
|
||||
import org.elasticsearch.test.ESSingleNodeTestCase;
|
||||
|
||||
import java.util.Arrays;
|
||||
import java.util.Collections;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
|
||||
import static org.elasticsearch.common.unit.TimeValue.timeValueMillis;
|
||||
import static org.hamcrest.Matchers.equalTo;
|
||||
|
||||
public class LicensesExpirationCallbackTests extends ESSingleNodeTestCase {
|
||||
static {
|
||||
MetaData.registerPrototype(LicensesMetaData.TYPE, LicensesMetaData.PROTO);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected boolean resetNodeAfterTest() {
|
||||
return true;
|
||||
}
|
||||
|
||||
public void testPostExpiration() throws Exception {
|
||||
int postExpirySeconds = randomIntBetween(5, 10);
|
||||
TimeValue postExpiryDuration = TimeValue.timeValueSeconds(postExpirySeconds);
|
||||
TimeValue min = TimeValue.timeValueSeconds(postExpirySeconds - randomIntBetween(1, 3));
|
||||
TimeValue max = TimeValue.timeValueSeconds(postExpirySeconds + randomIntBetween(1, 10));
|
||||
|
||||
final ExpirationCallback.Post post = new ExpirationCallback.Post(min, max, timeValueMillis(10)) {
|
||||
@Override
|
||||
public void on(License license) {
|
||||
}
|
||||
};
|
||||
long now = System.currentTimeMillis();
|
||||
assertThat(post.matches(now - postExpiryDuration.millis(), now), equalTo(true));
|
||||
assertThat(post.matches(now + postExpiryDuration.getMillis(), now), equalTo(false));
|
||||
}
|
||||
|
||||
public void testPostExpirationWithNullMax() throws Exception {
|
||||
int postExpirySeconds = randomIntBetween(5, 10);
|
||||
TimeValue postExpiryDuration = TimeValue.timeValueSeconds(postExpirySeconds);
|
||||
TimeValue min = TimeValue.timeValueSeconds(postExpirySeconds - randomIntBetween(1, 3));
|
||||
|
||||
final ExpirationCallback.Post post = new ExpirationCallback.Post(min, null, timeValueMillis(10)) {
|
||||
@Override
|
||||
public void on(License license) {
|
||||
}
|
||||
};
|
||||
long now = System.currentTimeMillis();
|
||||
assertThat(post.matches(now - postExpiryDuration.millis(), now), equalTo(true));
|
||||
}
|
||||
|
||||
public void testPreExpirationWithNullMin() throws Exception {
|
||||
int expirySeconds = randomIntBetween(5, 10);
|
||||
TimeValue expiryDuration = TimeValue.timeValueSeconds(expirySeconds);
|
||||
TimeValue max = TimeValue.timeValueSeconds(expirySeconds + randomIntBetween(1, 10));
|
||||
|
||||
final ExpirationCallback.Pre pre = new ExpirationCallback.Pre(null, max, timeValueMillis(10)) {
|
||||
@Override
|
||||
public void on(License license) {
|
||||
}
|
||||
};
|
||||
long now = System.currentTimeMillis();
|
||||
assertThat(pre.matches(expiryDuration.millis() + now, now), equalTo(true));
|
||||
}
|
||||
|
||||
public void testPreExpiration() throws Exception {
|
||||
int expirySeconds = randomIntBetween(5, 10);
|
||||
TimeValue expiryDuration = TimeValue.timeValueSeconds(expirySeconds);
|
||||
TimeValue min = TimeValue.timeValueSeconds(expirySeconds - randomIntBetween(0, 3));
|
||||
TimeValue max = TimeValue.timeValueSeconds(expirySeconds + randomIntBetween(1, 10));
|
||||
|
||||
final ExpirationCallback.Pre pre = new ExpirationCallback.Pre(min, max, timeValueMillis(10)) {
|
||||
@Override
|
||||
public void on(License license) {
|
||||
}
|
||||
};
|
||||
long now = System.currentTimeMillis();
|
||||
assertThat(pre.matches(expiryDuration.millis() + now, now), equalTo(true));
|
||||
assertThat(pre.matches(now - expiryDuration.getMillis(), now), equalTo(false));
|
||||
}
|
||||
|
||||
public void testPreExpirationNotification() throws Exception {
|
||||
final LicensesService licensesService = getInstanceFromNode(LicensesService.class);
|
||||
licensesService.setTrialLicenseDuration(TimeValue.timeValueSeconds(5));
|
||||
AtomicInteger counter = new AtomicInteger(0);
|
||||
// 2000, 1600, 1200
|
||||
licensesService.setExpirationCallbacks(Collections.singletonList(
|
||||
preCallbackLatch(TimeValue.timeValueSeconds(1), TimeValue.timeValueSeconds(2), timeValueMillis(400), counter))
|
||||
);
|
||||
licensesService.start();
|
||||
TestUtils.AssertingLicensee licensee = new TestUtils.AssertingLicensee("testPreExpirationNotification", logger);
|
||||
licensesService.register(licensee);
|
||||
boolean success = awaitBusy(() -> (counter.get() == 3 || counter.get() == 2));
|
||||
assertThat("counter: actual: " + counter.get() + "vs expected: 3", success, equalTo(true));
|
||||
licensesService.stop();
|
||||
}
|
||||
|
||||
public void testPostExpirationNotification() throws Exception {
|
||||
final LicensesService licensesService = getInstanceFromNode(LicensesService.class);
|
||||
licensesService.setTrialLicenseDuration(TimeValue.timeValueSeconds(3));
|
||||
AtomicInteger counter = new AtomicInteger(0);
|
||||
// 700, 1700, 2700
|
||||
licensesService.setExpirationCallbacks(Collections.singletonList(
|
||||
postCallbackLatch(timeValueMillis(700), TimeValue.timeValueSeconds(3), TimeValue.timeValueSeconds(1), counter))
|
||||
);
|
||||
licensesService.start();
|
||||
TestUtils.AssertingLicensee licensee = new TestUtils.AssertingLicensee("testPostExpirationNotification", logger);
|
||||
licensesService.register(licensee);
|
||||
// callback can be called only twice if the third notification is triggered with a delay
|
||||
// causing the trigger time to be out of the post expiry callback window
|
||||
boolean success = awaitBusy(() -> (counter.get() == 3 || counter.get() == 2));
|
||||
assertThat("counter: actual: " + counter.get() + "vs expected: 3", success, equalTo(true));
|
||||
licensesService.stop();
|
||||
}
|
||||
|
||||
public void testMultipleExpirationNotification() throws Exception {
|
||||
final LicensesService licensesService = getInstanceFromNode(LicensesService.class);
|
||||
licensesService.setTrialLicenseDuration(TimeValue.timeValueSeconds(4));
|
||||
AtomicInteger postCounter = new AtomicInteger(0);
|
||||
AtomicInteger preCounter = new AtomicInteger(0);
|
||||
licensesService.setExpirationCallbacks(Arrays.asList(
|
||||
// 2000, 1600, 1200
|
||||
preCallbackLatch(TimeValue.timeValueSeconds(1), TimeValue.timeValueSeconds(2), timeValueMillis(400), preCounter),
|
||||
// 100, 500, 900, 1300, 1700
|
||||
postCallbackLatch(timeValueMillis(100), TimeValue.timeValueSeconds(2), timeValueMillis(400), postCounter))
|
||||
);
|
||||
licensesService.start();
|
||||
TestUtils.AssertingLicensee licensee = new TestUtils.AssertingLicensee("testMultipleExpirationNotification", logger);
|
||||
licensesService.register(licensee);
|
||||
// callback can be called one less than expected if the last notification is triggered
|
||||
// with a delay, causing the trigger time to be out of the expiry callback window
|
||||
boolean success = awaitBusy(() -> ((preCounter.get() == 3 || preCounter.get() == 2)
|
||||
&& (postCounter.get() == 5 || postCounter.get() == 4)));
|
||||
assertThat("post count: actual: " + postCounter.get() + "vs expected: 5 " +
|
||||
"pre count: actual: " + preCounter.get() + " vs expected: 3", success, equalTo(true));
|
||||
licensesService.stop();
|
||||
}
|
||||
|
||||
private static ExpirationCallback preCallbackLatch(TimeValue min, TimeValue max, TimeValue frequency,
|
||||
final AtomicInteger count) {
|
||||
return new ExpirationCallback.Pre(min, max, frequency) {
|
||||
@Override
|
||||
public void on(License license) {
|
||||
count.incrementAndGet();
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
private static ExpirationCallback postCallbackLatch(TimeValue min, TimeValue max, TimeValue frequency,
|
||||
final AtomicInteger count) {
|
||||
return new ExpirationCallback.Post(min, max, frequency) {
|
||||
@Override
|
||||
public void on(License license) {
|
||||
count.incrementAndGet();
|
||||
}
|
||||
};
|
||||
}
|
||||
}
|
||||
|
|
@ -1,223 +0,0 @@
|
|||
/*
|
||||
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
|
||||
* or more contributor license agreements. Licensed under the Elastic License;
|
||||
* you may not use this file except in compliance with the Elastic License.
|
||||
*/
|
||||
package org.elasticsearch.license.plugin.core;
|
||||
|
||||
import org.apache.lucene.util.LuceneTestCase.BadApple;
|
||||
import org.elasticsearch.cluster.metadata.MetaData;
|
||||
import org.elasticsearch.common.unit.TimeValue;
|
||||
import org.elasticsearch.license.plugin.TestUtils.AssertingLicensee;
|
||||
import org.elasticsearch.test.ESSingleNodeTestCase;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
import static org.elasticsearch.license.plugin.TestUtils.awaitNoBlock;
|
||||
import static org.elasticsearch.license.plugin.TestUtils.awaitNoPendingTasks;
|
||||
import static org.elasticsearch.license.plugin.TestUtils.generateSignedLicense;
|
||||
import static org.elasticsearch.license.plugin.TestUtils.registerAndAckSignedLicenses;
|
||||
import static org.hamcrest.Matchers.equalTo;
|
||||
|
||||
//test is just too slow, please fix it to not be sleep-based
|
||||
@BadApple(bugUrl = "https://github.com/elastic/x-plugins/issues/1007")
|
||||
public class LicensesExpiryNotificationTests extends ESSingleNodeTestCase {
|
||||
static {
|
||||
MetaData.registerPrototype(LicensesMetaData.TYPE, LicensesMetaData.PROTO);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected boolean resetNodeAfterTest() {
|
||||
return true;
|
||||
}
|
||||
|
||||
public void testTrialLicenseEnforcement() throws Exception {
|
||||
LicensesService licensesService = getInstanceFromNode(LicensesService.class);
|
||||
licensesService.setTrialLicenseDuration(TimeValue.timeValueSeconds(5));
|
||||
licensesService.setGracePeriodDuration(TimeValue.timeValueSeconds(3));
|
||||
licensesService.start();
|
||||
String id1 = "testTrialLicenseEnforcement";
|
||||
final AssertingLicensee licensee = new AssertingLicensee(id1, logger);
|
||||
awaitNoBlock(client());
|
||||
licensesService.register(licensee);
|
||||
awaitNoPendingTasks(client());
|
||||
boolean success = awaitBusy(() -> licensee.statuses.size() == 3);
|
||||
// trail license: enable, grace, disabled
|
||||
assertLicenseStates(licensee, LicenseState.ENABLED, LicenseState.GRACE_PERIOD, LicenseState.DISABLED);
|
||||
assertTrue(dumpLicensingStates(licensee.statuses), success);
|
||||
licensesService.stop();
|
||||
}
|
||||
|
||||
public void testTrialLicenseEnforcementMultipleLicensees() throws Exception {
|
||||
LicensesService licensesService = getInstanceFromNode(LicensesService.class);
|
||||
licensesService.setTrialLicenseDuration(TimeValue.timeValueSeconds(5));
|
||||
licensesService.setGracePeriodDuration(TimeValue.timeValueSeconds(3));
|
||||
licensesService.start();
|
||||
String id1 = "testTrialLicenseEnforcementMultipleLicensees_1";
|
||||
final AssertingLicensee licensee1 = new AssertingLicensee(id1, logger);
|
||||
String id12 = "testTrialLicenseEnforcementMultipleLicensees_2";
|
||||
final AssertingLicensee licensee2 = new AssertingLicensee(id12, logger);
|
||||
awaitNoBlock(client());
|
||||
licensesService.register(licensee1);
|
||||
licensesService.register(licensee2);
|
||||
awaitNoPendingTasks(client());
|
||||
boolean success = awaitBusy(() -> licensee1.statuses.size() == 3);
|
||||
assertTrue(dumpLicensingStates(licensee1.statuses), success);
|
||||
success = awaitBusy(() -> licensee2.statuses.size() == 3);
|
||||
assertTrue(dumpLicensingStates(licensee2.statuses), success);
|
||||
// trail license: enable, grace, disabled
|
||||
assertLicenseStates(licensee1, LicenseState.ENABLED, LicenseState.GRACE_PERIOD, LicenseState.DISABLED);
|
||||
assertLicenseStates(licensee2, LicenseState.ENABLED, LicenseState.GRACE_PERIOD, LicenseState.DISABLED);
|
||||
licensesService.stop();
|
||||
}
|
||||
|
||||
public void testTrialSignedLicenseEnforcement() throws Exception {
|
||||
LicensesService licensesService = getInstanceFromNode(LicensesService.class);
|
||||
licensesService.setTrialLicenseDuration(TimeValue.timeValueSeconds(2));
|
||||
licensesService.setGracePeriodDuration(TimeValue.timeValueSeconds(3));
|
||||
licensesService.start();
|
||||
String id1 = "testTrialSignedLicenseEnforcement";
|
||||
final AssertingLicensee licensee = new AssertingLicensee(id1, logger);
|
||||
awaitNoBlock(client());
|
||||
licensesService.register(licensee);
|
||||
awaitNoPendingTasks(client());
|
||||
boolean success = awaitBusy(() -> licensee.statuses.size() == 1);
|
||||
assertTrue(dumpLicensingStates(licensee.statuses), success);
|
||||
registerAndAckSignedLicenses(licensesService, generateSignedLicense(TimeValue.timeValueSeconds(4)), LicensesStatus.VALID);
|
||||
success = awaitBusy(() -> licensee.statuses.size() == 4);
|
||||
// trial: enable, signed: enable, signed: grace, signed: disabled
|
||||
assertLicenseStates(licensee, LicenseState.ENABLED, LicenseState.ENABLED, LicenseState.GRACE_PERIOD, LicenseState.DISABLED);
|
||||
assertTrue(dumpLicensingStates(licensee.statuses), success);
|
||||
licensesService.stop();
|
||||
}
|
||||
|
||||
public void testSignedLicenseEnforcement() throws Exception {
|
||||
LicensesService licensesService = getInstanceFromNode(LicensesService.class);
|
||||
licensesService.setTrialLicenseDuration(TimeValue.timeValueSeconds(4));
|
||||
licensesService.setGracePeriodDuration(TimeValue.timeValueSeconds(3));
|
||||
licensesService.start();
|
||||
String id1 = "testSignedLicenseEnforcement";
|
||||
final AssertingLicensee licensee = new AssertingLicensee(id1, logger);
|
||||
awaitNoBlock(client());
|
||||
registerAndAckSignedLicenses(licensesService, generateSignedLicense(TimeValue.timeValueSeconds(2)), LicensesStatus.VALID);
|
||||
licensesService.register(licensee);
|
||||
awaitNoPendingTasks(client());
|
||||
boolean success = awaitBusy(() -> licensee.statuses.size() == 3);
|
||||
// signed: enable, signed: grace, signed: disabled
|
||||
assertLicenseStates(licensee, LicenseState.ENABLED, LicenseState.GRACE_PERIOD, LicenseState.DISABLED);
|
||||
assertTrue(dumpLicensingStates(licensee.statuses), success);
|
||||
licensesService.stop();
|
||||
}
|
||||
|
||||
public void testSingedLicenseEnforcementMultipleLicensees() throws Exception {
|
||||
LicensesService licensesService = getInstanceFromNode(LicensesService.class);
|
||||
licensesService.setTrialLicenseDuration(TimeValue.timeValueSeconds(4));
|
||||
licensesService.setGracePeriodDuration(TimeValue.timeValueSeconds(3));
|
||||
licensesService.start();
|
||||
String id1 = "testSingedLicenseEnforcementMultipleLicensees_1";
|
||||
final AssertingLicensee licensee1 = new AssertingLicensee(id1, logger);
|
||||
String id12 = "testSingedLicenseEnforcementMultipleLicensees_2";
|
||||
final AssertingLicensee licensee2 = new AssertingLicensee(id12, logger);
|
||||
awaitNoBlock(client());
|
||||
registerAndAckSignedLicenses(licensesService, generateSignedLicense(TimeValue.timeValueSeconds(2)), LicensesStatus.VALID);
|
||||
licensesService.register(licensee1);
|
||||
licensesService.register(licensee2);
|
||||
awaitNoPendingTasks(client());
|
||||
boolean success = awaitBusy(() -> licensee1.statuses.size() == 3);
|
||||
assertTrue(dumpLicensingStates(licensee1.statuses), success);
|
||||
success = awaitBusy(() -> licensee2.statuses.size() == 3);
|
||||
assertTrue(dumpLicensingStates(licensee2.statuses), success);
|
||||
// signed license: enable, grace, disabled
|
||||
assertLicenseStates(licensee1, LicenseState.ENABLED, LicenseState.GRACE_PERIOD, LicenseState.DISABLED);
|
||||
assertLicenseStates(licensee2, LicenseState.ENABLED, LicenseState.GRACE_PERIOD, LicenseState.DISABLED);
|
||||
licensesService.stop();
|
||||
}
|
||||
|
||||
public void testMultipleSignedLicenseEnforcement() throws Exception {
|
||||
// register with trial license and assert onEnable and onDisable notification
|
||||
LicensesService licensesService = getInstanceFromNode(LicensesService.class);
|
||||
licensesService.setTrialLicenseDuration(TimeValue.timeValueSeconds(4));
|
||||
licensesService.setGracePeriodDuration(TimeValue.timeValueSeconds(1));
|
||||
licensesService.start();
|
||||
String id1 = "testMultipleSignedLicenseEnforcement";
|
||||
final AssertingLicensee licensee = new AssertingLicensee(id1, logger);
|
||||
awaitNoBlock(client());
|
||||
licensesService.register(licensee);
|
||||
awaitNoPendingTasks(client());
|
||||
// trial license enabled
|
||||
boolean success = awaitBusy(() -> licensee.statuses.size() == 1);
|
||||
assertTrue(dumpLicensingStates(licensee.statuses), success);
|
||||
registerAndAckSignedLicenses(licensesService, generateSignedLicense("basic", TimeValue.timeValueSeconds(3)), LicensesStatus.VALID);
|
||||
// signed license enabled
|
||||
success = awaitBusy(() -> licensee.statuses.size() == 2);
|
||||
assertTrue(dumpLicensingStates(licensee.statuses), success);
|
||||
registerAndAckSignedLicenses(licensesService, generateSignedLicense("gold", TimeValue.timeValueSeconds(2)), LicensesStatus.VALID);
|
||||
// second signed license enabled, grace and expired
|
||||
success = awaitBusy(() ->licensee.statuses.size() == 5);
|
||||
assertLicenseStates(licensee, LicenseState.ENABLED, LicenseState.ENABLED, LicenseState.ENABLED, LicenseState.GRACE_PERIOD,
|
||||
LicenseState.DISABLED);
|
||||
assertTrue(dumpLicensingStates(licensee.statuses), success);
|
||||
licensesService.stop();
|
||||
}
|
||||
|
||||
public void testNonOverlappingMultipleLicensesEnforcement() throws Exception {
|
||||
// register with trial license and assert onEnable and onDisable notification
|
||||
LicensesService licensesService = getInstanceFromNode(LicensesService.class);
|
||||
licensesService.setTrialLicenseDuration(TimeValue.timeValueSeconds(3));
|
||||
licensesService.setGracePeriodDuration(TimeValue.timeValueSeconds(1));
|
||||
licensesService.start();
|
||||
String id1 = "testNonOverlappingMultipleLicensesEnforcement";
|
||||
final AssertingLicensee licensee = new AssertingLicensee(id1, logger);
|
||||
awaitNoBlock(client());
|
||||
licensesService.register(licensee);
|
||||
// trial license: enabled, grace, disabled
|
||||
boolean success = awaitBusy(() -> licensee.statuses.size() == 3);
|
||||
|
||||
assertTrue(dumpLicensingStates(licensee.statuses), success);
|
||||
// install license
|
||||
registerAndAckSignedLicenses(licensesService, generateSignedLicense("basic", TimeValue.timeValueSeconds(2)), LicensesStatus.VALID);
|
||||
// trial license: enabled, grace, disabled, signed license: enabled, grace, disabled
|
||||
success = awaitBusy(() -> licensee.statuses.size() == 6);
|
||||
assertLicenseStates(licensee, LicenseState.ENABLED, LicenseState.GRACE_PERIOD, LicenseState.DISABLED, LicenseState.ENABLED,
|
||||
LicenseState.GRACE_PERIOD, LicenseState.DISABLED);
|
||||
assertTrue(dumpLicensingStates(licensee.statuses), success);
|
||||
licensesService.stop();
|
||||
}
|
||||
|
||||
private void assertLicenseStates(AssertingLicensee licensee, LicenseState... states) {
|
||||
StringBuilder msg = new StringBuilder();
|
||||
msg.append("Actual: ");
|
||||
msg.append(dumpLicensingStates(licensee.statuses));
|
||||
msg.append(" Expected: ");
|
||||
msg.append(dumpLicensingStates(states));
|
||||
assertThat(msg.toString(), licensee.statuses.size(), equalTo(states.length));
|
||||
for (int i = 0; i < states.length; i++) {
|
||||
assertThat(msg.toString(), licensee.statuses.get(i).getLicenseState(), equalTo(states[i]));
|
||||
}
|
||||
}
|
||||
|
||||
private String dumpLicensingStates(List<Licensee.Status> statuses) {
|
||||
return dumpLicensingStates(statuses.toArray(new Licensee.Status[statuses.size()]));
|
||||
}
|
||||
|
||||
private String dumpLicensingStates(Licensee.Status... statuses) {
|
||||
LicenseState[] states = new LicenseState[statuses.length];
|
||||
for (int i = 0; i < statuses.length; i++) {
|
||||
states[i] = statuses[i].getLicenseState();
|
||||
}
|
||||
return dumpLicensingStates(states);
|
||||
}
|
||||
|
||||
private String dumpLicensingStates(LicenseState... states) {
|
||||
StringBuilder sb = new StringBuilder();
|
||||
sb.append("[");
|
||||
for (int i = 0; i < states.length; i++) {
|
||||
sb.append(states[i].name());
|
||||
if (i != states.length - 1) {
|
||||
sb.append(", ");
|
||||
}
|
||||
}
|
||||
sb.append("]");
|
||||
return sb.toString();
|
||||
}
|
||||
}
|
||||
|
|
@ -9,12 +9,21 @@ import org.elasticsearch.action.ActionListener;
|
|||
import org.elasticsearch.cluster.ack.ClusterStateUpdateResponse;
|
||||
import org.elasticsearch.cluster.metadata.MetaData;
|
||||
import org.elasticsearch.cluster.service.ClusterService;
|
||||
import org.elasticsearch.common.settings.Settings;
|
||||
import org.elasticsearch.common.unit.TimeValue;
|
||||
import org.elasticsearch.graph.Graph;
|
||||
import org.elasticsearch.license.core.License;
|
||||
import org.elasticsearch.license.plugin.TestUtils;
|
||||
import org.elasticsearch.license.plugin.action.delete.DeleteLicenseRequest;
|
||||
import org.elasticsearch.marvel.Monitoring;
|
||||
import org.elasticsearch.plugins.Plugin;
|
||||
import org.elasticsearch.shield.Security;
|
||||
import org.elasticsearch.test.ESSingleNodeTestCase;
|
||||
import org.elasticsearch.xpack.XPackPlugin;
|
||||
import org.elasticsearch.xpack.watcher.Watcher;
|
||||
|
||||
import java.util.Collection;
|
||||
import java.util.Collections;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
|
||||
|
|
@ -26,8 +35,19 @@ import static org.hamcrest.Matchers.not;
|
|||
|
||||
public class LicensesManagerServiceTests extends ESSingleNodeTestCase {
|
||||
|
||||
static {
|
||||
MetaData.registerPrototype(LicensesMetaData.TYPE, LicensesMetaData.PROTO);
|
||||
@Override
|
||||
protected Collection<Class<? extends Plugin>> getPlugins() {
|
||||
return Collections.singletonList(XPackPlugin.class);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected Settings nodeSettings() {
|
||||
return Settings.builder().
|
||||
put(XPackPlugin.featureEnabledSetting(Security.NAME), false)
|
||||
.put(XPackPlugin.featureEnabledSetting(Monitoring.NAME), false)
|
||||
.put(XPackPlugin.featureEnabledSetting(Watcher.NAME), false)
|
||||
.put(XPackPlugin.featureEnabledSetting(Graph.NAME), false)
|
||||
.build();
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
@ -91,7 +111,7 @@ public class LicensesManagerServiceTests extends ESSingleNodeTestCase {
|
|||
// ensure that the invalid license never made it to cluster state
|
||||
LicensesMetaData licensesMetaData = clusterService.state().metaData().custom(LicensesMetaData.TYPE);
|
||||
if (licensesMetaData != null) {
|
||||
assertThat(licensesMetaData.getLicense(), equalTo(LicensesMetaData.LICENSE_TOMBSTONE));
|
||||
assertThat(licensesMetaData.getLicense(), not(equalTo(tamperedLicense)));
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -185,4 +205,4 @@ public class LicensesManagerServiceTests extends ESSingleNodeTestCase {
|
|||
}
|
||||
assertThat("remove license(s) failed", success.get(), equalTo(true));
|
||||
}
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue