Logstash count events using ruby filter

gnam · March 22, 2022, 12:22am

I have following pipeline (example, not full pipeline) where I need count the number of events for EVERY file I am writing as output. For example, this pipleline creates new file every 30 seconds with the events processed in that 30 seconds. This code is sort of working except two issues:

Writing each trap_id as an entry in out put file (I just want ONE entry with final count) -- Though I am doing just event count, my end goal is, count different log_level events, for example, how many "Error", "Information" windows events etc.,
When I reset the trap_id counter, the first entry in output file starting with ZERO instead of ONE.

Can someone please advise how I can address these issues?

Posted same question on SO too.

input {
  beats {
    port => 5045
  }
}
filter
{
    ruby {
        init => '
                @trap_id = 0
                
                @lasttimestmp = 0
                
        '
        code => '
            
            evnttme = event.get("[@metadata][ms]")
            
            if @lasttimestmp == evnttme
                    @trap_id += 1
                    
                    event.set("lsttimestmp", @lasttimestmp)
                    event.set("trap_id", @trap_id)
                
                    
            else
                    @trap_id = 0
                                        
                    @lasttimestmp = evnttme
                    event.set("lsttimestmp", evnttme)
                    
                    event.set("[@metadata][ms]",  evnttme)
                    
                    event.set("trap_id", @trap_id)
                    
                
            end
    
        '
    }

}

output {

   file {
      path => "output.log"
   }    
   
   file {
            flush_interval => 30
            codec => line { format => "%{[@metadata][ts]}, %{[trap_id]}"}
            path => "C:/lgstshop/local/csv/output%{[@metadata][ms]}.csv"
        }
}

Badger · March 22, 2022, 1:14am

If I had to do this I would use a ruby script file. Keep track of the current 30 second window and count events of each type. If it is time to change files, then return an array of two events, one with the old file name, which includes the counts, and one with the new file name. Otherwise, just return an array of one event, being the current event.

gnam · March 22, 2022, 3:07am

Thank you! Is there any difference between with-in filter (or) separate file? I tried exactly what you suggested but with-in filter (like the example posted in question) by defining @trap_id instance variable to keep track of the count. The problem is, output file having entries like below (instead of one single entry with total 1647918042696, 3)

1647918042696, 0, 
1647918042696, 1, 
1647918042696, 2, 
1647918042696, 3,

I can't do event.cancel, because I need original event written to another output file at same time.

Here is my complete pipeline to give you an idea (I have zero experience with Logstash/Ruby, trying my best to put together based on inputs from you and other members, thank you very much for help!):

I need the data in "file" output one line with "counts" for different log_level, while "CSV" contains the real events for that duration.

# Sample Logstash configuration for creating a simple
# Beats -> Logstash -> Elasticsearch pipeline.


input {
  beats {
    port => 5045
  }
}
filter {
   ruby {
		
		init => "
            begin
				path = 'C:/lgstshop/local/'
				@ctime = Time.now.to_f
				id = (@ctime * 1000).to_i/30
				@evt_file = 'output'+id.to_s+'.evt'
                csv_headers = ['Timestamp','Windows Event Timestamp','Log Level','Event ID']
                if File.zero?(@evt_file) || !File.exist?(@evt_file)
                    CSV.open(path+@evt_file, 'w') do |csv|
                        csv << csv_headers
						csv << ['0','0','Error','0']
						csv << ['1','1','Error','1']
						csv << ['2','2','Information','2']
						csv << ['3','3','Error','3']
						csv << ['4','4','Error','4']
                    end
					
                end
				
				path2 = 'C:/lgstshop/local/csv/'
				@csv_file = 'output'+id.to_s+'.csv'
                csv2_headers = ['Timestamp','Info_Sum','Warn_Sum','Error_Sum']
                if File.zero?(@csv_file) || !File.exist?(@csv_file)
                    CSV.open(path2+@csv_file, 'w') do |csv2|
                        csv2 << csv2_headers
						csv2 << ['0','0','0','0']
						csv2 << ['1','1','1','1']
						csv2 << ['2','2','2','2']
						csv2 << ['3','3','3','3']
						csv2 << ['4','4','4','4']
                    end
					
                end
            end
        "
		
		code => '
				@ctime = Time.now.to_f
				event.set("[@metadata][ts]",  (@ctime * 1000).to_i)
				event.set("[@metadata][ms]",  (@ctime * 1000).to_i/30)
        '
    }
}

