Convert timestamp and add it to email body

mancharagopan · October 12, 2018, 6:39am

My logstash debug message is,

Oct 12 11:10:16 localhost.localdomain logstash[17086]: [2018-10-12T11:10:16,683][DEBUG][org.logstash.beats.BeatsHandler] [local: 172.20.11.19:5044, remote: 172.20.11.16:51070] Received a new payload
Oct 12 11:10:16 localhost.localdomain logstash[17086]: [2018-10-12T11:10:16,683][DEBUG][org.logstash.beats.BeatsHandler] [local: 172.20.11.19:5044, remote: 172.20.11.16:51070] Sending a new message for the listener, sequence: 1
Oct 12 11:10:16 localhost.localdomain logstash[17086]: [2018-10-12T11:10:16,684][DEBUG][org.logstash.beats.BeatsHandler] 3a0abf13: batches pending: false
Oct 12 11:10:16 localhost.localdomain logstash[17086]: [2018-10-12T11:10:16,788][DEBUG][logstash.pipeline        ] filter received {"event"=>{"record_number"=>"11403578", "beat"=>{"hostname"=>"COMPANY-HO-DC1", "name"=>"COMPANY-HO-DC1", "version"=>"6.4.2"}, "message"=>"An account was logged off.\n\nSubject:\n\tSecurity ID:\t\tS-1-5-18\n\tAccount Name:\t\tCOMPANY-HO-DC1$\n\tAccount Domain:\t\tCOMPANY\n\tLogon ID:\t\t0x23277f0\n\nLogon Type:\t\t\t3\n\nThis event is generated when a logon session is destroyed. It may be positively correlated with a logon event using the Logon ID value. Logon IDs are only unique between reboots on the same computer.", "type"=>"wineventlog", "log_name"=>"Security", "host"=>{"name"=>"COMPANY-HO-DC1"}, "computer_name"=>"COMPANY-HO-DC1.COMPANY.com", "process_id"=>464, "keywords"=>["Audit Success"], "@version"=>"1", "@timestamp"=>2018-10-12T05:35:23.425Z, "level"=>"Information", "thread_id"=>2764, "event_data"=>{"TargetUserName"=>"COMPANY-HO-DC1$", "LogonType"=>"3", "TargetDomainName"=>"COMPANY", "TargetLogonId"=>"0x23277f0", "TargetUserSid"=>"S-1-5-18"}, "provider_guid"=>"{54849625-5478-4994-A5BA-3E3B0328C30D}", "opcode"=>"Info", "task"=>"Logoff", "tags"=>["beats_input_codec_plain_applied"], "source_name"=>"Microsoft-Windows-Security-Auditing", "event_id"=>4634}}
Oct 12 11:10:16 localhost.localdomain logstash[17086]: [2018-10-12T11:10:16,791][DEBUG][logstash.filters.grok    ] Running grok filter {:event=>#<LogStash::Event:0x6ebdf3cd>}
Oct 12 11:10:16 localhost.localdomain logstash[17086]: [2018-10-12T11:10:16,791][DEBUG][logstash.filters.grok    ] Event now:  {:event=>#<LogStash::Event:0x6ebdf3cd>}
Oct 12 11:10:16 localhost.localdomain logstash[17086]: [2018-10-12T11:10:16,791][DEBUG][logstash.pipeline        ] output received {"event"=>{"record_number"=>"11403578", "beat"=>{"hostname"=>"Company-HO-DC1", "name"=>"COMPANY-HO-DC1", "version"=>"6.4.2"}, "message"=>"An account was logged off.\n\nSubject:\n\tSecurity ID:\t\tS-1-5-18\n\tAccount Name:\t\tCOMPANY-HO-DC1$\n\tAccount Domain:\t\tCOMPANY\n\tLogon ID:\t\t0x23277f0\n\nLogon Type:\t\t\t3\n\nThis event is generated when a logon session is destroyed. It may be positively correlated with a logon event using the Logon ID value. Logon IDs are only unique between reboots on the same computer.", "type"=>"wineventlog", "log_name"=>"Security", "host"=>{"name"=>"Company-HO-DC1"}, "computer_name"=>"Company-HO-DC1.Company.com", "process_id"=>464, "keywords"=>["Audit Success"], "@version"=>"1", "@timestamp"=>2018-10-12T05:35:23.425Z, "level"=>"Information", "thread_id"=>2764, "event_data"=>{"TargetUserName"=>"COMPANY-HO-DC1$", "LogonType"=>"3", "TargetDomainName"=>"COMPANY", "TargetLogonId"=>"0x23277f0", "TargetUserSid"=>"S-1-5-18"}, "provider_guid"=>"{54849625-5478-4994-A5BA-3E3B0328C30D}", "opcode"=>"Info", "task"=>"Logoff", "tags"=>["beats_input_codec_plain_applied", "_grokparsefailure"], "source_name"=>"Microsoft-Windows-Security-Auditing", "event_id"=>4634}}

I am receiving logs from windows server.

My logstash conf file is,

input {
  beats {
    port => 5044
  }
}

filter {
  grok {
      match => { "event_id" => "1149" }
      add_tag => ["warning"]
  }
}

output {
  if "warning" in [tags] {
      email {
         to => "its@company.com"
         address => "smtp.gmail.com"
         port => "587"
         via => "smtp"
         use_tls => "true"
         username => "it.support@company.com"
         password => ""
         from => "it.support@company.com"
         subject => "critical event spotted by ELK from : %{host}, %{TIMESTAMP_ISO8601:timestamp}, EventID = %{event_id}"
         body => "%{message}"
      }
  }
  elasticsearch {
    hosts => ["172.30.11.19:9200"]
    sniffing => true
    manage_template => true
    index => "%{[@metadata][beat]}-%{+YYYY.MM.dd}"
    document_type => "%{[@metadata][type]}"
  }
}

I am filtering event id and sending email alerts when certain event id matched. I want to add the timestamp to the email. How do i convert "@timestamp"=>2018-10-12T05:35:23.425Z,
to normal date and time?

guyboertje · October 22, 2018, 1:19pm

Use a ruby filter to set a date_time field to a string representation of the timestamp.

github.com

elastic/logstash/blob/master/logstash-core/src/main/java/org/logstash/ext/JrubyEventExtLibrary.java#L260


}


@JRubyMethod(name = "tag", required = 1)
public IRubyObject ruby_tag(ThreadContext context, RubyString value)
{
    //TODO(guy) should these tags be BiValues?
    this.event.tag(value.asJavaString());
    return context.nil;
}


@JRubyMethod(name = "timestamp")
public IRubyObject ruby_timestamp(ThreadContext context) {
    // We can just cast to IRubyObject here, because we know that Event stores a
    // RubyTimestamp internally.
    return (IRubyObject) event.getUnconvertedField(FieldReference.TIMESTAMP_REFERENCE);
}


@JRubyMethod(name = "timestamp=", required = 1)
public IRubyObject ruby_set_timestamp(ThreadContext context, IRubyObject value)
{
    if (!(value instanceof JrubyTimestampExtLibrary.RubyTimestamp)) {

github.com

elastic/logstash/blob/master/logstash-core/src/main/java/org/logstash/ext/JrubyTimestampExtLibrary.java#L79


                true
            );


        }
    } else {
        throw context.runtime.newTypeError("wrong argument type " + time.getMetaClass() + " (expected Time)");
    }
    return this;
}


