input {
 tcp {
      port => <PORT_NUMBER>
      type => "syslog-cockroachdb"
      ssl_enabled => true
      # ssl_certificate_authorities => ["${SSL_DIR}/ca.pem"]
      # ssl_certificate => "${SSL_DIR}/tls.crt"
      # ssl_key => "${SSL_DIR}/tls.key"
      # Uncomment the following line for Guardium Data Protection v12.2.3 (or patch 5008) and above to use encrypted private keys
      # ssl_key_passphrase => "${ssl_key_passphrase}"
      # ssl_client_authentication => "none"
 }
}

filter {
  if [type] == "syslog-cockroachdb" {

    # Drop system-generated queries FIRST (before any parsing) - optimized for performance
    # Check for internal execution markers and system users
    if [message] =~ /"intExec":/ or
       [message] =~ /"User":"node"/ or
       [message] =~ /"User":"root"/ or
       [message] =~ /"ApplicationName":"\$ internal/ or
       [message] =~ /"ExecMode":"exec-internal"/ or
       [message] =~ /system\.(jobs|lease|sql_instances|job_info|statement_statistics|transaction_statistics|job_progress_history|reports_meta|statement_diagnostics_requests|scheduled_jobs|settings|zones|tenants|migrations|eventlog|privileges|protected_ts_meta)/ {
      drop { }
    }

    # Drop CockroachDB configuration/informational messages that don't have JSON payloads
    if [message] =~ /\[config\]/ and [message] !~ /\{.*\}$/ {
      drop { }
    }

    # Extract CockroachDB JSON. If the rsyslog template injects server metadata
    # (serverHostname=<host> serverPort=<port>) as a prefix, extract those too.
    # See README for rsyslog template configuration.
    grok {
      match => {
        "message" => [
          "^serverHostname=%{NOTSPACE:server_hostname} serverPort=(?<server_port>[^ {]+)(?:.*\s)?(?<cockroach_wrapper_json>\{.*\})$",
          "(?:.*\s)?(?<cockroach_wrapper_json>\{.*\})$"
        ]
      }
    }

    # Remove grok failure tags if parsing succeeded
    if [cockroach_wrapper_json] {
      mutate {
        remove_tag => ["_grokparsefailure", "_grokparsefailure_sysloginput"]
      }
    }

    # Parse the outer wrapper JSON (contains channel, tags, and nested event)
    if [cockroach_wrapper_json] {
      json {
        source => "cockroach_wrapper_json"
        target => "cockroach_wrapper"
      }

      # Extract the nested "event" object which contains the actual audit data
      if [cockroach_wrapper][event] {
        # Copy the event object to cockroachdb field
        mutate {
          copy => { "[cockroach_wrapper][event]" => "cockroachdb" }
        }

        # Extract client IP and port from tags if available
        # Structured logging format has tags like: "tags":{"client":"ip:port"}
        if [cockroach_wrapper][tags][client] {
          grok {
            match => {
              "[cockroach_wrapper][tags][client]" => "%{IP:client_ip}:%{NUMBER:client_port}"
            }
          }
        }

        # Add client IP and port to cockroachdb object
        if [client_ip] {
          mutate {
            add_field => {
              "[cockroachdb][ClientIP]" => "%{client_ip}"
              "[cockroachdb][ClientPort]" => "%{client_port}"
            }
          }
        }

        # Add ServerHost and ServerPort from rsyslog template prefix if present.
        # If not present, [host] is left intact so the Java filter can use the
        # TCP socket IP as a fallback for ServerHost.
        if [server_hostname] {
          mutate {
            add_field => {
              "[cockroachdb][ServerHost]" => "%{server_hostname}"
              "[cockroachdb][ServerPort]" => "%{server_port}"
            }
            remove_field => ["host"]
          }
        }

        # Convert nanosecond timestamp to readable format
        if [cockroachdb][Timestamp] {
          ruby {
            code => "
              timestamp_ns = event.get('[cockroachdb][Timestamp]')
              timestamp_sec = timestamp_ns / 1_000_000_000.0
              event.set('[cockroachdb][TimestampReadable]', Time.at(timestamp_sec).utc.strftime('%Y-%m-%d %H:%M:%S.%6N UTC'))
            "
          }
        }

        # Apply the Guardium filter (system queries already dropped early)
        cockroachdb_guardium_filter {}
      } else {
        # No event field found - tag as invalid
        mutate {
          add_tag => ["_no_event_field"]
        }
      }
    }

    # Clean up temporary fields
    mutate {
      remove_field => [
        "cockroach_wrapper_json",
        "cockroach_wrapper",
        "client_ip",
        "client_port",
        "server_hostname",
        "server_port",
        "message"
      ]
    }
  }
}