Consul Has a Free API: Here's How to Use It for Service Discovery Automation

#consul #microservices #devops #api

HashiCorp Consul provides a powerful HTTP API for service discovery, health checking, key-value storage, and network configuration. It's the backbone of service mesh architectures and it's completely free.

Why Use the Consul API?

Discover services dynamically without hardcoded IPs
Health check all services automatically
Store configuration in a distributed key-value store
Automate DNS and load balancing for microservices

Getting Started

# Start Consul agent
consul agent -dev

# Register a service
curl -X PUT http://localhost:8500/v1/agent/service/register -d '{
  "Name": "web-api",
  "Port": 8080,
  "Tags": ["v2", "production"],
  "Check": {
    "HTTP": "http://localhost:8080/health",
    "Interval": "10s"
  }
}'

# Discover services
curl -s http://localhost:8500/v1/catalog/services | jq .

# Get service instances
curl -s http://localhost:8500/v1/health/service/web-api?passing=true | jq '.[] | {node: .Node.Node, address: .Service.Address, port: .Service.Port}'

Python Client

import requests

class ConsulClient:
    def __init__(self, url='http://localhost:8500'):
        self.url = f"{url}/v1"

    def register_service(self, name, port, tags=None, health_url=None):
        service = {'Name': name, 'Port': port}
        if tags:
            service['Tags'] = tags
        if health_url:
            service['Check'] = {'HTTP': health_url, 'Interval': '10s'}
        resp = requests.put(f"{self.url}/agent/service/register", json=service)
        return resp.status_code == 200

    def deregister_service(self, service_id):
        resp = requests.put(f"{self.url}/agent/service/deregister/{service_id}")
        return resp.status_code == 200

    def get_service(self, name, passing_only=True):
        params = {'passing': 'true'} if passing_only else {}
        resp = requests.get(f"{self.url}/health/service/{name}", params=params)
        return resp.json()

    def kv_get(self, key):
        resp = requests.get(f"{self.url}/kv/{key}?raw=true")
        if resp.status_code == 200:
            return resp.text
        return None

    def kv_put(self, key, value):
        resp = requests.put(f"{self.url}/kv/{key}", data=value)
        return resp.json()

    def kv_delete(self, key):
        resp = requests.delete(f"{self.url}/kv/{key}")
        return resp.json()

    def get_all_services(self):
        resp = requests.get(f"{self.url}/catalog/services")
        return resp.json()

# Usage
consul = ConsulClient()

# Register services
consul.register_service('api-v2', 8080, tags=['v2', 'production'], health_url='http://localhost:8080/health')
consul.register_service('worker', 9090, tags=['background'])

# Discover healthy instances
instances = consul.get_service('api-v2')
for inst in instances:
    addr = inst['Service']['Address'] or inst['Node']['Address']
    port = inst['Service']['Port']
    print(f"  {addr}:{port} - healthy")

Service Discovery Load Balancer

import random

class ConsulLoadBalancer:
    def __init__(self, consul_client):
        self.consul = consul_client
        self._cache = {}

    def get_endpoint(self, service_name, strategy='random'):
        instances = self.consul.get_service(service_name, passing_only=True)

        if not instances:
            raise Exception(f"No healthy instances of {service_name}")

        endpoints = []
        for inst in instances:
            addr = inst['Service']['Address'] or inst['Node']['Address']
            port = inst['Service']['Port']
            endpoints.append(f"http://{addr}:{port}")

        if strategy == 'random':
            return random.choice(endpoints)
        elif strategy == 'round-robin':
            idx = self._cache.get(service_name, 0)
            endpoint = endpoints[idx % len(endpoints)]
            self._cache[service_name] = idx + 1
            return endpoint

    def call_service(self, service_name, path, method='GET', **kwargs):
        endpoint = self.get_endpoint(service_name)
        url = f"{endpoint}{path}"
        return requests.request(method, url, **kwargs)

lb = ConsulLoadBalancer(consul)

# Call a service without knowing its address
response = lb.call_service('api-v2', '/api/users')
print(response.json())

Distributed Configuration

import json

class ConsulConfig:
    def __init__(self, consul_client, prefix='config'):
        self.consul = consul_client
        self.prefix = prefix

    def set(self, key, value):
        if isinstance(value, (dict, list)):
            value = json.dumps(value)
        self.consul.kv_put(f"{self.prefix}/{key}", str(value))

    def get(self, key, default=None):
        value = self.consul.kv_get(f"{self.prefix}/{key}")
        if value is None:
            return default
        try:
            return json.loads(value)
        except (json.JSONDecodeError, TypeError):
            return value

config = ConsulConfig(consul)

# Store configuration
config.set('database/max_connections', '100')
config.set('feature_flags', {'dark_mode': True, 'new_checkout': False})
config.set('rate_limits', {'api': 1000, 'webhook': 100})

# Read configuration
flags = config.get('feature_flags')
print(f"Dark mode: {flags['dark_mode']}")

Health Dashboard

def cluster_health_report(consul):
    services = consul.get_all_services()

    print(f"{'Service':25s} {'Healthy':>8s} {'Total':>8s} {'Tags'}")
    print('-' * 70)

    for service_name, tags in services.items():
        if service_name == 'consul':
            continue

        all_instances = consul.get_service(service_name, passing_only=False)
        healthy = [i for i in all_instances if all(c['Status'] == 'passing' for c in i.get('Checks', []))]

        status = 'OK' if len(healthy) == len(all_instances) else 'DEGRADED' if healthy else 'DOWN'
        print(f"{service_name:25s} {len(healthy):>8d} {len(all_instances):>8d}  {', '.join(tags)}  [{status}]")

cluster_health_report(consul)

Real-World Use Case

A logistics company ran 200+ microservices with Consul for service discovery. When a service instance crashes, Consul's health check detects it within 10 seconds and removes it from the pool. The load balancer automatically routes traffic to healthy instances. When a new instance spins up, Consul auto-registers it. Zero manual intervention, 99.99% uptime.