Merge pull request #904 from bboreham/inercia-weave-821

Remove stale records for short-lived containers (bis)

Fixes #821. Replaces and hence closes #865.

common/docker/client.go

@@ -0,0 +1,75 @@
+package docker
+
+import (
+	"regexp"
+
+	"github.com/fsouza/go-dockerclient"
+	. "github.com/weaveworks/weave/common"
+)
+
+// Regexp that matches container IDs
+var containerIDregexp = regexp.MustCompile("^[a-f0-9]+$")
+
+// IsContainerID returns True if the string provided is a valid container id
+func IsContainerID(idStr string) bool {
+	return containerIDregexp.MatchString(idStr)
+}
+
+// An observer for container events
+type ContainerObserver interface {
+	ContainerDied(ident string) error
+}
+
+type Client struct {
+	*docker.Client
+}
+
+// NewClient creates a new Docker client and checks we can talk to Docker
+func NewClient(apiPath string) (*Client, error) {
+	dc, err := docker.NewClient(apiPath)
+	if err != nil {
+		return nil, err
+	}
+	client := &Client{dc}
+
+	env, err := client.Version()
+	if err != nil {
+		return nil, err
+	}
+	Info.Printf("[docker] Using Docker API on %s: %v", apiPath, env)
+	return client, nil
+}
+
+// AddObserver adds an observer for docker events
+func (c *Client) AddObserver(ob ContainerObserver) error {
+	events := make(chan *docker.APIEvents)
+	if err := c.AddEventListener(events); err != nil {
+		Error.Printf("[docker] Unable to add listener to Docker API: %s", err)
+		return err
+	}
+
+	go func() {
+		for event := range events {
+			switch event.Status {
+			case "die":
+				id := event.ID
+				ob.ContainerDied(id)
+			}
+		}
+	}()
+	return nil
+}
+
+// IsContainerRunning returns true if the container is up and running
+func (c *Client) IsContainerRunning(idStr string) (bool, error) {
+	// check with Docker whether the container is really running
+	container, err := c.InspectContainer(idStr)
+	if err != nil {
+		if _, notThere := err.(*docker.NoSuchContainer); notThere {
+			return false, nil
+		}
+		return false, err
+	}
+
+	return container.State.Running, nil
+}

common/updater/updater.go

@@ -1,47 +0,0 @@
-package updater
-
-import (
-	"github.com/fsouza/go-dockerclient"
-	. "github.com/weaveworks/weave/common"
-)
-
-type ContainerObserver interface {
-	ContainerDied(ident string) error
-}
-
-func checkError(err error, apiPath string) {
-	if err != nil {
-		Error.Fatalf("[updater] Unable to connect to Docker API on %s: %s",
-			apiPath, err)
-	}
-}
-
-func Start(apiPath string, ob ContainerObserver) error {
-	client, err := docker.NewClient(apiPath)
-	checkError(err, apiPath)
-
-	env, err := client.Version()
-	checkError(err, apiPath)
-
-	events := make(chan *docker.APIEvents)
-	err = client.AddEventListener(events)
-	checkError(err, apiPath)
-
-	Info.Printf("[updater] Using Docker API on %s: %v", apiPath, env)
-
-	go func() {
-		for event := range events {
-			handleEvent(ob, event, client)
-		}
-	}()
-	return nil
-}
-
-func handleEvent(ob ContainerObserver, event *docker.APIEvents, client *docker.Client) error {
-	switch event.Status {
-	case "die":
-		id := event.ID
-		ob.ContainerDied(id)
-	}
-	return nil
-}

nameserver/http.go

