@Test public void parsingDefaultEventTest() throws RedisSerializerException { LogstashSerializer serializer = new LogstashSerializer(); byte[] body = new byte[] { '1', '2', '3', '4', '5' }; String testHost = "testhost"; String sourcePath = "/my/source/path"; String type = "mytype"; String source = "my source"; List<String> tags = new ArrayList<String>(); tags.add("one tag"); tags.add("another tag"); long now = System.currentTimeMillis(); Map<String, String> headers = new HashMap<String, String>(); headers.put(LogstashSerializer.TIMESTAMP, Long.toString(now)); headers.put(LogstashSerializer.HOST, testHost); headers.put(LogstashSerializer.SRC_PATH, sourcePath); headers.put(LogstashSerializer.TYPE, type); headers.put(LogstashSerializer.SOURCE, source); headers.put(LogstashSerializer.TAGS, StringUtils.join(tags, LogstashSerializer.DEFAULT_TAGS_SEPARATOR)); Event event = new SimpleEvent(); event.setBody(body); event.setHeaders(headers); LogstashEvent logstashEvent = serializer.convertToLogstashEvent(event); Assert.assertNotNull(logstashEvent.getMessage()); Assert.assertEquals(new String(body), logstashEvent.getMessage()); Assert.assertNotNull(logstashEvent.getTimestamp()); Assert.assertEquals(new Date(now), logstashEvent.getTimestamp()); Assert.assertNotNull(logstashEvent.getSourceHost()); Assert.assertEquals(testHost, logstashEvent.getSourceHost()); Assert.assertNotNull(logstashEvent.getSourcePath()); Assert.assertEquals(sourcePath, logstashEvent.getSourcePath()); Assert.assertNotNull(logstashEvent.getSource()); Assert.assertEquals(source, logstashEvent.getSource()); Assert.assertNotNull(logstashEvent.getType()); Assert.assertEquals(type, logstashEvent.getType()); Assert.assertNotNull(logstashEvent.getTags()); Assert.assertEquals(tags.size(), logstashEvent.getTags().size()); Assert.assertTrue(CollectionUtils.isEqualCollection(tags, logstashEvent.getTags())); Assert.assertNotNull(logstashEvent.getFields()); Assert.assertEquals(headers.size(), logstashEvent.getFields().size()); for(String key: headers.keySet()) { Assert.assertTrue(logstashEvent.getFields().containsKey(key)); Assert.assertEquals(headers.get(key), logstashEvent.getFields().get(key)); } }
public LogstashEvent convertToLogstashEvent(Event event) throws RedisSerializerException { if (event == null) { throw new RedisSerializerException("Event cannot be null"); } LogstashEvent logstashEvent = new LogstashEvent(); final Map<String, String> headers = event.getHeaders(); logstashEvent.setMessage(new String(event.getBody())); for (Entry<String, String> entry : headers.entrySet()) { logstashEvent.putField(entry.getKey(), entry.getValue()); } String timestamp = null; if (StringUtils.isNotBlank(headers.get(LogstashConstant.ATTRIBUTE_AT_TIMESTAMP))) { timestamp = headers.get(LogstashConstant.ATTRIBUTE_AT_TIMESTAMP); } else if (StringUtils.isNotBlank(headers.get(TIMESTAMP))) { timestamp = headers.get(TIMESTAMP); } if (timestamp != null) { long timestampMs = Long.parseLong(timestamp); logstashEvent.setTimestamp(new Date(timestampMs)); } String source = null; if (StringUtils.isNotBlank(headers.get(LogstashConstant.ATTRIBUTE_SOURCE))) { source = headers.get(LogstashConstant.ATTRIBUTE_SOURCE); } else if (StringUtils.isNotBlank(headers.get(SOURCE))) { source = headers.get(SOURCE); } if (source != null) { logstashEvent.setSource(source); } String host = null; if (StringUtils.isNotBlank(headers.get(LogstashConstant.ATTRIBUTE_SOURCE_HOST))) { host = headers.get(LogstashConstant.ATTRIBUTE_SOURCE_HOST); } else if (StringUtils.isNotBlank(headers.get(HOST))) { host = headers.get(HOST); } if (host != null) { logstashEvent.setSourceHost(host); } String sourcePath = null; if (StringUtils.isNotBlank(headers.get(LogstashConstant.ATTRIBUTE_SOURCE_PATH))) { sourcePath = headers.get(LogstashConstant.ATTRIBUTE_SOURCE_PATH); } else if (StringUtils.isNotBlank(headers.get(SRC_PATH))) { sourcePath = headers.get(SRC_PATH); } if (sourcePath != null) { logstashEvent.setSourcePath(sourcePath); } String type = null; if (StringUtils.isNotBlank(headers.get(LogstashConstant.ATTRIBUTE_TYPE))) { type = headers.get(LogstashConstant.ATTRIBUTE_TYPE); } else if (StringUtils.isNotBlank(headers.get(TYPE))) { type = headers.get(TYPE); } if (type != null) { logstashEvent.setType(type); } String tags = null; if (StringUtils.isNotBlank(headers.get(LogstashConstant.ATTRIBUTE_TAGS))) { tags = headers.get(LogstashConstant.ATTRIBUTE_TAGS); } else if (StringUtils.isNotBlank(headers.get(TAGS))) { tags = headers.get(TAGS); } if (StringUtils.isNotBlank(tags)) { String[] allTags = tags.split(tagsSeparator); List<String> tagsList = new ArrayList<String>(); for(String tag: allTags) { if(StringUtils.isNotBlank(tag)) { tagsList.add(tag); } } logstashEvent.setTags(tagsList); } return logstashEvent; }
LogstashSerializer implements Serializer { public LogstashEvent convertToLogstashEvent(Event event) throws RedisSerializerException { if (event == null) { throw new RedisSerializerException("Event cannot be null"); } LogstashEvent logstashEvent = new LogstashEvent(); final Map<String, String> headers = event.getHeaders(); logstashEvent.setMessage(new String(event.getBody())); for (Entry<String, String> entry : headers.entrySet()) { logstashEvent.putField(entry.getKey(), entry.getValue()); } String timestamp = null; if (StringUtils.isNotBlank(headers.get(LogstashConstant.ATTRIBUTE_AT_TIMESTAMP))) { timestamp = headers.get(LogstashConstant.ATTRIBUTE_AT_TIMESTAMP); } else if (StringUtils.isNotBlank(headers.get(TIMESTAMP))) { timestamp = headers.get(TIMESTAMP); } if (timestamp != null) { long timestampMs = Long.parseLong(timestamp); logstashEvent.setTimestamp(new Date(timestampMs)); } String source = null; if (StringUtils.isNotBlank(headers.get(LogstashConstant.ATTRIBUTE_SOURCE))) { source = headers.get(LogstashConstant.ATTRIBUTE_SOURCE); } else if (StringUtils.isNotBlank(headers.get(SOURCE))) { source = headers.get(SOURCE); } if (source != null) { logstashEvent.setSource(source); } String host = null; if (StringUtils.isNotBlank(headers.get(LogstashConstant.ATTRIBUTE_SOURCE_HOST))) { host = headers.get(LogstashConstant.ATTRIBUTE_SOURCE_HOST); } else if (StringUtils.isNotBlank(headers.get(HOST))) { host = headers.get(HOST); } if (host != null) { logstashEvent.setSourceHost(host); } String sourcePath = null; if (StringUtils.isNotBlank(headers.get(LogstashConstant.ATTRIBUTE_SOURCE_PATH))) { sourcePath = headers.get(LogstashConstant.ATTRIBUTE_SOURCE_PATH); } else if (StringUtils.isNotBlank(headers.get(SRC_PATH))) { sourcePath = headers.get(SRC_PATH); } if (sourcePath != null) { logstashEvent.setSourcePath(sourcePath); } String type = null; if (StringUtils.isNotBlank(headers.get(LogstashConstant.ATTRIBUTE_TYPE))) { type = headers.get(LogstashConstant.ATTRIBUTE_TYPE); } else if (StringUtils.isNotBlank(headers.get(TYPE))) { type = headers.get(TYPE); } if (type != null) { logstashEvent.setType(type); } String tags = null; if (StringUtils.isNotBlank(headers.get(LogstashConstant.ATTRIBUTE_TAGS))) { tags = headers.get(LogstashConstant.ATTRIBUTE_TAGS); } else if (StringUtils.isNotBlank(headers.get(TAGS))) { tags = headers.get(TAGS); } if (StringUtils.isNotBlank(tags)) { String[] allTags = tags.split(tagsSeparator); List<String> tagsList = new ArrayList<String>(); for(String tag: allTags) { if(StringUtils.isNotBlank(tag)) { tagsList.add(tag); } } logstashEvent.setTags(tagsList); } return logstashEvent; } }
LogstashSerializer implements Serializer { public LogstashEvent convertToLogstashEvent(Event event) throws RedisSerializerException { if (event == null) { throw new RedisSerializerException("Event cannot be null"); } LogstashEvent logstashEvent = new LogstashEvent(); final Map<String, String> headers = event.getHeaders(); logstashEvent.setMessage(new String(event.getBody())); for (Entry<String, String> entry : headers.entrySet()) { logstashEvent.putField(entry.getKey(), entry.getValue()); } String timestamp = null; if (StringUtils.isNotBlank(headers.get(LogstashConstant.ATTRIBUTE_AT_TIMESTAMP))) { timestamp = headers.get(LogstashConstant.ATTRIBUTE_AT_TIMESTAMP); } else if (StringUtils.isNotBlank(headers.get(TIMESTAMP))) { timestamp = headers.get(TIMESTAMP); } if (timestamp != null) { long timestampMs = Long.parseLong(timestamp); logstashEvent.setTimestamp(new Date(timestampMs)); } String source = null; if (StringUtils.isNotBlank(headers.get(LogstashConstant.ATTRIBUTE_SOURCE))) { source = headers.get(LogstashConstant.ATTRIBUTE_SOURCE); } else if (StringUtils.isNotBlank(headers.get(SOURCE))) { source = headers.get(SOURCE); } if (source != null) { logstashEvent.setSource(source); } String host = null; if (StringUtils.isNotBlank(headers.get(LogstashConstant.ATTRIBUTE_SOURCE_HOST))) { host = headers.get(LogstashConstant.ATTRIBUTE_SOURCE_HOST); } else if (StringUtils.isNotBlank(headers.get(HOST))) { host = headers.get(HOST); } if (host != null) { logstashEvent.setSourceHost(host); } String sourcePath = null; if (StringUtils.isNotBlank(headers.get(LogstashConstant.ATTRIBUTE_SOURCE_PATH))) { sourcePath = headers.get(LogstashConstant.ATTRIBUTE_SOURCE_PATH); } else if (StringUtils.isNotBlank(headers.get(SRC_PATH))) { sourcePath = headers.get(SRC_PATH); } if (sourcePath != null) { logstashEvent.setSourcePath(sourcePath); } String type = null; if (StringUtils.isNotBlank(headers.get(LogstashConstant.ATTRIBUTE_TYPE))) { type = headers.get(LogstashConstant.ATTRIBUTE_TYPE); } else if (StringUtils.isNotBlank(headers.get(TYPE))) { type = headers.get(TYPE); } if (type != null) { logstashEvent.setType(type); } String tags = null; if (StringUtils.isNotBlank(headers.get(LogstashConstant.ATTRIBUTE_TAGS))) { tags = headers.get(LogstashConstant.ATTRIBUTE_TAGS); } else if (StringUtils.isNotBlank(headers.get(TAGS))) { tags = headers.get(TAGS); } if (StringUtils.isNotBlank(tags)) { String[] allTags = tags.split(tagsSeparator); List<String> tagsList = new ArrayList<String>(); for(String tag: allTags) { if(StringUtils.isNotBlank(tag)) { tagsList.add(tag); } } logstashEvent.setTags(tagsList); } return logstashEvent; }  }
LogstashSerializer implements Serializer { public LogstashEvent convertToLogstashEvent(Event event) throws RedisSerializerException { if (event == null) { throw new RedisSerializerException("Event cannot be null"); } LogstashEvent logstashEvent = new LogstashEvent(); final Map<String, String> headers = event.getHeaders(); logstashEvent.setMessage(new String(event.getBody())); for (Entry<String, String> entry : headers.entrySet()) { logstashEvent.putField(entry.getKey(), entry.getValue()); } String timestamp = null; if (StringUtils.isNotBlank(headers.get(LogstashConstant.ATTRIBUTE_AT_TIMESTAMP))) { timestamp = headers.get(LogstashConstant.ATTRIBUTE_AT_TIMESTAMP); } else if (StringUtils.isNotBlank(headers.get(TIMESTAMP))) { timestamp = headers.get(TIMESTAMP); } if (timestamp != null) { long timestampMs = Long.parseLong(timestamp); logstashEvent.setTimestamp(new Date(timestampMs)); } String source = null; if (StringUtils.isNotBlank(headers.get(LogstashConstant.ATTRIBUTE_SOURCE))) { source = headers.get(LogstashConstant.ATTRIBUTE_SOURCE); } else if (StringUtils.isNotBlank(headers.get(SOURCE))) { source = headers.get(SOURCE); } if (source != null) { logstashEvent.setSource(source); } String host = null; if (StringUtils.isNotBlank(headers.get(LogstashConstant.ATTRIBUTE_SOURCE_HOST))) { host = headers.get(LogstashConstant.ATTRIBUTE_SOURCE_HOST); } else if (StringUtils.isNotBlank(headers.get(HOST))) { host = headers.get(HOST); } if (host != null) { logstashEvent.setSourceHost(host); } String sourcePath = null; if (StringUtils.isNotBlank(headers.get(LogstashConstant.ATTRIBUTE_SOURCE_PATH))) { sourcePath = headers.get(LogstashConstant.ATTRIBUTE_SOURCE_PATH); } else if (StringUtils.isNotBlank(headers.get(SRC_PATH))) { sourcePath = headers.get(SRC_PATH); } if (sourcePath != null) { logstashEvent.setSourcePath(sourcePath); } String type = null; if (StringUtils.isNotBlank(headers.get(LogstashConstant.ATTRIBUTE_TYPE))) { type = headers.get(LogstashConstant.ATTRIBUTE_TYPE); } else if (StringUtils.isNotBlank(headers.get(TYPE))) { type = headers.get(TYPE); } if (type != null) { logstashEvent.setType(type); } String tags = null; if (StringUtils.isNotBlank(headers.get(LogstashConstant.ATTRIBUTE_TAGS))) { tags = headers.get(LogstashConstant.ATTRIBUTE_TAGS); } else if (StringUtils.isNotBlank(headers.get(TAGS))) { tags = headers.get(TAGS); } if (StringUtils.isNotBlank(tags)) { String[] allTags = tags.split(tagsSeparator); List<String> tagsList = new ArrayList<String>(); for(String tag: allTags) { if(StringUtils.isNotBlank(tag)) { tagsList.add(tag); } } logstashEvent.setTags(tagsList); } return logstashEvent; }  void configure(Context context); void configure(ComponentConfiguration conf); LogstashEvent convertToLogstashEvent(Event event); byte[] serialize(Event event); }
LogstashSerializer implements Serializer { public LogstashEvent convertToLogstashEvent(Event event) throws RedisSerializerException { if (event == null) { throw new RedisSerializerException("Event cannot be null"); } LogstashEvent logstashEvent = new LogstashEvent(); final Map<String, String> headers = event.getHeaders(); logstashEvent.setMessage(new String(event.getBody())); for (Entry<String, String> entry : headers.entrySet()) { logstashEvent.putField(entry.getKey(), entry.getValue()); } String timestamp = null; if (StringUtils.isNotBlank(headers.get(LogstashConstant.ATTRIBUTE_AT_TIMESTAMP))) { timestamp = headers.get(LogstashConstant.ATTRIBUTE_AT_TIMESTAMP); } else if (StringUtils.isNotBlank(headers.get(TIMESTAMP))) { timestamp = headers.get(TIMESTAMP); } if (timestamp != null) { long timestampMs = Long.parseLong(timestamp); logstashEvent.setTimestamp(new Date(timestampMs)); } String source = null; if (StringUtils.isNotBlank(headers.get(LogstashConstant.ATTRIBUTE_SOURCE))) { source = headers.get(LogstashConstant.ATTRIBUTE_SOURCE); } else if (StringUtils.isNotBlank(headers.get(SOURCE))) { source = headers.get(SOURCE); } if (source != null) { logstashEvent.setSource(source); } String host = null; if (StringUtils.isNotBlank(headers.get(LogstashConstant.ATTRIBUTE_SOURCE_HOST))) { host = headers.get(LogstashConstant.ATTRIBUTE_SOURCE_HOST); } else if (StringUtils.isNotBlank(headers.get(HOST))) { host = headers.get(HOST); } if (host != null) { logstashEvent.setSourceHost(host); } String sourcePath = null; if (StringUtils.isNotBlank(headers.get(LogstashConstant.ATTRIBUTE_SOURCE_PATH))) { sourcePath = headers.get(LogstashConstant.ATTRIBUTE_SOURCE_PATH); } else if (StringUtils.isNotBlank(headers.get(SRC_PATH))) { sourcePath = headers.get(SRC_PATH); } if (sourcePath != null) { logstashEvent.setSourcePath(sourcePath); } String type = null; if (StringUtils.isNotBlank(headers.get(LogstashConstant.ATTRIBUTE_TYPE))) { type = headers.get(LogstashConstant.ATTRIBUTE_TYPE); } else if (StringUtils.isNotBlank(headers.get(TYPE))) { type = headers.get(TYPE); } if (type != null) { logstashEvent.setType(type); } String tags = null; if (StringUtils.isNotBlank(headers.get(LogstashConstant.ATTRIBUTE_TAGS))) { tags = headers.get(LogstashConstant.ATTRIBUTE_TAGS); } else if (StringUtils.isNotBlank(headers.get(TAGS))) { tags = headers.get(TAGS); } if (StringUtils.isNotBlank(tags)) { String[] allTags = tags.split(tagsSeparator); List<String> tagsList = new ArrayList<String>(); for(String tag: allTags) { if(StringUtils.isNotBlank(tag)) { tagsList.add(tag); } } logstashEvent.setTags(tagsList); } return logstashEvent; }  void configure(Context context); void configure(ComponentConfiguration conf); LogstashEvent convertToLogstashEvent(Event event); byte[] serialize(Event event); static final String CONFIGURATION_TAGS_SEPARATOR; static final String HOST; static final String SOURCE; static final String SRC_PATH; static final String TIMESTAMP; static final String TYPE; static final String TAGS; static final String DEFAULT_TAGS_SEPARATOR; }