creating data stream through logstash

Question

I have installed elasticsearch cluster v 7.14.

I have created ILM policy and Index template. However data stream parameters mentioned under logstash pipeline file are giving error.

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 pipeline config file -

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
  }
}

After all deployment done, can only see system indices but data stream is not created.

[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}

error is saying parameters like template"=>"/tmp/pingaccess_template", "ilm_pattern"=>"000001", "template_name"=>"pingaccess_template", "ilm_rollover_alias"=>"pingaccess", "ilm_policy"=>"testpolicy" are not valid but in below link they are mentioned.

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

Answer 1

The solution is to use logstash without be "aware" of data_stream.

1. FIRST of all (before running logstash) create your ILM and index_template BUT adding the "index.lifecycle.name" in the settings. That way, you are linking the template and ILM. Also, don't forget the data_stream in the index template.

{
  "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. Keep Logstash output like if data_stream doesn't exist but add action => create. This is because you can't use "index" API with data streams. Need the _create API call.

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

That way, logstash will output to ES, but the index template will be applied automatically (because of pattern match) and also the ILM and data_stream will be applied.

Important: To make it work, you need to start from scratch. If the index "test-data-stream" already exists in ES (as a traditional index), then data_stream will NOT be created. Make the test with another index name to make sure it works.

Answer 2

The documentation is unclear, but the plugin does not support those options when datastream output is enabled. The plugin is logging the options returned by the invalid_data_stream_params function, which allows action, routing, data_stream, anything else that starts with data_stream_, the shared options defined by the mixin, and the common options defined by the output plugin base.