Service Discovery Service

import requests
import socket
import threading
import time
from typing import Dict, List, Optional
 
class ServiceDiscoveryClient:
    def __init__(self, discovery_url: str, service_name: str, 
                 service_port: int, health_check_path: str = '/health'):
        self.discovery_url = discovery_url
        self.service_name = service_name
        self.service_port = service_port
        self.health_check_path = health_check_path
        self.service_id = f"{service_name}-{socket.gethostname()}-{service_port}"
        self.registration_data = None
        self.heartbeat_thread = None
        self.running = False
    
    def register(self, tags: List[str] = None, metadata: Dict[str, str] = None):
        """Register service with discovery service"""
        self.registration_data = {
            'id': self.service_id,
            'name': self.service_name,
            'address': self._get_local_ip(),
            'port': self.service_port,
            'tags': tags or [],
            'metadata': metadata or {},
            'health_check': {
                'http': f"http://{self._get_local_ip()}:{self.service_port}{self.health_check_path}",
                'interval': '10s',
                'timeout': '3s',
                'deregister_critical_service_after': '30s'
            }
        }
        
        response = requests.post(
            f"{self.discovery_url}/v1/agent/service/register",
            json=self.registration_data
        )
        
        if response.status_code == 200:
            print(f"Service {self.service_name} registered successfully")
            self._start_heartbeat()
        else:
            raise Exception(f"Service registration failed: {response.text}")
    
    def deregister(self):
        """Deregister service from discovery service"""
        self.running = False
        if self.heartbeat_thread:
            self.heartbeat_thread.join()
        
        response = requests.put(
            f"{self.discovery_url}/v1/agent/service/deregister/{self.service_id}"
        )
        
        if response.status_code == 200:
            print(f"Service {self.service_name} deregistered successfully")
        else:
            print(f"Service deregistration failed: {response.text}")
    
    def discover_service(self, service_name: str, healthy_only: bool = True) -> List[Dict]:
        """Discover instances of a service"""
        params = {'passing': 'true'} if healthy_only else {}
        
        response = requests.get(
            f"{self.discovery_url}/v1/health/service/{service_name}",
            params=params
        )
        
        if response.status_code == 200:
            services = response.json()
            return [
                {
                    'id': service['Service']['ID'],
                    'address': service['Service']['Address'],
                    'port': service['Service']['Port'],
                    'tags': service['Service']['Tags'],
                    'metadata': service['Service']['Meta']
                }
                for service in services
            ]
        else:
            return []
    
    def get_service_url(self, service_name: str, load_balance: bool = True) -> Optional[str]:
        """Get URL for a service instance"""
        instances = self.discover_service(service_name)
        
        if not instances:
            return None
        
        if load_balance:
            # Simple round-robin load balancing
            instance = instances[hash(time.time()) % len(instances)]
        else:
            instance = instances[0]
        
        return f"http://{instance['address']}:{instance['port']}"
    
    def _get_local_ip(self):
        """Get local IP address"""
        try:
            # Connect to a non-routable address to determine local IP
            with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
                s.connect(('10.254.254.254', 80))
                return s.getsockname()[0]
        except Exception:
            return '127.0.0.1'
    
    def _start_heartbeat(self):
        """Start heartbeat thread for service health"""
        self.running = True
        self.heartbeat_thread = threading.Thread(target=self._heartbeat_loop)
        self.heartbeat_thread.daemon = True
        self.heartbeat_thread.start()
    
    def _heartbeat_loop(self):
        """Heartbeat loop to maintain service registration"""
        while self.running:
            try:
                # Send heartbeat
                response = requests.put(
                    f"{self.discovery_url}/v1/agent/check/pass/service:{self.service_id}"
                )
                
                if response.status_code != 200:
                    print(f"Heartbeat failed: {response.text}")
                
            except Exception as e:
                print(f"Heartbeat error: {e}")
            
            time.sleep(30)  # Send heartbeat every 30 seconds
 
