AdminToken = "4axaw8zxe0qm22wa6urpp5nskcne8z88cvbupv653y1njyi05h"
AnonymousToken = "4kg6k6lzmp9kj4cpkcoxie964cmvjahbt4fod9zru44k4jqdmi"
DataManagerToken = "320mkve8qkswstz7ff61glpk3mhgghmg67wmic7elw4z41pke1"
+ ManagementToken = "jg3ajndnq63sywcd50gbs5dskdc9ckkysb0nsqmfz08nwf17nl"
ActiveUserUUID = "zzzzz-tpzed-xurymjxw79nv3jz"
SpectatorUserUUID = "zzzzz-tpzed-l1s2piq4t4mps8r"
UserAgreementCollection = "zzzzz-4zz18-uukreo9rbgwsujr" // user_agreement_in_anonymously_accessible_project
PingTimeout arvados.Duration
ClientEventQueue int
ServerEventQueue int
+
+ ManagementToken string
}
func defaultConfig() wsConfig {
type eventSource interface {
NewSink() eventSink
DB() *sql.DB
+ DBHealth() error
}
type event struct {
return ps.db
}
+func (ps *pgEventSource) DBHealth() error {
+ ctx, _ := context.WithDeadline(context.Background(), time.Now().Add(time.Second))
+ var i int
+ return ps.db.QueryRowContext(ctx, "SELECT 1").Scan(&i)
+}
+
func (ps *pgEventSource) DebugStatus() interface{} {
ps.mtx.Lock()
defer ps.mtx.Unlock()
"QueueDelay": stats.Duration(ps.lastQDelay),
"Sinks": len(ps.sinks),
"SinksBlocked": blocked,
+ "DBStats": ps.db.Stats(),
}
}
case <-time.After(10 * time.Second):
c.Fatal("timed out")
}
+
+ c.Check(pges.DBHealth(), check.IsNil)
}
rtr.mux = http.NewServeMux()
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))
+ rtr.mux.Handle("/debug.json", rtr.jsonHandler(rtr.DebugStatus))
+ rtr.mux.Handle("/status.json", rtr.jsonHandler(rtr.Status))
+
+ health := http.NewServeMux()
+ rtr.mux.Handle("/_health/", rtr.mgmtAuth(health))
+ health.Handle("/_health/ping", rtr.jsonHandler(rtr.HealthFunc(func() error { return nil })))
+ health.Handle("/_health/db", rtr.jsonHandler(rtr.HealthFunc(rtr.eventSource.DBHealth)))
}
func (rtr *router) makeServer(newSession sessionFactory) *websocket.Server {
return s
}
+var pingResponseOK = map[string]string{"health": "OK"}
+
+func (rtr *router) HealthFunc(f func() error) func() interface{} {
+ return func() interface{} {
+ err := f()
+ if err == nil {
+ return pingResponseOK
+ }
+ return map[string]string{
+ "health": "ERROR",
+ "error": err.Error(),
+ }
+ }
+}
+
func (rtr *router) Status() interface{} {
return map[string]interface{}{
"Clients": atomic.LoadInt64(&rtr.status.ReqsActive),
rtr.mux.ServeHTTP(resp, req)
}
-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)
+func (rtr *router) mgmtAuth(h http.Handler) http.Handler {
+ return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ if rtr.Config.ManagementToken == "" {
+ http.Error(w, "disabled", http.StatusNotFound)
+ } else if ah := r.Header.Get("Authorization"); ah == "" {
+ http.Error(w, "authorization required", http.StatusUnauthorized)
+ } else if ah != "Bearer "+rtr.Config.ManagementToken {
+ http.Error(w, "authorization error", http.StatusForbidden)
+ } else {
+ h.ServeHTTP(w, r)
+ }
+ })
+}
+
+func (rtr *router) jsonHandler(fn func() interface{}) http.Handler {
+ return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ logger := logger(r.Context())
+ w.Header().Set("Content-Type", "application/json")
+ enc := json.NewEncoder(w)
err := enc.Encode(fn())
if err != nil {
msg := "encode failed"
logger.WithError(err).Error(msg)
- http.Error(resp, msg, http.StatusInternalServerError)
+ http.Error(w, msg, http.StatusInternalServerError)
}
- }
+ })
}
package main
import (
+ "io/ioutil"
+ "net/http"
"sync"
"time"
"git.curoverse.com/arvados.git/sdk/go/arvados"
+ "git.curoverse.com/arvados.git/sdk/go/arvadostest"
check "gopkg.in/check.v1"
)
var _ = check.Suite(&serverSuite{})
type serverSuite struct {
+ cfg *wsConfig
+ srv *server
+ wg sync.WaitGroup
}
-func testConfig() *wsConfig {
+func (s *serverSuite) SetUpTest(c *check.C) {
+ s.cfg = s.testConfig()
+ s.srv = &server{wsConfig: s.cfg}
+}
+
+func (*serverSuite) testConfig() *wsConfig {
cfg := defaultConfig()
cfg.Client = *(arvados.NewClientFromEnv())
cfg.Postgres = testDBConfig()
cfg.Listen = ":"
+ cfg.ManagementToken = arvadostest.ManagementToken
return &cfg
}
// TestBadDB ensures Run() returns an error (instead of panicking or
// deadlocking) if it can't connect to the database server at startup.
func (s *serverSuite) TestBadDB(c *check.C) {
- cfg := testConfig()
- cfg.Postgres["password"] = "1234"
- srv := &server{wsConfig: cfg}
+ s.cfg.Postgres["password"] = "1234"
var wg sync.WaitGroup
wg.Add(1)
go func() {
- err := srv.Run()
+ err := s.srv.Run()
c.Check(err, check.NotNil)
wg.Done()
}()
wg.Add(1)
go func() {
- srv.WaitReady()
+ s.srv.WaitReady()
wg.Done()
}()
}
}
-func newTestServer() *server {
- srv := &server{wsConfig: testConfig()}
- go srv.Run()
- srv.WaitReady()
- return srv
+func (s *serverSuite) TestHealth(c *check.C) {
+ go s.srv.Run()
+ defer s.srv.Close()
+ s.srv.WaitReady()
+ for _, token := range []string{"", "foo", s.cfg.ManagementToken} {
+ req, err := http.NewRequest("GET", "http://"+s.srv.listener.Addr().String()+"/_health/ping", nil)
+ c.Assert(err, check.IsNil)
+ if token != "" {
+ req.Header.Add("Authorization", "Bearer "+token)
+ }
+ resp, err := http.DefaultClient.Do(req)
+ c.Check(err, check.IsNil)
+ if token == s.cfg.ManagementToken {
+ c.Check(resp.StatusCode, check.Equals, http.StatusOK)
+ buf, err := ioutil.ReadAll(resp.Body)
+ c.Check(err, check.IsNil)
+ c.Check(string(buf), check.Equals, `{"health":"OK"}`+"\n")
+ } else {
+ c.Check(resp.StatusCode, check.Not(check.Equals), http.StatusOK)
+ }
+ }
+}
+
+func (s *serverSuite) TestHealthDisabled(c *check.C) {
+ s.cfg.ManagementToken = ""
+
+ go s.srv.Run()
+ defer s.srv.Close()
+ s.srv.WaitReady()
+
+ for _, token := range []string{"", "foo", arvadostest.ManagementToken} {
+ req, err := http.NewRequest("GET", "http://"+s.srv.listener.Addr().String()+"/_health/ping", nil)
+ c.Assert(err, check.IsNil)
+ req.Header.Add("Authorization", "Bearer "+token)
+ resp, err := http.DefaultClient.Do(req)
+ c.Check(err, check.IsNil)
+ c.Check(resp.StatusCode, check.Equals, http.StatusNotFound)
+ }
}
sess.log.WithError(err).Error("db.Query failed")
return
}
+ defer rows.Close()
for rows.Next() {
var id uint64
err := rows.Scan(&id)
var _ = check.Suite(&v0Suite{})
type v0Suite struct {
- token string
- toDelete []string
+ serverSuite serverSuite
+ token string
+ toDelete []string
}
func (s *v0Suite) SetUpTest(c *check.C) {
+ s.serverSuite.SetUpTest(c)
s.token = arvadostest.ActiveToken
}
}
func (s *v0Suite) testClient() (*server, *websocket.Conn, *json.Decoder, *json.Encoder) {
- srv := newTestServer()
+ go s.serverSuite.srv.Run()
+ s.serverSuite.srv.WaitReady()
+ srv := s.serverSuite.srv
conn, err := websocket.Dial("ws://"+srv.listener.Addr().String()+"/websocket?api_token="+s.token, "", "http://"+srv.listener.Addr().String())
if err != nil {
panic(err)
projects / arvados.git / commitdiff