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

Jump to bottom

Remove blocking calls and change threat intel feed flow to event driven #871

Merged
merged 5 commits into from
Mar 2, 2024
Merged

Remove blocking calls and change threat intel feed flow to event driven #871

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
58f9727
remove actionGet() and change threat intel feed flow to event driven
eirsep Feb 16, 2024
5dce731
fix javadocs
eirsep Feb 27, 2024
b20270c
revert try catch removals
eirsep Mar 1, 2024
f222f41
use action listener wrap() in detector threat intel code paths
eirsep Mar 2, 2024
3bfb29d
add try catch
eirsep Mar 2, 2024
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
43 changes: 15 additions & 28 deletions src/main/java/org/opensearch/securityanalytics/threatIntel/DetectorThreatIntelService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,8 +32,6 @@
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import static org.opensearch.securityanalytics.model.Detector.DETECTORS_INDEX;
Expand Down Expand Up @@ -121,35 +119,24 @@
listener.onResponse(Collections.emptyList());
return;
}

CountDownLatch latch = new CountDownLatch(1);
Copy link
Collaborator

Choose a reason for hiding this comment

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

Do we know why the latches were initially implemented? Seems fine to remove them based on the testing performed but I'm puzzled as to why they would have been added in the first place if they are not required

Copy link
Member Author

Choose a reason for hiding this comment

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

bad practice. The right construct to use is a Countdown

Countdown is a  simple thread safe count-down class that in contrast to a CountDownLatch never blocks. This class is useful if a certain action has to wait for N concurrent tasks to return or a timeout to occur in order to proceed.

but safer to do it the event-driven way

threatIntelFeedDataService.getThreatIntelFeedData(new ActionListener<>() {
@Override
public void onResponse(List<ThreatIntelFeedData> threatIntelFeedData) {
if (threatIntelFeedData.isEmpty()) {
listener.onResponse(Collections.emptyList());
} else {
listener.onResponse(
createDocLevelQueriesFromThreatIntelList(iocFieldList, threatIntelFeedData, detector)
);
threatIntelFeedDataService.getThreatIntelFeedData(ActionListener.wrap(

Check warning on line 122 in src/main/java/org/opensearch/securityanalytics/threatIntel/DetectorThreatIntelService.java

View check run for this annotation

Codecov / codecov/patch

src/main/java/org/opensearch/securityanalytics/threatIntel/DetectorThreatIntelService.java#L122 

Added line #L122 was not covered by tests
threatIntelFeedData -> {
if (threatIntelFeedData.isEmpty()) {
listener.onResponse(Collections.emptyList());

Check warning on line 125 in src/main/java/org/opensearch/securityanalytics/threatIntel/DetectorThreatIntelService.java

View check run for this annotation

Codecov / codecov/patch

src/main/java/org/opensearch/securityanalytics/threatIntel/DetectorThreatIntelService.java#L125 

Added line #L125 was not covered by tests
} else {
listener.onResponse(
createDocLevelQueriesFromThreatIntelList(iocFieldList, threatIntelFeedData, detector)

Check warning on line 128 in src/main/java/org/opensearch/securityanalytics/threatIntel/DetectorThreatIntelService.java

View check run for this annotation

Codecov / codecov/patch

src/main/java/org/opensearch/securityanalytics/threatIntel/DetectorThreatIntelService.java#L127-L128 

Added lines #L127 - L128 were not covered by tests
);
}
}, e -> {
log.error("Failed to get threat intel feeds for doc level query creation", e);
listener.onFailure(e);

Check warning on line 133 in src/main/java/org/opensearch/securityanalytics/threatIntel/DetectorThreatIntelService.java

View check run for this annotation

Codecov / codecov/patch

src/main/java/org/opensearch/securityanalytics/threatIntel/DetectorThreatIntelService.java#L131-L133 

Added lines #L131 - L133 were not covered by tests
}
latch.countDown();
}

@Override
public void onFailure(Exception e) {
log.error("Failed to get threat intel feeds for doc level query creation", e);
listener.onFailure(e);
latch.countDown();
}
});

latch.await(30, TimeUnit.SECONDS);
} catch (InterruptedException e) {
log.error("Failed to create doc level queries from threat intel feeds", e);
));
} catch (Exception e) {
log.error("Failed to create doc level query from threat intel data", e);

Check warning on line 137 in src/main/java/org/opensearch/securityanalytics/threatIntel/DetectorThreatIntelService.java

View check run for this annotation

Codecov / codecov/patch

src/main/java/org/opensearch/securityanalytics/threatIntel/DetectorThreatIntelService.java#L136-L137 

Added lines #L136 - L137 were not covered by tests
listener.onFailure(e);
}

}

