Custom Indices

[Kobrik Kobrikovic]

Hi,
Is there a possibility for more detailed customization of the indexes? For example, to store data from Fortigate FW in its own index, etc.? I’m concerned about the data retention policy. Some log sources we don't need to keep for more than 3 months. Other log sources we need to archive for a year. So far, I haven’t found a way to achieve this. With the current indexing storage solution, we are forced to keep entire archives for a year, which unnecessarily requires significant capacity on the data storage resources.

[Javier Bejar]

Hi Kobrik,

You can achieve more detailed customization by modifying the Filebeat pipeline, allowing you to configure how and where events are indexed (e.g., storing Fortigate FW data in a separate index). However, this is done at your own risk, as Wazuh does not directly support this customization. Be aware that many views and dashboards in Wazuh rely on the default alerts index, so this change might affect them.

For more details on Wazuh indexer architecture and Filebeat configuration, you can refer to the following documentation:

• Wazuh Indexer architecture
• Filebeat configuration

Let us know if you have further questions!

Best regards,
Javier

[wazuh]

Hi Kobrik,

I've tried many ways before, and the best way i found was to direct filebeat to send all logs through a logstash server where you can do additional processing. and create more indices custom indices. In the example below (my configuration) for multi-index it changes from wazuh-alerts-4.x-2024.01.01 to wazuh-alerts-clientone-2024.01.01.

The idea is to create a keydb type of list and reference it for rule processing. in this example logstash is looking for a field called location (for Fortigate we use the location field, which is an IP address from where the fortigate logs come from), then it checks if the IP found in the log exists in the /etc/logstash/keydb/location.csv file and assigns the index_suffix a value - either "default" or the value that correlates the location key, so if the location was 111.111.111.111 logstash would send the log to the custom index wazuh-alerts-4.x-fortigateclient1-2024.01.01, and if it does not meet any of values it will be wazuh-alerts-4.x-default-2024.01.01.

location.csv file looks like this:
111.111.111.111,fortigateclient1

122.122.222.222,fortigateclient2

 if [location] {
    translate {
      source => "[location]"
      target => "[@metadata][index_suffix]"
      dictionary_path => "/etc/logstash/keydb/location.csv"
      fallback => "default"
    }


The reason i use a custom wazuh-template for output is because from creating multi-custom indexes you will also start creating instead of 6 shards per day an additional 6 per each indice you create, which becomes a problem if you have many custom indexes (default maximum amount of shards is 1000, but it can be increased by either adding more indexer nodes or increasing the maximum shard per node, which will have impact on your resource). I've made it so i create 1 primary and 1 replica instead of 3 primaries and 1 replica for each primary per index.

input {
  beats {
    port => 5044
  }
  beats {
    port => 5045
    ssl_enabled => true
    ssl_certificate_authorities => ["/etc/logstash/certs/root-ca.pem"]
    ssl_certificate => "/etc/logstash/certs/logstash.pem"
    ssl_key => "/etc/logstash/certs/logstash-key.pem"
    ssl_client_authentication => "required"
  }
}
filter {
  json {
    source => "message"
  }
  mutate {
    replace => { "[host]" => "%{[host][name]}" }
  }
  if [location] {
    translate {
      source => "[location]"
      target => "[@metadata][index_suffix]"
      dictionary_path => "/etc/logstash/keydb/location.csv"
      fallback => "default"
    }
  } else if [data][office365][OrganizationId] {
    translate {
      source => "[data][office365][OrganizationId]"
      target => "[@metadata][index_suffix]"
      dictionary_path => "/etc/logstash/keydb/o365.csv"
      fallback => "default"
    }
  } else if [data][ms-graph][tennantId] {
    translate {
      source => "[data][ms-graph][tennantId]"
      target => "[@metadata][index_suffix]"
      dictionary_path => "/etc/logstash/keydb/o365.csv"
      fallback => "default"
    }
  } else {
    mutate {
      add_field => { "[@metadata][index_suffix]" => "default" }
    }
  }

 if [previous_output] {
    mutate {
      remove_field => ["previous_output"]
    }
  }
  if [message] {
    mutate {
      remove_field => ["message"]
    }
  }
  mutate {
    remove_field => ["[event][original]"]
  }
}
output {
  opensearch {
    hosts => ["ipofindexer1:9200", "ipofindexer2:9200"]
    user => "$username"
    password => "$password"
    ssl => true
    cacert => "/etc/logstash/certs/root-ca.pem"
    index => "wazuh-alerts-4.x-%{[@metadata][index_suffix]}-%{+YYYY.MM.dd}"
    template => "/etc/logstash/templates/wazuh-template.json"
    template_name => "wazuh-template"
    template_overwrite => true
    manage_template => true
  }
}

[Kobrik Kobrikovic]

Hi,

thank you for your reply and the detailed instructions. This approach of using Logstash to route logs to custom indexes based on log sources looks like it could be a suitable solution for our needs.

I would like to look into this further and implement custom indexes for different log sources so that we can manage the length of retention effectively. It's exactly what I was looking for - the ability to separate logs from different devices, like Fortigate FW, and add a custom retention policy.

I also appreciate the tip on optimizing the number of shards per index. I'll consider the impact on our infrastructure and consider adjusting the shard setup so that it doesn't put unnecessary strain on our resources.

Thank you for your time and willingness to help!

Best regards,
Kobri

Dne úterý 1. října 2024 v 13:36:54 UTC+2 uživatel wazuh napsal: