转载别人的:https://www.jianshu.com/p/f4af986ca164
注：基于别人脚本上，进行针对修改。所以若直接使用清理脚本，请认真看清你的脚本是否是以下形式。如果不是，则脚本需要做适应性调整。
1.编写清理脚本

仅支持这种形式的索引清理
green  open   k8s-stdout-2018.11.01                     PHEIz5NXSw-ljRvOP1cj9Q   5   1    3748246            0      3.2gb          1.6gb
green  open   recom-nginxacclog-2018.10.27              jl5ZBPHsQBifN0-pXEp_yw   5   1   52743481            0     20.8gb         10.4gb
green  open   nginx-json-acclog-2018.11.09              r6jsGcHWRV6RP2mY6zFxDA   5   1      95681            0     50.4mb         25.1mb
green  open   watcher_alarms-2018.11.09                 r_GS_GQGRoCgyOgiCFCuQw   5   1         24            0    323.4kb        161.7kb
green  open   .monitoring-es-6-2018.11.12               o8H2S-iERnuIAWQT7wuBRA   1   1       4842            0      3.2mb          1.6mb
green  open   client-nginxacclog-2018.11.02             FxXdLPpiSnuBtrJOB2BRsw   5   1  179046160            0    220.8gb        110.3gb
green  open   k8s-stderr-2018.11.16                     Vggw7iYCQ6OHtW-sphgGrA   5   1      68546            0     20.4mb         10.2mb
green  open   k8s-stderr-2018.10.21                     ZCeZZFRWSVyyYDKMjzPnbw   5   1      15454            0      5.3mb          2.6mb
green  open   watcher_alarms-2018.10.30                 VqrbMbnuQ3ChPysgnwgm2w   5   1         44            0      371kb        185.5kb

#！/bin/bash
######################################################
# $Name:        clean_es_index.sh
# $Version:     v1.0
# $Function:    clean es log index
# $Author:      sjt
# $Create Date: 2018-05-14
# $Description: shell
######################################################
#本文未加文件锁，需要的可以加
#脚本的日志文件路径
CLEAN_LOG="/root/clean_es_index.log"
#索引前缀
INDEX_PRFIX=$1
DELTIME=$2
if [ "$DELTIME"x = ""x ]; then
   DELTIME=30
fi

echo "ready to delete index $DELTIME ago!!!"

#elasticsearch 的主机ip及端口
SERVER_PORT=10.2.1.2:9200
#取出已有的索引信息
if [ "$INDEX_PRFIX"x = ""x ]; then
     echo  "curl -s \"${SERVER_PORT}/_cat/indices?v\""
     INDEXS=$(curl -s "${SERVER_PORT}/_cat/indices?v" |awk '{print $3}')
else
     echo " curl -s \"${SERVER_PORT}/_cat/indices?v\""
     INDEXS=$(curl -s "${SERVER_PORT}/_cat/indices?v" |grep "${INDEX_PRFIX}"|awk '{print $3}')
fi
#删除多少天以前的日志，假设输入10，意味着10天前的日志都将会被删除
# seconds since 1970-01-01 00:00:00 seconds
SECONDS=$(date -d  "$(date  +%F) -${DELTIME} days" +%s)
#判断日志文件是否存在，不存在需要创建。
if [ ! -f  "${CLEAN_LOG}" ]
then
touch "${CLEAN_LOG}"
fi
#删除指定日期索引
echo "----------------------------clean time is $(date +%Y-%m-%d_%H:%M:%S) ------------------------------" >>${CLEAN_LOG}
for del_index in ${INDEXS}
do
    indexDate=$( echo ${del_index} | awk -F '-' '{print $NF}' )
    # echo "${del_index}"
    format_date=$(echo ${indexDate}| sed 's/\.//g')
    #根据索引的名称的长度进行切割，不同长度的索引在这里需要进行对应的修改
    indexSecond=$( date -d ${format_date} +%s )
    #echo "$SECONDS - $indexSecond = $(( $SECONDS - $indexSecond ))"
    if [ "$SECONDS" -gt "$indexSecond" ]; then
        echo "it will delete ${del_index}......."
        echo "${del_index}" >> ${CLEAN_LOG}
        #取出删除索引的返回结果
        delResult=`curl -s  -XDELETE "${SERVER_PORT}/"${del_index}"?pretty" |sed -n '2p'`
        #写入日志
        echo "clean time is $(date)" >>${CLEAN_LOG}
        echo "delResult is ${delResult}" >>${CLEAN_LOG}
    fi
done

2.加入crontab任务

# 进入crontab编辑模式
crontab -e 
# 在crontab编辑模式中输入
10 1 * * * sh /app/elk/es/es-index-clear.sh > /dev/null 2>&1

log.retention.hours=48 #数据最多保存48小时，默认是7天
log.retention.bytes=1073741824 #数据最多1G，默认是不限制大小。

在logstash.conf配置中，可以使用ruby动态修改某个字段数据。

filter {
    if [type] == "deployment" {
        drop {}
    }

    mutate {
        remove_field => ["kafka"]
    }

    ruby {
        code => "
        timestamp = event.get('@timestamp') #从字段中获取@timestamp字段，
        localtime = timestamp.time + 28800 #加上8个小时偏差
        localtimeStr = localtime.strftime('%Y.%m.%d')
        event.set('localtime', localtimeStr) #保存最新时间
        "
    }
}

