X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/8595030d0314e6f88a245e66f90cce0a306b6867..0fc6eaead0bf7691e99d19e74ec33636909001a7:/services/ws/router.go diff --git a/services/ws/router.go b/services/ws/router.go index 3f9800f04c..15b825f2ab 100644 --- a/services/ws/router.go +++ b/services/ws/router.go @@ -1,7 +1,6 @@ package main import ( - "database/sql" "encoding/json" "io" "net/http" @@ -10,6 +9,7 @@ import ( "sync/atomic" "time" + "git.curoverse.com/arvados.git/sdk/go/ctxlog" "github.com/Sirupsen/logrus" "golang.org/x/net/websocket" ) @@ -22,7 +22,7 @@ type wsConn interface { } type router struct { - Config *Config + Config *wsConfig eventSource eventSource newPermChecker func() permChecker @@ -41,20 +41,18 @@ type routerDebugStatus struct { ReqsActive int64 } -type DebugStatuser interface { +type debugStatuser interface { DebugStatus() interface{} } -type sessionFactory func(wsConn, chan<- interface{}, *sql.DB, permChecker) (session, error) - func (rtr *router) setup() { rtr.handler = &handler{ PingTimeout: rtr.Config.PingTimeout.Duration(), QueueSize: rtr.Config.ClientEventQueue, } rtr.mux = http.NewServeMux() - rtr.mux.Handle("/websocket", rtr.makeServer(NewSessionV0)) - rtr.mux.Handle("/arvados/v1/events.ws", rtr.makeServer(NewSessionV1)) + rtr.mux.Handle("/websocket", rtr.makeServer(newSessionV0)) + rtr.mux.Handle("/arvados/v1/events.ws", rtr.makeServer(newSessionV1)) rtr.mux.HandleFunc("/debug.json", jsonHandler(rtr.DebugStatus)) rtr.mux.HandleFunc("/status.json", jsonHandler(rtr.Status)) } @@ -71,12 +69,12 @@ func (rtr *router) makeServer(newSession sessionFactory) *websocket.Server { stats := rtr.handler.Handle(ws, rtr.eventSource, func(ws wsConn, sendq chan<- interface{}) (session, error) { - return newSession(ws, sendq, rtr.eventSource.DB(), rtr.newPermChecker()) + return newSession(ws, sendq, rtr.eventSource.DB(), rtr.newPermChecker(), &rtr.Config.Client) }) log.WithFields(logrus.Fields{ - "Elapsed": time.Now().Sub(t0).Seconds(), - "Stats": stats, + "elapsed": time.Now().Sub(t0).Seconds(), + "stats": stats, }).Info("disconnect") ws.Close() }), @@ -98,7 +96,7 @@ func (rtr *router) DebugStatus() interface{} { "HTTP": rtr.status, "Outgoing": rtr.handler.DebugStatus(), } - if es, ok := rtr.eventSource.(DebugStatuser); ok { + if es, ok := rtr.eventSource.(debugStatuser); ok { s["EventSource"] = es.DebugStatus() } return s @@ -118,11 +116,11 @@ func (rtr *router) ServeHTTP(resp http.ResponseWriter, req *http.Request) { logger := logger(req.Context()). WithField("RequestID", rtr.newReqID()) - ctx := contextWithLogger(req.Context(), logger) + ctx := ctxlog.Context(req.Context(), logger) req = req.WithContext(ctx) logger.WithFields(logrus.Fields{ - "RemoteAddr": req.RemoteAddr, - "X-Forwarded-For": req.Header.Get("X-Forwarded-For"), + "remoteAddr": req.RemoteAddr, + "reqForwardedFor": req.Header.Get("X-Forwarded-For"), }).Info("accept request") rtr.mux.ServeHTTP(resp, req) } @@ -130,6 +128,7 @@ func (rtr *router) ServeHTTP(resp http.ResponseWriter, req *http.Request) { func jsonHandler(fn func() interface{}) http.HandlerFunc { return func(resp http.ResponseWriter, req *http.Request) { logger := logger(req.Context()) + resp.Header().Set("Content-Type", "application/json") enc := json.NewEncoder(resp) err := enc.Encode(fn()) if err != nil {