Skip to content
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

Remove IP2Geo processor validation #336

Merged
merged 1 commit into from
Jun 16, 2023
Merged

Remove IP2Geo processor validation #336

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
72 changes: 19 additions & 53 deletions src/main/java/org/opensearch/geospatial/ip2geo/processor/Ip2GeoProcessor.java
View file
Open in desktop
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) {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

is check on exception class enough here, should we also check on a message to minimize chance of false positive?

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Check on exception class is enough. No other cases. It is only for error message display so not a big issue even if there is false positive.

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
Open in desktop
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
Open in desktop
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