Logstash 中的 JSON 转义

Question

我通过 TCP 将一些 JSON 数据从 Java 服务器发送到 Logstash（Logstash 将它们发送到 Elasticsearch），这些 JSON 数据似乎在 Elastic 中转义了。

Java 序列化：

Map<String, Object> jsonMap = new HashMap<>();
jsonMap.put("age", event.getAge());
for (Entry<String, Serializable> attribute : event.getAttributes().entrySet()) {
    jsonMap.put("attribute_" + attribute.getKey(), attribute.getValue());
}
jsonMap.put("message", event.getMessage());
jsonMap.put("cause", event.getCause());
jsonMap.put("timestamp", event.getTimestamp());
jsonMap.put("eventid", event.getEventId());
jsonMap.put("instanceid", event.getInstanceId());
jsonMap.put("origin", event.getOrigin());
jsonMap.put("severity", event.getSeverity());
jsonMap.put("durability", event.getDurability());
jsonMap.put("detail", event.getDetail());
int i = 0;
for (String tag : event.getTags()) {
    jsonMap.put("tag_" + String.valueOf(i), tag);
    i++;
}

return new JSONObject(jsonMap).toString();

Java 套接字：

try (Socket clientSocket = new Socket(url, port);
    OutputStreamWriter out = new OutputStreamWriter(
        clientSocket.getOutputStream(), "UTF-8")) {
    out.write(content.toString());
    out.flush();
}

Elastic 中的示例数据：

"message": "{\"detail\":null,\"cause\":null,\"attribute_start\":\"Mon Jan 11 16:15:28 CET 2016\",\"durability\":\"MOMENTARY\",\"attribute_login\":\"\",\"origin\":\"fortuna.ws.navipro\",\"severity\":\"ERROR\",\"attribute_type\":null,\"attribute_methodName\":\"Logout\",\"eventid\":\"ws.navipro.call\",\"attribute_call\":\"[57,7256538816272415441,,OK]{0 connections} CZ() Calling method 'Logout' at 1452525329029(Mon Jan 11 16:15:28 CET 2016-Mon Jan 11 16:15:29 CET 2016; roundtrip=36ms):\\n\\tRequest\\n\\t\\tCLASS com.etnetera.projects.jnp.fortuna.navipro.ws.ClientLogoutRequest\\n\\t\\tkeep: true\\n\\t\\tCLASS com.etnetera.projects.jnp.fortuna.navipro.ws.RequestBody\\n\\t\\tcountry: CZ\\n\\t\\tsessionCC: NULL\\n\\t\\tsessionID:\\n\\t\\tsessionIP:\\n\\t\\tdebug: NULL\\n\\t\\tns: NULL\\n\\t\\tCLASS com.etnetera.projects.jnp.fortuna.navipro.ws.RequestCorpus\\n\\tResponse\\n\\t\\tCLASS com.etnetera.projects.jnp.fortuna.navipro.ws.ClientLogoutResponse\\n\\t\\tCLASS com.etnetera.projects.jnp.fortuna.navipro.ws.ResponseBody\\n\\t\\tCLASS com.etnetera.projects.jnp.fortuna.navipro.ws.ResponseCorpus\\n\\t\\tmessage: \\n\\t\\t[1] \\n\\t\\t\\tCLASS com.etnetera.projects.jnp.fortuna.navipro.ws.Message\\n\\t\\t\\tparam: \\n\\t\\t\\t[1] \\n\\t\\t\\t\\tCLASS com.etnetera.projects.jnp.fortuna.navipro.ws.Message$Param\\n\\t\\t\\t\\tindex: 0\\n\\t\\t\\t\\ttype: NULL\\n\\t\\t\\t\\tvalue: 3\\n\\t\\t\\tid: 104\\n\\t\\t\\tseverity: NOTIFICATION\\n\\t\\t\\tlink: NULL\\n\\t\\tentryLink: NULL\\n\\t\\thint: NULL\\n\\t\\thintType: NULL\\n\\t\\tstatus: OK\\nEND\",\"timestamp\":1452525329030,\"message\":\"NaviPro method Logoutcalled.\",\"tag_1\":\"NaviPro\",\"attribute_end\":\"Mon Jan 11 16:15:29 CET 2016\",\"attribute_sessionId\":\"\",\"age\":0,\"tag_0\":\"Logout\",\"instanceid\":\"Logout\",\"attribute_address\":\""}"

Logstash 配置：

input {
  syslog {
    port => 1514
  }
  tcp {
    port => 3333
  }
}

filter {
  if [type] == "docker" {
    json {
      source => "message"
    }
    mutate {
      rename => [ "log", "message" ]
    }
    date {
      match => [ "time", "ISO8601" ]
    }
  }
}

output {
  elasticsearch {
    hosts => "elasticsearch:9200"
  }
}

我希望将 Elastic 中的数据作为 JSON 格式，以便在 Kibana 中过滤字段。

编辑：

如果我尝试将配置更改为：

input {
  tcp {
    port => 3333
    codec => json
  }
}

Logstash 拒绝在日志中使用此行启动：

logstash_1 | {:timestamp=>"2016-01-13T10:13:58.583000+0000", :message=>"SIGTERM received. Shutting down the pipeline.", :level=>:warn}

Answer 1

问题在于 Logstash 配置不正确。 Logstash 正在运行，但没有通过过滤器发送任何内容。这是正确的配置：

input {
  tcp {
    port => 3333
    type => "java"
  }
}

filter {
  if [type] == "java" {
    json {
      source => "message"
    }
  }
}

output {
  elasticsearch {
    hosts => "elasticsearch:9200"
  }
}

Answer 2

我创建了一个简单的测试，您需要添加到您的 logstash 配置中的只是 codec => json。默认值为"line" 并将转义字符串中的字符。

input {
  tcp {
    port => 3333
    codec => json
  }
}

output {
  stdout { 
    codec => rubydebug 
  }

  elasticsearch {
    hosts => "elasticsearch:9200"
  }
}