Skip to content

feat(wlm): Add additional search settings and override_request_values to WLM groups #21523

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
cwperks merged 3 commits into from
May 15, 2026
Merged

feat(wlm): Add additional search settings and override_request_values to WLM groups #21523

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
452e08b
feat(wlm): Add additional search settings and override_request_values…
dzane17 May 7, 2026
5bdf642
Fix tests and override in response
dzane17 May 8, 2026
6a38df0
Merge branch 'opensearch-project:main' into more-settings
dzane17 May 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
191 changes: 187 additions & 4 deletions ...rkload-management/src/javaRestTest/java/org/opensearch/rest/WorkloadManagementRestIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -148,14 +148,17 @@ public void testOperationWhenWlmDisabled() throws Exception {
}

public void testSearchSettings() throws Exception {
// Create with settings
// Create with all search settings
String createJson = """
{
"name": "search_test",
"resiliency_mode": "enforced",
"resource_limits": {"cpu": 0.3, "memory": 0.3},
"settings": {
"search.default_search_timeout": "30s"
"search.default_search_timeout": "30s",
"search.cancel_after_time_interval": "1m",
"search.max_concurrent_shard_requests": "5",
"search.batched_reduce_size": "512"
}
}""";
Response response = performOperation("PUT", "_wlm/workload_group", createJson);
Expand All @@ -165,13 +168,20 @@ public void testSearchSettings() throws Exception {
Response getResponse = performOperation("GET", "_wlm/workload_group/search_test", null);
String responseBody = EntityUtils.toString(getResponse.getEntity());
assertTrue(responseBody.contains("\"settings\""));
assertFalse(responseBody.contains("\"override_request_values\""));
assertTrue(responseBody.contains("\"search.default_search_timeout\":\"30s\""));
assertTrue(responseBody.contains("\"search.cancel_after_time_interval\":\"1m\""));
assertTrue(responseBody.contains("\"search.max_concurrent_shard_requests\":\"5\""));
assertTrue(responseBody.contains("\"search.batched_reduce_size\":\"512\""));

// Update settings
// Update search settings
String updateJson = """
{
"settings": {
"search.default_search_timeout": "1m"
"search.default_search_timeout": "1m",
"search.cancel_after_time_interval": "5m",
"search.max_concurrent_shard_requests": "10",
"search.batched_reduce_size": "256"
}
}""";
Response updateResponse = performOperation("PUT", "_wlm/workload_group/search_test", updateJson);
Expand All @@ -181,10 +191,183 @@ public void testSearchSettings() throws Exception {
Response getResponse2 = performOperation("GET", "_wlm/workload_group/search_test", null);
String responseBody2 = EntityUtils.toString(getResponse2.getEntity());
assertTrue(responseBody2.contains("\"search.default_search_timeout\":\"1m\""));
assertTrue(responseBody2.contains("\"search.cancel_after_time_interval\":\"5m\""));
assertTrue(responseBody2.contains("\"search.max_concurrent_shard_requests\":\"10\""));
assertTrue(responseBody2.contains("\"search.batched_reduce_size\":\"256\""));

performOperation("DELETE", "_wlm/workload_group/search_test", null);
}

public void testSearchSettingsOverrideRequestValues() throws Exception {
Comment thread
dzane17 marked this conversation as resolved.
// Create a WLM group with override_request_values=true so WLM settings win over request params
String createJson = """
{
"name": "override_test",
"resiliency_mode": "enforced",
"resource_limits": {"cpu": 0.3, "memory": 0.3},
"settings": {
"search.default_search_timeout": "30s",
"search.max_concurrent_shard_requests": "3",
"search.batched_reduce_size": "64",
"override_request_values": "true"
}
}""";
Response response = performOperation("PUT", "_wlm/workload_group", createJson);
assertEquals(200, response.getStatusLine().getStatusCode());

// Verify override_request_values is "true" in GET response
Response getResponse = performOperation("GET", "_wlm/workload_group/override_test", null);
assertTrue(EntityUtils.toString(getResponse.getEntity()).contains("\"override_request_values\":\"true\""));

// Toggle to false and verify
String toggleJson = """
{"settings": {"override_request_values": "false"}}""";
Response toggleResponse = performOperation("PUT", "_wlm/workload_group/override_test", toggleJson);
assertEquals(200, toggleResponse.getStatusLine().getStatusCode());
Response getResponse2 = performOperation("GET", "_wlm/workload_group/override_test", null);
assertTrue(EntityUtils.toString(getResponse2.getEntity()).contains("\"override_request_values\":\"false\""));

// Exercise the full request path: create an index, run a search with the WLM header.
// This confirms the listener is wired into the request flow without errors. Override
// semantics themselves are verified in WorkloadGroupRequestOperationListenerTests.
performOperation("PUT", "wlm-test-idx", "{\"settings\":{\"number_of_shards\":1,\"number_of_replicas\":0}}");
performOperation("POST", "wlm-test-idx/_doc", "{\"msg\":\"hello\"}");
performOperation("POST", "wlm-test-idx/_refresh", null);

Request searchRequest = new Request("POST", "wlm-test-idx/_search");
searchRequest.setJsonEntity("{\"query\":{\"match_all\":{}},\"timeout\":\"1m\"}");
searchRequest.setOptions(searchRequest.getOptions().toBuilder().addHeader("X-opaque-id", "wlm=override_test"));
Response searchResponse = client().performRequest(searchRequest);
assertEquals(200, searchResponse.getStatusLine().getStatusCode());
assertTrue(EntityUtils.toString(searchResponse.getEntity()).contains("\"hits\""));

performOperation("DELETE", "wlm-test-idx", null);
performOperation("DELETE", "_wlm/workload_group/override_test", null);
}

public void testSearchSettingsInvalidSettingsRejected() throws Exception {
// Unknown setting key should be rejected
String unknownKeyJson = """
{
"name": "invalid_test",
"resiliency_mode": "enforced",
"resource_limits": {"cpu": 0.3, "memory": 0.3},
"settings": {
"unknown_setting": "value"
}
}""";
ResponseException unknownKeyException = expectThrows(
ResponseException.class,
() -> performOperation("PUT", "_wlm/workload_group", unknownKeyJson)
);
assertTrue(EntityUtils.toString(unknownKeyException.getResponse().getEntity()).contains("Unknown WLM setting: unknown_setting"));

// Invalid value for max_concurrent_shard_requests (must be >= 1)
String invalidIntJson = """
{
"name": "invalid_test",
"resiliency_mode": "enforced",
"resource_limits": {"cpu": 0.3, "memory": 0.3},
"settings": {
"search.max_concurrent_shard_requests": "0"
}
}""";
ResponseException invalidIntException = expectThrows(
ResponseException.class,
() -> performOperation("PUT", "_wlm/workload_group", invalidIntJson)
);
String invalidIntBody = EntityUtils.toString(invalidIntException.getResponse().getEntity());
assertTrue(invalidIntBody.contains("search.max_concurrent_shard_requests"));
assertTrue(invalidIntBody.contains("must be >= 1"));

// Invalid value for batched_reduce_size (must be >= 2)
String invalidBatchJson = """
{
"name": "invalid_test",
"resiliency_mode": "enforced",
"resource_limits": {"cpu": 0.3, "memory": 0.3},
"settings": {
"search.batched_reduce_size": "1"
}
}""";
ResponseException invalidBatchException = expectThrows(
ResponseException.class,
() -> performOperation("PUT", "_wlm/workload_group", invalidBatchJson)
);
String invalidBatchBody = EntityUtils.toString(invalidBatchException.getResponse().getEntity());
assertTrue(invalidBatchBody.contains("search.batched_reduce_size"));
assertTrue(invalidBatchBody.contains("must be >= 2"));

// Invalid time value
String invalidTimeJson = """
{
"name": "invalid_test",
"resiliency_mode": "enforced",
"resource_limits": {"cpu": 0.3, "memory": 0.3},
"settings": {
"search.cancel_after_time_interval": "not_a_time"
}
}""";
ResponseException invalidTimeException = expectThrows(
ResponseException.class,
() -> performOperation("PUT", "_wlm/workload_group", invalidTimeJson)
);
String invalidTimeBody = EntityUtils.toString(invalidTimeException.getResponse().getEntity());
assertTrue(invalidTimeBody.contains("search.cancel_after_time_interval"));
assertTrue(invalidTimeBody.contains("Invalid value"));
}

public void testSearchSettingsMergeSemantics() throws Exception {
// Create with multiple settings
String createJson = """
{
"name": "merge_test",
"resiliency_mode": "enforced",
"resource_limits": {"cpu": 0.3, "memory": 0.3},
"settings": {
"search.default_search_timeout": "30s",
"search.max_concurrent_shard_requests": "5",
"override_request_values": "true"
}
}""";
Response response = performOperation("PUT", "_wlm/workload_group", createJson);
assertEquals(200, response.getStatusLine().getStatusCode());

// Update only timeout — other settings should persist
String updateJson = """
{
"settings": {
"search.default_search_timeout": "1m"
}
}""";
Response updateResponse = performOperation("PUT", "_wlm/workload_group/merge_test", updateJson);
assertEquals(200, updateResponse.getStatusLine().getStatusCode());

// Verify merge: timeout updated, max_concurrent persists, override persists
Response getResponse = performOperation("GET", "_wlm/workload_group/merge_test", null);
String responseBody = EntityUtils.toString(getResponse.getEntity());
assertTrue(responseBody.contains("\"search.default_search_timeout\":\"1m\""));
assertTrue(responseBody.contains("\"search.max_concurrent_shard_requests\":\"5\""));
assertTrue(responseBody.contains("\"override_request_values\":\"true\""));

// Clear all settings with empty object
String clearJson = """
{
"settings": {}
}""";
Response clearResponse = performOperation("PUT", "_wlm/workload_group/merge_test", clearJson);
assertEquals(200, clearResponse.getStatusLine().getStatusCode());

// Verify cleared — all settings should be gone
Response getResponse2 = performOperation("GET", "_wlm/workload_group/merge_test", null);
String responseBody2 = EntityUtils.toString(getResponse2.getEntity());
assertFalse(responseBody2.contains("\"override_request_values\""));
assertFalse(responseBody2.contains("\"search.default_search_timeout\""));
assertFalse(responseBody2.contains("\"search.max_concurrent_shard_requests\""));

performOperation("DELETE", "_wlm/workload_group/merge_test", null);
}

static String getCreateJson(String name, String resiliencyMode, double cpu, double memory) {
return String.format(Locale.ROOT, """
{
Expand Down
10 changes: 10 additions & 0 deletions server/src/main/java/org/opensearch/action/search/SearchRequest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@
import org.opensearch.action.IndicesRequest;
import org.opensearch.action.support.IndicesOptions;
import org.opensearch.common.Nullable;
import org.opensearch.common.annotation.ExperimentalApi;
import org.opensearch.common.annotation.PublicApi;
import org.opensearch.common.io.stream.BytesStreamOutput;
import org.opensearch.common.unit.TimeValue;
Expand Down Expand Up @@ -649,6 +650,15 @@ public int getMaxConcurrentShardRequests() {
return maxConcurrentShardRequests == 0 ? 5 : maxConcurrentShardRequests;
}

/**
* Returns the raw value of maxConcurrentShardRequests without applying the default.
* A value of {@code 0} means the user has not explicitly set this parameter.
*/
@ExperimentalApi
public int getMaxConcurrentShardRequestsRaw() {
return maxConcurrentShardRequests;
}

/**
* Sets the number of shard requests that should be executed concurrently on a single node. This value should be used as a
* protection mechanism to reduce the number of shard requests fired per high level search request. Searches that hit the entire
Expand Down
30 changes: 24 additions & 6 deletions server/src/main/java/org/opensearch/cluster/metadata/WorkloadGroup.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -114,18 +114,31 @@ public static WorkloadGroup updateExistingWorkloadGroup(
}
final ResiliencyMode mode = Optional.ofNullable(mutableWorkloadGroupFragment.getResiliencyMode())
.orElse(existingGroup.getResiliencyMode());
// Handle settings update:
// null = not specified (keep existing)
// empty Settings = explicitly clear (set to empty)
// non-empty Settings = replace with new values
// Handle settings update with merge semantics:
// null settings = not specified in request (keep existing)
// empty Settings = clear all settings
// non-empty Settings = merge with existing; keys with null values are removed
final Settings mutableFragmentSettings = mutableWorkloadGroupFragment.getSettings();
final Settings updatedSettings;
if (mutableFragmentSettings == null) {
// Not specified - keep existing
updatedSettings = Settings.builder().put(existingGroup.getSettings()).build();
} else if (mutableFragmentSettings.isEmpty()) {
// Explicitly empty - clear all settings
updatedSettings = Settings.EMPTY;
} else {
// Specified (empty or non-empty) - use the new value
updatedSettings = Settings.builder().put(mutableFragmentSettings).build();
// Merge: start with existing settings, overlay new values, remove null-valued keys
Settings.Builder builder = Settings.builder().put(existingGroup.getSettings());
for (String key : mutableFragmentSettings.keySet()) {
String value = mutableFragmentSettings.get(key);
if (value == null) {
// null value means "clear this setting"
builder.remove(key);
} else {
builder.put(key, value);
}
}
updatedSettings = builder.build();
}
return new WorkloadGroup(
existingGroup.getName(),
Expand Down Expand Up @@ -284,6 +297,11 @@ public static Builder fromXContent(XContentParser parser) throws IOException {
throw new IllegalArgumentException(fieldName + " is not a valid object in WorkloadGroup");
}
mutableWorkloadGroupFragment1.parseField(parser, fieldName);
} else if (token == XContentParser.Token.VALUE_NULL) {
if (fieldName.equals(MutableWorkloadGroupFragment.SETTINGS_STRING)) {
// "settings": null means clear all settings
mutableWorkloadGroupFragment1.parseField(parser, fieldName);
}
}
}
return builder.mutableWorkloadGroupFragment(mutableWorkloadGroupFragment1);
Expand Down
9 changes: 7 additions & 2 deletions server/src/main/java/org/opensearch/wlm/MutableWorkloadGroupFragment.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,7 +56,7 @@ public MutableWorkloadGroupFragment(ResiliencyMode resiliencyMode, Map<ResourceT
WorkloadGroupSearchSettings.validate(settings);
this.resiliencyMode = resiliencyMode;
this.resourceLimits = resourceLimits;
this.settings = settings;
this.settings = settings != null ? settings : Settings.EMPTY;
}

public MutableWorkloadGroupFragment(StreamInput in) throws IOException {
Expand Down Expand Up @@ -109,6 +109,10 @@ public Map<ResourceType, Double> parseField(XContentParser parser) throws IOExce

static class SearchSettingsParser implements FieldParser<Settings> {
public Settings parseField(XContentParser parser) throws IOException {
// "settings": null means clear all settings
if (parser.currentToken() == XContentParser.Token.VALUE_NULL) {
return Settings.EMPTY;
}
Settings settings = Settings.fromXContent(parser);
WorkloadGroupSearchSettings.validate(settings);
return settings;
Expand Down Expand Up @@ -292,6 +296,7 @@ void setResourceLimits(Map<ResourceType, Double> resourceLimits) {

void setSettings(Settings settings) {
WorkloadGroupSearchSettings.validate(settings);
this.settings = settings;
this.settings = settings != null ? settings : Settings.EMPTY;
}

}
Loading
Loading