private static String constructId(Detector detector, String iocType) {
Expand Down
145 changes: 59 additions & 86 deletions src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,12 +34,12 @@
import org.opensearch.core.xcontent.ToXContent;
import org.opensearch.core.xcontent.XContentBuilder;
import org.opensearch.securityanalytics.model.ThreatIntelFeedData;
import org.opensearch.securityanalytics.settings.SecurityAnalyticsSettings;
import org.opensearch.securityanalytics.threatIntel.action.PutTIFJobAction;
import org.opensearch.securityanalytics.threatIntel.action.PutTIFJobRequest;
import org.opensearch.securityanalytics.threatIntel.action.ThreatIntelIndicesResponse;
import org.opensearch.securityanalytics.threatIntel.common.TIFMetadata;
import org.opensearch.securityanalytics.threatIntel.common.StashedThreadContext;
import org.opensearch.securityanalytics.settings.SecurityAnalyticsSettings;
import org.opensearch.securityanalytics.threatIntel.common.TIFMetadata;
import org.opensearch.securityanalytics.threatIntel.jobscheduler.TIFJobParameterService;
import org.opensearch.securityanalytics.util.IndexUtils;
import org.opensearch.securityanalytics.util.SecurityAnalyticsException;
Expand All @@ -56,7 +56,6 @@
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CountDownLatch;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
Expand Down Expand Up @@ -104,21 +103,13 @@
ActionListener<List<ThreatIntelFeedData>> listener
) {
try {
Copy link
Collaborator

Choose a reason for hiding this comment

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

Do we no longer need this top-level try/catch? My observation has been that calls will hang it exceptions are not handled via the ActionListener

Copy link
Member Author

Choose a reason for hiding this comment

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

listener framework is event driven and no catch is required as ActionListener.onFailure() would need to implement whatever logic was written in catch block as callback mechanism will not throw an exception

Copy link
Collaborator

Choose a reason for hiding this comment

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

If this code throws an exception, we never make a call to ActionListener.onFailure(). I believe the original call just hangs in this case, at least that is what I have experienced while developing in this package.

Copy link
Collaborator

Choose a reason for hiding this comment

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

Let me know if I am missing something

Copy link
Member Author

@eirsep eirsep Feb 29, 2024 •
edited
Loading

Choose a reason for hiding this comment

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

true.
reverting try-catch removal


String tifdIndex = getLatestIndexByCreationDate();
if (tifdIndex == null) {
createThreatIntelFeedData(listener);
} else {
SearchRequest searchRequest = new SearchRequest(tifdIndex);
searchRequest.source().size(9999); //TODO: convert to scroll
String finalTifdIndex = tifdIndex;
client.search(searchRequest, ActionListener.wrap(r -> listener.onResponse(ThreatIntelFeedDataUtils.getTifdList(r, xContentRegistry)), e -> {
log.error(String.format(
"Failed to fetch threat intel feed data from system index %s", finalTifdIndex), e);
listener.onFailure(e);
}));
fetchThreatIntelFeedDataFromIndex(tifdIndex, listener);

Check warning on line 110 in src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java

View check run for this annotation

Codecov / codecov/patch

src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java#L110 

Added line #L110 was not covered by tests
}
} catch (InterruptedException e) {
} catch (Exception e) {

Check warning on line 112 in src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java

View check run for this annotation

Codecov / codecov/patch

src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java#L112 

Added line #L112 was not covered by tests
log.error("Failed to get threat intel feed data", e);
listener.onFailure(e);
}
Expand Down Expand Up @@ -150,21 +141,16 @@
.mapping(getIndexMapping()).timeout(clusterSettings.get(SecurityAnalyticsSettings.THREAT_INTEL_TIMEOUT));
StashedThreadContext.run(
client,
() -> client.admin().indices().create(createIndexRequest, new ActionListener<>() {
@Override
public void onResponse(CreateIndexResponse response) {
if (response.isAcknowledged()) {
listener.onResponse(response);
} else {
onFailure(new OpenSearchStatusException("Threat intel feed index creation failed", RestStatus.INTERNAL_SERVER_ERROR));
}
}

@Override
public void onFailure(Exception e) {
listener.onFailure(e);
}
})
() -> client.admin().indices().create(createIndexRequest,
ActionListener.wrap(

Check warning on line 145 in src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java

View check run for this annotation

Codecov / codecov/patch

src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java#L144-L145 

Added lines #L144 - L145 were not covered by tests
response -> {
if (response.isAcknowledged())
listener.onResponse(response);

Check warning on line 148 in src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java

View check run for this annotation

Codecov / codecov/patch

src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java#L148 

Added line #L148 was not covered by tests
else
listener.onFailure(new OpenSearchStatusException("Threat intel feed index creation failed", RestStatus.INTERNAL_SERVER_ERROR));

Check warning on line 150 in src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java

View check run for this annotation

Codecov / codecov/patch

src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java#L150 

Added line #L150 was not covered by tests

}, listener::onFailure

Check warning on line 152 in src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java

View check run for this annotation

Codecov / codecov/patch

src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java#L152 

Added line #L152 was not covered by tests
))
);
}

