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

Cache the creation of parsers within DateProcessor #92880

Merged
merged 5 commits into from
Jan 31, 2023
Merged

Cache the creation of parsers within DateProcessor #92880

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
37d1239
transform into a constant the `DateFormatter`
HiDAl Jan 13, 2023
0426778
introduce a cache to DateProcessor
HiDAl Jan 13, 2023
b8281f7
Update docs/changelog/92880.yaml
HiDAl Jan 12, 2023
d3eb4cd
uses `SoftReferences<>` for the cache values
HiDAl Jan 27, 2023
c004397
group cache variables and instances
HiDAl Jan 30, 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
5 changes: 5 additions & 0 deletions docs/changelog/92880.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
pr: 92880
summary: Cache the creation of parsers within DateProcessor
area: Ingest Node
type: enhancement
issues: []
11 changes: 9 additions & 2 deletions modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/DateFormat.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,11 +36,10 @@ enum DateFormat {
@Override
Function<String, ZonedDateTime> getFunction(String format, ZoneId timezone, Locale locale) {
return (date) -> {
TemporalAccessor accessor = DateFormatter.forPattern("iso8601").parse(date);
TemporalAccessor accessor = ISO_8601.parse(date);
// even though locale could be set to en-us, Locale.ROOT (following iso8601 calendar data rules) should be used
return DateFormatters.from(accessor, Locale.ROOT, timezone).withZoneSameInstant(timezone);
};

}
},
Unix {
Expand Down Expand Up @@ -115,6 +114,14 @@ Function<String, ZonedDateTime> getFunction(String format, ZoneId zoneId, Locale
}
};

/** It's important to keep this variable as a constant because {@link DateFormatter#forPattern(String)} is an expensive method and,
* in this case, it's a never changing value.
* <br>
* Also, we shouldn't inline it in the {@link DateFormat#Iso8601}'s enum because it'd make useless the cache used
* at {@link DateProcessor}).
*/
private static final DateFormatter ISO_8601 = DateFormatter.forPattern("iso8601");

abstract Function<String, ZonedDateTime> getFunction(String format, ZoneId timezone, Locale locale);

static DateFormat fromString(String format) {
Expand Down
56 changes: 55 additions & 1 deletion modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/DateProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,8 +9,10 @@
package org.elasticsearch.ingest.common;

import org.elasticsearch.ExceptionsHelper;
import org.elasticsearch.common.settings.SettingsException;
import org.elasticsearch.common.time.DateFormatter;
import org.elasticsearch.common.util.LocaleUtils;
import org.elasticsearch.common.util.concurrent.ConcurrentCollections;
import org.elasticsearch.core.Nullable;
import org.elasticsearch.ingest.AbstractProcessor;
import org.elasticsearch.ingest.ConfigurationUtils;
Expand All @@ -19,18 +21,32 @@
import org.elasticsearch.script.ScriptService;
import org.elasticsearch.script.TemplateScript;

import java.lang.ref.SoftReference;
import java.time.ZoneId;
import java.time.ZoneOffset;
import java.time.ZonedDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.concurrent.ConcurrentMap;
import java.util.function.Function;
import java.util.function.Supplier;

public final class DateProcessor extends AbstractProcessor {

public static final String TYPE = "date";
private static final String CACHE_CAPACITY_SETTING = "es.ingest.date_processor.cache_capacity";
private static final Cache CACHE;

static {
var cacheSizeStr = System.getProperty(CACHE_CAPACITY_SETTING, "256");
Copy link
Member

Choose a reason for hiding this comment

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

What's the reason for doing this as a system property instead of a Settings thing (and related, are we documenting this setting)?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

We don't expect users will have to change this value unless they see a big performance effect on their pipelines. In the opposite case, we wanted to provide an easy way to change the value if we receive an SDH (with the obvious memory consumption trade-off). I inspired by this other PR#80902 by @joegallo

Copy link
Contributor

Choose a reason for hiding this comment

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

+1, git grep 'System.getProperty("es' | grep -vi test/ shows a good number of examples of this as the pattern for non-customer-facing internal knobs that we don't expect to need to twiddle (but that we're making possible to twiddle as a favor to our future selves).

Copy link
Member

Choose a reason for hiding this comment

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

OK, just wanted to make sure it was intentional.

try {
CACHE = new Cache(Integer.parseInt(cacheSizeStr));
} catch (NumberFormatException e) {
throw new SettingsException("{} must be a valid number but was [{}]", CACHE_CAPACITY_SETTING, cacheSizeStr);
}
}
static final String DEFAULT_TARGET_FIELD = "@timestamp";
static final String DEFAULT_OUTPUT_FORMAT = "yyyy-MM-dd'T'HH:mm:ss.SSSXXX";

Expand Down Expand Up @@ -72,9 +88,17 @@ public final class DateProcessor extends AbstractProcessor {
this.targetField = targetField;
this.formats = formats;
this.dateParsers = new ArrayList<>(this.formats.size());

for (String format : formats) {
DateFormat dateFormat = DateFormat.fromString(format);
dateParsers.add((params) -> dateFormat.getFunction(format, newDateTimeZone(params), newLocale(params)));
dateParsers.add((params) -> {
var documentZoneId = newDateTimeZone(params);
var documentLocale = newLocale(params);
return CACHE.getOrCompute(
new Cache.Key(format, documentZoneId, documentLocale),
() -> dateFormat.getFunction(format, documentZoneId, documentLocale)
);
});
}
this.outputFormat = outputFormat;
formatter = DateFormatter.forPattern(this.outputFormat);
Expand Down Expand Up @@ -198,4 +222,34 @@ public DateProcessor create(
);
}
}

