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

[ML] fix x-pack usage regression caused by index migration (#36936) 

Changes the feature usage retrieval to use the job manager rather than
directly talking to the cluster state, because jobs can now be either in
cluster state or stored in an index

This is a follow-up of #36702 / #36698
This commit is contained in:
Hendrik Muhs 2018-12-31 08:30:08 +01:00 committed by GitHub
parent d3f1fe46d3
commit 632c7fbed2
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 136 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
x-pack/plugin/ml/qa/native-multi-node-tests/src/test/java/org/elasticsearch/xpack/ml/integration/MlJobIT.java
View File

@ -13,6 +13,7 @@ import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.util.concurrent.ConcurrentCollections;
import org.elasticsearch.common.util.concurrent.ConcurrentMapLong;
import org.elasticsearch.common.util.concurrent.ThreadContext;
import org.elasticsearch.common.xcontent.support.XContentMapValues;
import org.elasticsearch.test.SecuritySettingsSourceField;
import org.elasticsearch.test.rest.ESRestTestCase;
import org.elasticsearch.xpack.core.ml.integration.MlRestTestStateCleaner;
@ -22,7 +23,9 @@ import org.elasticsearch.xpack.ml.MachineLearning;
import org.junit.After;
import java.io.IOException;
import java.util.Collections;
import java.util.Locale;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.regex.Matcher;
@ -111,6 +114,21 @@ public class MlJobIT extends ESRestTestCase {
assertThat(implicitAll, containsString("\"job_id\":\"given-multiple-jobs-job-3\""));
}
// tests the _xpack/usage endpoint
public void testUsage() throws IOException {
createFarequoteJob("job-1");
createFarequoteJob("job-2");
Map<String, Object> usage = entityAsMap(client().performRequest(new Request("GET", "_xpack/usage")));
assertEquals(2, XContentMapValues.extractValue("ml.jobs._all.count", usage));
assertEquals(2, XContentMapValues.extractValue("ml.jobs.closed.count", usage));
Response openResponse = client().performRequest(new Request("POST", MachineLearning.BASE_PATH + "anomaly_detectors/job-1/_open"));
assertEquals(Collections.singletonMap("opened", true), entityAsMap(openResponse));
usage = entityAsMap(client().performRequest(new Request("GET", "_xpack/usage")));
assertEquals(2, XContentMapValues.extractValue("ml.jobs._all.count", usage));
assertEquals(1, XContentMapValues.extractValue("ml.jobs.closed.count", usage));
assertEquals(1, XContentMapValues.extractValue("ml.jobs.opened.count", usage));
}
private Response createFarequoteJob(String jobId) throws IOException {
Request request = new Request("PUT", MachineLearning.BASE_PATH + "anomaly_detectors/" + jobId);
request.setJsonEntity(

8
x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/MachineLearning.java
View File

@ -165,6 +165,7 @@ import org.elasticsearch.xpack.ml.datafeed.DatafeedJobBuilder;
import org.elasticsearch.xpack.ml.datafeed.DatafeedManager;
import org.elasticsearch.xpack.ml.datafeed.persistence.DatafeedConfigProvider;
import org.elasticsearch.xpack.ml.job.JobManager;
import org.elasticsearch.xpack.ml.job.JobManagerHolder;
import org.elasticsearch.xpack.ml.job.UpdateJobProcessNotifier;
import org.elasticsearch.xpack.ml.job.categorization.MlClassicTokenizer;
import org.elasticsearch.xpack.ml.job.categorization.MlClassicTokenizerFactory;
@ -375,7 +376,8 @@ public class MachineLearning extends Plugin implements ActionPlugin, AnalysisPlu
NamedXContentRegistry xContentRegistry, Environment environment,
NodeEnvironment nodeEnvironment, NamedWriteableRegistry namedWriteableRegistry) {
if (enabled == false || transportClientMode) {
return emptyList();
// special holder for @link(MachineLearningFeatureSetUsage) which needs access to job manager, empty if ML is disabled
return Collections.singletonList(new JobManagerHolder());
}
Auditor auditor = new Auditor(client, clusterService.getNodeName());
@ -385,6 +387,9 @@ public class MachineLearning extends Plugin implements ActionPlugin, AnalysisPlu
UpdateJobProcessNotifier notifier = new UpdateJobProcessNotifier(client, clusterService, threadPool);
JobManager jobManager = new JobManager(env, settings, jobResultsProvider, clusterService, auditor, threadPool, client, notifier);
// special holder for @link(MachineLearningFeatureSetUsage) which needs access to job manager if ML is enabled
JobManagerHolder jobManagerHolder = new JobManagerHolder(jobManager);
JobDataCountsPersister jobDataCountsPersister = new JobDataCountsPersister(client);
JobResultsPersister jobResultsPersister = new JobResultsPersister(client);
@ -443,6 +448,7 @@ public class MachineLearning extends Plugin implements ActionPlugin, AnalysisPlu
jobConfigProvider,
datafeedConfigProvider,
jobManager,
jobManagerHolder,
autodetectProcessManager,
new MlInitializationService(settings, threadPool, clusterService, client),
jobDataCountsPersister,

39
x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/MachineLearningFeatureSet.java
View File

@ -25,12 +25,12 @@ import org.elasticsearch.xpack.core.XPackPlugin;
import org.elasticsearch.xpack.core.XPackSettings;
import org.elasticsearch.xpack.core.XPackField;
import org.elasticsearch.xpack.core.ml.MachineLearningFeatureSetUsage;
import org.elasticsearch.xpack.core.ml.MlMetadata;
import org.elasticsearch.xpack.core.ml.action.GetDatafeedsStatsAction;
import org.elasticsearch.xpack.core.ml.action.GetJobsStatsAction;
import org.elasticsearch.xpack.core.ml.datafeed.DatafeedState;
import org.elasticsearch.xpack.core.ml.job.config.Job;
import org.elasticsearch.xpack.core.ml.job.config.JobState;
import org.elasticsearch.xpack.ml.job.JobManagerHolder;
import org.elasticsearch.xpack.ml.process.NativeController;
import org.elasticsearch.xpack.ml.process.NativeControllerHolder;
import org.elasticsearch.xpack.core.ml.job.process.autodetect.state.ModelSizeStats;
@ -47,6 +47,7 @@ import java.util.Locale;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;
public class MachineLearningFeatureSet implements XPackFeatureSet {
@ -60,15 +61,17 @@ public class MachineLearningFeatureSet implements XPackFeatureSet {
private final XPackLicenseState licenseState;
private final ClusterService clusterService;
private final Client client;
private final JobManagerHolder jobManagerHolder;
private final Map<String, Object> nativeCodeInfo;
@Inject
public MachineLearningFeatureSet(Environment environment, ClusterService clusterService, Client client,
@Nullable XPackLicenseState licenseState) {
@Nullable XPackLicenseState licenseState, JobManagerHolder jobManagerHolder) {
this.enabled = XPackSettings.MACHINE_LEARNING_ENABLED.get(environment.settings());
this.clusterService = Objects.requireNonNull(clusterService);
this.client = Objects.requireNonNull(client);
this.licenseState = licenseState;
this.jobManagerHolder = jobManagerHolder;
Map<String, Object> nativeCodeInfo = NativeController.UNKNOWN_NATIVE_CODE_INFO;
// Don't try to get the native code version if ML is disabled - it causes too much controversy
// if ML has been disabled because of some OS incompatibility. Also don't try to get the native
@ -133,7 +136,7 @@ public class MachineLearningFeatureSet implements XPackFeatureSet {
@Override
public void usage(ActionListener<XPackFeatureSet.Usage> listener) {
ClusterState state = clusterService.state();
new Retriever(client, MlMetadata.getMlMetadata(state), available(), enabled(), mlNodeCount(state)).execute(listener);
new Retriever(client, jobManagerHolder, available(), enabled(), mlNodeCount(state)).execute(listener);
}
private int mlNodeCount(final ClusterState clusterState) {
@ -153,16 +156,16 @@ public class MachineLearningFeatureSet implements XPackFeatureSet {
public static class Retriever {
private final Client client;
private final MlMetadata mlMetadata;
private final JobManagerHolder jobManagerHolder;
private final boolean available;
private final boolean enabled;
private Map<String, Object> jobsUsage;
private Map<String, Object> datafeedsUsage;
private int nodeCount;
public Retriever(Client client, MlMetadata mlMetadata, boolean available, boolean enabled, int nodeCount) {
public Retriever(Client client, JobManagerHolder jobManagerHolder, boolean available, boolean enabled, int nodeCount) {
this.client = Objects.requireNonNull(client);
this.mlMetadata = mlMetadata;
this.jobManagerHolder = jobManagerHolder;
this.available = available;
this.enabled = enabled;
this.jobsUsage = new LinkedHashMap<>();
@ -171,7 +174,8 @@ public class MachineLearningFeatureSet implements XPackFeatureSet {
}
public void execute(ActionListener<Usage> listener) {
if (enabled == false) {
// empty holder means either ML disabled or transport client mode
if (jobManagerHolder.isEmpty()) {
listener.onResponse(
new MachineLearningFeatureSetUsage(available, enabled, Collections.emptyMap(), Collections.emptyMap(), 0));
return;
@ -191,20 +195,19 @@ public class MachineLearningFeatureSet implements XPackFeatureSet {
GetJobsStatsAction.Request jobStatsRequest = new GetJobsStatsAction.Request(MetaData.ALL);
ActionListener<GetJobsStatsAction.Response> jobStatsListener = ActionListener.wrap(
response -> {
addJobsUsage(response);
GetDatafeedsStatsAction.Request datafeedStatsRequest =
new GetDatafeedsStatsAction.Request(GetDatafeedsStatsAction.ALL);
client.execute(GetDatafeedsStatsAction.INSTANCE, datafeedStatsRequest,
datafeedStatsListener);
},
listener::onFailure
);
jobManagerHolder.getJobManager().expandJobs(MetaData.ALL, true, ActionListener.wrap(jobs -> {
addJobsUsage(response, jobs.results());
GetDatafeedsStatsAction.Request datafeedStatsRequest = new GetDatafeedsStatsAction.Request(
GetDatafeedsStatsAction.ALL);
client.execute(GetDatafeedsStatsAction.INSTANCE, datafeedStatsRequest, datafeedStatsListener);
}, listener::onFailure));
}, listener::onFailure);
// Step 0. Kick off the chain of callbacks by requesting jobs stats
client.execute(GetJobsStatsAction.INSTANCE, jobStatsRequest, jobStatsListener);
}
private void addJobsUsage(GetJobsStatsAction.Response response) {
private void addJobsUsage(GetJobsStatsAction.Response response, List<Job> jobs) {
StatsAccumulator allJobsDetectorsStats = new StatsAccumulator();
StatsAccumulator allJobsModelSizeStats = new StatsAccumulator();
ForecastStats allJobsForecastStats = new ForecastStats();
@ -214,11 +217,11 @@ public class MachineLearningFeatureSet implements XPackFeatureSet {
Map<JobState, StatsAccumulator> modelSizeStatsByState = new HashMap<>();
Map<JobState, ForecastStats> forecastStatsByState = new HashMap<>();
Map<String, Job> jobs = mlMetadata.getJobs();
List<GetJobsStatsAction.Response.JobStats> jobsStats = response.getResponse().results();
Map<String, Job> jobMap = jobs.stream().collect(Collectors.toMap(Job::getId, item -> item));
for (GetJobsStatsAction.Response.JobStats jobStats : jobsStats) {
ModelSizeStats modelSizeStats = jobStats.getModelSizeStats();
int detectorsCount = jobs.get(jobStats.getJobId()).getAnalysisConfig()
int detectorsCount = jobMap.get(jobStats.getJobId()).getAnalysisConfig()
.getDetectors().size();
double modelSize = modelSizeStats == null ? 0.0
: jobStats.getModelSizeStats().getModelBytes();

46
x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/JobManagerHolder.java Normal file
View File

@ -0,0 +1,46 @@
/*
* 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.xpack.ml.job;
import org.elasticsearch.ElasticsearchException;
public class JobManagerHolder {
private final JobManager instance;
/**
* Create an empty holder which also means that no job manager gets created.
*/
public JobManagerHolder() {
this.instance = null;
}
/**
* Create a holder that allows lazy creation of a job manager.
*
*/
public JobManagerHolder(JobManager jobManager) {
this.instance = jobManager;
}
public boolean isEmpty() {
return instance == null;
}
/**
* Get the instance of the held JobManager.
*
* @return job manager instance
* @throws ElasticsearchException if holder has been created with the empty constructor
*/
public JobManager getJobManager() {
if (instance == null) {
throw new ElasticsearchException("Tried to get job manager although Machine Learning is disabled");
}
return instance;
}
}

59
x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/MachineLearningFeatureSetTests.java
View File

@ -31,7 +31,6 @@ import org.elasticsearch.xpack.core.XPackFeatureSet.Usage;
import org.elasticsearch.xpack.core.XPackField;
import org.elasticsearch.xpack.core.ml.MachineLearningFeatureSetUsage;
import org.elasticsearch.xpack.core.ml.MachineLearningField;
import org.elasticsearch.xpack.core.ml.MlMetadata;
import org.elasticsearch.xpack.core.ml.action.GetDatafeedsStatsAction;
import org.elasticsearch.xpack.core.ml.action.GetJobsStatsAction;
import org.elasticsearch.xpack.core.ml.action.util.QueryPage;
@ -46,6 +45,8 @@ import org.elasticsearch.xpack.core.ml.job.process.autodetect.state.ModelSizeSta
import org.elasticsearch.xpack.core.ml.stats.ForecastStats;
import org.elasticsearch.xpack.core.ml.stats.ForecastStatsTests;
import org.elasticsearch.xpack.core.watcher.support.xcontent.XContentSource;
import org.elasticsearch.xpack.ml.job.JobManager;
import org.elasticsearch.xpack.ml.job.JobManagerHolder;
import org.junit.Before;
import java.util.Arrays;
@ -62,6 +63,7 @@ import static org.hamcrest.Matchers.notNullValue;
import static org.hamcrest.Matchers.nullValue;
import static org.hamcrest.core.Is.is;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.eq;
import static org.mockito.Matchers.same;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
@ -72,6 +74,8 @@ public class MachineLearningFeatureSetTests extends ESTestCase {
private Settings commonSettings;
private ClusterService clusterService;
private Client client;
private JobManager jobManager;
private JobManagerHolder jobManagerHolder;
private XPackLicenseState licenseState;
@Before
@ -82,7 +86,11 @@ public class MachineLearningFeatureSetTests extends ESTestCase {
.build();
clusterService = mock(ClusterService.class);
client = mock(Client.class);
jobManager = mock(JobManager.class);
jobManagerHolder = new JobManagerHolder(jobManager);
licenseState = mock(XPackLicenseState.class);
ClusterState clusterState = new ClusterState.Builder(ClusterState.EMPTY_STATE).build();
when(clusterService.state()).thenReturn(clusterState);
givenJobs(Collections.emptyList(), Collections.emptyList());
givenDatafeeds(Collections.emptyList());
}
@ -104,7 +112,7 @@ public class MachineLearningFeatureSetTests extends ESTestCase {
public void testAvailable() throws Exception {
MachineLearningFeatureSet featureSet = new MachineLearningFeatureSet(TestEnvironment.newEnvironment(commonSettings), clusterService,
client, licenseState);
client, licenseState, jobManagerHolder);
boolean available = randomBoolean();
when(licenseState.isMachineLearningAllowed()).thenReturn(available);
assertThat(featureSet.available(), is(available));
@ -129,7 +137,7 @@ public class MachineLearningFeatureSetTests extends ESTestCase {
}
boolean expected = enabled || useDefault;
MachineLearningFeatureSet featureSet = new MachineLearningFeatureSet(TestEnvironment.newEnvironment(settings.build()),
clusterService, client, licenseState);
clusterService, client, licenseState, jobManagerHolder);
assertThat(featureSet.enabled(), is(expected));
PlainActionFuture<Usage> future = new PlainActionFuture<>();
featureSet.usage(future);
@ -163,7 +171,7 @@ public class MachineLearningFeatureSetTests extends ESTestCase {
));
MachineLearningFeatureSet featureSet = new MachineLearningFeatureSet(TestEnvironment.newEnvironment(settings.build()),
clusterService, client, licenseState);
clusterService, client, licenseState, jobManagerHolder);
PlainActionFuture<Usage> future = new PlainActionFuture<>();
featureSet.usage(future);
XPackFeatureSet.Usage mlUsage = future.get();
@ -232,6 +240,28 @@ public class MachineLearningFeatureSetTests extends ESTestCase {
}
}
public void testUsageDisabledML() throws Exception {
when(licenseState.isMachineLearningAllowed()).thenReturn(true);
Settings.Builder settings = Settings.builder().put(commonSettings);
settings.put("xpack.ml.enabled", false);
JobManagerHolder emptyJobManagerHolder = new JobManagerHolder();
MachineLearningFeatureSet featureSet = new MachineLearningFeatureSet(TestEnvironment.newEnvironment(settings.build()),
clusterService, client, licenseState, emptyJobManagerHolder);
PlainActionFuture<Usage> future = new PlainActionFuture<>();
featureSet.usage(future);
XPackFeatureSet.Usage mlUsage = future.get();
BytesStreamOutput out = new BytesStreamOutput();
mlUsage.writeTo(out);
XPackFeatureSet.Usage serializedUsage = new MachineLearningFeatureSetUsage(out.bytes().streamInput());
for (XPackFeatureSet.Usage usage : Arrays.asList(mlUsage, serializedUsage)) {
assertThat(usage, is(notNullValue()));
assertThat(usage.name(), is(XPackField.MACHINE_LEARNING));
assertThat(usage.enabled(), is(false));
}
}
public void testNodeCount() throws Exception {
when(licenseState.isMachineLearningAllowed()).thenReturn(true);
int nodeCount = randomIntBetween(1, 3);
@ -239,7 +269,7 @@ public class MachineLearningFeatureSetTests extends ESTestCase {
Settings.Builder settings = Settings.builder().put(commonSettings);
settings.put("xpack.ml.enabled", true);
MachineLearningFeatureSet featureSet = new MachineLearningFeatureSet(TestEnvironment.newEnvironment(settings.build()),
clusterService, client, licenseState);
clusterService, client, licenseState, jobManagerHolder);
PlainActionFuture<Usage> future = new PlainActionFuture<>();
featureSet.usage(future);
@ -282,7 +312,7 @@ public class MachineLearningFeatureSetTests extends ESTestCase {
when(clusterService.state()).thenReturn(ClusterState.EMPTY_STATE);
MachineLearningFeatureSet featureSet = new MachineLearningFeatureSet(TestEnvironment.newEnvironment(settings.build()),
clusterService, client, licenseState);
clusterService, client, licenseState, jobManagerHolder);
PlainActionFuture<Usage> future = new PlainActionFuture<>();
featureSet.usage(future);
XPackFeatureSet.Usage usage = future.get();
@ -319,15 +349,14 @@ public class MachineLearningFeatureSetTests extends ESTestCase {
}
private void givenJobs(List<Job> jobs, List<GetJobsStatsAction.Response.JobStats> jobsStats) {
MlMetadata.Builder mlMetadataBuilder = new MlMetadata.Builder();
for (Job job : jobs) {
mlMetadataBuilder.putJob(job, false);
}
ClusterState clusterState = new ClusterState.Builder(ClusterState.EMPTY_STATE)
.metaData(new MetaData.Builder()
.putCustom(MlMetadata.TYPE, mlMetadataBuilder.build()))
.build();
when(clusterService.state()).thenReturn(clusterState);
doAnswer(invocationOnMock -> {
@SuppressWarnings("unchecked")
ActionListener<QueryPage<Job>> jobListener =
(ActionListener<QueryPage<Job>>) invocationOnMock.getArguments()[2];
jobListener.onResponse(
new QueryPage<>(jobs, jobs.size(), Job.RESULTS_FIELD));
return Void.TYPE;
}).when(jobManager).expandJobs(eq(MetaData.ALL), eq(true), any(ActionListener.class));
doAnswer(invocationOnMock -> {
ActionListener<GetJobsStatsAction.Response> listener =