From 864cabcfe0680e01f52b140d9b2efbaa75da7588 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 11 Dec 2014 22:27:41 +0200 Subject: [PATCH] INT-3450 Support RFC 5424/6587 for SYSLOG JIRA: https://jira.spring.io/browse/INT-3450 Use LinkedHashMap - natural iteration order INT-3450 Polishing - PR Comments INT-3450: Polishing --- .../syslog/DefaultMessageConverter.java | 4 + .../syslog/RFC5424MessageConverter.java | 89 ++++ .../syslog/RFC5424SyslogParser.java | 386 ++++++++++++++++++ .../integration/syslog/SyslogHeaders.java | 29 ++ .../inbound/RFC6587SyslogDeserializer.java | 119 ++++++ ...ivingChannelAdapterParserTests-context.xml | 13 +- ...logReceivingChannelAdapterParserTests.java | 9 +- .../inbound/SyslogDeserializerTests.java | 141 +++++++ .../SyslogReceivingChannelAdapterTests.java | 93 ++++- src/reference/docbook/syslog.xml | 51 ++- 10 files changed, 927 insertions(+), 7 deletions(-) create mode 100644 spring-integration-syslog/src/main/java/org/springframework/integration/syslog/RFC5424MessageConverter.java create mode 100644 spring-integration-syslog/src/main/java/org/springframework/integration/syslog/RFC5424SyslogParser.java create mode 100644 spring-integration-syslog/src/main/java/org/springframework/integration/syslog/inbound/RFC6587SyslogDeserializer.java create mode 100644 spring-integration-syslog/src/test/java/org/springframework/integration/syslog/inbound/SyslogDeserializerTests.java diff --git a/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/DefaultMessageConverter.java b/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/DefaultMessageConverter.java index 3613fa2ef0..2f6deb8fa2 100644 --- a/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/DefaultMessageConverter.java +++ b/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/DefaultMessageConverter.java @@ -59,6 +59,10 @@ public class DefaultMessageConverter implements MessageConverter, BeanFactoryAwa this.asMap = asMap; } + protected boolean asMap() { + return asMap; + } + @Override public final void setBeanFactory(BeanFactory beanFactory) { this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(beanFactory); diff --git a/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/RFC5424MessageConverter.java b/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/RFC5424MessageConverter.java new file mode 100644 index 0000000000..fbbcc181ac --- /dev/null +++ b/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/RFC5424MessageConverter.java @@ -0,0 +1,89 @@ +/* + * Copyright 2014 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.syslog; + +import java.util.Map; + +import org.springframework.integration.support.AbstractIntegrationMessageBuilder; +import org.springframework.messaging.Message; +import org.springframework.util.Assert; + +/** + * {@link MessageConverter}; delegates to a {@link RFC5424SyslogParser} if + * necessary (TCP will have already done the syslog conversion because it needs + * to handle different message framing). Copies the resulting {@link Map} to + * the message headers if {@link #asMap()} is false. + * + * @author Gary Russell + * @since 4.1.1 + */ +public class RFC5424MessageConverter extends DefaultMessageConverter { + + private final RFC5424SyslogParser parser; + + private String charset = "UTF-8"; + + /** + * Construct an instance with a default {@link RFC5424SyslogParser}. + */ + public RFC5424MessageConverter() { + this(new RFC5424SyslogParser()); + } + + /** + * Construct an instance with a non-standard parser. + * @param parser the parser. + */ + public RFC5424MessageConverter(RFC5424SyslogParser parser) { + this.parser = parser; + } + + /** + * @param charset the charset to set + */ + protected void setCharset(String charset) { + this.charset = charset; + } + + @SuppressWarnings("unchecked") + @Override + public Message fromSyslog(Message message) throws Exception { + boolean isMap = message.getPayload() instanceof Map; + Map map; + Object originalContent; + if (!isMap) { + Assert.isInstanceOf(byte[].class, message.getPayload(), "Only byte[] and Map payloads are supported"); + map = this.parser.parse(new String(((byte[]) message.getPayload()), this.charset), 0, false); + originalContent = message.getPayload(); + } + else { + map = (Map) message.getPayload(); + originalContent = map.get(SyslogHeaders.UNDECODED); + if (originalContent == null) { + originalContent = map; + } + } + + AbstractIntegrationMessageBuilder builder = getMessageBuilderFactory().withPayload( + asMap() ? map : originalContent); + if (!asMap() && isMap) { + builder.copyHeaders(map); + } + return builder.build(); + } + +} diff --git a/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/RFC5424SyslogParser.java b/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/RFC5424SyslogParser.java new file mode 100644 index 0000000000..c0b4cc2f8d --- /dev/null +++ b/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/RFC5424SyslogParser.java @@ -0,0 +1,386 @@ +/* + * Copyright 2014 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.syslog; + +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +import org.springframework.util.Assert; + +/** + * Parse for RFC 5424 syslog messages; when used with TCP, requires the use + * of a {@code RFC6587SyslogDeserializer} which decodes the framing. + * + * @author Duncan McIntyre + * @author Gary Russell + * @since 1.4.1 + * + */ +public class RFC5424SyslogParser { + + protected static final char NILVALUE = '-'; + + protected static final char SPACE = ' '; + + protected final boolean retainOriginal; + + + /** + * Construct a default parser; do not retain the original message content unless there + * is an error. + */ + public RFC5424SyslogParser() { + this(false); + } + + /** + * @param retainOriginal when true, include the original message content intact in the + * map. + */ + public RFC5424SyslogParser(boolean retainOriginal) { + this.retainOriginal = retainOriginal; + } + + public Map parse(String line, int octetCount, boolean shortRead) { + + Map map = new LinkedHashMap(); + + Reader r = new Reader(line); + + try { + if (shortRead) { + int n = line.length() - 1; + while (n >= 0 && line.charAt(n) == 0x00) { + n--; + } + line = line.substring(0, n); + throw new IllegalStateException("Insufficient data; expected " + octetCount + " got " + (n + 1)); + } + r.expect('<'); + int pri = r.readInt(); + r.expect('>'); + + int version = r.readInt(); + r.expect(SPACE); + + Object timestamp = getTimestamp(r); + + String host = r.getIdentifier(); + String app = r.getIdentifier(); + String procId = r.getIdentifier(); + String msgId = r.getIdentifier(); + + Object structuredData = getStructuredData(r); + + String message; + if(r.is(SPACE)) { + r.getc(); + message = r.rest(); + } + else { + message = ""; + } + + int severity = pri & 0x7; + int facility = pri >> 3; + map.put(SyslogHeaders.FACILITY, facility); + + map.put(SyslogHeaders.SEVERITY, severity); + map.put(SyslogHeaders.SEVERITY_TEXT, Severity.parseInt(severity).label()); + + if (timestamp != null) { + map.put(SyslogHeaders.TIMESTAMP, timestamp); + } + + if (host != null) { + map.put(SyslogHeaders.HOST, host); + } + if (app != null) { + map.put(SyslogHeaders.APP_NAME, app); + } + if (procId != null) { + map.put(SyslogHeaders.PROCID, procId); + } + if (msgId != null) { + map.put(SyslogHeaders.MSGID, msgId); + } + map.put(SyslogHeaders.VERSION, version); + + if (structuredData != null) { + map.put(SyslogHeaders.STRUCTURED_DATA, structuredData); + } + + map.put(SyslogHeaders.MESSAGE, message); + map.put(SyslogHeaders.DECODE_ERRORS, "false"); + + if (this.retainOriginal) { + map.put(SyslogHeaders.UNDECODED, line); + } + } + catch(IllegalStateException e) { + map.put(SyslogHeaders.DECODE_ERRORS, "true"); + map.put(SyslogHeaders.ERRORS, e.getMessage()); + map.put(SyslogHeaders.UNDECODED, line); + } + catch(StringIndexOutOfBoundsException sob) { + map.put(SyslogHeaders.DECODE_ERRORS, "true"); + map.put(SyslogHeaders.ERRORS, "Unexpected end of message: " + sob.getMessage()); + map.put(SyslogHeaders.UNDECODED, line); + } + return map; + } + + /** + * Default implementation returns the date as a String (if present). + * @param r the reader. + * @return the timestamp. + */ + protected Object getTimestamp(Reader r) { + + int c = r.getc(); + + if(c == NILVALUE) { + return null; + } + + if(!Character.isDigit(c)) { + throw new IllegalStateException("Year expected @" + r.idx); + } + + StringBuilder dateBuilder = new StringBuilder(); + dateBuilder.append((char) c); + while ((c = r.getc()) != SPACE) { + dateBuilder.append((char) c); + } + + return dateBuilder.toString(); + } + + private Object getStructuredData(Reader r) { + if(r.is(NILVALUE)) { + r.getc(); + return null; + } + return parseStructuredDataElements(r); + } + + /** + * Default implementation returns a list of structured data elements with + * no internal parsing. + * @param r the reader. + * @return the structured data. + */ + protected Object parseStructuredDataElements(Reader r) { + List fragments = new ArrayList(); + while (r.is('[')) { + r.mark(); + r.skipTo(']'); + fragments.add(r.getMarkedSegment()); + } + return fragments; + } + + protected class Reader { + + private final String line; + + public int idx; + + private int mark; + + public Reader(String l) { + line = l; + } + + public void mark() { + this.mark = this.idx; + } + + public String getMarkedSegment() { + Assert.state(this.mark <= this.idx, "mark is greater than this.idx"); + return this.line.substring(mark, this.idx); + } + + public int current() { + return line.charAt(this.idx); + } + + public int prev() { + return line.charAt(this.idx - 1); + } + + public int getc() { + return line.charAt(this.idx++); + } + + public int peek() { + return line.charAt(this.idx + 1); + } + + public void ungetc() { + this.idx--; + } + + public int getInt() { + int c = getc(); + if (!Character.isDigit(c)) { + ungetc(); + return -1; + } + + return c - '0'; + } + + /** + * Read characters building an int until a non-digit is found + * @return int + */ + public int readInt() { + int val = 0; + while (isDigit()) { + val = (val * 10) + getInt(); + } + return val; + } + + public double readFraction() { + int val = 0; + int order = 0; + while (isDigit()) { + val = (val * 10) + getInt(); + order *= 10; + } + return (double) val / order; + + } + + public boolean is(char c) { + return line.charAt(this.idx) == c; + } + + public boolean was(char c) { + return line.charAt(this.idx - 1) == c; + } + + public boolean isDigit() { + return Character.isDigit(line.charAt(this.idx)); + } + + public void expect(char c) { + if (line.charAt(this.idx++) != c) { + throw new IllegalStateException("Expected '" + c + "' @" + this.idx); + } + } + + public void skipTo(char searchChar) { + while (!is(searchChar) || was('\\')) { + getc(); + } + getc(); + } + + public String rest() { + return line.substring(this.idx); + } + + public String getIdentifier() { + StringBuilder sb = new StringBuilder(); + int c; + while (true) { + c = getc(); + if (c >= 33 && c <= 127) { + sb.append((char) c); + } + else { + break; + } + } + return sb.toString(); + } + + } + + protected enum Severity { + + DEBUG(7, "DEBUG"), + + INFO(6, "INFO"), + + NOTICE(5, "NOTICE"), + + WARN(4, "WARN"), + + ERROR(3, "ERRORS"), + + CRITICAL(2, "CRITICAL"), + + ALERT(1, "ALERT"), + + EMERGENCY(0, "EMERGENCY"), + + UNDEFINED(-1, "UNDEFINED"); + + private final int level; + + private final String label; + + private Severity(int level, String label) { + this.level = level; + this.label = label; + } + + public int level() { + return level; + } + + public String label() { + return label; + } + + public static Severity parseInt(int syslogSeverity) { + if(syslogSeverity == 7) { + return DEBUG; + } + if(syslogSeverity == 6) { + return INFO; + } + if(syslogSeverity == 5) { + return NOTICE; + } + if(syslogSeverity == 4) { + return WARN; + } + if(syslogSeverity == 3) { + return ERROR; + } + if(syslogSeverity == 2) { + return CRITICAL; + } + if(syslogSeverity == 1) { + return ALERT; + } + if(syslogSeverity == 0) { + return EMERGENCY; + } + return UNDEFINED; + } + + } + + +} diff --git a/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/SyslogHeaders.java b/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/SyslogHeaders.java index 47e96e899e..a0c727dc96 100644 --- a/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/SyslogHeaders.java +++ b/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/SyslogHeaders.java @@ -38,4 +38,33 @@ public class SyslogHeaders { public static final String TAG = PREFIX + SyslogToMapTransformer.TAG; + //// RFC 5424 Parser: + + public static final String MESSAGE = PREFIX + "MESSAGE"; + + public static final String APP_NAME = PREFIX + "APP_NAME" ; + + public static final String PROCID = PREFIX + "PROCID" ; + + public static final String MSGID = PREFIX + "MSGID" ; + + public static final String VERSION = PREFIX + "VERSION" ; + + public static final String STRUCTURED_DATA = PREFIX + "STRUCTURED_DATA" ; + + // Text versions of syslog numeric values + public static final String SEVERITY_TEXT = PREFIX + "SEVERITY_TEXT" ; + + // Additional fields + public static final String SOURCE_TYPE = PREFIX + "SOURCE_TYPE" ; + + public static final String SOURCE = PREFIX + "SOURCE" ; + + // full line when parse errors or retained original + public static final String UNDECODED = PREFIX + "UNDECODED" ; + + public static final String DECODE_ERRORS = PREFIX + "DECODE_ERRORS" ; + + public static final String ERRORS = PREFIX + "ERRORS"; + } diff --git a/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/inbound/RFC6587SyslogDeserializer.java b/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/inbound/RFC6587SyslogDeserializer.java new file mode 100644 index 0000000000..fc78ec5ee6 --- /dev/null +++ b/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/inbound/RFC6587SyslogDeserializer.java @@ -0,0 +1,119 @@ +/* + * Copyright 2014 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.syslog.inbound; + +import java.io.DataInputStream; +import java.io.EOFException; +import java.io.IOException; +import java.io.InputStream; +import java.util.Map; + +import org.springframework.core.serializer.Deserializer; +import org.springframework.integration.ip.tcp.serializer.ByteArrayLfSerializer; +import org.springframework.integration.ip.tcp.serializer.SoftEndOfStreamException; +import org.springframework.integration.syslog.RFC5424SyslogParser; +import org.springframework.util.Assert; + +/** + * RFC5424/6587 Deserializer. Implemented as a {@link Deserializer} instead of a + * transformer because we may receive a mixture of octet counting and non-transparent + * framing - see RFC 6587. + * + * @author Duncan McIntyre + * @author Gary Russell + * @since 4.1.1 + * + */ +public class RFC6587SyslogDeserializer implements Deserializer> { + + private final Deserializer delimitedDeserializer; + + private RFC5424SyslogParser parser = new RFC5424SyslogParser(); + + /** + * Construct an instance using a {@link ByteArrayLfSerializer} for + * non-transparent frames. + */ + public RFC6587SyslogDeserializer() { + this.delimitedDeserializer = new ByteArrayLfSerializer(); + } + + /** + * Construct an instance using the specified {@link Deserializer} for + * non-transparent frames. + * @param delimitedDeserializer the Deserializer. + */ + public RFC6587SyslogDeserializer(Deserializer delimitedDeserializer) { + this.delimitedDeserializer = delimitedDeserializer; + } + + /** + * @param parser the parser to set + */ + public void setParser(RFC5424SyslogParser parser) { + this.parser = parser; + } + + @Override + public Map deserialize(InputStream inputStream) throws IOException { + DataInputStream stream = new DataInputStream(inputStream); + String line; + int octetCount = 0; + boolean shortRead = false; + int peek = stream.read(); + if (isDigit(peek)) { + octetCount = calculateLength(stream, peek); + Assert.state(octetCount > 0, "Expected length > 0"); + byte[] bytes = new byte[octetCount]; + try { + stream.readFully(bytes); + } + catch (EOFException e) { + shortRead = true; + } + line = new String(bytes, getCharset()); + } + else if (peek == '<') { + byte[] bytes = this.delimitedDeserializer.deserialize(inputStream); + line = "<" + new String(bytes, getCharset()); + } + else if (peek < 0) { + throw new SoftEndOfStreamException(); + } + else { + throw new IllegalStateException("Expected a digit or '<', got 0x" + Integer.toHexString(peek)); + } + return parser.parse(line, octetCount, shortRead); + } + + private boolean isDigit(int peek) { + return peek >= 0x30 && peek <= 0x39; + } + + private int calculateLength(DataInputStream stream, int peek) throws IOException { + int length = peek & 0xf; + int c; + while (isDigit((c = stream.read()))) { + length = length*10 + (c & 0xf); + } + return length; + } + + protected String getCharset() { + return "UTF-8"; + } + +} diff --git a/spring-integration-syslog/src/test/java/org/springframework/integration/syslog/config/SyslogReceivingChannelAdapterParserTests-context.xml b/spring-integration-syslog/src/test/java/org/springframework/integration/syslog/config/SyslogReceivingChannelAdapterParserTests-context.xml index 419dfa6d2d..9b04b71808 100644 --- a/spring-integration-syslog/src/test/java/org/springframework/integration/syslog/config/SyslogReceivingChannelAdapterParserTests-context.xml +++ b/spring-integration-syslog/src/test/java/org/springframework/integration/syslog/config/SyslogReceivingChannelAdapterParserTests-context.xml @@ -54,9 +54,18 @@ connection-factory="cf" auto-startup="false" phase="123" - converter="converter" + converter="rfc5424" send-timeout="456" error-channel="errors" /> - + + + + + + diff --git a/spring-integration-syslog/src/test/java/org/springframework/integration/syslog/config/SyslogReceivingChannelAdapterParserTests.java b/spring-integration-syslog/src/test/java/org/springframework/integration/syslog/config/SyslogReceivingChannelAdapterParserTests.java index 36744d41b8..e9759999b9 100644 --- a/spring-integration-syslog/src/test/java/org/springframework/integration/syslog/config/SyslogReceivingChannelAdapterParserTests.java +++ b/spring-integration-syslog/src/test/java/org/springframework/integration/syslog/config/SyslogReceivingChannelAdapterParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-2014 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -37,9 +37,9 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.beans.factory.parsing.BeanDefinitionParsingException; import org.springframework.context.support.ClassPathXmlApplicationContext; -import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory; import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory; import org.springframework.integration.syslog.MessageConverter; +import org.springframework.integration.syslog.RFC5424MessageConverter; import org.springframework.integration.syslog.inbound.TcpSyslogReceivingChannelAdapter; import org.springframework.integration.syslog.inbound.UdpSyslogReceivingChannelAdapter; import org.springframework.integration.test.util.TestUtils; @@ -81,6 +81,9 @@ public class SyslogReceivingChannelAdapterParserTests { @Autowired private PassThruConverter converter; + @Autowired + private RFC5424MessageConverter rfc5424; + @Autowired @Qualifier("bar.adapter") private TcpSyslogReceivingChannelAdapter adapter2; @@ -152,7 +155,7 @@ public class SyslogReceivingChannelAdapterParserTests { assertFalse(fullBoatTcp.isAutoStartup()); assertEquals(123, fullBoatTcp.getPhase()); assertEquals(456L, TestUtils.getPropertyValue(fullBoatUdp, "messagingTemplate.sendTimeout")); - assertSame(converter, TestUtils.getPropertyValue(fullBoatTcp, "converter")); + assertSame(rfc5424, TestUtils.getPropertyValue(fullBoatTcp, "converter")); assertSame(errors, TestUtils.getPropertyValue(fullBoatTcp, "errorChannel")); assertSame(cf, TestUtils.getPropertyValue(fullBoatTcp, "connectionFactory")); } diff --git a/spring-integration-syslog/src/test/java/org/springframework/integration/syslog/inbound/SyslogDeserializerTests.java b/spring-integration-syslog/src/test/java/org/springframework/integration/syslog/inbound/SyslogDeserializerTests.java new file mode 100644 index 0000000000..f06a65cc7a --- /dev/null +++ b/spring-integration-syslog/src/test/java/org/springframework/integration/syslog/inbound/SyslogDeserializerTests.java @@ -0,0 +1,141 @@ +/* + * Copyright 2014 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.syslog.inbound; + +import static org.junit.Assert.assertEquals; + +import java.io.ByteArrayInputStream; +import java.util.List; +import java.util.Map; + +import org.junit.Test; + +import org.springframework.integration.syslog.SyslogHeaders; + +/** + * @author Duncan McIntyre + * @author Gary Russell + * @since 4.1.1 + * + */ +public class SyslogDeserializerTests { + + static final String VALID_UNFRAMED_ENTRY = + "<14>1 2014-06-20T09:14:07+00:00 loggregator d0602076-b14a-4c55-852a-981e7afeed38 DEA - - Removing instance\n"; + static final String VALID_FRAMED_ENTRY = + "106 <14>1 2014-06-20T09:14:07+00:00 loggregator d0602076-b14a-4c55-852a-981e7afeed38 DEA - - Removing instance"; + static final String SHORT_FRAMED_ENTRY = + "107 <14>1 2014-06-20T09:14:07+00:00 loggregator d0602076-b14a-4c55-852a-981e7afeed38 DEA - - Removing instance"; + static final String SD_ENTRY_1 = + "179 <14>1 2014-06-20T09:14:07+00:00 loggregator d0602076-b14a-4c55-852a-981e7afeed38 DEA - " + + "[exampleSDID@32473 iut=\\\"3\\\" eventSource=\\\"Application\\\" eventID=\\\"1011\\\"] Removing instance"; + static final String SD_ENTRY_2 = + "253 <14>1 2014-06-20T09:14:07+00:00 loggregator d0602076-b14a-4c55-852a-981e7afeed38 DEA - " + + "[exampleSDID@32473 iut=\\\"3\\\" eventSource=\\\"Application\\\" eventID=\\\"1011\\\"][exampleSDID@32473 " + + "iut=\\\"3\\\" eventSource=\\\"Application\\\" eventID=\\\"1011\\\"] Removing instance"; + static final String SD_ENTRY_3 = "275 <14>1 2014-06-20T09:14:07+00:00 loggregator d0602076-b14a-4c55-852a-981e7afeed38 DEA - " + + "[exampleSDID@32473 iut=\\\"3\\\" eventSource=\\\"Application\\\" eventID=\\\"1011\\\"][exampleSDID@32473 " + + "iut=\\\"3\\\" eventSource=\\\"Application\\\" escapedBracket=\\\"\\]\\\" eventID=\\\"1011\\\"] Removing instance"; + + @Test + public void shouldParseAValidFramedEntry() throws Exception { + + RFC6587SyslogDeserializer deserializer = new RFC6587SyslogDeserializer(); + + Map map = deserializer.deserialize(new ByteArrayInputStream(VALID_FRAMED_ENTRY.getBytes())); + + assertEquals(1, map.get(SyslogHeaders.FACILITY)); + assertEquals(6, map.get(SyslogHeaders.SEVERITY)); + assertEquals(1, map.get(SyslogHeaders.VERSION)); + assertEquals("2014-06-20T09:14:07+00:00", map.get(SyslogHeaders.TIMESTAMP)); + assertEquals("loggregator", map.get(SyslogHeaders.HOST)); + assertEquals("d0602076-b14a-4c55-852a-981e7afeed38", map.get(SyslogHeaders.APP_NAME)); + assertEquals("DEA", map.get(SyslogHeaders.PROCID)); + assertEquals("-", map.get(SyslogHeaders.MSGID)); + assertEquals("Removing instance", map.get(SyslogHeaders.MESSAGE)); + } + + @Test + public void shouldParseAValidUnframedEntry() throws Exception { + + RFC6587SyslogDeserializer deserializer = new RFC6587SyslogDeserializer(); + + Map map = deserializer.deserialize(new ByteArrayInputStream(VALID_UNFRAMED_ENTRY.getBytes())); + + assertEquals(1, map.get(SyslogHeaders.FACILITY)); + assertEquals(6, map.get(SyslogHeaders.SEVERITY)); + assertEquals(1, map.get(SyslogHeaders.VERSION)); + assertEquals("2014-06-20T09:14:07+00:00", map.get(SyslogHeaders.TIMESTAMP)); + assertEquals("loggregator", map.get(SyslogHeaders.HOST)); + assertEquals("d0602076-b14a-4c55-852a-981e7afeed38", map.get(SyslogHeaders.APP_NAME)); + assertEquals("DEA", map.get(SyslogHeaders.PROCID)); + assertEquals("-", map.get(SyslogHeaders.MSGID)); + assertEquals("Removing instance", map.get(SyslogHeaders.MESSAGE)); + } + + @Test + public void shouldGetStructuredData() throws Exception { + + RFC6587SyslogDeserializer deserializer = new RFC6587SyslogDeserializer(); + + Map map = deserializer.deserialize(new ByteArrayInputStream(SD_ENTRY_1.getBytes())); + + assertEquals(1, map.get(SyslogHeaders.FACILITY)); + assertEquals(6, map.get(SyslogHeaders.SEVERITY)); + assertEquals(1, map.get(SyslogHeaders.VERSION)); + assertEquals("2014-06-20T09:14:07+00:00", map.get(SyslogHeaders.TIMESTAMP)); + assertEquals("loggregator", map.get(SyslogHeaders.HOST)); + assertEquals("d0602076-b14a-4c55-852a-981e7afeed38", map.get(SyslogHeaders.APP_NAME)); + assertEquals("DEA", map.get(SyslogHeaders.PROCID)); + assertEquals("-", map.get(SyslogHeaders.MSGID)); + assertEquals("Removing instance", map.get(SyslogHeaders.MESSAGE)); + assertEquals(1, ((List) map.get(SyslogHeaders.STRUCTURED_DATA)).size()); + } + + @Test + public void shouldGetMultipleStructuredData() throws Exception { + + RFC6587SyslogDeserializer deserializer = new RFC6587SyslogDeserializer(); + + Map map = deserializer.deserialize(new ByteArrayInputStream(SD_ENTRY_2.getBytes())); + + assertEquals("false", map.get(SyslogHeaders.DECODE_ERRORS)); + assertEquals("Removing instance", map.get(SyslogHeaders.MESSAGE)); + assertEquals(2, ((List) map.get(SyslogHeaders.STRUCTURED_DATA)).size()); + } + + @Test + public void shouldGetMultipleStructuredDataWithEscapedBracket() throws Exception { + + RFC6587SyslogDeserializer deserializer = new RFC6587SyslogDeserializer(); + + Map map = deserializer.deserialize(new ByteArrayInputStream(SD_ENTRY_3.getBytes())); + + assertEquals("false", map.get(SyslogHeaders.DECODE_ERRORS)); + assertEquals("Removing instance", map.get(SyslogHeaders.MESSAGE)); + } + + @Test + public void shouldErrorOnShortFramedData() throws Exception { + + RFC6587SyslogDeserializer deserializer = new RFC6587SyslogDeserializer(); + + Map map = deserializer.deserialize(new ByteArrayInputStream(SHORT_FRAMED_ENTRY.getBytes())); + + assertEquals("true", map.get(SyslogHeaders.DECODE_ERRORS)); + } + +} diff --git a/spring-integration-syslog/src/test/java/org/springframework/integration/syslog/inbound/SyslogReceivingChannelAdapterTests.java b/spring-integration-syslog/src/test/java/org/springframework/integration/syslog/inbound/SyslogReceivingChannelAdapterTests.java index 4d70c5194f..8681dce8af 100644 --- a/spring-integration-syslog/src/test/java/org/springframework/integration/syslog/inbound/SyslogReceivingChannelAdapterTests.java +++ b/spring-integration-syslog/src/test/java/org/springframework/integration/syslog/inbound/SyslogReceivingChannelAdapterTests.java @@ -29,6 +29,7 @@ import java.net.DatagramPacket; import java.net.DatagramSocket; import java.net.InetSocketAddress; import java.net.Socket; +import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -44,7 +45,10 @@ import org.springframework.beans.factory.BeanFactory; import org.springframework.context.ApplicationEvent; import org.springframework.context.ApplicationEventPublisher; import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory; +import org.springframework.integration.ip.tcp.connection.TcpNioServerConnectionFactory; import org.springframework.integration.syslog.DefaultMessageConverter; +import org.springframework.integration.syslog.RFC5424MessageConverter; import org.springframework.integration.syslog.config.SyslogReceivingChannelAdapterFactoryBean; import org.springframework.integration.test.util.SocketUtils; import org.springframework.integration.test.util.TestUtils; @@ -109,7 +113,7 @@ public class SyslogReceivingChannelAdapterTests { Log logger = spy(TestUtils.getPropertyValue(adapter, "logger", Log.class)); doReturn(true).when(logger).isDebugEnabled(); final CountDownLatch sawLog = new CountDownLatch(1); - doAnswer(new Answer(){ + doAnswer(new Answer() { @Override public Void answer(InvocationOnMock invocation) throws Throwable { @@ -163,4 +167,91 @@ public class SyslogReceivingChannelAdapterTests { adapter.stop(); } + @Test + public void testTcpRFC5424() throws Exception { + SyslogReceivingChannelAdapterFactoryBean factory = new SyslogReceivingChannelAdapterFactoryBean( + SyslogReceivingChannelAdapterFactoryBean.Protocol.tcp); + int port = SocketUtils.findAvailableServerSocket(1514); + PollableChannel outputChannel = new QueueChannel(); + factory.setOutputChannel(outputChannel); + ApplicationEventPublisher publisher = mock(ApplicationEventPublisher.class); + final CountDownLatch latch = new CountDownLatch(2); + doAnswer(new Answer() { + + @Override + public Object answer(InvocationOnMock invocation) throws Throwable { + latch.countDown(); + return null; + } + }).when(publisher).publishEvent(any(ApplicationEvent.class)); + factory.setBeanFactory(mock(BeanFactory.class)); + AbstractServerConnectionFactory connectionFactory = new TcpNioServerConnectionFactory(port); + connectionFactory.setDeserializer(new RFC6587SyslogDeserializer()); + connectionFactory.setApplicationEventPublisher(publisher); + factory.setConnectionFactory(connectionFactory); + factory.setConverter(new RFC5424MessageConverter()); + factory.afterPropertiesSet(); + factory.start(); + TcpSyslogReceivingChannelAdapter adapter = (TcpSyslogReceivingChannelAdapter) factory.getObject(); + Log logger = spy(TestUtils.getPropertyValue(adapter, "logger", Log.class)); + doReturn(true).when(logger).isDebugEnabled(); + final CountDownLatch sawLog = new CountDownLatch(1); + doAnswer(new Answer(){ + + @Override + public Void answer(InvocationOnMock invocation) throws Throwable { + if (((String) invocation.getArguments()[0]).contains("Error on syslog socket")) { + sawLog.countDown(); + } + invocation.callRealMethod(); + return null; + } + }).when(logger).debug(anyString()); + new DirectFieldAccessor(adapter).setPropertyValue("logger", logger); + Thread.sleep(1000); + byte[] buf = ("253 <14>1 2014-06-20T09:14:07+00:00 loggregator d0602076-b14a-4c55-852a-981e7afeed38 DEA - " + + "[exampleSDID@32473 iut=\\\"3\\\" eventSource=\\\"Application\\\" eventID=\\\"1011\\\"]" + + "[exampleSDID@32473 iut=\\\"3\\\" eventSource=\\\"Application\\\" eventID=\\\"1011\\\"] Removing instance") + .getBytes("UTF-8"); + Socket socket = SocketFactory.getDefault().createSocket("localhost", port); + socket.getOutputStream().write(buf); + socket.close(); + assertTrue(sawLog.await(10, TimeUnit.SECONDS)); + @SuppressWarnings("unchecked") + Message> message = (Message>) outputChannel.receive(10000); + assertNotNull(message); + assertEquals("loggregator", message.getPayload().get("syslog_HOST")); + adapter.stop(); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + } + + @Test + public void testUdpRFC5424() throws Exception { + SyslogReceivingChannelAdapterFactoryBean factory = new SyslogReceivingChannelAdapterFactoryBean( + SyslogReceivingChannelAdapterFactoryBean.Protocol.udp); + int port = SocketUtils.findAvailableUdpSocket(1514); + factory.setPort(port); + PollableChannel outputChannel = new QueueChannel(); + factory.setOutputChannel(outputChannel); + factory.setBeanFactory(mock(BeanFactory.class)); + factory.setConverter(new RFC5424MessageConverter()); + factory.afterPropertiesSet(); + factory.start(); + UdpSyslogReceivingChannelAdapter adapter = (UdpSyslogReceivingChannelAdapter) factory.getObject(); + Thread.sleep(1000); + byte[] buf = ("<14>1 2014-06-20T09:14:07+00:00 loggregator d0602076-b14a-4c55-852a-981e7afeed38 DEA - " + + "[exampleSDID@32473 iut=\\\"3\\\" eventSource=\\\"Application\\\" eventID=\\\"1011\\\"]" + + "[exampleSDID@32473 iut=\\\"3\\\" eventSource=\\\"Application\\\" eventID=\\\"1011\\\"] Removing instance") + .getBytes("UTF-8"); + DatagramPacket packet = new DatagramPacket(buf, buf.length, new InetSocketAddress("localhost", port)); + DatagramSocket socket = new DatagramSocket(); + socket.send(packet); + socket.close(); + @SuppressWarnings("unchecked") + Message> message = (Message>) outputChannel.receive(10000); + assertNotNull(message); + assertEquals("loggregator", message.getPayload().get("syslog_HOST")); + adapter.stop(); + } + } diff --git a/src/reference/docbook/syslog.xml b/src/reference/docbook/syslog.xml index c5727ad440..eb90575932 100644 --- a/src/reference/docbook/syslog.xml +++ b/src/reference/docbook/syslog.xml @@ -16,17 +16,26 @@ Spring Integration 3.0 introduced convenient namespace support for configuring a Syslog inbound adapter in a single element. + + Starting with version 4.1.1, the framework now supports the + extended Syslog format, as specified in RFC 5424>. In addition, when using TCP and RFC5424, both octet counting and + non-transparent framing described in RFC 6587 are supported. +
Syslog <inbound-channel-adapter> - This element encompases a UDP or TCP inbound channel adapter + This element encompasses a UDP or TCP inbound channel adapter and a MessageConverter to convert the Syslog message to a Spring Integration message. The DefaultMessageConverter delegates to the SyslogToMapTransformer, creating a message with its payload being the Map of Syslog fields. In addition, all fields except the message are also made available as headers in the message, prefixed with syslog_. + In this mode, only RFC 3164 + (BSD) syslogs are supported. Since version 4.1, @@ -35,6 +44,25 @@ the message payload as the original complete syslog message, in a byte[], while still setting the headers. + + Since version 4.1.1, RFC 5424 is also supported, using the + RFC5424MessageConverter; in this case the fields are not + copied as headers, unless asMap is set to false, in which + case the original message is the payload and the decoded fields are headers. + + + To use RFC 5424 with a TCP transport, additional configuration is required, to enable + the different framing techniques described in RFC 6587. The adapter needs a + TCP connection factory configured with a RFC6587SyslogDeserializer. + By default, this deserializer will handle octet counting and + non-transparent framing, using a linefeed (LF) to delimit syslog messages; + it uses a ByteArrayLfSerializer when octet counting + is not detected. To use different non-transparent framing, you can provide + it with some other deserializer. While the deserializer can support both octet counting + and non-transparent framing, only one form of the latter is supported. + If asMap is false on the converter, you must set the + retainOriginal constructor argument in the RFC6587SyslogDeserializer. +
Example Configuration ]]> @@ -94,6 +122,27 @@ The externally configured connection-factory must be of type server and, the port is defined there rather than on the inbound-channel-adapter element itself. + + + + + + +]]> + + A TCP adapter that sends messages to channel fromSyslog. It is configured + to use the RFC 5424 converter and is configured + with a reference to an externally defined connection factory with the RFC 6587 + deserializer (required for RFC 5424). +