# Author renyupeng # coding=utf-8 # @Time : 2021/10/20 3:39 下午 # @Site : # @File : ApplicationCDCMonitor.py # @Software: PyCharm # @contact: renyupeng@c-top.com.cn # @Tel 1501435553 import logging from Apietl.constant.ConfConstant import ConfConstant import requests import os class 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()