@@ -9,6 +9,7 @@ import (
 	"github.com/gorilla/mux"
 	"github.com/miekg/dns"
 	. "github.com/weaveworks/weave/common"
+	"github.com/weaveworks/weave/common/docker"
 )
 
 func httpErrorAndLog(level *log.Logger, w http.ResponseWriter, msg string,
@@ -24,14 +25,14 @@ func extractFQDN(r *http.Request) (string, string) {
 	return "*", ""
 }
 
-func ServeHTTP(listener net.Listener, version string, server *DNSServer, domain string, db Zone) {
+func ServeHTTP(listener net.Listener, version string, server *DNSServer, dockerCli *docker.Client) {
 
 	muxRouter := mux.NewRouter()
 
 	muxRouter.Methods("GET").Path("/status").HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		fmt.Fprintln(w, "weave DNS", version)
 		fmt.Fprintln(w, server.Status())
-		fmt.Fprintln(w, db.Status())
+		fmt.Fprintln(w, server.Zone.Status())
 	})
 
 	muxRouter.Methods("GET").Path("/domain").HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -59,18 +60,33 @@ func ServeHTTP(listener net.Listener, version string, server *DNSServer, domain
 			return
 		}
 
-		if dns.IsSubDomain(domain, name) {
-			Info.Printf("[http] Adding %s -> %s", name, ipStr)
-			if err := db.AddRecord(idStr, name, ip); err != nil {
-				if _, ok := err.(DuplicateError); !ok {
-					httpErrorAndLog(
-						Error, w, "Internal error", http.StatusInternalServerError,
-						"Unexpected error from DB: %s", err)
-					return
-				} // oh, I already know this. whatever.
-			}
-		} else {
+		domain := server.Zone.Domain()
+		if !dns.IsSubDomain(domain, name) {
 			Info.Printf("[http] Ignoring name %s, not in %s", name, domain)
+			return
+		}
+		Info.Printf("[http] Adding %s -> %s", name, ipStr)
+		if err := server.Zone.AddRecord(idStr, name, ip); err != nil {
+			if _, ok := err.(DuplicateError); !ok {
+				httpErrorAndLog(
+					Error, w, "Internal error", http.StatusInternalServerError,
+					"Unexpected error from DB: %s", err)
+				return
+			} // oh, I already know this. whatever.
+		}
+
+		if dockerCli != nil {
+			if !docker.IsContainerID(idStr) {
+				Debug.Printf("[http] '%s' does not seem to be a container id", idStr)
+				return
+			}
+
+			if running, err := dockerCli.IsContainerRunning(idStr); err != nil {
+				Error.Printf("[http] Could not check container status: %s", err)
+			} else if !running {
+				Info.Printf("[http] '%s' is not running: removing", idStr)
+				server.Zone.DeleteRecords(idStr, name, ip)
+			}
 		}
 	})
 
@@ -90,7 +106,7 @@ func ServeHTTP(listener net.Listener, version string, server *DNSServer, domain
 		fqdnStr, fqdn := extractFQDN(r)
 
 		Info.Printf("[http] Deleting ID %s, IP %s, FQDN %s", idStr, ipStr, fqdnStr)
-		db.DeleteRecords(idStr, fqdn, ip)
+		server.Zone.DeleteRecords(idStr, fqdn, ip)
 	})
 
 	muxRouter.Methods("DELETE").Path("/name/{id:.+}").HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -100,14 +116,14 @@ func ServeHTTP(listener net.Listener, version string, server *DNSServer, domain
 		fqdnStr, fqdn := extractFQDN(r)
 
 		Info.Printf("[http] Deleting ID %s, IP *, FQDN %s", idStr, fqdnStr)
-		db.DeleteRecords(idStr, fqdn, net.IP{})
+		server.Zone.DeleteRecords(idStr, fqdn, net.IP{})
 	})
 
 	muxRouter.Methods("DELETE").Path("/name").HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		fqdnStr, fqdn := extractFQDN(r)
 
 		Info.Printf("[http] Deleting ID *, IP *, FQDN %s", fqdnStr)
-		db.DeleteRecords("", fqdn, net.IP{})
+		server.Zone.DeleteRecords("", fqdn, net.IP{})
 	})
 
 	http.Handle("/", muxRouter)

