logstash占用cpu过高

发布于 2022-09-03 08:55:40 字数 6640 浏览 22 评论 0

之前用logstash读取文件然后入库,占用cpu平均在20%左右,最近加了一个监控目录,cpu平均就飙到80%多了

下面是logstash的配置文件,之前file都是监控单目录,加了一个目录后CPU就飙高了,麻烦各位看下什么问题?

input { 
    file {
                path => ["/data/rsync_files/assistants/web_100/*/users.log","/data/rsync_files/assistants/web_101/*/users.log"]
                type => "users"
                start_position => "beginning"
        codec => "json"
        }  
    file {
                path => ["/data/rsync_files/assistants/web_100/*/user_states.log","/data/rsync_files/assistants/web_101/*/user_states.log"]
                type => "user_states"
                start_position => "beginning"
        codec => "json"
        }
    file {
                path => ["/data/rsync_files/assistants/web_100/*/user_events.log","/data/rsync_files/assistants/web_101/*/user_events.log"]
                type => "user_events"
                start_position => "beginning"
        codec => "json"
        } 

    file {
                path => ["/data/rsync_files/assistants/web_100/*/user_locations.log","/data/rsync_files/assistants/web_101/*/user_locations.log"]
                type => "user_locations"
                start_position => "beginning"
        codec => "json"
        } 

    file {
                path => ["/data/rsync_files/assistants/web_100/*/user_snapshots.log","/data/rsync_files/assistants/web_101/*/user_snapshots.log"]
                type => "user_snapshots"
                start_position => "beginning"
        codec => "json"
        } 
}

filter {
    if [type] == "users" {
        if ![@fields][ctxt_type] {
        mutate {
            add_field => ["[@fields][ctxt_type]", "0"]
        }
        }
        mutate {
        convert => ["[@fields][ctxt_id]", "string"]
        convert => ["[@fields][ctxt_type]", "string"]
        }
    } else if [type] == "user_states" {
        mutate {
        convert => ["[@fields][ctxt_user_id]", "string"]
        }
    } else if [type] == "user_events"{
        mutate {
        convert => ["[@fields][ctxt_id]", "string"]
        convert => ["[@fields][ctxt_user_id]", "string"]
        convert => ["[@fields][ctxt_time]", "string"]
        }
    } else if [type] == "user_locations"{
        mutate {
        convert => ["[@fields][ctxt_user_id]", "string"]
        }
    } else if [type] == "user_snapshots"{
        mutate {
        convert => ["[@fields][ctxt_user_id]", "string"]
        }
    }
}

