1 // Copyright (C) The Arvados Authors. All rights reserved.
3 // SPDX-License-Identifier: AGPL-3.0
15 "git.curoverse.com/arvados.git/sdk/go/arvados"
16 "github.com/Sirupsen/logrus"
20 errQueueFull = errors.New("client queue full")
21 errFrameTooBig = errors.New("frame too big")
23 sendObjectAttributes = []string{"state", "name", "owner_uuid", "portable_data_hash"}
25 v0subscribeOK = []byte(`{"status":200}`)
26 v0subscribeFail = []byte(`{"status":400}`)
29 type v0session struct {
32 sendq chan<- interface{}
34 permChecker permChecker
35 subscriptions []v0subscribe
42 // newSessionV0 returns a v0 session: a partial port of the Rails/puma
43 // implementation, with just enough functionality to support Workbench
45 func newSessionV0(ws wsConn, sendq chan<- interface{}, db *sql.DB, pc permChecker, ac *arvados.Client) (session, error) {
52 log: logger(ws.Request().Context()),
55 err := ws.Request().ParseForm()
57 sess.log.WithError(err).Error("ParseForm failed")
60 token := ws.Request().Form.Get("api_token")
61 sess.permChecker.SetToken(token)
62 sess.log.WithField("token", token).Debug("set token")
67 func (sess *v0session) Receive(buf []byte) error {
69 if err := json.Unmarshal(buf, &sub); err != nil {
70 sess.log.WithError(err).Info("invalid message from client")
71 } else if sub.Method == "subscribe" {
73 sess.log.WithField("sub", sub).Debug("sub prepared")
74 sess.sendq <- v0subscribeOK
76 sess.subscriptions = append(sess.subscriptions, sub)
78 sub.sendOldEvents(sess)
81 sess.log.WithField("Method", sub.Method).Info("unknown method")
83 sess.sendq <- v0subscribeFail
87 func (sess *v0session) EventMessage(e *event) ([]byte, error) {
93 ok, err := sess.permChecker.Check(detail.ObjectUUID)
94 if err != nil || !ok {
98 kind, _ := sess.ac.KindForUUID(detail.ObjectUUID)
99 msg := map[string]interface{}{
100 "msgID": atomic.AddUint64(&sess.lastMsgID, 1),
103 "object_uuid": detail.ObjectUUID,
104 "object_owner_uuid": detail.ObjectOwnerUUID,
106 "event_type": detail.EventType,
107 "event_at": detail.EventAt,
109 if detail.Properties != nil && detail.Properties["text"] != nil {
110 msg["properties"] = detail.Properties
112 msgProps := map[string]map[string]interface{}{}
113 for _, ak := range []string{"old_attributes", "new_attributes"} {
114 eventAttrs, ok := detail.Properties[ak].(map[string]interface{})
118 msgAttrs := map[string]interface{}{}
119 for _, k := range sendObjectAttributes {
120 if v, ok := eventAttrs[k]; ok {
124 msgProps[ak] = msgAttrs
126 msg["properties"] = msgProps
128 return json.Marshal(msg)
131 func (sess *v0session) Filter(e *event) bool {
133 defer sess.mtx.Unlock()
134 for _, sub := range sess.subscriptions {
135 if sub.match(sess, e) {
142 func (sub *v0subscribe) sendOldEvents(sess *v0session) {
143 if sub.LastLogID == 0 {
146 sess.log.WithField("LastLogID", sub.LastLogID).Debug("getOldEvents")
147 // Here we do a "select id" query and queue an event for every
148 // log since the given ID, then use (*event)Detail() to
149 // retrieve the whole row and decide whether to send it. This
150 // approach is very inefficient if the subscriber asks for
151 // last_log_id==1, even if the filters end up matching very
154 // To mitigate this, filter on "created > 10 minutes ago" when
155 // retrieving the list of old event IDs to consider.
156 rows, err := sess.db.Query(
157 `SELECT id FROM logs WHERE id > $1 AND created_at > $2 ORDER BY id`,
159 time.Now().UTC().Add(-10*time.Minute).Format(time.RFC3339Nano))
161 sess.log.WithError(err).Error("db.Query failed")
167 err := rows.Scan(&id)
169 sess.log.WithError(err).Error("row Scan failed")
172 for len(sess.sendq)*2 > cap(sess.sendq) {
173 // Ugly... but if we fill up the whole client
174 // queue with a backlog of old events, a
175 // single new event will overflow it and
176 // terminate the connection, and then the
177 // client will probably reconnect and do the
178 // same thing all over again.
179 time.Sleep(100 * time.Millisecond)
188 if sub.match(sess, e) {
190 case sess.sendq <- e:
191 case <-sess.ws.Request().Context().Done():
196 if err := rows.Err(); err != nil {
197 sess.log.WithError(err).Error("db.Query failed")
201 type v0subscribe struct {
204 LastLogID int64 `json:"last_log_id"`
206 funcs []func(*event) bool
209 type v0filter [3]interface{}
211 func (sub *v0subscribe) match(sess *v0session, e *event) bool {
212 log := sess.log.WithField("LogID", e.LogID)
215 log.Error("match failed, no detail")
218 log = log.WithField("funcs", len(sub.funcs))
219 for i, f := range sub.funcs {
221 log.WithField("func", i).Debug("match failed")
225 log.Debug("match passed")
229 func (sub *v0subscribe) prepare(sess *v0session) {
230 for _, f := range sub.Filters {
234 if col, ok := f[0].(string); ok && col == "event_type" {
235 op, ok := f[1].(string)
236 if !ok || op != "in" {
239 arr, ok := f[2].([]interface{})
244 for _, s := range arr {
245 if s, ok := s.(string); ok {
246 strs = append(strs, s)
249 sub.funcs = append(sub.funcs, func(e *event) bool {
250 for _, s := range strs {
251 if s == e.Detail().EventType {
257 } else if ok && col == "created_at" {
258 op, ok := f[1].(string)
262 tstr, ok := f[2].(string)
266 t, err := time.Parse(time.RFC3339Nano, tstr)
268 sess.log.WithField("data", tstr).WithError(err).Info("time.Parse failed")
271 var fn func(*event) bool
274 fn = func(e *event) bool {
275 return !e.Detail().CreatedAt.Before(t)
278 fn = func(e *event) bool {
279 return !e.Detail().CreatedAt.After(t)
282 fn = func(e *event) bool {
283 return e.Detail().CreatedAt.After(t)
286 fn = func(e *event) bool {
287 return e.Detail().CreatedAt.Before(t)
290 fn = func(e *event) bool {
291 return e.Detail().CreatedAt.Equal(t)
294 sess.log.WithField("operator", op).Info("bogus operator")
297 sub.funcs = append(sub.funcs, fn)