This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch CAMEL-26172-splunk-hec-body in repository https://gitbox.apache.org/repos/asf/camel.git
commit 775e6b8713abbb6271312d70973db3a45dd78f43 Author: Claus Ibsen <[email protected]> AuthorDate: Mon Nov 2 23:30:16 2026 +0100 CAMEL-26172: camel-splunk-hec - send bytes, stream and file bodies as text, and write the JSON before sending The producer wrote the payload with Jackson inside an EntityTemplate, while the request was already being sent: a payload Jackson could not write (such as a stream body) failed half way after a truncated, invalid JSON request had reached Splunk. The JSON is now written to a byte array first. The body was put in the event as is, so Jackson wrote a byte array as base64, a File as its path and failed on a stream or the GenericFile of the file consumer. These bodies are now converted to text. Co-Authored-By: Claude Opus 5.5 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../camel/catalog/components/splunk-hec.json | 2 +- .../camel/catalog/docs/splunk-hec-component.adoc | 4 + .../camel/component/splunkhec/splunk-hec.json | 2 +- .../src/main/docs/splunk-hec-component.adoc | 4 + .../component/splunkhec/SplunkHECEndpoint.java | 7 +- .../component/splunkhec/SplunkHECProducer.java | 36 ++++++-- .../splunkhec/SplunkHECProducerHttpTest.java | 99 ++++++++++++++++++++++ .../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 9 ++ 8 files changed, 149 insertions(+), 14 deletions(-) diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/splunk-hec.json b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/splunk-hec.json index e1c35db33902..be2855f4497c 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/splunk-hec.json +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/splunk-hec.json @@ -33,7 +33,7 @@ "CamelSplunkHECIndexTime": { "index": 0, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "Long", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Epoch-formatted time. Specify with the time query string parameter. Sets a default for all events in the request. The default time can be overridden.", "constantName": "org.apache.camel.component.splunkhec.SplunkHECConstants#INDEX_TIME" } }, "bodyTypes": { - "producerIn": { "carriers": [ "text", "value", "map", "list", "pojo", "null" ], "byOption": [ { "option": "headersOnly", "value": "true", "carriers": [ "any" ] } ], "description": "The event, written as JSON by Jackson with the headers beside it, or alone with bodyOnly. A stream fails, bytes become base64 and a File its path: convert them to text first. With headersOnly the body is not sent" }, + "producerIn": { "carriers": [ "text", "bytes", "stream", "file", "value", "map", "list", "pojo", "null" ], "byOption": [ { "option": "headersOnly", "value": "true", "carriers": [ "any" ] } ], "description": "The event, written as JSON by Jackson with the headers beside it, or alone with bodyOnly. Bytes, a stream or a file are sent as text. With headersOnly the body is not sent" }, "producerOut": { "carriers": [ "unchanged" ] } }, "properties": { diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/splunk-hec-component.adoc b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/splunk-hec-component.adoc index 1baf659cc83b..b9847eb189c7 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/splunk-hec-component.adoc +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/splunk-hec-component.adoc @@ -48,6 +48,10 @@ The body must be serializable to JSON, so it may be sent to Splunk. A `String` or a `java.util.Map` object can be serialized easily. +A byte array, an `InputStream`, a `Reader` or a file (such as the body of the file consumer) is converted to text +and sent as a JSON string. The payload is written as JSON before the request is sent, so a body that cannot be +serialized fails the exchange without sending anything to Splunk. + == Use Cases The Splunk HEC endpoint may be used to stream data to Splunk for ingestion. diff --git a/components/camel-splunk-hec/src/generated/resources/META-INF/org/apache/camel/component/splunkhec/splunk-hec.json b/components/camel-splunk-hec/src/generated/resources/META-INF/org/apache/camel/component/splunkhec/splunk-hec.json index e1c35db33902..be2855f4497c 100644 --- a/components/camel-splunk-hec/src/generated/resources/META-INF/org/apache/camel/component/splunkhec/splunk-hec.json +++ b/components/camel-splunk-hec/src/generated/resources/META-INF/org/apache/camel/component/splunkhec/splunk-hec.json @@ -33,7 +33,7 @@ "CamelSplunkHECIndexTime": { "index": 0, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "Long", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Epoch-formatted time. Specify with the time query string parameter. Sets a default for all events in the request. The default time can be overridden.", "constantName": "org.apache.camel.component.splunkhec.SplunkHECConstants#INDEX_TIME" } }, "bodyTypes": { - "producerIn": { "carriers": [ "text", "value", "map", "list", "pojo", "null" ], "byOption": [ { "option": "headersOnly", "value": "true", "carriers": [ "any" ] } ], "description": "The event, written as JSON by Jackson with the headers beside it, or alone with bodyOnly. A stream fails, bytes become base64 and a File its path: convert them to text first. With headersOnly the body is not sent" }, + "producerIn": { "carriers": [ "text", "bytes", "stream", "file", "value", "map", "list", "pojo", "null" ], "byOption": [ { "option": "headersOnly", "value": "true", "carriers": [ "any" ] } ], "description": "The event, written as JSON by Jackson with the headers beside it, or alone with bodyOnly. Bytes, a stream or a file are sent as text. With headersOnly the body is not sent" }, "producerOut": { "carriers": [ "unchanged" ] } }, "properties": { diff --git a/components/camel-splunk-hec/src/main/docs/splunk-hec-component.adoc b/components/camel-splunk-hec/src/main/docs/splunk-hec-component.adoc index 1baf659cc83b..b9847eb189c7 100644 --- a/components/camel-splunk-hec/src/main/docs/splunk-hec-component.adoc +++ b/components/camel-splunk-hec/src/main/docs/splunk-hec-component.adoc @@ -48,6 +48,10 @@ The body must be serializable to JSON, so it may be sent to Splunk. A `String` or a `java.util.Map` object can be serialized easily. +A byte array, an `InputStream`, a `Reader` or a file (such as the body of the file consumer) is converted to text +and sent as a JSON string. The payload is written as JSON before the request is sent, so a body that cannot be +serialized fails the exchange without sending anything to Splunk. + == Use Cases The Splunk HEC endpoint may be used to stream data to Splunk for ingestion. diff --git a/components/camel-splunk-hec/src/main/java/org/apache/camel/component/splunkhec/SplunkHECEndpoint.java b/components/camel-splunk-hec/src/main/java/org/apache/camel/component/splunkhec/SplunkHECEndpoint.java index 122744454079..bd17dd6961ae 100644 --- a/components/camel-splunk-hec/src/main/java/org/apache/camel/component/splunkhec/SplunkHECEndpoint.java +++ b/components/camel-splunk-hec/src/main/java/org/apache/camel/component/splunkhec/SplunkHECEndpoint.java @@ -47,11 +47,12 @@ import org.apache.commons.validator.routines.InetAddressValidator; syntax = "splunk-hec:splunkURL", category = { Category.MONITORING }, headersClass = SplunkHECConstants.class) @BodyTypes(producerIn = @BodyType(value = { - Carrier.TEXT, Carrier.VALUE, Carrier.MAP, Carrier.LIST, Carrier.POJO, Carrier.NULL }, + Carrier.TEXT, Carrier.BYTES, Carrier.STREAM, Carrier.FILE, Carrier.VALUE, Carrier.MAP, Carrier.LIST, Carrier.POJO, + Carrier.NULL }, byOption = @BodyType.When(option = "headersOnly", value = "true", carriers = Carrier.ANY), description = "The event, written as JSON by Jackson with the headers beside it, or alone with" - + " bodyOnly. A stream fails, bytes become base64 and a File its path: convert" - + " them to text first. With headersOnly the body is not sent"), + + " bodyOnly. Bytes, a stream or a file are sent as text. With headersOnly the body" + + " is not sent"), producerOut = @BodyType(Carrier.UNCHANGED)) public class SplunkHECEndpoint extends DefaultEndpoint implements EndpointServiceLocation { diff --git a/components/camel-splunk-hec/src/main/java/org/apache/camel/component/splunkhec/SplunkHECProducer.java b/components/camel-splunk-hec/src/main/java/org/apache/camel/component/splunkhec/SplunkHECProducer.java index 390a06d3e6c2..e1ea84014123 100644 --- a/components/camel-splunk-hec/src/main/java/org/apache/camel/component/splunkhec/SplunkHECProducer.java +++ b/components/camel-splunk-hec/src/main/java/org/apache/camel/component/splunkhec/SplunkHECProducer.java @@ -17,7 +17,12 @@ package org.apache.camel.component.splunkhec; import java.io.ByteArrayOutputStream; +import java.io.File; +import java.io.InputStream; +import java.io.Reader; +import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; +import java.nio.file.Path; import java.util.HashMap; import java.util.Map; @@ -25,8 +30,10 @@ import javax.net.ssl.SSLContext; import com.fasterxml.jackson.databind.ObjectMapper; import org.apache.camel.Exchange; +import org.apache.camel.InvalidPayloadException; import org.apache.camel.Message; import org.apache.camel.RuntimeCamelException; +import org.apache.camel.WrappedFile; import org.apache.camel.support.DefaultProducer; import org.apache.hc.client5.http.classic.methods.HttpPost; import org.apache.hc.client5.http.impl.classic.CloseableHttpClient; @@ -38,7 +45,7 @@ import org.apache.hc.client5.http.ssl.NoopHostnameVerifier; import org.apache.hc.client5.http.ssl.SSLConnectionSocketFactory; import org.apache.hc.core5.http.ContentType; import org.apache.hc.core5.http.config.RegistryBuilder; -import org.apache.hc.core5.http.io.entity.EntityTemplate; +import org.apache.hc.core5.http.io.entity.ByteArrayEntity; import org.apache.hc.core5.http.io.entity.EntityUtils; import org.apache.hc.core5.http.message.StatusLine; import org.apache.hc.core5.ssl.SSLContextBuilder; @@ -102,10 +109,9 @@ public class SplunkHECProducer extends DefaultProducer { + endpoint.getSplunkURL() + endpoint.getConfiguration().getSplunkEndpoint()); httppost.addHeader("Authorization", " Splunk " + endpoint.getConfiguration().getToken()); - EntityTemplate entityTemplate = new EntityTemplate( - -1, ContentType.APPLICATION_JSON, null, outputStream -> MAPPER.writer().writeValue(outputStream, payload)); - - httppost.setEntity(entityTemplate); + // write the JSON before sending, so a payload that cannot be written fails before any request is made + byte[] json = MAPPER.writeValueAsBytes(payload); + httppost.setEntity(new ByteArrayEntity(json, ContentType.APPLICATION_JSON)); httpClient.execute( httppost, response -> { @@ -130,18 +136,17 @@ public class SplunkHECProducer extends DefaultProducer { } } - Map<String, Object> createPayload(Message message) { - Object body = message.getBody(); + Map<String, Object> createPayload(Message message) throws InvalidPayloadException { Map<String, Object> payload = new HashMap<>(); buildPayload(message, payload); if (endpoint.getConfiguration().isBodyOnly()) { - payload.put("event", body); + payload.put("event", eventBody(message)); } else if (endpoint.getConfiguration().isHeadersOnly()) { payload.put("event", message.getHeaders()); } else { Map<String, Object> eventPayload = new HashMap<>(); - eventPayload.put("body", body); + eventPayload.put("body", eventBody(message)); eventPayload.put("headers", message.getHeaders()); payload.put("event", eventPayload); } @@ -149,6 +154,19 @@ public class SplunkHECProducer extends DefaultProducer { return payload; } + /** + * The body to write as JSON in the event. Jackson writes a byte array as base64 and a File as its path, and fails + * half way on a stream, so these bodies (and the file of the file consumer) are converted to text. + */ + private static Object eventBody(Message message) throws InvalidPayloadException { + Object body = message.getBody(); + if (body instanceof byte[] || body instanceof InputStream || body instanceof Reader || body instanceof File + || body instanceof Path || body instanceof WrappedFile || body instanceof ByteBuffer) { + return message.getMandatoryBody(String.class); + } + return body; + } + private void buildPayload(Message message, Map<String, Object> payload) { if (endpoint.getConfiguration().getSourceType() != null) { payload.put("sourcetype", endpoint.getConfiguration().getSourceType()); diff --git a/components/camel-splunk-hec/src/test/java/org/apache/camel/component/splunkhec/SplunkHECProducerHttpTest.java b/components/camel-splunk-hec/src/test/java/org/apache/camel/component/splunkhec/SplunkHECProducerHttpTest.java index d8aba5146a8b..bbef42f40da0 100644 --- a/components/camel-splunk-hec/src/test/java/org/apache/camel/component/splunkhec/SplunkHECProducerHttpTest.java +++ b/components/camel-splunk-hec/src/test/java/org/apache/camel/component/splunkhec/SplunkHECProducerHttpTest.java @@ -16,22 +16,33 @@ */ package org.apache.camel.component.splunkhec; +import java.io.ByteArrayInputStream; import java.net.InetSocketAddress; import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; import java.util.List; import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.atomic.AtomicInteger; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; import com.sun.net.httpserver.HttpServer; import org.apache.camel.Exchange; import org.apache.camel.Producer; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.file.GenericFile; import org.apache.camel.support.jsse.SSLContextParameters; import org.apache.camel.test.junit6.CamelTestSupport; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -42,14 +53,23 @@ public class SplunkHECProducerHttpTest extends CamelTestSupport { private static final String TOKEN = "11111111-1111-1111-1111-111111111111"; + private static final ObjectMapper MAPPER = new ObjectMapper(); + private final List<String> requests = new CopyOnWriteArrayList<>(); + private final List<String> bodies = new CopyOnWriteArrayList<>(); + private final AtomicInteger arrived = new AtomicInteger(); private HttpServer server; + @TempDir + Path tempDir; + @BeforeEach void startServer() throws Exception { server = HttpServer.create(new InetSocketAddress("localhost", 0), 0); server.createContext("/", httpExchange -> { + arrived.incrementAndGet(); String body = new String(httpExchange.getRequestBody().readAllBytes(), StandardCharsets.UTF_8); + bodies.add(body); requests.add(httpExchange.getRequestMethod() + " " + httpExchange.getRequestURI().getPath() + " " + httpExchange.getRequestHeaders().getFirst("Authorization") + " " + body); byte[] answer = "{\"text\":\"Success\",\"code\":0}".getBytes(StandardCharsets.UTF_8); @@ -89,6 +109,85 @@ public class SplunkHECProducerHttpTest extends CamelTestSupport { assertTrue(requests.get(0).contains("\"event\":\"hello\""), requests.get(0)); } + @Test + void testStreamBody() throws Exception { + template.sendBody(uri("&bodyOnly=true"), new ByteArrayInputStream("hello stream".getBytes(StandardCharsets.UTF_8))); + + assertEquals("hello stream", event().asText()); + } + + @Test + void testStreamBodyWithHeaders() throws Exception { + template.sendBodyAndHeader(uri(""), new ByteArrayInputStream("hello stream".getBytes(StandardCharsets.UTF_8)), + "foo", "bar"); + + JsonNode event = event(); + assertEquals("hello stream", event.get("body").asText()); + assertEquals("bar", event.get("headers").get("foo").asText()); + } + + @Test + void testBytesBody() throws Exception { + template.sendBody(uri("&bodyOnly=true"), "hello bytes".getBytes(StandardCharsets.UTF_8)); + + assertEquals("hello bytes", event().asText()); + } + + @Test + void testFileBody() throws Exception { + Path file = Files.writeString(tempDir.resolve("event.txt"), "hello file"); + + template.sendBody(uri("&bodyOnly=true"), file.toFile()); + template.sendBody(uri("&bodyOnly=true"), file); + + assertEquals(2, bodies.size()); + assertEquals("hello file", MAPPER.readTree(bodies.get(0)).get("event").asText()); + assertEquals("hello file", MAPPER.readTree(bodies.get(1)).get("event").asText()); + } + + @Test + void testFileConsumerBody() throws Exception { + Files.writeString(tempDir.resolve("event.txt"), "hello file consumer"); + context.addRoutes(new RouteBuilder() { + @Override + public void configure() { + from("direct:file") + .pollEnrich("file:" + tempDir.toAbsolutePath() + "?noop=true", 5000) + .to(uri("&bodyOnly=true")); + } + }); + + Exchange exchange = template.send("direct:file", e -> { + }); + + assertNull(exchange.getException()); + assertInstanceOf(GenericFile.class, exchange.getMessage().getBody()); + assertEquals("hello file consumer", event().asText()); + } + + @Test + void testBodyNotWrittenAsJsonFailsBeforeSending() throws Exception { + // Jackson cannot write a plain Object: it fails after it has written the long text before it, + // and this must fail before any request reaches Splunk + List<Object> body = List.of("x".repeat(20000), new Object()); + Exchange exchange = template.send(uri("&bodyOnly=true"), e -> e.getIn().setBody(body)); + assertNotNull(exchange.getException()); + + // the server handles the requests one by one, so once this one is answered the failed one would have arrived + template.sendBody(uri("&bodyOnly=true"), "hello"); + assertEquals(1, arrived.get()); + assertEquals("hello", event().asText()); + } + + private String uri(String options) { + return "splunk-hec:localhost:" + server.getAddress().getPort() + "?token=" + TOKEN + "&https=false" + options; + } + + private JsonNode event() throws Exception { + assertEquals(1, bodies.size(), bodies::toString); + return MAPPER.readTree(bodies.get(0)).get("event"); + } + @Test void testStartWithoutSslContextParameters() throws Exception { // https=true (the default) without sslContextParameters uses the default SSL context of the JVM diff --git a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc index 10f5a47c1b89..bf69d2646d1a 100644 --- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc +++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc @@ -804,6 +804,15 @@ CamelContext of that provider. It now closes each of them only when no other `JC it returned the closed cache, whose operations throw `IllegalStateException`. This matters for code that calls a `JCacheManager` (for example one created by a custom `JCacheManagerFactory`) after it was closed. +=== camel-splunk-hec - bytes, stream and file bodies are sent as text + +The producer now sends a byte array, an `InputStream`, a `Reader` or a file body (`File`, `Path` or the file of the +file consumer) as text in the event. Before, a byte array was sent as a base64 string, a `File` as its path, and a +stream failed after part of the request had already been sent. + +The JSON of the event is now written before the request is sent, with a `Content-Length` instead of a chunked request +body. A body or a header that cannot be written as JSON fails the exchange without sending anything to Splunk. + === camel-spring-batch - the CamelSpringBatchJobName header needs jobFromHeader=true The producer now takes the job name from the `CamelSpringBatchJobName` header only when the endpoint has
