+// Route /arvados/v1/containers/{uuid}/log*, .../ssh, and
+// .../gateway_tunnel to rtr, pass everything else to next.
+//
+// (http.ServeMux doesn't let us route these without also routing
+// everything under /containers/, which we don't want yet.)
+func (h *Handler) routeContainerEndpoints(rtr http.Handler) middlewareFunc {
+ return func(w http.ResponseWriter, req *http.Request, next http.Handler) {
+ trim := strings.TrimPrefix(req.URL.Path, "/arvados/v1/containers/")
+ if trim != req.URL.Path && (strings.Index(trim, "/log") == 27 ||
+ strings.Index(trim, "/ssh") == 27 ||
+ strings.Index(trim, "/gateway_tunnel") == 27) {
+ rtr.ServeHTTP(w, req)
+ } else {
+ next.ServeHTTP(w, req)
+ }
+ }
+}
+
+func (h *Handler) limitLogCreateRequests(w http.ResponseWriter, req *http.Request, next http.Handler) {
+ if cap(h.limitLogCreate) > 0 && req.Method == http.MethodPost && strings.HasPrefix(req.URL.Path, "/arvados/v1/logs") {
+ select {
+ case h.limitLogCreate <- struct{}{}:
+ defer func() { <-h.limitLogCreate }()
+ next.ServeHTTP(w, req)
+ default:
+ http.Error(w, "Excess log messages", http.StatusServiceUnavailable)
+ }
+ return
+ }
+ next.ServeHTTP(w, req)
+}
+
+// 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)
+}
+