# Usage example
discovery_client = ServiceDiscoveryClient(
    discovery_url='http://consul.sindhan.ai:8500',
    service_name='user-service',
    service_port=8080,
    health_check_path='/health'
)
 
# Register service on startup
discovery_client.register(
    tags=['api', 'user-management'],
    metadata={'version': '1.2.3', 'environment': 'production'}
)
 
# Discover and call another service
payment_service_url = discovery_client.get_service_url('payment-service')
if payment_service_url:
    response = requests.post(f"{payment_service_url}/api/charge", json=payment_data)

# Service registration via Kubernetes service
apiVersion: v1
kind: Service
metadata:
  name: user-service
  namespace: sindhan-platform
  labels:
    app: user-service
    version: v1.2.3
  annotations:
    service.alpha.kubernetes.io/tolerate-unready-endpoints: "true"
spec:
  selector:
    app: user-service
  ports:
  - name: http
    port: 8080
    targetPort: 8080
  - name: grpc
    port: 9090
    targetPort: 9090
 
---
# Istio DestinationRule for load balancing
apiVersion: networking.istio.io/v1beta1
kind: DestinationRule
metadata:
  name: user-service
  namespace: sindhan-platform
spec:
  host: user-service
  trafficPolicy:
    loadBalancer:
      simple: LEAST_CONN
    connectionPool:
      tcp:
        maxConnections: 100
      http:
        http1MaxPendingRequests: 10
        maxRequestsPerConnection: 2
    outlierDetection:
      consecutiveErrors: 3
      interval: 30s
      baseEjectionTime: 30s
      maxEjectionPercent: 50
  subsets:
  - name: v1
    labels:
      version: v1.2.3
  - name: canary
    labels:
      version: v1.3.0-canary
 
---
# VirtualService for traffic routing
apiVersion: networking.istio.io/v1beta1
kind: VirtualService
metadata:
  name: user-service
  namespace: sindhan-platform
spec:
  hosts:
  - user-service
  http:
  - match:
    - headers:
        canary:
          exact: "true"
    route:
    - destination:
        host: user-service
        subset: canary
      weight: 100
  - route:
    - destination:
        host: user-service
        subset: v1
      weight: 90
    - destination:
        host: user-service
        subset: canary
      weight: 10
    fault:
      delay:
        percentage:
          value: 0.1
        fixedDelay: 5s
    retries:
      attempts: 3
      perTryTimeout: 10s

package servicediscovery
 
import (
    "context"
    "fmt"
    "math/rand"
    "net/http"
    "sync"
    "time"
    
    "github.com/hashicorp/consul/api"
)
 
type ServiceInstance struct {
    ID       string
    Address  string
    Port     int
    Tags     []string
    Metadata map[string]string
    Healthy  bool
}
 
type LoadBalancer interface {
    Choose(instances []ServiceInstance) *ServiceInstance
}
 
type RoundRobinLB struct {
    counter uint64
    mutex   sync.Mutex
}
 
func (lb *RoundRobinLB) Choose(instances []ServiceInstance) *ServiceInstance {
    if len(instances) == 0 {
        return nil
    }
    
    lb.mutex.Lock()
    defer lb.mutex.Unlock()
    
    lb.counter++
    index := lb.counter % uint64(len(instances))
    return &instances[index]
}
 
type ServiceClient struct {
    consulClient *api.Client
    loadBalancer LoadBalancer
    httpClient   *http.Client
    cache        map[string][]ServiceInstance
    cacheMutex   sync.RWMutex
    cacheTTL     time.Duration
}
 
func NewServiceClient(consulAddr string) (*ServiceClient, error) {
    config := api.DefaultConfig()
    config.Address = consulAddr
    
    consulClient, err := api.NewClient(config)
    if err != nil {
        return nil, err
    }
    
    return &ServiceClient{
        consulClient: consulClient,
        loadBalancer: &RoundRobinLB{},
        httpClient: &http.Client{
            Timeout: 30 * time.Second,
        },
        cache:    make(map[string][]ServiceInstance),
        cacheTTL: 30 * time.Second,
    }, nil
}
 