static final class Cache {
private final ConcurrentMap<Key, SoftReference<Function<String, ZonedDateTime>>> map;
private final int capacity;

Cache(int capacity) {
if (capacity <= 0) {
throw new IllegalArgumentException("cache capacity must be a value greater than 0 but was " + capacity);
}
this.capacity = capacity;
this.map = ConcurrentCollections.newConcurrentMapWithAggressiveConcurrency(this.capacity);
}

Function<String, ZonedDateTime> getOrCompute(Key key, Supplier<Function<String, ZonedDateTime>> supplier) {
Function<String, ZonedDateTime> fn;
var element = map.get(key);
// element exist and wasn't GCed
if (element != null && (fn = element.get()) != null) {
return fn;
}
if (map.size() >= capacity) {
map.clear();
Copy link
Member

Choose a reason for hiding this comment

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

This is kind of surprising. Whenever the cache is full, we empty it and start over? Is this to avoid the complexity of a concurrent LRU cache? Or is the idea that we'll rarely see more than 256 formats in practice, so this is just for safety and will rarely be hit?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Yes! this decision was made mainly because we want to avoid the complexity of LRU (or any eviction policy) within this code path. The decision to make the size 256 was primarily based on experiments with different benchmarks.

Copy link
Contributor

Choose a reason for hiding this comment

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

I think it's worth adding a one line comment in the code above the clear to explain its purpose.

}
fn = supplier.get();
map.put(key, new SoftReference<>(fn));
return fn;
}

record Key(String format, ZoneId zoneId, Locale locale) {}
}
}
33 changes: 33 additions & 0 deletions modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/DateProcessorTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,9 +24,15 @@
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.function.Function;
import java.util.function.Supplier;

import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.equalTo;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;

public class DateProcessorTests extends ESTestCase {

Expand Down Expand Up @@ -335,4 +341,31 @@ public void testOutputFormat() {
String expectedDate = "00:00:00." + Strings.format("%09d", nanosAfterEpoch);
assertThat(ingestDocument.getFieldValue("date_as_date", String.class), equalTo(expectedDate));
}

@SuppressWarnings("unchecked")
public void testCacheIsEvictedAfterReachMaxCapacity() {
Supplier<Function<String, ZonedDateTime>> supplier1 = mock(Supplier.class);
Supplier<Function<String, ZonedDateTime>> supplier2 = mock(Supplier.class);
Function<String, ZonedDateTime> zonedDateTimeFunction1 = str -> ZonedDateTime.now();
Function<String, ZonedDateTime> zonedDateTimeFunction2 = str -> ZonedDateTime.now();
var cache = new DateProcessor.Cache(1);
var key1 = new DateProcessor.Cache.Key("format-1", ZoneId.systemDefault(), Locale.ROOT);
var key2 = new DateProcessor.Cache.Key("format-2", ZoneId.systemDefault(), Locale.ROOT);

when(supplier1.get()).thenReturn(zonedDateTimeFunction1);
when(supplier2.get()).thenReturn(zonedDateTimeFunction2);

assertEquals(cache.getOrCompute(key1, supplier1), zonedDateTimeFunction1); // 1 call to supplier1
assertEquals(cache.getOrCompute(key2, supplier2), zonedDateTimeFunction2); // 1 call to supplier2
assertEquals(cache.getOrCompute(key1, supplier1), zonedDateTimeFunction1); // 1 more call to supplier1
assertEquals(cache.getOrCompute(key1, supplier1), zonedDateTimeFunction1); // should use cached value
assertEquals(cache.getOrCompute(key2, supplier2), zonedDateTimeFunction2); // 1 more call to supplier2
assertEquals(cache.getOrCompute(key2, supplier2), zonedDateTimeFunction2); // should use cached value
assertEquals(cache.getOrCompute(key2, supplier2), zonedDateTimeFunction2); // should use cached value
assertEquals(cache.getOrCompute(key2, supplier2), zonedDateTimeFunction2); // should use cached value
assertEquals(cache.getOrCompute(key1, supplier1), zonedDateTimeFunction1); // 1 more to call to supplier1

verify(supplier1, times(3)).get();
verify(supplier2, times(2)).get();
}
}