Skip to content

Commit

Permalink
Remove IP2Geo processor validation (#336)
Browse files Browse the repository at this point in the history 
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 <[email protected]>
  • Loading branch information
@heemin32
heemin32 authored Jun 16, 2023
1 parent 63e3a47 commit a04eeeb
Show file tree
Hide file tree
Showing 3 changed files with 23 additions and 105 deletions.
72 changes: 19 additions & 53 deletions src/main/java/org/opensearch/geospatial/ip2geo/processor/Ip2GeoProcessor.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,8 +4,6 @@
*/
package org.opensearch.geospatial.ip2geo.processor;

import static org.opensearch.cluster.service.ClusterApplierService.CLUSTER_UPDATE_THREAD_NAME;
import static org.opensearch.ingest.ConfigurationUtils.newConfigurationException;
import static org.opensearch.ingest.ConfigurationUtils.readBooleanProperty;
import static org.opensearch.ingest.ConfigurationUtils.readOptionalList;
import static org.opensearch.ingest.ConfigurationUtils.readStringProperty;
Expand All @@ -29,6 +27,7 @@
import org.opensearch.geospatial.ip2geo.common.DatasourceState;
import org.opensearch.geospatial.ip2geo.common.GeoIpDataFacade;
import org.opensearch.geospatial.ip2geo.jobscheduler.Datasource;
import org.opensearch.index.IndexNotFoundException;
import org.opensearch.ingest.AbstractProcessor;
import org.opensearch.ingest.IngestDocument;
import org.opensearch.ingest.IngestService;
Expand Down Expand Up @@ -154,7 +153,12 @@ protected void executeInternal(
@Override
public void onResponse(final Datasource datasource) {
if (datasource == null) {
handler.accept(null, new IllegalStateException("datasource does not exist"));
handler.accept(null, new IllegalStateException("datasource is not available"));
return;
}

if (DatasourceState.AVAILABLE.equals(datasource.getState()) == false) {
handler.accept(null, new IllegalStateException("datasource is not in an available state"));
return;
}

Expand All @@ -174,6 +178,10 @@ public void onResponse(final Datasource datasource) {

@Override
public void onFailure(final Exception e) {
if (e instanceof IndexNotFoundException) {
handler.accept(null, new IllegalStateException("datasource is not available"));
return;
}
handler.accept(null, e);
}
});
Expand Down Expand Up @@ -241,8 +249,8 @@ protected void executeInternal(
datasourceFacade.getDatasource(datasourceName, new ActionListener<>() {
@Override
public void onResponse(final Datasource datasource) {
if (datasource == null) {
handler.accept(null, new IllegalStateException("datasource does not exist"));
if (datasource == null || DatasourceState.AVAILABLE.equals(datasource.getState()) == false) {
handler.accept(null, new IllegalStateException("datasource is not available"));
return;
}

Expand All @@ -262,6 +270,10 @@ public void onResponse(final Datasource datasource) {

@Override
public void onFailure(final Exception e) {
if (e instanceof IndexNotFoundException) {
handler.accept(null, new IllegalStateException("datasource is not available"));
return;
}
handler.accept(null, e);
}
});
Expand Down Expand Up @@ -319,15 +331,8 @@ public Factory(final IngestService ingestService, final DatasourceFacade datasou
}

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

// Skip validation for the call by cluster applier service
if (Thread.currentThread().getName().contains(CLUSTER_UPDATE_THREAD_NAME) == false) {
validate(processorTag, datasourceName, propertyNames);
}

return new Ip2GeoProcessor(
processorTag,
description,
Expand All @@ -360,39 +360,5 @@ public Ip2GeoProcessor create(
geoIpDataFacade
);
}

private void validate(final String processorTag, final String datasourceName, final List<String> propertyNames) throws IOException {
Datasource datasource = datasourceFacade.getDatasource(datasourceName);

if (datasource == null) {
throw newConfigurationException(TYPE, processorTag, "datasource", "datasource [" + datasourceName + "] doesn't exist");
}

if (DatasourceState.AVAILABLE.equals(datasource.getState()) == false) {
throw newConfigurationException(
TYPE,
processorTag,
"datasource",
"datasource [" + datasourceName + "] is not in an available state"
);
}

if (propertyNames == null) {
return;
}

// Validate properties are valid. If not add all available properties.
final Set<String> availableProperties = new HashSet<>(datasource.getDatabase().getFields());
for (String fieldName : propertyNames) {
if (availableProperties.contains(fieldName) == false) {
throw newConfigurationException(
TYPE,
processorTag,
"properties",
"property [" + fieldName + "] is not available in the datasource [" + datasourceName + "]"
);
}
}
}
}
}
16 changes: 0 additions & 16 deletions src/test/java/org/opensearch/geospatial/ip2geo/processor/Ip2GeoProcessorIT.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,6 @@
import java.io.IOException;
import java.time.Duration;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Locale;
Expand All @@ -35,21 +34,6 @@ public class Ip2GeoProcessorIT extends GeospatialRestTestCase {
private static final String IP = "ip";
private static final String SOURCE = "_source";

@SneakyThrows
public void testCreateIp2GeoProcessor_whenNoSuchDatasourceExist_thenFails() {
String pipelineName = PREFIX + GeospatialTestHelper.randomLowerCaseString();

// Run
ResponseException exception = expectThrows(
ResponseException.class,
() -> createIp2GeoProcessorPipeline(pipelineName, Collections.emptyMap())
);

// Verify
assertTrue(exception.getMessage().contains("doesn't exist"));
assertEquals(RestStatus.BAD_REQUEST.getStatus(), exception.getResponse().getStatusLine().getStatusCode());
}

@SneakyThrows
public void testCreateIp2GeoProcessor_whenValidInput_thenAddData() {
Ip2GeoDataServer.start();
Expand Down
40 changes: 4 additions & 36 deletions src/test/java/org/opensearch/geospatial/ip2geo/processor/Ip2GeoProcessorTests.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,6 @@

import org.junit.Before;
import org.mockito.ArgumentCaptor;
import org.opensearch.OpenSearchException;
import org.opensearch.action.ActionListener;
import org.opensearch.common.Randomness;
import org.opensearch.geospatial.GeospatialTestHelper;
Expand All @@ -46,40 +45,6 @@ public void init() {
factory = new Ip2GeoProcessor.Factory(ingestService, datasourceFacade, geoIpDataFacade);
}

public void testCreateWithNoDatasource() {
Map<String, Object> config = new HashMap<>();
config.put("field", "ip");
config.put(CONFIG_DATASOURCE_KEY, "no_datasource");
OpenSearchException exception = expectThrows(
OpenSearchException.class,
() -> factory.create(
Collections.emptyMap(),
GeospatialTestHelper.randomLowerCaseString(),
GeospatialTestHelper.randomLowerCaseString(),
config
)
);
assertTrue(exception.getDetailedMessage().contains("doesn't exist"));
}

public void testCreateWithInvalidDatasourceState() {
Datasource datasource = new Datasource();
datasource.setName(GeospatialTestHelper.randomLowerCaseString());
datasource.setState(randomStateExcept(DatasourceState.AVAILABLE));
OpenSearchException exception = expectThrows(OpenSearchException.class, () -> createProcessor(datasource, Collections.emptyMap()));
assertTrue(exception.getDetailedMessage().contains("available state"));
}

public void testCreateIp2GeoProcessor_whenInvalidProperties_thenException() {
Map<String, Object> config = new HashMap<>();
config.put("properties", Arrays.asList(SUPPORTED_FIELDS.get(0), "invalid_property"));
OpenSearchException exception = expectThrows(
OpenSearchException.class,
() -> createProcessor(GeospatialTestHelper.randomLowerCaseString(), config)
);
assertTrue(exception.getDetailedMessage().contains("property"));
}

public void testExecuteWithNoIpAndIgnoreMissing() throws Exception {
String datasourceName = GeospatialTestHelper.randomLowerCaseString();
Map<String, Object> config = new HashMap<>();
Expand Down Expand Up @@ -125,13 +90,14 @@ public void testExecute_whenNonStringValue_thenException() throws Exception {
public void testExecuteWithNullDatasource() throws Exception {
BiConsumer<IngestDocument, Exception> handler = (doc, e) -> {
assertNull(doc);
assertTrue(e.getMessage().contains("datasource does not exist"));
assertTrue(e.getMessage().contains("datasource is not available"));
};
getActionListener(Collections.emptyMap(), handler).onResponse(null);
}

public void testExecuteWithExpiredDatasource() throws Exception {
Datasource datasource = mock(Datasource.class);
when(datasource.getState()).thenReturn(DatasourceState.AVAILABLE);
when(datasource.isExpired()).thenReturn(true);
BiConsumer<IngestDocument, Exception> handler = (doc, e) -> {
assertEquals("ip2geo_data_expired", doc.getFieldValue(DEFAULT_TARGET_FIELD + ".error", String.class));
Expand Down Expand Up @@ -174,6 +140,7 @@ public void testExecuteInternal_whenSingleIp_thenGetDatasourceIsCalled() {
verify(datasourceFacade).getDatasource(eq(datasourceName), captor.capture());
Datasource datasource = mock(Datasource.class);
when(datasource.isExpired()).thenReturn(false);
when(datasource.getState()).thenReturn(DatasourceState.AVAILABLE);
when(datasource.currentIndexName()).thenReturn(GeospatialTestHelper.randomLowerCaseString());

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

captor.getValue().onResponse(datasource);
Expand Down

0 comments on commit a04eeeb

Please sign in to comment.