Skip to content

Commit 023b4d0

Browse files
ruai0511ruai0511
committed
adding plugin-specific routing layer
Signed-off-by: Ruirui Zhang <[email protected]>
1 parent cfde69c commit 023b4d0

File tree

24 files changed

+579
-435
lines changed

24 files changed

+579
-435
lines changed

‎modules/autotagging-commons/common/src/main/java/org/opensearch/rule/RuleRoutingService.java

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,25 @@
1+
/*
2+
* SPDX-License-Identifier: Apache-2.0
3+
*
4+
* The OpenSearch Contributors require contributions made to
5+
* this file be licensed under the Apache-2.0 license or a
6+
* compatible open source license.
7+
*/
8+
9+
package org.opensearch.rule;
10+
11+
import org.opensearch.core.action.ActionListener;
12+
13+
/**
14+
* Interface that handles rule routing logic
15+
* @opensearch.experimental
16+
*/
17+
public interface RuleRoutingService {
18+
19+
/**
20+
* Handles a create rule request by routing it to the appropriate node.
21+
* @param request the create rule request
22+
* @param listener listener to handle the final response
23+
*/
24+
void handleCreateRuleRequest(CreateRuleRequest request, ActionListener<CreateRuleResponse> listener);
25+
}

‎modules/autotagging-commons/common/src/main/java/org/opensearch/rule/service/IndexStoredRulePersistenceService.java

Lines changed: 40 additions & 46 deletions
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,9 @@
1414
import org.opensearch.action.DocWriteResponse;
1515
import org.opensearch.action.delete.DeleteRequest;
1616
import org.opensearch.action.index.IndexRequest;
17+
import org.opensearch.action.index.IndexResponse;
1718
import org.opensearch.action.search.SearchRequestBuilder;
19+
import org.opensearch.action.search.SearchResponse;
1820
import org.opensearch.action.support.clustermanager.AcknowledgedResponse;
1921
import org.opensearch.cluster.service.ClusterService;
2022
import org.opensearch.common.util.concurrent.ThreadContext;
@@ -37,7 +39,6 @@
3739
import org.opensearch.search.sort.SortOrder;
3840
import org.opensearch.transport.client.Client;
3941