output {
    if [type] == "users" {
        if [@fields][ctxt_type] and [@fields][ctxt_imei] != '' {
            jdbc {
                driver_jar_path => "/usr/local/jar/mysql-connector-java-5.1.36-bin.jar"
                driver_class => "com.mysql.jdbc.Driver"
                connection_string => "jdbc:mysql://127.0.0.1/analysis?user=root&password=wgc@db123"
                statement => ["insert into user_lists(user_id,device,imei,v,lang,brand,ov,ctime,type) values(?,?,?,?,?,?,?,UNIX_TIMESTAMP(?),?) ON DUPLICATE KEY UPDATE device=VALUES(`device`),v=VALUES(`v`),lang=VALUES(`lang`),brand=VALUES(`brand`),ov=VALUES(`ov`),ctime=VALUES(`ctime`),user_id=VALUES(`user_id`),type=VALUES(type),imei=VALUES(`imei`)", "%{[@fields][ctxt_id]}","%{[@fields][ctxt_device]}","%{[@fields][ctxt_imei]}","%{[@fields][ctxt_v]}","%{[@fields][ctxt_lang]}","%{[@fields][ctxt_brand]}","%{[@fields][ctxt_ov]}","%{[@fields][ctxt_created_at]}","%{[@fields][ctxt_type]}"]
            }
        }
    } else if [type] == "user_states" {
        if [@fields][ctxt_user_id] != "0" {
            jdbc {
                driver_jar_path => "/usr/local/jar/mysql-connector-java-5.1.36-bin.jar"
                driver_class => "com.mysql.jdbc.Driver"
                connection_string => "jdbc:mysql://127.0.0.1/analysis?user=root&password=wgc@db123"
                statement => ["insert into user_lists(user_id,mtime) values(?, UNIX_TIMESTAMP(?) ) ON DUPLICATE KEY UPDATE mtime=VALUES(`mtime`)", "%{[@fields][ctxt_user_id]}", "%{[@fields][ctxt_updated_at]}"]
            }
        }
    } else if [type] == "user_events" {
        jdbc {
            driver_jar_path => "/usr/local/jar/mysql-connector-java-5.1.36-bin.jar"
            driver_class => "com.mysql.jdbc.Driver"
            connection_string => "jdbc:mysql://127.0.0.1/analysis?user=root&password=wgc@db123"        
            statement => [ "INSERT INTO user_events(id,user_id,event,time,name,package,version,created_at,updated_at) values(?,?,?,?,?,?,?,?,?)","%{[@fields][ctxt_id]}","%{[@fields][ctxt_user_id]}","%{[@fields][ctxt_event]}","%{[@fields][ctxt_time]}","%{[@fields][ctxt_name]}","%{[@fields][ctxt_package]}","%{[@fields][ctxt_version]}","%{[@fields][ctxt_created_at]}","%{[@fields][ctxt_updated_at]}" ]
        }
    } else if [type] == "user_locations" {
        if [@fields][ctxt_user_id] != "0" {
            jdbc {
                driver_jar_path => "/usr/local/jar/mysql-connector-java-5.1.36-bin.jar"
                driver_class => "com.mysql.jdbc.Driver"
                connection_string => "jdbc:mysql://127.0.0.1/analysis?user=root&password=wgc@db123"
                statement => [ "insert into user_lists(country_id,country,area_id,area,region_id,region,city_id,city,isp_id,isp,user_id) VALUES(?,?,?,?,?,?,?,?,?,?,?) ON DUPLICATE KEY UPDATE country_id=VALUES(`country_id`), country=VALUES(`country`), area_id=VALUES(`area_id`), area=VALUES(`area`), region_id=VALUES(`region_id`), region=VALUES(`region`), city_id=VALUES(`city_id`), city=VALUES(`city`), isp_id=VALUES(`isp_id`),isp=VALUES(`isp`)","%{[@fields][ctxt_country_id]}","%{[@fields][ctxt_country]}","%{[@fields][ctxt_area_id]}","%{[@fields][ctxt_area]}","%{[@fields][ctxt_region_id]}","%{[@fields][ctxt_region]}","%{[@fields][ctxt_city_id]}","%{[@fields][ctxt_city]}","%{[@fields][ctxt_isp_id]}","%{[@fields][ctxt_isp]}","%{[@fields][ctxt_user_id]}" ]
            }
        }
    } else if [type] == "user_snapshots" {
        if [@fields][ctxt_user_id] != "0" {
            jdbc {
                driver_jar_path => "/usr/local/jar/mysql-connector-java-5.1.36-bin.jar"
                driver_class => "com.mysql.jdbc.Driver"
                connection_string => "jdbc:mysql://127.0.0.1/analysis?user=root&password=wgc@db123"
                statement => [ "INSERT INTO apps_user_lists(`user_id`,`packages`,`decode`) VALUES(?,?,?) ON DUPLICATE KEY UPDATE packages=VALUES(`packages`),decode=0","%{[@fields][ctxt_user_id]}","%{[@fields][ctxt_snapshot]}","%[@fields][ctxt_snapshot]" ]
            }
        }
    }
    
}

如果你对这篇内容有疑问,欢迎到本站社区发帖提问 参与讨论,获取更多帮助,或者扫码二维码加入 Web 技术交流群。

扫码二维码加入Web技术交流群

发布评论

需要 登录 才能够评论, 你可以免费 注册 一个本站的账号。
列表为空,暂无数据
我们使用 Cookies 和其他技术来定制您的体验包括您的登录状态等。通过阅读我们的 隐私政策 了解更多相关信息。 单击 接受 或继续使用网站,即表示您同意使用 Cookies 和您的相关数据。
原文