filter{
 ruby {
        code => '
	    	event.remove("tags")
			event.remove("@version")
			event.remove("@timestamp")
	
			event.get("event").each { |k, v|
                event.set(k,v)
            }
			
			
			event.get("log").each { |k, v|
                event.set(k,v)
            }
			event.remove("log")
         	event.remove("message")
	    '
    }
}

filter
{
	ruby {
		init => '
				@trap_id = 0
				
				@info_cnt = 0
				@warn_cnt=0
				@error_cnt = 0
				
				@lasttimestmp = 0
				
			
		'
		code => '
			
			evnttme = event.get("[@metadata][ms]")
			loglevel = event.get("level")
			
			
			if loglevel == "information"
				@info_cnt += 1
			end
			
			if loglevel == "warn"
				@warn_cnt += 1
			end
			
			if loglevel == "error"
				@error_cnt += 1
			end
			
			
			if @lasttimestmp == 0
				@lasttimestmp = evnttme
			end
			
			
			
			if @lasttimestmp == evnttme
					@trap_id += 1
					
					event.set("lsttimestmp", @lasttimestmp)
					
					event.set("trap_id", @trap_id)
					event.set("info_cnt", @info_cnt)
					event.set("warn_cnt", @warn_cnt)
					event.set("error_cnt", @error_cnt)
					
			else
					@trap_id = 0
					
					@info_cnt = 0
					@warn_cnt = 0
					@error_cnt = 0
					
					@lasttimestmp = evnttme
					
					event.set("lsttimestmp", evnttme)
					
					event.set("[@metadata][ms]",  evnttme)
					
					event.set("trap_id", @trap_id)
					event.set("info_cnt", @info_cnt)
					event.set("warn_cnt", @warn_cnt)
					event.set("error_cnt", @error_cnt)
				
			end
	
		'
	}

}



filter {
    mutate {
        add_field => { "mycustomts" => "%{[@metadata][ts]}" }
    }
}


output {

   file {
      path => "output.log"
   }	
   
   file {
			flush_interval => 30
			codec => line { format => "%{[@metadata][ts]}, %{[info_cnt]}, %{[warn_cnt]}, %{[error_cnt]}"}
		 	path => "C:/lgstshop/local/csv/output%{[@metadata][ms]}.csv"
		}
		
	csv {
       flush_interval => 30
       fields => [ "mycustomts", "created", "level", "code"]
       path => "C:/lgstshop/local/output%{[@metadata][ms]}.evt"
    }
   
}

Badger · March 22, 2022, 3:20am

A script file returns an array of events (1 or 2 in this case), so that is an easy way to create a second event.

You may be able to use a new_event_block as described in the code option of an aggregate filter but I have not tested it. I do not know if that is specific to aggregate or whether you can use it a ruby filter. @Jenni's post here suggests it can be used in a ruby filter.

gnam · March 22, 2022, 4:56am

Thank you @Badger. new_event_block seems interesting option, will try.

system · April 19, 2022, 4:57am

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

Topic		Replies	Views
Ruby Filter output to Field Logstash	3	1951	July 6, 2017
Logstash - clone event (ruby code) Logstash	2	1757	July 6, 2017
Unable to create new document/event in logstash Ruby filter using yield Logstash	3	1129	July 14, 2020
Output writing wrong event data Logstash	3	320	October 2, 2019
Logstash ruby length does not count correctly Logstash	3	391	October 18, 2021

Logstash count events using ruby filter

Related topics