services/ws/session_v0.go

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