batcave.py 4.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130
  1. #!/usr/bin/python
  2. # -*- coding: utf-8 -*-
  3. from __future__ import print_function
  4. import argparse
  5. import daemon
  6. import logging
  7. import sys
  8. import time
  9. import threading
  10. from ffstatus import \
  11. dict_merge, merge_alfred_batman, \
  12. ApiServer, \
  13. AlfredParser, BatmanParser, \
  14. DashingClient, GraphitePush, \
  15. Storage
  16. from ffstatus.exceptions import SanityCheckError
  17. DEFAULT_INTERVAL = 15
  18. parser = argparse.ArgumentParser(description='Batman/Alfred Transmission Collection, Aggregation & Value Engine')
  19. parser.add_argument('--logfile', help='path for log file')
  20. parser.add_argument('--interval', type=int, default=DEFAULT_INTERVAL, help='data poll interval')
  21. parser.add_argument('-v', '--verbose', action='store_true', help='increase output verbosity')
  22. parser.add_argument('-d', '--no-detach', action='store_true', help='Don\'t detach (daemonize) ourself')
  23. parser.add_argument('-n', '--no-send', action='store_true', help='Fetch data but don\'t send it')
  24. parser.add_argument('-A', '--alfred-json', help='executable path for alfred-json')
  25. parser.add_argument('-B', '--batadv-vis', help='executable path for batadv-vis')
  26. parser.add_argument('-G', '--graphite-host', help='Graphite host')
  27. parser.add_argument('--graphite-port', type=int, default=2003, help='Graphite port')
  28. parser.add_argument('--dashing-url', help='Dashing URL')
  29. parser.add_argument('--dashing-token', help='Dashing\'s secret update token')
  30. parser.add_argument('--api-bind-host', default='', help='API-Server Hostname')
  31. parser.add_argument('--api-bind-port', type=int, default=8888, help='API-Server Port')
  32. parser.add_argument('-S', '--storage-dir', default='.', help='Path where to store data')
  33. args = parser.parse_args()
  34. if args.interval < 5:
  35. print('A poll interval lower than 5s is not supported.')
  36. sys.exit(1)
  37. shall_daemonize = not args.no_detach
  38. logger = logging.getLogger()
  39. logger.setLevel(logging.DEBUG if args.verbose else logging.INFO)
  40. if not args.logfile is None:
  41. fh = logging.FileHandler(args.logfile)
  42. fh.setFormatter(logging.Formatter('%(asctime)s [%(levelname)s] %(message)s', '%Y-%m-%d %H:%M:%S'))
  43. logger.addHandler(fh)
  44. if args.no_detach:
  45. ch = logging.StreamHandler(sys.stdout)
  46. ch.setFormatter(logging.Formatter('%(asctime)s [%(levelname)s] %(message)s', '%Y-%m-%d %H:%M:%S'))
  47. logger.addHandler(ch)
  48. logger.info('Starting up')
  49. storage = Storage(args.storage_dir)
  50. storage.open()
  51. logger.info('Storage: ' + str(storage))
  52. a = AlfredParser()
  53. b = BatmanParser()
  54. d = DashingClient(args.dashing_url, args.dashing_token) if not args.dashing_url is None else None
  55. g = GraphitePush(args.graphite_host, args.graphite_port) if not args.graphite_host is None else None
  56. if args.no_send:
  57. if not g is None: g.dont_send = True
  58. if not args.alfred_json is None: a.alfred_json = args.alfred_json
  59. if not args.batadv_vis is None: b.batadv_vis = args.batadv_vis
  60. logger.debug('Configured A.L.F.R.E.D. source: ' + str(a))
  61. logger.debug('Configured B.A.T.M.A.N. source: ' + str(b))
  62. logger.debug('Configured Dashing: ' + str(d))
  63. logger.debug('Configured Graphite: ' + str(g))
  64. # execute sanitycheck() where possible
  65. for i in [('AlfredParser', a), ('BatmanParser', b)]:
  66. try:
  67. i[1].sanitycheck()
  68. except SanityCheckError as err:
  69. logger.critical(i[0] + '.sanitycheck() failed: ' + str(err))
  70. print('FAILED SANITY CHECK: ' + str(err))
  71. sys.exit(1)
  72. server = ApiServer((args.api_bind_host, args.api_bind_port), storage)
  73. server_thread = threading.Thread(target=server.serve_forever)
  74. server_thread.daemon = True # exit thread when main thread terminates
  75. server_thread.start()
  76. logger.info('Started server: ' + str(server))
  77. if shall_daemonize:
  78. daemon_context = daemon.DaemonContext(
  79. files_preserve=[fh.stream],
  80. )
  81. daemon_context.open()
  82. while True:
  83. try:
  84. ts = int(time.time())
  85. logger.debug('Step 1/3: Fetching data ...')
  86. alfreddata = a.fetch()
  87. batmandata = b.fetch()
  88. newdata = merge_alfred_batman(alfreddata, batmandata)
  89. logger.debug('Fetched data: {0} ALFRED with {1} BATMAN makes {2} total'.format(len(alfreddata), len(batmandata), len(newdata)))
  90. logger.debug('Step 2/3: Pushing update data ...')
  91. if not g is None:
  92. graphitedata = g.push(newdata, ts=ts)
  93. logger.info('Sent ' + str(graphitedata.count('\n')+1) + ' lines to Graphite.')
  94. if not d is None:
  95. d.push(newdata)
  96. logger.debug('Step 3/3: Merging current data ...')
  97. storage.merge_new_data(newdata)
  98. logger.debug('I have data for %d nodes.', len(storage.data))
  99. except Exception as err:
  100. logger.error(str(err))
  101. logger.debug('Sleeping for {0} seconds'.format(args.interval))
  102. time.sleep(args.interval)
  103. storage.close()
  104. logger.info('Shut down.')