1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

# 下载软件包
PIKA_VERSION="$(wget -qO- https://api.github.com/repos/Qihoo360/pika/releases/latest | grep 'tag_name' | cut -d\" -f4)"
wget -q https://github.com/Qihoo360/pika/releases/download/${PIKA_VERSION}/pika-linux-x86_64-${PIKA_VERSION}.tar.bz2

# 解压软件包
tar xjf pika-linux-x86_64-${PIKA_VERSION}.tar.bz2

# 建立程序目录
mkdir -p /usr/local/pika

# 拷贝 pika 应用程序到程序目录
\cp -arf output/* /usr/local/pika

# 清理安装文件
rm -rf output pika-linux-x86_64-${PIKA_VERSION}.tar.bz2*

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

# 建立数据目录
mkdir -p /data/pika/db

# 建立同步目录
mkdir -p /data/pika/dbsync

# 建立备份目录
mkdir -p /data/pika/dump

# 建立日志目录
mkdir -p /data/pika/log

# 建立配置文件
touch /data/pika/pika.conf

# 设置权限
chown -R worker.worker /data/pika

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130

cat > /data/pika/pika.conf <<EOF
# Pika 端口
port : 9231

# Pika 进程数量，不建议超过核心数量，Pika 是多线程的
thread-num : 6

# Sync 线程数量
sync-thread-num : 6

# Sync 处理线程的任务队列大小，一般没有必要修改
sync-buffer-size : 10

# Pika 日志目录
log-path : /data/pika/log/

# Pika 的 log 级别（INFO|ERROR），任何一个级别均记录慢日志
loglevel : ERROR

# Pika 数据目录
db-path : /data/pika/db

# Pika 底层引擎的 write_buffer_size 配置
# 大会快，但越大刷盘越久，需要权衡
write-buffer-size : 268435456

# Pika 的连接超时时间，就是连接 sleep 多久了就把它断开
timeout : 3600

# 密码管理员密码，默认为空
requirepass :

# 主从同步密码，默认为空
masterauth :

# 用户密码，默认为空
userpass :

# 指令黑名单，普通用户将不能使用黑名单中的指令
# 指令之间使用 “,” 隔开，默认为空
userblacklist :

# Pika 的 dump 文件名称前缀
dump-prefix :

# 守护进程模式（yes|no）
daemonize : no

# 支持 migrate slot，作为 codis 节点时需开启（yes|no）
slotmigrate : no

# Pika 的 dump 目录
dump-path : /data/pika/dump/

# Pika 的 dump 文件过期时间，默认为 0，不过期
dump-expire : 0

# Path 的 pid 文件路径
pidfile : /data/pika/pika.pid

# 最大连接数
maxclients : 8192

# sstable 文件大小，默认是 20M
# 文件越小，速度越快，合并代价越低，但文件数量就会非常多
# 文件越大，速度相对变慢，合并代价大，但文件数量会很少
target-file-size-base : 20971520

# binlog 文件保留时间，最小为 1
expire-logs-days : 7

# binlog 文件最大数量，最小为 10
# 超过设定数量就开始自动清理，始终最多保留设定数量个文件
expire-logs-nums : 100

# root 用户连接保证数量
# Pika 达到最大连接数时，也能确保本地有指定个连接可以登入 Pika
root-connection-num : 2

# 慢日志记录时间，单位为微秒
slowlog-log-slower-than : 10000

# 作为从服务器时，是否为只读状态（yes|no，1|0）
slave-read-only : 0

# Pika 数据同步目录
db-sync-path : /data/pika/dbsync/

# 主从数据同步传输速度上限
# 最大值 125MB，最小值 0，如果设置的值超出范围会被校准为最大值
db-sync-speed : 60

# 指定监听的网卡
# network-interface : eth0

# 启用主从复制，并作为从服务器从指定服务器同步数据
# slaveof : master-ip:master-port

# 定时任务，格式 start-end/ratio
# 例如 02-04/60，Pika 将在每天 2 ～ 4 点检查磁盘使用量，如果超过 60% 则开始数据整理
# 当 compact-interval 被设置时，compact-cron 将会被自动禁用
# compact-cron : 20-24/60

# 整理间隔，格式 interval/ratio
# 例如 6/60，Pika 将每隔 6 小时检查磁盘使用量，如果超过 60% 则开始数据整理
# compact-interval 的优先级高于 compact-cron
# compact-interval :

###################
##    关键设置    ##
###################
# binlog 文件大小，默认为 100MB，启动后不可修改，范围为 [1K, 2G]
binlog-file-size : 104857600

# 压缩方式，默认为 snappy，启动后不可修改（snappy|none）
compression : snappy

# 后台 flush 线程数量，默认为1，范围为 [1, 4]
max-background-flushes : 2

# 指定后台压缩线程数量，默认为1，范围为[1, 8]
max-background-compactions : 4

# 缓存文件数量上限，默认为 5000
max-cache-files : 5000

# sstable 中 Level N+1 的最大容量是 Level N 的多少倍
# 默认为 10，可以设置为 5
max-bytes-for-level-multiplier : 10
EOF

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

cat > /usr/lib/systemd/system/pika.service <<EOF
[Unit]
Description=pika server

[Service]
WorkingDirectory=/data/pika/
ExecStart=/usr/local/pika/bin/pika -c /data/pika/pika.conf
Restart=always
User=worker
Group=worker

[Install]
WantedBy=multi-user.target
EOF

1
2
3
4
5

# reload service
systemctl daemon-reload

# enable service，auto start service after system boot
systemctl enable pika.service

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

# start service
systemctl start pika

# stop service
systemctl stop pika

# restart service
systemctl restart pika

# show service status
systemctl status pika

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

# run on slave

# stop service
systemctl stop pika

# clean data
rm -rf /data/pika/*/*

