Logstash datetime filter if else

Elastic Stack Logstash
1

Hi Team,

This is regarding the Logstash Input filter.

Sometimes the application from which I receive the data(Via logstash http) sends the value in one dateformat(ISO) and at times in another(UNIX).

Is there anyway I can check the format and if it's UNIX, i need to match it and convert it else - I can leave it like that itself. Please refer the attached snapshot to know more about the events I receive. Any suggestions would of great help.

image
image1547×424 45.5 KB

Regards
TJ

2

You can use a conditional in the filter section to see whether you need to pass the value through a date filter. Something like

if [repository][pushed_at] =~ /^date_time/ {
    mutate { gsub => [ "[repository][pushed_at]", "^date_time", "" ] }
    date { match => [ "[repository][pushed_at]", "UNIX" ] }
}
3

Actually from the source, the data is in Epoch format.

    "created_at": 1596548445,
    "pushed_at": 1604562857,

After it passes through logstash, it adds the date_time as prefix before the value. The thing is for other type of events, the [repository] created_at and [repository] pushed_at values is ISO but only for specific category, the value differs(Epoch). Is there anyway i can handle this situation?

Also I am trying to convert the epoch value of 1604595899233 to something like Nov 5, 2020 @ 22:34:22.659 but adding date filter (mentioned below) - converts it to something like 2020-11-05T16:21:17.094Z

date { match => [ "[entryTime]", "UNIX_MS" ] target => "[entryTime]"  }
4

A date filter converts the field to a LogStash::Timestamp, and what you have there is the string representation of such an object. If your fields are type date in elasticsearch then they are stored as milliseconds since the UNIX epoch in UTC.

If you want the field stored as a string in a particular format then use strftime.

5

Thank you. Currently logstash is receiving all the type of x_github_event in it's json file and I would like to apply the filter only when the input contains a key value pair. i.e., below : It has to perform a conversion. Else the conversion is not required.

if the value of the [headers][x_github_event]  is "push"

I need to set the value of [repository][created_at] from epoch to strftime. Can you please validate the below query

  if "status" in  [headers][x_github_event] {
        ruby {
        code => '
            t = Time.at(event.get("@timestamp").to_f)
            event.set("[repository][created_at]", t.strftime("%Y-%m-%d %H:%M:%s"))
        '
		}
	}

Kindly confirm if the above filter will do the trick.

Regards
TJ

6

That looks reasonable.

7

It doesn't works - The expected output remains same and logstash failed to index . I guess I am missing a small portion of something. This is the entire config of my logstash.

input {
 http {
        port => "8080"
        type => "json"
  }
}

