通过logstash创建数据流

Question

我已经安装了 elasticsearch cluster v 7.14。

我创建了 ILM 策略和索引模板。但是logstash管道文件下提到的数据流参数会报错。

ILM policy-

{
  "testpolicy" : {
    "version" : 1,
    "modified_date" : "2021-08-28T02:58:25.942Z",
    "policy" : {
      "phases" : {
        "hot" : {
          "min_age" : "0ms",
          "actions" : {
            "rollover" : {
              "max_primary_shard_size" : "900mb",
              "max_age" : "2d"
            },
            "set_priority" : {
              "priority" : 100
            }
          }
        },
        "delete" : {
          "min_age" : "2d",
          "actions" : {
            "delete" : {
              "delete_searchable_snapshot" : true
            }
          }
        }
      }
    },
    "in_use_by" : {
      "indices" : [ ],
      "data_streams" : [ ],
      "composable_templates" : [ ]
    }
  }
}

Index temaplate-

{
  "index_templates" : [
    {
      "name" : "access_template",
      "index_template" : {
        "index_patterns" : [
          "test-data-stream*"
        ],
        "template" : {
          "settings" : {
            "index" : {
              "number_of_shards" : "1",
              "number_of_replicas" : "0"
            }
          },
          "mappings" : {
            "_routing" : {
              "required" : false
            },
            "dynamic_date_formats" : [
              "strict_date_optional_time",
              "yyyy/MM/dd HH:mm:ss Z||yyyy/MM/dd Z"
            ],
            "numeric_detection" : true,
            "_source" : {
              "excludes" : [ ],
              "includes" : [ ],
              "enabled" : true
            },
            "dynamic" : true,
            "dynamic_templates" : [ ],
            "date_detection" : true
          }
        },
        "composed_of" : [ ],
        "priority" : 500,
        "version" : 1,
        "data_stream" : {
          "hidden" : false
        }
      }
    }
  ]
}

logstash 管道配置文件 -

input {
  beats {
    port => 5044
  }
}

filter {

 if [log_type] == "access_server" and [app_id] == "pa"
  {
     grok {
    match => {
    "message" => "%{YEAR}-%{MONTHNUM}-%{MONTHDAY}[T ]%{HOUR}:%{MINUTE}(?::?%{SECOND})\| %{USERNAME:exchangeId}\| %{DATA:trackingId}\| %{NUMBER:RoundTrip:int}%{SPACE}ms\| %{NUMBER:ProxyRoundTrip:int}%{SPACE}ms\| %{NUMBER:UserInfoRoundTrip:int}%{SPACE}ms\| %{DATA:Resource}\| %{DATA:subject}\| %{DATA:authmech}\| %{DATA:scopes}\| %{IPV4:Client}\| %{WORD:method}\| %{DATA:Request_URI}\| %{INT:response_code}\| %{DATA:failedRuleType}\| %{DATA:failedRuleName}\| %{DATA:APP_Name}\| %{DATA:Resource_Name}\| %{DATA:Path_Prefix}"    
    }
    }
    mutate {
             replace => {
               "[type]" => "access_server"
             }
           }
  }
}

output {
   if [log_type] == "access_server" {
  elasticsearch {
    hosts => ['http://10.10.10.76:9200']
        user => elastic
    password => xxx
     data_stream => "true"
     data_stream_type => "logs"
     data_stream_dataset => "access"
     data_stream_namespace => "default"
     ilm_rollover_alias => "access"
     ilm_pattern => "000001"
     ilm_policy => "testpolicy"
     template => "/tmp/access_template"
     template_name => "access_template"
      }
 }
   elasticsearch {
    hosts => ['http://10.10.10.76:9200']
    index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"
    user => elastic
    password => xxx
  }
}

所有部署完成后，只能看到系统索引，但没有创建数据流。

[2021-08-28T12:42:50,103][ERROR][logstash.outputs.elasticsearch][main] Invalid data stream configuration, following parameters are not supported: {"template"=>"/tmp/pingaccess_template", "ilm_pattern"=>"000001", "template_name"=>"pingaccess_template", "ilm_rollover_alias"=>"pingaccess", "ilm_policy"=>"testpolicy"}

[2021-08-28T12:42:50,547][ERROR][logstash.javapipeline    ][main] Pipeline error {:pipeline_id=>"main", :exception=>#<LogStash::ConfigurationError: Invalid data stream configuration: ["template", "ilm_pattern", "template_name", "ilm_rollover_alias", "ilm_policy"]>, :backtrace=>["/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-output-elasticsearch-11.0.2-java/lib/logstash/outputs/elasticsearch/data_stream_support.rb:57:in `check_data_stream_config!'"

[2021-08-28T12:42:50,702][ERROR][logstash.agent           ] Failed to execute action {:id=>:main, :action_type=>LogStash::ConvergeResult::FailedAction, :message=>"Could not execute action: PipelineAction::Create<main>, action_result: false", :backtrace=>nil}

错误是说像template"=>"/tmp/pingaccess_template", "ilm_pattern"=>"000001", "template_name"=>"pingaccess_template", "ilm_rollover_alias"=>"pingaccess", "ilm_policy"=>"testpolicy" 这样的参数无效，但在下面的链接中提到了它们。

https://www.elastic.co/guide/en/logstash/current/plugins-outputs-elasticsearch.html#plugins-outputs-elasticsearch-data-streams

Answer 1

文档不清楚，但启用数据流输出时插件不支持这些选项。该插件正在记录invalid_data_stream_params 函数返回的选项，该函数允许操作、路由、data_stream、以data_stream_ 开头的任何其他内容、mixin 定义的shared options 以及output plugin base 定义的常用选项。

Answer 2

解决方案是使用logstash而不“知道”data_stream。

1- 首先（在运行 logstash 之前）创建您的 ILM 和 index_template，但在设置中添加“index.lifecycle.name”。这样，您将链接模板和 ILM。另外，不要忘记索引模板中的 data_stream。

{
  "index_templates" : [
    {
      "name" : "access_template",
      "index_template" : {
        "index_patterns" : [
          "test-data-stream*"
        ],
        "template" : {
          "settings" : {
            "index" : {
              "number_of_shards" : "1",
              "number_of_replicas" : "0",
              "index.lifecycle.name": "testpolicy"
            }
          },
          "mappings" : {
          ...
          }
        },
        "composed_of" : [ ],
        "priority" : 500,
        "version" : 1,
        "data_stream" : {
          "hidden" : false
        }
      }
    }
  ]
}

2- 保持 Logstash 输出如果 data_stream 不存在，但添加操作 => 创建。这是因为您不能对数据流使用“索引”API。需要 _create API 调用。

    output { elasticsearch {
      hosts => ['http://10.10.10.76:9200']
      index => "test-data-stream"
      user => elastic
      password => xxx
      action => "create"
      }

这样，logstash 会输出到 ES，但是会自动应用索引模板（因为模式匹配），并且还会应用 ILM 和 data_stream。

重要提示：要使其发挥作用，您需要从头开始。如果 ES 中已经存在索引“test-data-stream”（作为传统索引），则不会创建 data_stream。使用另一个索引名称进行测试以确保它有效。