Logstash

filter { if [etltype] == "blocks" { #[fields][srctype] csv { columns => [ "number", "hash", "parent_hash", "nonce", "sha3_uncles", "logs_bloom", "transactions_root", "state_root", "receipts_root", "miner", "difficulty", "total_difficulty", "size", "extra_data", "gas_limit", "gas_used", "timestamp", "transaction_count" ] separator => "," remove_field => ["message"] skip_empty_columns => true skip_empty_rows => true } }else if [etltype] == "contracts" { #[fields][srctype] csv { columns => [ "address", "bytecode", "function_sighashes", "is_erc20", "is_erc721" ] separator => "," remove_field => ["message"] skip_empty_columns => true skip_empty_rows => true } }else if [etltype] == "logs" { #[fields][srctype] csv { columns => [ "log_index", "transaction_hash", "transaction_index", "block_hash", "block_number", "address", "data", "topics" ] separator => "," remove_field => ["message"] skip_empty_columns => true skip_empty_rows => true } }else if [etltype] == "receipts" { #[fields][srctype] csv { columns => [ "transaction_hash", "transaction_index", "block_hash", "block_number", "cumulative_gas_used", "gas_used", "contract_address", "root", "status" ] separator => "," remove_field => ["message"] skip_empty_columns => true skip_empty_rows => true } }else if [etltype] == "token_transfers" { #[fields][srctype] csv { columns => [ "" ] separator => "," remove_field => ["message"] skip_empty_columns => true skip_empty_rows => true } }else if [etltype] == "tokens" { #[fields][srctype] csv { columns => [ "" ] separator => "," remove_field => ["message"] skip_empty_columns => true skip_empty_rows => true } }else if [etltype] == "transactions" { #[fields][srctype] csv { columns => [ "hash", "nonce", "block_hash", "block_number", "transaction_index", "from_address", "to_address", "value", "gas", "gas_price", "inputcontext" ] separator => "," remove_field => ["message"] skip_empty_columns => true skip_empty_rows => true } } } output { if [etltype] == "blocks" { elasticsearch { hosts => "xxx.xxx.xxx.xxx:9200" manage_template => false index => "%{[@metadata][beat]}-%{[@metadata][version]}-blocks-%{+YYYY.MM.dd}" document_id => "%{[hash]}" } }else if [etltype] == "logs" { elasticsearch { hosts => "xxx.xxx.xxx.xxx:9200" manage_template => false index => "%{[@metadata][beat]}-%{[@metadata][version]}-logs-%{+YYYY.MM.dd}" } }else if [etltype] == "transactions" { elasticsearch { hosts => "xxx.xxx.xxx.xxx:9200" manage_template => false index => "%{[@metadata][beat]}-%{[@metadata][version]}-transactions-%{+YYYY.MM.dd}" document_id => "%{[hash]}" } }else if [etltype] == "contracts" { elasticsearch { hosts => "xxx.xxx.xxx.xxx:9200" manage_template => false index => "%{[@metadata][beat]}-%{[@metadata][version]}-contracts-%{+YYYY.MM.dd}" document_id => "%{[address]}" } }else{ elasticsearch { hosts => "xxx.xxx.xxx.xxx:9200" manage_template => false index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}" } } stdout { codec => rubydebug } } transactions csv fileds are ...

Use filebeat nginx module send nginx log to logstash or driect to elastick all get error!! BUT some nginx log record can send success. That success records try to copy to other VM that have filebeat and logstash, try to send again. all get error!!

https://segmentfault.com/a/1190000002972420 通配符 ? 匹配单个字符 * 匹配0到多个字符 kiba?a, el*search ? * 不能用作第一个字符，例如：?text *text ==================== 正则 es支持部分正则功能,性能较差 name:/joh?n(ath[oa]n)/ ==================== 模糊搜索 quikc~ brwn~ foks~ :在一个单词后面加上启用模糊搜索，可以搜到一些拼写错误的单词 first~ 这种也能匹配到 frist 还可以设置编辑距离（整数），指定需要多少相似度 cromm~1 会匹配到 from 和 chrome 默认2，越大越接近搜索的原始值，设置为1基本能搜到80%拼写错误的单词 ==================== 逻辑操作 AND OR +：搜索结果中必须包含此项 -：不能含有此项 +apache -jakarta test aaa bbb：结果中必须存在apache，不能有jakarta，剩余部分尽量都匹配到 ==================== 分组 (jakarta OR apache) AND jakarta ==================== 转义特殊字符 + - = && || > < ! ( ) { } [ ] ^ " ~ * ? : \ / 以上字符当作值搜索的时候需要用\转义 \(1\+1\)\=2用来查询(1+1)=2 ...

all output columns with logstash filter { if [srctype] == "etl" { #[fields][srctype] csv { columns => [ "number", "hash", "parent_hash", "nonce", "sha3_uncles", "logs_bloom", "transactions_root", "state_root", "receipts_root", "timestamp", "extra_data", "transaction_count", "gas_limit", "size", "total_difficulty", "difficulty", "miner", "block_hash", "block_number", "transaction_index", "from_address", "to_address", "value", "gas", "gas_price", "input", "address", "bytecode", "function_sighashes", "is_erc20", "is_erc721", "log_index", "transaction_hash", "data", "topics", "cumulative_gas_used", "gas_used", "contract_address", "root,status" ] separator => "," remove_field => ["message"] #autodetect_column_names => true #have problems #autogenerate_column_names => true #have problems skip_empty_columns => true skip_empty_rows => true } }

https://sueboy.blogspot.com/2018/11/elk60filebeatdocumenttype.html filebeat.yml - type: log paths: - /var/log/geth.log exclude_files: ['.gz$'] fields: srctype: "geth" pipleline logstah.conf if [fields][srctype] == “geth” { BUT fields_under_root: true - type: log paths: - /var/log/geth.log exclude_files: ['.gz$'] fields: srctype: "geth" fields_under_root: true if [srctype] == “geth” {