# start service
systemctl start pika

1
2

# option in pika config file
slaveof : 172.20.2.1:9231

1
2

# run on slave
redis-cli -p 9231 slaveof 172.20.2.1 9231

1
2
3
4
5
6
7

# run on master

# dump master data
redis-cli -p 9231 bgsave

# set master keep more binlog
redis-cli -p 9231 config set expire-logs-nums 300

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

# run on slave

# stop service
systemctl stop pika

# create sync directory
mkdir -p /data/pika/sync

# create rsync config file
cat > /data/pika/sync/rsync.conf <<EOF
uid = worker
gid = worker
use chroot = no
log file = /data/pika/sync/rsyncd.log
pid file = /data/pika/sync/rsyncd.pid
lock file = /data/pika/sync/rsync.lock

[pika]
uid = worker
gid = worker
path = /data/pika/db
comment = pika db files
ignore errors = yes
read only = no
list = no
auth users = pika
secrets file = /data/pika/sync/rsyncd.secrets
EOF

# create rsync secrets file
cat > /data/pika/sync/rsyncd.secrets <<EOF
pika:pika
EOF
chmod 600 /data/pika/sync/rsyncd.secrets

# start rsync service
rsync --daemon --config=/data/pika/sync/rsync.conf --address=<slave_ip> --port=12231

1
2

# run on master
rsync -avP --bwlimit=409600 --port=12231 /data/pika/dump/<today_date>/* pika@<slave_ip>::pika

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

# run on slave

# stop rsync service
kill `ps ax | grep rsync | grep 12231 | awk '{print $1}'`

# clean sync files
rm -rf /data/pika/sync

# delete sync info file
rm -f /data/pika/db/info

# start service
systemctl start pika

# start sync with assign binlog file and pos
redis-cli -p 9231 slaveof <master_ip> <master_port> <start_binlog_name> <start_binlog_pos>

1
2

# run on master and slave
redis-cli -p 9231 info log

1
2
3
4

# run on master

# clean dump files
redis-cli -p 9231 delbackup

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

pooling-pika-message-socket:
  listen: /usr/local/nutcracker/run/pika.sock 0777
  hash: fnv1a_64
  distribution: modula
  redis: true
  timeout: 1000
  backlog: 2048
  preconnect: false
  auto_eject_hosts: false
  server_retry_timeout: 2000
  server_failure_limit: 1
  servers:
  - pika01.may.ac.cn:9231:1

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181

#!/usr/bin/php
<?php
set_time_limit(0);
ini_set('default_socket_timeout', 300);

/**
 * 冗余键清理器
 */
class Cleaner {

  /**
   * 每次扫描的数量
   */
  const BATCH_NUM = 100;
  /**
   * 每批处理的休息时间（ms）
   */
  const REST_TIME = 10;
  /**
   * 重试间隔时间（ms）
   */
  const RETRY_TIME = 3000;
  /**
   * 重试次数
   */
  const RETRY_TIMES = 10;

