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 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_msg'] 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()
|