#!/bin/bash

max=2500000

#check permission
uid=$(id -u)
if [[ -z $uid ]];then
echo "please use root to run this"
exit 201
fi
if [[ $uid -ne 0 ]];then
echo "please use root to run this"
exit 201
fi

#check value
#linux default: 1024 * 1024 = 1048576
if [[ $max -lt 1048576 ]];then
echo "max value must >= 1048576"
exit 202
fi

#set temporary
sysctl -w fs.nr_open=$max
if [[ $? -ne 0 ]];then
echo "sysctl set failed"
exit 1
fi

#set forever
grep -P "^\s*fs.nr_open" /etc/sysctl.conf &>/dev/null
if [[ $? -eq 0 ]];then
#update
sed -i "s/^\s*fs.nr_open.*/fs.nr_open = $max /g" /etc/sysctl.conf
if [[ $? -ne 0 ]];then
echo "update value failed"
exit 2
fi
else
#add 
echo "fs.nr_open = $max" >> /etc/sysctl.conf
if [[ $? -ne 0 ]];then
echo "add value failed"
exit 3
fi
fi
sed -i "s/.*-.*nofile.*/\* - nofile $max/g" /etc/security/limits.conf

# allow user 'kafka' mlockall
kafka soft memlock unlimited
kafka hard memlock unlimited

node.name: 修改为本机hostname

bootstrap.mlockall: true 优化es内存设置，起来lock住内存

discovery.zen.ping.unicast.hosts:修改为集群所有的ip列表，利于发现其他节点

indices.recovery.concurrent_streams: 24
indices.recovery.concurrent_small_file_streams: 12
indices.recovery.max_bytes_per_sec: 200mb

indices.fielddata.cache.size: 20%
indices.breaker.fielddata.limit: 20%
indices.breaker.request.limit: 20%
indices.breaker.total.limit: 40%

export JAVA_HOME=/data/apps/jdk1.8.0_65
export ES_HEAP_SIZE=28g
cd /data/apps/elasticsearch-2.1.1/bin && ./elasticsearch -d --Xmx 28g --Xms 28g

broker.id= 每台kafka都有一个唯一id

sudo -u kafka nohup bin/kafka-server-start.sh config/server.properties &

bin/logstash -e 'input{file{ type=>".23.8" path=>"/data/logs/nginx/access.log"}}output{kafka{ bootstrap_servers => "xxx.68:10002" topic_id => "xxxslog" }}'

#type为标志本机ip

#path为日志路径

#output是kafka的地址列表，上表是武清机房的kafka地址

#topic_id是对应的topic

inputs: #input为kafka
- Kafka:
codec: json #json格式
topic: 
kxxx3log: 2

########################################################

从Kafka读数据. 下面示例是说, 开2个线程取app这个topic的数据, 编码格式为plain纯文本. consumer_settings中的参数, 参考[kafka]官方文档](http://kafka.apache.org/documentation.html#consumerconfigs), 所有参数都可以在这里配置. 比较重要的是group.id, zookeeper.connect这两个, 一定要有. zookeeper.connect的格式为 hostname1:port1,hostname2:port2,hostname3:port3.

########################################################
consumer_settings:
group.id: hangout
zookeeper.connect: 11.4.22.15:2181
auto.commit.interval.ms: "5000"
socket.receive.buffer.bytes: "1048576"
fetch.message.max.bytes: "1048576"
num.consumer.fetchers: "4"
filters:
- Grok:
src: message
match:
- '^\"(?<upstream_http_x_kss_Bucketowner>.+)\" \"(?<upstream_http_x_kss_Bucketname>.+)\" \"\[(?<time_local>.+)\]\" \"(?<remote_addr>\d{1,3}.\d{1,3}.\d{1,3}.\d{1,3})\" \"(?<upstrea
m_http_x_kss_Requester>.+)\" \"(?<upstream_http_x_kss_request_id>.+)\" \"(?<upstream_http_x_kss_Op>.+)\" \"(?<upstream_http_x_kss_Key>.+)\" \"(?<request>.+)\" \"(?<status>.+)\" \"(?<upstrea
m_http_x_kss_ErrCode>.+)\" \"(?<content_length>.+)\" \"(?<body_bytes_sent>.+)\" \"(?<upstream_http_x_kss_ObjSize>.+)\" \"(?<request_time>.+)\" \"(?<upstream_response_time>.+)\" \"(?<http_re
ferer>.+)\" \"(?<http_user_agent>.+)\" \"(?<upstream_addr>.+)\" \"(?<http_kscdn>.+)\"$'
remove_fields: ['path']
- Json:
field: message # required
- Gsub:
fields: #由于以下字段会有-，所以要替换为0,免得出错
upstream_http_x_kss_ObjSize: ['-','0']
body_bytes_sent: ['-','0']
content_length: ['-','0']
upstream_response_time: ['-','0.00']
request_time: ['-','0.00']
outputs: #向es输出
- Elasticsearch:
cluster: xxx-ecs #es的集群名
hosts: #杭州es host
- 10.4.69.70:9300
- 10.4.71.100:9300
- 10.4.71.103:9300
- 10.4.71.102:9300
index: 'hangout-ks3-%{+YYYYMMdd}' #日志格式
index_type: logs # default logs
bulk_actions: 20000 #default 20000
bulk_size: 30 # default 15 MB
flush_interval: 10 # default 10 seconds
concurrent_requests: 0 # default 0, concurrent_requests设置成大于0的数, 意思着多线程处理, 以我应用的经验,还有是一定OOM风险的,强烈建议设置为0
timezone: "Asia/Shanghai" # defaut UTC 时区. 只用于生成索引名字的字符串格式化