Skip to content

Commit 69245bd

Browse files
maxhniebergallelasticmachine
maxhniebergall
and
elasticmachine
authored
[ML] Wait for up to 2 seconds for yellow status before starting search (#115938)
* Wait for up to 2 seconds for yellow before starting search * Update docs/changelog/115938.yaml * Update docs/changelog/115938.yaml --------- Co-authored-by: Elastic Machine <[email protected]>
1 parent a2075f5 commit 69245bd

File tree

2 files changed

+100
-51
lines changed

2 files changed

+100
-51
lines changed

‎docs/changelog/115938.yaml‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,9 @@
1+
pr: 115938
2+
summary: Wait for up to 2 seconds for yellow status before starting search
3+
area: Machine Learning
4+
type: bug
5+
issues:
6+
- 107777
7+
- 105955
8+
- 107815
9+
- 112191

‎x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/inference/persistence/TrainedModelProvider.java‎

Lines changed: 91 additions & 51 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,8 @@
1414
import org.elasticsearch.ResourceNotFoundException;
1515
import org.elasticsearch.action.ActionListener;
1616
import org.elasticsearch.action.DocWriteRequest;
17+
import org.elasticsearch.action.admin.cluster.health.ClusterHealthRequest;
18+
import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
1719
import org.elasticsearch.action.admin.indices.refresh.RefreshAction;
1820
import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
1921
import org.elasticsearch.action.bulk.BulkItemResponse;
@@ -28,6 +30,7 @@
2830
import org.elasticsearch.action.search.SearchResponse;
2931
import org.elasticsearch.action.search.TransportSearchAction;
3032
import org.elasticsearch.action.support.IndicesOptions;
33+
import org.elasticsearch.action.support.SubscribableListener;
3134
import org.elasticsearch.action.support.WriteRequest;
3235
import org.elasticsearch.action.support.broadcast.BroadcastResponse;
3336
import org.elasticsearch.action.support.master.AcknowledgedResponse;
@@ -43,6 +46,7 @@
4346
import org.elasticsearch.common.xcontent.LoggingDeprecationHandler;
4447
import org.elasticsearch.common.xcontent.XContentHelper;
4548
import org.elasticsearch.core.Nullable;
49+
import org.elasticsearch.core.TimeValue;
4650
import org.elasticsearch.core.Tuple;
4751
import org.elasticsearch.index.IndexNotFoundException;
4852
import org.elasticsearch.index.engine.VersionConflictEngineException;
@@ -105,6 +109,7 @@
105109
import java.util.Objects;
106110
import java.util.Set;
107111
import java.util.TreeSet;
112+
import java.util.concurrent.TimeUnit;
108113
import java.util.function.Function;
109114
import java.util.stream.Collectors;
110115

@@ -1053,63 +1058,98 @@ public void expandIds(
10531058
}
10541059

10551060
public void getInferenceStats(String[] modelIds, @Nullable TaskId parentTaskId, ActionListener<List<InferenceStats>> listener) {
1056-
MultiSearchRequest multiSearchRequest = new MultiSearchRequest();
1057-
Arrays.stream(modelIds).map(TrainedModelProvider::buildStatsSearchRequest).forEach(multiSearchRequest::add);
1058-
if (multiSearchRequest.requests().isEmpty()) {
1059-
listener.onResponse(Collections.emptyList());
1060-
return;
1061-
}
1062-
if (parentTaskId != null) {
1063-
multiSearchRequest.setParentTask(parentTaskId);
1064-
}
1065-
executeAsyncWithOrigin(
1066-
client.threadPool().getThreadContext(),
1067-
ML_ORIGIN,
1068-
multiSearchRequest,
1069-
ActionListener.<MultiSearchResponse>wrap(responses -> {
1070-
List<InferenceStats> allStats = new ArrayList<>(modelIds.length);
1071-
int modelIndex = 0;
1072-
assert responses.getResponses().length == modelIds.length : "mismatch between search response size and models requested";
1073-
for (MultiSearchResponse.Item response : responses.getResponses()) {
1074-
if (response.isFailure()) {
1075-
if (ExceptionsHelper.unwrapCause(response.getFailure()) instanceof ResourceNotFoundException) {
1076-
modelIndex++;
1077-
continue;
1078-
}
1061+
1062+
SubscribableListener.<ClusterHealthResponse>newForked((delegate) -> {
1063+
// first wait for the index to be available
1064+
executeAsyncWithOrigin(
1065+
client.threadPool().getThreadContext(),
1066+
ML_ORIGIN,
1067+
new ClusterHealthRequest(new TimeValue(2, TimeUnit.SECONDS), MlStatsIndex.indexPattern()).waitForYellowStatus(),
1068+
delegate,
1069+
client.admin().cluster()::health
1070+
);
1071+
})
1072+
.<List<InferenceStats>>andThen(
1073+
client.threadPool().executor(MachineLearning.UTILITY_THREAD_POOL_NAME),
1074+
client.threadPool().getThreadContext(),
1075+
(delegate, clusterHealthResponse) -> {
1076+
if (clusterHealthResponse.isTimedOut()) {
10791077
logger.error(
1080-
() -> "[" + Strings.arrayToCommaDelimitedString(modelIds) + "] search failed for models",
1081-
response.getFailure()
1082-
);
1083-
listener.onFailure(
1084-
ExceptionsHelper.serverError(
1085-
"Searching for stats for models [{}] failed",
1086-
response.getFailure(),
1087-
Strings.arrayToCommaDelimitedString(modelIds)
1088-
)
1078+
"getInferenceStats Timed out waiting for index [{}] to be available, "
1079+
+ "this will probably cause the request to fail",
1080+
MlStatsIndex.indexPattern()
10891081
);
1090-
return;
10911082
}
1092-
try {
1093-
InferenceStats inferenceStats = handleMultiNodeStatsResponse(response.getResponse(), modelIds[modelIndex++]);
1094-
if (inferenceStats != null) {
1095-
allStats.add(inferenceStats);
1096-
}
1097-
} catch (Exception e) {
1098-
listener.onFailure(e);
1083+
1084+
MultiSearchRequest multiSearchRequest = new MultiSearchRequest();
1085+
Arrays.stream(modelIds).map(TrainedModelProvider::buildStatsSearchRequest).forEach(multiSearchRequest::add);
1086+
if (multiSearchRequest.requests().isEmpty()) {
1087+
delegate.onResponse(Collections.emptyList());
10991088
return;
11001089
}
1090+
if (parentTaskId != null) {
1091+
multiSearchRequest.setParentTask(parentTaskId);
1092+
}
1093+
executeAsyncWithOrigin(
1094+
client.threadPool().getThreadContext(),
1095+
ML_ORIGIN,
1096+
multiSearchRequest,
1097+
ActionListener.<MultiSearchResponse>wrap(responses -> {
1098+
List<InferenceStats> allStats = new ArrayList<>(modelIds.length);
1099+
int modelIndex = 0;
1100+
assert responses.getResponses().length == modelIds.length
1101+
: "mismatch between search response size and models requested";
1102+
for (MultiSearchResponse.Item response : responses.getResponses()) {
1103+
if (response.isFailure()) {
1104+
if (ExceptionsHelper.unwrapCause(response.getFailure()) instanceof ResourceNotFoundException) {
1105+
modelIndex++;
1106+
continue;
1107+
}
1108+
logger.error(
1109+
() -> "[" + Strings.arrayToCommaDelimitedString(modelIds) + "] search failed for models",
1110+
response.getFailure()
1111+
);
1112+
delegate.onFailure(
1113+
ExceptionsHelper.serverError(
1114+
"Searching for stats for models [{}] failed",
1115+
response.getFailure(),
1116+
Strings.arrayToCommaDelimitedString(modelIds)
1117+
)
1118+
);
1119+
return;
1120+
}
1121+
try {
1122+
InferenceStats inferenceStats = handleMultiNodeStatsResponse(
1123+
response.getResponse(),
1124+
modelIds[modelIndex++]
1125+
);
1126+
if (inferenceStats != null) {
1127+
allStats.add(inferenceStats);
1128+
}
1129+
} catch (Exception e) {
1130+
delegate.onFailure(e);
1131+
return;
1132+
}
1133+
}
1134+
delegate.onResponse(allStats);
1135+
}, e -> {
1136+
Throwable unwrapped = ExceptionsHelper.unwrapCause(e);
1137+
if (unwrapped instanceof ResourceNotFoundException) {
1138+
delegate.onResponse(Collections.emptyList());
1139+
return;
1140+
}
1141+
delegate.onFailure((Exception) unwrapped);
1142+
}),
1143+
client::multiSearch
1144+
);
1145+
11011146
}
1102-
listener.onResponse(allStats);
1103-
}, e -> {
1104-
Throwable unwrapped = ExceptionsHelper.unwrapCause(e);
1105-
if (unwrapped instanceof ResourceNotFoundException) {
1106-
listener.onResponse(Collections.emptyList());
1107-
return;
1108-
}
1109-
listener.onFailure((Exception) unwrapped);
1110-
}),
1111-
client::multiSearch
1112-
);
1147+
)
1148+
.addListener(
1149+
listener,
1150+
client.threadPool().executor(MachineLearning.UTILITY_THREAD_POOL_NAME),
1151+
client.threadPool().getThreadContext()
1152+
);
11131153
}
11141154

11151155
private static SearchRequest buildStatsSearchRequest(String modelId) {

0 commit comments

Comments
 (0)