Array datatype limit size?


(Roger) #1

Hi all,
I'm trying to collect information about login of accounts on a website. Currently I'm able to save, for instance, the last 10 login attempts using the array datatype.
To do that I use Filebeat that sends messages to Logstash to grok strings.
Is there a way to limit the array's size?


(Roger) #2

This is the logstash.conf used right now:

input { 
	beats {
		client_inactivity_timeout => 600
		port => 5044
	}
}

filter {
 grok {
        break_on_match => false
		patterns_dir => [".\patterns"]
        match => [
            "message", ".*\s%{TT:timestamp}.*login\=\"%{USERNAME:username}\""
        ]
    }
 date {
   match => ["timestamp", "yyyyMMdd HHmmss"]
 }

mutate {
    add_field => {
	 "logins" => ["%{@timestamp}"]
	}
	remove_field => ["timestamp"]
}
}

output {
  elasticsearch { 		
   hosts => ["localhost:9200"]
   index => "ftaudit"
   document_id => "%{[username]}_%{[beat][name]}"
  }
 }

unfortunately, when add_field is performed, it will overwrite the previous value. can anyone help me?
@magnusbaeck maybe you could know the answer?


(David Pilato) #3

Please don't ping people who are not yet involved in the thread.
I moved your question to #logstash.


(Roger) #4

sorry @dadoonet.
at the end I modified the output section of my logstash.conf in this way:

 output {
  elasticsearch { 		
   hosts => ["localhost:9200"]
   index => "ftaudit"
   document_id => "%{[username]}_%{[beat][name]}"
   doc_as_upsert => true
   action => "update"
   script => 'ctx._source.logins += ";""%{@timestamp}"'
  }
 }

but I'm facing this error:

[2018-11-30T19:37:46,744][WARN ][logstash.outputs.elasticsearch] Could not index event to Elasticsearch. {:status=>400, :action=>["update", {:_id=>"NoTmanaGe_CE", :_index=>"ftaudit", :_type=>"doc", :_routing=>nil, :_retry_on_conflict=>1}, #LogStash::Event:0x28acefed], :response=>{"update"=>{"_index"=>"ftaudit", "_type"=>"doc", "_id"=>"NoTmanaGe_CE_dxfbft03", "status"=>400, "error"=>{"type"=>"mapper_parsing_exception", "reason"=>"failed to parse field [logins] of type [date]", "caused_by"=>{"type"=>"illegal_argument_exception", "reason"=>"Invalid format: "2018-11-30T16:42:10.000Z2018-11-30T16:41:30.000Z" is malformed at "2018-11-30T16:41:30.000Z""}}}}}

so basically, in my index there is only the first document_id of each record. I'm not understanding how to manage my login field, composed by several date


(Roger) #5

just an update. I changed the output section of logstash.conf in this way:

output {
  elasticsearch { 		
   hosts => ["localhost:9200"]
   index => "ftaudit"
   document_id => "%{[username]}_%{[beat][name]}"
   doc_as_upsert => true
   action => "update"
   script => 'ctx._source.logins.add(%{@timestamp}); if(ctx._source.logins.length > 10) { ctx._source.logins.remove(0); }'
  }

it seems to be the correct way but I'm facing this error:

[2018-12-04T17:48:29,014][WARN ][logstash.outputs.elasticsearch] Could not index event to Elasticsearch. {:status=>400, :action=>["update", {:_id=>"NoTmanaGe_CE_dxfbft03", :_index=>"ftaudit", :_type=>"doc", :_routing=>nil, :_retry_on_conflict=>1}, #LogStash::Event:0x3ec54e10], :response=>{"update"=>{"_index"=>"ftaudit", "_type"=>"doc", "_id"=>"NoTmanaGe_CE_ftp", "status"=>400, "error"=>{"type"=>"illegal_argument_exception", "reason"=>"failed to execute script", "caused_by"=>{"type"=>"script_exception", "reason"=>"compile error", "script_stack"=>["... ource.list.add(2018-11-30T16:52:41.000Z)", " ^---- HERE"], "script"=>"ctx._source.list.add(2018-11-30T16:52:41.000Z)", "lang"=>"painless", "caused_by"=>{"type"=>"illegal_argument_exception", "reason"=>"invalid sequence of tokens near ['T16'].", "caused_by"=>{"type"=>"no_viable_alt_exception", "reason"=>nil}}}}}}}

here it is also the mapping of my index:

{
  "ftaudit" : {
    "mappings" : {
      "doc" : {
        "properties" : {
          "@timestamp" : {
            "type" : "date"
          },
          "@version" : {
            "type" : "text",
            "fields" : {
              "keyword" : {
                "type" : "keyword",
                "ignore_above" : 256
              }
            }
          },
          "software" : {
            "properties" : {
              "message" : {
                "type" : "text",
                "fields" : {
                  "keyword" : {
                    "type" : "keyword",
                    "ignore_above" : 256
                  }
                }
              }
            }
          },
          "beat" : {
            "properties" : {
              "hostname" : {
                "type" : "keyword"
              },
              "name" : {
                "type" : "text",
                "fields" : {
                  "keyword" : {
                    "type" : "keyword",
                    "ignore_above" : 256
                  }
                }
              },
              "version" : {
                "type" : "text",
                "fields" : {
                  "keyword" : {
                    "type" : "keyword",
                    "ignore_above" : 256
                  }
                }
              }
            }
          },
          "client" : {
            "properties" : {
              "ip" : {
                "type" : "keyword"
              },
              "port" : {
                "type" : "text",
                "fields" : {
                  "keyword" : {
                    "type" : "keyword",
                    "ignore_above" : 256
                  }
                }
              }
            }
          },
          "filebeat_source" : {
            "type" : "text",
            "fields" : {
              "keyword" : {
                "type" : "keyword",
                "ignore_above" : 256
              }
            }
          },
          "host" : {
            "properties" : {
              "name" : {
                "type" : "text",
                "fields" : {
                  "keyword" : {
                    "type" : "keyword",
                    "ignore_above" : 256
                  }
                }
              }
            }
          },
          "logins" : {
            "type" : "date"
          },
          "message" : {
            "type" : "text",
            "fields" : {
              "keyword" : {
                "type" : "keyword",
                "ignore_above" : 256
              }
            }
          },
          "name" : {
            "type" : "text",
            "fields" : {
              "keyword" : {
                "type" : "keyword",
                "ignore_above" : 256
              }
            }
          },
          "offset" : {
            "type" : "long"
          },
          "server" : {
            "properties" : {
              "ip" : {
                "type" : "keyword"
              },
              "port" : {
                "type" : "keyword"
              }
            }
          },
          "tags" : {
            "type" : "keyword"
          },
          "username" : {
            "type" : "keyword"
          }
        }
      }
    }
  }
}