honeymoose/OpenSearch
1
0
Fork 0
mirror of https://github.com/honeymoose/OpenSearch.git synced 2025-02-17 18:35:25 +00:00
Code Issues Packages Projects Releases Wiki Activity

Install prelert metadata and create required indices only once.

Browse Source 
Original commit: elastic/x-pack-elasticsearch@12c8ba0ce0
This commit is contained in:
Martijn van Groningen 2017-01-04 15:50:14 +01:00
parent 27c9f39bf5
commit 51e1199860
2 changed files with 80 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

86
elasticsearch/src/main/java/org/elasticsearch/xpack/prelert/job/metadata/PrelertInitializationService.java
View File

@ -18,12 +18,18 @@ import org.elasticsearch.threadpool.ThreadPool;
import org.elasticsearch.xpack.prelert.job.persistence.AnomalyDetectorsIndex; import org.elasticsearch.xpack.prelert.job.persistence.AnomalyDetectorsIndex;
import org.elasticsearch.xpack.prelert.job.persistence.JobProvider; import org.elasticsearch.xpack.prelert.job.persistence.JobProvider;
import java.util.concurrent.atomic.AtomicBoolean;
public class PrelertInitializationService extends AbstractComponent implements ClusterStateListener { public class PrelertInitializationService extends AbstractComponent implements ClusterStateListener {
private final ThreadPool threadPool; private final ThreadPool threadPool;
private final ClusterService clusterService; private final ClusterService clusterService;
private final JobProvider jobProvider; private final JobProvider jobProvider;
private final AtomicBoolean installPrelertMetadataCheck = new AtomicBoolean(false);
private final AtomicBoolean createPrelertUsageIndexCheck = new AtomicBoolean(false);
private final AtomicBoolean createStateIndexCheck = new AtomicBoolean(false);
public PrelertInitializationService(Settings settings, ThreadPool threadPool, ClusterService clusterService, public PrelertInitializationService(Settings settings, ThreadPool threadPool, ClusterService clusterService,
JobProvider jobProvider) { JobProvider jobProvider) {
super(settings); super(settings);
@ -38,54 +44,64 @@ public class PrelertInitializationService extends AbstractComponent implements C
if (event.localNodeMaster()) { if (event.localNodeMaster()) {
MetaData metaData = event.state().metaData(); MetaData metaData = event.state().metaData();
if (metaData.custom(PrelertMetadata.TYPE) == null) { if (metaData.custom(PrelertMetadata.TYPE) == null) {
threadPool.executor(ThreadPool.Names.GENERIC).execute(() -> { if (installPrelertMetadataCheck.compareAndSet(false, true)) {
clusterService.submitStateUpdateTask("install-prelert-metadata", new ClusterStateUpdateTask() { threadPool.executor(ThreadPool.Names.GENERIC).execute(() -> {
@Override clusterService.submitStateUpdateTask("install-prelert-metadata", new ClusterStateUpdateTask() {
public ClusterState execute(ClusterState currentState) throws Exception { @Override
ClusterState.Builder builder = new ClusterState.Builder(currentState); public ClusterState execute(ClusterState currentState) throws Exception {
MetaData.Builder metadataBuilder = MetaData.builder(currentState.metaData()); ClusterState.Builder builder = new ClusterState.Builder(currentState);
metadataBuilder.putCustom(PrelertMetadata.TYPE, PrelertMetadata.EMPTY_METADATA); MetaData.Builder metadataBuilder = MetaData.builder(currentState.metaData());
builder.metaData(metadataBuilder.build()); metadataBuilder.putCustom(PrelertMetadata.TYPE, PrelertMetadata.EMPTY_METADATA);
return builder.build(); builder.metaData(metadataBuilder.build());
} return builder.build();
}
@Override @Override
public void onFailure(String source, Exception e) { public void onFailure(String source, Exception e) {
logger.error("unable to install prelert metadata upon startup", e); logger.error("unable to install prelert metadata upon startup", e);
} }
});
}); });
}); }
} else {
installPrelertMetadataCheck.set(false);
} }
if (metaData.hasIndex(JobProvider.PRELERT_USAGE_INDEX) == false) { if (metaData.hasIndex(JobProvider.PRELERT_USAGE_INDEX) == false) {
threadPool.executor(ThreadPool.Names.GENERIC).execute(() -> { if (createPrelertUsageIndexCheck.compareAndSet(false, true)) {
jobProvider.createUsageMeteringIndex((result, error) -> { threadPool.executor(ThreadPool.Names.GENERIC).execute(() -> {
if (result) { jobProvider.createUsageMeteringIndex((result, error) -> {
logger.info("successfully created prelert-usage index"); if (result) {
} else { logger.info("successfully created prelert-usage index");
if (error instanceof ResourceAlreadyExistsException) { createPrelertUsageIndexCheck.set(false);
logger.debug("not able to create prelert-usage index as it already exists");
} else { } else {
logger.error("not able to create prelert-usage index", error); if (error instanceof ResourceAlreadyExistsException) {
logger.debug("not able to create prelert-usage index as it already exists");
} else {
logger.error("not able to create prelert-usage index", error);
}
} }
} });
}); });
}); }
} }
String stateIndexName = AnomalyDetectorsIndex.jobStateIndexName(); String stateIndexName = AnomalyDetectorsIndex.jobStateIndexName();
if (metaData.hasIndex(stateIndexName) == false) { if (metaData.hasIndex(stateIndexName) == false) {
threadPool.executor(ThreadPool.Names.GENERIC).execute(() -> { if (createStateIndexCheck.compareAndSet(false, true)) {
jobProvider.createJobStateIndex((result, error) -> { threadPool.executor(ThreadPool.Names.GENERIC).execute(() -> {
if (result) { jobProvider.createJobStateIndex((result, error) -> {
logger.info("successfully created {} index", stateIndexName); if (result) {
} else { logger.info("successfully created {} index", stateIndexName);
if (error instanceof ResourceAlreadyExistsException) { createStateIndexCheck.set(false);
logger.debug("not able to create {} index as it already exists", stateIndexName);
} else { } else {
logger.error("not able to create " + stateIndexName + " index", error); if (error instanceof ResourceAlreadyExistsException) {
logger.debug("not able to create {} index as it already exists", stateIndexName);
} else {
logger.error("not able to create " + stateIndexName + " index", error);
}
} }
} });
}); });
}); }
} }
} }
} }

29
elasticsearch/src/test/java/org/elasticsearch/xpack/prelert/job/metadata/PrelertInitializationServiceTests.java
View File

@ -124,4 +124,33 @@ public class PrelertInitializationServiceTests extends ESTestCase {
verify(jobProvider, times(0)).createUsageMeteringIndex(any()); verify(jobProvider, times(0)).createUsageMeteringIndex(any());
verify(jobProvider, times(0)).createJobStateIndex(any()); verify(jobProvider, times(0)).createJobStateIndex(any());
} }
public void testInitialize_onlyOnce() {
ThreadPool threadPool = mock(ThreadPool.class);
ExecutorService executorService = mock(ExecutorService.class);
doAnswer(invocation -> {
((Runnable) invocation.getArguments()[0]).run();
return null;
}).when(executorService).execute(any(Runnable.class));
when(threadPool.executor(ThreadPool.Names.GENERIC)).thenReturn(executorService);
ClusterService clusterService = mock(ClusterService.class);
JobProvider jobProvider = mock(JobProvider.class);
PrelertInitializationService initializationService =
new PrelertInitializationService(Settings.EMPTY, threadPool, clusterService, jobProvider);
ClusterState cs = ClusterState.builder(new ClusterName("_name"))
.nodes(DiscoveryNodes.builder()
.add(new DiscoveryNode("_node_id", new LocalTransportAddress("_id"), Version.CURRENT))
.localNodeId("_node_id")
.masterNodeId("_node_id"))
.metaData(MetaData.builder())
.build();
initializationService.clusterChanged(new ClusterChangedEvent("_source", cs, cs));
initializationService.clusterChanged(new ClusterChangedEvent("_source", cs, cs));
verify(clusterService, times(1)).submitStateUpdateTask(eq("install-prelert-metadata"), any());
verify(jobProvider, times(1)).createUsageMeteringIndex(any());
verify(jobProvider, times(1)).createJobStateIndex(any());
}
} }