chore(dns-server): implement resolver for service records that watches container changes

internal/machine/dns/resolver.go

@@ -0,0 +1,126 @@
+package dns
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"net/netip"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/psviderski/uncloud/internal/machine/store"
+	"github.com/psviderski/uncloud/pkg/api"
+)
+
+// ClusterResolver implements Resolver by tracking containers in the cluster and resolving service names
+// to their IP addresses.
+type ClusterResolver struct {
+	store *store.Store
+	// serviceIPs maps service names to container IPs.
+	serviceIPs map[string][]netip.Addr
+	// mu protects the serviceIPs map.
+	mu sync.RWMutex
+	// lastUpdate tracks when records were last updated.
+	lastUpdate time.Time
+	log        *slog.Logger
+}
+
+// NewClusterResolver creates a new cluster resolver using the cluster store.
+func NewClusterResolver(store *store.Store) *ClusterResolver {
+	return &ClusterResolver{
+		store:      store,
+		serviceIPs: make(map[string][]netip.Addr),
+		log:        slog.With("component", "dns-resolver"),
+	}
+}
+
+// Run starts watching for container changes and updates DNS records accordingly.
+func (r *ClusterResolver) Run(ctx context.Context) error {
+	containers, changes, err := r.store.SubscribeContainers(ctx)
+	if err != nil {
+		return fmt.Errorf("subscribe to container changes: %w", err)
+	}
+	r.log.Info("Subscribed to container changes in the cluster to keep DNS records updated.")
+
+	// TODO: implement machine membership check using Corrossion Admin client to filter available containers.
+	r.updateServiceIPs(containers)
+
+	for {
+		select {
+		case _, ok := <-changes:
+			if !ok {
+				return fmt.Errorf("containers subscription failed")
+			}
+			r.log.Debug("Cluster containers changed, updating DNS records.")
+
+			containers, err = r.store.ListContainers(ctx, store.ListOptions{})
+			if err != nil {
+				r.log.Error("Failed to list containers.", "err", err)
+				continue
+			}
+
+			// TODO: implement machine membership check using Corrossion Admin client to filter available containers.
+			r.updateServiceIPs(containers)
+		case <-ctx.Done():
+			return nil
+		}
+	}
+}
+
+// updateServiceIPs processes container records and updates the serviceIPs map.
+func (r *ClusterResolver) updateServiceIPs(containers []store.ContainerRecord) {
+	newServiceIPs := make(map[string][]netip.Addr, len(r.serviceIPs))
+
+	containersCount := 0
+	for _, record := range containers {
+		if !record.Container.Healthy() {
+			continue
+		}
+
+		ip := record.Container.UncloudNetworkIP()
+		if !ip.IsValid() {
+			// Container is not connected to the uncloud Docker network (could be host network).
+			continue
+		}
+
+		ctr := api.ServiceContainer{Container: record.Container}
+		if ctr.ServiceID() == "" || ctr.ServiceName() == "" {
+			// Container is not part of a service, skip it.
+			continue
+		}
+
+		// TODO: remove normalisation after implementing service name validation:
+		//.https://github.com/psviderski/uncloud/issues/53
+		serviceName := strings.ToLower(ctr.ServiceName())
+
+		newServiceIPs[serviceName] = append(newServiceIPs[serviceName], ip)
+		// Also add the service ID as a valid lookup.
+		newServiceIPs[ctr.ServiceID()] = append(newServiceIPs[ctr.ServiceID()], ip)
+		containersCount++
+	}
+
+	// Update the serviceIPs map atomically.
+	r.mu.Lock()
+	r.serviceIPs = newServiceIPs
+	r.mu.Unlock()
+
+	r.log.Debug("DNS records updated.", "services", len(newServiceIPs), "containers", containersCount)
+}
+
+// Resolve returns IP addresses of the service containers.
+func (r *ClusterResolver) Resolve(serviceName string) []netip.Addr {
+	r.mu.RLock()
+	defer r.mu.RUnlock()
+
+	ips, ok := r.serviceIPs[serviceName]
+	if !ok || len(ips) == 0 {
+		return nil
+	}
+
+	// Return a copy of the IPs slice to prevent modification of the original.
+	ipsCopy := make([]netip.Addr, len(ips))
+	copy(ipsCopy, ips)
+
+	return ipsCopy
+}

internal/machine/dns/server.go

@@ -31,7 +31,8 @@ const (
 
 // Resolver is an interface for resolving service names to IP addresses.
 type Resolver interface {
-	// Resolve returns a list of IP addresses of service containers. An empty list is returned if no service is found.
+	// Resolve returns a list of IP addresses of the service containers.
+	// An empty list is returned if no service is found.
 	Resolve(serviceName string) []netip.Addr
 }
 

pkg/api/container.go

@@ -3,11 +3,13 @@ package api
 import (
 	"encoding/json"
 	"fmt"
+	"net/netip"
 	"strings"
 	"time"
 
 	"github.com/docker/docker/api/types"
 	"github.com/docker/go-units"
+	"github.com/psviderski/uncloud/internal/machine/docker"
 )
 
 const (
@@ -90,6 +92,22 @@ func (c *Container) HumanState() (string, error) {
 		c.State.ExitCode, units.HumanDuration(time.Now().UTC().Sub(finishedAt))), nil
 }
 
+// UncloudNetworkIP returns the IP address of the container in the uncloud Docker network.
+func (c *Container) UncloudNetworkIP() netip.Addr {
+	network, ok := c.NetworkSettings.Networks[docker.NetworkName]
+	if !ok {
+		// Container is not connected to the uncloud Docker network (could be host network).
+		return netip.Addr{}
+	}
+
+	ip, err := netip.ParseAddr(network.IPAddress)
+	if err != nil {
+		return netip.Addr{}
+	}
+
+	return ip
+}
+
 func (c *Container) UnmarshalJSON(data []byte) error {
 	// A temporary type that's identical to Container but doesn't have the UnmarshalJSON method.
 	type ContainerAlias Container