分布式监控系统开发【day38】:报警阈值程序逻辑解析（四）

一、计算单条表达式的结果

1、解决了什么问题

1. 主机
2. 表达式
3. 多长时间进行一次监控
4. 拼出此服务在redis中存储的对应key
5. 获取要从redis中取多长时间的数据,单位为minute

2、代码实现

class ExpressionProcess(object):
    '''
    load data and calc it by different method
    '''
    def __init__(self,main_ins,host_obj,expression_obj,specified_item=None):
        '''
        :param main_ins:   DataHandler 实例
        :param host_obj: 具体的host obj
        :param expression_obj:
        :return:
        计算单条表达式的结果
        '''
        self.host_obj = host_obj
        self.expression_obj = expression_obj
        self.main_ins = main_ins
        self.service_redis_key = "StatusData_%s_%s_latest" %(host_obj.id,expression_obj.service.name) #拼出此服务在redis中存储的对应key
        self.time_range = self.expression_obj.data_calc_args.split(',')[0] #获取要从redis中取多长时间的数据,单位为minute

        print("33[31;1m------>%s33[0m" % self.service_redis_key)

二、如何拿到精确的数据

1、功能如下

1、我取出6个数据(下面的+60是默认多取一分钟数据,宁多勿少,多出来的后面会去掉)
2、approximate_data_range存的是大概的数据，要拿到精确的，我判断一下
3、把数据集合交给不同的方法去处理了
4、根据监控间隔去取数据，如果监控间隔改变了怎嘛办？

2、代码实现

    def load_data_from_redis(self):
        '''load data from redis according to expression's configuration'''
        time_in_sec = int(self.time_range) * 60  #下面的+60是默认多取一分钟数据,宁多勿少,多出来的后面会去掉
        approximate_data_points = (time_in_sec + 60) / self.expression_obj.service.interval #获取一个大概要取的值
        #stop_loading_flag = False #循环去redis里一个点一个点的取数据,直到变成True
        #while not stop_loading_flag:
        print("approximate dataset nums:", approximate_data_points,time_in_sec)
        data_range_raw = self.main_ins.redis.lrange(self.service_redis_key,-int(approximate_data_points),-1)
        #print("33[31;1m------>%s33[0m" % data_range)
        approximate_data_range = [json.loads(i.decode()) for i in data_range_raw]
        data_range = [] #精确的需要的数据 列表
        for point  in approximate_data_range:
            #print('bread point:', point)
            val,saving_time = point
            if time.time() - saving_time < time_in_sec :#代表数据有效
                data_range.append(point)
                #print("service index key:",self.expression_obj.service_index.key)
                #print(point)
                '''if val: #确保数据存在
                    if 'data' not in val:#代表这个dict没有sub_dict
                        print("33[44;1m%s33[0m" %val[self.expression_obj.service_index.key])
                        #如何处理这些数据 呢? 是求avg(5), hit(5,3)....? 看来只能把数据集合交给不同的方法去处理了
                        #self.process(self.)
                        #data_range.append(
                    else: #像disk , nic这种有多个item的数据
                        for k,v in val['data'].items():
                            print("33[45;1m%s, %s33[0m" %(k,v))
                            print("33[45;1m%s, %s33[0m" %(k,v[self.expression_obj.service_index.key]))
                '''
            #else:
            #    print("data is invalid")


        print(data_range)
        return data_range

三、算出单条expression表达式的结果

1、功能如下

1、按照用户的配置把数据 从redis里取出来了, 比如 最近5分钟,或10分钟的数据

2、确保上面的条件 有正确的返回

2、代码实现

    def process(self):
        """算出单条expression表达式的结果"""
        data_list = self.load_data_from_redis() #已经按照用户的配置把数据 从redis里取出来了, 比如 最近5分钟,或10分钟的数据
        data_calc_func = getattr(self,'get_%s' % self.expression_obj.data_calc_func)
        #data_calc_func = self.get_avg...
        single_expression_calc_res = data_calc_func(data_list) #[True,43,None]
        print("---res of single_expression_calc_res ",single_expression_calc_res)
        if single_expression_calc_res: #确保上面的条件 有正确的返回
            res_dic = {
                'calc_res':single_expression_calc_res[0],
                'calc_res_val':single_expression_calc_res[1],
                'expression_obj':self.expression_obj,
                'service_item':single_expression_calc_res[2],
            }

            print("33[41;1msingle_expression_calc_res:%s33[0m" % single_expression_calc_res)
            return res_dic
        else:
            return False

