services/ws/session_v0.go

   1 package main
   2
   3 import (
   4         "database/sql"
   5         "encoding/json"
   6         "errors"
   7         "sync"
   8         "sync/atomic"
   9         "time"
  10
  11         "git.curoverse.com/arvados.git/sdk/go/arvados"
  12         "github.com/Sirupsen/logrus"
  13 )
  14
  15 var (
  16         errQueueFull   = errors.New("client queue full")
  17         errFrameTooBig = errors.New("frame too big")
  18
  19         sendObjectAttributes = []string{"state", "name", "owner_uuid", "portable_data_hash"}
  20
  21         v0subscribeOK   = []byte(`{"status":200}`)
  22         v0subscribeFail = []byte(`{"status":400}`)
  23 )
  24
  25 type v0session struct {
  26         ac            *arvados.Client
  27         ws            wsConn
  28         sendq         chan<- interface{}
  29         db            *sql.DB
  30         permChecker   permChecker
  31         subscriptions []v0subscribe
  32         lastMsgID     uint64
  33         log           *logrus.Entry
  34         mtx           sync.Mutex
  35         setupOnce     sync.Once
  36 }
  37
  38 // newSessionV0 returns a v0 session: a partial port of the Rails/puma
  39 // implementation, with just enough functionality to support Workbench
  40 // and arv-mount.
  41 func newSessionV0(ws wsConn, sendq chan<- interface{}, db *sql.DB, pc permChecker, ac *arvados.Client) (session, error) {
  42         sess := &v0session{
  43                 sendq:       sendq,
  44                 ws:          ws,
  45                 db:          db,
  46                 ac:          ac,
  47                 permChecker: pc,
  48                 log:         logger(ws.Request().Context()),
  49         }
  50
  51         err := ws.Request().ParseForm()
  52         if err != nil {
  53                 sess.log.WithError(err).Error("ParseForm failed")
  54                 return nil, err
  55         }
  56         token := ws.Request().Form.Get("api_token")
  57         sess.permChecker.SetToken(token)
  58         sess.log.WithField("token", token).Debug("set token")
  59
  60         return sess, nil
  61 }
  62
  63 func (sess *v0session) Receive(buf []byte) error {
  64         var sub v0subscribe
  65         if err := json.Unmarshal(buf, &sub); err != nil {
  66                 sess.log.WithError(err).Info("invalid message from client")
  67         } else if sub.Method == "subscribe" {
  68                 sub.prepare(sess)
  69                 sess.log.WithField("sub", sub).Debug("sub prepared")
  70                 sess.sendq <- v0subscribeOK
  71                 sess.mtx.Lock()
  72                 sess.subscriptions = append(sess.subscriptions, sub)
  73                 sess.mtx.Unlock()
  74                 sub.sendOldEvents(sess)
  75                 return nil
  76         } else {
  77                 sess.log.WithField("Method", sub.Method).Info("unknown method")
  78         }
  79         sess.sendq <- v0subscribeFail
  80         return nil
  81 }
  82
  83 func (sess *v0session) EventMessage(e *event) ([]byte, error) {
  84         detail := e.Detail()
  85         if detail == nil {
  86                 return nil, nil
  87         }
  88
  89         ok, err := sess.permChecker.Check(detail.ObjectUUID)
  90         if err != nil || !ok {
  91                 return nil, err
  92         }
  93
  94         kind, _ := sess.ac.KindForUUID(detail.ObjectUUID)
  95         msg := map[string]interface{}{
  96                 "msgID":             atomic.AddUint64(&sess.lastMsgID, 1),
  97                 "id":                detail.ID,
  98                 "uuid":              detail.UUID,
  99                 "object_uuid":       detail.ObjectUUID,
 100                 "object_owner_uuid": detail.ObjectOwnerUUID,
 101                 "object_kind":       kind,
 102                 "event_type":        detail.EventType,
 103                 "event_at":          detail.EventAt,
 104         }
 105         if detail.Properties != nil && detail.Properties["text"] != nil {
 106                 msg["properties"] = detail.Properties
 107         } else {
 108                 msgProps := map[string]map[string]interface{}{}
 109                 for _, ak := range []string{"old_attributes", "new_attributes"} {
 110                         eventAttrs, ok := detail.Properties[ak].(map[string]interface{})
 111                         if !ok {
 112                                 continue
 113                         }
 114                         msgAttrs := map[string]interface{}{}
 115                         for _, k := range sendObjectAttributes {
 116                                 if v, ok := eventAttrs[k]; ok {
 117                                         msgAttrs[k] = v
 118                                 }
 119                         }
 120                         msgProps[ak] = msgAttrs
 121                 }
 122                 msg["properties"] = msgProps
 123         }
 124         return json.Marshal(msg)
 125 }
 126
 127 func (sess *v0session) Filter(e *event) bool {
 128         sess.mtx.Lock()
 129         defer sess.mtx.Unlock()
 130         for _, sub := range sess.subscriptions {
 131                 if sub.match(sess, e) {
 132                         return true
 133                 }
 134         }
 135         return false
 136 }
 137
 138 func (sub *v0subscribe) sendOldEvents(sess *v0session) {
 139         if sub.LastLogID == 0 {
 140                 return
 141         }
 142         sess.log.WithField("LastLogID", sub.LastLogID).Debug("getOldEvents")
 143         // Here we do a "select id" query and queue an event for every
 144         // log since the given ID, then use (*event)Detail() to
 145         // retrieve the whole row and decide whether to send it. This
 146         // approach is very inefficient if the subscriber asks for
 147         // last_log_id==1, even if the filters end up matching very
 148         // few events.
 149         //
 150         // To mitigate this, filter on "created > 10 minutes ago" when
 151         // retrieving the list of old event IDs to consider.
 152         rows, err := sess.db.Query(
 153                 `SELECT id FROM logs WHERE id > $1 AND created_at > $2 ORDER BY id`,
 154                 sub.LastLogID,
 155                 time.Now().UTC().Add(-10*time.Minute).Format(time.RFC3339Nano))
 156         if err != nil {
 157                 sess.log.WithError(err).Error("db.Query failed")
 158                 return
 159         }
 160         defer rows.Close()
 161         for rows.Next() {
 162                 var id uint64
 163                 err := rows.Scan(&id)
 164                 if err != nil {
 165                         sess.log.WithError(err).Error("row Scan failed")
 166                         continue
 167                 }
 168                 for len(sess.sendq)*2 > cap(sess.sendq) {
 169                         // Ugly... but if we fill up the whole client
 170                         // queue with a backlog of old events, a
 171                         // single new event will overflow it and
 172                         // terminate the connection, and then the
 173                         // client will probably reconnect and do the
 174                         // same thing all over again.
 175                         time.Sleep(100 * time.Millisecond)
 176                 }
 177                 now := time.Now()
 178                 e := &event{
 179                         LogID:    id,
 180                         Received: now,
 181                         Ready:    now,
 182                         db:       sess.db,
 183                 }
 184                 if sub.match(sess, e) {
 185                         select {
 186                         case sess.sendq <- e:
 187                         case <-sess.ws.Request().Context().Done():
 188                                 return
 189                         }
 190                 }
 191         }
 192         if err := rows.Err(); err != nil {
 193                 sess.log.WithError(err).Error("db.Query failed")
 194         }
 195 }
 196
 197 type v0subscribe struct {
 198         Method    string
 199         Filters   []v0filter
 200         LastLogID int64 `json:"last_log_id"`
 201
 202         funcs []func(*event) bool
 203 }
 204
 205 type v0filter [3]interface{}
 206
 207 func (sub *v0subscribe) match(sess *v0session, e *event) bool {
 208         log := sess.log.WithField("LogID", e.LogID)
 209         detail := e.Detail()
 210         if detail == nil {
 211                 log.Error("match failed, no detail")
 212                 return false
 213         }
 214         log = log.WithField("funcs", len(sub.funcs))
 215         for i, f := range sub.funcs {
 216                 if !f(e) {
 217                         log.WithField("func", i).Debug("match failed")
 218                         return false
 219                 }
 220         }
 221         log.Debug("match passed")
 222         return true
 223 }
 224
 225 func (sub *v0subscribe) prepare(sess *v0session) {
 226         for _, f := range sub.Filters {
 227                 if len(f) != 3 {
 228                         continue
 229                 }
 230                 if col, ok := f[0].(string); ok && col == "event_type" {
 231                         op, ok := f[1].(string)
 232                         if !ok || op != "in" {
 233                                 continue
 234                         }
 235                         arr, ok := f[2].([]interface{})
 236                         if !ok {
 237                                 continue
 238                         }
 239                         var strs []string
 240                         for _, s := range arr {
 241                                 if s, ok := s.(string); ok {
 242                                         strs = append(strs, s)
 243                                 }
 244                         }
 245                         sub.funcs = append(sub.funcs, func(e *event) bool {
 246                                 for _, s := range strs {
 247                                         if s == e.Detail().EventType {
 248                                                 return true
 249                                         }
 250                                 }
 251                                 return false
 252                         })
 253                 } else if ok && col == "created_at" {
 254                         op, ok := f[1].(string)
 255                         if !ok {
 256                                 continue
 257                         }
 258                         tstr, ok := f[2].(string)
 259                         if !ok {
 260                                 continue
 261                         }
 262                         t, err := time.Parse(time.RFC3339Nano, tstr)
 263                         if err != nil {
 264                                 sess.log.WithField("data", tstr).WithError(err).Info("time.Parse failed")
 265                                 continue
 266                         }
 267                         var fn func(*event) bool
 268                         switch op {
 269                         case ">=":
 270                                 fn = func(e *event) bool {
 271                                         return !e.Detail().CreatedAt.Before(t)
 272                                 }
 273                         case "<=":
 274                                 fn = func(e *event) bool {
 275                                         return !e.Detail().CreatedAt.After(t)
 276                                 }
 277                         case ">":
 278                                 fn = func(e *event) bool {
 279                                         return e.Detail().CreatedAt.After(t)
 280                                 }
 281                         case "<":
 282                                 fn = func(e *event) bool {
 283                                         return e.Detail().CreatedAt.Before(t)
 284                                 }
 285                         case "=":
 286                                 fn = func(e *event) bool {
 287                                         return e.Detail().CreatedAt.Equal(t)
 288                                 }
 289                         default:
 290                                 sess.log.WithField("operator", op).Info("bogus operator")
 291                                 continue
 292                         }
 293                         sub.funcs = append(sub.funcs, fn)
 294                 }
 295         }
 296 }