工作使用场景

工作中需要通过CPU、内存生成资源利用率报表，可以通过Prometheus的API写一个Python脚本。

可通过API获取数据，然后再进行数据排序、过滤、运算、聚合，最后写入Mysql数据库。

CPU峰值计算

取最近一周CPU数值，再排序取最高的值。

def get_cpu_average(self):    """        CPU忙时平均值：取最近一周CPU数据，每一天的TOP20除以20得到忙时平均值；        再将一周得到的忙时平均值相加，再除以7，得到时间范围内一周的忙时平均值。        :return:        """    cpu_average = {}    for t in range(len(self.time_list)):        if t + 1 < len(self.time_list):            start_time = self.time_list[t]            end_time = self. time_list[t + 1]            # print(start_time, end_time)            # 拼接URL            pre_url = server_ip + '/api/v1/query_range?query='            expr = '100 - (avg by(instance)  (irate(node_cpu_seconds_total{mode="idle"}[5m])) * 100) ' \            '&start=%s&end=%s&step=300' % (start_time, end_time - 1)            url = pre_url + expr            #  print(url)            # 请求接口数据            data = json.loads(requests.post(url=url, headers=self.headers).content.decode('utf8', 'ignore'))            for da in data.get('data').get('result'):   # 循环拿到result数据                values = da.get('values')                cpu_load = [float(v[1]) for v in values]    # 循环拿到values里面的所有值                ip = da.get('metric').get('instance')       # 拿到instance里面的ip                ip = ip[:ip.index(':')] if  ':' in ip else ip    # 去除个别后面带的端口号                # avg_cup_load = sum(sorted(cpu_load, reverse=True)[:20]) / 20                # 取top20% 再除以20%，得出top20%的平均值                # avg_cup_load =  round(sum(sorted(cpu_load, reverse=True)[:round(len(cpu_load) * 0.2)]) / round(len(cpu_load) * 0.2), 2)                # 倒序后取前面20%除以个数，得到前20%的平均值                avg_cup_load = sum(sorted(cpu_load,  reverse=True)[:round(len(cpu_load) * 0.2)]) / round(len(cpu_load) * 0.2)                # print(avg_cup_load)                # 将计算后的数据以ip为key写入字典                if cpu_average.get(ip):                    cpu_average[ip].append(avg_cup_load)                    else:                        cpu_average[ip] = [avg_cup_load]                        # 每日top20的平均值累加，共7天的再除以7                        for k, v in cpu_average.items():                            #  cpu_average[k] = round(sum(v) / 7, 2)                            cpu_average[k] = sum(v)                            # print(cpu_average)                            return cpu_average                        cpu_average[ip] = [avg_cup_load]                        # 每日top20的平均值累加，共7天的再除以7                        for k, v in cpu_average.items():                            # cpu_average[k] =  round(sum(v) / 7, 2)                            cpu_average[k] = sum(v)                            # print(cpu_average)                            return cpu_average                        cpu_average[ip] = [avg_cup_load]                        # 每日top20的平均值累加，共7天的再除以7                        for k, v in cpu_average.items():                            # cpu_average[k] =  round(sum(v) / 7, 2)                            cpu_average[k] = sum(v)                            # print(cpu_average)                            return cpu_average

内存峰值计算

取7天内存数值，排序后取最高峰值

