Add flag to override tokenLimit

api/machines.go

@@ -25,14 +25,15 @@ import (
 	"github.com/coreos/fleet/schema"
 )
 
-func wireUpMachinesResource(mux *http.ServeMux, prefix string, cAPI client.API) {
+func wireUpMachinesResource(mux *http.ServeMux, prefix string, tokenLimit int, cAPI client.API) {
 	res := path.Join(prefix, "machines")
-	mr := machinesResource{cAPI}
+	mr := machinesResource{cAPI, uint16(tokenLimit)}
 	mux.Handle(res, &mr)
 }
 
 type machinesResource struct {
-	cAPI client.API
+	cAPI       client.API
+	tokenLimit uint16
 }
 
 func (mr *machinesResource) ServeHTTP(rw http.ResponseWriter, req *http.Request) {
@@ -41,14 +42,14 @@ func (mr *machinesResource) ServeHTTP(rw http.ResponseWriter, req *http.Request)
 		return
 	}
 
-	token, err := findNextPageToken(req.URL)
+	token, err := findNextPageToken(req.URL, mr.tokenLimit)
 	if err != nil {
 		sendError(rw, http.StatusBadRequest, err)
 		return
 	}
 
 	if token == nil {
-		def := DefaultPageToken()
+		def := DefaultPageToken(mr.tokenLimit)
 		token = &def
 	}
 

api/machines_test.go

@@ -33,7 +33,7 @@ func TestMachinesList(t *testing.T) {
 		{ID: "YYY", PublicIP: "1.2.3.4", Metadata: map[string]string{"ping": "pong"}},
 	})
 	fAPI := &client.RegistryClient{Registry: fr}