filter {
    ruby {
            path => "/usr/share/logstash/filter/removeKeys.rb"
            script_params => { keys => [ "archive_url", "assignees_url", "avatar_url", "blobs_url", "branches_url", "clone_url", "collaborators_url", "comments_url", "commits_url", "compare_url", "contents_url", "contributors_url", "deployments_url", "diff_url", "downloads_url", "events_url", "followers_url", "following_url", "forks_url", "gists_url", "git_commits_url", "git_refs_url", "git_tags_url", "git_url", "hooks_url", "html_url", "issue_comment_url", "issue_events_url", "issues_url", "issue_url", "keys_url", "labels_url", "languages_url", "members_url", "merges_url", "milestones_url", "mirror_url", "notifications_url", "organizations_url", "patch_url", "public_members_url", "pull_request_url", "pulls_url", "received_events_url", "releases_url", "repos_url", "review_comments_url", "review_comment_url", "ssh_url", "stargazers_url", "starred_url", "statuses_url", "subscribers_url", "subscriptions_url", "subscription_url", "svn_url", "tags_url", "teams_url", "trees_url", "website_url", "ldap_dn", "gravatar_id", "node_id", "spdx_id", "stargazers_count", "has_wiki", "license", "stargazers", "mirror_url", "x_github_enterprise_host", "http_version", "x_github_enterprise_version", "http_host", "request_path", "http_accept", "request_method", "content_type", "organization.description", "repository.license", "commits" ] }
    }
        date { match => [ "[comment][created_at]", "ISO8601" ] target => "[comment][created_at]" }
        date { match => [ "[comment][updated_at]", "ISO8601" ] target => "[comment][updated_at]" }
        date { match => [ "[enterprise][created_at]", "ISO8601" ] target => "[enterprise][created_at]" }
        date { match => [ "[issue][closed_at]", "ISO8601" ] target => "[issue][closed_at]" }
        date { match => [ "[issue][created_at]", "ISO8601" ] target => "[issue][created_at]" }
        date { match => [ "[issue][updated_at]", "ISO8601" ] target => "[issue][updated_at]" }
        date { match => [ "[pull_request][base][repo][created_at]", "ISO8601" ] target => "[pull_request][base][repo][created_at]" }
        date { match => [ "[pull_request][base][repo][pushed_at]", "ISO8601" ] target => "[pull_request][base][repo][pushed_at]" }
        date { match => [ "[pull_request][base][repo][updated_at]", "ISO8601" ] target => "[pull_request][base][repo][updated_at]" }
        date { match => [ "[pull_request][closed_at]", "ISO8601" ] target => "[pull_request][closed_at]" }
        date { match => [ "[pull_request][created_at]", "ISO8601" ] target => "[pull_request][created_at]" }
        date { match => [ "[pull_request][updated_at]", "ISO8601" ] target => "[pull_request][updated_at]" }
        date { match => [ "[pull_request][head][repo][created_at]", "ISO8601" ] target => "[pull_request][head][repo][created_at]" }
        date { match => [ "[pull_request][head][repo][pushed_at]", "ISO8601" ] target => "[pull_request][head][repo][pushed_at]" }
        date { match => [ "[pull_request][head][repo][updated_at]", "ISO8601" ] target => "[pull_request][head][repo][updated_at]" }
        date { match => [ "[pull_request][merged_at]", "ISO8601" ] target => "[pull_request][merged_at]" }
        date { match => [ "[pull_request][merged_at]", "ISO8601" ] target => "[pull_request][merged_at]" }

        mutate {
                        lowercase => [ "[repository][name]" ]
        }
   if "push" in  [headers][x_github_event] {
        ruby {
        code => '
            t = Time.at(event.get("@timestamp").to_f)
            event.set("[repository][created_at]", t.strftime("%Y-%m-%d %H:%M:%s"))
        '
        }
   }
}

output {
  stdout { codec => rubydebug { metadata => true } }
  kafka {
        codec => json
        topic_id => "githubevent"
        bootstrap_servers => "host:9092"
        max_request_size => 104857600
  }
}

The error which I recieved from logstash is below.

[2020-11-23T07:24:12,198][WARN ][logstash.outputs.elasticsearch][github-kafka][githudata_elastic] Could not index event to Elasticsearch. {:status=>400, :action=>["index", {:_id=>nil, :_index=>"for-testing_project_pipeline-push-2020.11.23", :routing=>nil, :_type=>"_doc"}, #<LogStash::Event:0x12cc0b06>], :response=>{"index"=>{"_index"=>"for-testing_project_pipeline-push-2020.11.23", "_type"=>"_doc", "_id"=>"QEL883UBZ9G1Tbi08uzD", "status"=>400, "error"=>{"type"=>"mapper_parsing_exception", "reason"=>"failed to parse field [repository.created_at] of type [long] in document with id 'QEL883UBZ9G1Tbi08uzD'. Preview of field's value: '2020-11-23 07:24:1606116249'", "caused_by"=>{"type"=>"illegal_argument_exception", "reason"=>"For input string: \"2020-11-23 07:24:1606116249\""}}}}}

Requesting you to have a look at it.

Regards
TJ

8

Firstly %s is the number of seconds since the epoch, for the second of the minute use %S.

Even if you fix that you will get the same error, since elasticsearch expects the [repository][created_at] field to be type long, not a date. If you roll over to a new index that may fix itself. Otherwise you will need an index template (and still need to roll over to a new index to change the type).

9

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.