From dbc683f891f3dee16d8de1fcdc3f7ad41b18f0c2 Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Mon, 16 Mar 2015 17:46:01 +0000 Subject: [PATCH 1/5] send all gossip to new peers only Weave sends all gossip down new connections. We do this so that new peers learn about all gossip'ed information quickly, thus enabling them to process subsequent incremental updates. In particular, for topology gossip it ensures that the remote peer learns about all other peers. This ensures that the incremental update, which is sent as a result of the new connection getting established and which may reference a number of other peers, can be processed. So the intent really is to send all gossip to new peers. But what we actually do is send it down every new connection. Not every new connection is from a new peer. In fact most connections aren't. Hence the current approach is wasteful. To fix this we need a conservative detection of new peers, i.e. it is preferable to have false positives. This contrasts with existing logic in the handshake code. The criteria we use is whether our peer has a unicast route to the other peer. That is only the case when there is a continuos chain of symmetric, established connections. The establishment of these connections should have triggered the sending of all gossip to the other peer. So we don't need to send it ourself. Fixes #454. --- router/local_peer.go | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/router/local_peer.go b/router/local_peer.go index 6f2b1979..d719dd20 100644 --- a/router/local_peer.go +++ b/router/local_peer.go @@ -209,9 +209,14 @@ func (peer *LocalPeer) handleAddConnection(conn Connection) { conn.Shutdown(err) return } + _, isConnectedPeer := peer.Router.Routes.Unicast(toName) peer.addConnection(conn) - conn.Log("connection added") - peer.Router.SendAllGossipDown(conn) + if isConnectedPeer { + conn.Log("connection added") + } else { + conn.Log("connection added (new peer)") + peer.Router.SendAllGossipDown(conn) + } peer.broadcastPeerUpdate(conn.Remote()) } From 08a08be37080dd2e30095b6aca44a48e0d022748 Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Tue, 17 Mar 2015 15:11:28 +0000 Subject: [PATCH 2/5] cosmetic --- router/json.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/router/json.go b/router/json.go index e3091b24..9d27fdde 100644 --- a/router/json.go +++ b/router/json.go @@ -62,7 +62,7 @@ func (routes *Routes) MarshalJSON() ([]byte, error) { func (peer *Peer) MarshalJSON() ([]byte, error) { conns := make([]Connection, 0) - peer.ForEachConnection(func(remoteName PeerName, conn Connection) { + peer.ForEachConnection(func(_ PeerName, conn Connection) { conns = append(conns, conn) }) return json.Marshal(struct { From d5eb76f32ec61cefbe6270759a46f25d4911f4ae Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Tue, 17 Mar 2015 16:24:19 +0000 Subject: [PATCH 3/5] replace Peer.ForEachConnection with Peer.Connections The former is prone to triggering deadlocks since the lock is held on the peer during the iteration. --- router/connection_maker.go | 11 ++++++----- router/gossip.go | 7 +------ router/json.go | 6 +----- router/peer.go | 8 +++++--- router/peers.go | 6 +++--- router/routes.go | 5 +++-- 6 files changed, 19 insertions(+), 24 deletions(-) diff --git a/router/connection_maker.go b/router/connection_maker.go index 6042db51..236c7650 100644 --- a/router/connection_maker.go +++ b/router/connection_maker.go @@ -119,10 +119,10 @@ func (cm *ConnectionMaker) checkStateAndAttemptConnections() time.Duration { // copy the set of things we are connected to, so we can access them without locking ourConnectedPeers := make(map[PeerName]bool) ourConnectedTargets := make(map[string]bool) - cm.ourself.ForEachConnection(func(peer PeerName, conn Connection) { - ourConnectedPeers[peer] = true + for _, conn := range cm.ourself.Connections() { + ourConnectedPeers[conn.Remote().Name] = true ourConnectedTargets[conn.RemoteTCPAddr()] = true - }) + } addTarget := func(address string) { if !ourConnectedTargets[address] { @@ -139,7 +139,8 @@ func (cm *ConnectionMaker) checkStateAndAttemptConnections() time.Duration { // Add targets for peers that someone else is connected to, but we // aren't cm.peers.ForEach(func(name PeerName, peer *Peer) { - peer.ForEachConnection(func(otherPeer PeerName, conn Connection) { + for _, conn := range peer.Connections() { + otherPeer := conn.Remote().Name if otherPeer == cm.ourself.Name || ourConnectedPeers[otherPeer] { return } @@ -151,7 +152,7 @@ func (cm *ConnectionMaker) checkStateAndAttemptConnections() time.Duration { if host, _, err := net.SplitHostPort(address); err == nil { addTarget(NormalisePeerAddr(host)) } - }) + } }) now := time.Now() // make sure we catch items just added diff --git a/router/gossip.go b/router/gossip.go index 1e50b707..21e1d373 100644 --- a/router/gossip.go +++ b/router/gossip.go @@ -117,12 +117,7 @@ func deliverGossip(channel *GossipChannel, srcName PeerName, _ []byte, dec *gob. func (c *GossipChannel) SendGossipMsg(buf []byte) { protocolMsg := c.gossipMsg(buf) - // Copy the connections so we don't hold a lock - connections := make([]Connection, 0) - c.ourself.ForEachConnection(func(_ PeerName, conn Connection) { - connections = append(connections, conn) - }) - for _, conn := range connections { + for _, conn := range c.ourself.Connections() { conn.(ProtocolSender).SendProtocolMsg(protocolMsg) } } diff --git a/router/json.go b/router/json.go index 9d27fdde..dadd4ff6 100644 --- a/router/json.go +++ b/router/json.go @@ -61,16 +61,12 @@ func (routes *Routes) MarshalJSON() ([]byte, error) { } func (peer *Peer) MarshalJSON() ([]byte, error) { - conns := make([]Connection, 0) - peer.ForEachConnection(func(_ PeerName, conn Connection) { - conns = append(conns, conn) - }) return json.Marshal(struct { Name string UID uint64 Version uint64 Connections []Connection - }{peer.Name.String(), peer.UID, peer.version, conns}) + }{peer.Name.String(), peer.UID, peer.version, peer.Connections()}) } func (conn *RemoteConnection) MarshalJSON() ([]byte, error) { diff --git a/router/peer.go b/router/peer.go index 86047e3a..3e61e529 100644 --- a/router/peer.go +++ b/router/peer.go @@ -73,12 +73,14 @@ func (peer *Peer) ConnectionTo(name PeerName) (Connection, bool) { return conn, found // yes, you really can't inline that. FFS. } -func (peer *Peer) ForEachConnection(fun func(PeerName, Connection)) { +func (peer *Peer) Connections() []Connection { peer.RLock() defer peer.RUnlock() - for name, conn := range peer.connections { - fun(name, conn) + connections := make([]Connection, 0, len(peer.connections)) + for _, conn := range peer.connections { + connections = append(connections, conn) } + return connections } func (peer *Peer) SetVersionAndConnections(version uint64, connections map[PeerName]Connection) { diff --git a/router/peers.go b/router/peers.go index 514de2fd..9d9928ec 100644 --- a/router/peers.go +++ b/router/peers.go @@ -123,13 +123,13 @@ func (peers *Peers) String() string { var buf bytes.Buffer peers.ForEach(func(name PeerName, peer *Peer) { buf.WriteString(fmt.Sprint(peer, "\n")) - peer.ForEachConnection(func(remoteName PeerName, conn Connection) { + for _, conn := range peer.Connections() { established := "" if !conn.Established() { established = " (unestablished)" } - buf.WriteString(fmt.Sprintf(" -> %v (%v) [%v%s]\n", remoteName, conn.Remote().NickName, conn.RemoteTCPAddr(), established)) - }) + buf.WriteString(fmt.Sprintf(" -> %v (%v) [%v%s]\n", conn.Remote().Name, conn.Remote().NickName, conn.RemoteTCPAddr(), established)) + } }) return buf.String() } diff --git a/router/routes.go b/router/routes.go index 0e109fe4..89c639a6 100644 --- a/router/routes.go +++ b/router/routes.go @@ -140,17 +140,18 @@ func (routes *Routes) calculateBroadcast() map[PeerName][]PeerName { // This is rather similar to the inner loop on // peer.Routes(...); the main difference is in the // locking. - ourself.ForEachConnection(func(remoteName PeerName, conn Connection) { + for _, conn := range ourself.Connections() { if !conn.Established() { return } + remoteName := conn.Remote().Name if _, found := reached[remoteName]; found { return } if remoteConn, found := conn.Remote().ConnectionTo(ourself.Name); found && remoteConn.Established() { hops = append(hops, remoteName) } - }) + } } broadcast[name] = hops }) From bd7b232663edc9241af804de042c4290ec7b761e Mon Sep 17 00:00:00 2001 From: Adam Harrison Date: Tue, 17 Mar 2015 13:49:26 +0000 Subject: [PATCH 4/5] Replace hand parsing and validation of URLs with Gorilla mux --- nameserver/http.go | 121 ++++++++++++++++++++------------------------- weaver/main.go | 15 ++++-- 2 files changed, 66 insertions(+), 70 deletions(-) diff --git a/nameserver/http.go b/nameserver/http.go index c028c28b..78b4c5d1 100644 --- a/nameserver/http.go +++ b/nameserver/http.go @@ -1,7 +1,6 @@ package nameserver import ( - "errors" "fmt" "github.com/miekg/dns" . "github.com/zettio/weave/common" @@ -9,18 +8,9 @@ import ( "log" "net" "net/http" - "strings" + "github.com/gorilla/mux" ) -// Parse a URL of the form PUT /name// -func parseUrl(url string) (identifier string, ipaddr string, err error) { - parts := strings.Split(url, "/") - if len(parts) != 4 { - return "", "", errors.New("Unable to parse url: " + url) - } - return parts[2], parts[3], nil -} - func httpErrorAndLog(level *log.Logger, w http.ResponseWriter, msg string, status int, logmsg string, logargs ...interface{}) { http.Error(w, msg, status) @@ -29,78 +19,75 @@ func httpErrorAndLog(level *log.Logger, w http.ResponseWriter, msg string, func ListenHttp(domain string, db Zone, port int) { - http.HandleFunc("/status", func(w http.ResponseWriter, r *http.Request) { - if r.Method != "GET" { - http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) - } else { - io.WriteString(w, "ok") - } + muxRouter := mux.NewRouter() + + muxRouter.Methods("GET").Path("/status").HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + io.WriteString(w, "ok") }) - http.HandleFunc("/name/", func(w http.ResponseWriter, r *http.Request) { - + muxRouter.Methods("PUT").Path("/name/{identifier:.+}/{ip:.+}").HandlerFunc(func(w http.ResponseWriter, r *http.Request) { reqError := func(msg string, logmsg string, logargs ...interface{}) { - httpErrorAndLog(Warning, w, msg, http.StatusBadRequest, - logmsg, logargs...) + httpErrorAndLog(Warning, w, msg, http.StatusBadRequest, logmsg, logargs...) } - switch r.Method { - case "PUT": - ident, ipStr, err := parseUrl(r.URL.Path) - name := r.FormValue("fqdn") - if ident == "" || ipStr == "" || name == "" { - reqError("Invalid request", "Invalid request: %s, %s", r.URL, r.Form) - return - } + vars := mux.Vars(r) + ident := vars["identifier"] + ipStr := vars["ip"] + name := r.FormValue("fqdn") - ip := net.ParseIP(ipStr) - if ip == nil { - reqError("Invalid IP", "Invalid IP in request: %s", ipStr) - return - } + if name == "" { + reqError("Invalid FQDN", "Invalid FQDN in request: %s, %s", r.URL, r.Form) + return + } - if dns.IsSubDomain(domain, name) { - Info.Printf("[http] Adding %s -> %s", name, ipStr) - if err = db.AddRecord(ident, 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 { - Info.Printf("[http] Ignoring name %s, not in %s", name, domain) - } + ip := net.ParseIP(ipStr) + if ip == nil { + reqError("Invalid IP", "Invalid IP in request: %s", ipStr) + return + } - case "DELETE": - ident, ipStr, err := parseUrl(r.URL.Path) - if ident == "" || ipStr == "" { - reqError("Invalid Request", "Invalid request: %s, %s", r.URL, r.Form) - return - } - - ip := net.ParseIP(ipStr) - if ip == nil { - reqError("Invalid IP in request", "Invalid IP in request: %s", ipStr) - return - } - Info.Printf("[http] Deleting %s (%s)", ident, ipStr) - if err = db.DeleteRecord(ident, ip); err != nil { - if _, ok := err.(LookupError); !ok { + if dns.IsSubDomain(domain, name) { + Info.Printf("[http] Adding %s -> %s", name, ipStr) + if err := db.AddRecord(ident, 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. } - default: - msg := "Unexpected http method: " + r.Method - reqError(msg, msg) - return + } else { + Info.Printf("[http] Ignoring name %s, not in %s", name, domain) } }) + muxRouter.Methods("DELETE").Path("/name/{identifier:.+}/{ip:.+}").HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + reqError := func(msg string, logmsg string, logargs ...interface{}) { + httpErrorAndLog(Warning, w, msg, http.StatusBadRequest, logmsg, logargs...) + } + + vars := mux.Vars(r) + ident := vars["identifier"] + ipStr := vars["ip"] + + ip := net.ParseIP(ipStr) + if ip == nil { + reqError("Invalid IP in request", "Invalid IP in request: %s", ipStr) + return + } + Info.Printf("[http] Deleting %s (%s)", ident, ipStr) + if err := db.DeleteRecord(ident, ip); err != nil { + if _, ok := err.(LookupError); !ok { + httpErrorAndLog( + Error, w, "Internal error", http.StatusInternalServerError, + "Unexpected error from DB: %s", err) + return + } + } + }) + + http.Handle("/", muxRouter) + address := fmt.Sprintf(":%d", port) if err := http.ListenAndServe(address, nil); err != nil { Error.Fatal("[http] Unable to create http listener: ", err) diff --git a/weaver/main.go b/weaver/main.go index d1cc2097..8004c1c6 100644 --- a/weaver/main.go +++ b/weaver/main.go @@ -16,6 +16,7 @@ import ( "os/signal" "runtime" "syscall" + "github.com/gorilla/mux" ) var version = "(unreleased version)" @@ -152,16 +153,21 @@ func handleHttp(router *weave.Router) { if router.UsingPassword() { encryption = "on" } - http.HandleFunc("/status", func(w http.ResponseWriter, r *http.Request) { + + muxRouter := mux.NewRouter() + + muxRouter.Methods("GET").Path("/status").HandlerFunc(func(w http.ResponseWriter, r *http.Request) { io.WriteString(w, fmt.Sprintln("weave router", version)) io.WriteString(w, fmt.Sprintln("Encryption", encryption)) io.WriteString(w, router.Status()) }) - http.HandleFunc("/status-json", func(w http.ResponseWriter, r *http.Request) { + + muxRouter.Methods("GET").Path("/status-json").HandlerFunc(func(w http.ResponseWriter, r *http.Request) { json, _ := router.GenerateStatusJSON(version, encryption) w.Write(json) }) - http.HandleFunc("/connect", func(w http.ResponseWriter, r *http.Request) { + + muxRouter.Methods("POST").Path("/connect").HandlerFunc(func(w http.ResponseWriter, r *http.Request) { peer := r.FormValue("peer") if addr, err := net.ResolveTCPAddr("tcp4", weave.NormalisePeerAddr(peer)); err == nil { router.ConnectionMaker.InitiateConnection(addr.String()) @@ -169,6 +175,9 @@ func handleHttp(router *weave.Router) { http.Error(w, fmt.Sprint("invalid peer address: ", err), http.StatusBadRequest) } }) + + http.Handle("/", muxRouter) + address := fmt.Sprintf(":%d", weave.HttpPort) err := http.ListenAndServe(address, nil) if err != nil { From bae6865e33f55cc7ee5814354ceff74b09df196d Mon Sep 17 00:00:00 2001 From: Adam Harrison Date: Wed, 18 Mar 2015 11:45:04 +0000 Subject: [PATCH 5/5] Rename URL parameter --- nameserver/http.go | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/nameserver/http.go b/nameserver/http.go index 78b4c5d1..81e5d040 100644 --- a/nameserver/http.go +++ b/nameserver/http.go @@ -25,13 +25,13 @@ func ListenHttp(domain string, db Zone, port int) { io.WriteString(w, "ok") }) - muxRouter.Methods("PUT").Path("/name/{identifier:.+}/{ip:.+}").HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + muxRouter.Methods("PUT").Path("/name/{id:.+}/{ip:.+}").HandlerFunc(func(w http.ResponseWriter, r *http.Request) { reqError := func(msg string, logmsg string, logargs ...interface{}) { httpErrorAndLog(Warning, w, msg, http.StatusBadRequest, logmsg, logargs...) } vars := mux.Vars(r) - ident := vars["identifier"] + idStr := vars["id"] ipStr := vars["ip"] name := r.FormValue("fqdn") @@ -48,7 +48,7 @@ func ListenHttp(domain string, db Zone, port int) { if dns.IsSubDomain(domain, name) { Info.Printf("[http] Adding %s -> %s", name, ipStr) - if err := db.AddRecord(ident, name, ip); err != nil { + if err := db.AddRecord(idStr, name, ip); err != nil { if _, ok := err.(DuplicateError); !ok { httpErrorAndLog( Error, w, "Internal error", http.StatusInternalServerError, @@ -61,13 +61,13 @@ func ListenHttp(domain string, db Zone, port int) { } }) - muxRouter.Methods("DELETE").Path("/name/{identifier:.+}/{ip:.+}").HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + muxRouter.Methods("DELETE").Path("/name/{id:.+}/{ip:.+}").HandlerFunc(func(w http.ResponseWriter, r *http.Request) { reqError := func(msg string, logmsg string, logargs ...interface{}) { httpErrorAndLog(Warning, w, msg, http.StatusBadRequest, logmsg, logargs...) } vars := mux.Vars(r) - ident := vars["identifier"] + idStr := vars["id"] ipStr := vars["ip"] ip := net.ParseIP(ipStr) @@ -75,8 +75,8 @@ func ListenHttp(domain string, db Zone, port int) { reqError("Invalid IP in request", "Invalid IP in request: %s", ipStr) return } - Info.Printf("[http] Deleting %s (%s)", ident, ipStr) - if err := db.DeleteRecord(ident, ip); err != nil { + Info.Printf("[http] Deleting %s (%s)", idStr, ipStr) + if err := db.DeleteRecord(idStr, ip); err != nil { if _, ok := err.(LookupError); !ok { httpErrorAndLog( Error, w, "Internal error", http.StatusInternalServerError,