logstash 的基本配置与使用

安装kibana

• kibana不同版本下载地址https://www.elastic.co/cn/downloads/past-releases#kibana

• 解压下载的压缩文件

• 修改es服务器配置

  • vim config/kibana.yml
    • 修改server.host(kibana的服务器配置)
    • 修改elasticsearch.url(es服务器地址)

安装logstash

• logstash不同版本下载地址https://www.elastic.co/cn/downloads/past-releases#logstash

生成logstash模版

• 在kibana执行以下脚本，生成logstash模版
  或者
• 连接elasticsearch执行curl，创建logstash模版

PUT _template/logstash_template
{
    "template": "logstash-*",
    "settings": {
        "number_of_replicas": 1,
        "number_of_shards": 3
    },
    "mappings": {
        "logstash": {
            "properties": {
                "module": {
                    "type": "keyword"
                },
                "appid": {
                    "type": "keyword"
                },
                "table_name": {
                    "type": "keyword"
                },
                "cmd": {
                    "type": "keyword"
                },
                "action_type": {
                    "type": "keyword"
                },
                "timestamp": {
                    "type": "long"
                },
                "userid": {
                    "type": "keyword"
                },
                "cid": {
                    "type": "keyword"
                },
                "usercode": {
                    "type": "keyword"
                },
                "depart": {
                    "type": "keyword"
                },
                "orgcode": {
                    "type": "keyword"
                },
                "jobtitle": {
                    "type": "text",
                    "analyzer": "standard",
                    "fields": {
                        "keyword": {
                            "type": "keyword"
                        }
                    }
                },
                "joblevel": {
                    "type": "keyword"
                },
                "query": {
                    "type": "text",
                    "analyzer": "standard",
                    "fields": {
                        "keyword": {
                            "type": "keyword"
                        }
                    }
                },
                "from": {
                    "type": "long"
                },
                "size": {
                    "type": "long"
                },
                "result_count": {
                    "type": "long"
                },
                "resp_time": {
                    "type": "long"
                },
                "sort": {
                    "type": "keyword"
                },
                "itemid": {
                    "type": "keyword"
                },
                "other_info": {
                    "type": "nested"
                }
            }
        }
        "logs": {
            "properties": {
                "module": {
                    "type": "keyword"
                },
                "appid": {
                    "type": "keyword"
                },
                "table_name": {
                    "type": "keyword"
                },
                "cmd": {
                    "type": "keyword"
                },
                "action_type": {
                    "type": "keyword"
                },
                "timestamp": {
                    "type": "long"
                },
                "userid": {
                    "type": "keyword"
                },
                "cid": {
                    "type": "keyword"
                },
                "usercode": {
                    "type": "keyword"
                },
                "depart": {
                    "type": "keyword"
                },
                "orgcode": {
                    "type": "keyword"
                },
                "jobtitle": {
                    "type": "text",
                    "analyzer": "standard",
                    "fields": {
                        "keyword": {
                            "type": "keyword"
                        }
                    }
                },
                "joblevel": {
                    "type": "keyword"
                },
                "query": {
                    "type": "text",
                    "analyzer": "standard",
                    "fields": {
                        "keyword": {
                            "type": "keyword"
                        }
                    }
                },
                "from": {
                    "type": "long"
                },
                "size": {
                    "type": "long"
                },
                "result_count": {
                    "type": "long"
                },
                "resp_time": {
                    "type": "long"
                },
                "sort": {
                    "type": "keyword"
                },
                "itemid": {
                    "type": "keyword"
                },
                "other_info": {
                    "type": "nested"
                }
            }
        }  
    }
}

设置logstash pipeline

• 配置模版(kafka_logstash_pipeline.conf)

input{
    kafka{
                bootstrap_servers=>"kafka_host1:port,kafka_host2:port"
                topics=>["kafka_logstash_pipeline"]
                group_id=>"kafka_logstash"
                codec=>"json"
        }
}
filter{
    mutate{
        add_field=>{ "@fields"=> "%{fields}" }
    }
    json{
        source=>"@fields"
        remove_field=>["fields", "@fields"]
    }
    date{
        match=>["timestamp", "UNIX"]
        remove_field=>"timestamp"
    }
    ruby {
            code => "event.set('timestamp', event.get('@timestamp').time.localtime + 8*60*60)"
    }
    ruby {
            code => "event.set('@timestamp',event.get('timestamp'))"
            remove_field=>"timestamp"
    }
}
output{
        elasticsearch{
                hosts=>["elasticsearch_host1:port", "elasticsearch_host2:port"]
                ilm_pattern=>"{now/d}"
                template_name=>"logstash_template"
                index=>"logstash-%{+YYYY.MM.dd}"
        }
}

启动logstash

• 解压：tar -xzvf logstash-5.3.0.tar.gz
• cd logstash-5.3.0/pipeline
• 修改logstash 数据采集配置：将上述kafka_logstash_pipeline.conf放置在pipeline下
• 打开kafka_logstash_pipeline.conf，修改kafka和es集群连接配置
• 启动logstash

# check the validity of configuration file
./bin/logstash -f pipeline/kafka_logstash_pipeline.conf --config.test_and_exit
# run the log scribe process
nohup ./bin/logstash -f pipeline/kafka_logstash_pipeline.conf --config.reload.automatic &