|
@@ -0,0 +1,47 @@
|
|
|
+# 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
|
|
|
+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)
|
|
|
+ message = repo.json()
|
|
|
+ print(message)
|
|
|
+ for json in message:
|
|
|
+ task_id = json['id']
|
|
|
+ state = json['state']
|
|
|
+ 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))
|
|
|
+
|
|
|
+ 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/application.toml '
|
|
|
+ '--changefeed-id=application-task --start-ts {tso}'.format(tso=tso))
|
|
|
+
|
|
|
+
|
|
|
+if __name__ == '__main__':
|
|
|
+ ApplicationCDCMonitor().Headler()
|