logstash的ruby模块怎么用

今天小编给大家分享一下logstash的ruby模块怎么用的相关知识点，内容详细，逻辑清晰，相信大部分人都还太了解这方面的知识，所以分享这篇文章给大家参考一下，希望大家阅读完这篇文章后有所收获，下面我们一起来了解一下吧。

可以执行任意ruby代码.可是真正去实现的时候踩了不少吭,诸多报错. 第一步调用ruby脚本的时候就一直提示"Script does not define a filter! Please ensure that you have defined a filter method!,根本不加载指定的函数.翻阅logstash-filter-ruby源码得知

其方法中必须有filter函数,固将函数名称更改成filter后解决,但是后续提示没有定义event ,继续翻阅源码看到这个

在调用自定义ruby脚本的时候将event传递过去了.于是将event更改成参数传递解决.最终生产配置如下:

filter{  #执行解析逻辑  ruby{     id=>"ls133"        path =>"./rubyLogParse.rb"  }
  #格式化日期  date{    match=>["requestTime","yyyy-MM-dd HH:mm:ss"]    target =>"@timestamp"    locale=>"en"    timezone =>"+00:00"  }  #删除多余字段,减少es的存储空间  mutate{      remove_field => ["@version","offset","message","beat","input","prospector"]  }}

uby中String和json转换关系如下:

String--->JSON JSON.parse String

JSON--->String JSON.to_json

定义方法如下:

 #解析json格式日志 def parseJsonLog(event,logdata)    #获取日志内容    loginfo=logdata[(logdata.index('] ')+2)...(logdata.rindex('}')+1)]    #转成json对象    logInfoJson=JSON.parse loginfo    #获取流水号    event.set('transactionID',logInfoJson['transactionID'])    #获取业务账号    event.set('businessNumber',logInfoJson['businessNumber'])    #获取省份编码    event.set('provinceCode',logInfoJson['provinceCode'])    #获取执行耗时    event.set('callTime',logInfoJson['callTime'])    #获取接口编码    event.set('serverCode',logInfoJson['serverCode'])    #获取请求ip    event.set('reqIp',logInfoJson['reqIp'])    #获取渠道    event.set('channle',logInfoJson['reqMessage']['channel'])    #获取返回报文    event.set('respMessage',logInfoJson['respMessage'].to_json)    #获取请求报文    event.set('reqMessage',logInfoJson['reqMessage'].to_json)    if(loginfo.include?'errorInfo')      event.set('errorInfo',logInfoJson['errorInfo'])    end    #设置标示位    event.set('tags','log_success')  end
  #解析xml格式日志  def parseXmlLog(event,logdata)    #截取流水号字段    if (logdata.include?'transactionID') && (logdata.include?'businessNumber')      transactionID=logdata[(logdata.index('transactionID')+16)...(logdata.index('businessNumber')-3)]      event.set('transactionID',transactionID)      #根据流水号判断渠道      if(logdata.include?'Channel')        channle=logdata[(logdata.index('<Channel>')+9)...(logdata.index('</Channel>'))]        event.set('channle',channle)      end      #截取返回报文      if (logdata.include?'respMessage')        respMessage=logdata[(logdata.index('respMessage')+14)...(logdata.index('reqMessage')-3)]        event.set('respMessage',respMessage)      end        省略部分代码   ................            #设置标示位      event.set('tags','log_success')    end  end
  def filter(event)      #取出日志信息      logdata=event.get('message')      #如果日志不是短信日志,缓存监控日志,缓存日志 Access nodechanged      if (logdata.include?'reqMessage' )        #解析日志级别和请求时间,并保存到event对象中        logleve=logdata[(logdata.index('[')+1)...logdata.index(']')]        event.set('logleve',logleve)        event.set('requestTime',logdata[0,logdata.index(' - [')])        #如果是rest接口json日志        if(logdata.include?'restJson')          parseJsonLog event,logdata        else          parseXmlLog event,logdata        end      end    return [event]  end