Pregunta Logstash analizando el documento xml que contiene varias entradas de registro


Actualmente estoy evaluando si logstash y elasticsearch son útiles para nuestro caso de uso. Lo que tengo es un archivo de registro que contiene entradas múltiples que es de la forma

<root>
    <entry>
        <fieldx>...</fieldx>
        <fieldy>...</fieldy>
        <fieldz>...</fieldz>
        ...
        <fieldarray>
            <fielda>...</fielda>
            <fielda>...</fielda>
            ...
        </fieldarray>
    </entry>
    <entry>
    ...
    </entry>
    ...
<root>

Cada entry El elemento contendría un evento de registro. (Si está interesado, el archivo es en realidad una exportación de registro de trabajo de Tempo Timesheets (un complemento JIRA de Atlassian)).

¿Es posible transformar un archivo de este tipo en múltiples eventos de registro sin escribir mi propio códec?


8
2017-07-24 14:14


origen




Respuestas:


Muy bien, he encontrado una solución que funciona para mí. El mayor problema con la solución es que el complemento XML no es ... bastante inestable, pero está mal documentado y con errores o está mal documentado e incorrectamente.

TLDR

Línea de comando Bash:

gzcat -d file.xml.gz | tr -d "\n\r" | xmllint --format - | logstash -f logstash-csv.conf

Configuración de logstash:

input {
    stdin {}
}

filter {
    # add all lines that have more indentation than double-space to the previous line
    multiline {
        pattern => "^\s\s(\s\s|\<\/entry\>)"
        what => previous
    }
    # multiline filter adds the tag "multiline" only to lines spanning multiple lines
    # We _only_ want those here.
    if "multiline" in [tags] {
        # Add the encoding line here. Could in theory extract this from the
        # first line with a clever filter. Not worth the effort at the moment.
        mutate {
            replace => ["message",'<?xml version="1.0" encoding="UTF-8" ?>%{message}']
        }
        # This filter exports the hierarchy into the field "entry". This will
        # create a very deep structure that elasticsearch does not really like.
        # Which is why I used add_field to flatten it.
        xml {
            target => entry
            source => message
            add_field => {
                fieldx         => "%{[entry][fieldx]}"
                fieldy         => "%{[entry][fieldy]}"
                fieldz         => "%{[entry][fieldz]}"
                # With deeper nested fields, the xml converter actually creates
                # an array containing hashes, which is why you need the [0]
                # -- took me ages to find out.
                fielda         => "%{[entry][fieldarray][0][fielda]}"
                fieldb         => "%{[entry][fieldarray][0][fieldb]}"
                fieldc         => "%{[entry][fieldarray][0][fieldc]}"
            }
        }
        # Remove the intermediate fields before output. "message" contains the
        # original message (XML). You may or may-not want to keep that.
        mutate {
            remove_field => ["message"]
            remove_field => ["entry"]
        }
    }
}

output {
    ...
}

Detallado

Mi solución funciona porque al menos hasta el entry nivel, mi entrada XML es muy Uniforme y por lo tanto puede ser manejado por algún tipo de coincidencia de patrón.

Dado que la exportación es básicamente una línea muy larga de XML, y el complemento xml logstash esencialmente funciona solo con campos (leídos: columnas en líneas) que contienen datos XML, tuve que cambiar los datos a un formato más útil.

Shell: Preparando el archivo

  • gzcat -d file.xml.gz |: Era demasiada información, obviamente puedes omitir eso
  • tr -d "\n\r" |: Eliminar saltos de línea dentro de elementos XML: Algunos de los elementos pueden contener saltos de línea como datos de caracteres. El siguiente paso requiere que estos son eliminados, o codificados de alguna manera. Aunque asumió que en este punto tiene todo el código XML en una línea masiva, no importa si este comando elimina cualquier espacio en blanco entre los elementos.

  • xmllint --format - |: Formatea el XML con xmllint (viene con libxml)

    Aquí la única gran línea de espaguetis de XML (<root><entry><fieldx>...</fieldx></entry></root>) Tiene el formato correcto:

    <root>
      <entry>
        <fieldx>...</fieldx>
        <fieldy>...</fieldy>
        <fieldz>...</fieldz>
        <fieldarray>
          <fielda>...</fielda>
          <fieldb>...</fieldb>
          ...
        </fieldarray>
      </entry>
      <entry>
        ...
      </entry>
      ...
    </root>
    

Logstash

logstash -f logstash-csv.conf

(Ver contenido completo de la .conf archivo en la sección TL; DR.)

Aquí el multiline filtro hace el truco Puede combinar varias líneas en un solo mensaje de registro. Y es por esto que el formateo con xmllint fue necesario:

filter {
    # add all lines that have more indentation than double-space to the previous line
    multiline {
        pattern => "^\s\s(\s\s|\<\/entry\>)"
        what => previous
    }
}

Esto básicamente dice que cada línea con sangría que es más de dos espacios (o es </entry> / xmllint hace sangría con dos espacios por defecto) pertenece a una línea anterior. Esto también significa que los datos de caracteres no deben contener nuevas líneas (eliminadas con tr en shell) y que el xml debe estar normalizado (xmllint)


11
2017-08-06 10:41



Hola, ¿lograste hacer este trabajo? Tengo curiosidad ya que tengo una necesidad similar y la solución multilínea junto con la división no funcionó para mí. Gracias por sus comentarios - viz
@viz Esto funcionó, pero nunca lo usamos en producción. La línea múltiple solo funciona si tiene una estructura XML muy regular y la ha formateado primero con sangría (consulte la respuesta en la sección "Preparación del archivo") - dualed


Tuve un caso similar. Para analizar este xml:

<ROOT number="34">
  <EVENTLIST>
    <EVENT name="hey"/>
    <EVENT name="you"/>
  </EVENTLIST>
</ROOT>

Yo uso esta configuración para logstash:

input {
  file {
    path => "/path/events.xml"
    start_position => "beginning"
    sincedb_path => "/dev/null"
    codec => multiline {
      pattern => "<ROOT"
      negate => "true"
      what => "previous"
      auto_flush_interval => 1
    }
  }
}
filter {
  xml {
    source => "message"
    target => "xml_content"
  }
  split {
    field => "xml_content[EVENTLIST]"
  }
  split {
    field => "xml_content[EVENTLIST][EVENT]"
  }
  mutate {
    add_field => { "number" => "%{xml_content[number]}" }
    add_field => { "name" => "%{xml_content[EVENTLIST][EVENT][name]}" }
    remove_field => ['xml_content', 'message', 'path']
  }
}
output {
  stdout {
    codec => rubydebug
  }
}

Espero que esto pueda ayudar a alguien. He necesitado mucho tiempo para conseguirlo.


1
2017-12-23 12:14