fake-eureka-client.py

#!/usr/bin/env python3.11

import base64
import json
import requests
import select
import signal
import subprocess
import sys
import time
import logging
from collections import namedtuple

try:
    import curlify
    def to_curl(request):
        print("> " + curlify.to_curl(request))
except ImportError:
    def to_curl(request):
        print(f"> {request.method} {request.url}")

EurekaClient = namedtuple('EurekaClient', ['name', 'instance_id', 'port', 'secure_port', 'container_name'])
EurekaClient.__new__.__defaults__ = (None, None)

EUREKA_CLIENTS = [
    EurekaClient(name='myapi', instance_id='myapi-1', port=9001),
    EurekaClient(name='myApiWebSocket', instance_id='myApiWebSocket-1', port=8080, container_name='myapi')
]
EUREKA_CLIENT_HEARTBEAT_INTERVAL = 30

EUREKA_SERVER_HOST = 'localhost'
EUREKA_SERVER_PORT = '8761'
EUREKA_AUTH_BASIC = 'admin:password'
EUREKA_SERVER_URL = f'http://{EUREKA_AUTH_BASIC}@{EUREKA_SERVER_HOST}:{EUREKA_SERVER_PORT}'

EXIT_KEYS = ['q', 'e', 'x']

def docker_container_ip(container_name):
    try:
        return subprocess.run(
            ["docker", "inspect", "-f", "{{range.NetworkSettings.Networks}}{{.IPAddress}}{{end}}", container_name],
            check=True, capture_output=True, text=True
        ).stdout.strip()
    except subprocess.CalledProcessError:
        return None


def eureka_register_instance(client, eureka_url):
    payload = {
        "instance": {
            "instanceId": client.instance_id,
            "app": client.name,
            "hostName": docker_container_ip(client.container_name or client.name),
            "vipAddress": client.name,
            "secureVipAddress": client.name,
            "ipAddr": client.name,
            "port": {
                "$": str(client.port),
                "@enabled": "false" if client.port is None else "true"
            },
            "securePort": {
                "$": str(client.secure_port),
                "@enabled": "false" if client.secure_port is None else "true"
            },
            "status": "UP",
            "dataCenterInfo": {
                "@class": "com.netflix.appinfo.InstanceInfo$DefaultDataCenterInfo",
                "name": "MyOwn"
            }
        }
    }

    response = requests.post(f'{eureka_url}/eureka/apps/{client.name}',
        data=json.dumps(payload),
        headers={ 'Content-Type': 'application/json' })

    if response.status_code in [200, 204]:
        print(f"+ registered: {client.instance_id}")
    else:
        print(f"! failed ({response.status_code}) to register {client.instance_id}")
        print(to_curl(response.request))

# Function to send heartbeat
def eureka_send_heartbeat(client, eureka_url):
    heartbeat_url = f'{eureka_url}/eureka/apps/{client.name}/{client.instance_id}'
    response = requests.put(heartbeat_url)
    if response.status_code == 200:
        print(f"+ heartbeat: {client.instance_id}")
    else:
        print(f"! failed ({response.status_code}) to send heartbeat for {client.instance_id} with:")
        print(curlify.to_curl(response.request))
        print(f"<{response.text}")


def signal_handler(sig, frame):
    print('Shutting down...')
    sys.exit(0)

signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)

def listen_for_exit_keys(exit_keys, timeout):
    start_time = time.time()

    while True:
        ready, _, _ = select.select([sys.stdin], [], [], 0.1)
        if ready and sys.stdin.read(1) in exit_keys:
            return True
        if time.time() >= start_time + timeout:
            return False

    return False

# Main execution
if __name__ == "__main__":
    for client in EUREKA_CLIENTS: eureka_register_instance(client, EUREKA_SERVER_URL)

    try:
        while True:
            if listen_for_exit_keys(EXIT_KEYS, EUREKA_CLIENT_HEARTBEAT_INTERVAL):
                print("Exited...")
                break
            for client in EUREKA_CLIENTS: eureka_send_heartbeat(client, EUREKA_SERVER_URL)

    except KeyboardInterrupt:
        print('Interrupted...')