//
// Copyright IBM Corp. 2023, 2025 All rights reserved.
// SPDX-License-Identifier: Apache-2.0
//
input {
  kafka{
    #Insert "bootstrap_endpoints" string from the service credential in the Event stream.
    bootstrap_servers => "<bootstrap_endpoints>"
    #Insert the topic name which you created in Event stream.
    topics => "<topic_name>"
    security_protocol => "SASL_SSL"
    sasl_mechanism => "PLAIN"
    #Insert the username and password from the service credential in the Event stream.
    sasl_jaas_config => "org.apache.kafka.common.security.plain.PlainLoginModule required username='<user>'  password='<password>';"
    type =>"Postgress_UC"
    #consumer_thread should be equal to the number of partitions which you set while creating Topic in Event stream.
    consumer_threads => <number of partitions>
    #group_id should be set to a unique value in case you have more than one Kafka plugins configured
    #Please find more details : https://www.elastic.co/guide/en/logstash/current/plugins-inputs-kafka.html#plugins-inputs-kafka-group_id
    group_id => "<group_id>"
    #Set the below field if Account ID is to be included in the server hostname
    add_field => {"include_account_in_host" => "true"}
    # User can add one or more logSourceCRN to filter the event
    add_field => { "crnVal" => ["<CRN_VAL>"] } # i.e ["crn:v1:bluemix:public:databases-for-postgresql:{region}:a/{value1}:{value2}::","crn:v1:bluemix:public:databases-for-postgresql:{region}:a/{value1}:{value2}::"]
  }
}

filter {
    if [type] == "Postgress_UC" {

        mutate {
            add_field => {
                "json_event" => "%{message}"
            }
        }

        json {
            source => "json_event"
            target => "parsed_json"
        }

        mutate {
            add_field => {
                "logSourceCRN" => "%{[parsed_json][data][message][logSourceCRN]}"
            }
        }

        # Define isCRNMatched
        mutate {
            add_field => { "isCRNMatched" => "false" }
        }

        # Check if crnVal is defined
        if [crnVal] {
            # Ensure crn_vals is an array
            ruby {
                code => '
                    isMatched = false
                    crn_vals = event.get("crnVal")
                    log_crn = event.get("logSourceCRN")

                    # If crn_vals is a string, convert it into an array
                    crn_vals = [crn_vals] unless crn_vals.is_a?(Array)

                    crn_vals.each { |val|
                        if val == log_crn
                            isMatched = true
                        end
                    }

                    event.set("isCRNMatched", isMatched.to_s)
                '
            }
        }

        # Check if parsed_json does not contain "databases-for-postgres" and contains either "INFO:" or "INFO"
        if !( [parsed_json] =~ /databases-for-postgres/ ) and
            ( [parsed_json] =~ /INFO:/ or [parsed_json] =~ /INFO/ ) {
            drop { }
        }

        # Apply filter if isCRNMatched is true or crnVal is not defined.
        if [isCRNMatched] == "true" or ![crnVal] {
            if [message] {
                icd_postgresql_guardium_filter {}
            }
        }
    }
}