Expand Down Expand Up @@ -223,28 +209,20 @@
}
bulkRequestList.add(bulkRequest);

GroupedActionListener<BulkResponse> bulkResponseListener = new GroupedActionListener<>(new ActionListener<>() {
@Override
public void onResponse(Collection<BulkResponse> bulkResponses) {
int idx = 0;
for (BulkResponse response: bulkResponses) {
BulkRequest request = bulkRequestList.get(idx);
if (response.hasFailures()) {
throw new OpenSearchException(
"error occurred while ingesting threat intel feed data in {} with an error {}",
StringUtils.join(request.getIndices()),
response.buildFailureMessage()
);
}
GroupedActionListener<BulkResponse> bulkResponseListener = new GroupedActionListener<>(ActionListener.wrap(bulkResponses -> {
int idx = 0;

Check warning on line 213 in src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java

View check run for this annotation

Codecov / codecov/patch

src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java#L212-L213 

Added lines #L212 - L213 were not covered by tests
for (BulkResponse response : bulkResponses) {
BulkRequest request = bulkRequestList.get(idx);

Check warning on line 215 in src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java

View check run for this annotation

Codecov / codecov/patch

src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java#L215 

Added line #L215 was not covered by tests
if (response.hasFailures()) {
throw new OpenSearchException(

Check warning on line 217 in src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java

View check run for this annotation

Codecov / codecov/patch

src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java#L217 

Added line #L217 was not covered by tests
"error occurred while ingesting threat intel feed data in {} with an error {}",
StringUtils.join(request.getIndices()),
response.buildFailureMessage()

Check warning on line 220 in src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java

View check run for this annotation

Codecov / codecov/patch

src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java#L219-L220 

Added lines #L219 - L220 were not covered by tests
);
}
listener.onResponse(new ThreatIntelIndicesResponse(true, List.of(indexName)));
}

@Override
public void onFailure(Exception e) {
listener.onFailure(e);
}
}, bulkRequestList.size());
listener.onResponse(new ThreatIntelIndicesResponse(true, List.of(indexName)));
}, listener::onFailure), bulkRequestList.size());

Check warning on line 225 in src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java

View check run for this annotation

Codecov / codecov/patch

src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java#L224-L225 

Added lines #L224 - L225 were not covered by tests

for (int i = 0; i < bulkRequestList.size(); ++i) {
saveTifds(bulkRequestList.get(i), timeout, bulkResponseListener);
Expand Down Expand Up @@ -291,52 +269,47 @@
.prepareDelete(indices.toArray(new String[0]))
.setIndicesOptions(IndicesOptions.LENIENT_EXPAND_OPEN_CLOSED_HIDDEN)
.setTimeout(clusterSettings.get(SecurityAnalyticsSettings.THREAT_INTEL_TIMEOUT))
.execute(new ActionListener<>() {
@Override
public void onResponse(AcknowledgedResponse response) {
if (response.isAcknowledged() == false) {
onFailure(new OpenSearchException("failed to delete data[{}]", String.join(",", indices)));
}
}

@Override
public void onFailure(Exception e) {
log.error("unknown exception:", e);
}
})
.execute(ActionListener.wrap(

Check warning on line 272 in src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java

View check run for this annotation

Codecov / codecov/patch

src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java#L272 

Added line #L272 was not covered by tests
response -> {
if (response.isAcknowledged() == false) {
log.error(new OpenSearchException("failed to delete threat intel feed index[{}]",
String.join(",", indices)));

Check warning on line 276 in src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java

View check run for this annotation

Codecov / codecov/patch

src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java#L275-L276 

Added lines #L275 - L276 were not covered by tests
}
}, e -> log.error("failed to delete threat intel feed index [{}]", e)

Check warning on line 278 in src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java

View check run for this annotation

Codecov / codecov/patch

src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java#L278 

Added line #L278 was not covered by tests
))
);
}

private void createThreatIntelFeedData(ActionListener<List<ThreatIntelFeedData>> listener) throws InterruptedException {
CountDownLatch countDownLatch = new CountDownLatch(1);
private void createThreatIntelFeedData(ActionListener<List<ThreatIntelFeedData>> listener) {
client.execute(
PutTIFJobAction.INSTANCE,
new PutTIFJobRequest("feed_updater", clusterSettings.get(SecurityAnalyticsSettings.TIF_UPDATE_INTERVAL)),
new ActionListener<>() {
@Override
public void onResponse(AcknowledgedResponse acknowledgedResponse) {
log.debug("Acknowledged threat intel feed updater job created");
countDownLatch.countDown();
String tifdIndex = getLatestIndexByCreationDate();

SearchRequest searchRequest = new SearchRequest(tifdIndex);
searchRequest.source().size(9999); //TODO: convert to scroll
String finalTifdIndex = tifdIndex;
client.search(searchRequest, ActionListener.wrap(r -> listener.onResponse(ThreatIntelFeedDataUtils.getTifdList(r, xContentRegistry)), e -> {
log.error(String.format(
"Failed to fetch threat intel feed data from system index %s", finalTifdIndex), e);
ActionListener.wrap(

Check warning on line 287 in src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java

View check run for this annotation

Codecov / codecov/patch

src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java#L287 

Added line #L287 was not covered by tests
r -> {
if (false == r.isAcknowledged()) {
listener.onFailure(new Exception("Failed to acknowledge Put Tif job action"));
return;

Check warning on line 291 in src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java

View check run for this annotation

Codecov / codecov/patch

src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java#L290-L291 

Added lines #L290 - L291 were not covered by tests
}
log.debug("Acknowledged threat intel feed updater job created");
String tifdIndex = getLatestIndexByCreationDate();
fetchThreatIntelFeedDataFromIndex(tifdIndex, listener);
}, e -> {
log.debug("Failed to create threat intel feed updater job", e);

Check warning on line 297 in src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java

View check run for this annotation

Codecov / codecov/patch

src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java#L293-L297 

Added lines #L293 - L297 were not covered by tests
listener.onFailure(e);
}));
}

@Override
public void onFailure(Exception e) {
log.debug("Failed to create threat intel feed updater job", e);
countDownLatch.countDown();
}
}
}

Check warning on line 299 in src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java

View check run for this annotation

Codecov / codecov/patch

src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java#L299 

Added line #L299 was not covered by tests
)
);
countDownLatch.await();
}

Check warning on line 302 in src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java

View check run for this annotation

Codecov / codecov/patch

src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java#L302 

Added line #L302 was not covered by tests

private void fetchThreatIntelFeedDataFromIndex(String tifdIndex, ActionListener<List<ThreatIntelFeedData>> listener) {
SearchRequest searchRequest = new SearchRequest(tifdIndex);
searchRequest.source().size(9999); //TODO: convert to scroll
String finalTifdIndex = tifdIndex;
client.search(searchRequest, ActionListener.wrap(r -> listener.onResponse(ThreatIntelFeedDataUtils.getTifdList(r, xContentRegistry)), e -> {
log.error(String.format(

Check warning on line 309 in src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java

View check run for this annotation

Codecov / codecov/patch

src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java#L305-L309 

Added lines #L305 - L309 were not covered by tests
"Failed to fetch threat intel feed data from system index %s", finalTifdIndex), e);
listener.onFailure(e);
}));

Check warning on line 312 in src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java

View check run for this annotation

Codecov / codecov/patch

src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java#L311-L312 

Added lines #L311 - L312 were not covered by tests
}

private String getIndexMapping() {
Expand Down
Loading
Loading