# Logstash count events using ruby filter

**URL:** <https://discuss.elastic.co/t/logstash-count-events-using-ruby-filter/300256>\
**Category:** Logstash\
**Created:** [March 22, 2022, 12:22am UTC](https://discuss.elastic.co/t/logstash-count-events-using-ruby-filter/300256 "2022-03-22T00:22:24Z")\
**Posts on this page:** 6\
**Page:** 1

<div class="post-metadata">

**Author:** ![gnam](https://avatars.discourse-cdn.com/v4/letter/g/b9e5f3/32.png) [@gnam](https://discuss.elastic.co/u/gnam)\
**Post date:** [March 22, 2022, 12:22am UTC](https://discuss.elastic.co/t/logstash-count-events-using-ruby-filter/300256/1 "2022-03-22T00:22:24Z")

</div>

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:

1. 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.,
2. 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.](https://stackoverflow.com/questions/71565591/logstash-count-events-using-ruby-filter)

```auto
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"
        }
}

```

---

<div class="post-metadata">

**Author:** ![Badger](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/badger/32/25190_2.png) [@Badger](https://discuss.elastic.co/u/Badger)\
**Post date:** [March 22, 2022, 1:14am UTC](https://discuss.elastic.co/t/logstash-count-events-using-ruby-filter/300256/2 "2022-03-22T01:14:16Z")

</div>

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.

---

<div class="post-metadata">

**Author:** ![gnam](https://avatars.discourse-cdn.com/v4/letter/g/b9e5f3/32.png) [@gnam](https://discuss.elastic.co/u/gnam)\
**Post date:** [March 22, 2022, 3:07am UTC](https://discuss.elastic.co/t/logstash-count-events-using-ruby-filter/300256/3 "2022-03-22T03:07:34Z")

</div>

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)**

```auto
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.

```auto
# 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"
    }
   
}

```

---

<div class="post-metadata">

**Author:** ![Badger](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/badger/32/25190_2.png) [@Badger](https://discuss.elastic.co/u/Badger)\
**Post date:** [March 22, 2022, 3:20am UTC](https://discuss.elastic.co/t/logstash-count-events-using-ruby-filter/300256/4 "2022-03-22T03:20:35Z")

</div>

> [@gnam](#):
>
> Is there any difference between with-in filter (or) separate file?

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](https://www.elastic.co/guide/en/logstash/current/plugins-filters-aggregate.html#plugins-filters-aggregate-code) 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](https://discuss.elastic.co/t/unable-to-create-new-document-event-in-logstash-ruby-filter-using-yield/236995/2) suggests it _can_ be used in a ruby filter.

---

<div class="post-metadata">

**Author:** ![gnam](https://avatars.discourse-cdn.com/v4/letter/g/b9e5f3/32.png) [@gnam](https://discuss.elastic.co/u/gnam)\
**Post date:** [March 22, 2022, 4:56am UTC](https://discuss.elastic.co/t/logstash-count-events-using-ruby-filter/300256/5 "2022-03-22T04:56:58Z")

</div>

Thank you @Badger. new\_event\_block seems interesting option, will try.

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\
**Post date:** [April 19, 2022, 4:57am UTC](https://discuss.elastic.co/t/logstash-count-events-using-ruby-filter/300256/6 "2022-04-19T04:57:08Z")

</div>

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