-
Notifications
You must be signed in to change notification settings - Fork 69
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.
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 blocking calls and change threat intel feed flow to event driven #871
Changes from 2 commits
58f9727
5dce731
b20270c
f222f41
3bfb29d
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -116,13 +116,10 @@ private String buildQueryStringQueryWithIocList(Set<String> iocs) { | |
* Fetches threat intel data and creates doc level queries from threat intel data | ||
*/ | ||
public void createDocLevelQueryFromThreatIntel(List<LogType.IocFields> iocFieldList, Detector detector, ActionListener<List<DocLevelQuery>> listener) { | ||
try { | ||
if (false == detector.getThreatIntelEnabled() || iocFieldList.isEmpty()) { | ||
listener.onResponse(Collections.emptyList()); | ||
return; | ||
} | ||
|
||
CountDownLatch latch = new CountDownLatch(1); | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe 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 There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. bad practice. The right construct to use is a
but safer to do it the event-driven way |
||
threatIntelFeedDataService.getThreatIntelFeedData(new ActionListener<>() { | ||
@Override | ||
public void onResponse(List<ThreatIntelFeedData> threatIntelFeedData) { | ||
|
@@ -133,23 +130,14 @@ public void onResponse(List<ThreatIntelFeedData> threatIntelFeedData) { | |
createDocLevelQueriesFromThreatIntelList(iocFieldList, threatIntelFeedData, detector) | ||
); | ||
} | ||
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); | ||
listener.onFailure(e); | ||
} | ||
|
||
} | ||
|
||
private static String constructId(Detector detector, String iocType) { | ||
|
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -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; | ||
|
@@ -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; | ||
|
@@ -103,24 +102,11 @@ | |
public void getThreatIntelFeedData( | ||
ActionListener<List<ThreatIntelFeedData>> listener | ||
) { | ||
try { | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe 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 There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. listener framework is event driven and no catch is required as There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. If this code throws an exception, we never make a call to There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Let me know if I am missing something There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. true. |
||
|
||
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); | ||
})); | ||
} | ||
} catch (InterruptedException e) { | ||
log.error("Failed to get threat intel feed data", e); | ||
listener.onFailure(e); | ||
String tifdIndex = getLatestIndexByCreationDate(); | ||
Check warning on line 105 in src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java Codecov / codecov/patchsrc/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java#L105
|
||
if (tifdIndex == null) { | ||
createThreatIntelFeedData(listener); | ||
Check warning on line 107 in src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java Codecov / codecov/patchsrc/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java#L107
|
||
} else { | ||
fetchThreatIntelFeedDataFromIndex(tifdIndex, listener); | ||
Check warning on line 109 in src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java Codecov / codecov/patchsrc/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java#L109
|
||
} | ||
} | ||
|
||
|
@@ -307,36 +293,35 @@ | |
); | ||
} | ||
|
||
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); | ||
listener.onFailure(e); | ||
})); | ||
fetchThreatIntelFeedDataFromIndex(tifdIndex, listener); | ||
Check warning on line 305 in src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java Codecov / codecov/patchsrc/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java#L305
|
||
} | ||
|
||
@Override | ||
public void onFailure(Exception e) { | ||
log.debug("Failed to create threat intel feed updater job", e); | ||
countDownLatch.countDown(); | ||
} | ||
} | ||
); | ||
countDownLatch.await(); | ||
} | ||
Check warning on line 314 in src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java Codecov / codecov/patchsrc/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java#L314
|
||
|
||
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 321 in src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java Codecov / codecov/patchsrc/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java#L317-L321
|
||
"Failed to fetch threat intel feed data from system index %s", finalTifdIndex), e); | ||
listener.onFailure(e); | ||
})); | ||
Check warning on line 324 in src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java Codecov / codecov/patchsrc/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java#L323-L324
|
||
} | ||
|
||
private String getIndexMapping() { | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Similar try/catch comment as below
There was a problem hiding this comment.
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