jan/openreplay
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(backend): removed InputEvent, added use_brotli env variable

Browse source
This commit is contained in:
Alexander Zavorotynskiy 2023-10-09 14:09:08 +02:00
parent 9f0917ce5b
commit e698c76202
11 changed files with 4 additions and 141 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
backend/cmd/db/main.go
View file

@ -37,9 +37,9 @@ func main() {
messages.MsgUserID, messages.MsgUserAnonymousID, messages.MsgIntegrationEvent, messages.MsgPerformanceTrackAggr,
messages.MsgJSException, messages.MsgResourceTiming, messages.MsgCustomEvent, messages.MsgCustomIssue,
messages.MsgFetch, messages.MsgNetworkRequest, messages.MsgGraphQL, messages.MsgStateAction,
messages.MsgSetInputTarget, messages.MsgSetInputValue, messages.MsgCreateDocument, messages.MsgMouseClick,
messages.MsgCreateDocument, messages.MsgMouseClick,
messages.MsgSetPageLocation, messages.MsgPageLoadTiming, messages.MsgPageRenderTiming,
messages.MsgInputEvent, messages.MsgPageEvent, messages.MsgMouseThrashing, messages.MsgInputChange,
messages.MsgPageEvent, messages.MsgMouseThrashing, messages.MsgInputChange,
messages.MsgUnbindNodes}
// Init consumer

1
backend/cmd/heuristics/main.go
View file

