Cleanup dangling services

When a service was previously registered into the service registry
and registrator exits without unregistering, registrator now queries
the backend to see which services were registered, and checks against
it's internal list to determine which should be unregistered.

bridge/bridge.go

@@ -7,13 +7,16 @@ import (
 	"net/url"
 	"os"
 	"path"
+	"regexp"
 	"strconv"
 	"strings"
 	"sync"
 
 	dockerapi "github.com/fsouza/go-dockerclient"
 )
 
+var serviceIDPattern = regexp.MustCompile(`^(.+?):([a-zA-Z0-9][a-zA-Z0-9_.-]+):[0-9]+(?::udp)?$`)
+
 type Bridge struct {
 	sync.Mutex
 	registry       RegistryAdapter
@@ -98,8 +101,7 @@ func (b *Bridge) Sync(quiet bool) {
 
 	log.Printf("Syncing services on %d containers", len(containers))
 
-	// NOTE: This assumes reregistering will do the right thing, i.e. nothing.
-	// NOTE: This will NOT remove services.
+	// NOTE: This assumes reregistering will do the right thing, i.e. nothing..
 	for _, listing := range containers {
 		services := b.services[listing.ID]
 		if services == nil {
@@ -113,6 +115,47 @@ func (b *Bridge) Sync(quiet bool) {
 			}
 		}
 	}
+
+	// Clean up services that were registered previously, but aren't
+	// acknowledged within registrator
+	if b.config.Cleanup {
+		log.Println("Cleaning up dangling services")
+
+		extServices, err := b.registry.Services()
+		if err != nil {
+			log.Println("cleanup failed:", err)
+			return
+		}
+
+	Outer:
+		for _, extService := range extServices {
+			matches := serviceIDPattern.FindStringSubmatch(extService.ID)
+			if len(matches) != 3 {
+				// There's no way this was registered by us, so leave it
+				continue
+			}
+			serviceHostname := matches[1]
+			if serviceHostname != Hostname {
+				// ignore because registered on a different host
+				continue
+			}
+			serviceContainerName := matches[2]
+			for _, listing := range b.services {
+				for _, service := range listing {
+					if service.Name == extService.Name && serviceContainerName == service.Origin.container.Name[1:] {
+						continue Outer
+					}
+				}
+			}
+			log.Println("dangling:", extService.ID)
+			err := b.registry.Deregister(extService)
+			if err != nil {
+				log.Println("deregister failed:", extService.ID, err)
+				continue
+			}
+			log.Println(extService.ID, "removed")
+		}
+	}
 }
 
 func (b *Bridge) add(containerId string, quiet bool) {
@@ -179,15 +222,14 @@ func (b *Bridge) newService(port ServicePort, isgroup bool) *Service {
 	defaultName := strings.Split(path.Base(container.Config.Image), ":")[0]
 
 	// not sure about this logic. kind of want to remove it.
-	hostname, err := os.Hostname()
-	if err != nil {
+	hostname := Hostname
+	if hostname == "" {
 		hostname = port.HostIP
-	} else {
-		if port.HostIP == "0.0.0.0" {
-			ip, err := net.ResolveIPAddr("ip", hostname)
-			if err == nil {
-				port.HostIP = ip.String()
-			}
+	}
+	if port.HostIP == "0.0.0.0" {
+		ip, err := net.ResolveIPAddr("ip", hostname)
+		if err == nil {
+			port.HostIP = ip.String()
 		}
 	}
 
@@ -283,3 +325,11 @@ func (b *Bridge) didExitCleanly(containerId string) bool {
 	}
 	return !container.State.Running && container.State.ExitCode == 0
 }
+
+var Hostname string
+
+func init() {
+	// It's ok for Hostname to ultimately be an empty string
+	// An empty string will fall back to trying to make a best guess
+	Hostname, _ = os.Hostname()
+}

bridge/types.go

@@ -16,6 +16,7 @@ type RegistryAdapter interface {
 	Register(service *Service) error
 	Deregister(service *Service) error
 	Refresh(service *Service) error
+	Services() ([]*Service, error)
 }
 
 type Config struct {
@@ -25,6 +26,7 @@ type Config struct {
 	RefreshTtl      int
 	RefreshInterval int
 	DeregisterCheck string
+	Cleanup         bool
 }
 
 type Service struct {
@@ -52,5 +54,6 @@ type ServicePort struct {
 	PortType          string
 	ContainerHostname string
 	ContainerID       string
+	ContainerName     string
 	container         *dockerapi.Container
 }

consul/consul.go

@@ -96,3 +96,24 @@ func (r *ConsulAdapter) Deregister(service *bridge.Service) error {
 func (r *ConsulAdapter) Refresh(service *bridge.Service) error {
 	return nil
 }
+
+func (r *ConsulAdapter) Services() ([]*bridge.Service, error) {
+	services, err := r.client.Agent().Services()
+	if err != nil {
+		return []*bridge.Service{}, err
+	}
+	out := make([]*bridge.Service, len(services))
+	i := 0
+	for _, v := range services {
+		s := &bridge.Service{
+			ID:   v.ID,
+			Name: v.Service,
+			Port: v.Port,
+			Tags: v.Tags,
+			IP:   v.Address,
+		}
+		out[i] = s
+		i++
+	}
+	return out, nil
+}

consulkv/consulkv.go

@@ -68,3 +68,7 @@ func (r *ConsulKVAdapter) Deregister(service *bridge.Service) error {
 func (r *ConsulKVAdapter) Refresh(service *bridge.Service) error {
 	return nil
 }
+
+func (r *ConsulKVAdapter) Services() ([]*bridge.Service, error) {
+	return []*bridge.Service{}, nil
+}

etcd/etcd.go

@@ -123,3 +123,7 @@ func (r *EtcdAdapter) Deregister(service *bridge.Service) error {
 func (r *EtcdAdapter) Refresh(service *bridge.Service) error {
 	return r.Register(service)
 }
+
+func (r *EtcdAdapter) Services() ([]*bridge.Service, error) {
+	return []*bridge.Service{}, nil
+}

registrator.go

@@ -25,6 +25,8 @@ var resyncInterval = flag.Int("resync", 0, "Frequency with which services are re
 var deregister = flag.String("deregister", "always", "Deregister exited services \"always\" or \"on-success\"")
 var retryAttempts = flag.Int("retry-attempts", 0, "Max retry attempts to establish a connection with the backend. Use -1 for infinite retries")
 var retryInterval = flag.Int("retry-interval", 2000, "Interval (in millisecond) between retry-attempts.")
+var cleanup = flag.Bool("cleanup", false, "Remove dangling services")
+
 
 func getopt(name, def string) string {
 	if env := os.Getenv(name); env != "" {
@@ -76,6 +78,7 @@ func main() {
 		RefreshTtl:      *refreshTtl,
 		RefreshInterval: *refreshInterval,
 		DeregisterCheck: *deregister,
+		Cleanup:         *cleanup,
 	})
 
 	assert(err)

skydns2/skydns2.go

@@ -65,6 +65,10 @@ func (r *Skydns2Adapter) Refresh(service *bridge.Service) error {
 	return r.Register(service)
 }
 
+func (r *Skydns2Adapter) Services() ([]*bridge.Service, error) {
+	return []*bridge.Service{}, nil
+}
+
 func (r *Skydns2Adapter) servicePath(service *bridge.Service) string {
 	return r.path + "/" + service.Name + "/" + service.ID
 }