四、如何获取网卡的平均值

1、解决了什么问题

1、监控了特定的指标,比如有多个网卡,但这里只特定监控eth0，就是监控这个特定指标,match上了

2、在这里判断是否超越阈值

　　可能是由于最近这个服务没有数据汇报过来,取到的数据为空,所以没办法 判断阈值

3、监控这个服务的所有项, 比如一台机器的多个网卡, 任意一个超过了阈值,都算是问题的

1. 后面的循环不用走了,反正 已经成立了一个了
2. 能走到这一步,代表 上面的循环判段都未成立

2、代码实现

    def get_avg(self,data_set):
        '''
        return average value of given data set
        :param data_set:
        :return:
        '''
        clean_data_list = []
        clean_data_dic = {}
        for point in data_set:
            val,save_time = point
            #print('---point:>', val)

            if val:
                if 'data' not in val:#没有子dict
                    clean_data_list.append(val[self.expression_obj.service_index.key])

                else: #has sub dict
                    for k,v in val['data'].items():
                        if k not in clean_data_dic:
                            clean_data_dic[k]=[]

                        clean_data_dic[k].append(v[self.expression_obj.service_index.key])

        if clean_data_list:
            clean_data_list = [float(i) for i in clean_data_list]
            #avg_res = 0 if sum(clean_data_list) == 0 else  sum(clean_data_list)/ len(clean_data_list)
            avg_res = sum(clean_data_list)/ len(clean_data_list)
            print("33[46;1m----avg res:%s33[0m" % avg_res)
            return [self.judge(avg_res), avg_res,None]
            #print('clean data list:', clean_data_list)
        elif clean_data_dic:
            for k,v in clean_data_dic.items():
                clean_v_list = [float(i) for i in v]
                avg_res = 0 if sum(clean_v_list) == 0 else sum(clean_v_list) / len(clean_v_list)
                print("33[46;1m-%s---avg res:%s33[0m" % (k,avg_res))
                if self.expression_obj.specified_index_key:#监控了特定的指标,比如有多个网卡,但这里只特定监控eth0
                    if k == self.expression_obj.specified_index_key:#就是监控这个特定指标,match上了
                        #在这里判断是否超越阈值
                        print("test res [%s] [%s] [%s]=%s") %(avg_res,
                                                            self.expression_obj.operator_type,
                                                            self.expression_obj.threshold,
                                                            self.judge(avg_res),
                                                            )
                        calc_res = self.judge(avg_res)
                        if calc_res:
                            return  [calc_res,avg_res,k] #后面的循环不用走了,反正 已经成立了一个了
                else:#监控这个服务 的所有项, 比如一台机器的多个网卡, 任意一个超过了阈值,都 算是有问题的
                    calc_res = self.judge(avg_res)
                    if calc_res:
                        return [calc_res,avg_res,k]
                print('specified monitor key:',self.expression_obj.specified_index_key)
                print('clean data dic:',k,len(clean_v_list), clean_v_list)
            else: #能走到这一步,代表 上面的循环判段都未成立
                return [False,avg_res,k]
        else:#可能是由于最近这个服务 没有数据 汇报 过来,取到的数据 为空,所以没办法 判断阈值
            return [False,None,None]

3、监控这个服务的所有项, 比如一台机器的多个网卡, 任意一个超过了阈值,都算是有问题的

    def judge(self,calculated_val):
        '''
        determine whether the index has reached the alert benchmark
        :param calculated_val: #已经算好的结果,可能是avg(5) or ....
        :return:
        '''
        #expression_args = self.expression_obj.data_calc_args.split(',')
        #hit_times = expression_args[1] if len(expression_args)>1 else None
        #if hit_times:#定义了超过阈值几次的条件
        calc_func = getattr(operator,self.expression_obj.operator_type)
        #calc_func = operator.eq....
        return calc_func(calculated_val,self.expression_obj.threshold)

4、命中次数值返回给定数据集

    def get_hit(self,data_set):
        '''
        return hit times  value of given data set
        :param data_set:
        :return:
        '''
        pass