  /**
   * 哈希算法
   */
  const HASH_ALGO = 'fnv1a64';

  /**
   * Redis 主机地址
   *
   * @var string
   */
  private $host;
  /**
   * Redis 端口
   *
   * @var integer
   */
  private $port;
  /**
   * 服务器 Hash 模
   *
   * @var integer
   */
  private $modula;
  /**
   * 服务器 Hash 索引
   *
   * @var integer
   */
  private $index;
  /**
   * Redis实例
   *
   * @var \Redis
   */
  private $redis = NULL;

  /**
   * 初始化清理器
   *
   * @param string $host
   * @param int    $port
   * @param int    $modula
   * @param int    $index
   */
  public function __construct($host, $port, $modula, $index) {
    $this->host = $host;
    $this->port = $port;
    $this->index = $index;
    $this->modula = $modula;
  }

  /**
   * 获取键的哈希索引
   *
   * @param string $key
   * @return int
   */
  public function getKeyIndex($key) {
    return hexdec(substr(hash(self::HASH_ALGO, $key), -8)) % $this->modula;
  }

  /**
   * 执行扫描
   *
   * @param int $iterator
   * @param string $pattern
   * @param int $count
   * @return array|bool
   * @throws Exception
   */
  public function scan(&$iterator, $pattern = null, $count = 0) {
    for ($i = 0; $i < self::RETRY_TIMES; $i++) {
      try {
        if (empty($this->redis)) {
          $this->redis = new Redis();
          $this->redis->connect($this->host, $this->port);
        }

        return $this->redis->scan($iterator, $pattern, $count);
      } catch (Exception $e) {
        $this->redis = NULL;
        print_r($e);
        usleep(self::RETRY_TIME * 1000);
      }
    }

    throw new Exception("Too many retry times!");
  }

  /**
   * 执行查询
   *
   * @param string $command
   * @param array  ...$params
   * @return mixed
   * @throws Exception
   */
  public function exec($command, ...$params) {
    for ($i = 0; $i < self::RETRY_TIMES; $i++) {
      try {
        if (empty($this->redis)) {
          $this->redis = new Redis();
          $this->redis->connect($this->host, $this->port);
        }

        return $this->redis->$command(...$params);
      } catch (Exception $e) {
        $this->redis = NULL;
        print_r($e);
        usleep(self::RETRY_TIME * 1000);
      }
    }

    throw new Exception("Too many retry times!");
  }

  /**
   * 开始清理
   */
  public function start() {
    $cursor = NULL;

    do {
      $keys = $this->scan($cursor, '*', self::BATCH_NUM);

      if (!empty($keys)) {
        $delKeys = [];

        foreach ($keys as $key) {
          if ($this->getKeyIndex($key) != $this->index) {
            $delKeys[] = $key;
          }
        }

        if (count($delKeys) > 0) {
          echo implode("\n", $delKeys) . "\n";
          $this->exec('del', $delKeys);
        }
      }
    }
    while ($cursor > 0);
  }
}

if (count($argv) < 5) {
  echo "Error params!\n";
  echo "eg: ./keyCleaner 127.0.0.1 6379 4 1\n";
  exit(1);
}

list($script, $host, $port, $modula, $index) = $argv;

$cleaner = new Cleaner($host, $port, $modula, $index);
$cleaner->start();

1

./keyCleaner host port total_node_num node_index

1
2
3
4

  servers:
  - pika01.may.ac.cn:9231:1
  - pika02.may.ac.cn:9231:1
  - pika03.may.ac.cn:9231:1

1
2
3
4

  servers:
  - pika01.may.ac.cn:9231:1
  - pika03.may.ac.cn:9231:1
  - pika02.may.ac.cn:9231:1

1
2
3
4

  servers:
  - pika01.may.ac.cn:9231:1 server01
  - pika03.may.ac.cn:9231:1 server02
  - pika02.may.ac.cn:9231:1 server03

1

redis-cli -p 9231 info

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59

# Server
pika_version:2.2.6 ----------------------------- pika 版本信息
os:Linux 3.10.0-514.16.1.el7.x86_64 x86_64 ----- 操作系统信息
arch_bits:64 ----------------------------------- 操作系统位数
process_id:12226 ------------------------------- pika 进程 id
tcp_port:9231 ---------------------------------- pika 监听端口
thread_num:6 ----------------------------------- pika 线程数量
sync_thread_num:6 ------------------------------ pika 同步线程数量
uptime_in_seconds:92832 ------------------------ pika 运行时间（秒）
uptime_in_days:2 ------------------------------- pika 运行时间（天）
config_file:/data/pika/pika.conf --------------- pika 配置文件路径

