Reply
Highlighted
New Contributor
Posts: 1
Registered: ‎02-08-2018

morphline interceptor not work

[ Edited ]

Hello guys!!

 

I want to set a morphline interceptor in an agent wich send the dato to collector agent allocate in cloudera server.

But i have a the following problem:

 

 

tParser.hasAtLeastOneAttachment:164) - Command failed because of missing attachment for record: {hostname=[magnifix], plataforma=[linux]}
01 feb 2018 15:23:34,420 DEBUG [PollableSourceRunner-TaildirSource-linuxLog] (org.kitesdk.morphline.base.AbstractCommand.process:158) - Command failed!
01 feb 2018 15:23:34,420 DEBUG [PollableSourceRunner-TaildirSource-linuxLog] (org.kitesdk.morphline.base.AbstractCommand.process:158) - Command failed!
01 feb 2018 15:23:34,421 WARN [PollableSourceRunner-TaildirSource-linuxLog] (org.apache.flume.sink.solr.morphline.MorphlineHandlerImpl.process:132) - Morphline /home/flume/pruebaMorphline.conf@preMorphlineLinuxLog failed to process record: {hostname=[magnifix], plataforma=[linux]}

 

 

My agent configuration is:

 

############################################################################################
### LINUX-LOG ####
############################################################################################


# ------------------------- Declaracion sources, channels y sinks ------------------------ #

sourceAgent.sources = linuxLog
sourceAgent.channels = fileChannelLinux
sourceAgent.sinks = avroSinkLinux1
sourceAgent.sinkgroups = group


# --------------------------- Configuracion source linux-log ----------------------------- #

sourceAgent.sources.linuxLog.type = TAILDIR
sourceAgent.sources.linuxLog.positionFile = /home/flume/flume-metadata/flumedatadirLinuxLog/position_log_linux-events.json
sourceAgent.sources.linuxLog.filegroups = linuxLogs
sourceAgent.sources.linuxLog.filegroups.linuxLogs = /var/log/pruebas
sourceAgent.sources.linuxLog.filegroups.linuxLogs.headerkey=linuxLog


# ----------------------------- Configuracion de interceptores ---------------------------- #

#Definicion interceptores para adjuntar cabezeras al mensaje
sourceAgent.sources.linuxLog.interceptors = regex hostInterceptor plataforma morphlineinterceptor

sourceAgent.sources.linuxLog.interceptors.morphlineinterceptor.type = org.apache.flume.sink.solr.morphline.MorphlineInterceptor$Builder
sourceAgent.sources.linuxLog.interceptors.morphlineinterceptor.morphlineFile = /home/flume/pruebaMorphline.conf
sourceAgent.sources.linuxLog.interceptors.morphlineinterceptor.morphlineId = preMorphlineLinuxLog

sourceAgent.sources.linuxLog.interceptors.regex.type = regex_filter
sourceAgent.sources.linuxLog.interceptors.regex.regex = .*\\.(err|error|crit|alert|emerg|panic):\\s.*
sourceAgent.sources.linuxLog.interceptors.regex.excludeEvents = false

 

#Definicion interceptor que agrega el nombre del host
sourceAgent.sources.linuxLog.interceptors.hostInterceptor.type = org.apache.flume.interceptor.HostInterceptor$Builder
sourceAgent.sources.linuxLog.interceptors.hostInterceptor.hostHeader = hostname
sourceAgent.sources.linuxLog.interceptors.hostInterceptor.preserveExisting = false
sourceAgent.sources.linuxLog.interceptors.hostInterceptor.useIP = false

 

#Definicion interceptor que agrega el nombre de la plataforma a la cabezera
sourceAgent.sources.linuxLog.interceptors.plataforma.type = static
sourceAgent.sources.linuxLog.interceptors.plataforma.preserveExisting = true
sourceAgent.sources.linuxLog.interceptors.plataforma.key = plataforma
sourceAgent.sources.linuxLog.interceptors.plataforma.value = linux


# -------------------------------- Configuracion de los sink ------------------------------ #

sourceAgent.sinks.avroSinkLinux1.type = avro
sourceAgent.sinks.avroSinkLinux1.hostname = 172.36.1.32
sourceAgent.sinks.avroSinkLinux1.port = 9997

 

# ---------------------------- Configuracion de los channels ------------------------------ #

# Configure channel Linux-Log
sourceAgent.channels.fileChannelLinux.type = file
sourceAgent.channels.fileChannelLinux.checkpointDir = /home/flume/flume-metadata/flumecheckdirLinuxLog/
sourceAgent.channels.fileChannelLinux.dataDirs = /home/flume/flume-metadata/flumedatadirLinuxLog/


# ------------------- Union de los sink con el source y el channel ------------------------ #

sourceAgent.sources.linuxLog.channels = fileChannelLinux
sourceAgent.sinks.avroSinkLinux1.channel = fileChannelLinux

 

And the morphline is:

 

 

 

# Created by jromero

morphlines : [
{
# Morphline usado para agregar el campo de detail que es usado por el
id : preMorphlineLinuxLog

# Import all morphline commands in these java packages and their subpackages.
# Other commands that may be present on the classpath are not visible to this morphline.
importCommands : ["org.kitesdk.**", "org.apache.solr.**","com.cloudera.**"]

commands : [

{
readLine {
charset : UTF-8
}
}

{ addValuesIfAbsent {
bodybak : "@{_attachment_body}"
}
}


{
grok {

dictionaryFiles : ["/home/flume/grok-patterns"]

expressions : {
message : """%{TIMESTAMP_ISO8601}-%{INT}:%{INT} %{HOSTNAME} %{PROG}.%{WORD}%{GREEDYDATA:detalle}"""
}

}
}

{
logDebug {
format : "output record: {}", args : ["@{}"]
}
}
]
}
]

 

 

I hava try several options like:

-add java code in morphiline to try to agregate the _attachment_body field .

-modify de source code of the morphline core.

-Change the library versions

 

No one of this option work!! . I need their help !! Please Guys its necesary for me. Thanks !!!

 

 

 

 

 

 

 

Announcements
The Kite SDK is a collection of docs, sample code, APIs, and tools to make Hadoop application development faster. Learn more at http://kitesdk.org.