基础监控的同比告警主要是针对服务器监控采集的指标，包括负载（load1/load5/load15）、平均CPU使用率、内存使用率、内外网流量、端口数量等，具体采集方法可参考《基础监控-服务器监控》。

一、告警原理

　　多个指标每分钟1个数据，比较当前分钟的前10分钟的7天平均值，如果幅度超过100%并且绝对值相差达到M则算是一次异常，包括上升/下降异常，如果一个指标持续两次上升异常或者持续两次下降异常（不包括先上升后下降或者先下降后上升情况）则开始告警给机器的对应负责人（运维/开发）。例如当前是10号11:00，则比较的是10:59,10:58...10:50共10分钟的最近7天的平均值，例如10:59则是10号、9、8...4共7天的10:59这个点的数据的平均值。实际情况一般是第二分钟才解析了前一分钟的数据，相当于当前是11点，则解析的是10:59这个点和它之前的10分钟的7天平均值的比较。

二、数据来源

　　基础监控-服务器监控每分钟会采集一份数据保存到Redis中，保存格式是 reportTime - hash，hash的格式是{ip1:{item1:value, item2:value2...}, ip2:{item1:value, item2:value2...}, ip3...}，则每分钟一个redis hash，共7天 7*1440=10080个数据。实际情况保存的时候会多保留10分钟的数据，即7天+10分钟；由于reportTime是根据机器的实际时间来上报（这样画图才能保证是准确的），而某些机器没有NTP服务器或者其他原因导致时间不准，则reportTime则又会多种多样，所以导致的结果是redis的hash会变多一些，当然这并不影响我们的数据获取，因为整个同比告警就是根据画图来比较的，画图采用的是reportTime。采用hash保存到redis则不用每个ip读取一次redis，可以减少N次网络IO，大大提高程序速度。由于redis占据内存较多，大概10G左右，需要调整redis配置文件maxmemory的大小，不然redis会随机删除设置自动过期的key。

三、程序设计

1、DB设计

需要数据来源保存（redis）、异常展示表（mysql）、阈值配置表（mysql）、上次状态（redis）。异常展示表保存所有ip的异常描述信息、异常持续时间，可在页面展示；阈值配置表保存所有ip的阈值配置信息，每个ip的异常比率、各个指标的绝对值差、是否需要监控等；上次状态则用来判断是否需要告警的，持续2次同类异常则告警，使用redis保存即可。

mysql> show tables; +-----------------------------------+ | Tables_in_machineMonitor_overLast | +-----------------------------------+ | currentDisplay | | monitorConf | +-----------------------------------+

2、导入测试数据

测试数据是使用mysql和redis将数据从mysql导入redis中，线上数据是等程序完成后修改"服务器监控"的上报CGI来导入的。测试导入的时候遇到的坑参考《python处理json和redis hash的坑》

def initRedis(client): if client not in CONF.redisInfo: raise RedisNotFound("can not found redis %s in config" % client) try: pool = redis.ConnectionPool(**CONF.redisInfo[client]) # 线程安全 red = redis.Redis(connection_pool=pool) red.randomkey() # check except Exception, e: raise RedisException("connect redis %s failed: %s" % (client, str(e))) return red def initDb(client): if client not in CONF.dbInfo: raise MysqlNotFound("can not found mysql db %s in config" % client) try: db = opMysql.OP_MYSQL(**CONF.dbInfo[client]) except Exception, e: raise MysqlException("connect mysql %s failed: %s" % (client, str(e))) code, errMsg = db.connect() if 0 != code: raise MysqlException("connect mysql %s failed: %s" % (client, errMsg)) return db

3、告警接口

调用"告警平台"接口，项目配置后可发送RTX/SMS/Wechat，可在页面查看历史记录，轻松修改配置和临时屏蔽等。调用接口直接使用urllib/urllib2库即可

