import (
"context"
- "errors"
"fmt"
"net/http"
+ "net/http/httptest"
"net/url"
"strings"
"sync"
- "time"
"git.arvados.org/arvados.git/lib/controller/api"
"git.arvados.org/arvados.git/lib/controller/federation"
"git.arvados.org/arvados.git/lib/controller/router"
"git.arvados.org/arvados.git/lib/ctrlctx"
"git.arvados.org/arvados.git/sdk/go/arvados"
- "git.arvados.org/arvados.git/sdk/go/ctxlog"
"git.arvados.org/arvados.git/sdk/go/health"
"git.arvados.org/arvados.git/sdk/go/httpserver"
- "github.com/jmoiron/sqlx"
+
+ // sqlx needs lib/pq to talk to PostgreSQL
_ "github.com/lib/pq"
)
type Handler struct {
- Cluster *arvados.Cluster
+ Cluster *arvados.Cluster
+ BackgroundContext context.Context
setupOnce sync.Once
+ federation *federation.Conn
handlerStack http.Handler
proxy *proxy
secureClient *http.Client
insecureClient *http.Client
- pgdb *sqlx.DB
- pgdbMtx sync.Mutex
+ dbConnector ctrlctx.DBConnector
+ limitLogCreate chan struct{}
}
func (h *Handler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
req.URL.Path = strings.Replace(req.URL.Path, "//", "/", -1)
}
}
- if h.Cluster.API.RequestTimeout > 0 {
- ctx, cancel := context.WithDeadline(req.Context(), time.Now().Add(time.Duration(h.Cluster.API.RequestTimeout)))
- req = req.WithContext(ctx)
- defer cancel()
- }
-
h.handlerStack.ServeHTTP(w, req)
}
func (h *Handler) CheckHealth() error {
h.setupOnce.Do(h.setup)
- _, err := h.db(context.TODO())
+ _, err := h.dbConnector.GetDB(context.TODO())
if err != nil {
return err
}
_, _, err = railsproxy.FindRailsAPI(h.Cluster)
- return err
+ if err != nil {
+ return err
+ }
+ if h.Cluster.API.VocabularyPath != "" {
+ req, err := http.NewRequest("GET", "/arvados/v1/vocabulary", nil)
+ if err != nil {
+ return err
+ }
+ var resp httptest.ResponseRecorder
+ h.handlerStack.ServeHTTP(&resp, req)
+ if resp.Result().StatusCode != http.StatusOK {
+ return fmt.Errorf("%d %s", resp.Result().StatusCode, resp.Result().Status)
+ }
+ }
+ return nil
}
func (h *Handler) Done() <-chan struct{} {
func (h *Handler) setup() {
mux := http.NewServeMux()
+ healthFuncs := make(map[string]health.Func)
+
+ h.dbConnector = ctrlctx.DBConnector{PostgreSQL: h.Cluster.PostgreSQL}
+ oidcAuthorizer := localdb.OIDCAccessTokenAuthorizer(h.Cluster, h.dbConnector.GetDB)
+ h.federation = federation.New(h.Cluster, &healthFuncs)
+ rtr := router.New(h.federation, router.Config{
+ MaxRequestSize: h.Cluster.API.MaxRequestSize,
+ WrapCalls: api.ComposeWrappers(
+ ctrlctx.WrapCallsInTransactions(h.dbConnector.GetDB),
+ oidcAuthorizer.WrapCalls,
+ ctrlctx.WrapCallsWithAuth(h.Cluster)),
+ })
+
+ healthRoutes := health.Routes{"ping": func() error { _, err := h.dbConnector.GetDB(context.TODO()); return err }}
+ for name, f := range healthFuncs {
+ healthRoutes[name] = f
+ }
mux.Handle("/_health/", &health.Handler{
Token: h.Cluster.ManagementToken,
Prefix: "/_health/",
- Routes: health.Routes{"ping": func() error { _, err := h.db(context.TODO()); return err }},
+ Routes: healthRoutes,
})
-
- oidcAuthorizer := localdb.OIDCAccessTokenAuthorizer(h.Cluster, h.db)
- rtr := router.New(federation.New(h.Cluster), api.ComposeWrappers(ctrlctx.WrapCallsInTransactions(h.db), oidcAuthorizer.WrapCalls))
mux.Handle("/arvados/v1/config", rtr)
- mux.Handle("/"+arvados.EndpointUserAuthenticate.Path, rtr)
-
- if !h.Cluster.ForceLegacyAPI14 {
- mux.Handle("/arvados/v1/collections", rtr)
- mux.Handle("/arvados/v1/collections/", rtr)
- mux.Handle("/arvados/v1/users", rtr)
- mux.Handle("/arvados/v1/users/", rtr)
- mux.Handle("/login", rtr)
- mux.Handle("/logout", rtr)
- }
+ mux.Handle("/arvados/v1/vocabulary", rtr)
+ mux.Handle("/"+arvados.EndpointUserAuthenticate.Path, rtr) // must come before .../users/
+ mux.Handle("/arvados/v1/collections", rtr)
+ mux.Handle("/arvados/v1/collections/", rtr)
+ mux.Handle("/arvados/v1/users", rtr)
+ mux.Handle("/arvados/v1/users/", rtr)
+ mux.Handle("/arvados/v1/connect/", rtr)
+ mux.Handle("/arvados/v1/container_requests", rtr)
+ mux.Handle("/arvados/v1/container_requests/", rtr)
+ mux.Handle("/arvados/v1/groups", rtr)
+ mux.Handle("/arvados/v1/groups/", rtr)
+ mux.Handle("/arvados/v1/links", rtr)
+ mux.Handle("/arvados/v1/links/", rtr)
+ mux.Handle("/login", rtr)
+ mux.Handle("/logout", rtr)
+ mux.Handle("/arvados/v1/api_client_authorizations", rtr)
+ mux.Handle("/arvados/v1/api_client_authorizations/", rtr)
hs := http.NotFoundHandler()
hs = prepend(hs, h.proxyRailsAPI)
+ hs = prepend(hs, h.limitLogCreateRequests)
hs = h.setupProxyRemoteCluster(hs)
hs = prepend(hs, oidcAuthorizer.Middleware)
mux.Handle("/", hs)
ic.CheckRedirect = neverRedirect
h.insecureClient = &ic
- h.proxy = &proxy{
- Name: "arvados-controller",
+ logCreateLimit := int(float64(h.Cluster.API.MaxConcurrentRequests) * h.Cluster.API.LogCreateRequestFraction)
+ if logCreateLimit == 0 && h.Cluster.API.LogCreateRequestFraction > 0 {
+ logCreateLimit = 1
}
-}
-
-var errDBConnection = errors.New("database connection error")
+ h.limitLogCreate = make(chan struct{}, logCreateLimit)
-func (h *Handler) db(ctx context.Context) (*sqlx.DB, error) {
- h.pgdbMtx.Lock()
- defer h.pgdbMtx.Unlock()
- if h.pgdb != nil {
- return h.pgdb, nil
+ h.proxy = &proxy{
+ Name: "arvados-controller",
}
- db, err := sqlx.Open("postgres", h.Cluster.PostgreSQL.Connection.String())
- if err != nil {
- ctxlog.FromContext(ctx).WithError(err).Error("postgresql connect failed")
- return nil, errDBConnection
- }
- if p := h.Cluster.PostgreSQL.ConnectionPool; p > 0 {
- db.SetMaxOpenConns(p)
- }
- if err := db.Ping(); err != nil {
- ctxlog.FromContext(ctx).WithError(err).Error("postgresql connect succeeded but ping failed")
- return nil, errDBConnection
- }
- h.pgdb = db
- return db, nil
+ go h.trashSweepWorker()
+ go h.containerLogSweepWorker()
}
type middlewareFunc func(http.ResponseWriter, *http.Request, http.Handler)
return h.proxy.Do(req, urlOut, client)
}
+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)
+}
+
func (h *Handler) proxyRailsAPI(w http.ResponseWriter, req *http.Request, next http.Handler) {
resp, err := h.localClusterRequest(req)
n, err := h.proxy.ForwardResponse(w, resp, err)
projects / arvados.git / blobdiff