基于Codis的冷热缓存优化

Posted 2021-08-03 AggrxTech

基于Codis的冷热缓存优化

背景

1. 热数据运维挑战：缓存存储分别有memcache,redis,redis cluster,codis等缓存对维护和运维带来很大的挑战。

2. 冷数据成本挑战: 部分单服务内存超过500G根据三地两中心三副本部署共需要1.5T内存，未来随着用户增加内存不断的增加。

方案选择

1. 针对热数据运维挑战：统一为codis方面集群的扩缩，自动负载，数据迁移和监控等

   
   
   

2. 冷数据成本挑战：兼容codis协议使用磁盘存储kvrocks替换内存存储redis
   
   

   

性能及问题

codis-redis内存存储问题

1. 不支持多网卡造成codis-proxy添加集群auth认证失败

2. 缺少监控耗时：仅仅有codis-redis实例耗时没有codis-proxy监控耗时，造成业务反馈超时codis本身没有超时

3. 懒惰删除不及时造成集群空间浪费比较大

codis-kvrocks磁盘存储问题

1. 不支持codis协议，不支持水平扩缩容，不支持读写分离

2. 读耗时较高

解决方案

codis-redis内存存储问题

    var broadcaseAddrs []string
    interfaces, err := net.Interfaces()
    if err != nil {
        log.WarnErrorf(err, "Can not get local interface info")
    }
    for _, inter := range interfaces {
        flags := inter.Flags.String()
        if strings.Contains(flags, "up") && strings.Contains(flags, "broadcast") {
            addrs,err := inter.Addrs()
            if err!=nil {
                log.WarnErrorf( err,"Can not get inter.Addrs")
            }else {
                for _,value :=range addrs {
                    broadcaseAddrs = append (broadcaseAddrs,value.String())
                }
                log.Infof("broadcaseAddrs %s ", broadcaseAddrs)
            }
        }
    }

2: 针对耗时无监控添加耗时超时100ms日志（请求ip，redis ip，redis命令，命令长度等维度）,Filebeat采集日志针对耗时高请求进行监控报警：

func (s *Session) loopWriter(tasks *RequestChan) (err error) {
    p := s.Conn.FlushEncoder()
    p.MaxInterval = time.Millisecond
    p.MaxBuffered = maxPipelineLen / 2

    return tasks.PopFrontAll(func(r *Request) error {
        resp, err := s.handleResponse(r)
        if err != nil {
            resp = redis.NewErrorf("ERR loopWriter handle response, %s", err)
            if s.Conn!=nil {
                log.Errorf("ERR loopWriter handle response err=%s,remotip=%s", err.Error() ,s.Conn.RemoteAddr())
            }else{
                log.Errorf("ERR loopWriter handle response err=%s", err.Error())
            }
            return s.incrOpFails(r, err)
        }
        fflush := tasks.IsEmpty()
        if err := p.Flush(fflush); err != nil {
            return s.incrOpFails(r, err)
        } else {
            s.incrOpStats(r, resp.Type)
            cost:= time.Now().UnixNano() - r.UnixNano
            if s.config.SessionSlowlogTime > 0 && cost > s.config.SessionSlowlogTime*1000000 {
                log.Warnf("cmd=%s, resType=%s,resLen=%d,cost=%d,forward addr=%s,req key=%s, reqip=%s",r.OpStr, r.Resp.Type.String(),len(r.Resp.Value),cost/1000000,r.addr,r.key,s.Conn.RemoteAddr())
            }
        }
        if fflush {
            s.flushOpStats(false)
        }
        return nil
    })
}

3: 针对懒惰删除机制慢，加大低峰期删除key个数和频率,上线后删除过期的key加大10倍，节省内存资源。

void activeExpireCycle(int type) {
    time_t tmpcal_ptr;
    struct tm *tmp_ptr = NULL;
    time(&tmpcal_ptr);
    tmp_ptr = localtime(&tmpcal_ptr);
    if (tmp_ptr->tm_hour >3 && tmp_ptr->tm_hour<=6 ){
        timelimit = 1000000*ACTIVE_EXPIRE_CYCLE_SLOW_TIME_PERC/100;
    }else{
        timelimit = 1000000*ACTIVE_EXPIRE_CYCLE_SLOW_TIME_PERC/server.hz/100;
    }
    for (j = 0; j < dbs_per_call; j++) {
        int expired,lookup_num;
        redisDb *db = server.db+(current_db % server.dbnum);
        do {
            slots = dictSlots(db->expires);
            now = mstime();
            if (num && slots > DICT_HT_INITIAL_SIZE &&
                (num*100/slots < 1)) break;
            expired = 0;
            ttl_sum = 0;
            ttl_samples = 0;
            if (tmp_ptr->tm_hour >=3 && tmp_ptr->tm_hour<=6 ){
                num = ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP*10;
            }else {
                num = ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP;
            }
            lookup_num = num;

            while (num--) {
                dictEntry *de;
                long long ttl;

                if ((de = dictGetRandomKey(db->expires)) == NULL) break;
                ttl = dictGetSignedIntegerVal(de)-now;
                if (activeExpireCycleTryExpire(db,de,now)) expired++;
                if (ttl > 0) {
                    /* We want the average TTL of keys yet not expired. */
                    ttl_sum += ttl;
                    ttl_samples++;
                }
            }
        } while (expired > lookup_num/4);
    }
}