postData = urllib.urlencode(values) req = urllib2.Request(CONF.amcUrl, postData) response = urllib2.urlopen(req, timeout=CONF.AMC_TIME_OUT) amcDict = json.loads(response.read()) code = int(amcDict["returnCode"]) errMsg = amcDict["errorMsg"]

4、解析来源的数据

将数据转成可识别字典，并做排错处理，将错误数据拒绝掉

5、使用numpy和panda求平均值

将来源数据解析后存入panda.DataFrame中，如果数据不存在则使用numpy.nan代替，使用panda.DataFrame().mean().round(2)求平均值并保留2位数。如果某一分钟的7天数据全部获取不到则拒绝解析当前值，如果是7天数据的部分数据获取不到则剔除该点并在平均值的除数相对减一，使用NAN代替该点则在mean中可以解决这种情况。增加自定义函数比较每个列是否满足幅度上升/下降100%（可配置）并且绝对值差达到M（可配置），是则返回True，否则返回False，判断所有返回值是否均为True或者均为False，是则符合异常场景。

初始化DataFrame

for item in vd: if value is None or value[item] is None: vd[item][lastDayKey] = numpy.nan else: vd[item][lastDayKey] = value[item] vf = pandas.DataFrame(vd) columns.append(vf.mean().round(2)) indexes.append(lastMinKey) self.ipInfo[ip]["lastData"] = pandas.DataFrame(columns, index=indexes)

panda自定义函数比较并且判断是否需要告警

for item in curValue: if curValue[item] is None: # error continue else: curValue[item] = round(curValue[item], 2) def overLastCompute(v2, absSub): """ :param v2: float :param absSub: absolute subtract :return: high/low/null """ v1 = curValue[item] v2 = round(v2, 2) if 0 == v2: if v1 > absSub: return "HIGH" if v1 < -absSub: return "LOW" return "NULL" subVal = abs(v1 - v2) if subVal / v2 > CONF.RATIO and subVal > absSub: if v1 > v2: return "HIGH" return "LOW" return "NULL" self.ipInfo[ip]["result"][item] = self.ipInfo[ip]["lastData"][item].apply(overLastCompute, absSub=self.monitorConf[ip][item]) res = self.ipInfo[ip]["result"][item] == "HIGH" # Series if all(i for i in res): resErr[item] = CONF.HIGH_ERR if CONF.HIGH_ERR == self.lastCache[str(ip)][item]: # will Alert if switch on pass else: res = self.ipInfo[ip]["result"][item] == "LOW" if all(i for i in res): resErr[item] = CONF.LOW_ERR if CONF.LOW_ERR == self.lastCache[str(ip)][item]: # will Alert if switch on pass

6、由于IP较多，并且主要逻辑在解析数据和panda的计算上，使用CPU比较多，则需要使用多进程，并且结合线程池将进程跑满，别浪费进程资源。

step = ipNums / multiprocessing.cpu_count() ipList = list() i = 0 j = 1 processList = list() for ip in self.ipInfo: ipS = str(ip) if ipS not in self.lastCache: self.lastCache[ipS] = copy.deepcopy(self.value) ipList.append(ip) i += 1 if i == step * j or i == ipNums: j += 1 def innerRun(): wm = Pool.ThreadPool(CONF.POOL_SIZE) for myIp in ipList: kw = dict(ip=myIp, handlerKey=myIp) wm.addJob(self.handleOne, **kw) wm.waitForComplete() ipListNums = len(ipList) for tmp in xrange(ipListNums): res = wm.getResult() if res: handlerKey, code, handlerRet, errMsg = res if 0 != code: continue self.lastCache[str(handlerKey)] = handlerRet process = multiprocessing.Process(target=innerRun) process.start() processList.append(process) ipList = list() for process in processList: process.join()

四、优化

指标监控v2则是同比告警的升级版，数据量将会大好几倍，目前想到的优化如下

1、使用hbase替代redis

2、将程序做京广容灾，改成分布式运行，横向扩展多套程序并列运行