+// cacheEnt implements a basic stale-while-revalidate cache, suitable
+// for the Arvados discovery document.
+type cacheEnt struct {
+ validate func(body []byte) error
+ mtx sync.Mutex
+ header http.Header
+ body []byte
+ expireAfter time.Time
+ refreshAfter time.Time
+ refreshLock sync.Mutex
+}
+
+const (
+ cacheTTL = 5 * time.Minute
+ cacheExpire = 24 * time.Hour
+)
+
+func (ent *cacheEnt) refresh(path string, do func(*http.Request) (*http.Response, error)) (http.Header, []byte, error) {
+ ent.refreshLock.Lock()
+ defer ent.refreshLock.Unlock()
+ if header, body, needRefresh := ent.response(); !needRefresh {
+ // another goroutine refreshed successfully while we
+ // were waiting for refreshLock
+ return header, body, nil
+ } else if body != nil {
+ // Cache is present, but expired. We'll try to refresh
+ // below. Meanwhile, other refresh() calls will queue
+ // up for refreshLock -- and we don't want them to
+ // turn into N upstream requests, even if upstream is
+ // failing. (If we succeed we'll update the expiry
+ // time again below with the real cacheTTL -- this
+ // just takes care of the error case.)
+ ent.mtx.Lock()
+ ent.refreshAfter = time.Now().Add(time.Second)
+ ent.mtx.Unlock()
+ }
+
+ ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(time.Minute))
+ defer cancel()
+ // "http://localhost" is just a placeholder here -- we'll fill
+ // in req.URL.Path below, and then do(), which is
+ // localClusterRequest(), will replace the scheme and host
+ // parts with the real proxy destination.
+ req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://localhost", nil)
+ if err != nil {
+ return nil, nil, err
+ }
+ req.URL.Path = path
+ resp, err := do(req)
+ if err != nil {
+ return nil, nil, err
+ }
+ if resp.StatusCode != http.StatusOK {
+ return nil, nil, fmt.Errorf("HTTP status %d", resp.StatusCode)
+ }
+ body, err := ioutil.ReadAll(resp.Body)
+ if err != nil {
+ return nil, nil, fmt.Errorf("Read error: %w", err)
+ }
+ header := http.Header{}
+ for k, v := range resp.Header {
+ if !dropHeaders[k] && k != "X-Request-Id" {
+ header[k] = v
+ }
+ }
+ if ent.validate != nil {
+ if err := ent.validate(body); err != nil {
+ return nil, nil, err
+ }
+ } else if mediatype, _, err := mime.ParseMediaType(header.Get("Content-Type")); err == nil && mediatype == "application/json" {
+ if !json.Valid(body) {
+ return nil, nil, errors.New("invalid JSON encoding in response")
+ }
+ }
+ ent.mtx.Lock()
+ defer ent.mtx.Unlock()
+ ent.header = header
+ ent.body = body
+ ent.refreshAfter = time.Now().Add(cacheTTL)
+ ent.expireAfter = time.Now().Add(cacheExpire)
+ return ent.header, ent.body, nil
+}
+
+func (ent *cacheEnt) response() (http.Header, []byte, bool) {
+ ent.mtx.Lock()
+ defer ent.mtx.Unlock()
+ if ent.expireAfter.Before(time.Now()) {
+ ent.header, ent.body, ent.refreshAfter = nil, nil, time.Time{}
+ }
+ return ent.header, ent.body, ent.refreshAfter.Before(time.Now())
+}
+
+func (ent *cacheEnt) ServeHTTP(ctx context.Context, w http.ResponseWriter, path string, do func(*http.Request) (*http.Response, error)) {
+ header, body, needRefresh := ent.response()
+ if body == nil {
+ // need to fetch before we can return anything
+ var err error
+ header, body, err = ent.refresh(path, do)
+ if err != nil {
+ http.Error(w, err.Error(), http.StatusBadGateway)
+ return
+ }
+ } else if needRefresh {
+ // re-fetch in background
+ go func() {
+ _, _, err := ent.refresh(path, do)
+ if err != nil {
+ ctxlog.FromContext(ctx).WithError(err).WithField("path", path).Warn("error refreshing cache")
+ }
+ }()
+ }
+ for k, v := range header {
+ w.Header()[k] = v
+ }
+ w.WriteHeader(http.StatusOK)
+ w.Write(body)
+}
+