Logstash:运用 Logstash 对 Service API 数据进行分析
2021年8月9日 | by mebius
我记得在之前的文章 “Logstash:使用 ELK 堆栈进行 API 分析” 运用 Logstash 对一些指标的 API 进行分析。在今天的练习中,我将展示如何使用 Logstash 来对一些日志类的 Service API 进行分析。我们知道在很多的时候,我们可以很快速地运用一些脚本对一些 Service API 进行数据分析。这对tgcode于我们快速地分析一些数据进行分析是有帮助的。在数据导入中,我们可以使用 Logstash 所提供的丰富的 filter 来对数据进行清洗,丰富,转换等。
在今天的练习中,我将使用 shodan.io 网站为例:
Shodan 是一个万物互联的搜索引擎。在上面,我们搜索 china 这个单词的时候,它会显示那些主机中有这个单词的显示。关于 Shodan,我这里就不做更多的介绍。Shodan 实际上也提供一种 Service API 的接口并供我们的客户端进行调用。我们可以在地址来申请一个开发者账号,并得到开发者秘钥。
在下面,我使tgcode用一个 Python 的应用程序来获取查询的结果:
shodan_scanner.py
import os
from shodan import Shodan
import time
import requests
import pprint
import sys
import logging
logging.basicConfig(level=logging.INFO,
filename='shodan.log',
format='{%(message)s}')
# api_key = os.environ[SHODAN_API_KEY]
api_key = "YOUR DEVELOPER KEY"
api = Shodan(api_key)
search = sys.argv[1]
# Search Shodan
results = api.search(search)
# print(f"Total count: {len(results['matches'])}")
# Show the results
for result in results['matches']:
ip_address = result['ip_str']
domains = result['domains']
logging.info('"ip_addr":"{}","domains":"{}"'.format(ip_address, domains))
time.sleep(1)
为了运行上面的应用,我们必须安装 shodan:
pip3 install shodan
我们可以通过如下的方式来运行:
python3 shodan_scanner.py "china" > shodan.log
在上面,我们搜索 china 这个单词,并把搜索的结果保存到 shodan.log 里。等我们运行完后,我们可以在 shodan.log 文件里查看到如下的这样的日志:
shodan.log
{"ip_addr":"46.171.98.231","domains":"['tpnet.pl']"}
{"ip_addr":"50.212.192.99","domains":"[]"}
{"ip_addr":"189.68.169.82","domains":"['telesp.net.br']"}
{"ip_addr":"157.25.97.179","domains":"[]"}
{"ip_addr":"116.106.51.100","domains":"['viettel.vn']"}
{"ip_addr":"147.161.68.134","domains":"['adamo.es']"}
{"ip_addr":"37.252.122.48","domains":"['tilaa.com']"}
{"ip_addr":"134.73.16.32","domains":"['smbccojp14.com']"}
{"ip_addr":"177.182.214.4","domains":"['virtua.com.br']"}
{"ip_addr":"115.72.1.76","domains":"['viettel.vn']"}
{"ip_addr":"118.70.44.0","domains":"[]"}
{"ip_addr":"113.165.201.60","domains":"['vnpt.vn']"}
{"ip_addr":"45.201.208.251","domains":"[]"}
{"ip_addr":"42.119.67.22","domains":"[]"}
{"ip_addr":"14.166.150.83","domains":"['vnpt.vn']"}
{"ip_addr":"181.46.231.207","domains":"['telecentro-reversos.com.ar']"}
{"ip_addr":"109.50.128.189","domains":"['netcabo.pt']"}
{"ip_addr":"81.193.147.140","domains":"['telepac.pt']"}
{"ip_addr":"41.38.136.142","domains":"['tedata.net']"}
{"ip_addr":"85.187.243.54","domains":"['villagenet-bg.com']"}
{"ip_addr":"103.217.236.75","domains":"[]"}
{"ip_addr":"115.79.68.162","domains":"['viettel.vn']"}
从上面,我们可以看到,它有一个叫做 ip_addr 这样的 IP 地址。我们可以使用 Logstash 所提供的 geoip 过滤器来丰富这个数据。
logstash.conf
input {
file {
path => [ "/Users/liuxg/python/shodan/shodan.log" ]
start_position => "beginning"
sincedb_path => "/dev/null"
codec => "json"
}
}
filter {
geoip {
source => "ip_addr"
target => "geo"
}
if [geo][latitude] and [geo][logitude] {
mutate {
add_field => {
"location" => ["%{[geo][latitude]},%{[geo][logitude]}"]
}
}
mutate {
convert => {
"location" => "float"
}
}
}
}
output {
stdout {
codec => rubydebug
}
elasticsearch {
hosts => ["http://localhost:9200"]
index => "shodan"
}
}
在上面,我们通过 file input 把数据导入。在使用时,你需要使用自己的文件路径来代替上面的 path。在上面,我使用了 geoip 这个过滤器来对数据进行丰富。
在运行上面的 Logstash pipeltgcodeine 之前,我们在 Kibana 中打入如下的命令:
PUT shodan
{
"mappings": {
"properties": {
"geo": {
"properties": {
"location": {
"type": "geo_point"
}
}
}
}
}
}
上面的命令定义了 geo.location 的数据类型。它是一个 geo_point 的数据类型。
我们可以使用如下的命令来运行 Logstash:
sudo ./bin/logstash -f logstash.conf
我们可以在 Logstash 运行的 terminal 中看到如下的输出:
从上面,我们可以看出来,通过 geoip 的使用,我们得到了更多的关于位置的字段。
我们需要为 shaodan 索引创建一个索引模式。我们在 Discover 中可以对这些数据进行查找:
由于它也有位置字段,我们也可以使用 Maps 应用来对文档的位置来进行显示:
文章来源于互联网:Logstash:运用 Logstash 对 Service API 数据进行分析
在实际的使用过程中,我们有时希望正对 RSS 的数据来进行导入并进行搜索。在现实的许多微服务中,有很多的数据是根据 RSS feed 的形式来提供的,比如我们常见的点评网站等。那么我们有没有办法把这些数据导入到 Elasticsearch 并进行搜索呢?答案是…