Skip to content

Commit

Permalink
Improve caching logic for http.send
Browse files Browse the repository at this point in the history 
In order to calculate if a cached entry expired or not, we would previously
read the headers stored with the request, and parse each of the relevant date-related
ones and use those to compare with the current time. #4960 improved this by allowing
us to ignore the Date header completely when force_cache is enabled.

We now go one step further, and calculate and store an expiresAt value at the time the
entry is created. This allows for a much simpler check of expiry, and avoids re-parsing
the header values with each lookup.

Fixes #5300

Signed-off-by: Anders Eknert <anders@eknert.com>
  • Loading branch information
@anderseknert
anderseknert committed Oct 27, 2022
1 parent 51f9acb commit 41f6f73
Show file tree
Hide file tree
Showing 2 changed files with 93 additions and 119 deletions.
132 changes: 57 additions & 75 deletions topdown/http.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -505,7 +505,7 @@ func createHTTPRequest(bctx BuiltinContext, obj ast.Object) (*http.Request, *htt

if len(tlsCaCert) != 0 {
tlsCaCert = bytes.Replace(tlsCaCert, []byte("\\n"), []byte("\n"), -1)
pool, err := addCACertsFromBytes(tlsConfig.RootCAs, []byte(tlsCaCert))
pool, err := addCACertsFromBytes(tlsConfig.RootCAs, tlsCaCert)
if err != nil {
return nil, nil, err
}
Expand Down Expand Up @@ -692,21 +692,21 @@ func (c *interQueryCache) checkHTTPSendInterQueryCache() (ast.Value, error) {
return nil, nil
}

headers, err := parseResponseHeaders(cachedRespData.Headers)
if err != nil {
return nil, err
}

// check the freshness of the cached response
if isCachedResponseFresh(c.bctx, headers, c.forceCacheParams) {
if getCurrentTime(c.bctx).Before(cachedRespData.ExpiresAt) {
return cachedRespData.formatToAST(c.forceJSONDecode, c.forceYAMLDecode)
}

var err error
c.httpReq, c.httpClient, err = createHTTPRequest(c.bctx, c.key)
if err != nil {
return nil, handleHTTPSendErr(c.bctx, err)
}

headers, err := parseResponseHeaders(cachedRespData.Headers)
if err != nil {
return nil, err
}

// check with the server if the stale response is still up-to-date.
// If server returns a new response (ie. status_code=200), update the cache with the new response
// If server returns an unmodified response (ie. status_code=304), update the headers for the existing response
Expand All @@ -727,6 +727,12 @@ func (c *interQueryCache) checkHTTPSendInterQueryCache() (ast.Value, error) {
}
}

expiresAt, err := expiryFromHeaders(result.Header)
if err != nil {
return nil, err
}
cachedRespData.ExpiresAt = expiresAt

cachingMode, err := getCachingMode(c.key)
if err != nil {
return nil, err
Expand All @@ -753,16 +759,16 @@ func (c *interQueryCache) checkHTTPSendInterQueryCache() (ast.Value, error) {
return nil, err
}

if err := insertIntoHTTPSendInterQueryCache(c.bctx, c.key, result, respBody, c.forceCacheParams != nil); err != nil {
if err := insertIntoHTTPSendInterQueryCache(c.bctx, c.key, result, respBody, c.forceCacheParams); err != nil {
return nil, err
}

return newValue, nil
}

// insertIntoHTTPSendInterQueryCache inserts given key and value in the inter-query cache
func insertIntoHTTPSendInterQueryCache(bctx BuiltinContext, key ast.Value, resp *http.Response, respBody []byte, force bool) error {
if resp == nil || (!force && !canStore(resp.Header)) {
func insertIntoHTTPSendInterQueryCache(bctx BuiltinContext, key ast.Value, resp *http.Response, respBody []byte, cacheParams *forceCacheParams) error {
if resp == nil || (!forceCaching(cacheParams) && !canStore(resp.Header)) {
return nil
}

Expand All @@ -781,9 +787,9 @@ func insertIntoHTTPSendInterQueryCache(bctx BuiltinContext, key ast.Value, resp
var pcv cache.InterQueryCacheValue

if cachingMode == defaultCachingMode {
pcv, err = newInterQueryCacheValue(resp, respBody, force)
pcv, err = newInterQueryCacheValue(bctx, resp, respBody, cacheParams)
} else {
pcv, err = newInterQueryCacheData(resp, respBody, force)
pcv, err = newInterQueryCacheData(bctx, resp, respBody, cacheParams)
}

if err != nil {
Expand Down Expand Up @@ -855,15 +861,15 @@ func getCachingMode(req ast.Object) (cachingMode, error) {
return "", fmt.Errorf("invalid value specified for %v field: %v", key.String(), string(s))
}
}
return cachingMode(defaultCachingMode), nil
return defaultCachingMode, nil
}

type interQueryCacheValue struct {
Data []byte
}

func newInterQueryCacheValue(resp *http.Response, respBody []byte, force bool) (*interQueryCacheValue, error) {
data, err := newInterQueryCacheData(resp, respBody, force)
func newInterQueryCacheValue(bctx BuiltinContext, resp *http.Response, respBody []byte, cacheParams *forceCacheParams) (*interQueryCacheValue, error) {
data, err := newInterQueryCacheData(bctx, resp, respBody, cacheParams)
if err != nil {
return nil, err
}
Expand Down Expand Up @@ -893,20 +899,48 @@ type interQueryCacheData struct {
Status string
StatusCode int
Headers http.Header
ExpiresAt time.Time
}

func forceCaching(cacheParams *forceCacheParams) bool {
return cacheParams != nil && cacheParams.forceCacheDurationSeconds > 0
}

func expiryFromHeaders(headers http.Header) (time.Time, error) {
var expiresAt time.Time
maxAge, err := parseMaxAgeCacheDirective(parseCacheControlHeader(headers))
if err != nil {
return time.Time{}, err
}
if maxAge != -1 {
createdAt, err := getResponseHeaderDate(headers)
if err != nil {
return time.Time{}, err
}
expiresAt = createdAt.Add(time.Second * time.Duration(maxAge))
} else {
expiresAt = getResponseHeaderExpires(headers)
}
return expiresAt, nil
}

func newInterQueryCacheData(resp *http.Response, respBody []byte, force bool) (*interQueryCacheData, error) {
if force {
now := time.Now().UTC()
resp.Header["Date"] = []string{now.Format(http.TimeFormat)}
func newInterQueryCacheData(bctx BuiltinContext, resp *http.Response, respBody []byte, cacheParams *forceCacheParams) (*interQueryCacheData, error) {
var expiresAt time.Time

if forceCaching(cacheParams) {
createdAt := getCurrentTime(bctx)
expiresAt = createdAt.Add(time.Second * time.Duration(cacheParams.forceCacheDurationSeconds))
} else {
_, err := parseResponseHeaders(resp.Header)
var err error
expiresAt, err = expiryFromHeaders(resp.Header)
if err != nil {
return nil, err
}
}

cv := interQueryCacheData{RespBody: respBody,
cv := interQueryCacheData{
ExpiresAt: expiresAt,
RespBody: respBody,
Status: resp.Status,
StatusCode: resp.StatusCode,
Headers: resp.Header}
Expand Down Expand Up @@ -1013,58 +1047,6 @@ func canStore(headers http.Header) bool {
return true
}

func isCachedResponseFresh(bctx BuiltinContext, headers *responseHeaders, cacheParams *forceCacheParams) bool {
if headers.date.IsZero() {
return false
}

currentTime := getCurrentTime(bctx)
if currentTime.IsZero() {
return false
}

currentAge := currentTime.Sub(headers.date)

// The time.Sub operation uses wall clock readings and
// not monotonic clock readings as the parsed version of the response time
// does not contain monotonic clock readings. This can result in negative durations.
// Another scenario where a negative duration can occur, is when a server sets the Date
// response header. As per https://tools.ietf.org/html/rfc7231#section-7.1.1.2,
// an origin server MUST NOT send a Date header field if it does not
// have a clock capable of providing a reasonable approximation of the
// current instance in Coordinated Universal Time.
// Hence, consider the cached response as stale if a negative duration is encountered.
if currentAge < 0 {
return false
}

if cacheParams != nil {
// override the cache directives set by the server
maxAgeDur := time.Second * time.Duration(cacheParams.forceCacheDurationSeconds)
if maxAgeDur > currentAge {
return true
}
} else {
// Check "max-age" cache directive.
// The "max-age" response directive indicates that the response is to be
// considered stale after its age is greater than the specified number
// of seconds.
if headers.maxAge != -1 {
maxAgeDur := time.Second * time.Duration(headers.maxAge)
if maxAgeDur > currentAge {
return true
}
} else {
// Check "Expires" header.
// Note: "max-age" if set, takes precedence over "Expires"
if headers.expires.Sub(headers.date) > currentAge {
return true
}
}
}
return false
}

func getCurrentTime(bctx BuiltinContext) time.Time {
var current time.Time

Expand Down Expand Up @@ -1280,7 +1262,7 @@ func (c *interQueryCache) InsertIntoCache(value *http.Response) (ast.Value, erro
}

// fallback to the http send cache if error encountered while inserting response in inter-query cache
err = insertIntoHTTPSendInterQueryCache(c.bctx, c.key, value, respBody, c.forceCacheParams != nil)
err = insertIntoHTTPSendInterQueryCache(c.bctx, c.key, value, respBody, c.forceCacheParams)
if err != nil {
insertIntoHTTPSendCache(c.bctx, c.key, result)
}
Expand Down
Loading

0 comments on commit 41f6f73

Please sign in to comment.