grove/cache/handler.go - trafficcontrol - Git at Google

 package cache

 /*
    Licensed under the Apache License, Version 2.0 (the "License");
    you may not use this file except in compliance with the License.
    You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

    Unless required by applicable law or agreed to in writing, software
    distributed under the License is distributed on an "AS IS" BASIS,
    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    See the License for the specific language governing permissions and
    limitations under the License.
 */

 import (
 	"net/http"
 	"os"
 	"strconv"
 	"sync/atomic"
 	"time"
 	"unsafe"

 	"github.com/apache/trafficcontrol/grove/cachedata"
 	"github.com/apache/trafficcontrol/grove/plugin"

 	"github.com/apache/trafficcontrol/grove/remap"
 	"github.com/apache/trafficcontrol/grove/stat"
 	"github.com/apache/trafficcontrol/grove/thread"
 	"github.com/apache/trafficcontrol/grove/web"

 	"github.com/apache/trafficcontrol/lib/go-log"
 	"github.com/apache/trafficcontrol/lib/go-rfc"
 )

 type HandlerPointer struct {
 	realHandler *unsafe.Pointer
 }

 func NewHandlerPointer(realHandler *Handler) *HandlerPointer {
 	p := (unsafe.Pointer)(realHandler)
 	return &HandlerPointer{realHandler: &p}
 }

 func (h *HandlerPointer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	realHandler := (*Handler)(atomic.LoadPointer(h.realHandler))
 	realHandler.ServeHTTP(w, r)
 }

 func (h *HandlerPointer) Set(newHandler *Handler) {
 	p := (unsafe.Pointer)(newHandler)
 	atomic.StorePointer(h.realHandler, p)
 }

 type Handler struct {
 	remapper        remap.HTTPRequestRemapper
 	getter          thread.Getter
 	ruleThrottlers  map[string]thread.Throttler // doesn't need threadsafe keys, because it's never added to or deleted after creation. TODO fix for hot rule reloading
 	scheme          string
 	port            string
 	hostname        string
 	strictRFC       bool
 	stats           stat.Stats
 	conns           *web.ConnMap
 	connectionClose bool
 	plugins         plugin.Plugins
 	pluginContext   map[string]*interface{}
 	httpConns       *web.ConnMap
 	httpsConns      *web.ConnMap
 	interfaceName   string
 	requestID       uint64 // Atomic - DO NOT access or modify without atomic operations
 	// keyThrottlers     Throttlers
 	// nocacheThrottlers Throttlers
 }

 // func (h *cacheHandler) checkoutKeyThrottler(k string) Throttler {
 // 	keyThrottlersM.Lock()
 // 	defer keyThrottlersM.Unlock()
 // 	if t, ok := keyThrottlers[k]; !ok {
 // 		keyThrottlers[k] = NewThrottler
 // 	}
 // 	return keyThrottlers[k]
 // }

 // NewHandler returns an http.Handler object, which may be pipelined with other http.Handlers via `http.ListenAndServe`. If you prefer pipelining functions, use `GetHandlerFunc`.
 //
 // This needs rate-limited in 3 ways.
 // 1. ruleLimit - Simultaneous requests to the origin (remap rule) should be configurably limited. For example, "only allow 1000 simultaneous requests to the origin
 // 2. keyLimit - Simultaneous requests, on cache miss, for the same key (Method+Path+Qstring), should be configurably limited. For example, "Only allow 10 simultaneous requests per unique URL on cache miss. Additional requestors must wait until others complete. Once another requestor completes, all waitors for the same URL are signalled to use the cache, or proceed to the third uncacheable limiter"
 // 3. nocacheLimit - If simultaneous requestors exceed the URL limiter, and some request for the same key gets a result which is uncacheable, waitors for the same URL may then proceed at a third configurable limit for uncacheable requests.
 //
 // Note these only apply to cache misses. Cache hits are not limited in any way, the origin is not hit and the cache value is immediately returned to the client.
 //
 // This prevents a large number of uncacheable requests for the same URL from timing out because they're required to proceed serially from the low simultaneous-requests-per-URL limit, while at the same time only hitting the origin with a very low limit for many simultaneous cacheable requests.
 //
 // Example: Origin limit is 10,000, key limit is 1, the uncacheable limit is 1,000.
 // Then, 2,000 requests come in for the same URL, simultaneously. They are all within the Origin limit, so they are all allowed to proceed to the key limiter. Then, the first request is allowed to make an actual request to the origin, while the other 1,999 wait at the key limiter.
 //
 // The connectionClose parameter determines whether to send a `Connection: close` header. This is primarily designed for maintenance, to drain the cache of incoming requestors. This overrides rule-specific `connection-close: false` configuration, under the assumption that draining a cache is a temporary maintenance operation, and if connectionClose is true on the service and false on some rules, those rules' configuration is probably a permament setting whereas the operator probably wants to drain all connections if the global setting is true. If it's necessary to leave connection close false on some rules, set all other rules' connectionClose to true and leave the global connectionClose unset.
 func NewHandler(
 	remapper remap.HTTPRequestRemapper,
 	ruleLimit uint64,
 	stats stat.Stats,
 	scheme string,
 	port string,
 	conns *web.ConnMap,
 	strictRFC bool,
 	connectionClose bool,
 	plugins plugin.Plugins,
 	pluginContext map[string]*interface{},
 	httpConns *web.ConnMap,
 	httpsConns *web.ConnMap,
 	interfaceName string,
 ) *Handler {
 	hostname, err := os.Hostname()
 	if err != nil {
 		log.Errorf("getting  hostname: %v\n", err)
 	}

 	return &Handler{
 		remapper:        remapper,
 		getter:          thread.NewGetter(),
 		ruleThrottlers:  makeRuleThrottlers(remapper, ruleLimit),
 		strictRFC:       strictRFC,
 		scheme:          scheme,
 		port:            port,
 		hostname:        hostname,
 		stats:           stats,
 		conns:           conns,
 		connectionClose: connectionClose,
 		plugins:         plugins,
 		pluginContext:   pluginContext,
 		httpConns:       httpConns,
 		httpsConns:      httpsConns,
 		interfaceName:   interfaceName,
 		// keyThrottlers:     NewThrottlers(keyLimit),
 		// nocacheThrottlers: NewThrottlers(nocacheLimit),
 	}
 }

 func makeRuleThrottlers(remapper remap.HTTPRequestRemapper, limit uint64) map[string]thread.Throttler {
 	remapRules := remapper.Rules()
 	ruleThrottlers := make(map[string]thread.Throttler, len(remapRules))
 	for _, rule := range remapRules {
 		ruleLimit := uint64(rule.ConcurrentRuleRequests)
 		if rule.ConcurrentRuleRequests == 0 {
 			ruleLimit = limit
 		}
 		ruleThrottlers[rule.Name] = thread.NewThrottler(ruleLimit)
 	}
 	return ruleThrottlers
 }

 func copyPluginContext(context map[string]*interface{}) map[string]*interface{} {
 	new := make(map[string]*interface{}, len(context))
 	for k, v := range context {
 		newV := *v
 		new[k] = &newV
 	}
 	return new
 }

 func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	reqTime := time.Now()
 	reqID := atomic.AddUint64(&h.requestID, 1)
 	pluginContext := copyPluginContext(h.pluginContext) // must give each request a copy, because they can modify in parallel
 	srvrData := cachedata.SrvrData{Hostname: h.hostname, Port: h.port, Scheme: h.scheme}
 	onReqData := plugin.OnRequestData{W: w, R: r, Stats: h.stats, StatRules: h.remapper.StatRules(), HTTPConns: h.httpConns, HTTPSConns: h.httpsConns, InterfaceName: h.interfaceName, SrvrData: srvrData, RequestID: reqID}
 	stop := h.plugins.OnRequest(h.remapper.PluginCfg(), pluginContext, onReqData)
 	if stop {
 		return
 	}

 	conn := (*web.InterceptConn)(nil)
 	if realConn, ok := h.conns.Get(r.RemoteAddr); !ok {
 		log.Infof("RemoteAddr '%v' not in Conns (reqid %v)\n", r.RemoteAddr, reqID)
 	} else {
 		if conn, ok = realConn.(*web.InterceptConn); !ok {
 			log.Infof("Could not get Conn info: Conn is not an InterceptConn: %T (reqid %v)\n", realConn, reqID)
 		}
 	}

 	remappingProducer, err := h.remapper.RemappingProducer(r, h.scheme)

 	if err == nil { // if we failed to get a remapping, there's no DSCP to set.
 		if err := conn.SetDSCP(remappingProducer.DSCP()); err != nil {
 			log.Infoln(time.Now().Format(time.RFC3339Nano) + " " + r.RemoteAddr + " " + r.Method + " " + r.RequestURI + ": could not set DSCP: " + err.Error() + " (reqid " + strconv.FormatUint(reqID, 10) + ")")
 		}
 	}

 	reqHeader := web.CopyHeader(r.Header) // copy request header, because it's not guaranteed valid after actually issuing the request
 	clientIP, _ := web.GetClientIPPort(r)

 	toFQDN := ""
 	pluginCfg := map[string]interface{}{}
 	if remappingProducer != nil {
 		toFQDN = remappingProducer.FirstFQDN()
 		pluginCfg = remappingProducer.PluginCfg()
 	}

 	reqData := cachedata.ReqData{Req: r, Conn: conn, ClientIP: clientIP, ReqTime: reqTime, ToFQDN: toFQDN}
 	responder := NewResponder(w, pluginCfg, pluginContext, srvrData, reqData, h.plugins, h.stats, reqID)

 	if err != nil {
 		switch err {
 		case remap.ErrRuleNotFound:
 			log.Debugf("rule not found for %v (reqid %v)\n", r.RequestURI, reqID)
 			*responder.ResponseCode = http.StatusNotFound
 		case remap.ErrIPNotAllowed:
 			log.Debugf("IP %v not allowed (reqid %v)\n", r.RemoteAddr, reqID)
 			*responder.ResponseCode = http.StatusForbidden
 		default:
 			log.Debugf("request error: %v (reqid %v)\n", err, reqID)
 		}
 		responder.OriginConnectFailed = true
 		responder.Do()
 		return
 	}

 	reqCacheControl := rfc.ParseCacheControl(reqHeader)
 	log.Debugf("Serve got Cache-Control %+v (reqid %v)\n", reqCacheControl, reqID)

 	connectionClose := h.connectionClose || remappingProducer.ConnectionClose()

 	beforeCacheLookUpData := plugin.BeforeCacheLookUpData{Req: r, DefaultCacheKey: remappingProducer.CacheKey(), CacheKeyOverrideFunc: remappingProducer.OverrideCacheKey}
 	h.plugins.OnBeforeCacheLookup(remappingProducer.PluginCfg(), pluginContext, beforeCacheLookUpData)

 	cacheKey := remappingProducer.CacheKey()
 	retrier := NewRetrier(h, reqHeader, reqTime, reqCacheControl, remappingProducer, reqID)

 	cache := remappingProducer.Cache()

 	var reqHost *string
 	cacheObj, ok := cache.Get(cacheKey)
 	if !ok {
 		log.Debugf("cache.Handler.ServeHTTP: '%v' not in cache (reqid %v)\n", cacheKey, reqID)
 		beforeParentRequestData := plugin.BeforeParentRequestData{Req: r, RemapRule: remappingProducer.Name()}
 		h.plugins.OnBeforeParentRequest(remappingProducer.PluginCfg(), pluginContext, beforeParentRequestData)
 		cacheObj, reqHost, err = retrier.Get(r, nil)
 		if err != nil {
 			log.Errorf("retrying get error (in uncached): %v (reqid %v)\n", err, reqID)
 			responder.OriginConnectFailed = true
 			responder.ProxyStr = cacheObj.ProxyURL
 			if reqHost != nil {
 				responder.ToFQDN = *reqHost
 			}
 			responder.Do()
 			return
 		}

 		responder.OriginCode = cacheObj.OriginCode
 		// create new pointers, so plugins don't modify the cacheObj
 		codePtr, hdrsPtr, bodyPtr := cacheObj.Code, cacheObj.RespHeaders, cacheObj.Body
 		responder.SetResponse(&codePtr, &hdrsPtr, &bodyPtr, connectionClose)
 		responder.OriginReqSuccess = true
 		responder.ProxyStr = cacheObj.ProxyURL
 		if reqHost != nil {
 			responder.ToFQDN = *reqHost
 		}
 		beforeRespData := plugin.BeforeRespondData{Req: r, CacheObj: cacheObj, Code: &codePtr, Hdr: &hdrsPtr, Body: &bodyPtr, RemapRule: remappingProducer.Name()}
 		h.plugins.OnBeforeRespond(remappingProducer.PluginCfg(), pluginContext, beforeRespData)
 		responder.Do()
 		return
 	}

 	reqHeaders := r.Header
 	canReuseStored := rfc.CanReuseStored(reqHeaders, cacheObj.RespHeaders, reqCacheControl, cacheObj.RespCacheControl, cacheObj.ReqHeaders, cacheObj.ReqRespTime, cacheObj.RespRespTime, h.strictRFC)

 	if canReuseStored != rfc.ReuseCan { // run the BeforeParentRequest hook for revalidations / ReuseCannot
 		beforeParentRequestData := plugin.BeforeParentRequestData{Req: r, RemapRule: remappingProducer.Name()}
 		h.plugins.OnBeforeParentRequest(remappingProducer.PluginCfg(), pluginContext, beforeParentRequestData)
 	}

 	switch canReuseStored {
 	case rfc.ReuseCan:
 		log.Debugf("cache.Handler.ServeHTTP: '%v' cache hit! (reqid %v)\n", cacheKey, reqID)
 	case rfc.ReuseCannot:
 		log.Debugf("cache.Handler.ServeHTTP: '%v' can't reuse (reqid %v)\n", cacheKey, reqID)
 		cacheObj, reqHost, err = retrier.Get(r, nil)
 		if err != nil {
 			log.Errorf("retrying get error (in reuse-cannot): %v (reqid %v)\n", err, reqID)
 			responder.Do()
 			return
 		}
 	case rfc.ReuseMustRevalidate:
 		log.Debugf("cache.Handler.ServeHTTP: '%v' must revalidate (reqid %v)\n", cacheKey, reqID)
 		cacheObj, reqHost, err = retrier.Get(r, cacheObj)
 		if err != nil {
 			log.Errorf("retrying get error: %v (reqid %v)\n", err, reqID)
 			responder.Do()
 			return
 		}
 	case rfc.ReuseMustRevalidateCanStale:
 		log.Debugf("cache.Handler.ServeHTTP: '%v' must revalidate (but allowed stale) (reqid %v)\n", cacheKey, reqID)
 		oldCacheObj := cacheObj
 		cacheObj, reqHost, err = retrier.Get(r, cacheObj)
 		if err != nil {
 			log.Errorf("retrying get error - serving stale as allowed: %v (reqid %v)\n", err, reqID)
 			cacheObj = oldCacheObj
 		}
 	}
 	log.Debugf("cache.Handler.ServeHTTP: '%v' responding with %v (reqid %v)\n", cacheKey, cacheObj.Code, reqID)

 	// create new pointers, so plugins don't modify the cacheObj
 	codePtr, hdrsPtr, bodyPtr := cacheObj.Code, cacheObj.RespHeaders, cacheObj.Body
 	responder.SetResponse(&codePtr, &hdrsPtr, &bodyPtr, connectionClose)
 	responder.OriginReqSuccess = true
 	responder.Reuse = canReuseStored
 	responder.OriginCode = cacheObj.OriginCode
 	responder.OriginBytes = cacheObj.Size
 	responder.ProxyStr = cacheObj.ProxyURL
 	if reqHost != nil {
 		responder.ToFQDN = *reqHost
 	}
 	beforeRespData := plugin.BeforeRespondData{Req: r, CacheObj: cacheObj, Code: &codePtr, Hdr: &hdrsPtr, Body: &bodyPtr, RemapRule: remappingProducer.Name()}
 	h.plugins.OnBeforeRespond(remappingProducer.PluginCfg(), pluginContext, beforeRespData)
 	responder.Do()
 }
	package cache

	/*
	Licensed under the Apache License, Version 2.0 (the "License");
	you may not use this file except in compliance with the License.
	You may obtain a copy of the License at

	http://www.apache.org/licenses/LICENSE-2.0

	Unless required by applicable law or agreed to in writing, software
	distributed under the License is distributed on an "AS IS" BASIS,
	WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	See the License for the specific language governing permissions and
	limitations under the License.
	*/

	import (
	"net/http"
	"os"
	"strconv"
	"sync/atomic"
	"time"
	"unsafe"

	"github.com/apache/trafficcontrol/grove/cachedata"
	"github.com/apache/trafficcontrol/grove/plugin"

	"github.com/apache/trafficcontrol/grove/remap"
	"github.com/apache/trafficcontrol/grove/stat"
	"github.com/apache/trafficcontrol/grove/thread"
	"github.com/apache/trafficcontrol/grove/web"

	"github.com/apache/trafficcontrol/lib/go-log"
	"github.com/apache/trafficcontrol/lib/go-rfc"
	)

	type HandlerPointer struct {
	realHandler *unsafe.Pointer
	}

	func NewHandlerPointer(realHandler Handler) HandlerPointer {
	p := (unsafe.Pointer)(realHandler)
	return &HandlerPointer{realHandler: &p}
	}

	func (h HandlerPointer) ServeHTTP(w http.ResponseWriter, r http.Request) {
	realHandler := (*Handler)(atomic.LoadPointer(h.realHandler))
	realHandler.ServeHTTP(w, r)
	}

	func (h HandlerPointer) Set(newHandler Handler) {
	p := (unsafe.Pointer)(newHandler)
	atomic.StorePointer(h.realHandler, p)
	}

	type Handler struct {
	remapper remap.HTTPRequestRemapper
	getter thread.Getter
	ruleThrottlers map[string]thread.Throttler // doesn't need threadsafe keys, because it's never added to or deleted after creation. TODO fix for hot rule reloading
	scheme string
	port string
	hostname string
	strictRFC bool
	stats stat.Stats
	conns *web.ConnMap
	connectionClose bool
	plugins plugin.Plugins
	pluginContext map[string]*interface{}
	httpConns *web.ConnMap
	httpsConns *web.ConnMap
	interfaceName string
	requestID uint64 // Atomic - DO NOT access or modify without atomic operations
	// keyThrottlers Throttlers
	// nocacheThrottlers Throttlers
	}

	// func (h *cacheHandler) checkoutKeyThrottler(k string) Throttler {
	// keyThrottlersM.Lock()
	// defer keyThrottlersM.Unlock()
	// if t, ok := keyThrottlers[k]; !ok {
	// keyThrottlers[k] = NewThrottler
	// }
	// return keyThrottlers[k]
	// }

	// NewHandler returns an http.Handler object, which may be pipelined with other http.Handlers via `http.ListenAndServe`. If you prefer pipelining functions, use `GetHandlerFunc`.
	//
	// This needs rate-limited in 3 ways.
	// 1. ruleLimit - Simultaneous requests to the origin (remap rule) should be configurably limited. For example, "only allow 1000 simultaneous requests to the origin
	// 2. keyLimit - Simultaneous requests, on cache miss, for the same key (Method+Path+Qstring), should be configurably limited. For example, "Only allow 10 simultaneous requests per unique URL on cache miss. Additional requestors must wait until others complete. Once another requestor completes, all waitors for the same URL are signalled to use the cache, or proceed to the third uncacheable limiter"
	// 3. nocacheLimit - If simultaneous requestors exceed the URL limiter, and some request for the same key gets a result which is uncacheable, waitors for the same URL may then proceed at a third configurable limit for uncacheable requests.
	//
	// Note these only apply to cache misses. Cache hits are not limited in any way, the origin is not hit and the cache value is immediately returned to the client.
	//
	// This prevents a large number of uncacheable requests for the same URL from timing out because they're required to proceed serially from the low simultaneous-requests-per-URL limit, while at the same time only hitting the origin with a very low limit for many simultaneous cacheable requests.
	//
	// Example: Origin limit is 10,000, key limit is 1, the uncacheable limit is 1,000.
	// Then, 2,000 requests come in for the same URL, simultaneously. They are all within the Origin limit, so they are all allowed to proceed to the key limiter. Then, the first request is allowed to make an actual request to the origin, while the other 1,999 wait at the key limiter.
	//
	// The connectionClose parameter determines whether to send a `Connection: close` header. This is primarily designed for maintenance, to drain the cache of incoming requestors. This overrides rule-specific `connection-close: false` configuration, under the assumption that draining a cache is a temporary maintenance operation, and if connectionClose is true on the service and false on some rules, those rules' configuration is probably a permament setting whereas the operator probably wants to drain all connections if the global setting is true. If it's necessary to leave connection close false on some rules, set all other rules' connectionClose to true and leave the global connectionClose unset.
	func NewHandler(
	remapper remap.HTTPRequestRemapper,
	ruleLimit uint64,
	stats stat.Stats,
	scheme string,
	port string,
	conns *web.ConnMap,
	strictRFC bool,
	connectionClose bool,
	plugins plugin.Plugins,
	pluginContext map[string]*interface{},
	httpConns *web.ConnMap,
	httpsConns *web.ConnMap,
	interfaceName string,
	) *Handler {
	hostname, err := os.Hostname()
	if err != nil {
	log.Errorf("getting hostname: %v\n", err)
	}

	return &Handler{
	remapper: remapper,
	getter: thread.NewGetter(),
	ruleThrottlers: makeRuleThrottlers(remapper, ruleLimit),
	strictRFC: strictRFC,
	scheme: scheme,
	port: port,
	hostname: hostname,
	stats: stats,
	conns: conns,
	connectionClose: connectionClose,
	plugins: plugins,
	pluginContext: pluginContext,
	httpConns: httpConns,
	httpsConns: httpsConns,
	interfaceName: interfaceName,
	// keyThrottlers: NewThrottlers(keyLimit),
	// nocacheThrottlers: NewThrottlers(nocacheLimit),
	}
	}

	func makeRuleThrottlers(remapper remap.HTTPRequestRemapper, limit uint64) map[string]thread.Throttler {
	remapRules := remapper.Rules()
	ruleThrottlers := make(map[string]thread.Throttler, len(remapRules))
	for _, rule := range remapRules {
	ruleLimit := uint64(rule.ConcurrentRuleRequests)
	if rule.ConcurrentRuleRequests == 0 {
	ruleLimit = limit
	}
	ruleThrottlers[rule.Name] = thread.NewThrottler(ruleLimit)
	}
	return ruleThrottlers
	}

	func copyPluginContext(context map[string]interface{}) map[string]interface{} {
	new := make(map[string]*interface{}, len(context))
	for k, v := range context {
	newV := *v
	new[k] = &newV
	}
	return new
	}

	func (h Handler) ServeHTTP(w http.ResponseWriter, r http.Request) {
	reqTime := time.Now()
	reqID := atomic.AddUint64(&h.requestID, 1)
	pluginContext := copyPluginContext(h.pluginContext) // must give each request a copy, because they can modify in parallel
	srvrData := cachedata.SrvrData{Hostname: h.hostname, Port: h.port, Scheme: h.scheme}
	onReqData := plugin.OnRequestData{W: w, R: r, Stats: h.stats, StatRules: h.remapper.StatRules(), HTTPConns: h.httpConns, HTTPSConns: h.httpsConns, InterfaceName: h.interfaceName, SrvrData: srvrData, RequestID: reqID}
	stop := h.plugins.OnRequest(h.remapper.PluginCfg(), pluginContext, onReqData)
	if stop {
	return
	}

	conn := (*web.InterceptConn)(nil)
	if realConn, ok := h.conns.Get(r.RemoteAddr); !ok {
	log.Infof("RemoteAddr '%v' not in Conns (reqid %v)\n", r.RemoteAddr, reqID)
	} else {
	if conn, ok = realConn.(*web.InterceptConn); !ok {
	log.Infof("Could not get Conn info: Conn is not an InterceptConn: %T (reqid %v)\n", realConn, reqID)
	}
	}

	remappingProducer, err := h.remapper.RemappingProducer(r, h.scheme)

	if err == nil { // if we failed to get a remapping, there's no DSCP to set.
	if err := conn.SetDSCP(remappingProducer.DSCP()); err != nil {
	log.Infoln(time.Now().Format(time.RFC3339Nano) + " " + r.RemoteAddr + " " + r.Method + " " + r.RequestURI + ": could not set DSCP: " + err.Error() + " (reqid " + strconv.FormatUint(reqID, 10) + ")")
	}
	}

	reqHeader := web.CopyHeader(r.Header) // copy request header, because it's not guaranteed valid after actually issuing the request
	clientIP, _ := web.GetClientIPPort(r)

	toFQDN := ""
	pluginCfg := map[string]interface{}{}
	if remappingProducer != nil {
	toFQDN = remappingProducer.FirstFQDN()
	pluginCfg = remappingProducer.PluginCfg()
	}

	reqData := cachedata.ReqData{Req: r, Conn: conn, ClientIP: clientIP, ReqTime: reqTime, ToFQDN: toFQDN}
	responder := NewResponder(w, pluginCfg, pluginContext, srvrData, reqData, h.plugins, h.stats, reqID)

	if err != nil {
	switch err {
	case remap.ErrRuleNotFound:
	log.Debugf("rule not found for %v (reqid %v)\n", r.RequestURI, reqID)
	*responder.ResponseCode = http.StatusNotFound
	case remap.ErrIPNotAllowed:
	log.Debugf("IP %v not allowed (reqid %v)\n", r.RemoteAddr, reqID)
	*responder.ResponseCode = http.StatusForbidden
	default:
	log.Debugf("request error: %v (reqid %v)\n", err, reqID)
	}
	responder.OriginConnectFailed = true
	responder.Do()
	return
	}

	reqCacheControl := rfc.ParseCacheControl(reqHeader)
	log.Debugf("Serve got Cache-Control %+v (reqid %v)\n", reqCacheControl, reqID)

	connectionClose := h.connectionClose \|\| remappingProducer.ConnectionClose()

	beforeCacheLookUpData := plugin.BeforeCacheLookUpData{Req: r, DefaultCacheKey: remappingProducer.CacheKey(), CacheKeyOverrideFunc: remappingProducer.OverrideCacheKey}
	h.plugins.OnBeforeCacheLookup(remappingProducer.PluginCfg(), pluginContext, beforeCacheLookUpData)

	cacheKey := remappingProducer.CacheKey()
	retrier := NewRetrier(h, reqHeader, reqTime, reqCacheControl, remappingProducer, reqID)

	cache := remappingProducer.Cache()

	var reqHost *string
	cacheObj, ok := cache.Get(cacheKey)
	if !ok {
	log.Debugf("cache.Handler.ServeHTTP: '%v' not in cache (reqid %v)\n", cacheKey, reqID)
	beforeParentRequestData := plugin.BeforeParentRequestData{Req: r, RemapRule: remappingProducer.Name()}
	h.plugins.OnBeforeParentRequest(remappingProducer.PluginCfg(), pluginContext, beforeParentRequestData)
	cacheObj, reqHost, err = retrier.Get(r, nil)
	if err != nil {
	log.Errorf("retrying get error (in uncached): %v (reqid %v)\n", err, reqID)
	responder.OriginConnectFailed = true
	responder.ProxyStr = cacheObj.ProxyURL
	if reqHost != nil {
	responder.ToFQDN = *reqHost
	}
	responder.Do()
	return
	}

	responder.OriginCode = cacheObj.OriginCode
	// create new pointers, so plugins don't modify the cacheObj
	codePtr, hdrsPtr, bodyPtr := cacheObj.Code, cacheObj.RespHeaders, cacheObj.Body
	responder.SetResponse(&codePtr, &hdrsPtr, &bodyPtr, connectionClose)
	responder.OriginReqSuccess = true
	responder.ProxyStr = cacheObj.ProxyURL
	if reqHost != nil {
	responder.ToFQDN = *reqHost
	}
	beforeRespData := plugin.BeforeRespondData{Req: r, CacheObj: cacheObj, Code: &codePtr, Hdr: &hdrsPtr, Body: &bodyPtr, RemapRule: remappingProducer.Name()}
	h.plugins.OnBeforeRespond(remappingProducer.PluginCfg(), pluginContext, beforeRespData)
	responder.Do()
	return
	}

	reqHeaders := r.Header
	canReuseStored := rfc.CanReuseStored(reqHeaders, cacheObj.RespHeaders, reqCacheControl, cacheObj.RespCacheControl, cacheObj.ReqHeaders, cacheObj.ReqRespTime, cacheObj.RespRespTime, h.strictRFC)

	if canReuseStored != rfc.ReuseCan { // run the BeforeParentRequest hook for revalidations / ReuseCannot
	beforeParentRequestData := plugin.BeforeParentRequestData{Req: r, RemapRule: remappingProducer.Name()}
	h.plugins.OnBeforeParentRequest(remappingProducer.PluginCfg(), pluginContext, beforeParentRequestData)
	}

	switch canReuseStored {
	case rfc.ReuseCan:
	log.Debugf("cache.Handler.ServeHTTP: '%v' cache hit! (reqid %v)\n", cacheKey, reqID)
	case rfc.ReuseCannot:
	log.Debugf("cache.Handler.ServeHTTP: '%v' can't reuse (reqid %v)\n", cacheKey, reqID)
	cacheObj, reqHost, err = retrier.Get(r, nil)
	if err != nil {
	log.Errorf("retrying get error (in reuse-cannot): %v (reqid %v)\n", err, reqID)
	responder.Do()
	return
	}
	case rfc.ReuseMustRevalidate:
	log.Debugf("cache.Handler.ServeHTTP: '%v' must revalidate (reqid %v)\n", cacheKey, reqID)
	cacheObj, reqHost, err = retrier.Get(r, cacheObj)
	if err != nil {
	log.Errorf("retrying get error: %v (reqid %v)\n", err, reqID)
	responder.Do()
	return
	}
	case rfc.ReuseMustRevalidateCanStale:
	log.Debugf("cache.Handler.ServeHTTP: '%v' must revalidate (but allowed stale) (reqid %v)\n", cacheKey, reqID)
	oldCacheObj := cacheObj
	cacheObj, reqHost, err = retrier.Get(r, cacheObj)
	if err != nil {
	log.Errorf("retrying get error - serving stale as allowed: %v (reqid %v)\n", err, reqID)
	cacheObj = oldCacheObj
	}
	}
	log.Debugf("cache.Handler.ServeHTTP: '%v' responding with %v (reqid %v)\n", cacheKey, cacheObj.Code, reqID)

	// create new pointers, so plugins don't modify the cacheObj
	codePtr, hdrsPtr, bodyPtr := cacheObj.Code, cacheObj.RespHeaders, cacheObj.Body
	responder.SetResponse(&codePtr, &hdrsPtr, &bodyPtr, connectionClose)
	responder.OriginReqSuccess = true
	responder.Reuse = canReuseStored
	responder.OriginCode = cacheObj.OriginCode
	responder.OriginBytes = cacheObj.Size
	responder.ProxyStr = cacheObj.ProxyURL
	if reqHost != nil {
	responder.ToFQDN = *reqHost
	}
	beforeRespData := plugin.BeforeRespondData{Req: r, CacheObj: cacheObj, Code: &codePtr, Hdr: &hdrsPtr, Body: &bodyPtr, RemapRule: remappingProducer.Name()}
	h.plugins.OnBeforeRespond(remappingProducer.PluginCfg(), pluginContext, beforeRespData)
	responder.Do()
	}