Log Filtration Issue with Filebeat and Logstash Configuration

Elastic Stack Logstash
1

Hello everyone,

I'm using Filebeat to send three different logs to Logstash, where I'm applying parsing through filters. In the parsing process, I've taken into consideration the logs that are visible in the observation section of Kibana. While the logs are successfully being transferred to Elasticsearch, I'm encountering an issue where no filtration seems to occur while everything is working fine. Indices and data sync.

To elaborate, the format of the logs remains the same both before and after applying the filter section. I've attached my Logstash and Filebeat configuration files for reference.

Any insights or suggestions on why the filtration might not work as expected would be greatly appreciated.

Logstash Conf


input {
  beats {
    port => 5045
  }
}

filter {
  if "jicofo" in [tags] {
    grok {
      match => { "event.original" => "Jicofo %{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:log_level}: \[%{POSINT:process_id}\] \[room=%{DATA:meeting_name}@%{DATA}(?:\s*meeting_id=%{UUID:meeting_id})?\] %{GREEDYDATA:message}" }
    }
  }
}

output {
  if [tags] and "jicofo" in [tags] {
    elasticsearch {
      hosts => ["http://elasticSearch_ip:9200"]
      index => "jicofo-%{+YYYY}"
    }
  }
}

Thank you in advance for your help!

2

Can you provide a log sample? (2-3 lines)

3 
@timestamp:
    Aug 10, 2023 @ 15:34:47.730
@version:
    1
agent.ephemeral_id:
    048aeaca-ac5e-4582-99be-d524189902de
agent.id:
    48835a24-a1f1-4948-9741-870975a42f7b
agent.name:
    filebeat -ip-address
agent.type:
    filebeat
agent.version:
    8.9.0
app:
    jitsi-meet-jicofo
ecs.version:
    8.0.0
event.original:
    Jicofo 2023-08-10 10:04:47.463 INFO: [5097] [room=roomname@conference.filebeat-ip meeting_id=9d6998ec-1cf5-49d1-bdd2-5d27b3ce5bbf]

Jicofo 2023-08-10 10:04:39.909 INFO: [5097] [room=roomname@conferencece.filebeat-ip meeting_id=9d6998ec-1cf5-49d1-bdd2-5d27b3ce5bbf] ColibriV2SessionManager.removeParticipant#123: Removing 3ea9b689
4

Lines from the jicofo.log file, pls.

6

above logs are of jicofo

7

I don't see "jicofo" in tags. You should use [event][original] or the message field and make something like this:

 if ([event][original] =~ /^Jicofo/ ) {
    grok {
     match => { "[event][original]" => "Jicofo %{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:log_level}: \[%{POSINT:process_id}\] \[room=%{DATA:meeting_name}@%{DATA}(?:\s*meeting_id=%{UUID:meeting_id})?\] %{GREEDYDATA:message}" }
	  overwrite => [ "message" ]
    }
}

Also maybe is better to use GREEDYDATA:logmessage insted of overwriting GREEDYDATA:message. It's up to you.

8

jicofo tag is used in filebeat.yml
I am sharing that too with you and this Grok pattern is working fine in Grok debugger.

filebeat.inputs:
- type: log
  paths:
    - /var/log/jitsi/jicofo.log
  fields:
    app: jitsi-meet-jicofo
  fields_under_root: true
  tags: ["jicofo"]

- type: log
  paths:
    - /var/log/jitsi/jvb.log
  fields:
    app: jitsi-meet-jvb
  fields_under_root: true
  tags: ["jvb"]

- type: log
  paths:
    - /var/log/nginx/access.log
  fields:
    app: nginx
  fields_under_root: true
  tags: ["nginx-access"]

- type: log
  paths:
    - /var/log/nginx/error.log
  fields:
    app: nginx
  fields_under_root: true
  tags: ["nginx-error"]

output.logstash:
  hosts: ["logstash_ip.in:5045"]
  when.contains:
    tags: "jicofo"
  index: "jicofo-%{+yyyy}"

output.logstash:
  hosts: ["logstash_ip.in:5044"]
9

Yes, you can use tags, it's up to you

10

So where is the issue, why my logs are not getting filtered? no error nothing in the logs file too.

11

Should be [event][original]. However should check my post above

12

I did that still getting the same output on kibana discover.

13

Can you show what you get? Kibana or rubydebug

14

same log without any change

Jicofo 2023-08-11 09:34:36.453 INFO: [5795] [room=test8@conference.ipAddress meeting_id=44a3405d-270c-4469-9104-bf8b3b34af4d] JitsiMeetConferenceImpl.acceptSession#1302: Accepted initial sources from 1aac59c0: [audio=[2727804029], video=[3597009703, 868618095, 2972627672, 2019927256, 72401307, 4049052442], groups=[FID[3597009703, 868618095], SIM[3597009703, 2972627672, 2019927256], FID[2972627672, 72401307], FID[2019927256, 4049052442]]]
15

