Merge branch 'master' into 237_add_dns_to_weave_status

Conflicts:
	nameserver/http.go
This commit is contained in:
Adam Harrison committed 2015-03-18 11:58:39 +00:00
commit d23a6c86c3
9 files changed
+91 -95

No files matched your search

+53 -66
View File
@@ -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/<identifier>/<ip-address>
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,79 +19,76 @@ func httpErrorAndLog(level *log.Logger, w http.ResponseWriter, msg string,
func ListenHttp(version string, server *DNSServer, 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 {
muxRouter := mux.NewRouter()
muxRouter.Methods("GET").Path("/status").HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
io.WriteString(w, fmt.Sprintln("weave DNS", version))
io.WriteString(w, server.Status())
}
})
http.HandleFunc("/name/", 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...)
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)
idStr := vars["id"]
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(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.
}
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/{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)
idStr := vars["id"]
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)", idStr, ipStr)
if err := db.DeleteRecord(idStr, 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)
+6 -5
View File
@@ -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
+1 -6
View File
@@ -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)
}
}
+1 -5
View File
@@ -61,16 +61,12 @@ func (routes *Routes) MarshalJSON() ([]byte, error) {
}
func (peer *Peer) MarshalJSON() ([]byte, error) {
conns := make([]Connection, 0)
peer.ForEachConnection(func(remoteName 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) {
+7 -2
View File
@@ -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())
}
+5 -3
View File
@@ -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) {
+3 -3
View File
@@ -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()
}
+3 -2
View File
@@ -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
})
+12 -3
View File
@@ -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 {