ELK实现接口调用统计

具体思路

1. 设置打印日志格式
   日志格式：当前系统名称-调用类-调用方法
   各个字段间用空格分割，便于es分词切分
2. 使用logstash过滤器grok对固定格式的日期进行分词存储点击获取Grok 正则表达式相关语法
   使用正则表达式对上述固定格式日志进行切分
   过滤正则表达式为：
   %{DATA:current_system} %{DATA:current_class} %{DATA:current_method}
3. 使用es查询过滤出需要的数据
   可以使用系统名称、方法名称在es存储的数据中使用count操作,即可获取相关接口调用次数
4. 实现以上功能，需要配置logstash-sample.conf,样例如下：

 input {
  tcp {
    mode => "server"
    host => "localhost"
    port => 4560
    type => "tcp"
    codec => json_lines
  }
}
input {
  kafka {
    bootstrap_servers=> "localhost:9092"
    # group_id =>"es"
    topics =>"applog"
    consumer_threads =>1
    decorate_events =>true
    type => "kafka"
  }
}
filter{
  grok {
    match => {"message" => "%{DATA:current_system} %{DATA:current_class} %{DATA:current_method} "}
    overwrite => ["message"]
  }
  if[type] == "tcp" {
    mutate {
      add_tag => ["tcp1"]
    }
  }  
  if [type] == "kafka" {
     mutate {
       add_tag => ["kafka1"]
     }
  }
}
output {
  if "tcp1" in [tags]{
    elasticsearch {
      hosts => "localhost:9200"
      #index => "data_sharing_indicators008"
      index => "%{[appname]}"
      # template => "D:\bigdata\logstash-7.3.2\template"
      # template_overwrite => "true"
    }
  }
}
output {
  if "kafka1" in [tags]{
    elasticsearch {
      hosts => "localhost:9200"
      index => "data_sharing_indicators007"
      # index => "kafka_%{[appname]}"
      # template => "D:\bigdata\logstash-7.3.2\template"
      # template_overwrite => "true"
    }
  }
}

5.注意点
自定义的日志输出字段，不能与已有的系统日志输出字段重名，类似关键字的用法

"%{DATA:current_system} %{DATA:current_class} %{DATA:current_method} "