Gremlin Python - ошибка "Сервер отключен - попробуйте снова подключиться"

Question

У меня есть веб-приложение Flask, в котором я хочу поддерживать постоянное соединение с базой данных AWS Neptune graph. Это соединение устанавливается следующим образом:

from gremlin_python.process.anonymous_traversal import traversal
from gremlin_python.driver.driver_remote_connection import DriverRemoteConnection

neptune_endpt = 'db-instance-x.xxxxxxxxxx.xx-xxxxx-x.neptune.amazonaws.com'
remoteConn = DriverRemoteConnection(f'wss://{neptune_endpt}:8182/gremlin','g')
self.g = traversal().withRemote(remoteConn)

Проблема, с которой я сталкиваюсь, заключается в том, что соединение автоматически разрывается, если оно не используется, и я не могу найти способ определить, разорвалось ли соединение (чтобы я мог повторно подключиться, используя приведенный выше фрагмент кода).

Я видел аналогичную проблему: сервер Gremlin с удаленным подключением закрыто - как подключиться автоматически? однако и этот вопрос не имеет решения. На этот аналогичный вопрос тоже нет ответа.

Я пробовал следующие два решения (оба не сработали):

1. Я установил свое веб-приложение за четырьмя рабочими процессами Gunicorn с тайм-аутом в 100 секунд, надеясь, что перезапуск рабочего стола позаботится о тайм-аутах Gremlin.
2. Я попытался перехватить исключения, чтобы определить, разорвалось ли соединение. Каждый раз, когда я используюself.g Чтобы пройтись по моему графику, я пытаюсь "обновить" соединение, что я имею в виду:

def _refresh_neptune(self):
    try:
        self.g = traversal().withRemote(self.conn)
    except:
        self.conn = DriverRemoteConnection(f'wss://{neptune_endpt}:8182/gremlin','g')
        self.g = traversal().withRemote(self.conn)

Вот self.conn был инициализирован как:

self.conn = DriverRemoteConnection(f'wss://{neptune_endpt}:8182/gremlin','g')

Есть ли способ обойти эту ошибку подключения?

Благодарность

Обновление: добавлено сообщение об ошибке ниже:

  File "/home/ubuntu/.virtualenvs/rundev/lib/python3.6/site-packages/gremlin_python/process/traversal.py
", line 58, in toList
    return list(iter(self))
  File "/home/ubuntu/.virtualenvs/rundev/lib/python3.6/site-packages/gremlin_python/process/traversal.py
", line 48, in __next__
    self.traversal_strategies.apply_strategies(self)
  File "/home/ubuntu/.virtualenvs/rundev/lib/python3.6/site-packages/gremlin_python/process/traversal.py
", line 573, in apply_strategies
    traversal_strategy.apply(traversal)
  File "/home/ubuntu/.virtualenvs/rundev/lib/python3.6/site-packages/gremlin_python/driver/remote_connec
tion.py", line 149, in apply
    remote_traversal = self.remote_connection.submit(traversal.bytecode)
  File "/home/ubuntu/.virtualenvs/rundev/lib/python3.6/site-packages/gremlin_python/driver/driver_remote
_connection.py", line 56, in submit
    results = result_set.all().result()
  File "/usr/lib/python3.6/concurrent/futures/_base.py", line 425, in result
    return self.__get_result()
  File "/usr/lib/python3.6/concurrent/futures/_base.py", line 384, in __get_result
    raise self._exception
  File "/home/ubuntu/.virtualenvs/rundev/lib/python3.6/site-packages/gremlin_python/driver/resultset.py"
, line 90, in cb
    f.result()
  File "/usr/lib/python3.6/concurrent/futures/_base.py", line 425, in result
    return self.__get_result()
  File "/usr/lib/python3.6/concurrent/futures/_base.py", line 384, in __get_result
    raise self._exception
  File "/usr/lib/python3.6/concurrent/futures/thread.py", line 56, in run
    result = self.fn(*self.args, **self.kwargs)
  File "/home/ubuntu/.virtualenvs/rundev/lib/python3.6/site-packages/gremlin_python/driver/connection.py
", line 83, in _receive
    status_code = self._protocol.data_received(data, self._results)
  File "/home/ubuntu/.virtualenvs/rundev/lib/python3.6/site-packages/gremlin_python/driver/protocol.py",
 line 81, in data_received
    'message': 'Server disconnected - please try to reconnect', 'attributes': {}})
gremlin_python.driver.protocol.GremlinServerError: 500: Server disconnected - please try to reconnect

Answer 1

Я не уверен, что это лучший способ решить эту проблему, но я также использую gremlin-python и Neptune, и у меня была такая же проблема. Я обошел это, реализовав Transport, который вы можете предоставить DriverRemoteConnection.

DriverRemoteConnection(
    url=endpoint,
    traversal_source=self._traversal_source,
    transport_factory=Transport
)

gremlin-python возвращает соединения с пулом при исключении, а исключение, возвращаемое при закрытии соединения, - это GremlinServerError, которое также возникает для других ошибок.

gremlin_python / driver / connection.py#L69 -gremlin_python / driver / protocol.py#L80

Пользовательский транспорт такой же, как и TornadoTransport из gremlin-python, но методы чтения и записи расширены до:

• Повторно открывать закрытые соединения, если клиент веб-сокета закрыт
• Поднимите StreamClosedError, если клиент веб-сокета возвращает None из read_message

Мертвые соединения, которые добавляются обратно в пул, могут быть повторно открыты, а затем вы можете обработать StreamClosedError, чтобы применить некоторую логику повторных попыток. Я сделал это, переопределив методы submit и submitAsync в DriverRemoteConnection, но вы могли поймать и повторить попытку где угодно.

class Transport(AbstractBaseTransport):
    def __init__(self):
        self._ws = None
        self._loop = ioloop.IOLoop(make_current=False)
        self._url = None

        # Because the transport will try to reopen the underlying ws connection
        # track if the closed() method has been called to prevent the transport
        # from reopening.
        self._explicit_closed = True

    @property
    def closed(self):
        return not self._ws.protocol

    def connect(self, url, headers=None):
        self._forced_closed = False

        # Set the endpoint URL
        self._url = httpclient.HTTPRequest(url, headers=headers) if headers else url

        # Open the connection
        self._connect()

    def write(self, message):
        # Before writing, try to ensure that the connection is open.
        if self.closed:
            self._connect()

        self._loop.run_sync(lambda: self._ws.write_message(message, binary=True))

    def read(self):
        result = self._loop.run_sync(self._ws.read_message)

        # If the read call returns None, the stream has closed.
        if result is None:
            self._ws.close()  # Ensure we close the stream
            raise StreamClosedError()

        return result

    def close(self):
        self._ws.close()
        self._loop.close()
        self._explicit_closed = True

    def _connect(self):
        # If close() was called explicitly on the transport, don't allow
        # subsequent calls to write() to reopen the connection.
        if self._explicit_closed:
            raise TransportClosedError(
                "Transport has been closed and can not be reopened."
            )

        # Check if the ws is closed, if it is not, close it.
        if self._ws and not self.closed:
            self._ws.close()

        # Open the ws connection
        self._ws = self._loop.run_sync(
            lambda: websocket.websocket_connect(url=self._url)
        )


class TransportClosedError(Exception):
    pass

Это также будет работать с пулом соединений gremlin-pythons.

Если вам не нужен пул, альтернативный подход - установить размер пула равным 1 и реализовать некоторую форму поддержки активности, как это обсуждается здесь. ТИНКЕРПОП-2352

Похоже, что ping/keep-alive веб-сокета в gremlin-python еще не реализован TINKERPOP-1886.