Riak集群监控详解

/usr/sbin/riak-admin status

/usr/sbin/riak-admin status

/usr/sbin/riak-admin status

/usr/sbin/riak-admin status

find /data1/riak/data/leveldb -name "LOG" -exec grep -l 'Compaction error' {} \; | wc -l

/usr/sbin/riak-admin status

/usr/sbin/riak-admin member-status | grep `ifconfig | grep "inet addr:10" | awk -F':' '{print $2}' | awk '{print $1}'`

valid       9.0%      --      'riak@10.1.80.114'

Riak 日志路径：/data1/riak/logs 
采集文件：/data1/riak/logs/* 
采集时间段：最近一分钟 
采集内容：最近一分钟发生的错误数 
采集示例：grep error -rn /data1/riak/logs | wc -l 
说明：这个采集需要程序处理下逻辑，在此不给出完整的采集方法

mysql> use riakMonitor
show tabReading table information for completion of table and column names
You can turn off this feature to get a quicker startup with -A

Database changed
mysql> show tables;
+---------------------------+
| Tables_in_riakMonitor     |
+---------------------------+
| riakClusterConf           | 
| riakClusterDisplay        | 
| riakClusterStatus         | 
| riakClusterStatusTemplate | 
| riakSingleConf            | 
| riakSingleDisplay         | 
| riakSingleStatus          | 
| riakSingleStatusTemplate  | 
+---------------------------+
8 rows in set (0.00 sec)

CGI :  /data/riakMonitor   # daemon
agent: /home/opd/script/riakMonitor  #  crond
analyzer: /opdData/opdOnline/script/kmc/riakMonitor/analyzer  # crond
1、从 CMDB 更新 single/cluster conf 数据  
2、同步 conf 和 display
3、解析 status 数据到 display
4、异常数据写入
5、告警
riakTool: /opdData/opdOnline/script/kmc/riakMonitor/riakTool  # daemon
每分钟第 50s 运行一次
1、获取监控集群和集群的 ip，计算结果并汇聚
2、操作 redis，将集群数据入历史库

def getRiakMeta():
    thisFuncName = str(sys._getframe().f_code.co_name)
    cmdStr = "/usr/sbin/riak-admin status"
    cmdCode, cmdStdout, cmdStderr = getCmdResult(cmdStr)
    if 0 != cmdCode:
        msgTxt = "[%s] %s failed" % (thisFuncName, cmdStr)
        logger.error(msgTxt)
        sendRtx(MYCONF.riakAdmins, thisFuncName+" %s Fail:" % cmdStr)
        return 1

    data["node_gets"] = data["node_puts"] = data["node_get_fsm_time_mean"] = data["node_get_fsm_time_median"] = 0
    data["node_get_fsm_time_95"] = data["node_get_fsm_time_100"] = data["node_put_fsm_time_mean"] = 0
    data["node_put_fsm_time_median"] = data["node_put_fsm_time_95"] = data["node_put_fsm_time_100"] = 0
    data["sys_process_count"] = data["memory_processes"] = data["memory_processes_used"] = 0
    data["read_repairs"] = data["node_get_fsm_siblings_mean"] = data["node_get_fsm_siblings_median"] = 0
    data["node_get_fsm_siblings_95"] = data["node_get_fsm_siblings_100"] = data["node_get_fsm_objsize_mean"] = 0
    data["node_get_fsm_objsize_median"] = data["node_get_fsm_objsize_95"] = data["node_get_fsm_objsize_100"] = 0
    data["leveldb_read_block_error"] = 0

    riakItemInfo = cmdStdout.split('\n')
    for each in riakItemInfo:
        eachInfo = each.split(" : ")
        if 2 == len(eachInfo):
            itemKey = eachInfo[0]
            itemValue = eachInfo[1].replace('<<"', '').replace('">>', '')
            if itemKey in data: 
                logger.debug("%s:%s" % (itemKey, itemValue))
                try:
                    data[itemKey] = str(round(float(itemValue), 2))
                except ValueError:
                    data[itemKey] = itemValue
                except:
                    raise
    
    cmdStr = """ find /data1/riak/data/leveldb -name"LOG"-exec grep -l'Compaction error'{} \; | wc -l """
    cmdCode, cmdStdout, cmdStderr = getCmdResult(cmdStr)
    if 0 != cmdCode:
        msgTxt = "[%s] %s failed" % (thisFuncName, cmdStr)
        logger.error(msgTxt)
        sendRtx(MYCONF.riakAdmins, thisFuncName+" Fail:" + msgTxt)
        return 1
    data["leveldb_compaction_error"] = cmdStdout #不用转 int
    
    cmdStr = "/usr/sbin/riak-admin member-status | grep %s" % data["mainIp"]
    cmdCode, cmdStdout, cmdStderr = getCmdResult(cmdStr)
    logger.debug(cmdStdout)
    if 0 != cmdCode:
        msgTxt = "[%s] %s failed" % (thisFuncName, cmdStr)
        logger.error(msgTxt)
        sendRtx(MYCONF.riakAdmins, thisFuncName+" Fail:" + msgTxt)
        return 1
    if cmdStdout.strip().startswith('valid'):
        data["is_active"] = 1
    else:
        data["is_active"] = 0

    data["riak_error_log"] = 0
    riakLogPath = "/data1/riak/logs/"
    if not os.path.isdir(riakLogPath):
        msgTxt = "[%s] %s not exists" % (thisFuncName, riakLogPath)
        logger.error(msgTxt)
        sendRtx(MYCONF.riakAdmins, thisFuncName+" Fail:" + msgTxt)
        return 1
    riakLogInfo = os.listdir(riakLogPath)
    reportTimeSec = time.mktime(time.strptime(data["report_time"], "%Y-%m-%d %H:%M:%S"))
    for each in riakLogInfo:
        logger.debug("fileName: "+each)
        eachFile = os.path.join(riakLogPath, each)
        if os.path.isfile(eachFile):
            try:
                eachFd = open(eachFile, 'r')
            except IOError, e:
                msgTxt = "I/O error({}): {}".format(e.errno, e.strerror)
                logger.error(msgTxt)
                sendRtx(MYCONF.riakAdmins, thisFuncName+" Fail:" + msgTxt)
                return 1
            else:
                for eachLine in eachFd: #从头读，怕文件太大撑爆内存
                    if "error" in eachLine: #2016-03-20 04:57:09.704 [info] <0.19012.49>@riak_kv_index_h
                        eachInfo = eachLine.split(' ')
                        try:
                            eachTimeStr = "%s %s" % (eachInfo[0], eachInfo[1][:-4])
                            eachTimeSec = time.mktime(time.strptime(eachTimeStr, "%Y-%m-%d %H:%M:%S"))
                            if reportTimeSec - 60 <= eachTimeSec < reportTimeSec:
                                logger.debug(eachLine)
                                data["riak_error_log"] += 1
                            elif eachTimeSec >= reportTimeSec:
                                break
                        except:
                            msgTxt = "file(%s) format wrong " % eachFile
                            logger.error(msgTxt)
                            break
                            #sendRtx(MYCONF.riakAdmins, thisFuncName+"Fail:" + msgTxt)
                            #eachFile.close()
                            #return 1
                eachFd.close()
    return 0

def report2server(content, retry):
    '''上报到入库程序，根据 ip 求余获取优先的 server，如果上报失败会遍历 server 列表'''
    thisFuncName = ""
    try:
        thisFuncName = str(sys._getframe().f_code.co_name)
        pos = data["ip"] % len(MYCONF.reportServer)
        serverKeys = MYCONF.reportServer.keys()
        serverKeys.sort()
        serverKeys = serverKeys[pos:] + serverKeys[:pos]
        for serverId in serverKeys:
            cmdStr = "/usr/bin/curl -s --connect-timeout %d -m %d -d'%s&reTry=%d'%s" %(MYCONF.curlConnectTimeout, MYCONF.curlMaxTimeout, content, retry, MYCONF.reportServer[serverId])
            cmdCode, cmdStdout, cmdStderr = getCmdResult(cmdStr)
            logger.info(cmdStr + "\ncmdCode:" + str(cmdCode) + "\n" + cmdStdout + cmdStderr)
            if 0 == cmdCode:
                return 0
        return 1
    except:
        exceptmsg = StringIO.StringIO()
        traceback.print_exc(file=exceptmsg)
        msgTxt = exceptmsg.getvalue()
        sendRtx(MYCONF.admins, thisFuncName + " Fail:" + msgTxt)
        return 1

def reportScheduler(reportRecord=0):
    '''reportRecord = 0 表示上报 data 中采集的新数据，reportRecord = 1 表示从 reportFailFile 里面获取最新的一条数据上报到 server，然后需要处理 reportFailFile'''
    thisFuncName = ""
    try:
        thisFuncName = str(sys._getframe().f_code.co_name)
        if 1 == reportRecord: # 从上报失败文件中获取最后一条数据，上报之
            if not reportFail.has_section("index"): #这里不要去 add_section("index") 该谁 add 谁 add 去
                return 0
            if not reportFail.has_option("index", "index") or "" == reportFail.get("index", "index").strip():
                return 0
            indexVec = MYCONF.splitRe.split(reportFail.get("index", "index").strip())
            index = indexVec[-1]
            if "" == index:
                msgTxt = reportFail.get("index", "index").strip()
                sendRtx(MYCONF.admins, thisFuncName + "[系统错误] index.index 末尾有多余的逗号 " + msgTxt)
                return 1
            if not reportFail.has_option("content", index + "_c") or not reportFail.has_option("content", index + "_t"): 
                # _c 是内容  _t 是重试次数
                msgTxt = "content sector 缺少 %s_c 或 %s_t" %(index, index)
                sendRtx(MYCONF.admins, thisFuncName + "[系统错误] " + msgTxt)
                return 1
            content = reportFail.get("content", index + "_c")
            retry = reportFail.getint("content", index + "_t")
            retry += 1
            code = report2server(content, retry)
            if 0 == code: # 发送成功
                indexVec.remove(index)
                if indexVec:
                    reportFail.set("index", "index", ",".join(indexVec))
                else:
                    reportFail.set("index", "index", "")
                reportFail.remove_option("content", index + "_c")
                reportFail.remove_option("content", index + "_t")
            elif retry > MYCONF.maxRetry: # 重发失败，且超过最大重试次数
                indexVec.remove(index)
                if indexVec:
                    reportFail.set("index", "index", ",".join(indexVec))
                else:
                    reportFail.set("index", "index", "")
                reportFail.remove_option("content", index + "_c")
                reportFail.remove_option("content", index + "_t")
            else: # 重发失败，更新 _t (retry) 字段
                reportFail.set("content", index + "_t", retry)
        else: # 发送新数据
            index = data["report_time"].replace(" ", "").replace("-", "").replace(":", "")
            content = urllib.urlencode(data)
            retry = 0
            code = report2server(content, retry)
            if 0 == code:
                return 0
            if not reportFail.has_section("index"):
                reportFail.add_section("index")
                reportFail.set("index", "index", index)
                reportFail.add_section("content")
                reportFail.set("content", index + "_c", content)
                reportFail.set("content", index + "_t", retry)
            else:
                indexVec = MYCONF.splitRe.split(reportFail.get("index", "index").strip())
                indexVec.append(index)
                if len(indexVec) > MYCONF.maxFailRecord: # 超过最大 fail record 数
                    reportFail.set("index", "index", ",".join(indexVec[len(indexVec) - MYCONF.maxFailRecord:]))
                    reportFail.set("content", index + "_c", content)
                    reportFail.set("content", index + "_t", retry)
                    for i in range(0, len(indexVec) - MYCONF.maxFailRecord):
                        delIndex = indexVec[i]
                        reportFail.remove_option("content", delIndex + "_c")
                        reportFail.remove_option("content", delIndex + "_t")
                else:
                    reportFail.set("index", "index", ",".join(indexVec))
                    reportFail.set("content", index + "_c", content)
                    reportFail.set("content", index + "_t", retry)
        return 0
    except:
        exceptmsg = StringIO.StringIO()
        traceback.print_exc(file=exceptmsg)
        msgTxt = exceptmsg.getvalue()
        sendRtx(MYCONF.admins, thisFuncName + " Fail:" + msgTxt)
        return 1

def getCmdResult(cmdStr):
    '''获取 shell 命令的返回码，标准输出，标准错误'''
    #child = subprocess.Popen(cmdStr, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, close_fds=True)
    #cmdStdout, cmdStderr = child.communicate()
    #cmdCode = child.wait() 
    #return (cmdCode, cmdStdout, cmdStderr)
    thisFuncName = str(sys._getframe().f_code.co_name)
    nowTime = int(time.time())
    tmpstdout = os.path.join(MYCONF.basePath, "cmd.stdout.%d.tmp" % nowTime)
    tmpstderr = os.path.join(MYCONF.basePath, "cmd.stderr.%d.tmp" % nowTime)
    if "debug" == MYCONF.role:
        msgTxt = "[%d]Run Cmd: %s" % (nowTime, cmdStr)
        logger.debug(msgTxt)

    cmdStr = "(%s) 1>%s 2>%s" %(cmdStr, tmpstdout, tmpstderr)
    cmdCode = os.system(cmdStr) >> 8
    cdmStdout = cmdStderr = ""
    try:
        outfd = open(tmpstdout)
        cmdStdout = outfd.read()
        errfd = open(tmpstderr)
        cmdStderr = errfd.read()
    except:
        exceptmsg = StringIO.StringIO()
        traceback.print_exc(file=exceptmsg)
        msgTxt = exceptmsg.getvalue()
        sendRtx(MYCONF.admins, thisFuncName + " Fail:" + msgTxt)
        cmdCode = 110
    else:
        outfd.close()
        errfd.close()
        os.remove(tmpstderr)
        os.remove(tmpstdout)
    return (cmdCode, cmdStdout, cmdStderr)

def readLastCache():
    global lastCache
    lastCache = ConfigParser.ConfigParser()
    if not os.path.isfile(MYCONF.lastCacheFile):
        try:
            fd = open(MYCONF.lastCacheFile, "w")
        except IOError, e:
            logger.error("I/O error({}): {}".format(e.errno, e.strerror))
            return 1
        else:
            fd.close()
    lastCache.readfp(open(MYCONF.lastCacheFile), "rb")
    return 0

def writeCache():
    thisFuncName = ""
    try:
        thisFuncName = str(sys._getframe().f_code.co_name)
        lastCache.write(open(MYCONF.lastCacheFile, 'w'))
        return 0
    except:
        exceptmsg = StringIO.StringIO()
        traceback.print_exc(file=exceptmsg)
        msgTxt = exceptmsg.getvalue()
        logger.error(msgTxt)
    return 1

def readFailRecord():
    global reportFail
    reportFail = ConfigParser.ConfigParser()
    if not os.path.isfile(MYCONF.lastReportFailFile):
        try:
            fd = open(MYCONF.lastReportFailFile, "w")
        except IOError, e:
            logger.error("I/O error({}): {}".format(e.errno, e.strerror))
            return 1
        else:
            fd.close()
    reportFail.readfp(open(MYCONF.lastReportFailFile), "rb")
    return 0

def writeFailRecord():
    thisFuncName = ""
    try:
        thisFuncName = str(sys._getframe().f_code.co_name)
        reportFail.write(open(MYCONF.lastReportFailFile, 'w'))
        return 0
    except:
        exceptmsg = StringIO.StringIO()
        traceback.print_exc(file=exceptmsg)
        msgTxt = exceptmsg.getvalue()
        logger.error(msgTxt)
    return 1

def main():
    data["osType"] = 0 # 0 表示 linux
    data["version"] = MYCONF.version # 当前程序的自定义版本号
    data["report_time"] = time.strftime("%Y-%m-%d %H:%M:00") #上报时间，由于目前基础监控是分钟级监控粒度，因此秒取 00
    
    initLog()
    logger.info('='*80)
    if 0 == checkLastPid() and 0 == readLastCache() and 0 == getLoginIp():
        readFailRecord() # 读取早迁采集周期上报失败，需要重传的数据
        reportScheduler(reportRecord=1) #从 fail record 中选取最近的一条信息上报给服务器
        if 0 == getRiakMeta():
            reportScheduler(reportRecord=0)
        writeFailRecord()
        writeCache()
    logger.info('='*80)
    logging.shutdown()
    return

添加监控需要先添加集群（不支持先添加 IP），添加集群会默认把所有 IP 都添加监控（前台将在 clusterConf 新增记录, 并在 singleConf 增加对应的 ip 记录，然后调用调度平台，检测 ip 是否已经安装）如果该集群在 CMDB 里面新增 Ip，则需要手动添加监控（前台提供新增监控节点，插入 singleConf）

(1) 卸载监控可以卸载整个集群的监控（将 clusterConf needMonitor 置 0，同步将 singleConf 的 needMonitor 都置 0，然后调用

调度平台  卸载集群下的所有机器，如果该 ip 存在其他集群并且需要监控，则不用调用  调度平台  卸载）也可以卸载单个节点的监控 (前台将 singleConf 的 needMonitor 置 0，调用 调度平台，同样判断 ip 是否存在其他集群）(2) 添加卸载监控部由前台调用

调度平台  接口，并修改数据库（插入数据或者更新 need_monitor) (3) Single/cluster dislplay 表会同步 conf 表的数据，只保留 need_monitor= 1 的数据

后台一直同步 CMDB 的数据和 conf 表的数据，如果不在 CMDB 的则需要删掉 conf 里面的数据，不管 needMonitor 的值为多少。删除三级业务的话只需要删除 clusterConf 表对应的记录，single 会自动同步外键(尝试调用

调度平台卸载接口，卸载掉被删除的三级业务 ID 下面的所有已安装监控的 IP）