def get_mem_average(self):    """        内存忙时平均值：先取出7天的日期，根据多条链接循环取出每天数据，排序value取top20除以20，最终7天数据再除以7        :return:        """    avg_mem_util = {}    for t in range(len(self.time_list)):        if t + 1 < len(self.time_list):            start_time = self.time_list[t]            end_time = self.time_list[t + 1]            # 根据多条链接循环取出每天数据            pre_url = self.server_ip + '/api/v1/query_range?query='            # expr = '(node_memory_MemTotal_bytes - (node_memory_MemFree_bytes+node_memory_Buffers_bytes+node_memory_Cached_bytes )) / node_memory_MemTotal_bytes * 100&start=%s&end=%s&step=300' % (start_time, end_time)            expr_MenTotal = 'node_memory_MemTotal_bytes&start=%s&end=%s&step=600' % (start_time, end_time - 1)            expr_MemFree = 'node_memory_MemFree_bytes&start=%s&end=%s&step=600' % (start_time, end_time - 1)            expr_Buffers = 'node_memory_Buffers_bytes&start=%s&end=%s&step=600' % (start_time, end_time - 1)            expr_Cached = 'node_memory_Cached_bytes&start=%s&end=%s&step=600' % (start_time, end_time - 1)            result = {}            # 循环取出四个字段            for ur in expr_MenTotal, expr_MemFree, expr_Buffers, expr_Cached:                url = pre_url + ur                data = json.loads(requests.post(url=url, headers=self.headers).content.decode('utf8', 'ignore'))                ip_dict = {}                # 循环单个字段所有值                for da in data.get('data').get('result'):                    ip = da.get('metric').get('instance')                    ip = ip[:ip.index(':')] if ':' in ip else ip                    if ip_dict.get(ip):                        # print("重复ip：%s" % (ip))                        continue                        values = da.get('values')                        # 将列表里的值转为字典方便计算                        values_dict = {}                        for v in values:                            values_dict[str(v[0])] = v[1]                            # 标记ip存在                            ip_dict[ip] = True                            # 建立列表追加字典                            if result.get(ip):                                result[ip].append(values_dict)                                else:                                    result[ip] = [values_dict]                                    # print(result)                                    for ip, values in result.items():                                        values_list = []                                        for k, v in values[0].items():                                            try:                                                values_MenTotal = float(v)                                                values_MemFree = float(values[1].get(k, 0)) if values[1] else 0                                                values_Buffers = float(values[2].get(k, 0)) if values[2] else 0                                                values_Cached = float(values[3].get(k, 0)) if values[3] else 0                                                if values_MemFree == 0.0 or values_Buffers == 0.0 or values_Cached == 0.0:                                                    continue                                                    value_calc = (values_MenTotal - (values_MemFree + values_Buffers + values_Cached)) / values_MenTotal * 100                                                    if value_calc != float(0):                                                        values_list.append(value_calc)                                                        except Exception as e:                                                            print(values[0])                                                            # logging.exception(e)                                                            continue                                                            # 排序value取top20除以20                                                            # avg_mem = round(sum(sorted(values_list, reverse=True)[:round(len(values_list) * 0.2)]) / round(len(values_list) * 0.2), 2)                                                            try:                                                                avg_mem = sum(sorted(values_list, reverse=True)[:round(len(values_list) * 0.2)]) / round(len(values_list) * 0.2)                                                                except Exception as e:                                                                    avg_mem = 0                                                                    logging.exception(e)                                                                    if avg_mem_util.get(ip):                                                                        avg_mem_util[ip].append(avg_mem)                                                                        else:                                                                            avg_mem_util[ip] = [avg_mem]                                                                            # 最终7天数据再除以7                                                                            for k, v in avg_mem_util.items():                                                                                # avg_mem_util[k] = round(sum(v) / 7, 2)                                                                                avg_mem_util[k] = sum(v)                                                                                return avg_mem_util

导出excel

将采集到的数据导出excel

<pre class="hljs python" "="" style="overflow: auto; font-family: Menlo, Monaco, Consolas, "Courier New", monospace; font-size: 13px; padding: 0.5em; margin-top: 0px; margin-bottom: 10px; line-height: 1.42857; color: rgb(68, 68, 68); word-break: break-all; overflow-wrap: normal; background: rgb(240, 240, 240); border: 1px solid rgb(204, 204, 204); border-radius: 4px; max-width: ;">def export_excel(self, export):    """        将采集到的数据导出excel        :param export: 数据集合        :return:        """    try:        # 将字典列表转换为DataFrame        pf = pd.DataFrame(list(export))        # 指定字段顺序        order = ['ip', 'cpu_peak', 'cpu_average', 'mem_peak', 'mem_average', 'collector']        pf = pf[order]        # 将列名替换为中文        columns_map = {            'ip': 'ip',            'cpu_peak': 'CPU峰值利用率',            'cpu_average': 'CPU忙时平均峰值利用率',            'mem_peak': '内存峰值利用率',            'mem_average': '内存忙时平均峰值利用率',            'collector': '来源地址'        }        pf.rename(columns=columns_map, inplace=True)        # 指定生成的Excel表格名称        writer_name = self.Host + '.xlsx'        writer_name.replace(':18600', '')        # print(writer_name)        file_path = pd.ExcelWriter(writer_name.replace(':18600', ''))        # 替换空单元格        pf.fillna(' ', inplace=True)        # 输出        pf.to_excel(file_path, encoding='utf-8', index=False)        # 保存表格        file_path.save()        except Exception as e:            print(e)            logging.exception(e)