Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Unique limiters for each API listener #1904

Merged
merged 13 commits into from
Sep 29, 2022
Merged

Unique limiters for each API listener #1904

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
08fc24a
Unique limiters for each API listener
michel-laterman Sep 21, 2022
716e8d3
Merge branch 'main' into separate-limiters
michel-laterman Sep 22, 2022
2e626b8
Add listener address labeled logs to limiter
michel-laterman Sep 22, 2022
676a12b
Review feedback
michel-laterman Sep 23, 2022
bfb16e6
Apply suggestions from code review
michel-laterman Sep 26, 2022
b6b7ca2
Merge remote-tracking branch 'origin/main' into separate-limiters
michel-laterman Sep 26, 2022
ec7e9bf
review feedback
michel-laterman Sep 26, 2022
71193cf
fix import
michel-laterman Sep 26, 2022
8b78ecc
Fix test
michel-laterman Sep 26, 2022
bd1475e
Merge branch 'main' into separate-limiters
michel-laterman Sep 26, 2022
d1dde1f
Merge branch 'main' into separate-limiters
michel-laterman Sep 27, 2022
4c3efa2
Merge branch 'main' into separate-limiters
michel-laterman Sep 28, 2022
0be91b4
Merge branch 'main' into separate-limiters
michel-laterman Sep 29, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.next.asciidoc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
- Add error detail to catch-all HTTP error response. {pull}1854[1854]
- Fix issue were errors where being ignored written to elasticsearch. {pull}1896[1896]
- Update apikey.cache_hit log field name to match convention. {pull}1900[1900]
- Use seperate rate limiters for internal and external API listeners. {issue}1859[1859] {pull}1904[1904]

==== New Features

