def __call__(self, *args, **kwargs):
""" Call the method and handle the AutoReconnect exception gracefully """
start_time = time.time()
for attempt in count():
try:
return self._method(*args, **kwargs)
except AutoReconnect:
duration = time.time() - start_time
if duration >= WAIT_TIME:
break
logger.warning(
'Reconnecting to MongoDB, attempt {} ({:.3f} seconds elapsed)'.
format(attempt, duration))
time.sleep(self.calc_sleep(attempt))
return self._method(*args, **kwargs)
python类AutoReconnect()的实例源码
def _receive_data_on_socket(sock, length):
msg = b""
while length:
try:
chunk = sock.recv(length)
except (IOError, OSError) as exc:
err = None
if hasattr(exc, 'errno'):
err = exc.errno
elif exc.args:
err = exc.args[0]
if err == errno.EINTR:
continue
raise
if chunk == b"":
raise AutoReconnect("connection closed")
length -= len(chunk)
msg += chunk
return msg
def search_mongodb(url, times=0):
"""???????????mongodb?"""
try:
ret = db.weixin.find_one({'url': url})
if not ret:
return True
else:
return False
except AutoReconnect:
times += 1
if times <= 5:
print('?????????????mongodb')
search_mongodb(url, times)
else:
print('mongodb????')
return False
def _connect(self):
"""Connecting to mongo database."""
try:
self.connection = Connection(host=self.host, port=self.port)
except AutoReconnect, e:
if self.fail_silently:
return
else:
raise AutoReconnect(e)
self.db = self.connection[self.database_name]
if self.username is not None and self.password is not None:
self.authenticated = self.db.authenticate(self.username, self.password)
if self.collection_name not in self.db.collection_names():
self.db.create_collection(self.collection_name,size=512000000,capped=True,max=200)
self.collection = self.db[self.collection_name]
def autoreconnect_retry(func, retries=3):
"""Decorating checking connection to the database."""
def db_op_wrapper(*args, **kwargs):
"""Decorator wrapper"""
tries = 0
while tries < retries:
try:
return func(*args, **kwargs)
except AutoReconnect:
tries += 1
raise Exception(
"Couldn't connect to the database, even after %d retries" % retries)
return db_op_wrapper
def _receive_data_on_socket(sock, length):
msg = b""
while length:
try:
chunk = sock.recv(length)
except (IOError, OSError) as exc:
err = None
if hasattr(exc, 'errno'):
err = exc.errno
elif exc.args:
err = exc.args[0]
if err == errno.EINTR:
continue
raise
if chunk == b"":
raise AutoReconnect("connection closed")
length -= len(chunk)
msg += chunk
return msg
def _receive_data_on_socket(sock, length):
msg = b""
while length:
try:
chunk = sock.recv(length)
except (IOError, OSError) as exc:
err = None
if hasattr(exc, 'errno'):
err = exc.errno
elif exc.args:
err = exc.args[0]
if err == errno.EINTR:
continue
raise
if chunk == b"":
raise AutoReconnect("connection closed")
length -= len(chunk)
msg += chunk
return msg
def _receive_data_on_socket(sock, length):
msg = b""
while length:
try:
chunk = sock.recv(length)
except (IOError, OSError) as exc:
err = None
if hasattr(exc, 'errno'):
err = exc.errno
elif exc.args:
err = exc.args[0]
if err == errno.EINTR:
continue
raise
if chunk == b"":
raise AutoReconnect("connection closed")
length -= len(chunk)
msg += chunk
return msg
def _receive_data_on_socket(sock, length):
msg = b""
while length:
try:
chunk = sock.recv(length)
except (IOError, OSError) as exc:
err = None
if hasattr(exc, 'errno'):
err = exc.errno
elif exc.args:
err = exc.args[0]
if err == errno.EINTR:
continue
raise
if chunk == b"":
raise AutoReconnect("connection closed")
length -= len(chunk)
msg += chunk
return msg
def __send_message(self, msg):
"""Send a getmore message and handle the response.
"""
client = self.__collection.database.connection
try:
res = client._send_message_with_response(
msg, _connection_to_use=self.__conn_id)
self.__conn_id, (response, dummy0, dummy1) = res
except AutoReconnect:
# Don't try to send kill cursors on another socket
# or to another server. It can cause a _pinValue
# assertion on some server releases if we get here
# due to a socket timeout.
self.__killed = True
raise
try:
response = helpers._unpack_response(response,
self.__id,
*self.__decode_opts)
except CursorNotFound:
self.__killed = True
raise
except AutoReconnect:
# Don't send kill cursors to another server after a "not master"
# error. It's completely pointless.
self.__killed = True
client.disconnect()
raise
self.__id = response["cursor_id"]
assert response["starting_from"] == self.__retrieved, (
"Result batch started from %s, expected %s" % (
response['starting_from'], self.__retrieved))
self.__retrieved += response["number_returned"]
self.__data = deque(response["data"])
master_slave_connection.py 文件源码
项目：noc-orchestrator
作者: DirceuSilvaLabs
项目源码
文件源码
阅读 17
收藏 0
点赞 0
评论 0
def _send_message_with_response(self, message, _connection_to_use=None,
_must_use_master=False, **kwargs):
"""Receive a message from Mongo.
Sends the given message and returns a (connection_id, response) pair.
:Parameters:
- `operation`: opcode of the message to send
- `data`: data to send
"""
if _connection_to_use is not None:
if _connection_to_use == -1:
member = self.__master
conn = -1
else:
member = self.__slaves[_connection_to_use]
conn = _connection_to_use
return (conn,
member._send_message_with_response(message, **kwargs)[1])
# _must_use_master is set for commands, which must be sent to the
# master instance. any queries in a request must be sent to the
# master since that is where writes go.
if _must_use_master or self.in_request():
return (-1, self.__master._send_message_with_response(message,
**kwargs)[1])
# Iterate through the slaves randomly until we have success. Raise
# reconnect if they all fail.
for connection_id in helpers.shuffled(xrange(len(self.__slaves))):
try:
slave = self.__slaves[connection_id]
return (connection_id,
slave._send_message_with_response(message,
**kwargs)[1])
except AutoReconnect:
pass
raise AutoReconnect("failed to connect to slaves")
def __send_message(self, msg):
"""Send a getmore message and handle the response.
"""
client = self.__collection.database.connection
try:
res = client._send_message_with_response(
msg, _connection_to_use=self.__conn_id)
self.__conn_id, (response, dummy0, dummy1) = res
except AutoReconnect:
# Don't try to send kill cursors on another socket
# or to another server. It can cause a _pinValue
# assertion on some server releases if we get here
# due to a socket timeout.
self.__killed = True
raise
try:
response = helpers._unpack_response(response,
self.__id,
*self.__decode_opts)
except CursorNotFound:
self.__killed = True
raise
except AutoReconnect:
# Don't send kill cursors to another server after a "not master"
# error. It's completely pointless.
self.__killed = True
client.disconnect()
raise
self.__id = response["cursor_id"]
assert response["starting_from"] == self.__retrieved, (
"Result batch started from %s, expected %s" % (
response['starting_from'], self.__retrieved))
self.__retrieved += response["number_returned"]
self.__data = deque(response["data"])
master_slave_connection.py 文件源码
项目：noc-orchestrator
作者: DirceuSilvaLabs
项目源码
文件源码
阅读 17
收藏 0
点赞 0
评论 0
def _send_message_with_response(self, message, _connection_to_use=None,
_must_use_master=False, **kwargs):
"""Receive a message from Mongo.
Sends the given message and returns a (connection_id, response) pair.
:Parameters:
- `operation`: opcode of the message to send
- `data`: data to send
"""
if _connection_to_use is not None:
if _connection_to_use == -1:
member = self.__master
conn = -1
else:
member = self.__slaves[_connection_to_use]
conn = _connection_to_use
return (conn,
member._send_message_with_response(message, **kwargs)[1])
# _must_use_master is set for commands, which must be sent to the
# master instance. any queries in a request must be sent to the
# master since that is where writes go.
if _must_use_master or self.in_request():
return (-1, self.__master._send_message_with_response(message,
**kwargs)[1])
# Iterate through the slaves randomly until we have success. Raise
# reconnect if they all fail.
for connection_id in helpers.shuffled(xrange(len(self.__slaves))):
try:
slave = self.__slaves[connection_id]
return (connection_id,
slave._send_message_with_response(message,
**kwargs)[1])
except AutoReconnect:
pass
raise AutoReconnect("failed to connect to slaves")
def __send_message(self, msg):
"""Send a getmore message and handle the response.
"""
client = self.__collection.database.connection
try:
res = client._send_message_with_response(
msg, _connection_to_use=self.__conn_id)
self.__conn_id, (response, dummy0, dummy1) = res
except AutoReconnect:
# Don't try to send kill cursors on another socket
# or to another server. It can cause a _pinValue
# assertion on some server releases if we get here
# due to a socket timeout.
self.__killed = True
raise
try:
response = helpers._unpack_response(response,
self.__id,
*self.__decode_opts)
except CursorNotFound:
self.__killed = True
raise
except AutoReconnect:
# Don't send kill cursors to another server after a "not master"
# error. It's completely pointless.
self.__killed = True
client.disconnect()
raise
self.__id = response["cursor_id"]
assert response["starting_from"] == self.__retrieved, (
"Result batch started from %s, expected %s" % (
response['starting_from'], self.__retrieved))
self.__retrieved += response["number_returned"]
self.__data = deque(response["data"])
def __send_message(self, msg):
"""Send a getmore message and handle the response.
"""
client = self.__collection.database.connection
try:
res = client._send_message_with_response(
msg, _connection_to_use=self.__conn_id)
self.__conn_id, (response, dummy0, dummy1) = res
except AutoReconnect:
# Don't try to send kill cursors on another socket
# or to another server. It can cause a _pinValue
# assertion on some server releases if we get here
# due to a socket timeout.
self.__killed = True
raise
try:
response = helpers._unpack_response(response,
self.__id,
*self.__decode_opts)
except CursorNotFound:
self.__killed = True
raise
except AutoReconnect:
# Don't send kill cursors to another server after a "not master"
# error. It's completely pointless.
self.__killed = True
client.disconnect()
raise
self.__id = response["cursor_id"]
assert response["starting_from"] == self.__retrieved, (
"Result batch started from %s, expected %s" % (
response['starting_from'], self.__retrieved))
self.__retrieved += response["number_returned"]
self.__data = deque(response["data"])
master_slave_connection.py 文件源码
项目：noc-orchestrator
作者: DirceuSilvaLabs
项目源码
文件源码
阅读 21
收藏 0
点赞 0
评论 0
def _send_message_with_response(self, message, _connection_to_use=None,
_must_use_master=False, **kwargs):
"""Receive a message from Mongo.
Sends the given message and returns a (connection_id, response) pair.
:Parameters:
- `operation`: opcode of the message to send
- `data`: data to send
"""
if _connection_to_use is not None:
if _connection_to_use == -1:
member = self.__master
conn = -1
else:
member = self.__slaves[_connection_to_use]
conn = _connection_to_use
return (conn,
member._send_message_with_response(message, **kwargs)[1])
# _must_use_master is set for commands, which must be sent to the
# master instance. any queries in a request must be sent to the
# master since that is where writes go.
if _must_use_master or self.in_request():
return (-1, self.__master._send_message_with_response(message,
**kwargs)[1])
# Iterate through the slaves randomly until we have success. Raise
# reconnect if they all fail.
for connection_id in helpers.shuffled(xrange(len(self.__slaves))):
try:
slave = self.__slaves[connection_id]
return (connection_id,
slave._send_message_with_response(message,
**kwargs)[1])
except AutoReconnect:
pass
raise AutoReconnect("failed to connect to slaves")
def remove_host(module, client, host_name, timeout=180):
while True:
try:
admin_db = client['admin']
local_db = client['local']
if local_db.system.replset.count() > 1:
module.fail_json(msg='local.system.replset has unexpected contents')
cfg = local_db.system.replset.find_one()
if not cfg:
module.fail_json(msg='no config object retrievable from local.system.replset')
cfg['version'] += 1
if len(cfg['members']) == 1:
module.fail_json(msg="You can't delete last member of replica set")
for member in cfg['members']:
if host_name in member['host']:
cfg['members'].remove(member)
else:
fail_msg = "couldn't find member with hostname: {0} in replica set members list".format(host_name)
module.fail_json(msg=fail_msg)
except (OperationFailure, AutoReconnect) as e:
timeout = timeout - 5
if timeout <= 0:
module.fail_json(msg='reached timeout while waiting for rs.reconfig(): %s' % str(e))
time.sleep(5)
def insert_mongodb(item, times=0):
"""????mongodb?"""
try:
db.weixin.insert(item)
except AutoReconnect:
times += 1
if times <= 5:
print('?????????????mongodb')
insert_mongodb(item, times)
else:
print('mongodb????')
def __send_message(self, msg):
"""Send a getmore message and handle the response.
"""
client = self.__collection.database.connection
try:
res = client._send_message_with_response(
msg, _connection_to_use=self.__conn_id)
self.__conn_id, (response, dummy0, dummy1) = res
except AutoReconnect:
# Don't try to send kill cursors on another socket
# or to another server. It can cause a _pinValue
# assertion on some server releases if we get here
# due to a socket timeout.
self.__killed = True
raise
try:
response = helpers._unpack_response(response,
self.__id,
*self.__decode_opts)
except CursorNotFound:
self.__killed = True
raise
except AutoReconnect:
# Don't send kill cursors to another server after a "not master"
# error. It's completely pointless.
self.__killed = True
client.disconnect()
raise
self.__id = response["cursor_id"]
assert response["starting_from"] == self.__retrieved, (
"Result batch started from %s, expected %s" % (
response['starting_from'], self.__retrieved))
self.__retrieved += response["number_returned"]
self.__data = deque(response["data"])
def _send_message_with_response(self, message, _connection_to_use=None,
_must_use_master=False, **kwargs):
"""Receive a message from Mongo.
Sends the given message and returns a (connection_id, response) pair.
:Parameters:
- `operation`: opcode of the message to send
- `data`: data to send
"""
if _connection_to_use is not None:
if _connection_to_use == -1:
member = self.__master
conn = -1
else:
member = self.__slaves[_connection_to_use]
conn = _connection_to_use
return (conn,
member._send_message_with_response(message, **kwargs)[1])
# _must_use_master is set for commands, which must be sent to the
# master instance. any queries in a request must be sent to the
# master since that is where writes go.
if _must_use_master or self.in_request():
return (-1, self.__master._send_message_with_response(message,
**kwargs)[1])
# Iterate through the slaves randomly until we have success. Raise
# reconnect if they all fail.
for connection_id in helpers.shuffled(xrange(len(self.__slaves))):
try:
slave = self.__slaves[connection_id]
return (connection_id,
slave._send_message_with_response(message,
**kwargs)[1])
except AutoReconnect:
pass
raise AutoReconnect("failed to connect to slaves")
def __init__(self, obj, methods):
""" Initialize the MongoReconnectProxy.
Args:
obj: The object for which all calls should be wrapped in the AutoReconnect
exception handling block.
methods (set): The list of method names that should be wrapped.
"""
self._unproxied_object = obj
self._methods = methods
def ensure_mongo(func):
@wraps(func)
def func_wrapper(*args, **kwargs):
client = MongoClient(serverSelectionTimeoutMS=500, connectTimeoutMS=500)
try:
# The ismaster command is cheap and does not require auth.
client.admin.command('ismaster')
except (errors.ServerSelectionTimeoutError, errors.AutoReconnect):
raise MinionMongoError("Can't connect to mongodb")
else:
return func(*args, **kwargs)
finally:
client.close()
return func_wrapper
def remove_host(module, client, host_name, timeout=180):
while True:
try:
admin_db = client['admin']
local_db = client['local']
if local_db.system.replset.count() > 1:
module.fail_json(msg='local.system.replset has unexpected contents')
cfg = local_db.system.replset.find_one()
if not cfg:
module.fail_json(msg='no config object retrievable from local.system.replset')
cfg['version'] += 1
if len(cfg['members']) == 1:
module.fail_json(msg="You can't delete last member of replica set")
for member in cfg['members']:
if host_name in member['host']:
cfg['members'].remove(member)
else:
fail_msg = "couldn't find member with hostname: {0} in replica set members list".format(host_name)
module.fail_json(msg=fail_msg)
except (OperationFailure, AutoReconnect), e:
timeout = timeout - 5
if timeout <= 0:
module.fail_json(msg='reached timeout while waiting for rs.reconfig(): %s' % str(e))
time.sleep(5)
def _connect(self):
self._conn = motor.motor_asyncio.AsyncIOMotorClient(
self._build_connection_string(),
io_loop=self.loop
)
try:
await self.wait_db()
except AutoReconnect as e:
self.logger.error("Couldn't connect to db %s", self.fingerprint)
await self.wait_db()
def mongo_wait_connected_on_coro(coro, *args, **kwargs):
retry = 0
while True:
retry += 1
try:
res = await coro(*args, **kwargs)
if retry > 1:
print('Restored mongo connection in {}'.format(coro))
return res
except AutoReconnect:
print('Waiting for mongo connection in {}'.format(coro))
await asyncio.sleep(1.0)
def rs_reconfigure(client, rs_config):
try:
client.admin.command('replSetReconfig', rs_config)
except AutoReconnect:
pass
def add_host(module, client, host_name, host_port, host_type, timeout=180, **kwargs):
while True:
try:
admin_db = client['admin']
local_db = client['local']
if local_db.system.replset.count() > 1:
module.fail_json(msg='local.system.replset has unexpected contents')
cfg = local_db.system.replset.find_one()
if not cfg:
module.fail_json(msg='no config object retrievable from local.system.replset')
cfg['version'] += 1
max_id = max(cfg['members'], key=lambda x:x['_id'])
new_host = { '_id': max_id['_id'] + 1, 'host': "{0}:{1}".format(host_name, host_port) }
if host_type == 'arbiter':
new_host['arbiterOnly'] = True
if not kwargs['build_indexes']:
new_host['buildIndexes'] = False
if kwargs['hidden']:
new_host['hidden'] = True
if kwargs['priority'] != 1.0:
new_host['priority'] = kwargs['priority']
if kwargs['slave_delay'] != 0:
new_host['slaveDelay'] = kwargs['slave_delay']
if kwargs['votes'] != 1:
new_host['votes'] = kwargs['votes']
cfg['members'].append(new_host)
admin_db.command('replSetReconfig', cfg)
return
except (OperationFailure, AutoReconnect) as e:
timeout = timeout - 5
if timeout <= 0:
module.fail_json(msg='reached timeout while waiting for rs.reconfig(): %s' % str(e))
time.sleep(5)
def add_host(module, client, host_name, host_port, host_type, timeout=180, **kwargs):
while True:
try:
admin_db = client['admin']
local_db = client['local']
if local_db.system.replset.count() > 1:
module.fail_json(msg='local.system.replset has unexpected contents')
cfg = local_db.system.replset.find_one()
if not cfg:
module.fail_json(msg='no config object retrievable from local.system.replset')
cfg['version'] += 1
max_id = max(cfg['members'], key=lambda x:x['_id'])
new_host = { '_id': max_id['_id'] + 1, 'host': "{0}:{1}".format(host_name, host_port) }
if host_type == 'arbiter':
new_host['arbiterOnly'] = True
if not kwargs['build_indexes']:
new_host['buildIndexes'] = False
if kwargs['hidden']:
new_host['hidden'] = True
if kwargs['priority'] != 1.0:
new_host['priority'] = kwargs['priority']
if kwargs['slave_delay'] != 0:
new_host['slaveDelay'] = kwargs['slave_delay']
if kwargs['votes'] != 1:
new_host['votes'] = kwargs['votes']
cfg['members'].append(new_host)
admin_db.command('replSetReconfig', cfg)
return
except (OperationFailure, AutoReconnect), e:
timeout = timeout - 5
if timeout <= 0:
module.fail_json(msg='reached timeout while waiting for rs.reconfig(): %s' % str(e))
time.sleep(5)