-	resource := &machinesResource{cAPI: fAPI}
+	resource := &machinesResource{cAPI: fAPI, tokenLimit: testTokenLimit}
 	rw := httptest.NewRecorder()
 	req, err := http.NewRequest("GET", "http://example.com", nil)
 	if err != nil {
@@ -66,7 +66,7 @@ func TestMachinesList(t *testing.T) {
 func TestMachinesListBadNextPageToken(t *testing.T) {
 	fr := registry.NewFakeRegistry()
 	fAPI := &client.RegistryClient{Registry: fr}
-	resource := &machinesResource{fAPI}
+	resource := &machinesResource{fAPI, testTokenLimit}
 	rw := httptest.NewRecorder()
 	req, err := http.NewRequest("GET", "http://example.com/machines?nextPageToken=EwBMLg==", nil)
 	if err != nil {

api/mux.go

@@ -23,15 +23,16 @@ import (
 	"github.com/coreos/fleet/version"
 )
 
-func NewServeMux(reg registry.Registry) http.Handler {
+func NewServeMux(reg registry.Registry, tokenLimit int) http.Handler {
 	sm := http.NewServeMux()
 	cAPI := &client.RegistryClient{Registry: reg}
 
 	for _, prefix := range []string{"/v1-alpha", "/fleet/v1"} {
 		wireUpDiscoveryResource(sm, prefix)
-		wireUpMachinesResource(sm, prefix, cAPI)
-		wireUpStateResource(sm, prefix, cAPI)
-		wireUpUnitsResource(sm, prefix, cAPI)
+
+		wireUpMachinesResource(sm, prefix, tokenLimit, cAPI)
+		wireUpStateResource(sm, prefix, tokenLimit, cAPI)
+		wireUpUnitsResource(sm, prefix, tokenLimit, cAPI)
 		sm.HandleFunc(prefix, methodNotAllowedHandler)
 	}
 

api/mux_test.go

@@ -24,6 +24,8 @@ import (
 	"github.com/coreos/fleet/version"
 )
 
+const testTokenLimit = 100
+
 func TestDefaultHandlers(t *testing.T) {
 	tests := []struct {
 		method string
@@ -38,7 +40,7 @@ func TestDefaultHandlers(t *testing.T) {
 
 	for i, tt := range tests {
 		fr := registry.NewFakeRegistry()
-		hdlr := NewServeMux(fr)
+		hdlr := NewServeMux(fr, testTokenLimit)
 		rr := httptest.NewRecorder()
 
 		req, err := http.NewRequest(tt.method, tt.path, nil)

api/pagination.go

@@ -23,18 +23,13 @@ import (
 	"net/url"
 )
 
-const (
-	// Support a single value for PageToken.Limit to make life easy
-	tokenLimit = 100
-)
-
 type PageToken struct {
 	Limit uint16
 	Page  uint16
 }
 
-func DefaultPageToken() PageToken {
-	return PageToken{Limit: tokenLimit, Page: 1}
+func DefaultPageToken(limit uint16) PageToken {
+	return PageToken{Limit: limit, Page: 1}
 }
 
 func (tok PageToken) Next() PageToken {
@@ -63,7 +58,7 @@ func decodePageToken(value string) (*PageToken, error) {
 	return &tok, nil
 }
 
-func findNextPageToken(u *url.URL) (*PageToken, error) {
+func findNextPageToken(u *url.URL, limit uint16) (*PageToken, error) {
 	values := u.Query()["nextPageToken"]
 
 	if len(values) > 1 {
@@ -80,17 +75,17 @@ func findNextPageToken(u *url.URL) (*PageToken, error) {
 		return nil, err
 	}
 
-	err = validatePageToken(tok)
+	err = validatePageToken(tok, limit)
 	if err != nil {
 		return nil, err
 	}
 
 	return tok, nil
 }
 
-func validatePageToken(tok *PageToken) error {
-	if tok.Limit != tokenLimit {
-		return fmt.Errorf("token limit must be %d", tokenLimit)
+func validatePageToken(tok *PageToken, limit uint16) error {
+	if tok.Limit != limit {
+		return fmt.Errorf("token limit must be %d", limit)
 	}
 
 	if tok.Page == 0 {

api/pagination_test.go

@@ -21,7 +21,7 @@ import (
 )
 
 func TestDefaultPageToken(t *testing.T) {
-	tok := DefaultPageToken()
+	tok := DefaultPageToken(testTokenLimit)
 	expect := PageToken{Limit: 100, Page: 1}
 	if !reflect.DeepEqual(expect, tok) {
 		t.Errorf("Unexpected default PageToken: expect=%v, got=%v", expect, tok)
@@ -108,7 +108,7 @@ func TestFindNextPageToken(t *testing.T) {
 	}
 
 	for i, tt := range tests {
-		next, err := findNextPageToken(&tt.input)
+		next, err := findNextPageToken(&tt.input, testTokenLimit)
 
 		if tt.pass != (err == nil) {
 			t.Errorf("case %d: pass=%t, err=%v", i, tt.pass, err)
@@ -135,7 +135,7 @@ func TestValidatePageToken(t *testing.T) {
 	}
 
 	for i, tt := range tests {
-		err := validatePageToken(&tt.input)
+		err := validatePageToken(&tt.input, testTokenLimit)
 
 		if tt.pass != (err == nil) {
 			t.Errorf("case %d: pass=%t, err=%v", i, tt.pass, err)

api/state.go

@@ -24,15 +24,16 @@ import (
 	"github.com/coreos/fleet/schema"
 )
 
-func wireUpStateResource(mux *http.ServeMux, prefix string, cAPI client.API) {
+func wireUpStateResource(mux *http.ServeMux, prefix string, tokenLimit int, cAPI client.API) {
 	base := path.Join(prefix, "state")
-	sr := stateResource{cAPI, base}
+	sr := stateResource{cAPI, base, uint16(tokenLimit)}
 	mux.Handle(base, &sr)
 }
 
 type stateResource struct {
-	cAPI     client.API
-	basePath string
+	cAPI       client.API
+	basePath   string
+	tokenLimit uint16
 }
 
 func (sr *stateResource) ServeHTTP(rw http.ResponseWriter, req *http.Request) {
@@ -45,14 +46,14 @@ func (sr *stateResource) ServeHTTP(rw http.ResponseWriter, req *http.Request) {
 }
 
 func (sr *stateResource) list(rw http.ResponseWriter, req *http.Request) {
-	token, err := findNextPageToken(req.URL)
+	token, err := findNextPageToken(req.URL, sr.tokenLimit)
 	if err != nil {
 		sendError(rw, http.StatusBadRequest, err)
 		return
 	}
 
 	if token == nil {
-		def := DefaultPageToken()
+		def := DefaultPageToken(sr.tokenLimit)
 		token = &def
 	}
 

api/state_test.go

@@ -81,7 +81,7 @@ func TestUnitStateList(t *testing.T) {
 		fr := registry.NewFakeRegistry()
 		fr.SetUnitStates([]unit.UnitState{us1, us2, us3, us4})
 		fAPI := &client.RegistryClient{Registry: fr}
-		resource := &stateResource{fAPI, "/state"}
+		resource := &stateResource{fAPI, "/state", testTokenLimit}
 		rw := httptest.NewRecorder()
 		req, err := http.NewRequest("GET", tt.url, nil)
 		if err != nil {
@@ -130,7 +130,7 @@ func TestUnitStateList(t *testing.T) {
 		unit.UnitState{UnitName: "YYY", ActiveState: "inactive"},
 	})
 	fAPI := &client.RegistryClient{Registry: fr}
-	resource := &stateResource{fAPI, "/state"}
+	resource := &stateResource{fAPI, "/state", testTokenLimit}
 	rw := httptest.NewRecorder()
 	req, err := http.NewRequest("GET", "http://example.com/state", nil)
 	if err != nil {
@@ -178,7 +178,7 @@ func TestUnitStateList(t *testing.T) {
 func TestUnitStateListBadNextPageToken(t *testing.T) {
 	fr := registry.NewFakeRegistry()
 	fAPI := &client.RegistryClient{Registry: fr}
-	resource := &stateResource{fAPI, "/state"}
+	resource := &stateResource{fAPI, "/state", testTokenLimit}
 	rw := httptest.NewRecorder()
 	req, err := http.NewRequest("GET", "http://example.com/state?nextPageToken=EwBMLg==", nil)
 	if err != nil {

api/units.go

@@ -30,16 +30,17 @@ import (
 	"github.com/coreos/fleet/unit"
 )
 
-func wireUpUnitsResource(mux *http.ServeMux, prefix string, cAPI client.API) {
+func wireUpUnitsResource(mux *http.ServeMux, prefix string, tokenLimit int, cAPI client.API) {
 	base := path.Join(prefix, "units")
-	ur := unitsResource{cAPI, base}
+	ur := unitsResource{cAPI, base, uint16(tokenLimit)}
 	mux.Handle(base, &ur)
 	mux.Handle(base+"/", &ur)
 }
 
 type unitsResource struct {
-	cAPI     client.API
-	basePath string
+	cAPI       client.API
+	basePath   string
+	tokenLimit uint16
 }
 
 func (ur *unitsResource) ServeHTTP(rw http.ResponseWriter, req *http.Request) {
@@ -286,14 +287,14 @@ func (ur *unitsResource) get(rw http.ResponseWriter, req *http.Request, item str
 }
 
 func (ur *unitsResource) list(rw http.ResponseWriter, req *http.Request) {
-	token, err := findNextPageToken(req.URL)
+	token, err := findNextPageToken(req.URL, ur.tokenLimit)
 	if err != nil {
 		sendError(rw, http.StatusBadRequest, err)
 		return
 	}
 
 	if token == nil {
-		def := DefaultPageToken()
+		def := DefaultPageToken(ur.tokenLimit)
 		token = &def
 	}
 

api/units_test.go

@@ -43,7 +43,7 @@ func newUnit(t *testing.T, str string) unit.UnitFile {
 func TestUnitsSubResourceNotFound(t *testing.T) {
 	fr := registry.NewFakeRegistry()
 	fAPI := &client.RegistryClient{Registry: fr}
-	ur := &unitsResource{fAPI, "/units"}
+	ur := &unitsResource{fAPI, "/units", testTokenLimit}
 	rr := httptest.NewRecorder()
 
 	req, err := http.NewRequest("GET", "/units/foo/bar", nil)
@@ -66,7 +66,7 @@ func TestUnitsList(t *testing.T) {
 		{Name: "YYY.service"},
 	})
 	fAPI := &client.RegistryClient{Registry: fr}
-	resource := &unitsResource{fAPI, "/units"}
+	resource := &unitsResource{fAPI, "/units", testTokenLimit}
 	rw := httptest.NewRecorder()
 	req, err := http.NewRequest("GET", "http://example.com/units", nil)
 	if err != nil {
@@ -103,7 +103,7 @@ func TestUnitsList(t *testing.T) {
 func TestUnitsListBadNextPageToken(t *testing.T) {
 	fr := registry.NewFakeRegistry()
 	fAPI := &client.RegistryClient{Registry: fr}
-	resource := &unitsResource{fAPI, "/units"}
+	resource := &unitsResource{fAPI, "/units", testTokenLimit}
 	rw := httptest.NewRecorder()
 	req, err := http.NewRequest("GET", "http://example.com/units?nextPageToken=EwBMLg==", nil)
 	if err != nil {
@@ -184,7 +184,7 @@ func TestUnitGet(t *testing.T) {
 		{Name: "YYY.service"},
 	})
 	fAPI := &client.RegistryClient{Registry: fr}
-	resource := &unitsResource{fAPI, "/units"}
+	resource := &unitsResource{fAPI, "/units", testTokenLimit}
 
 	for i, tt := range tests {
 		rw := httptest.NewRecorder()
@@ -247,7 +247,7 @@ func TestUnitsDestroy(t *testing.T) {
 		}
 
 		fAPI := &client.RegistryClient{Registry: fr}
-		resource := &unitsResource{fAPI, "/units"}
+		resource := &unitsResource{fAPI, "/units", testTokenLimit}
 		rw := httptest.NewRecorder()
 		resource.destroy(rw, req, tt.arg)
 
@@ -415,7 +415,7 @@ func TestUnitsSetDesiredState(t *testing.T) {
 		req.Header.Set("Content-Type", "application/json")
 
 		fAPI := &client.RegistryClient{Registry: fr}
-		resource := &unitsResource{fAPI, "/units"}
+		resource := &unitsResource{fAPI, "/units", testTokenLimit}
 		rw := httptest.NewRecorder()
 		resource.set(rw, req, tt.item)
 
@@ -718,7 +718,7 @@ func TestValidateName(t *testing.T) {
 func TestUnitsSetDesiredStateBadContentType(t *testing.T) {
 	fr := registry.NewFakeRegistry()
 	fAPI := &client.RegistryClient{Registry: fr}
-	resource := &unitsResource{fAPI, "/units"}
+	resource := &unitsResource{fAPI, "/units", testTokenLimit}
 	rr := httptest.NewRecorder()
 
 	body := ioutil.NopCloser(bytes.NewBuffer([]byte(`{"foo":"bar"}`)))

config/config.go

@@ -30,6 +30,7 @@ type Config struct {
 	Verbosity               int
 	RawMetadata             string
 	AgentTTL                string
+	TokenLimit              int
 	VerifyUnits             bool
 	AuthorizedKeysFile      string
 }

fleetd/fleetd.go

@@ -75,6 +75,7 @@ func main() {
 	cfgset.String("public_ip", "", "IP address that fleet machine should publish")
 	cfgset.String("metadata", "", "List of key-value metadata to assign to the fleet machine")
 	cfgset.String("agent_ttl", agent.DefaultTTL, "TTL in seconds of fleet machine state in etcd")
+	cfgset.Int("token_limit", 100, "Maximum number of entries per page returned from API requests")
 	cfgset.Bool("verify_units", false, "DEPRECATED - This option is ignored")
 	cfgset.String("authorized_keys_file", "", "DEPRECATED - This option is ignored")
 
@@ -188,6 +189,7 @@ func getConfig(flagset *flag.FlagSet, userCfgFile string) (*config.Config, error
 		RawMetadata:             (*flagset.Lookup("metadata")).Value.(flag.Getter).Get().(string),
 		AgentTTL:                (*flagset.Lookup("agent_ttl")).Value.(flag.Getter).Get().(string),
 		VerifyUnits:             (*flagset.Lookup("verify_units")).Value.(flag.Getter).Get().(bool),
+		TokenLimit:              (*flagset.Lookup("token_limit")).Value.(flag.Getter).Get().(int),
 		AuthorizedKeysFile:      (*flagset.Lookup("authorized_keys_file")).Value.(flag.Getter).Get().(string),
 	}
 

server/server.go

@@ -114,7 +114,7 @@ func New(cfg config.Config) (*Server, error) {
 	hrt := heart.New(reg, mach)
 	mon := heart.NewMonitor(agentTTL)
 
-	apiServer := api.NewServer(listeners, api.NewServeMux(reg))
+	apiServer := api.NewServer(listeners, api.NewServeMux(reg, cfg.TokenLimit))
 	apiServer.Serve()
 
 	eIval := time.Duration(cfg.EngineReconcileInterval*1000) * time.Millisecond