etcd: Distributed Key-Value Store

# Basic key-value operations
# Set a key-value pair
etcdctl put /config/database/host "db.example.com"
etcdctl put /config/database/port "5432"
etcdctl put /config/database/name "myapp"

# Get a single key
etcdctl get /config/database/host

# Get all keys with a prefix
etcdctl get /config/database/ --prefix

# Get keys with values and metadata
etcdctl get /config/database/ --prefix --print-value-only
etcdctl get /config/database/ --prefix -w json

# Watch for changes
etcdctl watch /config/database/ --prefix

# Delete keys
etcdctl del /config/database/host
etcdctl del /config/database/ --prefix

# Atomic operations with transactions
etcdctl txn <<< '
mod("/config/feature/enabled") = "0"

put /config/feature/enabled "true"
put /config/feature/rollout "10"

get /config/feature/enabled
'

// Service registration and discovery
package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "net/http"
    "os"
    "time"

    "go.etcd.io/etcd/clientv3"
)

type ServiceInfo struct {
    Name     string    `json:"name"`
    Address  string    `json:"address"`
    Port     int       `json:"port"`
    Health   string    `json:"health"`
    Metadata map[string]string `json:"metadata"`
    RegisteredAt time.Time `json:"registered_at"`
}

type ServiceRegistry struct {
    client    *clientv3.Client
    leaseID   clientv3.LeaseID
    keepAlive <-chan *clientv3.LeaseKeepAliveResponse
}

func NewServiceRegistry(endpoints []string) (*ServiceRegistry, error) {
    client, err := clientv3.New(clientv3.Config{
        Endpoints:   endpoints,
        DialTimeout: 5 * time.Second,
    })
    if err != nil {
        return nil, err
    }

    return &ServiceRegistry{client: client}, nil
}

func (sr *ServiceRegistry) Register(service ServiceInfo) error {
    // Create lease with 30-second TTL
    resp, err := sr.client.Grant(context.Background(), 30)
    if err != nil {
        return err
    }
    sr.leaseID = resp.ID

    // Keep the lease alive
    sr.keepAlive, err = sr.client.KeepAlive(context.Background(), sr.leaseID)
    if err != nil {
        return err
    }

    // Process keep alive responses
    go func() {
        for ka := range sr.keepAlive {
            if ka != nil {
                log.Printf("Lease renewed: %x", ka.ID)
            }
        }
    }()

    // Serialize service info
    serviceData, err := json.Marshal(service)
    if err != nil {
        return err
    }

    // Register service with lease
    key := fmt.Sprintf("/services/%s/%s:%d", service.Name, service.Address, service.Port)
    _, err = sr.client.Put(context.Background(), key, string(serviceData), 
        clientv3.WithLease(sr.leaseID))

    return err
}

func (sr *ServiceRegistry) Discover(serviceName string) ([]ServiceInfo, error) {
    resp, err := sr.client.Get(context.Background(), 
        fmt.Sprintf("/services/%s/", serviceName), 
        clientv3.WithPrefix())
    if err != nil {
        return nil, err
    }

    var services []ServiceInfo
    for _, kv := range resp.Kvs {
        var service ServiceInfo
        if err := json.Unmarshal(kv.Value, &service); err != nil {
            continue
        }
        services = append(services, service)
    }

    return services, nil
}

func (sr *ServiceRegistry) Watch(serviceName string, callback func([]ServiceInfo)) {
    watchChan := sr.client.Watch(context.Background(), 
        fmt.Sprintf("/services/%s/", serviceName), 
        clientv3.WithPrefix())

    for watchResp := range watchChan {
        services, err := sr.Discover(serviceName)
        if err != nil {
            log.Printf("Error discovering services: %v", err)
            continue
        }
        callback(services)
    }
}

// Usage example
func main() {
    registry, err := NewServiceRegistry([]string{"localhost:2379"})
    if err != nil {
        log.Fatal(err)
    }
    defer registry.client.Close()

    // Register this service
    service := ServiceInfo{
        Name:    "api-server",
        Address: "192.168.1.100",
        Port:    8080,
        Health:  "healthy",
        Metadata: map[string]string{
            "version": "1.2.3",
            "region":  "us-east-1",
        },
        RegisteredAt: time.Now(),
    }

    if err := registry.Register(service); err != nil {
        log.Fatal(err)
    }

    // Start health check endpoint
    http.HandleFunc("/health", healthCheckHandler)
    go http.ListenAndServe(":8080", nil)

    // Watch for other API servers
    go registry.Watch("api-server", func(services []ServiceInfo) {
        log.Printf("Available API servers: %d", len(services))
        for _, svc := range services {
            log.Printf("  - %s:%d (health: %s)", svc.Address, svc.Port, svc.Health)
        }
        updateLoadBalancer(services)
    })

    // Keep running
    select {}
}

func healthCheckHandler(w http.ResponseWriter, r *http.Request) {
    w.WriteHeader(http.StatusOK)
    w.Write([]byte("OK"))
}

func updateLoadBalancer(services []ServiceInfo) {
    // Update load balancer configuration
    // Filter only healthy services
    healthyServices := make([]ServiceInfo, 0)
    for _, svc := range services {
        if svc.Health == "healthy" {
            healthyServices = append(healthyServices, svc)
        }
    }
    // Update routing rules...
}