server.py 文件源码

def _create_healthcheck(self, port):

        class HealthcheckHTTPRequestHandler(BaseHTTPRequestHandler):
            def do_GET(self):
                health = {'status': 'ok'}
                self.send_response(200)
                self.send_header('Content-Type', 'application/json')
                self.end_headers()
                self.wfile.write(bytes(json.dumps(health), 'UTF-8'))

            def log_message(self, format, *args):
                logger.debug("Healthcheck from %s %s" % (self.address_string(), format % args))

        self.healthcheck_http_server = HTTPServer(('', port), HealthcheckHTTPRequestHandler)

        healthcheck_thread = Thread(target=self._run_healthcheck_thread,
                                    name='healthcheck', args=(), daemon=True)
        healthcheck_thread.start()