/usr/share/logstash/bin/logstash-plugin install logstash-output-exec

vim /etc/logstash/conf.d/daemonset-filebeat.conf
input {
  kafka {
    bootstrap_servers => "10.0.7.53:9092,10.0.7.54:9092,10.0.7.55:9092"
    topics => ["daemonset-pod-console-log"]
  }
}

filter {
 if [fields][log_topic] == "daemonset-pod-console-log"  {
    mutate {
     remove_field => ["@version","agent","ecs"]
    }
  }
}

output {
  if [fields][log_topic] == "daemonset-pod-console-log"  {
    elasticsearch {
      hosts => ["10.0.7.46:9200","10.0.7.47:9200","10.0.7.48:9200"]
      index => "daemonset-pod-console-log-%{+YYYY.MM.dd}"
    }
  }
  if 'error' in [message] or 'Error' in [message] or 'ERROR' in [message] or 'failed' in [message] or 'fail' in [message] or 'Fail' in [message] or 'Failed' in [message] or '异常' in [message] or 'exception' in [message] or 'Undefined index' in [message] or 'Invalid argument' in [message] or 'not found' in [message] {
    # stdout { codec => rubydebug }
    exec {
        command => "sh /root/b.sh \"%{message}\""
    }
  }
}

调试:
/usr/share/logstash/bin/logstash -f daemonset-filebeat.conf  

重启:
systemctl restart logstash

input {
  kafka {
    bootstrap_servers => "172.18.10.78:9092"
    topics => ["test-lava-mall-client"] 
    codec => json {
        charset => "UTF-8"
    }
  }
}


filter {
 if [fields][log_topic] == "test-lava-mall-client"  {
    mutate {
     remove_field => ["ecs", "tags", "agent"]
    }
  }
}


output {

  if [fields][log_topic] == "test-lava-mall-client" {
    #stdout { codec => rubydebug }

    elasticsearch {
      hosts => ["172.18.10.96"]
      index => "test-lava-mall-client-%{+YYYY.MM.dd}"
    }

    if 'error' in [message] or 'Error' in [message] or 'ERROR' in [message] or 'failed' in [message] or 'fail' in [message] or 'Fail' in [message] or 'Failed' in [message] or '异常' in [message] or 'exception' in [message] or 'Undefined index' in [message] or 'Invalid argument' in [message] or 'not found' in [message] {
      exec {
        command => "sh /data/alert/error_alert.sh '%{message} %{log}'"
      }
    }


  }

}

vim /root/b.sh
#!/bin/bash
webhook="你的企业wx的webhook地址"
message=`echo $*`
error_message=`echo "${message}" |awk -F'message:' '{print $2}' | awk -F'"' '{print $1}'`
namespace=`echo "${message}" | awk -F'pods/' '{print $2}' |awk -F'_' '{print $1}'`
pod_name=`echo "${message}" |awk -F'pods/' '{print $2}' |awk -F'_' '{print $2}'`
curl ${webhook} -H "Content-Type: application/json" -d '{
          "msgtype": "markdown",
           "markdown": {
             "content": "<font color=\"info\">应用启动故障报警</font> 
              >pod_name: <font color=\"comment\">'${pod_name}'</font>
              >namespace: <font color=\"comment\">'${namespace}'</font>
              >报错日志: <font color=\"comment\">'"${error_message}"'</font>"
            }
        }'

[root@k8s-node1 filebeat-kafka]# pwd
/var/log/pods/kube-system_filebeat-kafka-kqnxn_86f08f9d-c56d-4b5b-800d-78e7574f4fcd/filebeat-kafka
[root@k8s-node1 filebeat-kafka]# vim a.log 
Undefined index afj jlj wqerlj ljaflafkj alfj afl af
Invalid argument lwqkjreljwr sja falkfj alfjka f
not found; jwle;qjekrlqwjr lqwjr qr
exception: 自定义异常
Fatl ERROR: Uncaugh Exxor: AppProcessAchievementAchievementProcess class not esist. in /var/www/EasySwooleEvent.php:248

vim /etc/logstash/conf.d/daemonset-filebeat-2.conf
input {
  beats {
    port => 5044
  }
}

output {
  if 'error' in [message] or 'Error' in [message] or 'ERROR' in [message] or 'failed' in [message] or 'fail' in [message] or 'Fail' in [message] or 'Failed' in [message] or '异常' in [message] or 'exception' in [message] or 'Undefined index' in [message] or 'Invalid argument' in [message] or 'not found' in [message] {
    stdout { codec => rubydebug }
    exec {
        command => "sh /data/alert/error_alert.sh '%{message} %{log}'"
    }
  }
}

vim /data/alert/error_alert.sh 
#!/bin/bash
webhook="https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=4b232609-ed90-4aee-b83c-05f0e5cf2734"
message=`echo $*`
error_message=`echo "${message}" |awk -F' {"offset' '{print $1}'`
namespace=`echo "${message}" | awk -F'pods/' '{print $2}' |awk -F'_' '{print $1}'`
pod_name=`echo "${message}" |awk -F'pods/' '{print $2}' |awk -F'_' '{print $2}'`
curl ${webhook} -H "Content-Type: application/json" -d '{
          "msgtype": "markdown",
           "markdown": {
             "content": "日志报警
              >pod_name: <font color=\"comment\">'${pod_name}'</font>
              >namespace: <font color=\"comment\">'${namespace}'</font>
              >报错日志: <font color=\"comment\">'"${error_message//\\/\\\\\\\\\\\\}"'</font>"
            }
        }'