echo "vm.swappiness=0" >> /etc/sysctl.conf
echo "vm.max_map_count=655350" >> /etc/sysctl.conf
sysctl -p

cat /etc/security/limits.conf|grep -v "^#"
* soft    nofile  1024000
* hard    nofile  1024000
* soft    nproc   unlimited
* hard    nproc   unlimited
* soft    core    unlimited
* hard    core    unlimited
* soft    memlock unlimited
* hard    memlock unlimited

[root@node7_2 ~]# ulimit -SHn 1024000

path.data: /var/lib/elasticsearch
path.logs: /var/log/elasticsearch
http.port: 9200
http.host: 0.0.0.0
xpack.security.enabled: true
xpack.security.transport.ssl.enabled: false
xpack.security.http.ssl.enabled: false

#先启动服务
systemctl start elasticsearch

#然后创建用户，其中 $(echo y | /usr/share/elasticsearch/bin/elasticsearch-reset-password -u elastic echo y | ./elasticsearch-reset-password -u elastic --url http://127.0.0.1:9200 | grep "New value" | awk '{print $NF}') 为elastic超级用户重置的密码
curl -X POST -u elastic:$(echo y | /usr/share/elasticsearch/bin/elasticsearch-reset-password -u elastic --url http://127.0.0.1:9200 | grep "New value" | awk '{print $NF}') "http://127.0.0.1:9200/_security/user/用户名" -H 'Content-Type: application/json' -d'{"password":"用户密码","roles":["superuser","kibana_system"]}'

path.data: /data/logstash    			#数据存储路径
path.config: /etc/logstash/conf.d/*.conf 	#配置文件目录

-Duser.timezone=Asia/Shanghai

input {
    tcp {
        port => 9601
        codec => "json"
    }
}
output {
    elasticsearch {
        hosts => ["127.0.0.1:9200"]
        user => "用户名"
        password => "用户密码"
        index => "logstash-%{+YYYY.MM.dd}"
    }
}

server.port: 5601
server.host: "0.0.0.0"
pid.file: /run/kibana/kibana.pid
path.data: /run/kibana
i18n.locale: "zh-CN"
elasticsearch.hosts: ["http://127.0.0.1:9200"]
elasticsearch.username: "用户名"
elasticsearch.password: "用户密码"

#logstash.conf配置文件，内容如下：
input {
#    stdin {}
    kafka {
        bootstrap_servers  => "192.168.8.100:9092"
        topics => ["logstash"]
        type => "test"
    }
}

filter {
    if [type] == "test" {
        json {
            source => "message"
            target => "jsoncontent"
        }
        mutate {
            split => ["jsoncontent",","]
            update => {"message" => "%{[jsoncontent][message]}"}
            remove_field => ["@timestamp","@version","jsoncontent"]
        }
    }
}

output {
    if [type] == "test" {
#        stdout {codec => rubydebug}
        elasticsearch {
            hosts => "http://172.17.0.2:9200"
            manage_template => false
            index => "test-%{+yyyy.MM}"
        }
    }
}

#检查配置文件语法是否正确
/usr/share/logstash/bin/logstash -f logstash.conf --config.test_and_exit

#--config.reload.automatic可以在Logstash不重启的情况下自动加载配置文件
/usr/share/logstash/bin/logstash -f conf.d/logstash.conf --config.reload.automatic

#---------------------------------------------------------------------------------------------

#配置文件filebeat.yml里面的内容：
filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /tmp/logs/*.log

output.kafka:
  enabled: true
  hosts: ["192.168.8.100:9092"]
  topic: logstash

#启动（最好先启动logstash不然没有消费者会崩）

curl -s "http://127.0.0.1:9200/_cat/nodes?v"        #查看集群的节点
curl -s "http://127.0.0.1:9200/_cat/health?v"       #查看集群健康状态
curl -s "http://127.0.0.1:9200/_cat/indices?v"      #查看索引
curl -s -X DELETE "http://127.0.0.1:9200/索引名"     #删除索引
curl -s "http://127.0.0.1:9200/索引名/_mapping"      #查看索引数据结构
curl -s "http://127.0.0.1:9200/_cat/allocation?v"   #查看集群所在磁盘的分配状况
curl -s "http://127.0.0.1:9200/_cat/shards?v"       #查看集群中所有索引的分片信息

#查询索引指定时间范围内含特定字符串的数据
curl -X GET "http://127.0.0.1:9200/索引名/_search" -H 'Content-Type: application/json' -d'
{
  "query": {
    "bool": {
      "must": [
        {
          "match_phrase": {
            "message": "要查询的字符串"
          }
        },
        {
          "range": {
            "@timestamp": {
              "gte": "2023-10-09 10:50:00",
              "lte": "2023-10-09 11:00:00",
              "format": "yyyy-MM-dd HH:mm:ss"
            }
          }
        }
      ]
    }
  }
}'

# 往logstash写入数据的方式：
# 1、input插件名为tcp时，curl和nc两种方式都可以写入，但curl方式会提示超时，需要设置超时时间，--max-time最长等待时间(秒)
# 2、input插件名为http是，只能curl方式
curl -X POST -H "Content-Type: application/json" -d '{"message": "中文测试777"}' --max-time 3 "http://127.0.0.1:9601/input插件名"
或者
echo '{"message": "中文测试222"}' | nc localhost 9601

bin/logstash-plugin prepare-offline-pack logstash-input-jdbc
bin/logstash-plugin prepare-offline-pack logstash-input-*
bin/logstash-plugin prepare-offline-pack logstash-output-* logstash-input-jdbc

bin/logstash-plugin install file:///path/to/logstash-offline-input-5.5.1.zip