@ -26,7 +26,6 @@ func main() {
// HandlersFabric returns the list of message handlers we want to be applied to each incoming message.
handlersFabric := func() []handlers.MessageProcessor {
return []handlers.MessageProcessor{
custom.NewInputEventBuilder(),
custom.NewPageEventBuilder(),
web.NewDeadClickDetector(),
&web.ClickRageDetector{},

1
backend/internal/config/storage/config.go
View file

@ -23,6 +23,7 @@ type Config struct {
MaxFileSize int64 `env:"MAX_FILE_SIZE,default=524288000"`
UseSort bool `env:"USE_SESSION_SORT,default=true"`
UseProfiler bool `env:"PROFILER_ENABLED,default=false"`
UseBrotli bool `env:"USE_BROTLI,default=false"`
}
func New() *Config {

2
backend/internal/db/datasaver/saver.go
View file

@ -65,8 +65,6 @@ func (s *saverImpl) handleMessage(msg Message) error {
return s.pg.InsertWebCustomEvent(m)
case *MouseClick:
return s.pg.InsertWebClickEvent(m)
case *InputEvent:
return s.pg.InsertWebInputEvent(m)
case *PageEvent:
return s.pg.InsertWebPageEvent(m)
case *NetworkRequest:

2
backend/internal/storage/storage.go
View file

@ -198,7 +198,7 @@ func (s *Storage) prepareSession(path string, tp FileType, task *Task) error {
func (s *Storage) packSession(task *Task, tp FileType) {
// If encryption key is empty, pack session using better algorithm
if task.key == "" {
if task.key == "" && s.cfg.UseBrotli {
s.packSessionBetter(task, tp)
return
}

9
backend/pkg/db/cache/messages-web.go vendored
View file

@ -172,15 +172,6 @@ func (c *PGCache) InsertWebClickEvent(e *MouseClick) error {
return c.Conn.InsertWebClickEvent(sessionID, session.ProjectID, e)
}
func (c *PGCache) InsertWebInputEvent(e *InputEvent) error {
sessionID := e.SessionID()
session, err := c.Cache.GetSession(sessionID)
if err != nil {
return err
}
return c.Conn.InsertWebInputEvent(sessionID, session.ProjectID, e)
}
func (c *PGCache) InsertWebInputDuration(e *InputChange) error {
sessionID := e.SessionID()
session, err := c.Cache.GetSession(sessionID)

12
backend/pkg/db/postgres/bulks.go
View file

@ -18,7 +18,6 @@ type BulkSet struct {
requests Bulk
customEvents Bulk
webPageEvents Bulk
webInputEvents Bulk
webInputDurations Bulk
webGraphQL Bulk
webErrors Bulk
@ -56,8 +55,6 @@ func (conn *BulkSet) Get(name string) Bulk {
return conn.customEvents
case "webPageEvents":
return conn.webPageEvents
case "webInputEvents":
return conn.webInputEvents
case "webInputDurations":
return conn.webInputDurations
case "webGraphQL":
@ -121,14 +118,6 @@ func (conn *BulkSet) initBulks() {
if err != nil {
log.Fatalf("can't create webPageEvents bulk: %s", err)
}
conn.webInputEvents, err = NewBulk(conn.c,
"events.inputs",
"(session_id, message_id, timestamp, label)",
"($%d, $%d, $%d, NULLIF(LEFT($%d, 2000),''))",
4, 200)
if err != nil {
log.Fatalf("can't create webPageEvents bulk: %s", err)
}
conn.webInputDurations, err = NewBulk(conn.c,
"events.inputs",
"(session_id, message_id, timestamp, label, hesitation, duration)",
@ -219,7 +208,6 @@ func (conn *BulkSet) Send() {
newTask.bulks = append(newTask.bulks, conn.requests)
newTask.bulks = append(newTask.bulks, conn.customEvents)
newTask.bulks = append(newTask.bulks, conn.webPageEvents)
newTask.bulks = append(newTask.bulks, conn.webInputEvents)
newTask.bulks = append(newTask.bulks, conn.webInputDurations)
newTask.bulks = append(newTask.bulks, conn.webGraphQL)
newTask.bulks = append(newTask.bulks, conn.webErrors)

12
backend/pkg/db/postgres/messages-web.go
View file

@ -76,18 +76,6 @@ func (conn *Conn) InsertWebClickEvent(sessionID uint64, projectID uint32, e *Mou
return nil
}
func (conn *Conn) InsertWebInputEvent(sessionID uint64, projectID uint32, e *InputEvent) error {
if e.Label == "" {
return nil
}
if err := conn.bulks.Get("webInputEvents").Append(sessionID, truncSqIdx(e.MessageID), e.Timestamp, e.Label); err != nil {
log.Printf("insert web input event err: %s", err)
}
conn.updateSessionEvents(sessionID, 1, 0)
conn.insertAutocompleteValue(sessionID, projectID, "INPUT", e.Label)
return nil
}
func (conn *Conn) InsertWebInputDuration(sessionID uint64, projectID uint32, e *InputChange) error {
if e.Label == "" {
return nil

77
backend/pkg/handlers/custom/inputEventBuilder.go
View file

@ -1,77 +0,0 @@
package custom
import (
. "openreplay/backend/pkg/messages"
)
const InputEventTimeout = 1 * 60 * 1000
type inputLabels map[uint64]string
type inputEventBuilder struct {
inputEvent *InputEvent
inputLabels inputLabels
inputID uint64
}
func NewInputEventBuilder() *inputEventBuilder {
ieBuilder := &inputEventBuilder{}
ieBuilder.clearLabels()
return ieBuilder
}
func (b *inputEventBuilder) clearLabels() {
b.inputLabels = make(inputLabels)
}
func (b *inputEventBuilder) Handle(message Message, timestamp uint64) Message {
var inputEvent Message = nil
switch msg := message.(type) {
case *SetInputTarget:
if b.inputID != msg.ID {
inputEvent = b.Build()
b.inputID = msg.ID
}
b.inputLabels[msg.ID] = msg.Label
return inputEvent
case *SetInputValue:
if b.inputID != msg.ID {
inputEvent = b.Build()
b.inputID = msg.ID
}
if b.inputEvent == nil {
b.inputEvent = &InputEvent{
MessageID: message.MsgID(),
Timestamp: timestamp,
Value: msg.Value,
ValueMasked: msg.Mask > 0,
}
} else {
b.inputEvent.Value = msg.Value
b.inputEvent.ValueMasked = msg.Mask > 0
}
return inputEvent
case *CreateDocument:
inputEvent = b.Build()
b.clearLabels()
return inputEvent
case *MouseClick:
return b.Build()
}
if b.inputEvent != nil && b.inputEvent.Timestamp+InputEventTimeout < timestamp {
return b.Build()
}
return nil
}
func (b *inputEventBuilder) Build() Message {
if b.inputEvent == nil {
return nil
}
inputEvent := b.inputEvent
inputEvent.Label = b.inputLabels[b.inputID] // might be empty string
b.inputEvent = nil
return inputEvent
}

3
ee/backend/internal/db/datasaver/methods.go
View file

@ -8,7 +8,6 @@ import (
"openreplay/backend/pkg/db/clickhouse"
"openreplay/backend/pkg/db/types"
"openreplay/backend/pkg/env"
. "openreplay/backend/pkg/messages"
"openreplay/backend/pkg/queue"
)
@ -52,8 +51,6 @@ func (s *saverImpl) handleExtraMessage(msg Message) error {
return s.ch.InsertWebPerformanceTrackAggr(session, m)
case *MouseClick:
return s.ch.InsertWebClickEvent(session, m)
case *InputEvent:
return s.ch.InsertWebInputEvent(session, m)
// Unique for Web
case *PageEvent:
return s.ch.InsertWebPageEvent(session, m)

22
ee/backend/pkg/db/clickhouse/connector.go
View file

@ -3,7 +3,6 @@ package clickhouse
import (
"errors"
"fmt"
"github.com/ClickHouse/clickhouse-go/v2"
"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
"log"
"openreplay/backend/pkg/db/types"
@ -25,7 +24,6 @@ type Connector interface {
InsertWebResourceEvent(session *types.Session, msg *messages.ResourceTiming) error
InsertWebPageEvent(session *types.Session, msg *messages.PageEvent) error
InsertWebClickEvent(session *types.Session, msg *messages.MouseClick) error
InsertWebInputEvent(session *types.Session, msg *messages.InputEvent) error
InsertWebErrorEvent(session *types.Session, msg *types.ErrorEvent) error
InsertWebPerformanceTrackAggr(session *types.Session, msg *messages.PerformanceTrackAggr) error
InsertAutocomplete(session *types.Session, msgType, msgValue string) error
@ -373,26 +371,6 @@ func (c *connectorImpl) InsertWebClickEvent(session *types.Session, msg *message
return nil
}
func (c *connectorImpl) InsertWebInputEvent(session *types.Session, msg *messages.InputEvent) error {
if msg.Label == "" {
return nil
}
if err := c.batches["inputs"].Append(
session.SessionID,
uint16(session.ProjectID),
msg.MessageID,
datetime(msg.Timestamp),
msg.Label,
"INPUT",
nil,
nil,
); err != nil {
c.checkError("inputs", err)
return fmt.Errorf("can't append to inputs batch: %s", err)
}
return nil
}
func (c *connectorImpl) InsertWebErrorEvent(session *types.Session, msg *types.ErrorEvent) error {
keys, values := make([]string, 0, len(msg.Tags)), make([]*string, 0, len(msg.Tags))
for k, v := range msg.Tags {