Merge pull request kubernetes#188 from eyakubovich/cursor-fix

Make sure the cursor is a string in JSON

remote/client.go

@@ -124,7 +124,12 @@ func (m *RemoteManager) WatchLeases(ctx context.Context, network string, cursor
 	url := m.mkurl(network, "leases")
 
 	if cursor != nil {
-		url = fmt.Sprintf("%v?next=%v", url, cursor)
+		c, ok := cursor.(string)
+		if !ok {
+			return subnet.WatchResult{}, fmt.Errorf("internal error: RemoteManager.WatchLeases received non-string cursor")
+		}
+
+		url = fmt.Sprintf("%v?next=%v", url, c)
 	}
 
 	resp, err := httpGet(ctx, url)
@@ -140,6 +145,9 @@ func (m *RemoteManager) WatchLeases(ctx context.Context, network string, cursor
 	if err := json.NewDecoder(resp.Body).Decode(&wr); err != nil {
 		return subnet.WatchResult{}, err
 	}
+	if _, ok := wr.Cursor.(string); !ok {
+		return subnet.WatchResult{}, fmt.Errorf("lease watch returned non-string cursor")
+	}
 
 	return wr, nil
 }

remote/server.go

@@ -20,7 +20,6 @@ import (
 	"net"
 	"net/http"
 	"net/url"
-	"strconv"
 
 	log "github.com/coreos/flannel/Godeps/_workspace/src/github.com/golang/glog"
 	"github.com/coreos/flannel/Godeps/_workspace/src/github.com/gorilla/mux"
@@ -109,13 +108,12 @@ func handleRenewLease(ctx context.Context, sm subnet.Manager, w http.ResponseWri
 	jsonResponse(w, http.StatusOK, lease)
 }
 
-func getCursor(u *url.URL) (interface{}, error) {
+func getCursor(u *url.URL) interface{} {
 	vals, ok := u.Query()["next"]
 	if !ok {
-		return nil, nil
+		return nil
 	}
-	index, err := strconv.ParseUint(vals[0], 10, 64)
-	return index, err
+	return vals[0]
 }
 
 // GET /{network}/leases?next=cursor
@@ -127,12 +125,7 @@ func handleWatchLeases(ctx context.Context, sm subnet.Manager, w http.ResponseWr
 		network = ""
 	}
 
-	cursor, err := getCursor(r.URL)
-	if err != nil {
-		w.WriteHeader(http.StatusBadRequest)
-		fmt.Fprint(w, "invalid 'next' value: ", err)
-		return
-	}
+	cursor := getCursor(r.URL)
 
 	wr, err := sm.WatchLeases(ctx, network, cursor)
 	if err != nil {
@@ -141,6 +134,16 @@ func handleWatchLeases(ctx context.Context, sm subnet.Manager, w http.ResponseWr
 		return
 	}
 
+	switch wr.Cursor.(type) {
+	case string:
+	case fmt.Stringer:
+		wr.Cursor = wr.Cursor.(fmt.Stringer).String()
+	default:
+		w.WriteHeader(http.StatusInternalServerError)
+		fmt.Fprint(w, fmt.Errorf("internal error: watch cursor is of unknown type"))
+		return
+	}
+
 	jsonResponse(w, http.StatusOK, wr)
 }
 

subnet/etcd.go

@@ -49,6 +49,14 @@ var (
 	subnetRegex *regexp.Regexp = regexp.MustCompile(`(\d+\.\d+.\d+.\d+)-(\d+)`)
 )
 
+type watchCursor struct {
+	index uint64
+}
+
+func (c watchCursor) String() string {
+	return strconv.FormatUint(c.index, 10)
+}
+
 func NewEtcdManager(config *EtcdConfig) (Manager, error) {
 	r, err := newEtcdSubnetRegistry(config)
 	if err != nil {
@@ -279,7 +287,20 @@ func (m *EtcdManager) WatchLeases(ctx context.Context, network string, cursor in
 		return m.watchReset(ctx, network)
 	}
 
-	nextIndex := cursor.(uint64)
+	nextIndex := uint64(0)
+
+	if wc, ok := cursor.(watchCursor); ok {
+		nextIndex = wc.index
+	} else if s, ok := cursor.(string); ok {
+		var err error
+		nextIndex, err = strconv.ParseUint(s, 10, 64)
+		if err != nil {
+			return WatchResult{}, fmt.Errorf("failed to parse cursor: %v", err)
+		}
+	} else {
+		return WatchResult{}, fmt.Errorf("internal error: watch cursor is of unknown type")
+	}
+
 	resp, err := m.registry.watchSubnets(ctx, network, nextIndex)
 
 	switch {
@@ -337,7 +358,7 @@ func parseSubnetWatchResponse(resp *etcd.Response) (WatchResult, error) {
 		}
 	}
 
-	cursor := resp.Node.ModifiedIndex + 1
+	cursor := watchCursor{resp.Node.ModifiedIndex + 1}
 
 	return WatchResult{
 		Cursor: cursor,
@@ -354,7 +375,7 @@ func (m *EtcdManager) watchReset(ctx context.Context, network string) (WatchResu
 		return wr, fmt.Errorf("failed to retrieve subnet leases: %v", err)
 	}
 
-	cursor := index + 1
+	cursor := watchCursor{index + 1}
 	wr.Snapshot = leases
 	wr.Cursor = cursor
 	return wr, nil