Merge "Retry on EAI_AGAIN name resolution failures"
This commit is contained in:
commit
701fc86f81
|
@ -24,6 +24,7 @@ import os.path
|
||||||
import re
|
import re
|
||||||
import signal
|
import signal
|
||||||
import threading
|
import threading
|
||||||
|
import time
|
||||||
import yaml
|
import yaml
|
||||||
import zmq
|
import zmq
|
||||||
|
|
||||||
|
@ -164,10 +165,25 @@ class Server(object):
|
||||||
self.processors.append(log_processor)
|
self.processors.append(log_processor)
|
||||||
self.processors.append(subunit_processor)
|
self.processors.append(subunit_processor)
|
||||||
|
|
||||||
|
def wait_for_name_resolution(self, host, port):
|
||||||
|
while True:
|
||||||
|
try:
|
||||||
|
socket.getaddrinfo(host, port)
|
||||||
|
except socket.gaierror as e:
|
||||||
|
if e.errno == socket.EAI_AGAIN:
|
||||||
|
logging.debug("Temporary failure in name resolution")
|
||||||
|
time.sleep(2)
|
||||||
|
continue
|
||||||
|
else:
|
||||||
|
raise
|
||||||
|
break
|
||||||
|
|
||||||
def main(self):
|
def main(self):
|
||||||
statsd_host = os.environ.get('STATSD_HOST')
|
statsd_host = os.environ.get('STATSD_HOST')
|
||||||
statsd_port = int(os.environ.get('STATSD_PORT', 8125))
|
statsd_port = int(os.environ.get('STATSD_PORT', 8125))
|
||||||
statsd_prefix = os.environ.get('STATSD_PREFIX', 'logstash.geard')
|
statsd_prefix = os.environ.get('STATSD_PREFIX', 'logstash.geard')
|
||||||
|
if statsd_host:
|
||||||
|
self.wait_for_name_resolution(statsd_host, statsd_port)
|
||||||
self.gearserver = gear.Server(
|
self.gearserver = gear.Server(
|
||||||
statsd_host=statsd_host,
|
statsd_host=statsd_host,
|
||||||
statsd_port=statsd_port,
|
statsd_port=statsd_port,
|
||||||
|
|
|
@ -359,9 +359,23 @@ class Server(object):
|
||||||
logging.basicConfig(level=logging.CRITICAL)
|
logging.basicConfig(level=logging.CRITICAL)
|
||||||
logging.debug("Log pusher starting.")
|
logging.debug("Log pusher starting.")
|
||||||
|
|
||||||
|
def wait_for_name_resolution(self, host, port):
|
||||||
|
while True:
|
||||||
|
try:
|
||||||
|
socket.getaddrinfo(host, port)
|
||||||
|
except socket.gaierror as e:
|
||||||
|
if e.errno == socket.EAI_AGAIN:
|
||||||
|
logging.debug("Temporary failure in name resolution")
|
||||||
|
time.sleep(2)
|
||||||
|
continue
|
||||||
|
else:
|
||||||
|
raise
|
||||||
|
break
|
||||||
|
|
||||||
def setup_retriever(self):
|
def setup_retriever(self):
|
||||||
hostname = socket.gethostname()
|
hostname = socket.gethostname()
|
||||||
gearman_worker = gear.Worker(hostname + b'-pusher')
|
gearman_worker = gear.Worker(hostname + b'-pusher')
|
||||||
|
self.wait_for_name_resolution(self.gearman_host, self.gearman_port)
|
||||||
gearman_worker.addServer(self.gearman_host,
|
gearman_worker.addServer(self.gearman_host,
|
||||||
self.gearman_port)
|
self.gearman_port)
|
||||||
gearman_worker.registerFunction(b'push-log')
|
gearman_worker.registerFunction(b'push-log')
|
||||||
|
|
Loading…
Reference in New Issue