# Data
db_size:167421439361 --------------------------- pika db 当前大小（byte）
db_size_human:159665M -------------------------- pika db 当前大小（自适应）
compression:snappy ----------------------------- pika 当前压缩方案
used_memory:2658139096 ------------------------- pika 占用的内存大小（byte）
used_memory_human:2534M ------------------------ pika 占用的内存大小（自适应）
db_memtable_usage:335232088 -------------------- pika memtable 用量
db_tablereader_usage:2322907008 ---------------- pika tablereader 用量

# Log
log_size:10601924353 --------------------------- pika log 当前大小（byte）
log_size_human:10110M -------------------------- pika log 当前大小（自适应）
safety_purge:write2file111795 ------------------ pika 当前能够安全清理的最新日志
expire_logs_days:7 ----------------------------- pika 日志过期时间
expire_logs_nums:100 --------------------------- pika 日志保留最大数量
binlog_offset:111805 28779269 ------------------ pika 当前的 binlog id 和位置

# Clients
connected_clients:1 ---------------------------- 当前连接数

# Stats
total_connections_received:1617 ---------------- 总连接次数统计
instantaneous_ops_per_sec:4679 ----------------- 当前 qps
total_commands_processed:354452092 ------------- 总执行命令次数统计
is_bgsaving:No, , 0 ---------------------------- pika 备份信息：是否在备份，备份名称
is_scaning_keyspace:No ------------------------- pika 是否在扫描 keyspace
is_compact:No ---------------------------------- pika 是否在执行数据压缩

# Replication(MASTER)
role:master ------------------------------------ pika 在主从同步中的角色（slave|master）
connected_slaves:1 ----------------------------- 连接的从实例数量
slave0: host_port=172.20.2.230:93 state=online - 连接到主服务器的从实例信息列表

# Replication(SLAVE)
role:slave ------------------------------------- pika 在主从同步中的角色（slave|master）
master_host:172.20.2.230 ----------------------- pika 主服务器地址
master_port:9231 ------------------------------- pika 主服务器端口
master_link_status:up -------------------------- 与主服务器的链接状态（up|down）
slave_read_only:1 ------------------------------ 从服务器是否只读（0|1）

# Keyspace ------------------------------------- key 数量展示，仅在执行 info keyspace 1 时更新
# Time:1970-01-01 08:00:00 --------------------- 上一次统计的时间
kv keys:0
hash keys:0
list keys:0
zset keys:0
set keys:0

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

# /home/worker/monitor/pika_monitor.py
import re
import commands
import time
import requests
import json
import socket

_stat_regex = re.compile(r'(\w+):([0-9]+\.?[0-9]*)\r')


def get_info(host, port):
    command = 'redis-cli -h %s -p %s info' % (host, port)
    info = commands.getoutput(command)
    return dict(_stat_regex.findall(info))


def format_falcon_data(info_dict, endpoint, metric):
    timestamp = int(time.time())
    expected_keys = {
        'db_size': 'GAUGE',
        'connected_clients': 'GAUGE',
        'used_memory': 'GAUGE',
        'total_commands_processed': 'COUNTER'
    }
    event_list = []

    for key, vtype in expected_keys.items():
        if key not in info_dict:
            continue
        else:
            try:
                value = int(info_dict[key])
            except:
                continue
            event_list.append({
                'Metric': '%s.%s' % (metric, key),
                'Endpoint': endpoint,
                'Timestamp': timestamp,
                'Step': 60,
                'Value': value,
                'CounterType': vtype,
            })

    return event_list


if __name__ == "__main__":
    url: http://127.0.0.1:1988/v1/push"
    host = '127.0.0.1'
    port = 9231
    endpoint = socket.gethostname()

    info_dict = get_info(host, port)
    event_list = format_falcon_data(info_dict, endpoint, "pika")
    result = requests.post(url, data=json.dumps(event_list))
    print "falcon result: %s" % result.content

1

* * * * * python /home/worker/monitor/pika_monitor.py >> /home/worker/monitor/pika_monitor.log