nameserver/http_test.go

@@ -27,7 +27,7 @@ func TestHttp(t *testing.T) {
 		dockerIP        = "9.8.7.6"
 	)
 
-	zone, err := NewZoneDb(ZoneConfig{})
+	zone, err := NewZoneDb(ZoneConfig{Domain: testDomain})
 	wt.AssertNoErr(t, err)
 	err = zone.Start()
 	wt.AssertNoErr(t, err)
@@ -38,8 +38,11 @@ func TestHttp(t *testing.T) {
 		t.Fatal("Unable to create http listener: ", err)
 	}
 
+	// Create a useless server... it will not even be started
+	srv, _ := NewDNSServer(DNSServerConfig{Zone: zone})
+
 	port := httpListener.Addr().(*net.TCPAddr).Port
-	go ServeHTTP(httpListener, "", nil, testDomain, zone)
+	go ServeHTTP(httpListener, "", srv, nil)
 
 	// Ask the http server to add our test address into the database
 	addrParts := strings.Split(testAddr1, "/")

prog/weavedns/main.go

@@ -8,7 +8,7 @@ import (
 
 	"github.com/miekg/dns"
 	. "github.com/weaveworks/weave/common"
-	"github.com/weaveworks/weave/common/updater"
+	"github.com/weaveworks/weave/common/docker"
 	weavedns "github.com/weaveworks/weave/nameserver"
 	weavenet "github.com/weaveworks/weave/net"
 )
@@ -18,6 +18,7 @@ var version = "(unreleased version)"
 func main() {
 	var (
 		justVersion     bool
+		dockerCli       *docker.Client
 		ifaceName       string
 		apiPath         string
 		domain          string
@@ -95,14 +96,18 @@ func main() {
 	if err != nil {
 		Error.Fatal("[main] Unable to initialize the Zone database", err)
 	}
-	err = zone.Start()
-	if err != nil {
+	if err := zone.Start(); err != nil {
 		Error.Fatal("[main] Unable to start the Zone database", err)
 	}
 	defer zone.Stop()
 
+	dockerCli, err = docker.NewClient(apiPath)
+	if err != nil {
+		Error.Fatal("[main] Unable to start docker client: ", err)
+	}
+
 	if watch {
-		err := updater.Start(apiPath, zone)
+		err := dockerCli.AddObserver(zone)
 		if err != nil {
 			Error.Fatal("[main] Unable to start watcher", err)
 		}
@@ -140,7 +145,7 @@ func main() {
 	}
 
 	go SignalHandlerLoop(srv)
-	go weavedns.ServeHTTP(httpListener, version, srv, domain, zone)
+	go weavedns.ServeHTTP(httpListener, version, srv, dockerCli)
 
 	err = srv.Start()
 	if err != nil {

prog/weaver/main.go

@@ -15,7 +15,7 @@ import (
 	"github.com/davecheney/profile"
 	"github.com/gorilla/mux"
 	. "github.com/weaveworks/weave/common"
-	"github.com/weaveworks/weave/common/updater"
+	"github.com/weaveworks/weave/common/docker"
 	"github.com/weaveworks/weave/ipam"
 	"github.com/weaveworks/weave/ipam/address"
 	weavenet "github.com/weaveworks/weave/net"
@@ -222,10 +222,15 @@ func createAllocator(router *weave.Router, apiPath string, ipRangeStr string, de
 
 	allocator.SetInterfaces(router.NewGossip("IPallocation", allocator))
 	allocator.Start()
-	err := updater.Start(apiPath, allocator)
+
+	dockerCli, err := docker.NewClient(apiPath)
 	if err != nil {
-		log.Fatal("Unable to start watcher", err)
+		Error.Fatal("Unable to start docker client: ", err)
 	}
+	if err = dockerCli.AddObserver(allocator); err != nil {
+		Error.Fatal("Unable to start watcher", err)
+	}
+
 	return allocator, defaultSubnet
 }