Can you share the entire document you are receiving? Go on discover, expand one of the documents, click on the JSON to see the Json structure and copy and share everything to make it clear what you are receiving.

16

As Leandro said, we need more details. Here is .conf which parse a line.

input {
  generator {
       message => "Jicofo 2023-08-11 09:34:36.453 INFO: [5795] [room=test8@conference.ipAddress meeting_id=44a3405d-270c-4469-9104-bf8b3b34af4d] JitsiMeetConferenceImpl.acceptSession#1302: Accepted initial sources from 1aac59c0: [audio=[2727804029], video=[3597009703, 868618095, 2972627672, 2019927256, 72401307, 4049052442], groups=[FID[3597009703, 868618095], SIM[3597009703, 2972627672, 2019927256], FID[2972627672, 72401307], FID[2019927256, 4049052442]]]"
       count => 1
  }
} 
filter {

 if ([event][original] =~ /^Jicofo/ ) {
    grok {
     match => { "[event][original]" => "Jicofo %{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:log_level}: \[%{POSINT:process_id}\] \[room=%{DATA:meeting_name}@%{DATA}(?:\s*meeting_id=%{UUID:meeting_id})?\] %{GREEDYDATA:message}" }
	  overwrite => [ "message" ]
    }
}

}

output {
   stdout { codec => rubydebug{} }
}

Result:

{
       "log_level" => "INFO",
           "event" => {
        "original" => "Jicofo 2023-08-11 09:34:36.453 INFO: [5795] [room=test8@conference.ipAddress meeting_id=44a3405d-270c-4469-9104-bf8b3b34af4d] JitsiMeetConferenceImpl.acceptSession#1302: Accepted initial sources from 1aac59c0: [audio=[2727804029], video=[3597009703, 868618095, 2972627672, 2019927256, 72401307, 4049052442], groups=[FID[3597009703, 868618095], SIM[3597009703, 2972627672, 2019927256], FID[2972627672, 72401307], FID[2019927256, 4049052442]]]",
        "sequence" => 0
    },
      "meeting_id" => "44a3405d-270c-4469-9104-bf8b3b34af4d",
    "meeting_name" => "test8",
      "@timestamp" => 2023-08-11T12:08:45.964922400Z,
         "message" => "JitsiMeetConferenceImpl.acceptSession#1302: Accepted initial sources from 1aac59c0: [audio=[2727804029], video=[3597009703, 868618095, 2972627672, 2019927256, 72401307, 4049052442], groups=[FID[3597009703, 868618095], SIM[3597009703, 2972627672, 2019927256], FID[2972627672, 72401307], FID[2019927256, 4049052442]]]",
      "process_id" => "5795",
       "timestamp" => "2023-08-11 09:34:36.453"
}
17

Just saw this in your filebeat config, this is not supported, filebeat can have only one output, you need to remove one of those outputs.

I'm not sure which one Filebeat will use, if I'm not wrong it will use the last one. What is the pipeline listening on port 5044?

18

Exactly, one of mentioned outputs of filebeat is elasticsearch which bypass logstash and all it's filters. Seems like that is the issue.

19

Port 5045 is dedicated to the Jicofo log with the "jicofo" tag, and port 5044 is assigned for other logs like Nginx and JVB. In the index management of the Kibana dashboard, I am observing complete outputs for both Jicofo and the others, indicating that both outputs are functioning properly. Additionally, I haven't come across any errors or warnings in the log files.

jicofo.conf file(1st conf file)


input {
  beats {
    port => 5045
  }
}

filter {
  if "jicofo" in [tags] {
    grok {
      match => { "message" => "Jicofo %{TIMESTAMP_ISO8601:my_timestamp} %{LOGLEVEL:my_log_level}: \[%{POSINT:my_process_id}\] \[room=%{DATA:meeting_name}@%{DATA}(?:\s*meeting_id=%{UUID:meeting_id})?\] %{GREEDYDATA:message}" }
    }
  }
}

output {
  if [tags] and "jicofo" in [tags] {
    elasticsearch {
      hosts => ["http://elastic-search-ip:9200"]
      index => "jicofo-%{+YYYY}"
    }
  }
}

Logstash.conf(second conf file)

input {
  beats {
    port => 5044
  }
}

filter {
  if "jvb" in [tags] {
    mutate {
      add_field => { "index_prefix" => "jvb-" }
    }
  } else if "nginx-access" in [tags] {
    mutate {
      add_field => { "index_prefix" => "nginx-access-" }
    }
  } else if "nginx-error" in [tags] {
    mutate {
      add_field => { "index_prefix" => "nginx-error-" }
    }
  }
}

output {
  if [index_prefix] {
    elasticsearch {
      hosts => ["http://elasticSearch_ip:9200"]
      index => "%{[index_prefix]}%{+YYYY}"
    }
  }
}
20

Hey, It worked. I am really thankful to all of you for your time and respond.

1 Like
21

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