Processar a data de registo como data-limite, usando o filtro de datas

Bem, depois de olhar muito em volta, não consegui encontrar uma solução para o meu problema, pois "deveria" funcionar, mas obviamente não funciona. Estou a usar um Logstash de máquina Ubuntu 14.04 LTS 1.4.2-1-2-2c0f5a1, e eu estou recebendo mensagens como a seguinte:

2014-08-05 10:21:13,618 [17] INFO  Class.Type - This is a log message from the class:
  BTW, I am also multiline

na configuração de entrada, tenho um codec multiline e o evento é processado correctamente. Eu também separo o texto do evento em várias partes para que seja mais fácil de ler.

No final, eu obtenho, como visto em Kibana, algo como o seguinte (JSON view):

{
  "_index": "logstash-2014.08.06",
  "_type": "customType",
  "_id": "PRtj-EiUTZK3HWAm5RiMwA",
  "_score": null,
  "_source": {
    "@timestamp": "2014-08-06T08:51:21.160Z",
    "@version": "1",
    "tags": [
      "multiline"
    ],
    "type": "utg-su",
    "host": "ubuntu-14",
    "path": "/mnt/folder/thisIsTheLogFile.log",
    "logTimestamp": "2014-08-05;10:21:13.618",
    "logThreadId": "17",
    "logLevel": "INFO",
    "logMessage": "Class.Type - This is a log message from the class:\r\n  BTW, I am also multiline\r"
  },
  "sort": [
    "21",
    1407315081160
  ]
}

Deves ter reparado que pus um"; " no intervalo de tempo. A razão é que eu quero ser capaz de classificar os logs usando a string timestamp, e aparentemente logstash não é tão bom nisso (por exemplo: http://www.elasticsearch.org/guide/en/elasticsearch/guide/current/multi-fields.html).

Eu tentei sem sucesso usar o filtro {[[5]} de várias maneiras, e aparentemente não trabalhar.

date {
            locale => "en"
            match => ["logTimestamp", "YYYY-MM-dd;HH:mm:ss.SSS", "ISO8601"]
            timezone => "Europe/Vienna"
            target => "@timestamp"
            add_field => { "debug" => "timestampMatched"}
        }

, uma vez que li que o Joda biblioteca pode ter problemas se a seqüência de caracteres não é estritamente ISO 8601-compatível (muito exigente e espera um T, https://logstash.jira.com/browse/LOGSTASH-180), eu também tentei usar mutate para converter a seqüência de caracteres para algo como 2014-08-05T10:21:13.618 e, em seguida, use "YYYY-MM-dd'T'HH:mm:ss.SSS". Isso também não funcionou.

Eu não quero ter que colocar manualmente um +02: 00 no tempo, porque isso daria problemas com o horário de Verão.

em qualquer um dos estes casos, o evento vai para a pesquisa elastica, mas date aparentemente não faz nada, pois @timestamp e logTimestamp são diferentes e nenhum campo debug é adicionado.

Alguma ideia de como posso tornar as cordas do logTime adequadamente atracáveis? Concentrei-me em convertê-los a um horário adequado, mas qualquer outra solução seria também bem-vinda.

Como pode ver abaixo:

ao classificar sobre @timestamp, elasticsearch pode fazê-lo corretamente, mas uma vez que este não é o "real" tempo de registo, mas sim melhor quando o evento logstash foi lido, eu preciso (obviamente) ser capaz de classificar também sobre logTimestamp. Isto é o que então é saída. Obviamente não é assim tão útil.

Qualquer ajuda é bem-vinda! Avisa-me se me esquecer de alguma informação que possa ser útil.

actualizar:

Aqui está o ficheiro de configuração do filtro que finalmente funcionou:

# Filters messages like this:
# 2014-08-05 10:21:13,618 [17] INFO  Class.Type - This is a log message from the class:
#  BTW, I am also multiline

# Take only type- events (type-componentA, type-componentB, etc)
filter {
    # You cannot write an "if" outside of the filter!
    if "type-" in [type] {
        grok {
            # Parse timestamp data. We need the "(?m)" so that grok (Oniguruma internally) correctly parses multi-line events
            patterns_dir => "./patterns"
            match => [ "message", "(?m)%{TIMESTAMP_ISO8601:logTimestampString}[ ;]\[%{DATA:logThreadId}\][ ;]%{LOGLEVEL:logLevel}[ ;]*%{GREEDYDATA:logMessage}" ]
        }

        # The timestamp may have commas instead of dots. Convert so as to store everything in the same way
        mutate {
            gsub => [
                # replace all commas with dots
                "logTimestampString", ",", "."
                ]
        }

        mutate {
            gsub => [
                # make the logTimestamp sortable. With a space, it is not! This does not work that well, in the end
                # but somehow apparently makes things easier for the date filter
                "logTimestampString", " ", ";"
                ]
        }

        date {
            locale => "en"
            match => ["logTimestampString", "YYYY-MM-dd;HH:mm:ss.SSS"]
            timezone => "Europe/Vienna"
            target => "logTimestamp"
        }
    }
}

filter {
    if "type-" in [type] {
        # Remove already-parsed data
        mutate {
            remove_field => [ "message" ]
        }
    }
}