@@ -0,0 +1,1519 @@
+package com.opensoc.parsing.parsers;%0A%0Aimport java.io.UnsupportedEncodingException;%0A%0Aimport org.json.simple.JSONObject;%0Aimport org.json.simple.parser.JSONParser;%0Aimport org.json.simple.parser.ParseException;%0A%0Apublic class BasicLogstashParser extends AbstractParser %7B%0A%0A%09@Override%0A%09public JSONObject parse(byte%5B%5D raw_message) %7B%0A%09%09%0A%09%09try %7B%0A%09%09%09%0A%09%09%09/*%0A%09%09%09 * We need to create a new JSONParser each time because its %0A%09%09%09 * not serializable and the parser is created on the storm nimbus%0A%09%09%09 * node, then transfered to the workers.%0A%09%09%09 */%0A%09%09%09JSONParser jsonParser = new JSONParser();%0A%09%09%09String rawString = new String(raw_message, %22UTF-8%22);%0A%09%09%09JSONObject rawJson = (JSONObject) jsonParser.parse(rawString);%0A%09%09%09%0A%09%09%09// remove logstash meta fields%0A%09%09%09rawJson.remove(%22@vesrion%22);%0A%09%09%09rawJson.remove(%22type%22);%0A%09%09%09rawJson.remove(%22host%22);%0A%09%09%09%0A%09%09%09// rename other keys%0A%09%09%09rawJson = mutate(rawJson, %22@timestamp%22, %22timestamp%22);%0A%09%09%09rawJson = mutate(rawJson, %22message%22, %22original_string%22);%0A%09%09%09rawJson = mutate(rawJson, %22src_ip%22, %22ip_src_addr%22);%0A%09%09%09rawJson = mutate(rawJson, %22dst_ip%22, %22ip_dst_addr%22);%0A%09%09%09rawJson = mutate(rawJson, %22src_port%22, %22ip_src_port%22);%0A%09%09%09rawJson = mutate(rawJson, %22dst_port%22, %22ip_dst_port%22);%0A%09%09%09rawJson = mutate(rawJson, %22src_ip%22, %22ip_src_addr%22);%0A%0A%09%09%09return rawJson;%0A%09%09%7D catch (Exception e) %7B%0A%09%09%09e.printStackTrace();%0A%09%09%09return null;%0A%09%09%7D%0A%09%09%0A%09%7D%0A%09%0A%09private JSONObject mutate(JSONObject json, String oldKey, String newKey) %7B%0A%09%09if (json.containsKey(oldKey)) %7B%0A%09%09%09json.put(newKey, json.remove(oldKey));%0A%09%09%7D%0A%09%09return json;%0A%09%7D%0A%0A%7D%0A

add BasicLogstashParser