failover_client_sentinel.py

import argparse
import time
import numpy as np
from datetime import datetime
from redis.sentinel import (Sentinel, MasterNotFoundError)
from redis import exceptions

# this client is based on Yossi failover_client
class TestClientSentinel(object):
    def __init__(self, args):
        self.connect_timeout = args.connect_timeout
        self.connect_retry_interval = args.connect_retry_interval
        self.heartbeat_timeout = args.heartbeat_timeout
        self.heartbeat_interval = args.heartbeat_interval
        self.sock = None
        self.last_pong_time = None
        self.dbname = args.dbname
        self.hosts = [str(item) for item in args.hosts.split(',')]
        self.ports = [int(item) for item in args.ports.split(',')]
        sentinel_list = np.column_stack((self.hosts, self.ports))
        self.sentinel = Sentinel(sentinel_list, socket_timeout=0.1)
        self.master = self.sentinel.master_for(self.dbname, socket_timeout=self.heartbeat_timeout)
#        self.sentinel_list = [[0 for x in range(args.hosts)] for i in range(args.hosts)]
#        print(self.sentinel_list)

    @staticmethod
    def log_event(text):
        print '[{}] {}'.format(datetime.now().strftime('%d-%b-%g %H:%M:%S.%f', ), text)

    def connect(self):
        while True:
            try:
                self.log_event('[I] Trying to connect %s (%s), port %s' % (self.sentinel.master_for(self.dbname), "3", 222))
                self.master = self.sentinel.master_for(self.dbname, socket_timeout=self.heartbeat_timeout)
                print(self.master)
            except (exceptions.ConnectionError, exceptions.TimeoutError) as err:
                self.log_event('[E] %s' % str(err))
            if self.master is not None:
                self.log_event('[I] Connection established.')

                return
            time.sleep(self.connect_retry_interval)

    def heartbeat(self):
        responses = 0
        while True:
            try:

                response = self.master.ping()

                if not response:
                    self.log_event('[E] Server connection dropped')
                    break
                if not response:
                    self.log_event('[E] Unexpected protocol response: %s' % response)
                    break
            except (exceptions.ConnectionError, exceptions.TimeoutError) as err:
                self.log_event('[E] %s' % str(err))
                break

            now = time.time()
            if responses == 0 and self.last_pong_time:
                self.log_event('[I] First successful response, %.2f after previous one' % (
                    now - self.last_pong_time))
            responses += 1
            self.last_pong_time = now
            time.sleep(self.heartbeat_interval)

    def run(self):
        while True:
            self.connect()
            self.heartbeat()
            time.sleep(1)
#            try:
#                self.sock.shutdown(socket.SHUT_RDWR)
#            except socket.error:
#                pass
#            self.sock = None


def main():
    parser = argparse.ArgumentParser(
        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
        description='Redis failover test client')
    parser.add_argument(
        '--hosts', type=str, help='Servers address',
        default='localhost')
    parser.add_argument(
        '--ports', type=str, help='Server port',
        default=6379)
    parser.add_argument(
        '--dbname', type=str, help='db name',
        default='mymaster')
    parser.add_argument(
        '--connect-timeout', type=int, help='Timeout (secs) for individual connect attempts',
        default=5)
    parser.add_argument(
        '--connect-retry-interval', type=float, help='Connect (secs) retry interval time',
        default=0.5)
    parser.add_argument(
        '--heartbeat-timeout', type=float, help='PING heartbeat  timeout (secs)',
        default=3)
    parser.add_argument(
        '--heartbeat-interval', type=float, help='PING heartbeat interval time (secs)',
        default=1)
    args = parser.parse_args()

    TestClientSentinel(args).run()


if __name__ == '__main__':
    main()