Skip to content

Significantly Lower Monitoring HttpExport Memory Footprint #48854

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

Merged
merged 3 commits into from
Nov 11, 2019
Merged

Significantly Lower Monitoring HttpExport Memory Footprint #48854

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
c9b6800
Significantly Lower Monitoring HttpExport Memory Footprint
original-brownbear Nov 4, 2019
0dd10fc
no assertion error
original-brownbear Nov 11, 2019
a0eee7f
Merge remote-tracking branch 'elastic/master' into improve-monitoring…
original-brownbear Nov 11, 2019
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
80 changes: 38 additions & 42 deletions ...toring/src/main/java/org/elasticsearch/xpack/monitoring/exporter/http/HttpExportBulk.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,24 +6,22 @@
package org.elasticsearch.xpack.monitoring.exporter.http;

import org.apache.http.entity.ContentType;
import org.apache.http.nio.entity.NByteArrayEntity;
import org.apache.http.entity.InputStreamEntity;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.apache.logging.log4j.message.ParameterizedMessage;
import org.apache.logging.log4j.util.Supplier;
import org.apache.lucene.util.BytesRef;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.client.Request;
import org.elasticsearch.client.Response;
import org.elasticsearch.client.ResponseListener;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.common.bytes.BytesReference;
import org.elasticsearch.common.io.stream.BytesStreamOutput;
import org.elasticsearch.common.io.stream.StreamOutput;
import org.elasticsearch.common.time.DateFormatter;
import org.elasticsearch.common.util.concurrent.ThreadContext;
import org.elasticsearch.common.xcontent.ToXContent;
import org.elasticsearch.common.xcontent.XContent;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentHelper;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.xpack.core.monitoring.exporter.MonitoringDoc;
import org.elasticsearch.xpack.core.monitoring.exporter.MonitoringTemplateUtils;
Expand Down Expand Up @@ -60,7 +58,7 @@ class HttpExportBulk extends ExportBulk {
/**
* The bytes payload that represents the bulk body is created via {@link #doAdd(Collection)}.
*/
private byte[] payload = null;
private BytesReference payload = null;

HttpExportBulk(final String name, final RestClient client, final Map<String, String> parameters,
final DateFormatter dateTimeFormatter, final ThreadContext threadContext) {
Expand All @@ -77,12 +75,11 @@ public void doAdd(Collection<MonitoringDoc> docs) throws ExportException {
if (docs != null && docs.isEmpty() == false) {
try (BytesStreamOutput payload = new BytesStreamOutput()) {
for (MonitoringDoc monitoringDoc : docs) {
// any failure caused by an individual doc will be written as an empty byte[], thus not impacting the rest
payload.write(toBulkBytes(monitoringDoc));
writeDocument(monitoringDoc, payload);
}

// store the payload until we flush
this.payload = BytesReference.toBytes(payload.bytes());
this.payload = payload.bytes();
}
}
} catch (Exception e) {
Expand All @@ -94,12 +91,19 @@ public void doAdd(Collection<MonitoringDoc> docs) throws ExportException {
public void doFlush(ActionListener<Void> listener) throws ExportException {
if (payload == null) {
listener.onFailure(new ExportException("unable to send documents because none were loaded for export bulk [{}]", name));
} else if (payload.length != 0) {
} else if (payload.length() != 0) {
final Request request = new Request("POST", "/_bulk");
for (Map.Entry<String, String> param : params.entrySet()) {
request.addParameter(param.getKey(), param.getValue());
}
request.setEntity(new NByteArrayEntity(payload, ContentType.APPLICATION_JSON));
try {
request.setEntity(new InputStreamEntity(payload.streamInput(), payload.length(), ContentType.APPLICATION_JSON));
} catch (IOException e) {
listener.onFailure(e);
return;
}
// null out serialized docs to make things easier on the GC
payload = null;

client.performRequestAsync(request, new ResponseListener() {
@Override
Expand All @@ -123,51 +127,43 @@ public void onFailure(Exception exception) {
}
}

private byte[] toBulkBytes(final MonitoringDoc doc) throws IOException {
private void writeDocument(MonitoringDoc doc, StreamOutput out) throws IOException {
final XContentType xContentType = XContentType.JSON;
final XContent xContent = xContentType.xContent();

final String index = MonitoringTemplateUtils.indexName(formatter, doc.getSystem(), doc.getTimestamp());
final String id = doc.getId();

try (BytesStreamOutput out = new BytesStreamOutput()) {
try (XContentBuilder builder = new XContentBuilder(xContent, out)) {
// Builds the bulk action metadata line
builder.startObject();
try (XContentBuilder builder = new XContentBuilder(xContent, out)) {
// Builds the bulk action metadata line
builder.startObject();
{
builder.startObject("index");
{
builder.startObject("index");
{
builder.field("_index", index);
if (id != null) {
builder.field("_id", id);
}
builder.field("_index", index);
if (id != null) {
builder.field("_id", id);
}
builder.endObject();
}
builder.endObject();
}
builder.endObject();
}

// Adds action metadata line bulk separator
out.write(xContent.streamSeparator());

// Adds the source of the monitoring document
final BytesRef source = XContentHelper.toXContent(doc, xContentType, false).toBytesRef();
out.write(source.bytes, source.offset, source.length);

// Adds final bulk separator
out.write(xContent.streamSeparator());
// Adds action metadata line bulk separator
out.write(xContent.streamSeparator());

logger.trace(
"http exporter [{}] - added index request [index={}, id={}, monitoring data type={}]",
name, index, id, doc.getType()
);
// Adds the source of the monitoring document
try (XContentBuilder builder = new XContentBuilder(xContent, out)) {
doc.toXContent(builder, ToXContent.EMPTY_PARAMS);
}

return BytesReference.toBytes(out.bytes());
} catch (Exception e) {
logger.warn((Supplier<?>) () -> new ParameterizedMessage("failed to render document [{}], skipping it [{}]", doc, name), e);
Copy link
Contributor Author

@original-brownbear original-brownbear Nov 4, 2019

Choose a reason for hiding this comment

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

This is hard to model with streaming IO, but to me catch and appending an empty byte array seems bogus here and not worth retaining (there's no test covering this case and we should simply make sure out documents that we have full control over serialise shouldn't we?).

// Adds final bulk separator
out.write(xContent.streamSeparator());

return BytesRef.EMPTY_BYTES;
}
logger.trace(
"http exporter [{}] - added index request [index={}, id={}, monitoring data type={}]",
name, index, id, doc.getType()
);
}

}