| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687 | # Author renyupeng# coding=utf-8# @Time    : 2021/10/20 3:39 下午# @Site    : # @File    : ApplicationCDCMonitor.py# @Software: PyCharm# @contact: renyupeng@c-top.com.cn# @Tel 1501435553import loggingfrom Apietl.constant.ConfConstant import ConfConstantimport requestsimport osclass ApplicationCDCMonitor:    def __init__(self):        self.cdc_url = ConfConstant.CDC_URL        self.pd_url = ConfConstant.TIDB_HOST        self.cdc_port = 8300        self.pd_port = 2379        self.cdc_job = 'application'        self.cdc_status_url = '/api/v1/status'        self.cdc_health_url = 'GET /api/v1/health'        self.cdc_changefeeds_url = '/api/v1/changefeeds'    def Headler(self):        headers = {'Content-Type': 'application/json'}        repo = requests.get(url='http://{}:{}{}?'.format(self.cdc_url, self.cdc_port, self.cdc_changefeeds_url),                            headers=headers)        errrepo = requests.get(            url='http://{}:{}{}?state=error'.format(self.cdc_url, self.cdc_port, self.cdc_changefeeds_url),            headers=headers)        errtasks = errrepo.json()        if errtasks is not None:            for errtask in errtasks:                errtask_id = errtask['id']                print(errtask, '-----error--------')                requests.delete(url='http://{}:{}{}/{}'.format(self.cdc_url, self.cdc_port,                                                               self.cdc_changefeeds_url, errtask_id))                os.system(                    'cd /data/tidb-deploy/cdc-8300/bin; ./cdc cli changefeed create --pd=http://192.168.0.184:2379 '                    '--sink-uri="mysql://data:hcst@2021@139.186.27.96:3390" --config ../conf/{task_id}.toml '                    '--changefeed-id={task_id} '.format(task_id=errtask_id))        message = repo.json()        # print(message)        for json in message:            task_id = json['id']            tso = json['checkpoint_tso']            error = json['error']            if error is not None:                requests.delete(url='http://{}:{}{}/{}'.format(self.cdc_url, self.cdc_port,                                                               self.cdc_changefeeds_url, task_id))                status = os.system(                    'cd /data/tidb-deploy/cdc-8300/bin; ./cdc cli changefeed create --pd=http://192.168.0.184:2379 '                    '--sink-uri="mysql://data:hcst@2021@139.186.27.96:3390" --config ../conf/{task_id}.toml '                    '--changefeed-id={task_id} --start-ts {tso}'.format(tso=tso, task_id=task_id))                # print(status)                backResult = requests.get(                    url='http://{}:{}{}/{}'.format(self.cdc_url, self.cdc_port, self.cdc_changefeeds_url, task_id),                    headers=headers)                back = backResult.json()                print(back)                if back is None:                    os.system(                        'cd /data/tidb-deploy/cdc-8300/bin; ./cdc cli changefeed create --pd=http://192.168.0.184:2379 '                        '--sink-uri="mysql://data:hcst@2021@139.186.27.96:3390" --config ../conf/{task_id}.toml '                        '--changefeed-id={task_id} '.format(task_id=task_id))                else:                    if back['error'] is not None:                        requests.delete(url='http://{}:{}{}/{}'.format(self.cdc_url, self.cdc_port,                                                                       self.cdc_changefeeds_url, task_id))                        os.system(                            'cd /data/tidb-deploy/cdc-8300/bin; ./cdc cli changefeed create '                            '--pd=http://192.168.0.184:2379 '                            '--sink-uri="mysql://data:hcst@2021@139.186.27.96:3390" --config ../conf/{task_id}.toml '                            '--changefeed-id={task_id} '.format(task_id=task_id))                    else:                        logging.info('新创建没有问题')            else:                logging.info("无异常")if __name__ == '__main__':    ApplicationCDCMonitor().Headler()
 |