func (sc *ServiceClient) DiscoverService(serviceName string) ([]ServiceInstance, error) {
    // Check cache first
    sc.cacheMutex.RLock()
    if instances, exists := sc.cache[serviceName]; exists {
        sc.cacheMutex.RUnlock()
        return instances, nil
    }
    sc.cacheMutex.RUnlock()
    
    // Query Consul for healthy service instances
    services, _, err := sc.consulClient.Health().Service(serviceName, "", true, nil)
    if err != nil {
        return nil, fmt.Errorf("failed to discover service %s: %w", serviceName, err)
    }
    
    var instances []ServiceInstance
    for _, service := range services {
        instance := ServiceInstance{
            ID:       service.Service.ID,
            Address:  service.Service.Address,
            Port:     service.Service.Port,
            Tags:     service.Service.Tags,
            Metadata: service.Service.Meta,
            Healthy:  true,
        }
        instances = append(instances, instance)
    }
    
    // Update cache
    sc.cacheMutex.Lock()
    sc.cache[serviceName] = instances
    sc.cacheMutex.Unlock()
    
    // Set cache expiration
    go func() {
        time.Sleep(sc.cacheTTL)
        sc.cacheMutex.Lock()
        delete(sc.cache, serviceName)
        sc.cacheMutex.Unlock()
    }()
    
    return instances, nil
}
 
func (sc *ServiceClient) CallService(serviceName, path string, method string, body []byte) (*http.Response, error) {
    instances, err := sc.DiscoverService(serviceName)
    if err != nil {
        return nil, err
    }
    
    if len(instances) == 0 {
        return nil, fmt.Errorf("no healthy instances found for service %s", serviceName)
    }
    
    // Choose instance using load balancer
    instance := sc.loadBalancer.Choose(instances)
    if instance == nil {
        return nil, fmt.Errorf("load balancer returned no instance")
    }
    
    // Construct URL
    url := fmt.Sprintf("http://%s:%d%s", instance.Address, instance.Port, path)
    
    // Create request
    req, err := http.NewRequest(method, url, strings.NewReader(string(body)))
    if err != nil {
        return nil, err
    }
    
    req.Header.Set("Content-Type", "application/json")
    req.Header.Set("X-Service-Discovery", "consul")
    
    // Make request with retry logic
    maxRetries := 3
    for i := 0; i < maxRetries; i++ {
        resp, err := sc.httpClient.Do(req)
        if err == nil && resp.StatusCode < 500 {
            return resp, nil
        }
        
        if i < maxRetries-1 {
            // Try different instance on failure
            instances = sc.removeInstance(instances, instance.ID)
            if len(instances) > 0 {
                instance = sc.loadBalancer.Choose(instances)
                url = fmt.Sprintf("http://%s:%d%s", instance.Address, instance.Port, path)
                req.URL, _ = req.URL.Parse(url)
            }
            
            time.Sleep(time.Duration(i+1) * 100 * time.Millisecond)
        }
    }
    
    return nil, fmt.Errorf("all retry attempts failed for service %s", serviceName)
}
 
func (sc *ServiceClient) removeInstance(instances []ServiceInstance, instanceID string) []ServiceInstance {
    var result []ServiceInstance
    for _, instance := range instances {
        if instance.ID != instanceID {
            result = append(result, instance)
        }
    }
    return result
}
 
// Usage example
client, err := NewServiceClient("consul.sindhan.ai:8500")
if err != nil {
    log.Fatal(err)
}
 
// Call payment service
response, err := client.CallService("payment-service", "/api/charge", "POST", paymentData)
if err != nil {
    log.Printf("Failed to call payment service: %v", err)
    return
}
defer response.Body.Close()