Expand Down
4 changes: 2 additions & 2 deletions cmd/fleet/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -907,10 +907,10 @@ func (f *FleetServer) runSubsystems(ctx context.Context, cfg *config.Config, g *
ack := api.NewAckT(&cfg.Inputs[0].Server, bulker, f.cache)
st := api.NewStatusT(&cfg.Inputs[0].Server, bulker, f.cache)

router := api.NewRouter(ctx, bulker, ct, et, at, ack, st, sm, tracer, f.bi)
router := api.NewRouter(&cfg.Inputs[0].Server, bulker, ct, et, at, ack, st, sm, tracer, f.bi)

g.Go(loggedRunFunc(ctx, "Http server", func(ctx context.Context) error {
return api.Run(ctx, router, &cfg.Inputs[0].Server)
return router.Run(ctx)
}))

return err
Expand Down
20 changes: 0 additions & 20 deletions internal/pkg/api/error.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,6 @@ import (
"strings"

"github.com/elastic/fleet-server/v7/internal/pkg/dl"
"github.com/elastic/fleet-server/v7/internal/pkg/limit"
"github.com/elastic/fleet-server/v7/internal/pkg/logger"

"github.com/pkg/errors"
Expand Down Expand Up @@ -43,7 +42,6 @@ type HTTPErrResp struct {

// NewHTTPErrResp creates an ErrResp from a go error
func NewHTTPErrResp(err error) HTTPErrResp {

errTable := []struct {
target error
meta HTTPErrResp
Expand All @@ -57,24 +55,6 @@ func NewHTTPErrResp(err error) HTTPErrResp {
zerolog.WarnLevel,
},
},
{
limit.ErrRateLimit,
HTTPErrResp{
http.StatusTooManyRequests,
"RateLimit",
"exceeded the rate limit",
zerolog.DebugLevel,
},
},
{
limit.ErrMaxLimit,
HTTPErrResp{
http.StatusTooManyRequests,
"MaxLimit",
"exceeded the max limit",
zerolog.DebugLevel,
},
},
Comment on lines -60 to -77
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

rate limiting responses are done by the wrapping limiter

{
ErrAPIKeyNotEnabled,
HTTPErrResp{
Expand Down
21 changes: 2 additions & 19 deletions internal/pkg/api/handleAck.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,6 @@ import (
"github.com/elastic/fleet-server/v7/internal/pkg/config"
"github.com/elastic/fleet-server/v7/internal/pkg/dl"
"github.com/elastic/fleet-server/v7/internal/pkg/es"
"github.com/elastic/fleet-server/v7/internal/pkg/limit"
"github.com/elastic/fleet-server/v7/internal/pkg/logger"
"github.com/elastic/fleet-server/v7/internal/pkg/model"
"github.com/elastic/fleet-server/v7/internal/pkg/policy"
Expand All @@ -42,27 +41,21 @@ func (e *HTTPError) Error() string {

type AckT struct {
cfg *config.Server
limit *limit.Limiter
bulk bulk.Bulk
cache cache.Cache
}

func NewAckT(cfg *config.Server, bulker bulk.Bulk, cache cache.Cache) *AckT {
log.Info().
Interface("limits", cfg.Limits.AckLimit).
Msg("Setting config ack_limits")

return &AckT{
cfg: cfg,
bulk: bulker,
cache: cache,
limit: limit.NewLimiter(&cfg.Limits.AckLimit),
}
}

func (rt Router) handleAcks(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
//nolint:dupl // function body calls different internal hander then handleCheckin
michel-laterman marked this conversation as resolved.
Show resolved Hide resolved
func (rt *Router) handleAcks(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
start := time.Now()

id := ps.ByName("id")

reqID := r.Header.Get(logger.HeaderRequestID)
Expand Down Expand Up @@ -91,12 +84,6 @@ func (rt Router) handleAcks(w http.ResponseWriter, r *http.Request, ps httproute
}

func (ack *AckT) handleAcks(zlog *zerolog.Logger, w http.ResponseWriter, r *http.Request, id string) error {
limitF, err := ack.limit.Acquire()
if err != nil {
return err
}
defer limitF()

agent, err := authAgent(r, &id, ack.bulk, ack.cache)
if err != nil {
return err
Expand All @@ -107,10 +94,6 @@ func (ack *AckT) handleAcks(zlog *zerolog.Logger, w http.ResponseWriter, r *http
return ctx.Str(LogAccessAPIKeyID, agent.AccessAPIKeyID)
})

// Metrics; serenity now.
dfunc := cntAcks.IncStart()
defer dfunc()
Comment on lines -110 to -112
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In order to not change the structure of the stats we collect, and still track rate-limit errors per route. I've moved the incrementer/decrement to the rate limiter wrapper, this means that the total calls (and active calls) will always be tracked, even in the case where authentication fails.
Currently if auth fails on an endpoint like this the total count will not increase (but the error count might)


return ack.processRequest(*zlog, w, r, agent)
}

Expand Down
20 changes: 1 addition & 19 deletions internal/pkg/api/handleArtifacts.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,6 @@ import (
"github.com/elastic/fleet-server/v7/internal/pkg/cache"
"github.com/elastic/fleet-server/v7/internal/pkg/config"
"github.com/elastic/fleet-server/v7/internal/pkg/dl"
"github.com/elastic/fleet-server/v7/internal/pkg/limit"
"github.com/elastic/fleet-server/v7/internal/pkg/logger"
"github.com/elastic/fleet-server/v7/internal/pkg/model"
"github.com/elastic/fleet-server/v7/internal/pkg/throttle"
Expand All @@ -46,24 +45,17 @@ type ArtifactT struct {
bulker bulk.Bulk
cache cache.Cache
esThrottle *throttle.Throttle
limit *limit.Limiter
}

func NewArtifactT(cfg *config.Server, bulker bulk.Bulk, cache cache.Cache) *ArtifactT {
log.Info().
Interface("limits", cfg.Limits.ArtifactLimit).
Int("maxParallel", defaultMaxParallel).
Msg("Artifact install limits")

return &ArtifactT{
bulker: bulker,
cache: cache,
limit: limit.NewLimiter(&cfg.Limits.ArtifactLimit),
esThrottle: throttle.NewThrottle(defaultMaxParallel),
}
}

func (rt Router) handleArtifacts(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
func (rt *Router) handleArtifacts(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
start := time.Now()

var (
Expand Down Expand Up @@ -112,12 +104,6 @@ func (rt Router) handleArtifacts(w http.ResponseWriter, r *http.Request, ps http
}

func (at ArtifactT) handleArtifacts(zlog *zerolog.Logger, r *http.Request, id, sha2 string) (io.Reader, error) {
limitF, err := at.limit.Acquire()
if err != nil {
return nil, err
}
defer limitF()

// Authenticate the APIKey; retrieve agent record.
// Note: This is going to be a bit slow even if we hit the cache on the api key.
// In order to validate that the agent still has that api key, we fetch the agent record from elastic.
Expand All @@ -131,10 +117,6 @@ func (at ArtifactT) handleArtifacts(zlog *zerolog.Logger, r *http.Request, id, s
return ctx.Str(LogAccessAPIKeyID, agent.AccessAPIKeyID)
})

// Metrics; serenity now.
dfunc := cntArtifacts.IncStart()
defer dfunc()

return at.processRequest(r.Context(), *zlog, agent, id, sha2)
}

Expand Down
32 changes: 2 additions & 30 deletions internal/pkg/api/handleCheckin.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,6 @@ import (
"github.com/elastic/fleet-server/v7/internal/pkg/checkin"
"github.com/elastic/fleet-server/v7/internal/pkg/config"
"github.com/elastic/fleet-server/v7/internal/pkg/dl"
"github.com/elastic/fleet-server/v7/internal/pkg/limit"
"github.com/elastic/fleet-server/v7/internal/pkg/logger"
"github.com/elastic/fleet-server/v7/internal/pkg/model"
"github.com/elastic/fleet-server/v7/internal/pkg/monitor"
Expand All @@ -48,7 +47,8 @@ const (
kEncodingGzip = "gzip"
)

func (rt Router) handleCheckin(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
//nolint:dupl // function body calls different internal hander then handleAck
func (rt *Router) handleCheckin(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
start := time.Now()

id := ps.ByName("id")
Expand All @@ -65,12 +65,6 @@ func (rt Router) handleCheckin(w http.ResponseWriter, r *http.Request, ps httpro
cntCheckin.IncError(err)
resp := NewHTTPErrResp(err)

// Log this as warn for visibility that limit has been reached.
// This allows customers to tune the configuration on detection of threshold.
if errors.Is(err, limit.ErrMaxLimit) {
resp.Level = zerolog.WarnLevel
}

zlog.WithLevel(resp.Level).
Err(err).
Int(ECSHTTPResponseCode, resp.StatusCode).
Expand All @@ -93,7 +87,6 @@ type CheckinT struct {
ad *action.Dispatcher
tr *action.TokenResolver
bulker bulk.Bulk
limit *limit.Limiter
}

func NewCheckinT(
Expand All @@ -107,14 +100,6 @@ func NewCheckinT(
tr *action.TokenResolver,
bulker bulk.Bulk,
) *CheckinT {

log.Info().
Interface("limits", cfg.Limits.CheckinLimit).
Dur("long_poll_timeout", cfg.Timeouts.CheckinLongPoll).
Dur("long_poll_timestamp", cfg.Timeouts.CheckinTimestamp).
Dur("long_poll_jitter", cfg.Timeouts.CheckinJitter).
Msg("Checkin install limits")

ct := &CheckinT{
verCon: verCon,
cfg: cfg,
Expand All @@ -124,23 +109,15 @@ func NewCheckinT(
gcp: gcp,
ad: ad,
tr: tr,
limit: limit.NewLimiter(&cfg.Limits.CheckinLimit),
bulker: bulker,
}

return ct
}

func (ct *CheckinT) handleCheckin(zlog *zerolog.Logger, w http.ResponseWriter, r *http.Request, id string) error {

start := time.Now()

limitF, err := ct.limit.Acquire()
if err != nil {
return err
}
defer limitF()

agent, err := authAgent(r, &id, ct.bulker, ct.cache)
if err != nil {
return err
Expand All @@ -158,11 +135,6 @@ func (ct *CheckinT) handleCheckin(zlog *zerolog.Logger, w http.ResponseWriter, r

// Safely check if the agent version is different, return empty string otherwise
newVer := agent.CheckDifferentVersion(ver)

// Metrics; serenity now.
dfunc := cntCheckin.IncStart()
defer dfunc()

return ct.processRequest(*zlog, w, r, start, agent, newVer)
}

Expand Down
20 changes: 1 addition & 19 deletions internal/pkg/api/handleEnroll.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,6 @@ import (
"github.com/elastic/fleet-server/v7/internal/pkg/cache"
"github.com/elastic/fleet-server/v7/internal/pkg/config"
"github.com/elastic/fleet-server/v7/internal/pkg/dl"
"github.com/elastic/fleet-server/v7/internal/pkg/limit"
"github.com/elastic/fleet-server/v7/internal/pkg/logger"
"github.com/elastic/fleet-server/v7/internal/pkg/model"
"github.com/elastic/fleet-server/v7/internal/pkg/rollback"
Expand Down Expand Up @@ -49,25 +48,19 @@ type EnrollerT struct {
cfg *config.Server
bulker bulk.Bulk
cache cache.Cache
limit *limit.Limiter
}

func NewEnrollerT(verCon version.Constraints, cfg *config.Server, bulker bulk.Bulk, c cache.Cache) (*EnrollerT, error) {
log.Info().
Interface("limits", cfg.Limits.EnrollLimit).
Msg("Setting config enroll_limit")

return &EnrollerT{
verCon: verCon,
cfg: cfg,
limit: limit.NewLimiter(&cfg.Limits.EnrollLimit),
bulker: bulker,
cache: c,
}, nil

}

func (rt Router) handleEnroll(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
func (rt *Router) handleEnroll(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
start := time.Now()

// Work around wonky router rule
Expand Down Expand Up @@ -129,13 +122,6 @@ func (rt Router) handleEnroll(w http.ResponseWriter, r *http.Request, ps httprou
}

func (et *EnrollerT) handleEnroll(rb *rollback.Rollback, zlog *zerolog.Logger, w http.ResponseWriter, r *http.Request) (*EnrollResponse, error) {

limitF, err := et.limit.Acquire()
if err != nil {
return nil, err
}
defer limitF()

key, err := authAPIKey(r, et.bulker, et.cache)
if err != nil {
return nil, err
Expand All @@ -151,10 +137,6 @@ func (et *EnrollerT) handleEnroll(rb *rollback.Rollback, zlog *zerolog.Logger, w
return nil, err
}

// Metrics; serenity now.
dfunc := cntEnroll.IncStart()
defer dfunc()

return et.processRequest(rb, *zlog, w, r, key.ID, ver)
}

Expand Down
Loading