Export message metadata from rspamd metadata_exporter to logstash http input

curl

curl -X PUT -H 'Content-Type: application/json' -d @mail_log_policy.json http://localhost:9200/_ilm/policy/mail_log_policy
curl -X PUT -H 'Content-Type: application/json' -d @mail_log_template.json http://localhost:9200/_index_template/mail_log_template
curl -X PUT -H 'Content-Type: application/json' -d @mail_log_index.json http://localhost:9200/mail-000001

logstash.conf

input {
  http {
    host => "0.0.0.0"
    port => "1514"
  }
}

filter {
  json {
    source => "message"
    remove_field => ["event", "message"]
  }
  mutate {
    convert => ["[symbols][*][score]", "float"]
  }
}

output {
  elasticsearch {
    hosts => "es01.example.com"
    index => "mail_log_template"
    ilm_enabled => "true"
    ilm_pattern => "-000001"
    ilm_policy => "mail_log_policy"
    ilm_rollover_alias => "mail"
  }
}

mail_log_index.json

{
  "aliases": {
    "mail": {
      "is_write_index": true
    }
  }
}

mail_log_policy.json

{
  "policy": {
    "phases": {
      "hot": {
        "actions": {
          "rollover": {
            "max_size": "50gb",
            "max_age": "14d"
          },
          "set_priority": {
            "priority": 50
          }
        }
      },
      "warm": {
        "min_age": "6m",
        "actions": {
          "forcemerge": {
            "max_num_segments": 1
          },
          "shrink": {
            "number_of_shards": 1
          },
          "allocate": {
            "require": {
              "data": "warm"
            }
          },
          "set_priority": {
            "priority": 25
          }
        }
      },
      "cold": {
        "min_age": "12m",
        "actions": {
          "set_priority": {
            "priority": 0
          },
          "freeze": {},
          "allocate": {
            "require": {
              "data": "cold"
            }
          }
        }
      },
      "delete": {
        "min_age": "36m",
        "actions": {
          "delete": {}
        }
      }
    }
  }
}

mail_log_template.json

{
  "index_patterns": ["mail-*"],
  "template": {
    "settings": {
      "index": {
        "lifecycle": {
          "name": "mail_log_policy",
          "rollover_alias": "mail"
        },
        "routing": {
          "allocation": {
            "include": {
              "_tier_preference": "data_content"
            }
          }
        },
        "number_of_shards": "1",
        "number_of_replicas": "1"
      }
    },
    "mappings": {
      "dynamic_templates": [
        {
          "strings_as_keyword": {
            "match_mapping_type": "string",
            "mapping": {
              "ignore_above": 1024,
              "type": "keyword"
            }
          }
        }
      ],
      "properties": {
        "@timestamp": {
          "type": "date"
        },
        "@version": {
          "type": "long"
        },
        "action": {
          "type": "keyword"
        },
        "from": {
          "type": "keyword"
        },
        "fuzzy": {
          "type": "keyword"
        },
        "header_date": {
          "type": "keyword"
        },
        "header_from": {
          "type": "keyword"
        },
        "header_subject": {
          "type": "keyword"
        },
        "header_to": {
          "type": "keyword"
        },
        "host": {
          "properties": {
            "ip": {
              "type": "keyword"
            }
          }
        },
        "http": {
          "properties": {
            "method": {
              "type": "keyword"
            },
            "request": {
              "properties": {
                "body": {
                  "properties": {
                    "bytes": {
                      "type": "long"
                    }
                  }
                },
                "mime_type": {
                  "type": "keyword"
                }
              }
            },
            "version": {
              "type": "keyword"
            }
          }
        },
        "ip": {
          "type": "keyword"
        },
        "message_id": {
          "type": "keyword"
        },
        "qid": {
          "type": "keyword"
        },
        "rcpt": {
          "type": "keyword"
        },
        "score": {
          "type": "double",
          "ignore_malformed": false,
          "coerce": true
        },
        "subject": {
          "type": "keyword"
        },
        "symbols": {
          "type": "nested",
          "properties": {
            "group": {
              "type": "keyword"
            },
            "groups": {
              "type": "keyword"
            },
            "name": {
              "type": "keyword"
            },
            "options": {
              "type": "keyword"
            },
            "score": {
              "type": "double",
              "ignore_malformed": false,
              "coerce": true
            }
          }
        },
        "tags": {
          "type": "keyword"
        },
        "url": {
          "properties": {
            "domain": {
              "type": "keyword"
            },
            "path": {
              "type": "keyword"
            }
          }
        },
        "user": {
          "type": "keyword"
        },
        "user_agent": {
          "properties": {
            "original": {
              "type": "keyword"
            }
          }
        }
      }
    },
    "aliases": {}
  }
}

rspamd.conf.local

# rspamd.conf.local
metadata_exporter {
  rules {
    LOGSTASH {
      backend = "http";
      url = "http://your.logstash.tld:1514";
      formatter = "json";
      selector = "not_from_watchdog";
    }
  }

  custom_select {
    not_from_watchdog = <<EOD
return function(task)
  local mime_from = task:get_from{"mime", "orig"}[1].addr
  if mime_from then
    if mime_from ~= "watchdog@localhost" then
      return true
    else
      return false
    end
  end
end
EOD;
  }
}