logstash ruby filter elasticsearch bulk insert

 

input 부분

input {
    exec {
        command  => "/usr/bin/python3 /home/kimjh/Desktop/logstashQuery/01/jsonRead.py"
        interval => 120
    }
}

python 코드 부분

 

json 파일 

=> logstash 는 ndjson 형태의 파일을 read 한다. 하지만 다음은 일반적인 json 파일이다. 이를 읽기 위해 위와 같은 파이썬 코드를 짰고 그것을 logstash의 input에 넣었다.

$ cat test_sample.json
{
        "2010": {"name2": "kim1"},
        "2011": {"name2": "kim2"},
        "2012": {"name2": "kim3"},
        "2013": {"name2": "kim4"},
        "2014": {"name2": "kim5"}
}

filter 부분

filter {
    ruby {
        code => '
                require "elasticsearch"
                require "json"
                require "yaml"

                esClient = Elasticsearch::Client.new
                actionArray = Array.new

                jsonDatas = event.get("message")
                if ! jsonDatas.nil?
                    jsonArry  = jsonDatas.split("\n")
                    jsonArry.each_with_index do |n, c|
                        #puts n.class
                        strHash = YAML.load(n)
                        hashData = {
                                    "index" => {
                                        "_index"  => "sleep4",
                                        "_id"     => c,
                                        "data" => {}
                                        }
                                    }

                        hashData["index"]["data"].merge!strHash
                        actionArray.push(hashData)
                    end
                end
                if ! actionArray.empty?
                    actionArray.each do |n|
                        p n
                    end
                    esClient.bulk body: actionArray
                end
                '
    }

    mutate {
        remove_field => ["@timestamp", "host", "command", "@version"]
    }
}

 

1) split : 문자열 분리 

=> 특정 char를 기준으로 문자열을 자르게 된다.

 

2) each_with_index : python 과 비교하자면 enumerate 와 동일하게 작동

 

3) empty?

=> 데이터가 비어있는지 확인 비어있으면 true 반환 아니면 false 반환

 

4) push

=> array에 데이터 삽입