codis-kvrocks

性能数据（第三方数据显示qps可以和redis媲美，耗时较高）

整体架构

问题

1. 针对不支持codis增加对codis slot命令支持：

    ADD_CMD("slotsscan",              CommandSlotsScan),
    ADD_CMD("slotsdel",               CommandSlotsDel),
    ADD_CMD("slotsmgrtslot",          CommandSlotsMgrtSlot),
    ADD_CMD("slotsmgrtone",           CommandSlotsMgrtOne),
    ADD_CMD("slotsmgrttagslot",       CommandSlotsMgrtTagSlot),
    ADD_CMD("slotsmgrttagone",        CommandSlotsMgrtTagOne),
    ADD_CMD("slotsrestore",           CommandSlotsRestore),
    ADD_CMD("slotshashkey",           CommandSlotsHashKey),
    ADD_CMD("slotscheck",             CommandSlotsCheck),
    ADD_CMD("slotsmgrtslot-async",    CommandSlotsMgrtSlotAsync),
    ADD_CMD("slotsmgrttagslot-async", CommandSlotsMgrtTagSlotAsync),
    ADD_CMD("slotsmgrt-exec-wrapper", CommandSlotsMgrtExecWrapper),
    ADD_CMD("slotsmgrt-async-status", CommandSlotsMgrtAsyncStatus),
    ADD_CMD("slotsmgrt-async-cancel", CommandSlotsMgrtAsyncCancel),

   如果codis扩容实例需要迁移slot中keys代码如下：

void SlotsMgrtSenderThread::loop() {
  Redis::Slot slot_db(storage_);
  while (!IsStopped()) {
    auto s = Util::SockConnect(dest_ip_, dest_port_, &sock_fd, timeout_ms_, timeout_ms_);
    moved_keys_all_ = 0;
    while (is_migrating_) {
      if (keys_num_ <= 0) {
        sleep(1);
        continue;
      }
      std::vector<std::string> migrate_batch_keys;
      auto s = ElectMigrateKeys(&migrate_batch_keys);
      std::lock_guard<std::mutex> ones_guard(ones_mu_);
      std::copy(migrating_ones_.begin(), migrating_ones_.end(), std::back_inserter(migrate_batch_keys));
      if (migrate_batch_keys.size() != 0) {
          moved_keys_num_ = 0;
      }
      std::vector<std::string>().swap(migrating_ones_);
      for (auto const &key : migrate_batch_keys) {
        auto s = slot_db.MigrateOneKey(sock_fd, key);
        moved_keys_num_++;
        moved_keys_all_++;
        remained_keys_num_--;
      }
      if (error_) {
        break;
 }
 if (remained_keys_num_ == 0) {
        LOG(INFO) << "[slots-mgrt-sender-thread] Migrate slot: " << slot_num_ << " finished";
        slotsmgrt_cond_.Signal();
        StopMigrateSlot();
        break;
      }
    }
}  // namespace Redis

    2. 读耗时优化

RocksDB 的全量 Compact 导致磁盘 IO 从而造成业务访问的毛刺点问题，之前策略是每天凌晨低峰时段进行一次，过于频繁会导致访问毛刺点，频率过低会导致磁盘空间回收不及时。所以增加另外一种部分 Compact 策略，优先对那些比较老以及无效 KV 比较多的 SST进行 Compact。开启只需要在配置文件里面增加一行，那么则会在凌晨 3 到 7 点之间去检查这些 SST 文件并做Compact。

另外由于RocksDB定时Compact sst文件造成大量磁盘IO，而像SATA磁盘IO瓶颈最大200M/s超过这个阀值读写都要等待，所以还需要设置磁盘Compact最大磁盘IO阀值小于磁盘IO的瓶颈。

# The maximum allowed aggregated write rate of flush and compaction (in MB/s).
# If the rate exceeds max-io-mb, io will slow down.
# 0 is no limit
# Default: 500
max-io-mb 150