ApplicationCDCMonitor.py 4.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687
  1. # Author renyupeng
  2. # coding=utf-8
  3. # @Time : 2021/10/20 3:39 下午
  4. # @Site :
  5. # @File : ApplicationCDCMonitor.py
  6. # @Software: PyCharm
  7. # @contact: renyupeng@c-top.com.cn
  8. # @Tel 1501435553
  9. import logging
  10. from Apietl.constant.ConfConstant import ConfConstant
  11. import requests
  12. import os
  13. class ApplicationCDCMonitor:
  14. def __init__(self):
  15. self.cdc_url = ConfConstant.CDC_URL
  16. self.pd_url = ConfConstant.TIDB_HOST
  17. self.cdc_port = 8300
  18. self.pd_port = 2379
  19. self.cdc_job = 'application'
  20. self.cdc_status_url = '/api/v1/status'
  21. self.cdc_health_url = 'GET /api/v1/health'
  22. self.cdc_changefeeds_url = '/api/v1/changefeeds'
  23. def Headler(self):
  24. headers = {'Content-Type': 'application/json'}
  25. repo = requests.get(url='http://{}:{}{}?'.format(self.cdc_url, self.cdc_port, self.cdc_changefeeds_url),
  26. headers=headers)
  27. errrepo = requests.get(
  28. url='http://{}:{}{}?state=error'.format(self.cdc_url, self.cdc_port, self.cdc_changefeeds_url),
  29. headers=headers)
  30. errtasks = errrepo.json()
  31. if errtasks is not None:
  32. for errtask in errtasks:
  33. errtask_id = errtask['id']
  34. print(errtask, '-----error--------')
  35. requests.delete(url='http://{}:{}{}/{}'.format(self.cdc_url, self.cdc_port,
  36. self.cdc_changefeeds_url, errtask_id))
  37. os.system(
  38. 'cd /data/tidb-deploy/cdc-8300/bin; ./cdc cli changefeed create --pd=http://192.168.0.184:2379 '
  39. '--sink-uri="mysql://data:hcst@2021@139.186.27.96:3390" --config ../conf/{task_id}.toml '
  40. '--changefeed-id={task_id} '.format(task_id=errtask_id))
  41. message = repo.json()
  42. # print(message)
  43. for json in message:
  44. task_id = json['id']
  45. tso = json['checkpoint_tso']
  46. error = json['error']
  47. if error is not None:
  48. requests.delete(url='http://{}:{}{}/{}'.format(self.cdc_url, self.cdc_port,
  49. self.cdc_changefeeds_url, task_id))
  50. status = os.system(
  51. 'cd /data/tidb-deploy/cdc-8300/bin; ./cdc cli changefeed create --pd=http://192.168.0.184:2379 '
  52. '--sink-uri="mysql://data:hcst@2021@139.186.27.96:3390" --config ../conf/{task_id}.toml '
  53. '--changefeed-id={task_id} --start-ts {tso}'.format(tso=tso, task_id=task_id))
  54. # print(status)
  55. backResult = requests.get(
  56. url='http://{}:{}{}/{}'.format(self.cdc_url, self.cdc_port, self.cdc_changefeeds_url, task_id),
  57. headers=headers)
  58. back = backResult.json()
  59. print(back)
  60. if back is None:
  61. os.system(
  62. 'cd /data/tidb-deploy/cdc-8300/bin; ./cdc cli changefeed create --pd=http://192.168.0.184:2379 '
  63. '--sink-uri="mysql://data:hcst@2021@139.186.27.96:3390" --config ../conf/{task_id}.toml '
  64. '--changefeed-id={task_id} '.format(task_id=task_id))
  65. else:
  66. if back['error_msg'] is not None:
  67. requests.delete(url='http://{}:{}{}/{}'.format(self.cdc_url, self.cdc_port,
  68. self.cdc_changefeeds_url, task_id))
  69. os.system(
  70. 'cd /data/tidb-deploy/cdc-8300/bin; ./cdc cli changefeed create '
  71. '--pd=http://192.168.0.184:2379 '
  72. '--sink-uri="mysql://data:hcst@2021@139.186.27.96:3390" --config ../conf/{task_id}.toml '
  73. '--changefeed-id={task_id} '.format(task_id=task_id))
  74. else:
  75. logging.info('新创建没有问题')
  76. else:
  77. logging.info("无异常")
  78. if __name__ == '__main__':
  79. ApplicationCDCMonitor().Headler()