# Logstash Aggregate Problem tag in subdirectories Please Help
**URL:** https://discuss.elastic.co/t/logstash-aggregate-problem-tag-in-subdirectories-please-help/295116
**Category:** Logstash
**Created:** [January 22, 2022, 12:32pm UTC](https://discuss.elastic.co/t/logstash-aggregate-problem-tag-in-subdirectories-please-help/295116 "2022-01-22T12:32:29Z")
**Posts on this page:** 9
**Page:** 1
### Author:  [@melorium](https://discuss.elastic.co/u/melorium)
#### Post date: [January 22, 2022, 12:32pm UTC](https://discuss.elastic.co/t/logstash-aggregate-problem-tag-in-subdirectories-please-help/295116/1 "2022-01-22T12:32:29Z")
Hi i trying to use logstash aggregate to tag all log events i main and sub folders.
I want all logs in APP1 folders and subfolders to have tag "APP1" and all logs in app2 folders to have APP2 etc.
My friend did some kind of loop down here.
But i dont understand what he did.
The events in each app folder has a stop and end event.
This below here is not for tagging every event but i want to have the tagging to work.
Please help me.
BR Dennis
Main
├───APP1-LOG-MAIN-DIR
│ ├───SUB-DIR
| │ ├───APP1-LOG-SUB-DIR
| | | ├───file1
| | | ├───file2
| | | ├───file3
| | | ├───file4
├───APP2-LOG-MAIN-DIR
│ ├───SUB-DIR
| │ ├───APP2-LOG-SUB-DIR
| | | ├───file1
| | | ├───file2
| | | ├───file3
| | | ├───file4
filter {
grok {
match =\> {"message" =\> "%{TIMESTAMP\_ISO8601:nuix\_timestamp} %{NUMBER:nuix\_timezone} [%{DATA:nuix\_source}] %{NUMBER:nuix\_thread} %{LOGLEVEL:nuix\_loglevel}%{SPACE}%{DATA:nuix\_class} - %{GREEDYDATA:automation\_status}"}
}
```
if "NUIX-RUNNER-START" in [tags] {
ruby {
code => '
event.set("[nuix_log]", {
"runner" => event.get("nuix_class").delete_prefix("SCRIPT.").delete_suffix(".gen"),
"path" => File.dirname(event.get("[log][file][path]")),
"host" => event.get("[agent][hostname]"),
})
'
}
aggregate {
task_id => "%{[nuix_log][host]}"
code => "map[event.get('[nuix_log][path]')] = event.get('[nuix_log]')"
map_action => "create"
}
} else if "NUIX-RUNNER-END" in [tags] {
ruby {
code => 'event.set("[nuix_log][host]", event.get("[agent][hostname]"))'
}
aggregate {
task_id => "%{[nuix_log][host]}"
code => '
log_dir = File.dirname(event.get("[log][file][path]"))
(0..6).each do |i|
if map.key?(log_dir)
event.set("[nuix_log]", map[log_dir])
break
end
log_dir = File.dirname(log_dir)
end
'
map_action => "update"
end_of_task => true
timeout => 120
}
} else {
ruby {
code => 'event.set("[nuix_log][host]", event.get("[agent][hostname]"))'
}
aggregate {
task_id => "%{[nuix_log][host]}"
code => '
log_dir = File.dirname(event.get("[log][file][path]"))
(0..6).each do |i|
if map.key?(log_dir)
event.set("[nuix_log]", map[log_dir])
break
end
log_dir = File.dirname(log_dir)
end
'
map_action => "update"
}
}
```
}
---
### Author:  [@Badger](https://discuss.elastic.co/u/Badger)
#### Post date: [January 22, 2022, 5:28pm UTC](https://discuss.elastic.co/t/logstash-aggregate-problem-tag-in-subdirectories-please-help/295116/2 "2022-01-22T17:28:02Z")
Those aggregates will add
```
"nuix_log" => {
"path" => "/Main/APP1-LOG-MAIN-DIR/SUB-DIR/APP1-LOG-SUB-DIR",
"host" => "foo",
"runner" => "blah"
},
```
to each event where [log][file][path] is "/Main/APP1-LOG-MAIN-DIR/SUB-DIR/APP1-LOG-SUB-DIR". If you literally want to add a tag with the app name I would suggest adding
```
grok { match => { "[nuix_log][path]" => "/\w+/%{WORD:[@metadata][appname]}-" } add_tag => ["%{[@metadata][appname]}" ] }
```
after the aggregates.
---
### Author:  [@melorium](https://discuss.elastic.co/u/melorium)
#### Post date: [January 23, 2022, 11:49am UTC](https://discuss.elastic.co/t/logstash-aggregate-problem-tag-in-subdirectories-please-help/295116/3 "2022-01-23T11:49:40Z")
Thanks a lot Badger.
Your help is invaluable for me.
I will explain a little more what I want to do.
Nuix is a program for forensic examination's and it create logs everytime you start a new ingestion.
Let say case1 starting and ingesting it creates a main folder for main log.
that long is called Nuix.log and it for the main log.
2 steps below it creates something called worker logs these also belong to case1
All logs in 20211230131416 folder belongs to same case.
as you see here in the list all cases starts with a main folder with a muix.log and 2 folder below its creates something called worker logs filename is also nuix.log.
What I'm trying to do here is to add the case name in all logs events below to the case.
first I use filebeat to tag the files and read them in
- type: log
paths:
- type: log
paths:
* * *
1. 0-9
2. 0-9
---
### Author:  [@melorium](https://discuss.elastic.co/u/melorium)
#### Post date: [January 23, 2022, 12:03pm UTC](https://discuss.elastic.co/t/logstash-aggregate-problem-tag-in-subdirectories-please-help/295116/4 "2022-01-23T12:03:02Z")
After that sens it to Logstash.
The case name can be found in Nuix.log from the main log.
Here is a snippet from the log
2021-12-30 13:15:08.614 +0100 [AWT-EventQueue-0] 51818 INFO com.nuix.data.keystore.b - Read 0 passwords from C:\CASE\Test Log 1-1\Stores\DecryptionKeys\pgpkeylist.dat
so the case name is "Test Log 1-1"
So in logstach a can get it by using
else if "NUIX-RUNNER-START-WS" in [tags] and "NUIX-LOG" in [tags] {
grok {
match =\> {"message" =\> "%{TIMESTAMP\_ISO8601:nuix\_timestamp} %{NUMBER:nuix\_timezone} [%{DATA:nuix\_source}] %{NUMBER:nuix\_thread} %{LOGLEVEL:nuix\_loglevel}%{SPACE}%{DATA:nuix\_class} - (?\.+?(?=\Stores\DecryptionKeys\pgpkeylist.dat))"}
}
mutate {
gsub =\> [
# replace backslashes and forward slashes with a hyphen
"case\_name", "[\/]", "€ "
]
}
grok {
match =\> {"case\_name" =\> " (?\([^€]+$))"}
So what I'm doing here is within the aggregate loop I sent here yestoday I want to tag every event with case name or add a filed with the case name to every event.
I post my whole lostasch conf here so maybe you can se what I'm doing wrong .
The Logstash do alto more but my main problem is to tag it with case name.
input {
beats {
port =\> 5044
}
}
filter {
if "AUTOMATION" in [tags] {
grok {
match =\> { "ts" =\> "(?^.{0,23})" }
}
date {
match =\> ["ts1", "ISO8601", "YYYY-MM-dd HH:mm:ss", "YYYY-MM-dd HH:mm:ss.ZZZ", "UNIX"]
target =\> "@timestamp"
locale =\> "en"
}
ruby {
code =\> '
t = event.get("@timestamp")
event.set("automation\_timestamp\_fixed", Time.at(t.to\_f).strftime("%Y-%m-%d %H:%M:%S"))
'
}
}
}
filter {
if "MULTILINE" in [tags] and "NUIX-LOG" in [tags] {
grok {
match =\> {"message" =\> "%{TIMESTAMP\_ISO8601:nuix\_timestamp} %{NUMBER:nuix\_timezone} [%{DATA:nuix\_source}] %{NUMBER:nuix\_thread} %{LOGLEVEL:nuix\_loglevel}%{SPACE}%{DATA:nuix\_class} - (?\[\s\S]{0,50})%{DATA:nuix\_second\_short\_message}\n(?m)%{GREEDYDATA:nuix\_multilines}"}
}
} else if "AUTOMATION" in [tags] {
grok {
match =\> {"message" =\> "%{LOGLEVEL:level}%{GREEDYDATA:log\_message}"}
}
} else if "INVESTIGATE" in [tags] and "MULTILINE" in [tags] {
grok {
match =\> {"message" =\> "%{TIMESTAMP\_ISO8601:investigate\_timestamp} %{DATA:message}\n(?m)%{GREEDYDATA:message\_multilines}"}
}
} else if "METRICBEAT" in [tags] {
grok {
match =\> {"message" =\> "%{GREEDYDATA:log\_message}"}
}
} else if "INVESTIGATE" in [tags] {
grok {
match =\> {"message" =\> "%{TIMESTAMP\_ISO8601:investigate\_timestamp}%{GREEDYDATA:log\_message}"}
}
} else if "\_grokparsefailure" in [tags] {
grok {
match =\> {"message" =\> "%{GREEDYDATA:log\_message}"}
}
} else if "NUIX-RUNNER-START-WS" in [tags] and "NUIX-LOG" in [tags] {
grok {
match =\> {"message" =\> "%{TIMESTAMP\_ISO8601:nuix\_timestamp} %{NUMBER:nuix\_timezone} [%{DATA:nuix\_source}] %{NUMBER:nuix\_thread} %{LOGLEVEL:nuix\_loglevel}%{SPACE}%{DATA:nuix\_class} - (?\.+?(?=\Stores\DecryptionKeys\pgpkeylist.dat))"}
}
mutate {
gsub =\> [
# replace backslashes and forward slashes with a hyphen
"case\_name", "[\/]", "€ "
]
}
grok {
match =\> {"case\_name" =\> " (?\([^€]+$))"}
}
} else if "OCR-LOG" in [tags] {
grok {
match =\> {"message" =\> "%{GREEDYDATA:log\_message}"}
}
} else if "ABBY-WORKER-LOG" in [tags] {
grok {
match =\> {"message" =\> "%{GREEDYDATA:log\_message}"}
}
} else {
grok {
match =\> {"message" =\> "%{TIMESTAMP\_ISO8601:nuix\_timestamp} %{NUMBER:nuix\_timezone} [%{DATA:nuix\_source}] %{NUMBER:nuix\_thread} %{LOGLEVEL:nuix\_loglevel}%{SPACE}%{DATA:nuix\_class} - %{GREEDYDATA:automation\_status}"}
add\_tag =\> ["SINGLE-LINE"]
}
}
```
if "NUIX-RUNNER-START" in [tags] {
ruby {
code => '
event.set("[nuix_log]", {
"runner" => event.get("nuix_class").delete_prefix("SCRIPT.").delete_suffix(".gen"),
"path" => File.dirname(event.get("[log][file][path]")),
"host" => event.get("[agent][hostname]"),
})
'
}
aggregate {
task_id => "%{[nuix_log][host]}"
code => "map[event.get('[nuix_log][path]')] = event.get('[nuix_log]')"
map_action => "create"
}
} else if "NUIX-RUNNER-END" in [tags] {
ruby {
code => 'event.set("[nuix_log][host]", event.get("[agent][hostname]"))'
}
aggregate {
task_id => "%{[nuix_log][host]}"
code => '
log_dir = File.dirname(event.get("[log][file][path]"))
if map.key?(log_dir)
event.set("[nuix_log]", map[log_dir])
end
'
timeout => 2629743
inactivity_timeout => 300
map_action => "update"
end_of_task => true
}
} else {
ruby {
code => 'event.set("[nuix_log][host]", event.get("[agent][hostname]"))'
}
aggregate {
task_id => "%{[nuix_log][host]}"
code => '
log_dir = File.dirname(event.get("[log][file][path]"))
(0..6).each do |i|
if map.key?(log_dir)
event.set("[nuix_log]", map[log_dir])
break
end
log_dir = File.dirname(log_dir)
end
'
map_action => "update"
}
```
# grok { match =\> { "[nuix\_log][path]" =\> "/\w+/%{WORD:[@metadata][appname]}-" } add\_tag =\> ["%{[@metadata][appname]}" ] }
```
}
```
}
##########################################################
filter {
if "MULTILINE" in [tags] and "NUIX-LOG" in [tags] and "NUIX-MAIN-LOG-ERROR" in [tags] {
mutate {
add\_field =\> { "nuix\_error\_message" =\> "%{nuix\_short\_message}%{nuix\_second\_short\_message}" }
}
}
}
filter {
if "MULTILINE" in [tags] and "NUIX-LOG" in [tags] and "NUIX-WORKER-LOG-ERROR" in [tags] {
mutate {
add\_field =\> { "nuix\_error\_message" =\> "%{nuix\_short\_message}%{nuix\_second\_short\_message}" }
}
}
}
filter {
if "SINGLE-LINE" in [tags] and "NUIX-LOG" in [tags] and "NUIX-WORKER-LOG-ERROR" in [tags] {
grok {
match =\> {"message" =\> "%{TIMESTAMP\_ISO8601:nuix\_timestamp\_blaj} %{NUMBER:nuix\_timezone\_blaj} [%{DATA:nuix\_source\_blaj}] %{NUMBER:nuix\_thread-blaj} %{GREEDYDATA:nuix\_error\_message}"}
}
mutate {
remove\_field =\> ["nuix\_timestamp\_blaj", "nuix\_timezone\_blaj", "nuix\_source\_blaj", "uix\_thread-blaj"]
}
}
}
filter {
if "SINGLE-LINE" in [tags] and "NUIX-LOG" in [tags] and "NUIX-MAIN-LOG-ERROR" in [tags] {
grok {
match =\> {"message" =\> "%{TIMESTAMP\_ISO8601:nuix\_timestamp-blaj} %{NUMBER:nuix\_timezone\_blaj} [%{DATA:nuix\_source\_blaj}] %{NUMBER:nuix\_thread\_blaj} %{GREEDYDATA:nuix\_error\_message}"}
}
mutate {
remove\_field =\> ["nuix\_timestamp\_blaj", "nuix\_timezone\_blaj", "nuix\_source\_blaj", "uix\_thread-blaj"]
}
}
}
###########################################################
filter {
if "NUIX-SINGLE-WORKER" in [tags] {
mutate { convert =\> ["process\_pid", "string"] }
ruby {
code =\> '
event.set("nuix\_running\_worker", event.get("[process][name]") + "-" + event.get("process\_pid"))
'
}
}
}
#filter {
# if "DISK-SPACE-UTIL" in [tags] {
# mutate { convert =\> ["[system][filesystem][used][bytes]", "string"] }
# mutate { convert =\> ["[system][filesystem][total]", "string"] }
# mutate {
# add\_field =\> {
# "nuix\_disc\_event" =\> " Drive %{[system][filesystem][device\_name]} %{[system][filesystem][used][bytes]} of %{[system][filesystem][total]}"
# }
# }
# }
#}
#filter {
# if "DISK-SPACE-UTIL" in [tags] {
# mutate { convert =\> ["system\_filesystem\_free", "string"] }
# mutate { convert =\> ["system\_filesystem\_total", "string"] }
# mutate {
# add\_field =\> {
# "nuix\_disc\_event" =\> " Drive %{[system][filesystem][device\_name]} %{system\_filesystem\_free} of %{system\_filesystem\_total}"
# }
# }
#}
############ From here Metricbeat
#event.set("nuix\_running\_worker", "Host=" + event.get("[agent][hostname]") + event.get("[process][name]") + event.get("process\_pid"))
filter {
if "NUIX-LOG" in [tags] {
date {
match =\> ["nuix\_timestamp", "ISO8601", "YYYY-MM-dd HH:mm:ss", "YYYY-MM-dd HH:mm:ss.ZZZ"]
target =\> "@timestamp"
locale =\> "en"
}
}
}
filter {
if "INVESTIGATE" in [tags] {
date {
match =\> ["investigate\_timestamp", "ISO8601", "YYYY-MM-dd HH:mm:ss", "YYYY-MM-dd HH:mm:ss.ZZZ"]
target =\> "@timestamp"
locale =\> "en"
}
}
}
filter {
if "RU" in [tags] {
elapsed {
start\_tag =\> "STARTING-RUNNER"
end\_tag =\> "ENDING-RUNNER"
periodic\_flush =\> false
unique\_id\_field =\> "runner"
timeout =\> 2629743
new\_event\_on\_match =\> false
}
} else if "ST" in [tags] {
elapsed {
start\_tag =\> "STARTING-STAGE"
end\_tag =\> "ENDING-STAGE"
periodic\_flush =\> false
unique\_id\_field =\> "[automation][stage\_id]"
timeout =\> 2629743
new\_event\_on\_match =\> false
}
}
}
filter {
if [elapsed\_time] {
ruby {
code =\> '
event.set("[elapsed\_hr]", Time.at(event.get("[elapsed\_time]")).utc.strftime("%T.%L"))
'
}
}
}
filter {
if "STARTING-RUNNER" in [tags] {
ruby {
code =\> '
t = event.get("@timestamp")
event.set("elapsed\_timestamp\_event\_start\_runner", Time.at(t.to\_f).strftime("%Y-%m-%d %H:%M:%S"))
'
}
mutate {
add\_field =\> { "elapsed\_timestamp\_event\_end\_runner" =\> "---------------\>" }
}
mutate {
add\_field =\> { "elapsed\_hr" =\> "Runner Started" }
}
}
}
filter {
if "STARTING-STAGE" in [tags] {
ruby {
code =\> '
t = event.get("@timestamp")
event.set("elapsed\_timestamp\_event\_start\_stage", Time.at(t.to\_f).strftime("%Y-%m-%d %H:%M:%S"))
'
}
mutate {
add\_field =\> { "elapsed\_timestamp\_event\_end\_stage" =\> "---------------\>" }
}
mutate {
add\_field =\> { "elapsed\_hr" =\> "Stage Started" }
}
}
}
filter {
if "ENDING-STAGE" in [tags] {
ruby {
code =\> '
t = event.get("elapsed\_timestamp\_start")
event.set("elapsed\_timestamp\_event\_start\_stage", Time.at(t.to\_f).strftime("%Y-%m-%d %H:%M:%S"))
'
}
}
}
filter {
if "ENDING-RUNNER" in [tags] {
ruby {
code =\> '
t = event.get("elapsed\_timestamp\_start")
event.set("elapsed\_timestamp\_event\_start\_runner", Time.at(t.to\_f).strftime("%Y-%m-%d %H:%M:%S"))
'
}
}
}
filter {
if "AUTOMATION-FINISHED-STAGE" in [tags] {
ruby {
code =\> '
t = event.get("@timestamp")
event.set("elapsed\_timestamp\_event\_end\_stage", Time.at(t.to\_f).strftime("%Y-%m-%d %H:%M:%S"))
'
}
}
}
filter {
if "AUTOMATION-FINISHED-RUNNER" in [tags] {
ruby {
code =\> '
t = event.get("@timestamp")
event.set("elapsed\_timestamp\_event\_end\_runner", Time.at(t.to\_f).strftime("%Y-%m-%d %H:%M:%S"))
'
}
}
}
filter {
mutate {
remove\_tag =\> ["beats\_input\_codec\_plain\_applied", "beats\_input\_raw\_event"]
}
}
filter {
if "\_grokparsefailure" in [tags] {
drop{}
}
}
#filter {
# if "NUIX-LOG" in [tags] {
# mutate {
# copy =\> { "[nuix\_log][runner]" =\> "nuix\_case" }
# }
# }
#}
output {
elasticsearch {
hosts =\> ["[http://automation.avian.dk:9200](http://automation.avian.dk:9200)", "[http://automation.avian.dk:9201](http://automation.avian.dk:9201)"]
index =\> "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"
}
}
---
### Author:  [@melorium](https://discuss.elastic.co/u/melorium)
#### Post date: [January 23, 2022, 12:05pm UTC](https://discuss.elastic.co/t/logstash-aggregate-problem-tag-in-subdirectories-please-help/295116/5 "2022-01-23T12:05:08Z")
1. input {
2. beats {
3.
```
port => 5044
```
4. }
5. }
6. filter {
7.
```
if "AUTOMATION" in [tags] {
```
8.
```
grok {
```
9.
```
match => { "ts" => "(?^.{0,23})" }
```
10.
```
}
```
11.
```
date {
```
12.
```
match => ["ts1", "ISO8601", "YYYY-MM-dd HH:mm:ss", "YYYY-MM-dd HH:mm:ss.ZZZ", "UNIX"]
```
13.
```
target => "@timestamp"
```
14.
```
locale => "en"
```
15.
```
}
```
16.
```
ruby {
```
17.
```
code => '
```
18.
```
t = event.get("@timestamp")
```
19.
```
event.set("automation_timestamp_fixed", Time.at(t.to_f).strftime("%Y-%m-%d %H:%M:%S"))
```
20.
```
'
```
21.
```
}
```
22.
```
}
```
23.
24. }
25.
26. filter {
27.
```
if "MULTILINE" in [tags] and "NUIX-LOG" in [tags] {
```
28.
```
grok {
```
29.
```
match => {"message" => "%{TIMESTAMP_ISO8601:nuix_timestamp} %{NUMBER:nuix_timezone} \[%{DATA:nuix_source}] %{NUMBER:nuix_thread} %{LOGLEVEL:nuix_loglevel}%{SPACE}%{DATA:nuix_class} - (?[\s\S]{0,50})%{DATA:nuix_second_short_message}\n(?m)%{GREEDYDATA:nuix_multilines}"}
```
30.
```
}
```
31.
```
} else if "AUTOMATION" in [tags] {
```
32.
```
grok {
```
33.
```
match => {"message" => "%{LOGLEVEL:level}%{GREEDYDATA:log_message}"}
```
34.
```
}
```
35.
```
} else if "INVESTIGATE" in [tags] and "MULTILINE" in [tags] {
```
36.
```
grok {
```
37.
```
match => {"message" => "%{TIMESTAMP_ISO8601:investigate_timestamp} %{DATA:message}\n(?m)%{GREEDYDATA:message_multilines}"}
```
38.
```
}
```
39.
```
} else if "METRICBEAT" in [tags] {
```
40.
```
grok {
```
41.
```
match => {"message" => "%{GREEDYDATA:log_message}"}
```
42.
```
}
```
43.
```
} else if "INVESTIGATE" in [tags] {
```
44.
```
grok {
```
45.
```
match => {"message" => "%{TIMESTAMP_ISO8601:investigate_timestamp}%{GREEDYDATA:log_message}"}
```
46.
```
}
```
47.
```
} else if "_grokparsefailure" in [tags] {
```
48.
```
grok {
```
49.
```
match => {"message" => "%{GREEDYDATA:log_message}"}
```
50.
```
}
```
51.
```
} else if "NUIX-RUNNER-START-WS" in [tags] and "NUIX-LOG" in [tags] {
```
52.
```
grok {
```
53.
```
match => {"message" => "%{TIMESTAMP_ISO8601:nuix_timestamp} %{NUMBER:nuix_timezone} \[%{DATA:nuix_source}] %{NUMBER:nuix_thread} %{LOGLEVEL:nuix_loglevel}%{SPACE}%{DATA:nuix_class} - (?.+?(?=\\Stores\\DecryptionKeys\\pgpkeylist.dat))"}
```
54.
```
}
```
55.
```
mutate {
```
56.
```
gsub => [
```
57.
```
# replace backslashes and forward slashes with a hyphen
```
58.
```
"case_name", "[\\/]", "€ "
```
59.
```
]
```
60.
```
}
```
61.
```
grok {
```
62.
```
match => {"case_name" => " (?([^€]+$))"}
```
63.
```
}
```
64.
```
} else if "OCR-LOG" in [tags] {
```
65.
```
grok {
```
66.
```
match => {"message" => "%{GREEDYDATA:log_message}"}
```
67.
```
}
```
68.
```
} else if "ABBY-WORKER-LOG" in [tags] {
```
69.
```
grok {
```
70.
```
match => {"message" => "%{GREEDYDATA:log_message}"}
```
71.
```
}
```
72.
```
} else {
```
73.
```
grok {
```
74.
```
match => {"message" => "%{TIMESTAMP_ISO8601:nuix_timestamp} %{NUMBER:nuix_timezone} \[%{DATA:nuix_source}] %{NUMBER:nuix_thread} %{LOGLEVEL:nuix_loglevel}%{SPACE}%{DATA:nuix_class} - %{GREEDYDATA:automation_status}"}
```
75.
```
add_tag => ["SINGLE-LINE"]
```
76.
```
}
```
77.
```
}
```
78.
79.
80.
```
if "NUIX-RUNNER-START" in [tags] {
```
81.
```
ruby {
```
82.
```
code => '
```
83.
```
event.set("[nuix_log]", {
```
84.
```
"runner" => event.get("nuix_class").delete_prefix("SCRIPT.").delete_suffix(".gen"),
```
85.
```
"path" => File.dirname(event.get("[log][file][path]")),
```
86.
```
"host" => event.get("[agent][hostname]"),
```
87.
```
})
```
88.
```
'
```
89.
```
}
```
90.
```
aggregate {
```
91.
```
task_id => "%{[nuix_log][host]}"
```
92.
```
code => "map[event.get('[nuix_log][path]')] = event.get('[nuix_log]')"
```
93.
```
map_action => "create"
```
94.
```
}
```
95.
```
} else if "NUIX-RUNNER-END" in [tags] {
```
96.
```
ruby {
```
97.
```
code => 'event.set("[nuix_log][host]", event.get("[agent][hostname]"))'
```
98.
```
}
```
99.
```
aggregate {
```
100.
```
task_id => "%{[nuix_log][host]}"
```
101.
```
code => '
```
102.
```
log_dir = File.dirname(event.get("[log][file][path]"))
```
103.
```
if map.key?(log_dir)
```
104.
```
event.set("[nuix_log]", map[log_dir])
```
105.
```
end
```
106.
```
'
```
107.
```
timeout => 2629743
```
108.
```
inactivity_timeout => 300
```
109.
```
map_action => "update"
```
110.
```
end_of_task => true
```
111.
```
}
```
112.
```
} else {
```
113.
```
ruby {
```
114.
```
code => 'event.set("[nuix_log][host]", event.get("[agent][hostname]"))'
```
115.
```
}
```
116.
```
aggregate {
```
117.
```
task_id => "%{[nuix_log][host]}"
```
118.
```
code => '
```
119.
```
log_dir = File.dirname(event.get("[log][file][path]"))
```
120.
```
(0..6).each do |i|
```
121.
```
if map.key?(log_dir)
```
122.
```
event.set("[nuix_log]", map[log_dir])
```
123.
```
break
```
124.
```
end
```
125.
```
log_dir = File.dirname(log_dir)
```
126.
```
end
```
127.
```
'
```
128.
```
map_action => "update"
```
129.
130.
131.
132.
```
}
```
133.
# grok { match =\> { "[nuix\_log][path]" =\> "/\w+/%{WORD:[@metadata][appname]}-" } add\_tag =\> ["%{[@metadata][appname]}" ] }
134.
135.
136.
```
}
```
137. }
138. ##########################################################
139. filter {
140.
```
if "MULTILINE" in [tags] and "NUIX-LOG" in [tags] and "NUIX-MAIN-LOG-ERROR" in [tags] {
```
141.
```
mutate {
```
142.
```
add_field => { "nuix_error_message" => "%{nuix_short_message}%{nuix_second_short_message}" }
```
143.
```
}
```
144.
```
}
```
145. }
146. filter {
147.
```
if "MULTILINE" in [tags] and "NUIX-LOG" in [tags] and "NUIX-WORKER-LOG-ERROR" in [tags] {
```
148.
```
mutate {
```
149.
```
add_field => { "nuix_error_message" => "%{nuix_short_message}%{nuix_second_short_message}" }
```
150.
```
}
```
151.
```
}
```
152. }
153. filter {
154.
```
if "SINGLE-LINE" in [tags] and "NUIX-LOG" in [tags] and "NUIX-WORKER-LOG-ERROR" in [tags] {
```
155.
```
grok {
```
156.
```
match => {"message" => "%{TIMESTAMP_ISO8601:nuix_timestamp_blaj} %{NUMBER:nuix_timezone_blaj} \[%{DATA:nuix_source_blaj}] %{NUMBER:nuix_thread-blaj} %{GREEDYDATA:nuix_error_message}"}
```
157.
```
}
```
158.
```
mutate {
```
159.
```
remove_field => ["nuix_timestamp_blaj", "nuix_timezone_blaj", "nuix_source_blaj", "uix_thread-blaj"]
```
160.
```
}
```
161.
```
}
```
162. }
163. filter {
164.
```
if "SINGLE-LINE" in [tags] and "NUIX-LOG" in [tags] and "NUIX-MAIN-LOG-ERROR" in [tags] {
```
165.
```
grok {
```
166.
```
match => {"message" => "%{TIMESTAMP_ISO8601:nuix_timestamp-blaj} %{NUMBER:nuix_timezone_blaj} \[%{DATA:nuix_source_blaj}] %{NUMBER:nuix_thread_blaj} %{GREEDYDATA:nuix_error_message}"}
```
167.
```
}
```
168.
```
mutate {
```
169.
```
remove_field => ["nuix_timestamp_blaj", "nuix_timezone_blaj", "nuix_source_blaj", "uix_thread-blaj"]
```
170.
```
}
```
171.
```
}
```
172. }
173. ###########################################################
174. filter {
175.
```
if "NUIX-SINGLE-WORKER" in [tags] {
```
176.
```
mutate { convert => ["process_pid", "string"] }
```
177.
```
ruby {
```
178.
```
code => '
```
179.
```
event.set("nuix_running_worker", event.get("[process][name]") + "-" + event.get("process_pid"))
```
180.
```
'
```
181.
```
}
```
182.
```
}
```
183. }
184. #filter {
185.
# if "DISK-SPACE-UTIL" in [tags] {
186.
# mutate { convert =\> ["[system][filesystem][used][bytes]", "string"] }
187.
# mutate { convert =\> ["[system][filesystem][total]", "string"] }
188.
# mutate {
189.
# add\_field =\> {
190.
# "nuix\_disc\_event" =\> " Drive %{[system][filesystem][device\_name]} %{[system][filesystem][used][bytes]} of %{[system][filesystem][total]}"
191.
# }
192.
# }
193.
# }
194. #}
195. #filter {
196.
# if "DISK-SPACE-UTIL" in [tags] {
197.
# mutate { convert =\> ["system\_filesystem\_free", "string"] }
198.
# mutate { convert =\> ["system\_filesystem\_total", "string"] }
199.
# mutate {
200.
# add\_field =\> {
201.
# "nuix\_disc\_event" =\> " Drive %{[system][filesystem][device\_name]} %{system\_filesystem\_free} of %{system\_filesystem\_total}"
202.
# }
203.
# }
204. #}
205. ############ From here Metricbeat
206. #event.set("nuix\_running\_worker", "Host=" + event.get("[agent][hostname]") + event.get("[process][name]") + event.get("process\_pid"))
207. filter {
208.
```
if "NUIX-LOG" in [tags] {
```
209.
```
date {
```
210.
```
match => ["nuix_timestamp", "ISO8601", "YYYY-MM-dd HH:mm:ss", "YYYY-MM-dd HH:mm:ss.ZZZ"]
```
211.
```
target => "@timestamp"
```
212.
```
locale => "en"
```
213.
```
}
```
214.
```
}
```
215. }
216. filter {
217.
```
if "INVESTIGATE" in [tags] {
```
218.
```
date {
```
219.
```
match => ["investigate_timestamp", "ISO8601", "YYYY-MM-dd HH:mm:ss", "YYYY-MM-dd HH:mm:ss.ZZZ"]
```
220.
```
target => "@timestamp"
```
221.
```
locale => "en"
```
222.
```
}
```
223.
```
}
```
224. }
225. filter {
226.
```
if "RU" in [tags] {
```
227.
```
elapsed {
```
228.
```
start_tag => "STARTING-RUNNER"
```
229.
```
end_tag => "ENDING-RUNNER"
```
230.
```
periodic_flush => false
```
231.
```
unique_id_field => "runner"
```
232.
```
timeout => 2629743
```
233.
```
new_event_on_match => false
```
234.
```
}
```
235.
```
} else if "ST" in [tags] {
```
236.
```
elapsed {
```
237.
```
start_tag => "STARTING-STAGE"
```
238.
```
end_tag => "ENDING-STAGE"
```
239.
```
periodic_flush => false
```
240.
```
unique_id_field => "[automation][stage_id]"
```
241.
```
timeout => 2629743
```
242.
```
new_event_on_match => false
```
243.
```
}
```
244.
```
}
```
245. }
246. filter {
247.
```
if [elapsed_time] {
```
248.
```
ruby {
```
249.
```
code => '
```
250.
```
event.set("[elapsed_hr]", Time.at(event.get("[elapsed_time]")).utc.strftime("%T.%L"))
```
251.
```
'
```
252.
```
}
```
253.
```
}
```
254. }
255. filter {
256. if "STARTING-RUNNER" in [tags] {
257.
```
ruby {
```
258.
```
code => '
```
259.
```
t = event.get("@timestamp")
```
260.
```
event.set("elapsed_timestamp_event_start_runner", Time.at(t.to_f).strftime("%Y-%m-%d %H:%M:%S"))
```
261.
```
'
```
262.
```
}
```
263.
```
mutate {
```
264.
```
add_field => { "elapsed_timestamp_event_end_runner" => "--------------->" }
```
265.
```
}
```
266.
```
mutate {
```
267.
```
add_field => { "elapsed_hr" => "Runner Started" }
```
268.
```
}
```
269.
```
}
```
270. }
271. filter {
272. if "STARTING-STAGE" in [tags] {
273.
```
ruby {
```
274.
```
code => '
```
275.
```
t = event.get("@timestamp")
```
276.
```
event.set("elapsed_timestamp_event_start_stage", Time.at(t.to_f).strftime("%Y-%m-%d %H:%M:%S"))
```
277.
```
'
```
278.
```
}
```
279.
```
mutate {
```
280.
```
add_field => { "elapsed_timestamp_event_end_stage" => "--------------->" }
```
281.
```
}
```
282.
```
mutate {
```
283.
```
add_field => { "elapsed_hr" => "Stage Started" }
```
284.
```
}
```
285.
```
}
```
286. }
287. filter {
288. if "ENDING-STAGE" in [tags] {
289.
```
ruby {
```
290.
```
code => '
```
291.
```
t = event.get("elapsed_timestamp_start")
```
292.
```
event.set("elapsed_timestamp_event_start_stage", Time.at(t.to_f).strftime("%Y-%m-%d %H:%M:%S"))
```
293.
```
'
```
294.
```
}
```
295. }
296. }
297. filter {
298. if "ENDING-RUNNER" in [tags] {
299.
```
ruby {
```
300.
```
code => '
```
301.
```
t = event.get("elapsed_timestamp_start")
```
302.
```
event.set("elapsed_timestamp_event_start_runner", Time.at(t.to_f).strftime("%Y-%m-%d %H:%M:%S"))
```
303.
```
'
```
304.
```
}
```
305. }
306. }
307. filter {
308. if "AUTOMATION-FINISHED-STAGE" in [tags] {
309.
```
ruby {
```
310.
```
code => '
```
311.
```
t = event.get("@timestamp")
```
312.
```
event.set("elapsed_timestamp_event_end_stage", Time.at(t.to_f).strftime("%Y-%m-%d %H:%M:%S"))
```
313.
```
'
```
314.
```
}
```
315. }
316. }
317. filter {
318. if "AUTOMATION-FINISHED-RUNNER" in [tags] {
319.
```
ruby {
```
320.
```
code => '
```
321.
```
t = event.get("@timestamp")
```
322.
```
event.set("elapsed_timestamp_event_end_runner", Time.at(t.to_f).strftime("%Y-%m-%d %H:%M:%S"))
```
323.
```
'
```
324.
```
}
```
325. }
326. }
327. filter {
328. mutate {
329.
```
remove_tag => ["beats_input_codec_plain_applied", "beats_input_raw_event"]
```
330. }
331. }
332. filter {
333. if "\_grokparsefailure" in [tags] {
334.
```
drop{}
```
335.
```
}
```
336. }
337. #filter {
338.
# if "NUIX-LOG" in [tags] {
339.
# mutate {
340.
# copy =\> { "[nuix\_log][runner]" =\> "nuix\_case" }
341.
# }
342.
# }
343. #}
344. output {
345. elasticsearch {
346.
```
hosts => ["http://automation.avian.dk:9200", "http://automation.avian.dk:9201"]
```
347.
```
index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"
```
348. }
349. }
350. emphasized text
---
### Author:  [@melorium](https://discuss.elastic.co/u/melorium)
#### Post date: [January 23, 2022, 12:06pm UTC](https://discuss.elastic.co/t/logstash-aggregate-problem-tag-in-subdirectories-please-help/295116/6 "2022-01-23T12:06:13Z")
Sorry I dont know how to post code here as in a windows
---
### Author:  [@melorium](https://discuss.elastic.co/u/melorium)
#### Post date: [January 23, 2022, 12:11pm UTC](https://discuss.elastic.co/t/logstash-aggregate-problem-tag-in-subdirectories-please-help/295116/7 "2022-01-23T12:11:48Z")
The case name is just showing upp in one row in the logs. So I need to remember that over all events in the log. and also I have a stop event in the logs.
Like the old memorize filter but I can't get that filter to work in 7.16
All events between start and stop should have a tags with case name.
I feel stupid that I can't solve it
Dennis
---
### Author:  [@Badger](https://discuss.elastic.co/u/Badger)
#### Post date: [January 23, 2022, 3:20pm UTC](https://discuss.elastic.co/t/logstash-aggregate-problem-tag-in-subdirectories-please-help/295116/8 "2022-01-23T15:20:04Z")
You need to use markdown. If you Google "markdown tutorial" you will find multiple sites that provide one. Use the preview pane on the right of the edit pane to make sure the code is formatted correctly.
---
### Author:  [@system](https://discuss.elastic.co/u/system)
#### Post date: [February 20, 2022, 3:20pm UTC](https://discuss.elastic.co/t/logstash-aggregate-problem-tag-in-subdirectories-please-help/295116/9 "2022-02-20T15:20:15Z")
This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.