40-
import java.io.IOException;
4142
import java.util.Arrays;
4243
import java.util.List;
4344
import java.util.Map;
@@ -95,10 +96,10 @@ public IndexStoredRulePersistenceService(
9596
* @param listener ActionListener for CreateRuleResponse
9697
*/
9798
public void createRule(CreateRuleRequest request, ActionListener<CreateRuleResponse> listener) {
98-
try (ThreadContext.StoredContext ctx = getContext()) {
99+
try (ThreadContext.StoredContext ctx = stashContext()) {
99100
if (!clusterService.state().metadata().hasIndex(indexName)) {
100101
logger.error("Index {} does not exist", indexName);
101-
throw new IllegalStateException("Index" + indexName + " does not exist");
102+
listener.onFailure(new IllegalStateException("Index" + indexName + " does not exist"));
102103
} else {
103104
Rule rule = request.getRule();
104105
validateNoDuplicateRule(rule, ActionListener.wrap(unused -> persistRule(rule, listener), listener::onFailure));
@@ -107,30 +108,28 @@ public void createRule(CreateRuleRequest request, ActionListener<CreateRuleRespo
107108
}
108109

109110
/**
110-
* Validates that no duplicate rule exists with the same attribute map.
111-
* If a conflict is found, fails the listener
111+
* Validates that no existing rule has the same attribute map as the given rule.
112+
* This validation must be performed one at a time to prevent writing duplicate rules.
112113
* @param rule - the rule we check duplicate against
113114
* @param listener - listener for validateNoDuplicateRule response
114115
*/
115116
private void validateNoDuplicateRule(Rule rule, ActionListener<Void> listener) {
116-
try (ThreadContext.StoredContext ctx = getContext()) {
117-
QueryBuilder query = queryBuilder.from(new GetRuleRequest(null, rule.getAttributeMap(), null, rule.getFeatureType()));
118-
getRuleFromIndex(null, query, null, new ActionListener<>() {
119-
@Override
120-
public void onResponse(GetRuleResponse getRuleResponse) {
121-
Optional<String> duplicateRuleId = RuleUtils.getDuplicateRuleId(rule, getRuleResponse.getRules());
122-
duplicateRuleId.ifPresentOrElse(
123-
id -> listener.onFailure(new IllegalArgumentException("Duplicate rule exists under id " + id)),
124-
() -> listener.onResponse(null)
125-
);
126-
}
117+
QueryBuilder query = queryBuilder.from(new GetRuleRequest(null, rule.getAttributeMap(), null, rule.getFeatureType()));
118+
getRuleFromIndex(null, query, null, new ActionListener<>() {
119+
@Override
120+
public void onResponse(GetRuleResponse getRuleResponse) {
121+
Optional<String> duplicateRuleId = RuleUtils.getDuplicateRuleId(rule, getRuleResponse.getRules());
122+
duplicateRuleId.ifPresentOrElse(
123+
id -> listener.onFailure(new IllegalArgumentException("Duplicate rule exists under id " + id)),
124+
() -> listener.onResponse(null)
125+
);
126+
}
127127

128-
@Override
129-
public void onFailure(Exception e) {
130-
listener.onFailure(e);
131-
}
132-
});
133-
}
128+
@Override
129+
public void onFailure(Exception e) {
130+
listener.onFailure(e);
131+
}
132+
});
134133
}
135134

136135
/**
@@ -139,17 +138,13 @@ public void onFailure(Exception e) {
139138
* @param listener - ActionListener for CreateRuleResponse
140139
*/
141140
private void persistRule(Rule rule, ActionListener<CreateRuleResponse> listener) {
142-
try (ThreadContext.StoredContext ctx = getContext()) {
141+
try {
143142
IndexRequest indexRequest = new IndexRequest(indexName).source(
144143
rule.toXContent(XContentFactory.jsonBuilder(), ToXContent.EMPTY_PARAMS)
145144
);
146-
client.index(indexRequest, ActionListener.wrap(indexResponse -> {
147-
listener.onResponse(new CreateRuleResponse(indexResponse.getId(), rule));
148-
}, e -> {
149-
logger.warn("Failed to save Rule object due to error: {}", e.getMessage());
150-
listener.onFailure(e);
151-
}));
152-
} catch (IOException e) {
145+
IndexResponse indexResponse = client.index(indexRequest).get();
146+
listener.onResponse(new CreateRuleResponse(indexResponse.getId(), rule));
147+
} catch (Exception e) {
153148
logger.error("Error saving rule to index: {}", indexName);
154149
listener.onFailure(new RuntimeException("Failed to save rule to index."));
155150
}
@@ -161,8 +156,10 @@ private void persistRule(Rule rule, ActionListener<CreateRuleResponse> listener)
161156
* @param listener the listener for GetRuleResponse.
162157
*/
163158
public void getRule(GetRuleRequest getRuleRequest, ActionListener<GetRuleResponse> listener) {
164-
final QueryBuilder getQueryBuilder = queryBuilder.from(getRuleRequest);
165-
getRuleFromIndex(getRuleRequest.getId(), getQueryBuilder, getRuleRequest.getSearchAfter(), listener);
159+
try (ThreadContext.StoredContext context = stashContext()) {
160+
final QueryBuilder getQueryBuilder = queryBuilder.from(getRuleRequest);
161+
getRuleFromIndex(getRuleRequest.getId(), getQueryBuilder, getRuleRequest.getSearchAfter(), listener);
162+
}
166163
}
167164

168165
/**
@@ -173,22 +170,19 @@ public void getRule(GetRuleRequest getRuleRequest, ActionListener<GetRuleRespons
173170
* @param listener - ActionListener for GetRuleResponse
174171
*/
175172
private void getRuleFromIndex(String id, QueryBuilder queryBuilder, String searchAfter, ActionListener<GetRuleResponse> listener) {
176-
// Stash the current thread context when interacting with system index to perform
177-
// operations as the system itself, bypassing authorization checks. This ensures that
178-
// actions within this block are trusted and executed with system-level privileges.
179-
try (ThreadContext.StoredContext context = getContext()) {
173+
try {
180174
SearchRequestBuilder searchRequest = client.prepareSearch(indexName).setQuery(queryBuilder).setSize(maxRulesPerPage);
181175
if (searchAfter != null) {
182176
searchRequest.addSort(_ID_STRING, SortOrder.ASC).searchAfter(new Object[] { searchAfter });
183177
}
184-
searchRequest.execute(ActionListener.wrap(searchResponse -> {
185-
List<SearchHit> hits = Arrays.asList(searchResponse.getHits().getHits());
186-
if (hasNoResults(id, listener, hits)) return;
187-
handleGetRuleResponse(hits, listener);
188-
}, e -> {
189-
logger.error("Failed to fetch all rules: {}", e.getMessage());
190-
listener.onFailure(e);
191-
}));
178+
179+
SearchResponse searchResponse = searchRequest.get();
180+
List<SearchHit> hits = Arrays.asList(searchResponse.getHits().getHits());
181+
if (hasNoResults(id, listener, hits)) return;
182+
handleGetRuleResponse(hits, listener);
183+
} catch (Exception e) {
184+
logger.error("Failed to fetch all rules: {}", e.getMessage());
185+
listener.onFailure(e);
192186
}
193187
}
194188

@@ -214,7 +208,7 @@ void handleGetRuleResponse(List<SearchHit> hits, ActionListener<GetRuleResponse>
214208

215209
@Override
216210
public void deleteRule(DeleteRuleRequest request, ActionListener<AcknowledgedResponse> listener) {
217-
try (ThreadContext.StoredContext context = getContext()) {
211+
try (ThreadContext.StoredContext context = stashContext()) {
218212
DeleteRequest deleteRequest = new DeleteRequest(indexName).id(request.getRuleId());
219213
client.delete(deleteRequest, ActionListener.wrap(deleteResponse -> {
220214
boolean acknowledged = deleteResponse.getResult() == DocWriteResponse.Result.DELETED;
@@ -241,7 +235,7 @@ public String getIndexName() {
241235
return indexName;
242236
}
243237

244-
private ThreadContext.StoredContext getContext() {
238+
private ThreadContext.StoredContext stashContext() {
245239
return client.threadPool().getThreadContext().stashContext();
246240
}
247241
}

‎modules/autotagging-commons/common/src/test/java/org/opensearch/rule/RuleUtilsTests.java

Lines changed: 38 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -9,22 +9,23 @@
99
package org.opensearch.rule;
1010

1111
import org.opensearch.rule.autotagging.Rule;
12+
import org.opensearch.rule.autotagging.RuleTests;
1213
import org.opensearch.rule.utils.RuleTestUtils;
1314
import org.opensearch.test.OpenSearchTestCase;
1415

1516
import java.util.Map;
1617
import java.util.Optional;
1718
import java.util.Set;
1819

19-
import static org.opensearch.rule.action.GetRuleRequestTests.ATTRIBUTE_VALUE_ONE;
20-
import static org.opensearch.rule.action.GetRuleRequestTests.ATTRIBUTE_VALUE_TWO;
21-
import static org.opensearch.rule.action.GetRuleRequestTests.DESCRIPTION_ONE;
22-
import static org.opensearch.rule.action.GetRuleRequestTests.FEATURE_VALUE_ONE;
23-
import static org.opensearch.rule.action.GetRuleRequestTests.TIMESTAMP_ONE;
24-
import static org.opensearch.rule.action.GetRuleRequestTests._ID_ONE;
25-
import static org.opensearch.rule.action.GetRuleRequestTests._ID_TWO;
26-
import static org.opensearch.rule.action.GetRuleRequestTests.ruleTwo;
2720
import static org.opensearch.rule.action.GetRuleResponseTests.ruleOne;
21+
import static org.opensearch.rule.utils.RuleTestUtils.ATTRIBUTE_VALUE_ONE;
22+
import static org.opensearch.rule.utils.RuleTestUtils.ATTRIBUTE_VALUE_TWO;
23+
import static org.opensearch.rule.utils.RuleTestUtils.DESCRIPTION_ONE;
24+
import static org.opensearch.rule.utils.RuleTestUtils.FEATURE_VALUE_ONE;
25+
import static org.opensearch.rule.utils.RuleTestUtils.TIMESTAMP_ONE;
26+
import static org.opensearch.rule.utils.RuleTestUtils._ID_ONE;
27+
import static org.opensearch.rule.utils.RuleTestUtils._ID_TWO;
28+
import static org.opensearch.rule.utils.RuleTestUtils.ruleTwo;
2829

2930
public class RuleUtilsTests extends OpenSearchTestCase {
3031

@@ -34,12 +35,12 @@ public void testDuplicateRuleFound() {
3435
assertEquals(_ID_ONE, result.get());
3536
}
3637

37-
public void testNoDuplicate_NoAttributeIntersection() {
38+
public void testNoAttributeIntersection() {
3839
Optional<String> result = RuleUtils.getDuplicateRuleId(ruleOne, Map.of(_ID_TWO, ruleTwo));
3940
assertTrue(result.isEmpty());
4041
}
4142

42-
public void testNoDuplicate_AttributeSizeMismatch() {
43+
public void testAttributeSizeMismatch() {
4344
Rule testRule = Rule.builder()
4445
.description(DESCRIPTION_ONE)
4546
.featureType(RuleTestUtils.MockRuleFeatureType.INSTANCE)
@@ -57,4 +58,31 @@ public void testNoDuplicate_AttributeSizeMismatch() {
5758
Optional<String> result = RuleUtils.getDuplicateRuleId(ruleOne, Map.of(_ID_TWO, testRule));
5859
assertTrue(result.isEmpty());
5960
}
61+
62+
public void testPartialAttributeValueIntersection() {
63+
Rule ruleWithPartialOverlap = Rule.builder()
64+
.description(DESCRIPTION_ONE)
65+
.featureType(RuleTestUtils.MockRuleFeatureType.INSTANCE)
66+
.featureValue(FEATURE_VALUE_ONE)
67+
.attributeMap(Map.of(RuleTestUtils.MockRuleAttributes.MOCK_RULE_ATTRIBUTE_ONE, Set.of(ATTRIBUTE_VALUE_ONE, "extra_value")))
68+
.updatedAt(TIMESTAMP_ONE)
69+
.build();
70+
71+
Optional<String> result = RuleUtils.getDuplicateRuleId(ruleWithPartialOverlap, Map.of(_ID_ONE, ruleOne));
72+
assertTrue(result.isPresent());
73+
assertEquals(_ID_ONE, result.get());
74+
}
75+
76+
public void testDifferentFeatureTypes() {
77+
Rule differentFeatureTypeRule = Rule.builder()
78+
.description(DESCRIPTION_ONE)
79+
.featureType(RuleTests.TestFeatureType.INSTANCE)
80+
.featureValue(FEATURE_VALUE_ONE)
81+
.attributeMap(RuleTests.ATTRIBUTE_MAP)
82+
.updatedAt(TIMESTAMP_ONE)
83+
.build();
84+
85+
Optional<String> result = RuleUtils.getDuplicateRuleId(differentFeatureTypeRule, Map.of(_ID_ONE, ruleOne));
86+
assertTrue(result.isEmpty());
87+
}
6088
}

‎modules/autotagging-commons/common/src/test/java/org/opensearch/rule/action/CreateRuleRequestTests.java

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -15,8 +15,8 @@
1515

1616
import java.io.IOException;
1717

18-
import static org.opensearch.rule.action.GetRuleRequestTests.assertEqualRule;
19-
import static org.opensearch.rule.action.GetRuleRequestTests.ruleOne;
18+
import static org.opensearch.rule.utils.RuleTestUtils.assertEqualRule;
19+
import static org.opensearch.rule.utils.RuleTestUtils.ruleOne;
2020

2121
public class CreateRuleRequestTests extends OpenSearchTestCase {
2222

‎modules/autotagging-commons/common/src/test/java/org/opensearch/rule/action/CreateRuleResponseTests.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,9 +20,9 @@
2020
import java.io.IOException;
2121
import java.util.Map;
2222

23-
import static org.opensearch.rule.action.GetRuleRequestTests.assertEqualRules;
2423
import static org.opensearch.rule.action.GetRuleResponseTests.ruleOne;
2524
import static org.opensearch.rule.utils.RuleTestUtils._ID_ONE;
25+
import static org.opensearch.rule.utils.RuleTestUtils.assertEqualRules;
2626
import static org.mockito.Mockito.mock;
2727

2828
public class CreateRuleResponseTests extends OpenSearchTestCase {

‎modules/autotagging-commons/common/src/test/java/org/opensearch/rule/action/DeleteRuleRequestTests.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,7 @@
1616

1717
import java.io.IOException;
1818

19-
import static org.opensearch.rule.action.GetRuleRequestTests._ID_ONE;
19+
import static org.opensearch.rule.utils.RuleTestUtils._ID_ONE;
2020

2121
public class DeleteRuleRequestTests extends OpenSearchTestCase {
2222

‎modules/autotagging-commons/common/src/test/java/org/opensearch/rule/action/GetRuleRequestTests.java

Lines changed: 4 additions & 63 deletions
Original file line numberDiff line numberDiff line change
@@ -11,15 +11,15 @@
1111
import org.opensearch.common.io.stream.BytesStreamOutput;
1212
import org.opensearch.core.common.io.stream.StreamInput;
1313
import org.opensearch.rule.GetRuleRequest;
14-
import org.opensearch.rule.autotagging.Attribute;
15-
import org.opensearch.rule.autotagging.Rule;
1614
import org.opensearch.rule.utils.RuleTestUtils;
1715
import org.opensearch.test.OpenSearchTestCase;
1816

1917
import java.io.IOException;
2018
import java.util.HashMap;
21-
import java.util.Map;
22-
import java.util.Set;
19+
20+
import static org.opensearch.rule.utils.RuleTestUtils.ATTRIBUTE_MAP;
21+
import static org.opensearch.rule.utils.RuleTestUtils.SEARCH_AFTER;
22+
import static org.opensearch.rule.utils.RuleTestUtils._ID_ONE;
2323

2424
public class GetRuleRequestTests extends OpenSearchTestCase {
2525
/**
@@ -64,63 +64,4 @@ public void testValidate() {
6464
request = new GetRuleRequest(_ID_ONE, ATTRIBUTE_MAP, "", RuleTestUtils.MockRuleFeatureType.INSTANCE);
6565
assertThrows(IllegalArgumentException.class, request::validate);
6666
}
67-
68-
public static final String _ID_ONE = "id_1";
69-
public static final String SEARCH_AFTER = "search_after";
70-
public static final String _ID_TWO = "G5iIq84j7eK1qIAAAAIH53=1";
71-
public static final String FEATURE_VALUE_ONE = "feature_value_one";
72-
public static final String FEATURE_VALUE_TWO = "feature_value_two";
73-
public static final String ATTRIBUTE_VALUE_ONE = "mock_attribute_one";
74-
public static final String ATTRIBUTE_VALUE_TWO = "mock_attribute_two";
75-
public static final String DESCRIPTION_ONE = "description_1";
76-
public static final String DESCRIPTION_TWO = "description_2";
77-
public static final String TIMESTAMP_ONE = "2024-01-26T08:58:57.558Z";
78-
public static final String TIMESTAMP_TWO = "2023-01-26T08:58:57.558Z";
79-
public static final Map<Attribute, Set<String>> ATTRIBUTE_MAP = Map.of(
80-
RuleTestUtils.MockRuleAttributes.MOCK_RULE_ATTRIBUTE_ONE,
81-
Set.of(ATTRIBUTE_VALUE_ONE)
82-
);
83-
84-
public static final Rule ruleOne = Rule.builder()
85-
.description(DESCRIPTION_ONE)
86-
.featureType(RuleTestUtils.MockRuleFeatureType.INSTANCE)
87-
.featureValue(FEATURE_VALUE_ONE)
88-
.attributeMap(ATTRIBUTE_MAP)
89-
.updatedAt(TIMESTAMP_ONE)
90-
.build();
91-
92-
public static final Rule ruleTwo = Rule.builder()
93-
.description(DESCRIPTION_TWO)
94-
.featureType(RuleTestUtils.MockRuleFeatureType.INSTANCE)
95-
.featureValue(FEATURE_VALUE_TWO)
96-
.attributeMap(Map.of(RuleTestUtils.MockRuleAttributes.MOCK_RULE_ATTRIBUTE_TWO, Set.of(ATTRIBUTE_VALUE_TWO)))
97-
.updatedAt(TIMESTAMP_TWO)
98-
.build();
99-
100-
public static Map<String, Rule> ruleMap() {
101-
return Map.of(_ID_ONE, ruleOne, _ID_TWO, ruleTwo);
102-
}
103-
104-
public static void assertEqualRules(Map<String, Rule> mapOne, Map<String, Rule> mapTwo, boolean ruleUpdated) {
105-
assertEquals(mapOne.size(), mapTwo.size());
106-
for (Map.Entry<String, Rule> entry : mapOne.entrySet()) {
107-
String id = entry.getKey();
108-
assertTrue(mapTwo.containsKey(id));
109-
Rule one = mapOne.get(id);
110-
Rule two = mapTwo.get(id);
111-
assertEqualRule(one, two, ruleUpdated);
112-
}
113-
}
114-
115-
public static void assertEqualRule(Rule one, Rule two, boolean ruleUpdated) {
116-
if (ruleUpdated) {
117-
assertEquals(one.getDescription(), two.getDescription());
118-
assertEquals(one.getFeatureType(), two.getFeatureType());
119-
assertEquals(one.getFeatureValue(), two.getFeatureValue());
120-
assertEquals(one.getAttributeMap(), two.getAttributeMap());
121-
assertEquals(one.getAttributeMap(), two.getAttributeMap());
122-
} else {
123-
assertEquals(one, two);
124-
}
125-
}
12667
}

0 commit comments

Comments
 (0)