1 // Copyright (C) The Arvados Authors. All rights reserved.
3 // SPDX-License-Identifier: AGPL-3.0
21 "git.arvados.org/arvados.git/lib/controller/api"
22 "git.arvados.org/arvados.git/lib/controller/federation"
23 "git.arvados.org/arvados.git/lib/controller/localdb"
24 "git.arvados.org/arvados.git/lib/controller/railsproxy"
25 "git.arvados.org/arvados.git/lib/controller/router"
26 "git.arvados.org/arvados.git/lib/ctrlctx"
27 "git.arvados.org/arvados.git/sdk/go/arvados"
28 "git.arvados.org/arvados.git/sdk/go/ctxlog"
29 "git.arvados.org/arvados.git/sdk/go/health"
30 "git.arvados.org/arvados.git/sdk/go/httpserver"
32 // sqlx needs lib/pq to talk to PostgreSQL
37 Cluster *arvados.Cluster
38 BackgroundContext context.Context
41 federation *federation.Conn
42 handlerStack http.Handler
44 secureClient *http.Client
45 insecureClient *http.Client
46 dbConnector ctrlctx.DBConnector
47 limitLogCreate chan struct{}
49 cache map[string]*cacheEnt
52 func (h *Handler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
53 h.setupOnce.Do(h.setup)
54 if req.Method != "GET" && req.Method != "HEAD" {
55 // http.ServeMux returns 301 with a cleaned path if
56 // the incoming request has a double slash. Some
57 // clients (including the Go standard library) change
58 // the request method to GET when following a 301
59 // redirect if the original method was not HEAD
60 // (RFC7231 6.4.2 specifically allows this in the case
61 // of POST). Thus "POST //foo" gets misdirected to
62 // "GET /foo". To avoid this, eliminate double slashes
63 // before passing the request to ServeMux.
64 for strings.Contains(req.URL.Path, "//") {
65 req.URL.Path = strings.Replace(req.URL.Path, "//", "/", -1)
68 h.handlerStack.ServeHTTP(w, req)
71 func (h *Handler) CheckHealth() error {
72 h.setupOnce.Do(h.setup)
73 _, err := h.dbConnector.GetDB(context.TODO())
77 _, _, err = railsproxy.FindRailsAPI(h.Cluster)
81 if h.Cluster.API.VocabularyPath != "" {
82 req, err := http.NewRequest("GET", "/arvados/v1/vocabulary", nil)
86 var resp httptest.ResponseRecorder
87 h.handlerStack.ServeHTTP(&resp, req)
88 if resp.Result().StatusCode != http.StatusOK {
89 return fmt.Errorf("%d %s", resp.Result().StatusCode, resp.Result().Status)
95 func (h *Handler) Done() <-chan struct{} {
99 func neverRedirect(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }
101 func (h *Handler) setup() {
102 mux := http.NewServeMux()
103 healthFuncs := make(map[string]health.Func)
105 h.dbConnector = ctrlctx.DBConnector{PostgreSQL: h.Cluster.PostgreSQL}
107 <-h.BackgroundContext.Done()
108 h.dbConnector.Close()
110 oidcAuthorizer := localdb.OIDCAccessTokenAuthorizer(h.Cluster, h.dbConnector.GetDB)
111 h.federation = federation.New(h.BackgroundContext, h.Cluster, &healthFuncs, h.dbConnector.GetDB)
112 rtr := router.New(h.federation, router.Config{
113 MaxRequestSize: h.Cluster.API.MaxRequestSize,
114 WrapCalls: api.ComposeWrappers(
115 ctrlctx.WrapCallsInTransactions(h.dbConnector.GetDB),
116 oidcAuthorizer.WrapCalls,
117 ctrlctx.WrapCallsWithAuth(h.Cluster)),
120 healthRoutes := health.Routes{"ping": func() error { _, err := h.dbConnector.GetDB(context.TODO()); return err }}
121 for name, f := range healthFuncs {
122 healthRoutes[name] = f
124 mux.Handle("/_health/", &health.Handler{
125 Token: h.Cluster.ManagementToken,
127 Routes: healthRoutes,
129 mux.Handle("/arvados/v1/config", rtr)
130 mux.Handle("/arvados/v1/vocabulary", rtr)
131 mux.Handle("/"+arvados.EndpointUserAuthenticate.Path, rtr) // must come before .../users/
132 mux.Handle("/arvados/v1/collections", rtr)
133 mux.Handle("/arvados/v1/collections/", rtr)
134 mux.Handle("/arvados/v1/users", rtr)
135 mux.Handle("/arvados/v1/users/", rtr)
136 mux.Handle("/arvados/v1/connect/", rtr)
137 mux.Handle("/arvados/v1/container_requests", rtr)
138 mux.Handle("/arvados/v1/container_requests/", rtr)
139 mux.Handle("/arvados/v1/groups", rtr)
140 mux.Handle("/arvados/v1/groups/", rtr)
141 mux.Handle("/arvados/v1/links", rtr)
142 mux.Handle("/arvados/v1/links/", rtr)
143 mux.Handle("/login", rtr)
144 mux.Handle("/logout", rtr)
145 mux.Handle("/arvados/v1/api_client_authorizations", rtr)
146 mux.Handle("/arvados/v1/api_client_authorizations/", rtr)
148 hs := http.NotFoundHandler()
149 hs = prepend(hs, h.proxyRailsAPI)
150 hs = prepend(hs, h.routeContainerEndpoints(rtr))
151 hs = prepend(hs, h.limitLogCreateRequests)
152 hs = h.setupProxyRemoteCluster(hs)
153 hs = prepend(hs, oidcAuthorizer.Middleware)
157 sc := *arvados.DefaultSecureClient
158 sc.CheckRedirect = neverRedirect
161 ic := *arvados.InsecureHTTPClient
162 ic.CheckRedirect = neverRedirect
163 h.insecureClient = &ic
165 logCreateLimit := int(float64(h.Cluster.API.MaxConcurrentRequests) * h.Cluster.API.LogCreateRequestFraction)
166 if logCreateLimit == 0 && h.Cluster.API.LogCreateRequestFraction > 0 {
169 h.limitLogCreate = make(chan struct{}, logCreateLimit)
172 Name: "arvados-controller",
174 h.cache = map[string]*cacheEnt{
175 "/discovery/v1/apis/arvados/v1/rest": &cacheEnt{validate: validateDiscoveryDoc},
178 go h.trashSweepWorker()
179 go h.containerLogSweepWorker()
182 type middlewareFunc func(http.ResponseWriter, *http.Request, http.Handler)
184 func prepend(next http.Handler, middleware middlewareFunc) http.Handler {
185 return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
186 middleware(w, req, next)
190 func (h *Handler) localClusterRequest(req *http.Request) (*http.Response, error) {
191 urlOut, insecure, err := railsproxy.FindRailsAPI(h.Cluster)
196 Scheme: urlOut.Scheme,
199 RawPath: req.URL.RawPath,
200 RawQuery: req.URL.RawQuery,
202 client := h.secureClient
204 client = h.insecureClient
206 return h.proxy.Do(req, urlOut, client)
209 // Route /arvados/v1/containers/{uuid}/log*, .../ssh, and
210 // .../gateway_tunnel to rtr, pass everything else to next.
212 // (http.ServeMux doesn't let us route these without also routing
213 // everything under /containers/, which we don't want yet.)
214 func (h *Handler) routeContainerEndpoints(rtr http.Handler) middlewareFunc {
215 return func(w http.ResponseWriter, req *http.Request, next http.Handler) {
216 trim := strings.TrimPrefix(req.URL.Path, "/arvados/v1/containers/")
217 if trim != req.URL.Path && (strings.Index(trim, "/log") == 27 ||
218 strings.Index(trim, "/ssh") == 27 ||
219 strings.Index(trim, "/gateway_tunnel") == 27) {
220 rtr.ServeHTTP(w, req)
222 next.ServeHTTP(w, req)
227 func (h *Handler) limitLogCreateRequests(w http.ResponseWriter, req *http.Request, next http.Handler) {
228 if cap(h.limitLogCreate) > 0 && req.Method == http.MethodPost && strings.HasPrefix(req.URL.Path, "/arvados/v1/logs") {
230 case h.limitLogCreate <- struct{}{}:
231 defer func() { <-h.limitLogCreate }()
232 next.ServeHTTP(w, req)
234 http.Error(w, "Excess log messages", http.StatusServiceUnavailable)
238 next.ServeHTTP(w, req)
241 // cacheEnt implements a basic stale-while-revalidate cache, suitable
242 // for the Arvados discovery document.
243 type cacheEnt struct {
244 validate func(body []byte) error
248 expireAfter time.Time
249 refreshAfter time.Time
250 refreshLock sync.Mutex
254 cacheTTL = 5 * time.Minute
255 cacheExpire = 24 * time.Hour
258 func (ent *cacheEnt) refresh(path string, do func(*http.Request) (*http.Response, error)) (http.Header, []byte, error) {
259 ent.refreshLock.Lock()
260 defer ent.refreshLock.Unlock()
261 if header, body, needRefresh := ent.response(); !needRefresh {
262 // another goroutine refreshed successfully while we
263 // were waiting for refreshLock
264 return header, body, nil
265 } else if body != nil {
266 // Cache is present, but expired. We'll try to refresh
267 // below. Meanwhile, other refresh() calls will queue
268 // up for refreshLock -- and we don't want them to
269 // turn into N upstream requests, even if upstream is
270 // failing. (If we succeed we'll update the expiry
271 // time again below with the real cacheTTL -- this
272 // just takes care of the error case.)
274 ent.refreshAfter = time.Now().Add(time.Second)
278 ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(time.Minute))
280 // 0.0.0.0:0 is just a placeholder here -- do(), which is
281 // localClusterRequest(), will replace the scheme and host
282 // parts with the real proxy destination.
283 req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://0.0.0.0:0/"+path, nil)
291 if resp.StatusCode != http.StatusOK {
292 return nil, nil, fmt.Errorf("HTTP status %d", resp.StatusCode)
294 body, err := ioutil.ReadAll(resp.Body)
296 return nil, nil, fmt.Errorf("Read error: %w", err)
298 header := http.Header{}
299 for k, v := range resp.Header {
300 if !dropHeaders[k] && k != "X-Request-Id" {
304 if ent.validate != nil {
305 if err := ent.validate(body); err != nil {
308 } else if mediatype, _, err := mime.ParseMediaType(header.Get("Content-Type")); err == nil && mediatype == "application/json" {
309 if !json.Valid(body) {
310 return nil, nil, errors.New("invalid JSON encoding in response")
314 defer ent.mtx.Unlock()
317 ent.refreshAfter = time.Now().Add(cacheTTL)
318 ent.expireAfter = time.Now().Add(cacheExpire)
319 return ent.header, ent.body, nil
322 func (ent *cacheEnt) response() (http.Header, []byte, bool) {
324 defer ent.mtx.Unlock()
325 if ent.expireAfter.Before(time.Now()) {
326 ent.header, ent.body, ent.refreshAfter = nil, nil, time.Time{}
328 return ent.header, ent.body, ent.refreshAfter.Before(time.Now())
331 func (ent *cacheEnt) ServeHTTP(ctx context.Context, w http.ResponseWriter, path string, do func(*http.Request) (*http.Response, error)) {
332 header, body, needRefresh := ent.response()
334 // need to fetch before we can return anything
336 header, body, err = ent.refresh(path, do)
338 http.Error(w, err.Error(), http.StatusBadGateway)
341 } else if needRefresh {
342 // re-fetch in background
344 _, _, err := ent.refresh(path, do)
346 ctxlog.FromContext(ctx).WithError(err).WithField("path", path).Warn("error refreshing cache")
350 for k, v := range header {
353 w.WriteHeader(http.StatusOK)
357 func (h *Handler) proxyRailsAPI(w http.ResponseWriter, req *http.Request, next http.Handler) {
358 if ent, ok := h.cache[req.URL.Path]; ok && req.Method == http.MethodGet {
359 ent.ServeHTTP(req.Context(), w, req.URL.Path, h.localClusterRequest)
362 resp, err := h.localClusterRequest(req)
363 n, err := h.proxy.ForwardResponse(w, resp, err)
365 httpserver.Logger(req).WithError(err).WithField("bytesCopied", n).Error("error copying response body")
369 // Use a localhost entry from Services.RailsAPI.InternalURLs if one is
370 // present, otherwise choose an arbitrary entry.
371 func findRailsAPI(cluster *arvados.Cluster) (*url.URL, bool, error) {
373 for target := range cluster.Services.RailsAPI.InternalURLs {
374 target := url.URL(target)
376 if strings.HasPrefix(target.Host, "localhost:") || strings.HasPrefix(target.Host, "127.0.0.1:") || strings.HasPrefix(target.Host, "[::1]:") {
381 return nil, false, fmt.Errorf("Services.RailsAPI.InternalURLs is empty")
383 return best, cluster.TLS.Insecure, nil
386 func validateDiscoveryDoc(body []byte) error {
387 var dd arvados.DiscoveryDocument
388 err := json.Unmarshal(body, &dd)
390 return fmt.Errorf("error decoding JSON response: %w", err)
392 if dd.BasePath == "" {
393 return errors.New("error in discovery document: no value for basePath")