Skip to content

Commit 110db37

Browse files
heemin32heemin32
committed
Remove IP2Geo processor validation
Cannot query index to get data to validate IP2Geo processor. Will add validation when we decide to store some of data in cluster state metadata. Signed-off-by: Heemin Kim <heemin@amazon.com>
1 parent e082a6f commit 110db37

File tree

2 files changed

+19
-90
lines changed

2 files changed

+19
-90
lines changed

‎src/main/java/org/opensearch/geospatial/ip2geo/processor/Ip2GeoProcessor.java‎

Lines changed: 15 additions & 54 deletions
Original file line numberDiff line numberDiff line change
@@ -4,8 +4,6 @@
44
*/
55
package org.opensearch.geospatial.ip2geo.processor;
66

7-
import static org.opensearch.cluster.service.ClusterApplierService.CLUSTER_UPDATE_THREAD_NAME;
8-
import static org.opensearch.ingest.ConfigurationUtils.newConfigurationException;
97
import static org.opensearch.ingest.ConfigurationUtils.readBooleanProperty;
108
import static org.opensearch.ingest.ConfigurationUtils.readOptionalList;
119
import static org.opensearch.ingest.ConfigurationUtils.readStringProperty;
@@ -29,6 +27,7 @@
2927
import org.opensearch.geospatial.ip2geo.common.DatasourceState;
3028
import org.opensearch.geospatial.ip2geo.common.GeoIpDataFacade;
3129
import org.opensearch.geospatial.ip2geo.jobscheduler.Datasource;
30+
import org.opensearch.index.IndexNotFoundException;
3231
import org.opensearch.ingest.AbstractProcessor;
3332
import org.opensearch.ingest.IngestDocument;
3433
import org.opensearch.ingest.IngestService;
@@ -153,8 +152,8 @@ protected void executeInternal(
153152
datasourceFacade.getDatasource(datasourceName, new ActionListener<>() {
154153
@Override
155154
public void onResponse(final Datasource datasource) {
156-
if (datasource == null) {
157-
handler.accept(null, new IllegalStateException("datasource does not exist"));
155+
if (datasource == null || DatasourceState.AVAILABLE.equals(datasource.getState()) == false) {
156+
handler.accept(null, new IllegalStateException("datasource is not available"));
158157
return;
159158
}
160159

@@ -174,6 +173,10 @@ public void onResponse(final Datasource datasource) {
174173

175174
@Override
176175
public void onFailure(final Exception e) {
176+
if (e instanceof IndexNotFoundException) {
177+
handler.accept(null, new IllegalStateException("datasource is not available"));
178+
return;
179+
}
177180
handler.accept(null, e);
178181
}
179182
});
@@ -241,8 +244,8 @@ protected void executeInternal(
241244
datasourceFacade.getDatasource(datasourceName, new ActionListener<>() {
242245
@Override
243246
public void onResponse(final Datasource datasource) {
244-
if (datasource == null) {
245-
handler.accept(null, new IllegalStateException("datasource does not exist"));
247+
if (datasource == null || DatasourceState.AVAILABLE.equals(datasource.getState()) == false) {
248+
handler.accept(null, new IllegalStateException("datasource is not available"));
246249
return;
247250
}
248251

@@ -262,6 +265,10 @@ public void onResponse(final Datasource datasource) {
262265

263266
@Override
264267
public void onFailure(final Exception e) {
268+
if (e instanceof IndexNotFoundException) {
269+
handler.accept(null, new IllegalStateException("datasource is not available"));
270+
return;
271+
}
265272
handler.accept(null, e);
266273
}
267274
});
@@ -319,15 +326,8 @@ public Factory(final IngestService ingestService, final DatasourceFacade datasou
319326
}
320327

321328
/**
322-
* When a user create a processor, this method is called twice. Once to validate the new processor and another
323-
* to apply cluster state change after the processor is added.
324-
*
325-
* The second call is made by ClusterApplierService. Therefore, we cannot access cluster state in the call.
326-
* That means, we cannot even query an index inside the call.
327-
*
328-
* Because the processor is validated in the first call, we skip the validation in the second call.
329-
*
330-
* @see org.opensearch.cluster.service.ClusterApplierService#state()
329+
* Within this method, blocking request cannot be called because this method is executed in a transport thread.
330+
* This means, validation using data in an index won't work.
331331
*/
332332
@Override
333333
public Ip2GeoProcessor create(
@@ -342,11 +342,6 @@ public Ip2GeoProcessor create(
342342
List<String> propertyNames = readOptionalList(TYPE, processorTag, config, CONFIG_PROPERTIES);
343343
boolean ignoreMissing = readBooleanProperty(TYPE, processorTag, config, CONFIG_IGNORE_MISSING, false);
344344

345-
// Skip validation for the call by cluster applier service
346-
if (Thread.currentThread().getName().contains(CLUSTER_UPDATE_THREAD_NAME) == false) {
347-
validate(processorTag, datasourceName, propertyNames);
348-
}
349-
350345
return new Ip2GeoProcessor(
351346
processorTag,
352347
description,
@@ -360,39 +355,5 @@ public Ip2GeoProcessor create(
360355
geoIpDataFacade
361356
);
362357
}
363-
364-
private void validate(final String processorTag, final String datasourceName, final List<String> propertyNames) throws IOException {
365-
Datasource datasource = datasourceFacade.getDatasource(datasourceName);
366-
367-
if (datasource == null) {
368-
throw newConfigurationException(TYPE, processorTag, "datasource", "datasource [" + datasourceName + "] doesn't exist");
369-
}
370-
371-
if (DatasourceState.AVAILABLE.equals(datasource.getState()) == false) {
372-
throw newConfigurationException(
373-
TYPE,
374-
processorTag,
375-
"datasource",
376-
"datasource [" + datasourceName + "] is not in an available state"
377-
);
378-
}
379-
380-
if (propertyNames == null) {
381-
return;
382-
}
383-
384-
// Validate properties are valid. If not add all available properties.
385-
final Set<String> availableProperties = new HashSet<>(datasource.getDatabase().getFields());
386-
for (String fieldName : propertyNames) {
387-
if (availableProperties.contains(fieldName) == false) {
388-
throw newConfigurationException(
389-
TYPE,
390-
processorTag,
391-
"properties",
392-
"property [" + fieldName + "] is not available in the datasource [" + datasourceName + "]"
393-
);
394-
}
395-
}
396-
}
397358
}
398359
}

‎src/test/java/org/opensearch/geospatial/ip2geo/processor/Ip2GeoProcessorTests.java‎

Lines changed: 4 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,6 @@
2525

2626
import org.junit.Before;
2727
import org.mockito.ArgumentCaptor;
28-
import org.opensearch.OpenSearchException;
2928
import org.opensearch.action.ActionListener;
3029
import org.opensearch.common.Randomness;
3130
import org.opensearch.geospatial.GeospatialTestHelper;
@@ -46,40 +45,6 @@ public void init() {
4645
factory = new Ip2GeoProcessor.Factory(ingestService, datasourceFacade, geoIpDataFacade);
4746
}
4847

49-
public void testCreateWithNoDatasource() {
50-
Map<String, Object> config = new HashMap<>();
51-
config.put("field", "ip");
52-
config.put(CONFIG_DATASOURCE_KEY, "no_datasource");
53-
OpenSearchException exception = expectThrows(
54-
OpenSearchException.class,
55-
() -> factory.create(
56-
Collections.emptyMap(),
57-
GeospatialTestHelper.randomLowerCaseString(),
58-
GeospatialTestHelper.randomLowerCaseString(),
59-
config
60-
)
61-
);
62-
assertTrue(exception.getDetailedMessage().contains("doesn't exist"));
63-
}
64-
65-
public void testCreateWithInvalidDatasourceState() {
66-
Datasource datasource = new Datasource();
67-
datasource.setName(GeospatialTestHelper.randomLowerCaseString());
68-
datasource.setState(randomStateExcept(DatasourceState.AVAILABLE));
69-
OpenSearchException exception = expectThrows(OpenSearchException.class, () -> createProcessor(datasource, Collections.emptyMap()));
70-
assertTrue(exception.getDetailedMessage().contains("available state"));
71-
}
72-
73-
public void testCreateIp2GeoProcessor_whenInvalidProperties_thenException() {
74-
Map<String, Object> config = new HashMap<>();
75-
config.put("properties", Arrays.asList(SUPPORTED_FIELDS.get(0), "invalid_property"));
76-
OpenSearchException exception = expectThrows(
77-
OpenSearchException.class,
78-
() -> createProcessor(GeospatialTestHelper.randomLowerCaseString(), config)
79-
);
80-
assertTrue(exception.getDetailedMessage().contains("property"));
81-
}
82-
8348
public void testExecuteWithNoIpAndIgnoreMissing() throws Exception {
8449
String datasourceName = GeospatialTestHelper.randomLowerCaseString();
8550
Map<String, Object> config = new HashMap<>();
@@ -125,13 +90,14 @@ public void testExecute_whenNonStringValue_thenException() throws Exception {
12590
public void testExecuteWithNullDatasource() throws Exception {
12691
BiConsumer<IngestDocument, Exception> handler = (doc, e) -> {
12792
assertNull(doc);
128-
assertTrue(e.getMessage().contains("datasource does not exist"));
93+
assertTrue(e.getMessage().contains("datasource is not available"));
12994
};
13095
getActionListener(Collections.emptyMap(), handler).onResponse(null);
13196
}
13297

13398
public void testExecuteWithExpiredDatasource() throws Exception {
13499
Datasource datasource = mock(Datasource.class);
100+
when(datasource.getState()).thenReturn(DatasourceState.AVAILABLE);
135101
when(datasource.isExpired()).thenReturn(true);
136102
BiConsumer<IngestDocument, Exception> handler = (doc, e) -> {
137103
assertEquals("ip2geo_data_expired", doc.getFieldValue(DEFAULT_TARGET_FIELD + ".error", String.class));
@@ -174,6 +140,7 @@ public void testExecuteInternal_whenSingleIp_thenGetDatasourceIsCalled() {
174140
verify(datasourceFacade).getDatasource(eq(datasourceName), captor.capture());
175141
Datasource datasource = mock(Datasource.class);
176142
when(datasource.isExpired()).thenReturn(false);
143+
when(datasource.getState()).thenReturn(DatasourceState.AVAILABLE);
177144
when(datasource.currentIndexName()).thenReturn(GeospatialTestHelper.randomLowerCaseString());
178145

179146
captor.getValue().onResponse(datasource);
@@ -301,6 +268,7 @@ public void testExecuteInternal_whenMultiIps_thenGetDatasourceIsCalled() {
301268
verify(datasourceFacade).getDatasource(eq(datasourceName), captor.capture());
302269
Datasource datasource = mock(Datasource.class);
303270
when(datasource.isExpired()).thenReturn(false);
271+
when(datasource.getState()).thenReturn(DatasourceState.AVAILABLE);
304272
when(datasource.currentIndexName()).thenReturn(GeospatialTestHelper.randomLowerCaseString());
305273

306274
captor.getValue().onResponse(datasource);

0 commit comments

Comments
 (0)