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