前提条件

操作步骤

1. 进入 阿里云Elasticsearch控制台的Logstash页面 。
2. 进入目标实例。
   1. 在顶部菜单栏处，选择地域。
   2. 在 Logstash实例 中单击目标实例ID。
3. 在左侧导航栏，单击 管道管理 。
4. 单击 创建管道 。
5. 在 创建管道任务 页面，输入管道ID并配置管道。
   本文使用的管道配置如下。

   input {
       beats {
           port => 8001
   filter {
       mutate {
           gsub => ["message","\|","| "]
           split => ["message","|"]
           add_field => {
               "mobile" => "%{[message][1]}"
               "appName" => "%{[message][2]}"
               "type" => "%{[message][3]}"
               "timestamp" => "%{[message][4]}"
               "status" => "%{[message][5]}"
               "code" => "%{[message][6]}"
               "component" => "%{[message][7]}"
               "cid" => "%{[message][8]}"
               "serviceId" => "%{[message][9]}"
               "serviceName" => "%{[message][10]}"
               "serviceType" => "%{[message][11]}"
               "param" => "%{[message][12]}"
       mutate {
           strip => ["mobile","appName","type","timestamp","status","code","component","cid","serviceId","serviceName","serviceType","param"]
   output {
       elasticsearch {
           index => "<yourIndexName>"
           hosts => ["es-cn-7mz2mu1zp0006****.elasticsearch.aliyuncs.com:9200"]
           user => "elastic"
           password => "<yourPassword>"
   }                

   重要

   • input.beat.port 为Beats采集日志输入到当前Logstash管道的端口，需要使用8000~9000范围内的端口。
   • 以上管道配置中的 gsub => ["message","\|","| "] ，第二个 | 后有一个空格。
   • 以上管道配置中的 index 、 hosts 和 password 参数值需要替换为您实际业务的索引名称、Elasticsearch集群的访问地址和集群的elastic账号对应的密码。

   以上管道配置的原理说明如下：

   1. 通过Logstash的 filter.mutate.gsub 参数，使用正则表达式 \| 去匹配 LogMessage 中的 | ，将 | 替换为 | （即 | +空格）。替换后的效果如下。

      LogMessage: | 1390000****| jop| byORP| 2022-04-18T14:18:16.633| /log/cms/send| 200| pluginNums=0,pluginStatus=0| | | | | |

   2. 通过 filter.mutate.split 参数将 LogMessage 按照 | 进行切分。
   3. 通过 filter.mutate.add_field 参数添加字段，即将切分后的 LogMessage 一一添加到对应的字段中。添加后的效果如下。

      "mobile":" 1390000****",
      "appName":" jop",
      "type":" byORP",
      "timestamp":" 2022-04-18T14:18:16.633",
      "status":" /log/cms/sen",
      "code":" 200",
      "component":" pluginNums=0,pluginStatus=0",
      "cid":" ",
      "serviceId":" ",
      "serviceName":" ",
      "serviceType":" ",
      "param":" "

   4. 通过 filter.mutate.strip 参数去除字段空格。由于添加后的每个字段前面都有一个空格，因此需要去除这些空格。

   更多管道配置说明，请参见 通过配置文件管理管道 和 Logstash配置文件说明 。

   警告 配置完成后，需要保存并部署才能生效。保存并部署操作会触发实例重启，请在不影响业务的前提下，继续执行以下步骤。
6. 单击 保存 或者 保存并部署 。
   • 保存 ：将管道信息保存在Logstash里并触发实例变更，配置不会生效。保存后，系统会返回 管道管理 页面。可在 管道列表 区域，单击 操作 列下的 立即部署 ，触发实例重启，使配置生效。
   • 保存并部署 ：保存并且部署后，会触发实例重启，使配置生效。

验证结果

1. 登录目标阿里云Elasticsearch实例的Kibana控制台，根据页面提示进入Kibana主页。
   登录Kibana控制台的具体操作，请参见 登录Kibana控制台 。

   说明 本文以阿里云Elasticsearch 7.10.0版本为例，其他版本操作可能略有差别，请以实际界面为准。
2. 单击右上角的 Dev tools 。
3. 在 Console 中，执行以下脚本，查询目标索引中的信息。

   GET <yourIndexName>/_search
     "query": {
       "match_all": {}
   }

   说明 <yourIndexName> 需要与管道配置中的 index 参数值保持一致。

   预期结果如下。

   {
     "took" : 1,
     "timed_out" : false,
     "_shards" : {
       "total" : 1,
       "successful" : 1,
       "skipped" : 0,
       "failed" : 0
     "hits" : {
       "total" : {
         "value" : 1,
         "relation" : "eq"
       "max_score" : 1.0,
       "hits" : [
           "_index" : "<yourIndexName>",
           "_type" : "_doc",
           "_id" : "Lb1UWoAB-6Zo6en4luDi",
           "_score" : 1.0,
           "_source" : {
             "mobile" : "1390000****",
             "appName" : "jop",
             "type" : "byORP",
             "timestamp" : "2022-04-18T14:18:16.633",
             "status" : "/log/cms/sen",
             "code" : "200",
             "component" : "pluginNums=0,pluginStatus=0",
             "cid" : "",
             "serviceId" : "",
             "serviceName" : "",
             "serviceType" : "",
             "param" : ""