@JRubyMethod(name = "time")
public RubyTime ruby_time(ThreadContext context)
{
    return RubyTime.newTime(context.runtime, this.timestamp.getTime());
}


@JRubyMethod(name = "to_i")
public IRubyObject ruby_to_i(ThreadContext context)
{
    return RubyFixnum.newFixnum(context.runtime, this.timestamp.getTime().getMillis() / 1000);
}

https://ruby-doc.org/core-2.3.0/Time.html#method-i-strftime

The ruby code will look something like this:
event.set("datetime_in_email", event.timestamp.strftime("%B, %d %Y at %H:%M:%S")

You probably want a conditional branch to ensure that only events meant for the email get the extra field.

mancharagopan · November 2, 2018, 3:47am

I am new to ELK Stack and i don't have a lot of programming knowledge. Is there any simple method for this? like adding the timezone with the timestamp inside the email plugin or inside any filter plugins.

guyboertje · November 2, 2018, 2:57pm

At this time, there is no other mechanism. sorry.

Here is a test evaluation config.

input {
  generator {
    message => '{"msg":"Julius, stay away from the Forum today", "tags": ["warning"]}'
    count => 1
  }
}

filter {
  json {
    source => "message"
  }
  if "warning" in [tags]  {
    ruby {
      code => 'event.set("datetime_in_email", event.timestamp.time.strftime("%B, %d %Y at %H:%M:%S"))'
      remove_field => ["message"]
    }
  }
}

output {
  stdout {
    codec => rubydebug
  }
}

Result:

{
           "@timestamp" => 2018-11-02T14:55:21.194Z,
             "sequence" => 0,
                  "msg" => "Julius, stay away from the Forum today",
                 "tags" => [
        [0] "warning"
    ],
                 "host" => "Elastics-MacBook-Pro.local",
    "datetime_in_email" => "November, 02 2018 at 14:55:21",
             "@version" => "1"
}

system · November 30, 2018, 2:57pm

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