consumer-scheduler/ip_watchdog.py

#!/usr/bin/env python3
from typing import Tuple
import logging
import socket

from redis_super_storage import RedisSuperStorage


class IPWatchdog:
    """
    This is very simple class, that is used to determine if the ip address of the host have changed.

    Internally this class relies on `RedisSuperStorage` to fetch the last used ip address.

    The ip address of the current host is acquired using python's builtin `socket` interface, by requesting an address resolve agains the localhost's host name.
    In some scenarios this may result in wrongly identifying the loopback address instead of the assigned address.
    In most cases, where the application is run inside a docker container this method will work just fine.
    """


    def __init__(self, redis_store: RedisSuperStorage):
        """
        During the construction of the object, the host name is of the current machine is cached, for quicker lookups.

        :param redis_store: a RedisSuperStorage instance.
        """
        self._redis_store = redis_store
        self._host_name = socket.gethostname()

    def ip_changed(self) -> Tuple[bool, str]:
        """
        This method fetches the last ip address from the RedisSuperStorage instance, and compares it to the current local address.
        If the ip address changes the new value is automatically stored in the RedisSuperStorage instance.

        Detection is performed upon calling this method, as well as storing the updated address.

        :return: (changed, ip_address) A tuple with two members, where the first member indicates if the ip address is changed, the second member is the current ip address.
        """
        old_ip = self._redis_store.current_ip
        current_ip = socket.gethostbyname(self._host_name)

        if current_ip != old_ip:
            logging.info(f'IP changed: {old_ip} -> {current_ip}')
            self._redis_store.current_ip = current_ip
            return True, current_ip

        return False, old_ip