filter {
    grok {
        #match => {"message"=>'(?\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}) - (?(\s+)|-) (?.*) "(?.*?) (?.*?)\?d=(?.*?) (?\S+)" (?\d+) (?\d+) "(?.*?)" "(?.*?)" "(?.*?)'}
        #match => {"message"=>'(?\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}) - (?(\s+)|-) (?.*) "(?.*?) (?.*?)\?d=(?.*?) (?\S+)" (?\d+) (?\d+) "(?.*?)" "(?.*?)".*?'}
        match => {"message"=>'(?\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}) - (?(\s+)|-) \[(?.*)\] "(?.*?) (?.*?)\?(d=)?(?.*?) (?\S+)" (?\d+) (?\d+) "(?.*?)" "(?.*?)".*?'}
    }

    if [tags]{
        drop {}
    }

    if [status] != "200" {
        drop {}
    }

    ruby {
        init => "require 'base64'"
        code => "
        string = event.get('string')
        if string
            begin
                b64 = Base64.decode64(string).force_encoding('utf-8')
                #puts b64, event.get('message')
                event.set('b64_decode', b64)
            rescue ArgumentError
                event.set('b64_decode', '')
            end
        else
            event.set('b64_decode', '')
        end
        "
    }

    if [b64_decode == ""]{
        drop {}
    }

    kv {
        source => "b64_decode"
        field_split => "&?"
        value_split => "="
    }

    if [type] == "template" {
        mutate {
            remove_field => ["@timestamp", "@version", "b64_decode", "message", "string", "body_bytes_sent", "timelocal", "http_user_agent", "http_referer", "status", "protocol", "uri", "ver", "remote_user", "remote_addr", "host", "method", "path"]
        }
    } else {
        date {
            match => ["timelocal", "dd/MMM/yyyy:HH:mm:ss Z"]
            target => "@timestamp"
        }

        mutate {
            remove_field => ["@version", "b64_decode", "message", "string", "body_bytes_sent", "timelocal", "http_user_agent", "http_referer", "status", "protocol", "uri", "ver", "remote_user", "remote_addr", "host", "method", "path"]
        }
    }

}

1.配置nginx的日志格式

log_format appstore escape=json '{"@timestamp":"$time_iso8601",'
                        '"@source":"$server_addr",'
                        '"hostname":"$hostname",'
                        '"xforward":"$http_x_forwarded_for",'
                        '"remoteaddr":"$remote_addr",'
                        '"method":"$request_method",'
                        '"scheme":"$scheme",'
                        '"domain":"$server_name",'
                        '"referer":"$http_referer",'
                        '"url":"$request_uri",'
                        '"args":"$args",'
                        '"requestbody":"$request_body",'  #根据实际情况开放，否则会导致内容太大。
                        '"bodybytessend":$body_bytes_sent,'
                        '"status":$status,'
                        '"requesttime":$request_time,'
                        '"upstreamtime":"$upstream_response_time",'  #需要加引号，某些upstream放弃处理时，其时间会为空的。
                        '"upstreamaddr":"$upstream_addr",'
                        '"useragent":"$http_user_agent"'
                        '}';

当输出的不是JSON格式时，因为logstash的透传关系，仍会把内容保存在ES中，并增加了[tag]failure的标签，如下所示：

2.使用方式

location /api/app/ {
        access_log /data/log/nginx/appstore.app.abc.cn.access.log abc buffer=32k flush=5s;
        error_log /data/log/nginx/appstore.app.abc.cn.error.log;
        proxy_pass http://appstore.app.abc.cn:8001/;
    }

3.配置filebeat的yml配置
vim /etc/filebeat/filebeat.yml

filebeat.prospectors:
- input_type: log
  paths:
    - /data/log/nginx/appstore.app.wps.cn.access.log
  document_type: simplejson
  fields_under_root: true
  fields:
    es_index_type: nginx-access
    es_index_prefix: appstore-app-wps-cn
  tail_files: true

#================================ Processors ===================================
processors:
- drop_event:
    when:
       contains:
           message: "HEAD /lb_health.php"


#----------------------------- output ----------------------------------
output.kafka:
  hosts: ["10.0.0.33:9092","10.0.0.59:9092","10.0.0.67:9092"]
  topic: "simple_json_for_all"
  partition.round_robin:
    reachable_only: false

  required_acks: 0
  compression: none
  max_message_bytes: 1000000


#----------------------------- file output for debug ----------------------------------
output.file:
  # true: will output to  file, false: disable output to file.
  enabled: false
  path: "/data/log/filebeat"
  filename: debug.log

#================================ Logging =====================================

logging.level: info

3.logstash的logstash.conf的nginx-access的配置。

input {
    kafka {
        bootstrap_servers => "10.0.0.33:9092,10.0.0.59:9092,10.0.0.67:9092"
        topics => ["simple_json_for_all"]
        group_id => "simple_json_consumer"
        consumer_threads => 5
        codec => 'json'
        decorate_events => true
    }
}

filter {
   if [type] == "simplejson" {
       json {
           source => "message"
           remove_field => ["message"]
           remove_field => ["kafka"]
           remove_field => ["beat"]
       }
   }
}

output {
    #stdout{codec=>rubydebug}
    if [type] == "simplejson" {
        if [es_index_prefix] and [es_index_type] {
            elasticsearch {
                hosts => ["10.2.1.2:9200", "10.2.1.11:9200", "10.2.1.15:9200"]
                # es_index_prefix & es_index_type were defined in filebeat.yml
                index => "%{es_index_prefix}-%{es_index_type}-%{+YYYY.MM.dd}"
                manage_template => true
            }
        } else {
            elasticsearch {
                hosts => ["10.2.1.2:9200", "10.2.1.11:9200", "10.2.1.15:9200"]
                index => "default-simplejson-%{+YYYY.MM.dd}"
                manage_template => true
            }
        }
    }
}

5.检验索引是否存在
curl -XGET ‘http://10.2.1.2:9200/_cat/indices?v’|grep nginx-access