ELFK结合Kafka报错：InstanceAlreadyExistsException

向右看齐 2021-12-23 23:19 597阅读 0赞

### InstanceAlreadyExistsException错误 ###

在ELFK集群结合Kafka集群的配置过程中，通过filebet收集日志传输到kafka，然后Kafka传输日志到logstash，最后到elasticsearch，并结合zabbix告警。

但在kafka传输日志到logstash的配置完成后，启动logstash时报错：

[WARN ][org.apache.kafka.common.utils.AppInfoParser] Error registering AppInfo mbean
    javax.management.InstanceAlreadyExistsException

kafka通过logstash输入到elasticsearch的配置文件：

input { 
        kafka { 
            bootstrap_servers => "192.168.1.253:9092"
            group_id => "webservice"
    	    topics => "253-webservice"
            auto_offset_reset => "latest"
            consumer_threads => 5
            decorate_events => true 
            type => "253_webservice"
        }
    }
    
    filter { 
        if [type] == "253_ciphermachine" { 
            ruby { 
                code => "event.set('log_time',event.get('@timestamp').time.localtime + 8*60*60)"
            }
            #grok{ 
            # match => [ "log_time","(?<thisdate>^\d{4}-\d{1,2}-\d{1,2})" ]
            #}
    
            grok { 
                match => [ "message", "%{TIME:thistime} %{NOTSPACE:level}" ]
            }
    
            mutate { 
                add_field => [ "[zabbix_key]", "ciphermachine_error" ]
                add_field => [ "[zabbix_host]", "192.168.1.253" ]
    
                remove_field => "@version"
                remove_field => "host"
                remove_field => "path"
                remove_field => "_type"
                remove_field => "_score"
                remove_field => "_id"
                remove_field => "thread-id"
                remove_field => "log_time"
                remove_field => "thisdate"
                remove_field => "thistime"
                remove_field => "score"
                remove_field => "id"
                remove_field => "name"
                remove_field => "beat.*"
                remove_field => "fields.*"
                remove_field => "host.*"
                remove_field => "input.type"
                remove_field => "log.file.path"
                remove_field => "offset"
                remove_field => "prospector.type"
                remove_field => "source"
            }
    
            #ruby { 
            # code => "event.set('logtime',event.get('thisdate') + ' ' + event.get('thistime') )"
            #}
    
            date { 
                match => [ "logtime","yyyy-MM-dd HH:mm:ss,SSS",'ISO8601' ]
                target => "@timestamp"
            }
    
            ruby { 
                code => "event.set('logtime',event.get('@timestamp').time.localtime + 8*60*60)"
            }
        }
    }
    
    output { 
        if [type] == "253_ciphermachine" { 
            elasticsearch { 
                hosts => ["192.168.1.253:9200"]
                user => "elastic"
                password => "changeme"
                index => "webservice.log-%{+YYYY.MM.dd}"
            }
    
            if [level]  =~ /(ERR|error|ERROR|Failed)/ { 
                zabbix { 
                    zabbix_host => "[zabbix_host]"
                    zabbix_key => "[zabbix_key]"
                    zabbix_server_host => "192.168.1.252"
                    zabbix_server_port => "10051"
                    zabbix_value => "message"
                }
            }
        }
    }

网上搜了一下，提示是kafka并发问题。于是重新修改了配置文件：

input { 
        kafka { 
            bootstrap_servers => "192.168.1.253:9092"
            group_id => "ciphermachine"
            client_id => "ciphermachine-1"                #需要指定client_id，否则logstash启动报错
            topics => "253-ciphermachine"
            auto_offset_reset => "latest"
            codec => "json"
            consumer_threads => 5
            decorate_events => false
            type => "253_ciphermachine"
        }
    }
    
    filter { 
        if [type] == "253_ciphermachine" { 
            ruby { 
                code => "event.set('log_time',event.get('@timestamp').time.localtime + 8*60*60)"
            }
            #grok{ 
            # match => [ "log_time","(?<thisdate>^\d{4}-\d{1,2}-\d{1,2})" ]
            #}
    
            grok { 
                match => [ "message", "%{TIME:thistime} %{NOTSPACE:level}" ]
            }
    
            mutate { 
                add_field => [ "[zabbix_key]", "ciphermachine_error" ]
                add_field => [ "[zabbix_host]", "192.168.1.253" ]
    
                remove_field => "@version"
                remove_field => "host"
                remove_field => "path"
                remove_field => "_type"
                remove_field => "_score"
                remove_field => "_id"
                remove_field => "thread-id"
                remove_field => "log_time"
                remove_field => "thisdate"
                remove_field => "thistime"
                remove_field => "score"
                remove_field => "id"
                remove_field => "name"
                remove_field => "beat.*"
                remove_field => "fields.*"
                remove_field => "host.*"
                remove_field => "input.type"
                remove_field => "log.file.path"
                remove_field => "offset"
                remove_field => "prospector.type"
                remove_field => "source"
            }
    
            #ruby { 
            # code => "event.set('logtime',event.get('thisdate') + ' ' + event.get('thistime') )"
            #}
    
            date { 
                match => [ "logtime","yyyy-MM-dd HH:mm:ss,SSS",'ISO8601' ]
                target => "@timestamp"
            }
    
            ruby { 
                code => "event.set('logtime',event.get('@timestamp').time.localtime + 8*60*60)"
            }
        }
    }
    
    output { 
        if [type] == "253_ciphermachine" { 
            elasticsearch { 
                hosts => ["192.168.1.253:9200"]
                user => "elastic"
                password => "changeme"
                index => "webservice.log-%{+YYYY.MM.dd}"
            }
    
            if [level]  =~ /(ERR|error|ERROR|Failed)/ { 
                zabbix { 
                    zabbix_host => "[zabbix_host]"
                    zabbix_key => "[zabbix_key]"
                    zabbix_server_host => "192.168.1.252"
                    zabbix_server_port => "10051"
                    zabbix_value => "message"
                }
            }
        }
    }

反复测试，发现是input部分没有指定`client_id`，任意指定`client_id`即可，重启logstash后不再报错。

--------------------

### 删除topic错误 ###

在部署过程中，原先直接使用logstash收集日志传输到kafka存在问题，kafka无法接收到logstash传入的数据，而手动去kafka的topic中生产数据，又会导致kafka重复消费的问题，但是我其实配置了`auto_offset_reset => "latest"`，可能是我的配置文件存在问题。于是最终放弃了直接用logstash收集日志到kafka，改由filebeat收集日志到kafka。

但是在此之前，我需要删除之前的topic。但是在删除topic时，一直提示zookeeper marked，实际topic没有删除，只是被zookeeper标记，可以通过下面解决。

*  修改kafka配置：

# vim /usr/local/kafka/config/server.properties
    
    delete.topic.enable=true            #允许删除topic

*  删除topic：

# /usr/local/kafka/bin/kafka-topics.sh --delete --zookeeper 192.168.1.253:2181 --topic 253-webservice

如果这样无法删除，进入zookeeper命令行删除：

# /usr/local/kafka/bin/zookeeper-shell.sh 192.168.1.253:2181
    
    ls /brokers/topics
    
    rmr /brokers/topics/253-ciphermachine
    
    rmr /brokers/topics/253-webservice

*  列出当前topic：

# /usr/local/kafka/bin/kafka-topics.sh --list --zookeeper 192.168.1.253:2181

--------------------