источник фильтра logstash json

Я не могу заставить поле сообщения декодировать из моей строки журнала json при получении через filebeat.

Вот строка в моих журналах:

{"levelname": "WARNING", "asctime": "2016-07-01 18:06:37", "message": "One or more gateways are offline", "name": "ep.management.commands.import", "funcName": "check_gateway_online", "lineno": 103, "process": 44551, "processName": "MainProcess", "thread": 140735198597120, "threadName": "MainThread", "server": "default"}

Здесь файл logstash config. Я пробовал с codec и без него. Единственная разница в том, что сообщение экранируется, когда я использую кодек.

 input {
  beats {
    port => 5044   
    codec => "json"
  }
}

filter {
 json{
 source => "message"
 }
}

Вот json в том виде, в каком он поступает в elasticsearch:

    {
  "_index": "filebeat-2016.07.01",
  "_type": "json",
  "_id": "AVWnpK519vJkh3Ry-Q9B",
  "_score": null,
  "_source": {
    "@timestamp": "2016-07-01T18:07:13.522Z",
    "beat": {
      "hostname": "59b378d40b2e",
      "name": "59b378d40b2e"
    },
    "count": 1,
    "fields": null,
    "input_type": "log",
    "message": "{\"levelname\": \"WARNING\", \"asctime\": \"2016-07-01 18:07:12\", \"message\": \"One or more gateways are offline on server default\", \"name\": \"ep.controllers.secure_client\", \"funcName\": \"check_gateways_online\", \"lineno\": 80, \"process\": 44675, \"processName\": \"MainProcess\", \"thread\": 140735198597120, \"threadName\": \"MainThread\"}",
    "offset": 251189,
    "source": "/mnt/ep_logs/ep_.json",
    "type": "json"
  },
  "fields": {
    "@timestamp": [
      1467396433522
    ]
  },
  "sort": [
    1467396433522
  ]
}

Я бы хотел, чтобы содержимое объекта message декодировалось.

Большое спасибо