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

W-13716428 filter events after the queue #591

Merged
merged 7 commits into from
Aug 3, 2023
Merged

W-13716428 filter events after the queue #591

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
857fe01
filter events after the queue
valeriaares Aug 1, 2023
6dc8037
filter events after the queue
valeriaares Aug 1, 2023
2382f29
fix tests
valeriaares Aug 2, 2023
71a5473
turn value to boolean instead of String
valeriaares Aug 2, 2023
0f57253
pass name and id
valeriaares Aug 2, 2023
89f13d3
Boolean to boolean
valeriaares Aug 3, 2023
630ad8b
Merge branch 'master' into W-13716428-filter-events-on-queue-drain
valeriaares Aug 3, 2023
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
35 changes: 33 additions & 2 deletions .../timbermill-local/src/main/java/com/datorama/oss/timbermill/common/ElasticsearchUtil.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
import java.time.temporal.ChronoUnit;
import java.util.*;
import java.util.concurrent.BlockingQueue;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

import static com.datorama.oss.timbermill.TaskIndexer.logErrorInEventsMap;
Expand Down Expand Up @@ -304,16 +305,23 @@ public class ElasticsearchUtil {
private static final String OLD_SUFFIX = "old";

private static final Set<String> envsSet = Sets.newConcurrentHashSet();
private static final Pattern metadataPatten = Pattern.compile("metadata.*");
private static Pattern notToSkipRegexPattern;

public static Set<String> getEnvSet() {
return envsSet;
}

public static void drainAndIndex(BlockingQueue<Event> eventsQueue, TaskIndexer taskIndexer, int maxElement) {
drainAndIndex(eventsQueue, taskIndexer, maxElement, "false", ".*");
}

public static void drainAndIndex(BlockingQueue<Event> eventsQueue, TaskIndexer taskIndexer, int maxElement, String skipEventsAtDrainFlag, String notToSkipRegex) {
valeriaares marked this conversation as resolved.
Show resolved Hide resolved
while (!eventsQueue.isEmpty()) {
try {
Collection<Event> events = new ArrayList<>();
eventsQueue.drainTo(events, maxElement);
Collection<Event> unfilteredEvents = new ArrayList<>();
eventsQueue.drainTo(unfilteredEvents, maxElement);
Collection<Event> events = filterEvents(unfilteredEvents, skipEventsAtDrainFlag, notToSkipRegex);
KamonConstants.MESSAGES_IN_INPUT_QUEUE_RANGE_SAMPLER.withoutTags().decrement(events.size());
logErrorInEventsMap(events.stream().filter(event -> event.getTaskId() != null).collect(Collectors.groupingBy(Event::getTaskId)), "drainAndIndex");

Expand Down Expand Up @@ -346,6 +354,29 @@ public static void drainAndIndex(BlockingQueue<Event> eventsQueue, TaskIndexer t
}
}

private static Collection<Event> filterEvents(Collection<Event> unfilteredEvents, String skipEventsAtDrainFlag, String notToSkipRegex) {
if (Boolean.parseBoolean(skipEventsAtDrainFlag)) {
return unfilteredEvents.stream()
.filter(event-> shouldKeep(event, notToSkipRegex))
valeriaares marked this conversation as resolved.
Show resolved Hide resolved
valeriaares marked this conversation as resolved.
Show resolved Hide resolved
.collect(Collectors.toList());
}
return unfilteredEvents;
}

private static Boolean shouldKeep(Event event, String notToSkipRegex) {
if (metadataPatten.matcher(event.getName()).matches()) {
valeriaares marked this conversation as resolved.
Show resolved Hide resolved
return true;
}
if (notToSkipRegexPattern == null) {
notToSkipRegexPattern = Pattern.compile(notToSkipRegex);
}
boolean match = notToSkipRegexPattern.matcher(event.getName()).matches();
if (match) {
LOG.info("skipEvents | keeping task {} task id: {} at drain", event.getName(), event.getTaskId());
}
return match;
}

public static long getTimesDuration(ZonedDateTime taskIndexerStartTime, ZonedDateTime taskIndexerEndTime) {
return ChronoUnit.MILLIS.between(taskIndexerStartTime, taskIndexerEndTime);
}
Expand Down
15 changes: 9 additions & 6 deletions ...ermill-server/src/main/java/com/datorama/timbermill/server/service/TimbermillService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,8 @@ public class TimbermillService {
private static Pattern notToSkipRegexPattern = null;
private static Pattern metadataPatten = Pattern.compile("metadata.*");

private String skipEventsFlag;
private String skipEventsAtInsertFlag;
private String skipEventsAtDrainFlag;
private String notToSkipRegex;


Expand Down Expand Up @@ -113,15 +114,17 @@ public TimbermillService(@Value("${INDEX_BULK_SIZE:200000}") Integer indexBulkSi
@Value("${LIMIT_FOR_PERIOD:30000}") int limitForPeriod,
@Value("${LIMIT_REFRESH_PERIOD_MINUTES:1}") int limitRefreshPeriod,
@Value("${RATE_LIMITER_CAPACITY:1000000}") int rateLimiterCapacity,
@Value("${skip.events.flag:false}") String skipEventsFlag,
@Value("${skip.events.at.insert.flag:false}") String skipEventsAtInsertFlag,
valeriaares marked this conversation as resolved.
Show resolved Hide resolved
@Value("${skip.events.at.drain.flag:false}") String skipEventsAtDrainFlag,
valeriaares marked this conversation as resolved.
Show resolved Hide resolved
@Value("${not.to.skip.events.regex:.*}") String notToSkipRegex){


eventsQueue = new LinkedBlockingQueue<>(eventsQueueCapacity);
overflowedQueue = new LinkedBlockingQueue<>(overFlowedQueueCapacity);
terminationTimeout = terminationTimeoutSeconds * 1000;

this.skipEventsFlag = skipEventsFlag;
this.skipEventsAtInsertFlag = skipEventsAtInsertFlag;
this.skipEventsAtDrainFlag = skipEventsAtDrainFlag;
this.notToSkipRegex = notToSkipRegex;

RedisService redisService = null;
Expand Down Expand Up @@ -168,7 +171,7 @@ private void startWorkingThread() {
Thread workingThread = new Thread(() -> {
LOG.info("Timbermill has started");
while (keepRunning) {
ElasticsearchUtil.drainAndIndex(eventsQueue, taskIndexer, eventsMaxElement);
ElasticsearchUtil.drainAndIndex(eventsQueue, taskIndexer, eventsMaxElement, skipEventsAtDrainFlag, notToSkipRegex);
}
stoppedRunning = true;
});
Expand Down Expand Up @@ -211,7 +214,7 @@ void handleEvents(Collection<Event> events) {
}

private Boolean shouldKeep(Event event) {
if (Boolean.parseBoolean(skipEventsFlag)) {
if (Boolean.parseBoolean(skipEventsAtInsertFlag)) {
if (metadataPatten.matcher(event.getName()).matches()) {
return true;
}
Expand All @@ -220,7 +223,7 @@ private Boolean shouldKeep(Event event) {
}
boolean match = notToSkipRegexPattern.matcher(event.getName()).matches();
if (match) {
LOG.info("skipEvents | keeping task {} task id: {}", event.getName(), event.getTaskId());
LOG.info("skipEvents | keeping task {} task id: {} at insert", event.getName(), event.getTaskId());
}
return match;
}
Expand Down
3 changes: 2 additions & 1 deletion timbermill-java/timbermill-server/src/main/resources/application.properties
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,5 +9,6 @@ management.health.elasticsearch.enabled=false
management.endpoint.health.show-details=always
PLUGINS_JSON=[{"class":"SwitchCasePlugin","taskMatcher":{"name":"Eventplugin"},"searchField":"exception","outputAttribute":"errorType","switchCase":[{"match":["TOO_MANY_SERVER_ROWS"],"output":"TOO_MANY_SERVER_ROWS"}]}]

#skip.events.flag = true
#skip.events.at.insert.flag = true
#skip.events.at.drain.flag = false
#not.to.skip.events.regex = "(account_analytics.*)|(page_view.*)|(validate_login)|(last_workspace_update)|(interactive_dingo_query)|(widget_init)|(widget_rendered)|(page_load)|(iframe_route_states_log)|(workspace_analytics_average_query_load_time)|(top_workspace_analytics_orphan_count)|(workspace_analytics_orphan_count)"
8 changes: 4 additions & 4 deletions ...ll-server/src/test/java/com/datorama/timbermill/server/service/TimbermillServiceTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,7 +56,7 @@ public void init() {
@TimberLogTask(name = EVENT)
public void testHandleEventsOneMatchingTask() {

ReflectionTestUtils.setField(service, "skipEventsFlag", "true");
ReflectionTestUtils.setField(service, "skipEventsAtInsertFlag", "true");
ReflectionTestUtils.setField(service, "notToSkipRegex", ".*keep.*");

Event event = new StartEvent(TimberLogger.getCurrentTaskId(), KEEP, new LogParams(), null);
Expand All @@ -68,7 +68,7 @@ public void testHandleEventsOneMatchingTask() {
@TimberLogTask(name = EVENT)
public void testHandleEventsOneNotMatchingTask() {

ReflectionTestUtils.setField(service, "skipEventsFlag", "true");
ReflectionTestUtils.setField(service, "skipEventsAtInsertFlag", "true");
ReflectionTestUtils.setField(service, "notToSkipRegex", ".*keep.*");

Event event = new StartEvent(TimberLogger.getCurrentTaskId(), SKIP, new LogParams(), null);
Expand All @@ -82,7 +82,7 @@ public void testHandleEventsOneNotMatchingTask() {
public void testHandleEventsMultipleTasks() {


ReflectionTestUtils.setField(service, "skipEventsFlag", "true");
ReflectionTestUtils.setField(service, "skipEventsAtInsertFlag", "true");
ReflectionTestUtils.setField(service, "notToSkipRegex", ".keep.*");

Event eventToKeep1 = new StartEvent(TimberLogger.getCurrentTaskId(), KEEP + "1", new LogParams(), null);
Expand All @@ -103,7 +103,7 @@ public void testHandleEventsMultipleTasks() {
@TimberLogTask(name = EVENT)
public void testHandleEventsMultipleTasksFlagClosed() {

ReflectionTestUtils.setField(service, "skipEventsFlag", "false");
ReflectionTestUtils.setField(service, "skipEventsAtInsertFlag", "false");
ReflectionTestUtils.setField(service, "notToSkipRegex", ".keep.*");

Event eventToKeep1 = new StartEvent(TimberLogger.getCurrentTaskId(), KEEP + "1", new LogParams(), null);
Expand Down
Loading