## 1.下载
 wget https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-8.10.2-x86_64.rpm
 rpm -ivh filebeat-7.17.5-x86_64.rpm
 
## 2.cat /etc/ilebeat/filebeat.yml
filebeat.inputs:
filebeat.config.inputs:
  path: ${path.config}/child/*.yml
  enabled: true
  reload.enabled: true
  reload.period: 60s
filebeat.config.modules:
  path: ${path.config}/modules.d/*.yml
  reload.enabled: true
  reload.period: 5s
#==================== Elasticsearch template setting ==========================
setup.template.settings:
  index.number_of_shards: 3
#-----------------------kafka output ----------------------------------------
output.kafka:
  hosts: ["kafka1:9092","kafka2:9092","kafka3:9092"]
  topic: '%{[fields.modletype]}'
  username: 'admin'
  password: '123456'
  sasl.mechanism: 'SCRAM-SHA-256'
  partition.round_robin:
    reachable_only: false
  required_acks: 1
  compression: gzip
  max_message_bytes: 1000000
  keep_alive: 10s 
  
## 3.cat /etc/ilebeat/child/test-nginx.yml 
- type: log 
  paths:
    - /data/nginx/logs/*access*.log
  exclude_lines: ['^DBG']   #不包含匹配该格式的日志
  #clean_inactive: 24h
  #ignore_older: 20h
  close_inactive: 1m
  close_timeout: 2h
  fields:
    ip:  36.134.227.197
    modletype: test-nginx
    datacenter: gy
  #  level: debug
  #  review: 1
  multiline.pattern: ^\{ 
  multiline.negate: true   #false，匹配pattern的行合并到上一行；true，不匹配pattern的行合并到上一行
  multiline.match: after  #合并到上一行的末尾或开头
  processors:
  - decode_json_fields:
      fields: ["message"]
      process_array: false
      max_depth: 3
      target: ""
      overwrite_keys: false
      
## 启动服务
sudo systemctl daemon-reload
sudo systemctl enable filebeat  # 设置开机自启
sudo systemctl start filebeat   # 启动服务
journalctl -u filebeat -f  # 查看日志

apiVersion: v1
kind: ConfigMap
metadata:
  name: logstash-config
  namespace: elk
data:
  logstash.conf: |-
    input{
      kafka{
        bootstrap_servers => "kafka1:9093, kafka29093, kafka3:9093"
        group_id => "elk-logstash-consumer"
        security_protocol => "SASL_PLAINTEXT"
        sasl_mechanism => "SCRAM-SHA-256"
        sasl_jaas_config => "org.apache.kafka.common.security.scram.ScramLoginModule required username='admin' password='admin-secret';" 
        type => "test-nginx"
        topics => "test-nginx"
        consumer_threads => 6
        decorate_events => true
        #enable_auto_commit => false
        codec => "json"
        auto_offset_reset => "latest"
        heartbeat_interval_ms => "30000"
        session_timeout_ms => "60000"
        fetch_max_bytes => "1048576"
        fetch_max_wait_ms => "10000"
        receive_buffer_bytes => "2097152"
      }
    }
    filter {
      truncate {
        fields => [ "message" ]
        length_bytes => 32764
      }
      if ( "Exception" in [message] and "ERROR" in [message] ){
        mutate {
            add_field => [ "show","exception" ]
        }
      }
    # ["1-半日","2-日","3-周","4-月"]
      ruby {
        code => "event.set('index_day', (event.get('@timestamp').time.localtime + 28800).strftime('%Y.%m.%Ww'))"
      }
    }

    output {
      elasticsearch {
        action => "index"
        hosts => ["es-ip:9092"]
        #workers => 5
        password => "123456"
        user => "elastic"
        index => "test-nginx"
        http_compression => true
        timeout => 280
      }
    }
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: logstash
  namespace: elk
  labels: 
    name: logstash
spec:
  replicas: 1
  selector:
    matchLabels: 
      name: logstash
  template:
    metadata:
      labels: 
        app: logstash
        name: logstash
    spec:
      containers:
      - name: logstash
        image: crpi-39k23aqntb3h78w8.cn-hangzhou.personal.cr.aliyuncs.com/wdsheng/logstash:7.12.0
        ports:
        - containerPort: 5044
          protocol: TCP
        - containerPort: 9600
          protocol: TCP
        volumeMounts:
        - name: logstash-config
          mountPath: /usr/share/logstash/pipeline/logstash.conf
          subPath: logstash.conf
      volumes:
      - name: logstash-config
        configMap:
          name: logstash-config

---
apiVersion: v1
kind: Service
metadata:
  namespace: elk
  name: logstash
  labels:
    app: logstash
spec:
  type: ClusterIP
  ports:
  - port: 5044
    name: logstash
  selector: 
    app: logstash

apiVersion: apps/v1
kind: Deployment
metadata:
  generation: 1
  labels:
    app: elasticsearch-logging
    version: v1
  name: elasticsearch
  namespace: elk
spec:
  minReadySeconds: 10
  progressDeadlineSeconds: 600
  replicas: 1
  revisionHistoryLimit: 10
  selector:
    matchLabels:
      app: elasticsearch-logging
      version: v1
  strategy:
    type: Recreate
  template:
    metadata:
      creationTimestamp: null
      labels:
        app: elasticsearch-logging
        version: v1
    spec:
      affinity:
        nodeAffinity: 
      containers:
      - env:
        - name: discovery.type
          value: single-node
        - name: ES_JAVA_OPTS
          value: -Xms512m -Xmx512m
        - name: MINIMUM_MASTER_NODES
          value: "1"
        image: docker.elastic.co/elasticsearch/elasticsearch:7.12.0-amd64
        imagePullPolicy: IfNotPresent
        name: elasticsearch-logging
        ports:
        - containerPort: 9200
          name: db
          protocol: TCP
        - containerPort: 9300
          name: transport
          protocol: TCP
        resources:
          limits:
            cpu: "1"
            memory: 1Gi
          requests:
            cpu: "1"
            memory: 1Gi
        terminationMessagePath: /dev/termination-log
        terminationMessagePolicy: File
        volumeMounts:
        - mountPath: /data
          name: es-persistent-storage
      dnsPolicy: ClusterFirst
      imagePullSecrets:
      - name: user-1-registrysecret
      initContainers:
      - command:
        - /sbin/sysctl
        - -w
        - vm.max_map_count=262144
        image: alpine:3.6
        imagePullPolicy: IfNotPresent
        name: elasticsearch-logging-init
        resources: 
        securityContext:
          privileged: true
          procMount: Default
        terminationMessagePath: /dev/termination-log
        terminationMessagePolicy: File
      restartPolicy: Always
      schedulerName: default-scheduler
      securityContext: 
      terminationGracePeriodSeconds: 30
      volumes:
      - hostPath:
          path: /opt/paas/hanju/es_data
          type: ""
        name: es-persistent-storage
      - name: config
        configMap:
          name: elasticsearch-config
---
apiVersion: v1
kind: Service
metadata:
  namespace: elk
  name: elasticsearch
  labels:
    app: elasticsearch-logging
spec:
  type: ClusterIP
  ports:
  - port: 9200
    name: elasticsearch
  selector: 
    app: elasticsearch-logging

---
kind: ConfigMap
apiVersion: v1
metadata:
  name: elasticsearch-config
  namespace: elk
  labels:
   app: elasticsearch
   role: master
data:
  elasticsearch.yml: |-
   action.auto_create_index: true

---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: kibana
  namespace: elk
  labels:
    name: kibana
spec:
  replicas: 1
  selector:
    matchLabels:
      name: kibana
  template:
    metadata:
      labels:
        app: kibana
        name: kibana
    spec:
      containers:
      - name: kibana
        image: docker.elastic.co/kibana/kibana:7.12.0
        ports:
          - containerPort: 5601
            protocol: TCP
        env:
          - name: ELASTICSEARCH_URL
            value: http://elasticsearch:9200
        volumeMounts:
          - name: config
            mountPath: /usr/share/kibana/config/kibana.yml
            readOnly: true
            subPath: kibana.yml
      volumes:
      - name: config
        configMap:
          name: kibana-config

---
apiVersion: v1
kind: ConfigMap
metadata:
  namespace: elk
  name: kibana-config
data:
  kibana.yml: |-
    server:
      name: kibana
      host: "0"
    elasticsearch:
      hosts: [ "http://elasticsearch:9200" ] 
    i18n:
      locale: "zh-CN"
---
apiVersion: v1
kind: Service
metadata:
  name: kibana
  namespace: elk
spec:
  type: NodePort
  ports:
  - protocol: TCP
    port: 80
    targetPort: 5601
    nodePort: 31100
  selector:
    app: kibana

#查看集群健康状态
curl -u elastic:123456  -X GET "192.168.0.1:9200/_cluster/health?pretty"

#查看集群配置
curl -u elastic:123456  -X GET "192.168.0.1:9200/_cluster/settings?pretty"

#查看集群nodes
curl -u elastic:123456  -X GET "192.168.0.1:9200/_cat/nodes?pretty"

#查看集群集群分片
curl -u elastic:123456 -X GET 192.168.0.1:9200/_cluster/settings?pretty

# 修改集群分片上上限，默认1000
curl -u elastic:123456 -X PUT 192.168.0.1:9200/_cluster/settings -H "Content-Type: application/json" -d '{ "persistent": { "cluster.max_shards_per_node": "2000" } }'

# 下载logstash
根据es集群的版本，选择相应版本的logstash
这里的es集群版本为7.1.1，选择了7.11.2版本的logstash 
wget https://artifacts.elastic.co/downloads/logstash/logstash-7.11.2-linux-x86_64.tar.gz
tar -zxvf logstash-7.11.2-linux-x86_64.tar.gz

# 插件安装 
./logstash-plugin list | grep jdbc
./logstash-plugin install logstash-output-jdbc
# mysql-connector-java-5.1.40.jar可以找研发
mv mysql-connector-java-5.1.40.jar /etc/logstash/conf.d/

# 编辑导数据的配置文件
input中配置es集群的信息，query字段配置的查询参数，需要提前使用curl测试查询
output中配置mysql数据库的信息，注意connection_string和statement中库和表的参数配置
output中statement的表字段，值需要一一对应 

vim convert_sql_from_es.conf
 
 
input {
    elasticsearch {
        hosts => "192.168.0.2:9200"
        index => "test-operation-2024-04"
        user => "elastic"
        password => "123456"
        query => '{"query":{"match_all":{}}}'
        #query => '{"query": { "range": { "@timestamp": { "gte": "now-1y/d", "lt": "now" } } }}'
        }
}
output {
    jdbc {
        connection_string => "jdbc:mysql://192.168.0.1:3306/test?characterEncoding=utf8&useSSL=false&serverTimezone=UTC&rewriteBatchedStatements=true"
        username => "root"
        password => "123456"
        driver_jar_path => "/etc/logstash/conf.d/mysql-connector-java-5.1.40.jar"
        driver_class => "com.mysql.jdbc.Driver"
        statement => ["INSERT INTO tezt_operationlog (phone, operationContent, actionId, userId, orgId, serialNo, timestamp, userName, mediaName) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", "[phone]", "[operationContent]", "[actionId]", "[userId]", "[orgId]", "[serialNo]", "[timestamp]", "[userName]", "[mediaName]"]
    }
}

# 导入 
/data/logstash-7.11.2/bin/logstash -f /data/logstash-7.11.2/config/convert_sql_from_es.conf