Разделить запись журнала в двух событиях с помощью logstash-grok, клонировать и мутировать

Question

Я пытаюсь разбить журнал на два события с grok, mutate и clone в моей конфигурации logstash. Мой стек очень стандартный ELK (Elasticsearch, Logstash, Kibana).

Я создаю журнал, который отформатирован так:

timestamp float integer 

Например:

2015/01/19 21:48:12 24.7 32 
2015/01/19 22:00:20 24.7 32 
2015/01/19 22:01:11 24.7 32 
2015/01/19 22:01:58 24.7 28 
2015/01/19 22:02:28 23.7 28 
(etc ...) 

В конце концов, я хочу два события в logstash, одна с ключами timestamp очевидно, type=sensorA и value=%{the value of the integer}, а другой с той же отметкой времени и value=%{value of the float} и type=sensorB.

До сих пор, я пришел с этой конфигурацией в logstash.conf:

1.My журнал помечается type=sensor в моих входов:

input { 
    file { 
     path => "/var/log/sensors.log" 
     type => "sensor" 
    } 
    } 

2.Then, я использую Grok, клонировать и мутировать, чтобы попытаться разделить их

if [type] == "sensor" { 
    # Extracts the values 
    grok { 
    match => { "message" => "(?<timestamp>%{YEAR}/%{MONTHNUM:month}/%{MONTHDAY:day} %{TIME}) %{NUMBER:sensorA:float} %{NUMBER:sensorB:int}" } 
    } 
    mutate { 
    update => [ "type", "sensorA" ] 
    } 
    # Clones the event 
    clone { 
    clones => ["sensorB"] 
    } 
} 
# So now I should have two events, 
# one with type sensorA and one with type sensorB, no ? : 
if [type] == "sensorA" { 
    mutate { 
    add_field => { "value" => "%{sensorA}" } 
    convert => ["value", "float"] 
    } 
} 
if [type] == "sensorB" { 
    mutate { 
    add_field => { "value" => "%{sensorB}" } 
    convert => ["value", "integer"] 
    } 
} 

Но что на самом деле не работает, так как Eventhough я получаю два события с другим типом, они оба имеют одинаковое значение (которое всегда соответствует датчику B).

Почему? У меня возникло ощущение, что файл logstash.conf на самом деле не читается линейным способом, но я не могу найти для этого никакого решения.

Подсказка? Я пропустил что-то действительно очевидное здесь? Спасибо большое

Answer 1

Вместо использования мутата для преобразования значения в поле value вы можете использовать плагин ruby, чтобы сделать то, что вам нужно.

if [type] == "sensor" { 
    # Extracts the values 
    grok { 
     match => { "message" => "(?<timestamp>%{YEAR}/%{MONTHNUM:month}/%{MONTHDAY:day} %{TIME}) %{NUMBER:sensorA:float} %{NUMBER:sensorB:int}" } 
    } 
    mutate { 
     update => [ "type", "sensorA" ] 
    } 
    # Clones the event 
    clone { 
     clones => ["sensorB"] 
    } 
} 
# So now I should have two events, 
# one with type sensorA and one with type sensorB, no ? : 
ruby { 
    code => " 
     if event['type'] == 'sensorA' 
       event['value'] = event['sensorA'] 
     elsif event['type'] == 'sensorB' 
       event['value'] = event['sensorB'] 
     end 
    " 
} 

С помощью этой конфигурации я могу удовлетворить ваши требования. Надеюсь, это вам поможет :)