Merge branch '8784-dir-listings'
[arvados.git] / services / ws / handler.go
index 09ab945c76fe89bcb167402e9931394326f0d28e..f9f7f53edc58430f231e9a52d5d95bb1a025084a 100644 (file)
@@ -1,18 +1,27 @@
+// Copyright (C) The Arvados Authors. All rights reserved.
+//
+// SPDX-License-Identifier: AGPL-3.0
+
 package main
 
 import (
+       "context"
        "io"
+       "sync"
        "time"
 
        "git.curoverse.com/arvados.git/sdk/go/arvados"
-       log "github.com/Sirupsen/logrus"
+       "git.curoverse.com/arvados.git/sdk/go/stats"
 )
 
 type handler struct {
        Client      arvados.Client
        PingTimeout time.Duration
        QueueSize   int
-       NewSession  func(wsConn, chan<- interface{}) (session, error)
+
+       mtx       sync.Mutex
+       lastDelay map[chan interface{}]stats.Duration
+       setupOnce sync.Once
 }
 
 type handlerStats struct {
@@ -22,27 +31,39 @@ type handlerStats struct {
        EventCount   uint64
 }
 
-func (h *handler) Handle(ws wsConn, incoming <-chan *event) (stats handlerStats) {
-       ctx := contextWithLogger(ws.Request().Context(), log.WithFields(log.Fields{
-               "RemoteAddr": ws.Request().RemoteAddr,
-       }))
+func (h *handler) Handle(ws wsConn, eventSource eventSource, newSession func(wsConn, chan<- interface{}) (session, error)) (hStats handlerStats) {
+       h.setupOnce.Do(h.setup)
 
-       queue := make(chan interface{}, h.QueueSize)
-       sess, err := h.NewSession(ws, queue)
+       ctx, cancel := context.WithCancel(ws.Request().Context())
+       defer cancel()
        log := logger(ctx)
+
+       incoming := eventSource.NewSink()
+       defer incoming.Stop()
+
+       queue := make(chan interface{}, h.QueueSize)
+       h.mtx.Lock()
+       h.lastDelay[queue] = 0
+       h.mtx.Unlock()
+       defer func() {
+               h.mtx.Lock()
+               delete(h.lastDelay, queue)
+               h.mtx.Unlock()
+       }()
+
+       sess, err := newSession(ws, queue)
        if err != nil {
-               log.WithError(err).Error("NewSession failed")
+               log.WithError(err).Error("newSession failed")
                return
        }
 
-       stopped := make(chan struct{})
-       stop := make(chan error, 5)
-
+       // Receive websocket frames from the client and pass them to
+       // sess.Receive().
        go func() {
                buf := make([]byte, 2<<20)
                for {
                        select {
-                       case <-stopped:
+                       case <-ctx.Done():
                                return
                        default:
                        }
@@ -57,19 +78,33 @@ func (h *handler) Handle(ws wsConn, incoming <-chan *event) (stats handlerStats)
                                if err != io.EOF {
                                        log.WithError(err).Info("read error")
                                }
-                               stop <- err
+                               cancel()
                                return
                        }
                        err = sess.Receive(buf)
                        if err != nil {
-                               stop <- err
+                               log.WithError(err).Error("sess.Receive() failed")
+                               cancel()
                                return
                        }
                }
        }()
 
+       // Take items from the outgoing queue, serialize them using
+       // sess.EventMessage() as needed, and send them to the client
+       // as websocket frames.
        go func() {
-               for data := range queue {
+               for {
+                       var ok bool
+                       var data interface{}
+                       select {
+                       case <-ctx.Done():
+                               return
+                       case data, ok = <-queue:
+                               if !ok {
+                                       return
+                               }
+                       }
                        var e *event
                        var buf []byte
                        var err error
@@ -84,7 +119,7 @@ func (h *handler) Handle(ws wsConn, incoming <-chan *event) (stats handlerStats)
                                buf, err = sess.EventMessage(e)
                                if err != nil {
                                        log.WithError(err).Error("EventMessage failed")
-                                       stop <- err
+                                       cancel()
                                        break
                                } else if len(buf) == 0 {
                                        log.Debug("skip")
@@ -101,37 +136,35 @@ func (h *handler) Handle(ws wsConn, incoming <-chan *event) (stats handlerStats)
                        _, err = ws.Write(buf)
                        if err != nil {
                                log.WithError(err).Error("write failed")
-                               stop <- err
+                               cancel()
                                break
                        }
                        log.Debug("sent")
 
                        if e != nil {
-                               stats.QueueDelayNs += t0.Sub(e.Received)
+                               hStats.QueueDelayNs += t0.Sub(e.Ready)
+                               h.mtx.Lock()
+                               h.lastDelay[queue] = stats.Duration(time.Since(e.Ready))
+                               h.mtx.Unlock()
                        }
-                       stats.WriteDelayNs += time.Since(t0)
-                       stats.EventBytes += uint64(len(buf))
-                       stats.EventCount++
-               }
-               for _ = range queue {
-                       // Ensure queue can't fill up and block other
-                       // goroutines after we hit a write error.
+                       hStats.WriteDelayNs += time.Since(t0)
+                       hStats.EventBytes += uint64(len(buf))
+                       hStats.EventCount++
                }
        }()
 
        // Filter incoming events against the current subscription
        // list, and forward matching events to the outgoing message
-       // queue. Close the queue and return when the "stopped"
-       // channel closes or the incoming event stream ends. Shut down
-       // the handler if the outgoing queue fills up.
+       // queue. Close the queue and return when the request context
+       // is done/cancelled or the incoming event stream ends. Shut
+       // down the handler if the outgoing queue fills up.
        go func() {
                ticker := time.NewTicker(h.PingTimeout)
                defer ticker.Stop()
 
                for {
                        select {
-                       case <-stopped:
-                               close(queue)
+                       case <-ctx.Done():
                                return
                        case <-ticker.C:
                                // If the outgoing queue is empty,
@@ -140,12 +173,15 @@ func (h *handler) Handle(ws wsConn, incoming <-chan *event) (stats handlerStats)
                                // socket, and prevent an idle socket
                                // from being closed.
                                if len(queue) == 0 {
-                                       queue <- []byte(`{}`)
+                                       select {
+                                       case queue <- []byte(`{}`):
+                                       default:
+                                       }
                                }
                                continue
-                       case e, ok := <-incoming:
+                       case e, ok := <-incoming.Channel():
                                if !ok {
-                                       close(queue)
+                                       cancel()
                                        return
                                }
                                if !sess.Filter(e) {
@@ -154,14 +190,50 @@ func (h *handler) Handle(ws wsConn, incoming <-chan *event) (stats handlerStats)
                                select {
                                case queue <- e:
                                default:
-                                       stop <- errQueueFull
+                                       log.WithError(errQueueFull).Error("terminate")
+                                       cancel()
+                                       return
                                }
                        }
                }
        }()
 
-       <-stop
-       close(stopped)
-
+       <-ctx.Done()
        return
 }
+
+func (h *handler) DebugStatus() interface{} {
+       h.mtx.Lock()
+       defer h.mtx.Unlock()
+
+       var s struct {
+               QueueCount    int
+               QueueMin      int
+               QueueMax      int
+               QueueTotal    uint64
+               QueueDelayMin stats.Duration
+               QueueDelayMax stats.Duration
+       }
+       for q, lastDelay := range h.lastDelay {
+               s.QueueCount++
+               n := len(q)
+               s.QueueTotal += uint64(n)
+               if s.QueueMax < n {
+                       s.QueueMax = n
+               }
+               if s.QueueMin > n || s.QueueCount == 1 {
+                       s.QueueMin = n
+               }
+               if (s.QueueDelayMin > lastDelay || s.QueueDelayMin == 0) && lastDelay > 0 {
+                       s.QueueDelayMin = lastDelay
+               }
+               if s.QueueDelayMax < lastDelay {
+                       s.QueueDelayMax = lastDelay
+               }
+       }
+       return &s
+}
+
+func (h *handler) setup() {
+       h.lastDelay = make(map[chan interface{}]stats.Duration)
+}