gRPC Naming and Discovery — Service Resolution with DNS, Consul, and Kubernetes

package main

import (
    "google.golang.org/grpc"
    "google.golang.org/grpc/credentials/insecure"
    "google.golang.org/grpc/resolver"
)

func init() {
    // Register DNS resolver scheme
    resolver.SetDefaultScheme("dns")
}

func main() {
    // DNS-based resolution with round-robin
    conn, _ := grpc.Dial(
        "dns:///threat-service.default.svc.cluster.local:50051",
        grpc.WithTransportCredentials(insecure.NewCredentials()),
        grpc.WithDefaultServiceConfig(`{
            "loadBalancingConfig": [{"round_robin": {}}],
            "methodConfig": [{
                "name": [{"service": ""}],
                "retryPolicy": {
                    "maxAttempts": 3,
                    "initialBackoff": "0.1s",
                    "maxBackoff": "1s",
                    "backoffMultiplier": 2,
                    "retryableStatusCodes": ["UNAVAILABLE"]
                }
            }]
        }`),
        grpc.WithResolvers(&customResolver{}),
    )
    
    client := pb.NewThreatServiceClient(conn)
}

// Custom resolver builder
type customResolver struct{}

func (r *customResolver) Build(target resolver.Target,
    cc resolver.ClientConn,
    opts resolver.BuildOptions,
) (resolver.Resolver, error) {
    
    // Resolve addresses
    addrs := []resolver.Address{
        {Addr: "10.0.0.1:50051"},
        {Addr: "10.0.0.2:50051"},
    }
    
    cc.UpdateState(resolver.State{
        Addresses: addrs,
    })
    
    return &resolverState{cc: cc}, nil
}

func (r *customResolver) Scheme() string {
    return "custom"
}

type resolverState struct {
    cc resolver.ClientConn
}

func (r *resolverState) ResolveNow(o resolver.ResolveNowOptions) {
    // Trigger re-resolution
}

func (r *resolverState) Close() {}

import grpc
import consul
import random

class ConsulResolver:
    def __init__(self, consul_host="localhost",
                 consul_port=8500):
        self.consul = consul.Consul(
            host=consul_host, port=consul_port)
        self.cache = {}
    
    def resolve(self, service_name):
        """Resolve service to available instances."""
        if service_name in self.cache:
            instances, index = self.cache[service_name]
            # Use blocking query for long polling
            instances, index = self.consul.health.service(
                service_name, index=index, wait="30s")
            self.cache[service_name] = (instances, index)
        else:
            instances, index = self.consul.health.service(
                service_name)
            self.cache[service_name] = (instances, index)
        
        # Filter healthy instances
        healthy = [
            i for i in instances
            if i["Checks"][0]["Status"] == "passing"
        ]
        
        if not healthy:
            raise Exception(
                f"No healthy instances for {service_name}")
        
        # Pick random instance (load balance)
        instance = random.choice(healthy)
        service = instance["Service"]
        address = service.get("Address")
        port = service["Port"]
        
        return f"{address}:{port}"
    
    def create_channel(self, service_name):
        address = self.resolve(service_name)
        return grpc.insecure_channel(address)

# Usage
resolver = ConsulResolver()
channel = resolver.create_channel("threat-service")
stub = pb.ThreatServiceStub(channel)

const grpc = require('@grpc/grpc-js');
const { K8sResolver } = require('@grpc/resolver-k8s');

// Register Kubernetes resolver
grpc.resolver.register(K8sResolver);

// Kubernetes-aware connection
const client = new ThreatServiceClient(
  'k8s:///default/threat-service:50051',
  grpc.credentials.createInsecure(),
  {
    'grpc.lb_policy_name': 'round_robin',
    'grpc.dns_min_time_between_resolutions_ms': 5000,
  },
);

// Watch for endpoint changes
function watchEndpoints(namespace, service) {
  const watch = new K8sResolver.EndpointWatcher(
    namespace, service);
  
  watch.on('endpoints', (endpoints) => {
    console.log('Available endpoints:', endpoints);
  });
  
  watch.on('error', (error) => {
    console.error('Endpoint watch error:', error);
  });
  
  return watch;
}

// Custom health-aware resolver
class HealthAwareResolver {
  constructor() {
    this.healthyBackends = new Map();
  }
  
  addBackend(address) {
    this.healthyBackends.set(address, {
      healthy: true,
      failCount: 0,
    });
  }
  
  markUnhealthy(address) {
    const backend = this.healthyBackends.get(address);
    if (backend) {
      backend.failCount++;
      if (backend.failCount > 3) {
        backend.healthy = false;
        console.log(`Marked ${address} as unhealthy`);
        // Trigger re-resolution after 30s
        setTimeout(() => {
          backend.healthy = true;
          backend.failCount = 0;
        }, 30000);
      }
    }
  }
  
  getAddresses() {
    return Array.from(this.healthyBackends.entries())
      .filter(([_, status]) => status.healthy)
      .map(([addr]) => ({ addr }));
  }
}