resource_data.py

    1. #!/usr/bin/env python
    2. # -*- coding: utf-8 -*
    4. """
    5.     hbase资源数据上报
    6. """
    8. import requests, json
    9. import pymysql
    10. from subprocess import check_output
    11. from datetime import datetime, timedelta
    12. from fastapi import FastAPI
    14. from sqlalchemy import create_engine, Column, String, Integer, FLOAT, Boolean
    15. from sqlalchemy.orm import sessionmaker
    16. from sqlalchemy.ext.declarative import declarative_base
    17. from sqlalchemy import and_, or_
    18. from logger import FastapiLoguru
    20. # API服务
    21. api = FastAPI()
    23. # 当前环境
    24. CURRENT_IP = check_output('hostname -i', shell=True)
    25. ENV = 'prd' if CURRENT_IP.startswith(b'10.117') or CURRENT_IP.startswith(b'10.150') else ('sit' if CURRENT_IP.startswith(b'10.203') else 'dev')
    26. DB_URL = "mysql+pymysql://bdpops:LTejd42ml6@bdpops-m.db.sfdc.com.cn:3306/bdpops?charset=utf8" if ENV == "prd" else "mysql+pymysql://bdpops:Oo7glydkwk@bdpops-m.dbsit.sfcloud.local:3306/bdpops?charset=utf8"
    28. # 日志处理
    29. logger = FastapiLoguru(ENV).get_logger()
    31. # MySQL连接引擎
    32. engine = create_engine(DB_URL)
    33. DBsession = sessionmaker(bind=engine)
    35. # MySQL连接会话
    36. session = DBsession()
    38. # ORM实体继承对象
    39. Base = declarative_base()
    41. # ORM实体类
    42. class HRMCluster(Base):
    43.     """
    44.         集群信息表
    45.     """
    46.     __tablename__ = "hrm_cluster_info"
    47.     id = Column(Integer, primary_key=True, index=True)
    48.     clusterName = Column(String(50), nullable=False, comment="集群名称")
    49.     clusterId = Column(String(50), nullable=False, comment="集群ID")
    50.     clusterLoad = Column(String(16), default=None, comment="集群Load")
    51.     clusterRequests = Column(FLOAT, default=None, comment="集群请求量")
    52.     numRegionServers = Column(Integer, default=None, comment="集群节点数")
    53.     numDeadRegionServers = Column(Integer, default=None, comment="集群Dead节点数")
    54.     zookeeperQuorum = Column(String(100), default=None, comment="集群Zookeeper")
    55.     liveRegionServers = Column(String(5000), default=None, comment="集群Live节点信息")
    56.     deadRegionServers = Column(String(5000), default=None, comment="集群Dead节点信息")
    57.     storageType = Column(String(20), comment="集群存储磁盘类型")
    60. class HRMRsgroup(Base):
    61.     """
    62.         集群分组信息表
    63.     """
    64.     __tablename__ = "hrm_rsgroup_info"
    65.     id = Column(Integer, primary_key=True, index=True)
    66.     groupName = Column(String(50), nullable=False, comment="分组名称")
    67.     groupDepartment = Column(String(500), default=None, comment="分组归属组织")
    68.     groupUser = Column(String(200), default=None, comment="分组使用者")
    69.     systemCode = Column(String(50), default=None, comment="系统编码")
    70.     systemLevel = Column(String(20), default=None, comment="系统级别")
    71.     clusterName = Column(String(50), nullable=False, comment="集群名称")
    72.     numServers = Column(Integer, default=None, comment="分组节点数")
    73.     numTables = Column(Integer, default=None, comment="分组表数量")
    74.     region_capacity = Column(Integer, default=None, comment="分组region容量")
    75.     region_used = Column(Integer, comment="分组region使用量")
    76.     region_free = Column(Integer, comment="分组region可用量")
    77.     region_utillization = Column(String(30), comment="分组region使用率")
    80. class HRMTable(Base):
    81.     """
    82.         集群Hbase表信息表
    83.     """
    84.     __tablename__ = "hrm_table_info"
    85.     id = Column(Integer, primary_key=True, index=True)
    86.     fulltableName = Column(String(300), nullable=False, comment="表全名")
    87.     namespcae = Column(String(50), nullable=False, comment="表命名空间")
    88.     tableName = Column(String(300), nullable=False, comment="表名")
    89.     tableRegions = Column(Integer, default=None, comment="表region数")
    90.     isEnable = Column(Boolean, default=None, comment="是否在线")
    91.     groupName = Column(String(50), nullable=False, comment="分组名称")
    92.     clusterName = Column(String(50), nullable=False, comment="集群名称")
    93.     tableDepartment = Column(String(500), default=None, comment="表归属组中")
    94.     tableUser = Column(String(200), default=None, comment="表使用者")
    95.     systemCode = Column(String(50), comment="系统编码")
    96.     systemLevel = Column(String(20), comment="系统级别")
    97.     overMatch = Column(Boolean, comment="表region数是否过多")
    98.     space = Column(String(50), comment="表所占存储空间")
    99.     dateMark = Column(String(20), comment="日期标识位")
    102. # Hbase资源数据采集上报函数
    103. def report_hbase_data():
    104.     # 采集结果列表
    105.     report_list = []
    107.     # 固定上报项
    108.     #classification = "realtime"
    109.     #component = "hbase"
    110.     consumer_classification = "bdpapp"
    111.     consumption_frequency = "day"
    112.     #consumption_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    113.     consumption_time = (datetime.now() + timedelta(days=-1)).strftime("%Y-%m-%d %H:%M:%S")
    115.     # 今日日期
    116.     todaydate = datetime.today().date()
    118.     rsgroup = session.query(HRMRsgroup).all()
    119.     for gn in rsgroup:
    120.         cluster = session.query(HRMCluster).filter(HRMCluster.clusterName == gn.clusterName).one()
    121.         volume_disk = cluster.storageType
    123.         if gn.groupName == 'default' and not (str(gn.clusterName).startswith("ebil") or str(gn.clusterName).startswith("rdmp")):
    124.             logger.info("Begin ------ 集群：{} 分组：{} 数据采集".format(gn.clusterName, gn.groupName))
    125.             table = session.query(HRMTable).filter(and_(HRMTable.groupName == 'default', HRMTable.clusterName == gn.clusterName, HRMTable.dateMark == todaydate)).all()
    126.             for t in table:
    127.                 report = {}
    128.                 #report['classification'] = classification
    129.                 #report['component'] = component
    130.                 report['consumer_classification'] = consumer_classification
    131.                 report['consumption_frequency'] = consumption_frequency
    132.                 report['consumption_time'] = consumption_time
    134.                 report['volume_unit'] = 'hbase_region'
    135.                 #report['volume_info'] = 'hbase集群名称_表名'
    136.                 #report['consumption_info'] = '表region数量'
    138.                 report['volume_area'] = "sit" if str(t.clusterName).rfind("test") != -1 else ( "prd" if str(t.clusterName).rfind("sc") != -1 else "dr")
    139.                 if t.systemCode is None or len(t.systemCode) == 0:
    140.                     logger.info("{} 该Table系统编码需要更新".format(t.fulltableName))
    141.                     continue
    142.                 #report['consumer_name'] = t.systemCode
    143.                 else:
    144.                     app_info = get_bdp_appid(t.systemCode)
    145.                     if app_info is not None:
    146.                         report['consumer_name'] = str(app_info['app_id'])
    147.                     else:
    148.                         logger.info("ROP 系统编码： {} 对应APP为空".format(t.systemCode))
    149.                         continue
    151.                 report['volume_name'] = t.clusterName + '_' + t.fulltableName
    152.                 report['volume_info'] = 'hbase集群：{}_公共分组表：{}'.format(t.clusterName, t.fulltableName)
    153.                 report['consumption_amount'] = t.tableRegions
    154.                 #report['consumption_info'] = '[消费组件：{}]，[消费方：{}]，[消费标识：hbase集群 {} | 公共分组表 {}]，[消费说明：表当前分区数量]，[消费量：{}]'.format(component, t.systemCode, t.clusterName, t.fulltableName, t.tableRegions)
    156.                 report['volume_disk'] = volume_disk
    158.                 report_list.append(report)
    160.         elif gn.groupName != 'discard':
    161.             logger.info("Begin ------ 集群：{} 分组：{} 数据采集".format(gn.clusterName, gn.groupName))
    162.             report = {}
    163.             #report['classification'] = classification
    164.             #report['component'] = component
    165.             report['consumer_classification'] = consumer_classification
    166.             report['consumption_frequency'] = consumption_frequency
    167.             report['consumption_time'] = consumption_time
    169.             report['volume_unit'] = 'hbase_group'
    170.             #report['volume_info'] = 'hbase集群名称_分组名'
    171.             #report['consumption_info'] = '分组RegionServer节点数'
    173.             report['volume_area'] = "sit" if str(gn.clusterName).rfind("test") != -1 else ( "prd" if str(gn.clusterName).rfind("sc") != -1 else "dr")
    174.             if gn.systemCode is None or len(gn.systemCode) == 0:
    175.                 logger.info("{} 该RsGroup系统编码需要更新".format(gn.groupName))
    176.                 continue
    177.             #report['consumer_name'] = gn.systemCode
    178.             else:
    179.                 app_info = get_bdp_appid(gn.systemCode)
    180.                 if app_info is not None:
    181.                     report['consumer_name'] = str(app_info['app_id'])
    182.                 else:
    183.                     logger.info("ROP 系统编码： {} 对应APP为空".format(gn.systemCode))
    184.                     continue
    186.             report['volume_name'] = gn.clusterName + '_' + gn.groupName
    187.             report['volume_info'] = 'hbase集群：{}_独立分组：{}'.format(gn.clusterName, gn.groupName)
    188.             report['consumption_amount'] = gn.numServers
    189.             #report['consumption_info'] = '[消费组件：{}]，[消费方：{}]，[消费标识：hbase集群 {} | 独立分组 {}]，[消费说明：分组当前服务器数量]，[消费量：{}]'.format(component, gn.systemCode, gn.clusterName, gn.groupName, gn.numServers)
    191.             report['volume_disk'] = volume_disk
    193.             report_list.append(report)
    194.         else:
    195.             pass
    197.     logger.info("End ------ 采集结束,采集条数: {}".format(len(report_list)))
    198.     return report_list
    201. # 获取BDP平台对应应用ID
    202. def get_bdp_appid(systemcode):
    203.     conn = pymysql.connect(
    204.         host='rccprd-m.db.sfcloud.local',
    205.         port=3306,
    206.         user='rcc',
    207.         password='LTejd42ml6',
    208.         database='rcc',
    209.         charset='utf8'
    210.     )
    211.     # 获取一个光标
    212.     cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)  # 返回字典数据类型
    214.     # 定义将要执行的sql语句
    215.     sql = "select system_code, app_id from rcc_v2_sys_app_mapping where system_code = '{}'".format(systemcode)
    216.     # 拼接并执行sql语句
    217.     cursor.execute(sql)
    218.     # 取到查询结果
    219.     ret = cursor.fetchone()  # 取一条
    221.     cursor.close()
    222.     conn.close()
    223.     # 可以获取指定数量的数据
    224.     #cursor.fetchmany(3)
    225.     # 光标按绝对位置移动1
    226.     #cursor.scroll(1, mode="absolute")
    227.     # 光标按照相对位置(当前位置)移动1
    228.     #cursor.scroll(1, mode="relative")
    230.     return ret
    233. # 数据上报
    234. def data_report(data):
    235.     headers = {'content-type': 'application/json'}
    236.     #request_url = "http://10.206.55.18:9527/api/v1.0/billings/collection"
    237.     request_url = "http://10.216.28.150:9527/api/v2.0/volumes/collection"
    239.     logger.info("Begin ------ 开始上报")
    240.     #for record in data:
    241.     logger.info("--------------- 分隔行 ----------------")
    242.     logger.info("上报数据：{}".format(data))
    244.     try:
    245.         response = requests.post(request_url, data=json.dumps(data), headers=headers, timeout=(3, 2))
    247.         logger.info("返回状态码：{}".format(response.status_code))
    248.         logger.info("返回信息：{}".format(json.loads(response.text)))
    249.     except Exception as e:
    250.         logger.error(e)
    252.     logger.info("End ------ 上报结束")
    254. if __name__ == "__main__":
    255.     result = report_hbase_data()
    257.     data_report(result)
    259. session.close()

    logger.py

    1. #!/usr/bin/env python
    2. # -*- coding: utf-8 -*
    5. import os
    6. import sys
    7. import time
    8. from loguru import logger
    10. """
    11. basedir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
    12. # print(basedir)
    14. # 定位到log日志文件
    15. log_path = os.path.join(basedir, 'logs')
    17. if not os.path.exists(log_path):
    18.     os.mkdir(log_path)
    20. log_path_error = os.path.join(log_path, f'error_{time.strftime("%Y-%m-%d")}.log')
    22. # 日志简单配置
    23. # 具体其他配置 可自行参考 https://github.com/Delgan/loguru
    24. logger.add(log_path_error, rotation="12:00", retention="5 days", enqueue=True)
    25. """
    27. class FastapiLoguru(object):
    28.     def __init__(self, env: str):
    29.         #self.base_path = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
    30.         self.base_path = os.getcwd()
    31.         self.log_path = os.path.join(self.base_path, 'logs')
    32.         if not os.path.exists(self.log_path):
    33.             os.mkdir(self.log_path)
    34.         self.log_file_path = os.path.join(
    35.             self.log_path,
    36.             f'log_{time.strftime("%Y-%m-%d")}.log'
    37.         )
    39.         #self.log_format = '<level>[{time:YYYY-MM-DD at HH:mm:ss}] {level} {message}</level>'
    40.         self.log_format = '[{time:YYYY-MM-DD at HH:mm:ss}] {level} {message}'
    42.         self._logger = logger
    43.         self._logger.remove()
    45.         if env == 'dev':
    46.             self._logger.add(
    47.                 sys.stdout,
    48.                 colorize=True,
    49.                 format=self.log_format,
    50.                 backtrace=True,
    51.                 diagnose=True,
    52.                 level='DEBUG'
    53.             )
    55.         if env == 'sit':
    56.             self._logger.add(
    57.                 self.log_file_path,
    58.                 colorize=True,
    59.                 format=self.log_format,
    60.                 rotation='50 MB',
    61.                 retention='7 days',
    62.                 enqueue=True,
    63.                 level='INFO'
    64.             )
    66.         if env == 'prd':
    67.             self._logger.add(
    68.                 self.log_file_path,
    69.                 colorize=True,
    70.                 format=self.log_format,
    71.                 rotation='100 MB',
    72.                 retention='30 days',
    73.                 enqueue=True,
    74.                 level='INFO'
    75.             )
    77.     def get_logger(self) -> logger:
    78.         return self._logger