old heuristics

backend/go.mod

@@ -3,19 +3,20 @@ module openreplay/backend
 go 1.13
 
 require (
+	github.com/ClickHouse/clickhouse-go v1.4.3
+	github.com/aws/aws-sdk-go v1.35.23
 	github.com/btcsuite/btcutil v1.0.2
 	github.com/confluentinc/confluent-kafka-go v1.5.2 // indirect
 	github.com/go-redis/redis v6.15.9+incompatible
 	github.com/google/uuid v1.1.1
 	github.com/jackc/pgconn v1.6.0
-	github.com/jackc/pgx/v4 v4.6.0
 	github.com/jackc/pgerrcode v0.0.0-20201024163028-a0d42d470451
+	github.com/jackc/pgx/v4 v4.6.0
 	github.com/klauspost/compress v1.11.9
 	github.com/klauspost/pgzip v1.2.5
 	github.com/oschwald/maxminddb-golang v1.7.0
+	github.com/pkg/errors v0.9.1
 	github.com/tomasen/realip v0.0.0-20180522021738-f0c99a92ddce
 	github.com/ua-parser/uap-go v0.0.0-20200325213135-e1c09f13e2fe
 	gopkg.in/confluentinc/confluent-kafka-go.v1 v1.5.2
-	github.com/ClickHouse/clickhouse-go v1.4.3
-	github.com/aws/aws-sdk-go v1.35.23
 )

backend/go.sum

@@ -1,5 +1,6 @@
 github.com/ClickHouse/clickhouse-go v1.4.3/go.mod h1:EaI/sW7Azgz9UATzd5ZdZHRUhHgv5+JMS9NSr2smCJI=
 github.com/aead/siphash v1.0.1/go.mod h1:Nywa3cDsYNNK3gaciGTWPwHt0wlpNV15vwmswBAUSII=
+github.com/aws/aws-sdk-go v1.35.23 h1:SCP0d0XvyJTDmfnHEQPvBaYi3kea1VNUo7uQmkVgFts=
 github.com/aws/aws-sdk-go v1.35.23/go.mod h1:tlPOdRjfxPBpNIwqDj61rmsnA85v9jc0Ps9+muhnW+k=
 github.com/bkaradzic/go-lz4 v1.0.0/go.mod h1:0YdlkowM3VswSROI7qDxhRvJ3sLhlFrRRwjwegp5jy4=
 github.com/btcsuite/btcd v0.20.1-beta/go.mod h1:wVuoA8VJLEcwgqHBwHmzLRazpKxTv13Px/pDuV7OomQ=
@@ -24,6 +25,7 @@ github.com/davecgh/go-spew v0.0.0-20171005155431-ecdeabc65495/go.mod h1:J7Y8YcW2
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
+github.com/go-redis/redis v6.15.9+incompatible h1:K0pv1D7EQUjfyoMql+r/jZqCLizCGKFlFgcHWWmHQjg=
 github.com/go-redis/redis v6.15.9+incompatible/go.mod h1:NAIEuMOZ/fxfXJIrKDQDz8wamY7mA7PouImQ2Jvg6kA=
 github.com/go-sql-driver/mysql v1.4.0/go.mod h1:zAC/RDZ24gD3HViQzih4MyKcchzm+sOG5ZlKdlhCg5w=
 github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
@@ -73,6 +75,7 @@ github.com/jackc/puddle v0.0.0-20190413234325-e4ced69a3a2b/go.mod h1:m4B5Dj62Y0f
 github.com/jackc/puddle v0.0.0-20190608224051-11cab39313c9/go.mod h1:m4B5Dj62Y0fbyuIc15OsIqK0+JU8nkqQjsgx7dvjSWk=
 github.com/jackc/puddle v1.1.0/go.mod h1:m4B5Dj62Y0fbyuIc15OsIqK0+JU8nkqQjsgx7dvjSWk=
 github.com/jessevdk/go-flags v0.0.0-20141203071132-1679536dcc89/go.mod h1:4FA24M0QyGHXBuZZK/XkWh8h0e1EYbRYJSGM75WSRxI=
+github.com/jmespath/go-jmespath v0.4.0 h1:BEgLn5cpjn8UN1mAw4NjwDrS35OdebyEtFe+9YPoQUg=
 github.com/jmespath/go-jmespath v0.4.0/go.mod h1:T8mJZnbsbmF+m6zOOFylbeCJqk5+pHWvzYPziyZiYoo=
 github.com/jmespath/go-jmespath/internal/testify v1.5.1/go.mod h1:L3OGu8Wl2/fWfCI6z80xFu9LTZmf1ZRjMHUOPmWr69U=
 github.com/jmoiron/sqlx v1.2.0/go.mod h1:1FEQNm3xlJgrMD+FBdI9+xvCksHtbpVBBw5dYhBSsks=
@@ -103,6 +106,7 @@ github.com/oschwald/maxminddb-golang v1.7.0 h1:JmU4Q1WBv5Q+2KZy5xJI+98aUwTIrPPxZ
 github.com/oschwald/maxminddb-golang v1.7.0/go.mod h1:RXZtst0N6+FY/3qCNmZMBApR19cdQj43/NM9VkrNAis=
 github.com/pierrec/lz4 v2.0.5+incompatible/go.mod h1:pdkljMzZIN41W+lC3N2tnIh5sFi+IEE17M5jbnwPHcY=
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/rs/xid v1.2.1/go.mod h1:+uKXf+4Djp6Md1KODXJxgGQPKngRmWyn10oCKFzNHOQ=

backend/services/detectors/builder/builder.go

@@ -0,0 +1,255 @@
+package builder
+
+import (
+	"log"
+	"openreplay/backend/pkg/intervals"
+	. "openreplay/backend/pkg/messages"
+)
+
+type ReducedHandler struct {
+	handlers []handler
+}
+
+func (rh *ReducedHandler) HandleMessage(msg Message) []Message {
+	var resultMessages []Message
+	for _, h := range rh.handlers {
+		resultMessages = append(resultMessages, h.HandleMessage(msg)...)
+	}
+	return resultMessages
+}
+
+//-------------------------------------------------------------------------
+
+type CombinedHandler struct {
+	handlers []handler
+}
+
+func (ch *CombinedHandler) HandleMessage(msg Message) []Message {
+	resultMessages := []Message{msg}
+	for _, h := range ch.handlers {
+		var nextResultMessages []Message
+		for _, m := resultMessages {
+			nextResultMessages = append(nextResultMessages, h.HandleMessage(m)...)
+		}
+		resultMessages = nextResultMessages
+	}
+	return resultMessages
+}
+
+type builder struct {
+	readyMessages       []Message // a collection of built events
+	timestamp           uint64    // current timestamp
+	peBuilder           *pageEventBuilder
+	ptaBuilder          *performanceTrackAggrBuilder
+	ieBuilder           *inputEventBuilder
+	reBuilder           *resourceEventBuilder
+	ciDetector          *cpuIssueDetector
+	miDetector          *memoryIssueDetector
+	ddDetector          *domDropDetector
+	crDetector          *clickRageDetector
+	crshDetector        *crashDetector
+	dcDetector          *deadClickDetector
+	qrdetector          *quickReturnDetector
+	ridetector          *repetitiveInputDetector
+	exsdetector         *excessiveScrollingDetector
+	chdetector          *clickHesitationDetector
+	integrationsWaiting bool
+	sid                 uint64
+	sessionEventsCache  []*IssueEvent // a cache of selected ready messages used to detect events that depend on other events
+}
+
+func NewBuilder() *builder {
+	return &builder{
+		peBuilder:           &pageEventBuilder{},
+		ptaBuilder:          &performanceTrackAggrBuilder{},
+		ieBuilder:           NewInputEventBuilder(),
+		reBuilder:           &resourceEventBuilder{},
+		ciDetector:          &cpuIssueDetector{},
+		miDetector:          &memoryIssueDetector{},
+		ddDetector:          &domDropDetector{},
+		crDetector:          &clickRageDetector{},
+		crshDetector:        &crashDetector{},
+		dcDetector:          &deadClickDetector{},
+		qrdetector:          &quickReturnDetector{},
+		ridetector:          &repetitiveInputDetector{},
+		exsdetector:         &excessiveScrollingDetector{},
+		chdetector:          &clickHesitationDetector{},
+		integrationsWaiting: true,
+	}
+}
+
+// Additional methods for builder
+func (b *builder) appendReadyMessage(msg Message) { // interface is never nil even if it holds nil value
+	b.readyMessages = append(b.readyMessages, msg)
+}
+func (b *builder) appendSessionEvent(msg *IssueEvent) { // interface is never nil even if it holds nil value
+	b.sessionEventsCache = append(b.sessionEventsCache, msg)
+}
+
+func (b *builder) iterateReadyMessage(iter func(msg Message)) {
+	for _, readyMsg := range b.readyMessages {
+		iter(readyMsg)
+	}
+	b.readyMessages = nil
+}
+
+func (b *builder) buildSessionEnd() {
+	sessionEnd := &SessionEnd{
+		Timestamp: b.timestamp, // + delay?
+	}
+	b.appendReadyMessage(sessionEnd)
+}
+
+// ==================== DETECTORS ====================
+
+func (b *builder) detectCpuIssue(msg Message, messageID uint64) {
+	//	handle message and append to ready messages if it's fully composed
+	if rm := b.ciDetector.HandleMessage(msg, messageID, b.timestamp); rm != nil {
+		b.appendReadyMessage(rm)
+	}
+}
+
+func (b *builder) detectMemoryIssue(msg Message, messageID uint64) {
+	//	handle message and append to ready messages if it's fully composed
+	if rm := b.miDetector.HandleMessage(msg, messageID, b.timestamp); rm != nil {
+		b.appendReadyMessage(rm)
+	}
+}
+
+func (b *builder) detectDomDrop(msg Message) {
+	//	handle message and append to ready messages if it's fully composed
+	if dd := b.ddDetector.HandleMessage(msg, b.timestamp); dd != nil {
+		b.appendSessionEvent(dd) // not to ready messages, since we don't put it as anomaly
+	}
+}
+
+func (b *builder) detectDeadClick(msg Message, messageID uint64) {
+	if rm := b.dcDetector.HandleMessage(msg, messageID, b.timestamp); rm != nil {
+		b.appendReadyMessage(rm)
+	}
+}
+func (b *builder) detectQuickReturn(msg Message, messageID uint64) {
+	if rm := b.qrdetector.HandleMessage(msg, messageID, b.timestamp); rm != nil {
+		b.appendReadyMessage(rm)
+	}
+}
+func (b *builder) detectClickHesitation(msg Message, messageID uint64) {
+	if rm := b.chdetector.HandleMessage(msg, messageID, b.timestamp); rm != nil {
+		b.appendReadyMessage(rm)
+	}
+}
+func (b *builder) detectRepetitiveInput(msg Message, messageID uint64) {
+	if rm := b.ridetector.HandleMessage(msg, messageID, b.timestamp); rm != nil {
+		b.appendReadyMessage(rm)
+	}
+}
+func (b *builder) detectExcessiveScrolling(msg Message, messageID uint64) {
+	if rm := b.exsdetector.HandleMessage(msg, messageID, b.timestamp); rm != nil {
+		b.appendReadyMessage(rm)
+	}
+}
+
+// ==================== BUILDERS ====================
+
+func (b *builder) handlePerformanceTrackAggr(message Message, messageID uint64) {
+	if msg := b.ptaBuilder.HandleMessage(message, messageID, b.timestamp); msg != nil {
+		b.appendReadyMessage(msg)
+	}
+}
+func (b *builder) buildPerformanceTrackAggr() {
+	if msg := b.ptaBuilder.Build(); msg != nil {
+		b.appendReadyMessage(msg)
+	}
+}
+
+func (b *builder) handleInputEvent(message Message, messageID uint64) {
+	if msg := b.ieBuilder.HandleMessage(message, messageID, b.timestamp); msg != nil {
+		b.appendReadyMessage(msg)
+	}
+}
+func (b *builder) buildInputEvent() {
+	if msg := b.ieBuilder.Build(); msg != nil {
+		b.appendReadyMessage(msg)
+	}
+}
+
+func (b *builder) handlePageEvent(message Message, messageID uint64) {
+	if msg := b.peBuilder.HandleMessage(message, messageID, b.timestamp); msg != nil {
+		b.appendReadyMessage(msg)
+	}
+}
+func (b *builder) buildPageEvent() {
+	if msg := b.peBuilder.Build(); msg != nil {
+		b.appendReadyMessage(msg)
+	}
+}
+
+func (b *builder) handleResourceEvent(message Message, messageID uint64) {
+	if msg := b.reBuilder.HandleMessage(message, messageID); msg != nil {
+		b.appendReadyMessage(msg)
+	}
+}
+
+func (b *builder) handleMessage(message Message, messageID uint64) {
+
+	// update current timestamp
+	switch msg := message.(type) {
+	//case *SessionDisconnect:
+	//	b.timestamp = msg.Timestamp
+	case *SessionStart:
+		b.timestamp = msg.Timestamp
+	case *Timestamp:
+		b.timestamp = msg.Timestamp
+	}
+	// Start only from the first timestamp event.
+	if b.timestamp == 0 {
+		return
+	}
+
+	// Pass message to detector handlers
+	b.detectCpuIssue(message, messageID)
+	b.detectMemoryIssue(message, messageID)
+	b.detectDomDrop(message)
+	b.detectDeadClick(message, messageID)
+	b.detectQuickReturn(message, messageID)
+	b.detectClickHesitation(message, messageID)
+	b.detectRepetitiveInput(message, messageID)
+	b.detectExcessiveScrolling(message, messageID)
+
+	// Pass message to eventBuilders handler
+	b.handleInputEvent(message, messageID)
+	b.handlePageEvent(message, messageID)
+	b.handlePerformanceTrackAggr(message, messageID)
+	b.handleResourceEvent(message, messageID)
+
+	// Handle messages which translate to events without additional operations
+	b.HandleSimpleMessages(message, messageID)
+}
+
+func (b *builder) buildEvents(ts int64) {
+	if b.timestamp == 0 {
+		return // There was no timestamp events yet
+	}
+
+	if b.peBuilder.HasInstance() && int64(b.peBuilder.GetTimestamp())+intervals.EVENTS_PAGE_EVENT_TIMEOUT < ts {
+		b.buildPageEvent()
+	}
+	if b.ieBuilder.HasInstance() && int64(b.ieBuilder.GetTimestamp())+intervals.EVENTS_INPUT_EVENT_TIMEOUT < ts {
+		b.buildInputEvent()
+	}
+	if b.ptaBuilder.HasInstance() && int64(b.ptaBuilder.GetStartTimestamp())+intervals.EVENTS_PERFORMANCE_AGGREGATION_TIMEOUT < ts {
+		b.buildPerformanceTrackAggr()
+	}
+}
+
+func (b *builder) checkTimeouts(ts int64) bool {
+	if b.timestamp == 0 {
+		return false // There was no timestamp events yet
+	}
+	lastTsGap := ts - int64(b.timestamp)
+	log.Printf("checking timeouts for sess %v: %v now, %v sesstime; gap %v", b.sid, ts, b.timestamp, lastTsGap)
+	if lastTsGap > intervals.EVENTS_SESSION_END_TIMEOUT {
+		return true
+	}
+	return false
+}

backend/services/detectors/builder/builderMap.go

@@ -0,0 +1,51 @@
+package builder
+
+import (
+	. "openreplay/backend/pkg/messages"
+)
+
+type builderMap map[uint64]*builder
+
+func NewBuilderMap() builderMap {
+	return make(builderMap)
+}
+
+func (m builderMap) getBuilder(sessionID uint64) *builder {
+	b := m[sessionID]
+	if b == nil {
+		b = NewBuilder()
+		m[sessionID] = b
+		b.sid = sessionID
+	}
+	return b
+}
+
+func (m builderMap) HandleMessage(sessionID uint64, msg Message, messageID uint64) {
+	b := m.getBuilder(sessionID)
+	b.handleMessage(msg, messageID)
+}
+
+func (m builderMap) IterateSessionReadyMessages(sessionID uint64, operatingTs int64, iter func(msg Message)) {
+	b, ok := m[sessionID]
+	if !ok {
+		return
+	}
+	b.buildEvents(operatingTs)
+	sessionEnded := b.checkTimeouts(operatingTs)
+	b.iterateReadyMessage(iter)
+	if sessionEnded {
+		delete(m, sessionID)
+	}
+}
+
+func (m builderMap) IterateReadyMessages(operatingTs int64, iter func(sessionID uint64, msg Message)) {
+	for sessionID, b := range m {
+		sessionEnded := b.checkTimeouts(operatingTs)
+		b.iterateReadyMessage(func(msg Message) {
+			iter(sessionID, msg)
+		})
+		if sessionEnded {
+			delete(m, sessionID)
+		}
+	}
+}

backend/services/detectors/builder/clickHesitationDetector.go

@@ -0,0 +1,30 @@
+package builder
+
+import (
+	. "openreplay/backend/pkg/messages"
+)
+
+const HESITATION_THRESHOLD = 3000 // ms
+
+type clickHesitationDetector struct{}
+
+func (chd *clickHesitationDetector) HandleMouseClick(msg *MouseClick, messageID uint64, timestamp uint64) *IssueEvent {
+	if msg.HesitationTime > HESITATION_THRESHOLD {
+		return &IssueEvent{
+			Timestamp:     timestamp,
+			MessageID:     messageID,
+			Type:          "click_hesitation",
+			Context:       msg.Label,
+			ContextString: "Click hesitation above 3 seconds",
+		}
+	}
+	return nil
+}
+
+func (chd *clickHesitationDetector) HandleMessage(message Message, messageID uint64, timestamp uint64) *IssueEvent {
+	switch msg := message.(type) {
+	case *MouseClick:
+		return chd.HandleMouseClick(msg, messageID, timestamp)
+	}
+	return nil
+}

backend/services/detectors/builder/clikRageDetector.go

@@ -1,34 +1,42 @@
 package builder
 
 import (
-		"encoding/json"
+	"encoding/json"
 
 	. "openreplay/backend/pkg/messages"
 )
 
-
 const CLICK_TIME_DIFF = 200
 const MIN_CLICKS_IN_A_ROW = 3
 
 type clickRageDetector struct {
-	lastTimestamp  uint64
+	lastTimestamp        uint64
-	lastLabel string
+	lastLabel            string
 	firstInARawTimestamp uint64
 	firstInARawMessageId uint64
-	countsInARow int
+	countsInARow         int
 }
 
+func (crd *clickRageDetector) HandleMessage(message Message, messageID uint64, timestamp uint64) *IssueEvent {
+	switch msg := message.(type) {
+	case *MouseClick:
+		return crd.HandleMouseClick(msg, messageID, timestamp)
+	case *SessionEnd:
+		return crd.Build()
+	}
+	return nil
+}
 
 func (crd *clickRageDetector) Build() *IssueEvent {
 	var i *IssueEvent
 	if crd.countsInARow >= MIN_CLICKS_IN_A_ROW {
-		payload, _ := json.Marshal(struct{Count int }{crd.countsInARow,})
+		payload, _ := json.Marshal(struct{ Count int }{crd.countsInARow})
 		i = &IssueEvent{
-			Type: "click_rage",
+			Type:          "click_rage",
 			ContextString: crd.lastLabel,
-			Payload: string(payload), // TODO: json encoder
+			Payload:       string(payload), // TODO: json encoder
-			Timestamp: crd.firstInARawTimestamp,
+			Timestamp:     crd.firstInARawTimestamp,
-			MessageID: crd.firstInARawMessageId,
+			MessageID:     crd.firstInARawMessageId,
 		}
 	}
 	crd.lastTimestamp = 0
@@ -39,8 +47,8 @@ func (crd *clickRageDetector) Build() *IssueEvent {
 	return i
 }
 
-func (crd *clickRageDetector) HandleMouseClick(msg *MouseClick,  messageID uint64, timestamp uint64) *IssueEvent {
+func (crd *clickRageDetector) HandleMouseClick(msg *MouseClick, messageID uint64, timestamp uint64) *IssueEvent {
-	if crd.lastTimestamp + CLICK_TIME_DIFF < timestamp && crd.lastLabel == msg.Label {
+	if crd.lastTimestamp+CLICK_TIME_DIFF < timestamp && crd.lastLabel == msg.Label {
 		crd.lastTimestamp = timestamp
 		crd.countsInARow += 1
 		return nil
@@ -54,4 +62,4 @@ func (crd *clickRageDetector) HandleMouseClick(msg *MouseClick,  messageID uint6
 		crd.countsInARow = 1
 	}
 	return i
-}
+}

backend/services/detectors/builder/cpuIssueFinder.go

@@ -3,23 +3,34 @@ package builder
 import (
 	"encoding/json"
 
-	"openreplay/backend/pkg/messages/performance"
 	. "openreplay/backend/pkg/messages"
+	"openreplay/backend/pkg/messages/performance"
 )
 
-const CPU_THRESHOLD = 70  // % out of 100 
+const CPU_THRESHOLD = 70 // % out of 100
 const CPU_MIN_DURATION_TRIGGER = 6 * 1000
 
-
+type cpuIssueDetector struct {
-type cpuIssueFinder struct {
 	startTimestamp uint64
 	startMessageID uint64
-	lastTimestamp     uint64
+	lastTimestamp  uint64
-	maxRate uint64
+	maxRate        uint64
-	contextString string
+	contextString  string
 }
 
-func (f *cpuIssueFinder) Build() *IssueEvent {
+func (f *cpuIssueDetector) HandleMessage(message Message, messageID uint64, timestamp uint64) *IssueEvent {
+	switch msg := message.(type) {
+	case *SetPageLocation:
+		f.HandleSetPageLocation(msg)
+	case *PerformanceTrack:
+		return f.HandlePerformanceTrack(msg, messageID, timestamp);
+	case *SessionEnd:
+		return f.Build()
+	}
+	return nil
+}
+
+func (f *cpuIssueDetector) Build() *IssueEvent {
 	if f.startTimestamp == 0 {
 		return nil
 	}
@@ -35,26 +46,24 @@ func (f *cpuIssueFinder) Build() *IssueEvent {
 		return nil
 	}
 
-	payload, _ := json.Marshal(struct{
+	payload, _ := json.Marshal(struct {
 		Duration uint64
-		Rate uint64
+		Rate     uint64
-	}{duration,maxRate})
+	}{duration, maxRate})
 	return &IssueEvent{
-		Type: "cpu",
+		Type:          "cpu",
-		Timestamp: timestamp,
+		Timestamp:     timestamp,
-		MessageID: messageID,
+		MessageID:     messageID,
 		ContextString: f.contextString,
-		Payload: string(payload),
+		Payload:       string(payload),
 	}
 }
 
-func (f *cpuIssueFinder) HandleSetPageLocation(msg *SetPageLocation) {
+func (f *cpuIssueDetector) HandleSetPageLocation(msg *SetPageLocation) {
 	f.contextString = msg.URL
 }
 
-
+func (f *cpuIssueDetector) HandlePerformanceTrack(msg *PerformanceTrack, messageID uint64, timestamp uint64) *IssueEvent {
-
-func (f *cpuIssueFinder) HandlePerformanceTrack(msg *PerformanceTrack, messageID uint64, timestamp uint64) *IssueEvent {
 	dt := performance.TimeDiff(timestamp, f.lastTimestamp)
 	if dt == 0 {
 		return nil // TODO: handle error
@@ -83,4 +92,3 @@ func (f *cpuIssueFinder) HandlePerformanceTrack(msg *PerformanceTrack, messageID
 	return nil
 }
 
-

backend/services/detectors/builder/crashDetector.go

@@ -0,0 +1,71 @@
+package builder
+
+import (
+	. "openreplay/backend/pkg/messages"
+)
+
+const CPU_ISSUE_WINDOW = 3000
+const MEM_ISSUE_WINDOW = 4 * 1000
+const DOM_DROP_WINDOW = 200
+//const CLICK_RELATION_DISTANCE = 1200
+
+type crashDetector struct {
+	startTimestamp uint64
+}
+
+func (*crashDetector) buildCrashEvent(s []*IssueEvent) *IssueEvent{
+	var cpuIssues []*IssueEvent
+	var memIssues []*IssueEvent
+	var domDrops []*IssueEvent
+	for _, e := range s {
+		if e.Type == "cpu" {
+			cpuIssues = append(cpuIssues, e)
+		}
+		if e.Type == "memory" {
+			memIssues = append(memIssues, e)
+		}
+		if e.Type == "dom_drop" {
+			domDrops = append(domDrops, e)
+		}
+	}
+	var i, j, k int
+	for _, e := range s {
+		for i < len(cpuIssues) && cpuIssues[i].Timestamp+CPU_ISSUE_WINDOW < e.Timestamp {
+			i++
+		}
+		for j < len(memIssues) && memIssues[j].Timestamp+MEM_ISSUE_WINDOW < e.Timestamp {
+			j++
+		}
+		for k < len(domDrops) && domDrops[k].Timestamp+DOM_DROP_WINDOW < e.Timestamp { //Actually different type of issue
+			k++
+		}
+		if i == len(cpuIssues) && j == len(memIssues) && k == len(domDrops) {
+			break
+		}
+		if (i < len(cpuIssues) && cpuIssues[i].Timestamp < e.Timestamp+CPU_ISSUE_WINDOW) ||
+			(j < len(memIssues) && memIssues[j].Timestamp < e.Timestamp+MEM_ISSUE_WINDOW) ||
+			(k < len(domDrops) && domDrops[k].Timestamp < e.Timestamp+DOM_DROP_WINDOW) {
+			contextString := "UNKNOWN"
+			return &IssueEvent{
+				MessageID: e.MessageID,
+				Timestamp: e.Timestamp,
+				Type: "crash",
+				ContextString: contextString,
+				Context: "",
+			}
+		}
+	}
+	return nil
+}
+
+func (cr *crashDetector) HandleMessage(message Message, s []*IssueEvent) *IssueEvent{
+	// Only several message types can trigger crash
+	//	which is by our definition, a combination of DomDrop event, CPU and Memory issues
+	switch message.(type) {
+	case *SessionEnd:
+		return cr.buildCrashEvent(s)
+	case *PerformanceTrack:
+		return cr.buildCrashEvent(s)
+	}
+	return nil
+}

backend/services/detectors/builder/deadClickDetector.go

@@ -4,24 +4,22 @@ import (
 	. "openreplay/backend/pkg/messages"
 )
 
-
 const CLICK_RELATION_TIME = 1400
 
 type deadClickDetector struct {
-	lastMouseClick  *MouseClick
+	lastMouseClick *MouseClick
-	lastTimestamp uint64
+	lastTimestamp  uint64
-	lastMessageID   uint64
+	lastMessageID  uint64
 }
 
-
 func (d *deadClickDetector) HandleReaction(timestamp uint64) *IssueEvent {
 	var i *IssueEvent
-	if d.lastMouseClick != nil && d.lastTimestamp + CLICK_RELATION_TIME < timestamp {
+	if d.lastMouseClick != nil && d.lastTimestamp+CLICK_RELATION_TIME < timestamp {
 		i = &IssueEvent{
-			Type: "dead_click",
+			Type:          "dead_click",
 			ContextString: d.lastMouseClick.Label,
-			Timestamp: d.lastTimestamp,
+			Timestamp:     d.lastTimestamp,
-			MessageID: d.lastMessageID,
+			MessageID:     d.lastMessageID,
 		}
 	}
 	d.lastMouseClick = nil
@@ -38,8 +36,8 @@ func (d *deadClickDetector) HandleMessage(msg Message, messageID uint64, timesta
 		d.lastMouseClick = m
 		d.lastTimestamp = timestamp
 		d.lastMessageID = messageID
-	case *SetNodeAttribute, 
+	case *SetNodeAttribute,
-		*RemoveNodeAttribute, 
+		*RemoveNodeAttribute,
 		*CreateElementNode,
 		*CreateTextNode,
 		*MoveNode,
@@ -51,5 +49,3 @@ func (d *deadClickDetector) HandleMessage(msg Message, messageID uint64, timesta
 	}
 	return i
 }
-
-

backend/services/detectors/builder/domDropDetector.go

@@ -0,0 +1,52 @@
+package builder
+
+import (
+	. "openreplay/backend/pkg/messages"
+)
+
+type domDropDetector struct {
+	removedCount      int
+	lastDropTimestamp uint64
+}
+
+const DROP_WINDOW = 200  //ms
+const CRITICAL_COUNT = 1 // Our login page contains 20. But on crush it removes only roots (1-3 nodes).
+
+func (f *domDropDetector) HandleMessage(message Message, timestamp uint64) *IssueEvent {
+	switch message.(type) {
+	case *CreateElementNode,
+		*CreateTextNode:
+		f.HandleNodeCreation()
+	case *RemoveNode:
+		f.HandleNodeRemoval(timestamp)
+	case *CreateDocument:
+		return f.Build()
+	}
+	return nil
+}
+
+func (dd *domDropDetector) HandleNodeCreation() {
+	dd.removedCount = 0
+	dd.lastDropTimestamp = 0
+}
+
+func (dd *domDropDetector) HandleNodeRemoval(ts uint64) {
+	if dd.lastDropTimestamp+DROP_WINDOW > ts {
+		dd.removedCount += 1
+	} else {
+		dd.removedCount = 1
+	}
+	dd.lastDropTimestamp = ts
+}
+
+func (dd *domDropDetector) Build() *IssueEvent {
+	var domDrop *IssueEvent
+	if dd.removedCount >= CRITICAL_COUNT {
+		domDrop = &IssueEvent{
+			Type:      "dom_drop",
+			Timestamp: dd.lastDropTimestamp}
+	}
+	dd.removedCount = 0
+	dd.lastDropTimestamp = 0
+	return domDrop
+}

backend/services/detectors/builder/excessiveScrollingDetector.go

@@ -0,0 +1,52 @@
+package builder
+
+import (
+	. "openreplay/backend/pkg/messages"
+)
+
+const MAX_SCROLL_PER_PAGE = 20
+
+type excessiveScrollingDetector struct {
+	currentPage   string
+	scrollsNumber uint64
+}
+
+func (exs *excessiveScrollingDetector) HandleMouseClick() {
+	exs.scrollsNumber = 0
+}
+
+func (exs *excessiveScrollingDetector) HandleSetPageLocation(msg *SetPageLocation) {
+	if msg.Referrer != exs.currentPage {
+		exs.currentPage = msg.Referrer
+		exs.scrollsNumber = 0
+	}
+}
+
+func (exs *excessiveScrollingDetector) HandleScroll(msg *SetViewportScroll, messageID uint64, timestamp uint64) *IssueEvent {
+	if exs.scrollsNumber+1 >= MAX_SCROLL_PER_PAGE {
+		return &IssueEvent{
+			MessageID:     messageID,
+			Type:          "excessive_scrolling",
+			Timestamp:     timestamp,
+			ContextString: "Number of scrolling per page is above the threshold",
+			Context:       exs.currentPage,
+		}
+	} else {
+		exs.scrollsNumber += 1
+		return nil
+	}
+}
+
+func (exs *excessiveScrollingDetector) HandleMessage(message Message, messageID uint64, timestamp uint64) *IssueEvent {
+	switch msg := message.(type) {
+	case *MouseClick:
+		exs.HandleMouseClick()
+	case *SetPageLocation:
+		if msg.NavigationStart != 0 {
+			exs.HandleSetPageLocation(msg)
+		}
+	case *SetViewportScroll:
+		return exs.HandleScroll(msg, messageID, timestamp)
+	}
+	return nil
+}

backend/services/detectors/builder/inputEventBuilder.go

@@ -7,9 +7,9 @@ import (
 type inputLabels map[uint64]string
 
 type inputEventBuilder struct {
-	inputEvent         *InputEvent
+	inputEvent  *InputEvent
-	inputLabels        inputLabels
+	inputLabels inputLabels
-	inputID            uint64
+	inputID     uint64
 }
 
 func NewInputEventBuilder() *inputEventBuilder {
@@ -18,6 +18,25 @@ func NewInputEventBuilder() *inputEventBuilder {
 	return ieBuilder
 }
 
+func (ib *inputEventBuilder) HandleMessage(message Message, messageID uint64, timestamp uint64) *InputEvent {
+	switch msg := message.(type) {
+	//case *SessionDisconnect:
+	//	i := ib.Build()
+	//	ib.ClearLabels()
+	//	return i
+	case *SetPageLocation:
+		if msg.NavigationStart != 0 {
+			i := ib.Build()
+			ib.ClearLabels()
+			return i
+		}
+	case *SetInputTarget:
+		return ib.HandleSetInputTarget(msg)
+	case *SetInputValue:
+		return ib.HandleSetInputValue(msg, messageID, timestamp)
+	}
+	return nil
+}
 
 func (b *inputEventBuilder) ClearLabels() {
 	b.inputLabels = make(inputLabels)
@@ -57,7 +76,7 @@ func (b *inputEventBuilder) HasInstance() bool {
 	return b.inputEvent != nil
 }
 
-func (b * inputEventBuilder) GetTimestamp() uint64 {
+func (b *inputEventBuilder) GetTimestamp() uint64 {
 	if b.inputEvent == nil {
 		return 0
 	}
@@ -69,10 +88,10 @@ func (b *inputEventBuilder) Build() *InputEvent {
 		return nil
 	}
 	inputEvent := b.inputEvent
-	label, exists := b.inputLabels[b.inputID]
+	label := b.inputLabels[b.inputID]
-	if !exists {
+	// if !ok {
-		return nil
+	// 	return nil
-	}
+	// }
 	inputEvent.Label = label
 
 	b.inputEvent = nil

backend/services/detectors/builder/memoryIssueFinder.go

@@ -0,0 +1,82 @@
+package builder
+
+import (
+	"encoding/json"
+	"math"
+
+	. "openreplay/backend/pkg/messages"
+)
+
+const MIN_COUNT = 3
+const MEM_RATE_THRESHOLD = 300 // % to average 
+
+type memoryIssueDetector struct {
+	startMessageID uint64
+	startTimestamp uint64
+	rate           int
+	count          float64
+	sum            float64
+	contextString  string
+}
+
+func (f *memoryIssueDetector) HandleMessage(message Message, messageID uint64, timestamp uint64) *IssueEvent {
+	switch msg := message.(type) {
+	case *SetPageLocation:
+		f.HandleSetPageLocation(msg)
+	case *PerformanceTrack:
+		return f.HandlePerformanceTrack(msg, messageID, timestamp)
+	case *SessionEnd:
+		return f.Build()
+	}
+	return nil
+}
+
+func (f *memoryIssueDetector) Build() *IssueEvent {
+	if f.startTimestamp == 0 {
+		return nil
+	}
+	payload, _ := json.Marshal(struct{ Rate int }{f.rate - 100})
+	i := &IssueEvent{
+		Type:          "memory",
+		Timestamp:     f.startTimestamp,
+		MessageID:     f.startMessageID,
+		ContextString: f.contextString,
+		Payload:       string(payload),
+	}
+	f.startTimestamp = 0
+	f.startMessageID = 0
+	f.rate = 0
+	return i
+}
+
+func (f *memoryIssueDetector) HandleSetPageLocation(msg *SetPageLocation) {
+	f.contextString = msg.URL
+}
+
+func (f *memoryIssueDetector) HandlePerformanceTrack(msg *PerformanceTrack, messageID uint64, timestamp uint64) *IssueEvent {
+	if f.count < MIN_COUNT {
+		f.sum += float64(msg.UsedJSHeapSize)
+		f.count++
+		return nil
+	}
+
+	average := f.sum / f.count
+	rate := int(math.Round(float64(msg.UsedJSHeapSize) / average * 100))
+
+	f.sum += float64(msg.UsedJSHeapSize)
+	f.count++
+
+	if rate >= MEM_RATE_THRESHOLD {
+		if f.startTimestamp == 0 {
+			f.startTimestamp = timestamp
+			f.startMessageID = messageID
+		}
+		if f.rate < rate {
+			f.rate = rate
+		}
+	} else {
+		return f.Build()
+	}
+
+	return nil
+}

backend/services/detectors/builder/pageEventBuilder.go

@@ -5,8 +5,26 @@ import (
 )
 
 type pageEventBuilder struct {
-	pageEvent                   *PageEvent
+	pageEvent          *PageEvent
-	firstTimingHandled      bool
+	firstTimingHandled bool
+}
+
+func (pe *pageEventBuilder) HandleMessage(message Message, messageID uint64, timestamp uint64) *PageEvent {
+	switch msg := message.(type) {
+	case *SetPageLocation:
+		if msg.NavigationStart != 0 {
+			e := pe.Build()
+			pe.HandleSetPageLocation(msg, messageID, timestamp)
+			return e
+		}
+	case *PageLoadTiming:
+		return pe.HandlePageLoadTiming(msg)
+	case *PageRenderTiming:
+		return pe.HandlePageRenderTiming(msg)
+	//case *SessionDisconnect:
+	//	return pe.Build()
+	}
+	return nil
 }
 
 func (b *pageEventBuilder) buildIfTimingsComplete() *PageEvent {
@@ -28,7 +46,7 @@ func (b *pageEventBuilder) HandleSetPageLocation(msg *SetPageLocation, messageID
 	}
 }
 
-func (b * pageEventBuilder) HandlePageLoadTiming(msg *PageLoadTiming) *PageEvent {
+func (b *pageEventBuilder) HandlePageLoadTiming(msg *PageLoadTiming) *PageEvent {
 	if !b.HasInstance() {
 		return nil
 	}
@@ -62,7 +80,7 @@ func (b * pageEventBuilder) HandlePageLoadTiming(msg *PageLoadTiming) *PageEvent
 	return b.buildIfTimingsComplete()
 }
 
-func (b * pageEventBuilder) HandlePageRenderTiming(msg *PageRenderTiming) *PageEvent {
+func (b *pageEventBuilder) HandlePageRenderTiming(msg *PageRenderTiming) *PageEvent {
 	if !b.HasInstance() {
 		return nil
 	}
@@ -76,16 +94,16 @@ func (b *pageEventBuilder) HasInstance() bool {
 	return b.pageEvent != nil
 }
 
-func (b * pageEventBuilder) GetTimestamp() uint64 {
+func (b *pageEventBuilder) GetTimestamp() uint64 {
 	if b.pageEvent == nil {
 		return 0
 	}
-	return b.pageEvent.Timestamp;
+	return b.pageEvent.Timestamp
 }
 
-func (b * pageEventBuilder) Build() *PageEvent {
+func (b *pageEventBuilder) Build() *PageEvent {
 	pageEvent := b.pageEvent
 	b.pageEvent = nil
 	b.firstTimingHandled = false
 	return pageEvent
-}
+}

backend/services/detectors/builder/performanceTrackAggrBuilder.go

@@ -3,21 +3,29 @@ package builder
 import (
 	"math"
 
-	"openreplay/backend/pkg/messages/performance"
 	. "openreplay/backend/pkg/messages"
+	"openreplay/backend/pkg/messages/performance"
 )
 
-
 type performanceTrackAggrBuilder struct {
-	performanceTrackAggr   *PerformanceTrackAggr
+	performanceTrackAggr *PerformanceTrackAggr
-	lastTimestamp     uint64
+	lastTimestamp        uint64
-	count              float64
+	count                float64
-	sumFrameRate       float64
+	sumFrameRate         float64
-	sumTickRate        float64
+	sumTickRate          float64
-	sumTotalJSHeapSize float64
+	sumTotalJSHeapSize   float64
-	sumUsedJSHeapSize  float64
+	sumUsedJSHeapSize    float64
 }
 
+func (pta *performanceTrackAggrBuilder) HandleMessage(message Message, messageID uint64, timestamp uint64) *PerformanceTrackAggr {
+	switch msg := message.(type) {
+	//case *SessionDisconnect:
+	//	return pta.Build()
+	case *PerformanceTrack:
+		return pta.HandlePerformanceTrack(msg, timestamp)
+	}
+	return nil
+}
 
 func (b *performanceTrackAggrBuilder) start(timestamp uint64) {
 	b.performanceTrackAggr = &PerformanceTrackAggr{
@@ -39,7 +47,7 @@ func (b *performanceTrackAggrBuilder) HandlePerformanceTrack(msg *PerformanceTra
 	}
 
 	frameRate := performance.FrameRate(msg.Frames, dt)
-	tickRate :=  performance.TickRate(msg.Ticks, dt)
+	tickRate := performance.TickRate(msg.Ticks, dt)
 
 	fps := uint64(math.Round(frameRate))
 	cpu := performance.CPURateFromTickRate(tickRate)
@@ -84,7 +92,7 @@ func (b *performanceTrackAggrBuilder) GetStartTimestamp() uint64 {
 	if b.performanceTrackAggr == nil {
 		return 0
 	}
-	return b.performanceTrackAggr.TimestampStart;
+	return b.performanceTrackAggr.TimestampStart
 }
 
 func (b *performanceTrackAggrBuilder) Build() *PerformanceTrackAggr {
@@ -106,4 +114,3 @@ func (b *performanceTrackAggrBuilder) Build() *PerformanceTrackAggr {
 	b.lastTimestamp = 0
 	return performanceTrackAggr
 }
-

backend/services/detectors/builder/quickReturnDetector.go

@@ -0,0 +1,48 @@
+package builder
+
+import (
+	. "openreplay/backend/pkg/messages"
+)
+
+const QUICK_RETURN_THRESHOLD = 3 * 1000
+
+type quickReturnDetector struct {
+	timestamp   uint64
+	currentPage string
+	basePage    string
+}
+
+func (qrd *quickReturnDetector) HandleMessage(message Message, messageID uint64, timestamp uint64) *IssueEvent {
+	switch msg := message.(type) {
+	case *SetPageLocation:
+		if msg.NavigationStart != 0 {
+			return qrd.HandleSetPageLocation(msg, messageID, timestamp)
+		}
+	}
+	return nil
+}
+
+func (qrd *quickReturnDetector) HandleSetPageLocation(msg *SetPageLocation, messageID uint64, timestamp uint64) *IssueEvent {
+
+	if (timestamp-qrd.timestamp > 0) && (timestamp-qrd.timestamp < QUICK_RETURN_THRESHOLD) && (
+		msg.Referrer == qrd.basePage) {
+		i := &IssueEvent{
+			Type:          "quick_return",
+			Timestamp:     timestamp,
+			MessageID:     messageID,
+			ContextString: "Quick return from a page",
+			Context:       msg.Referrer + ", " + qrd.currentPage,
+			Payload:       ""}
+		qrd.basePage = qrd.currentPage
+		qrd.currentPage = msg.Referrer
+		qrd.timestamp = timestamp
+		return i
+	}
+
+	if msg.Referrer != qrd.currentPage {
+		qrd.timestamp = timestamp
+		qrd.basePage = qrd.currentPage
+		qrd.currentPage = msg.Referrer
+	}
+	return nil
+}

backend/services/detectors/builder/repetitiveInputDetector.go

@@ -0,0 +1,58 @@
+package builder
+
+import (
+	. "openreplay/backend/pkg/messages"
+)
+
+const MAX_HISTORY_OF_INPUTS = 10
+
+type repetitiveInputDetector struct {
+	lastInputs  []string
+	currentPage string
+}
+
+func (rid *repetitiveInputDetector) HandleMessage(message Message, messageID uint64, timestamp uint64) *IssueEvent {
+	switch msg := message.(type) {
+	case *InputEvent:
+		return rid.HandleInputEvent(msg, messageID, timestamp)
+	case *SetPageLocation:
+		if msg.NavigationStart != 0 {
+			rid.HandleSetPageLocation(msg)
+		}
+	}
+	return nil
+}
+
+func (rid *repetitiveInputDetector) HandleInputEvent(msg *InputEvent, messageID uint64, timestamp uint64) *IssueEvent {
+	// Check if the input is already in cache
+	for i, value := range rid.lastInputs {
+		if value == msg.Value {
+
+			// Update cache
+			rid.lastInputs = append(rid.lastInputs[:i], rid.lastInputs[i+1:]...)
+			rid.lastInputs = append(rid.lastInputs, msg.Value)
+
+			// Build an issue
+			return &IssueEvent{
+				MessageID:     messageID,
+				Timestamp:     timestamp,
+				Type:          "repetitive_input",
+				Payload:       "",
+				Context:       rid.currentPage,
+				ContextString: "The same input has recently been typed in"}
+		}
+	}
+
+	// Append a message value to cache
+	rid.lastInputs = append(rid.lastInputs, msg.Value)
+
+	// Discard last element from the queue
+	if len(rid.lastInputs) >= MAX_HISTORY_OF_INPUTS {
+		rid.lastInputs = rid.lastInputs[:len(rid.lastInputs)-1]
+	}
+	return nil
+}
+
+func (rid *repetitiveInputDetector) HandleSetPageLocation(msg *SetPageLocation) {
+	rid.currentPage = msg.Referrer
+}

backend/services/detectors/builder/resourceEventBuilder.go

@@ -0,0 +1,61 @@
+package builder
+
+import (
+	"net/url"
+	"strings"
+	. "openreplay/backend/pkg/messages"
+)
+
+type resourceEventBuilder struct {}
+
+func (reb *resourceEventBuilder) HandleMessage(message Message, messageID uint64) *ResourceEvent{
+	switch msg := message.(type) {
+	case *ResourceTiming:
+		tp := getResourceType(msg.Initiator, msg.URL)
+		success := msg.Duration != 0
+		return &ResourceEvent{
+			MessageID:       messageID,
+			Timestamp:       msg.Timestamp,
+			Duration:        msg.Duration,
+			TTFB:            msg.TTFB,
+			HeaderSize:      msg.HeaderSize,
+			EncodedBodySize: msg.EncodedBodySize,
+			DecodedBodySize: msg.DecodedBodySize,
+			URL:             msg.URL,
+			Type:            tp,
+			Success:         success,
+		}
+	}
+	return nil
+}
+
+func getURLExtention(URL string) string {
+	u, err := url.Parse(URL)
+	if err != nil {
+		return ""
+	}
+	i := strings.LastIndex(u.Path, ".")
+	return u.Path[i+1:]
+}
+
+func getResourceType(initiator string, URL string) string {
+	switch initiator {
+	case "xmlhttprequest", "fetch":
+		return "fetch"
+	case "img":
+		return "img"
+	default:
+		switch getURLExtention(URL) {
+		case "css":
+			return "stylesheet"
+		case "js":
+			return "script"
+		case "png", "gif", "jpg", "jpeg", "svg":
+			return "img"
+		case "mp4", "mkv", "ogg", "webm", "avi", "mp3":
+			return "media"
+		default:
+			return "other"
+		}
+	}
+}

backend/services/detectors/builder/simpleEventsBuilder.go

@@ -0,0 +1,89 @@
+package builder
+
+import (
+	. "openreplay/backend/pkg/messages"
+)
+
+func (b *builder) HandleSimpleMessages(message Message, messageID uint64) {
+	switch msg := message.(type) {
+	case *MouseClick:
+		if msg.Label != "" {
+			b.appendReadyMessage(&ClickEvent{
+				MessageID:      messageID,
+				Label:          msg.Label,
+				HesitationTime: msg.HesitationTime,
+				Timestamp:      b.timestamp,
+			})
+		}
+	case *RawErrorEvent:
+		b.appendReadyMessage(&ErrorEvent{
+			MessageID: messageID,
+			Timestamp: msg.Timestamp,
+			Source:    msg.Source,
+			Name:      msg.Name,
+			Message:   msg.Message,
+			Payload:   msg.Payload,
+		})
+	case *JSException:
+		b.appendReadyMessage(&ErrorEvent{
+			MessageID: messageID,
+			Timestamp: b.timestamp,
+			Source:    "js_exception",
+			Name:      msg.Name,
+			Message:   msg.Message,
+			Payload:   msg.Payload,
+		})
+	case *ResourceTiming:
+		success := msg.Duration != 0
+		tp := getResourceType(msg.Initiator, msg.URL)
+		if !success && tp == "fetch" {
+			b.appendReadyMessage(&IssueEvent{
+				Type:          "bad_request",
+				MessageID:     messageID,
+				Timestamp:     msg.Timestamp,
+				ContextString: msg.URL,
+				Context:       "",
+				Payload:       "",
+			})
+		}
+	case *RawCustomEvent:
+		b.appendReadyMessage(&CustomEvent{
+			MessageID: messageID,
+			Timestamp: b.timestamp,
+			Name:      msg.Name,
+			Payload:   msg.Payload,
+		})
+	case *CustomIssue:
+		b.appendReadyMessage(&IssueEvent{
+			Type:          "custom",
+			Timestamp:     b.timestamp,
+			MessageID:     messageID,
+			ContextString: msg.Name,
+			Payload:       msg.Payload,
+		})
+	case *Fetch:
+		b.appendReadyMessage(&ResourceEvent{
+			MessageID: messageID,
+			Timestamp: msg.Timestamp,
+			Duration:  msg.Duration,
+			URL:       msg.URL,
+			Type:      "fetch",
+			Success:   msg.Status < 300,
+			Method:    msg.Method,
+			Status:    msg.Status,
+		})
+	case *StateAction:
+		b.appendReadyMessage(&StateActionEvent{
+			MessageID: messageID,
+			Timestamp: b.timestamp,
+			Type:      msg.Type,
+		})
+	case *GraphQL:
+		b.appendReadyMessage(&GraphQLEvent{
+			MessageID: messageID,
+			Timestamp: b.timestamp,
+			Name:      msg.OperationName,
+		})
+	}
+
+}

backend/services/detectors/main.go

@@ -0,0 +1,72 @@
+package main
+
+import (
+	"log"
+	"time"
+
+	"os"
+	"os/signal"
+	"syscall"
+
+	"openreplay/backend/pkg/intervals"
+	"openreplay/backend/pkg/env"
+	"openreplay/backend/pkg/messages"
+	"openreplay/backend/pkg/queue"
+	"openreplay/backend/pkg/queue/types"
+	"openreplay/backend/services/detectors/builder"
+)
+
+func main() {
+	log.SetFlags(log.LstdFlags | log.LUTC | log.Llongfile)
+
+	GROUP_EVENTS := env.String("GROUP_DETECTOR") // env.String("GROUP_EVENTS")
+	TOPIC_RAW := env.String("TOPIC_RAW")
+	TOPIC_TRIGGER := env.String("TOPIC_TRIGGER")
+
+	builderMap := builder.NewBuilderMap()
+	var lastTs int64 = 0
+
+	producer := queue.NewProducer()
+	consumer := queue.NewMessageConsumer(
+		GROUP_EVENTS,
+		[]string{TOPIC_RAW},
+		func(sessionID uint64, msg messages.Message, meta *types.Meta) {
+			lastTs = meta.Timestamp
+			builderMap.HandleMessage(sessionID, msg, msg.Meta().Index)
+			builderMap.IterateSessionReadyMessages(sessionID, lastTs, func(readyMsg messages.Message) {
+				producer.Produce(TOPIC_TRIGGER, sessionID, messages.Encode(readyMsg))
+			})
+		},
+	)
+	consumer.DisableAutoCommit()
+
+	tick := time.Tick(intervals.EVENTS_COMMIT_INTERVAL * time.Millisecond)
+
+	sigchan := make(chan os.Signal, 1)
+	signal.Notify(sigchan, syscall.SIGINT, syscall.SIGTERM)
+
+	for {
+		select {
+		case sig := <-sigchan:
+			log.Printf("Caught signal %v: terminating\n", sig)
+			producer.Close(2000)
+			consumer.CommitBack(intervals.EVENTS_BACK_COMMIT_GAP)
+			consumer.Close()
+			os.Exit(0)
+		case <-tick:
+			builderMap.IterateReadyMessages(time.Now().UnixNano()/1e6, func(sessionID uint64, readyMsg messages.Message) {
+				if _, ok := readyMsg.(*messages.SessionEnd); ok {
+					log.Printf("ENDSOME %v", sessionID)
+				}
+				producer.Produce(TOPIC_TRIGGER, sessionID, messages.Encode(readyMsg))
+			})
+			// TODO: why exactly do we need Flush here and not in any other place?
+			producer.Flush(2000)
+			consumer.CommitBack(intervals.EVENTS_BACK_COMMIT_GAP)
+		default:
+			if err := consumer.ConsumeNext(); err != nil {
+				log.Printf("Error on consuming: %v", err)
+			}
+		}
+	}
+}

backend/services/ender/builder/builder.go

@@ -1,71 +1,21 @@
 package builder
 
 import (
-	"net/url"
+	"log"
-	"strings"
 
 	"openreplay/backend/pkg/intervals"
 	. "openreplay/backend/pkg/messages"
 )
 
-func getURLExtention(URL string) string {
-	u, err := url.Parse(URL)
-	if err != nil {
-		return ""
-	}
-	i := strings.LastIndex(u.Path, ".")
-	return u.Path[i+1:]
-}
-
-func getResourceType(initiator string, URL string) string {
-	switch initiator {
-	case "xmlhttprequest", "fetch":
-		return "fetch"
-	case "img":
-		return "img"
-	default:
-		switch getURLExtention(URL) {
-		case "css":
-			return "stylesheet"
-		case "js":
-			return "script"
-		case "png", "gif", "jpg", "jpeg", "svg":
-			return "img"
-		case "mp4", "mkv", "ogg", "webm", "avi", "mp3":
-			return "media"
-		default:
-			return "other"
-		}
-	}
-}
-
 type builder struct {
-	readyMsgs                         []Message
+	readyMsgs           []Message
-	timestamp                         uint64
+	timestamp           uint64
-	peBuilder 												*pageEventBuilder
+	integrationsWaiting bool
-	ptaBuilder       									*performanceTrackAggrBuilder
+	sid                 uint64
-	ieBuilder													*inputEventBuilder
-	ciFinder													*cpuIssueFinder
-	miFinder													*memoryIssueFinder
-	ddDetector												*domDropDetector
-	crDetector												*clickRageDetector
-	dcDetector												*deadClickDetector
-	integrationsWaiting 	   					bool
-
-
-	sid uint64
 }
 
 func NewBuilder() *builder {
 	return &builder{
-		peBuilder: &pageEventBuilder{},
-		ptaBuilder: &performanceTrackAggrBuilder{},
-		ieBuilder: NewInputEventBuilder(),
-		ciFinder: &cpuIssueFinder{},
-		miFinder: &memoryIssueFinder{},
-		ddDetector: &domDropDetector{},
-		crDetector: &clickRageDetector{},
-		dcDetector: &deadClickDetector{},
 		integrationsWaiting: true,
 	}
 }
@@ -82,234 +32,35 @@ func (b *builder) iterateReadyMessage(iter func(msg Message)) {
 }
 
 func (b *builder) buildSessionEnd() {
-	if b.timestamp == 0 {
-		return
-	}
 	sessionEnd := &SessionEnd{
 		Timestamp: b.timestamp, // + delay?
 	}
 	b.appendReadyMessage(sessionEnd)
 }
 
-func (b *builder) buildPageEvent() {
-	if msg := b.peBuilder.Build(); msg != nil {
-		b.appendReadyMessage(msg)
-	}
-}
-func (b *builder) buildPerformanceTrackAggr() {
-	if msg := b.ptaBuilder.Build(); msg != nil {
-		b.appendReadyMessage(msg)
-	}
-}
-func (b *builder) buildInputEvent() {
-	if msg := b.ieBuilder.Build(); msg != nil {
-		b.appendReadyMessage(msg)
-	}
-}
-
 func (b *builder) handleMessage(message Message, messageID uint64) {
-	timestamp := uint64(message.Meta().Timestamp)
-	if b.timestamp <= timestamp { // unnecessary. TODO: test and remove
-		b.timestamp = timestamp
-	}
-	// Before  the first timestamp.
 	switch msg := message.(type) {
-	case *SessionStart,
+	case *SessionDisconnect:
-			*Metadata,
+		b.timestamp = msg.Timestamp
-			*UserID,
+	case *SessionStart:
-	  	*UserAnonymousID:
+		b.timestamp = msg.Timestamp
-		b.appendReadyMessage(msg)
+	case *Timestamp:
-	case *RawErrorEvent:
+		b.timestamp = msg.Timestamp
-		b.appendReadyMessage(&ErrorEvent{
-			MessageID: messageID,
-			Timestamp: msg.Timestamp,
-			Source:    msg.Source,
-			Name:      msg.Name,
-			Message:   msg.Message,
-			Payload:   msg.Payload,
-		})
 	}
+	// Start from the first timestamp event.
 	if b.timestamp == 0 {
 		return
 	}
-	switch msg := message.(type) {
-	case *SetPageLocation:
-		if msg.NavigationStart == 0 {
-			b.appendReadyMessage(&PageEvent{
-				URL:       msg.URL,
-				Referrer:  msg.Referrer,
-				Loaded:    false,
-				MessageID: messageID,
-				Timestamp: b.timestamp,
-			})
-		} else {
-			b.buildPageEvent()
-			b.buildInputEvent()
-			b.ieBuilder.ClearLabels()
-			b.peBuilder.HandleSetPageLocation(msg, messageID, b.timestamp)
-			b.miFinder.HandleSetPageLocation(msg)
-			b.ciFinder.HandleSetPageLocation(msg)
-		}
-	case *PageLoadTiming:
-		if rm := b.peBuilder.HandlePageLoadTiming(msg); rm != nil {
-			b.appendReadyMessage(rm)
-		}
-	case *PageRenderTiming:
-		if rm := b.peBuilder.HandlePageRenderTiming(msg); rm != nil {
-			b.appendReadyMessage(rm)
-		}
-	case *PerformanceTrack:
-		if rm := b.ptaBuilder.HandlePerformanceTrack(msg, b.timestamp); rm != nil {
-			b.appendReadyMessage(rm)
-		}
-		if rm := b.ciFinder.HandlePerformanceTrack(msg, messageID, b.timestamp); rm != nil {
-			b.appendReadyMessage(rm)
-		}
-		if rm := b.miFinder.HandlePerformanceTrack(msg, messageID, b.timestamp); rm != nil {
-			b.appendReadyMessage(rm)
-		}
-	case *SetInputTarget:
-		if rm := b.ieBuilder.HandleSetInputTarget(msg); rm != nil {
-			b.appendReadyMessage(rm)
-		}
-	case *SetInputValue:
-		if rm := b.ieBuilder.HandleSetInputValue(msg, messageID, b.timestamp); rm != nil {
-			b.appendReadyMessage(rm)
-		}
-	case *MouseClick:
-		b.buildInputEvent()
-		if rm := b.crDetector.HandleMouseClick(msg, messageID, b.timestamp); rm != nil {
-			b.appendReadyMessage(rm)
-		}
-		if msg.Label != "" {
-			b.appendReadyMessage(&ClickEvent{
-				MessageID:      messageID,
-				Label:          msg.Label,
-				HesitationTime: msg.HesitationTime,
-				Timestamp:      b.timestamp,
-			})
-		}
-	case *JSException:
-		b.appendReadyMessage(&ErrorEvent{
-			MessageID: messageID,
-			Timestamp: b.timestamp,
-			Source:    "js_exception",
-			Name:      msg.Name,
-			Message:   msg.Message,
-			Payload:   msg.Payload,
-		})
-	case *ResourceTiming:
-		tp := getResourceType(msg.Initiator, msg.URL)
-		success := msg.Duration != 0
-		b.appendReadyMessage(&ResourceEvent{
-			MessageID:       messageID,
-			Timestamp:       msg.Timestamp,
-			Duration:        msg.Duration,
-			TTFB:            msg.TTFB,
-			HeaderSize:      msg.HeaderSize,
-			EncodedBodySize: msg.EncodedBodySize,
-			DecodedBodySize: msg.DecodedBodySize,
-			URL:             msg.URL,
-			Type:            tp,
-			Success:         success,
-		})
-	  if !success && tp == "fetch" {
-		  b.appendReadyMessage(&IssueEvent{
-		  	Type: "bad_request",
-				MessageID: messageID,
-				Timestamp: msg.Timestamp,
-				ContextString: msg.URL,
-				Context: "",
-				Payload: "",
-			})
-		}
-	case *RawCustomEvent:
-		b.appendReadyMessage(&CustomEvent{
-			MessageID: messageID,
-			Timestamp: b.timestamp,
-			Name:      msg.Name,
-			Payload:   msg.Payload,
-		})
-	case *CustomIssue:
-		b.appendReadyMessage(&IssueEvent{
-			Type: "custom",
-			Timestamp: b.timestamp,
-			MessageID: messageID,
-			ContextString: msg.Name,
-			Payload: msg.Payload,
-		})
-	case *Fetch:
-		b.appendReadyMessage(&ResourceEvent{
-			MessageID: messageID,
-			Timestamp: msg.Timestamp,
-			Duration:  msg.Duration,
-			URL:       msg.URL,
-			Type:      "fetch",
-			Success:   msg.Status < 300,
-			Method:    msg.Method,
-			Status:    msg.Status,
-		})
-	case *StateAction:
-		b.appendReadyMessage(&StateActionEvent{
-			MessageID: messageID,
-			Timestamp: b.timestamp,
-			Type:      msg.Type,
-		})
-	case *GraphQL:
-		b.appendReadyMessage(&GraphQLEvent{
-			MessageID: messageID,
-			Timestamp: b.timestamp,
-			Name:      msg.OperationName,
-		})
-	case *CreateElementNode,
-		*CreateTextNode:
-		b.ddDetector.HandleNodeCreation()
-	case *RemoveNode:
-		b.ddDetector.HandleNodeRemoval(b.timestamp)
-	case *CreateDocument:
-		if rm := b.ddDetector.Build(); rm != nil {
-			b.appendReadyMessage(rm)
-		}
-	}
-	if rm := b.dcDetector.HandleMessage(message, messageID, b.timestamp); rm != nil {
-		b.appendReadyMessage(rm)
-	}
 }
 
-
 func (b *builder) checkTimeouts(ts int64) bool {
-	if b.timestamp == 0 { 
+	if b.timestamp == 0 {
 		return false // There was no timestamp events yet
-	} 
-
-	if b.peBuilder.HasInstance() && int64(b.peBuilder.GetTimestamp())+intervals.EVENTS_PAGE_EVENT_TIMEOUT < ts {
-		b.buildPageEvent()
-	}
-	if b.ieBuilder.HasInstance() && int64(b.ieBuilder.GetTimestamp())+intervals.EVENTS_INPUT_EVENT_TIMEOUT < ts {
-		b.buildInputEvent()
-	}
-	if b.ptaBuilder.HasInstance() && int64(b.ptaBuilder.GetStartTimestamp())+intervals.EVENTS_PERFORMANCE_AGGREGATION_TIMEOUT < ts {
-		b.buildPerformanceTrackAggr()
 	}
 
 	lastTsGap := ts - int64(b.timestamp)
-	//log.Printf("checking timeouts for sess %v: %v now, %v sesstime; gap %v",b.sid,  ts, b.timestamp, lastTsGap)
+	log.Printf("checking timeouts for sess %v: %v now, %v sesstime; gap %v", b.sid, ts, b.timestamp, lastTsGap)
 	if lastTsGap > intervals.EVENTS_SESSION_END_TIMEOUT {
-		if rm := b.ddDetector.Build(); rm != nil {
-			b.appendReadyMessage(rm)
-		}
-		if rm := b.ciFinder.Build(); rm != nil {
-			b.appendReadyMessage(rm)
-		}
-		if rm := b.miFinder.Build(); rm != nil {
-			b.appendReadyMessage(rm)
-		}
-		if rm := b.crDetector.Build(); rm != nil {
-			b.appendReadyMessage(rm)
-		}
-		if rm := b.dcDetector.HandleReaction(b.timestamp); rm != nil {
-			b.appendReadyMessage(rm)
-		}
 		b.buildSessionEnd()
 		return true
 	}

backend/services/ender/builder/builderMap.go

@@ -11,7 +11,7 @@ func NewBuilderMap() builderMap {
 	return make(builderMap)
 }
 
-func (m builderMap) GetBuilder(sessionID uint64) *builder {
+func (m builderMap) getBuilder(sessionID uint64) *builder {
 	b := m[sessionID]
 	if b == nil {
 		b = NewBuilder()
@@ -23,7 +23,7 @@ func (m builderMap) GetBuilder(sessionID uint64) *builder {
 }
 
 func (m builderMap) HandleMessage(sessionID uint64, msg Message, messageID uint64) {
-	b := m.GetBuilder(sessionID)
+	b := m.getBuilder(sessionID)
 	b.handleMessage(msg, messageID)
 }
 

backend/services/ender/builder/domDropDetector.go

@@ -1,42 +0,0 @@
-package builder
-
-import (
-	. "openreplay/backend/pkg/messages"
-)
-
-
-type domDropDetector struct {
-	removedCount  int
-	lastDropTimestamp uint64
-}
-
-const DROP_WINDOW = 200 //ms
-const CRITICAL_COUNT = 1  // Our login page contains 20. But on crush it removes only roots (1-3 nodes).
-
-func (dd *domDropDetector) HandleNodeCreation() {
-	dd.removedCount = 0
-	dd.lastDropTimestamp = 0
-}
-
-func (dd *domDropDetector) HandleNodeRemoval(ts uint64) {
-	if dd.lastDropTimestamp + DROP_WINDOW > ts {
-		dd.removedCount += 1
-	} else {
-		dd.removedCount = 1
-	}
-	dd.lastDropTimestamp = ts
-}
-
-
-func (dd *domDropDetector) Build() *DOMDrop {
-	var domDrop *DOMDrop
-	if dd.removedCount >= CRITICAL_COUNT {
-		domDrop = &DOMDrop{
-			Timestamp: dd.lastDropTimestamp,
-		}
-	}
-	dd.removedCount = 0
-	dd.lastDropTimestamp = 0
-	return domDrop
-}
-

backend/services/ender/builder/memoryIssueFinder.go

@@ -1,72 +0,0 @@
-package builder
-
-import (
-	"math"
-	"encoding/json"
-	
-	. "openreplay/backend/pkg/messages"
-)
-
-const MIN_COUNT = 3
-const MEM_RATE_THRESHOLD = 300 // % to average 
-
-type memoryIssueFinder struct {
-	startMessageID uint64
-	startTimestamp uint64
-	rate           int
-	count          float64
-	sum  		       float64
-	contextString  string
-}
-
-func (f *memoryIssueFinder) Build() *IssueEvent {
-	if f.startTimestamp == 0 {
-		return nil
-	}
-	payload, _ := json.Marshal(struct{Rate int }{f.rate - 100,})
-	i := &IssueEvent{
-		Type: "memory",
-		Timestamp: f.startTimestamp,
-		MessageID: f.startMessageID,
-		ContextString: f.contextString,
-		Payload: string(payload),
-	}
-	f.startTimestamp = 0
-	f.startMessageID = 0
-	f.rate = 0
-	return i
-}
-
-func (f *memoryIssueFinder) HandleSetPageLocation(msg *SetPageLocation) {
-	f.contextString = msg.URL
-}
-
-func (f *memoryIssueFinder) HandlePerformanceTrack(msg *PerformanceTrack, messageID uint64, timestamp uint64) *IssueEvent {
-	if f.count < MIN_COUNT {
-		f.sum += float64(msg.UsedJSHeapSize)
-		f.count++
-		return nil
-	}
-
-	average := f.sum/f.count
-	rate := int(math.Round(float64(msg.UsedJSHeapSize)/average * 100))
-
-	f.sum += float64(msg.UsedJSHeapSize)
-	f.count++
-
-	if rate >= MEM_RATE_THRESHOLD {
-		if f.startTimestamp == 0 {
-			f.startTimestamp = timestamp
-			f.startMessageID = messageID
-		}
-		if f.rate < rate {
-			f.rate = rate
-		}
-	} else {
-		return f.Build()
-	}
-
-	return nil
-}
-
-

backend/services/ender/main.go

@@ -20,7 +20,8 @@ import (
 func main() {
 	log.SetFlags(log.LstdFlags | log.LUTC | log.Llongfile)
 
-	GROUP_EVENTS := env.String("GROUP_ENDER")
+	GROUP_EVENTS := env.String("GROUP_ENDER") // env.String("GROUP_EVENTS")
+	TOPIC_RAW := env.String("TOPIC_RAW")
 	TOPIC_TRIGGER := env.String("TOPIC_TRIGGER")
 
 	builderMap := builder.NewBuilderMap()
@@ -29,15 +30,13 @@ func main() {
 	producer := queue.NewProducer()
 	consumer := queue.NewMessageConsumer(
 		GROUP_EVENTS, 
-		[]string{ 
+		[]string{ TOPIC_RAW }, 
-			env.String("TOPIC_RAW"),
-		}, 
 		func(sessionID uint64, msg messages.Message, meta *types.Meta) {
 			lastTs = meta.Timestamp
-			builderMap.HandleMessage(sessionID, msg, msg.Meta().Index)
+			builderMap.HandleMessage(sessionID, msg, meta.ID)
-			// builderMap.IterateSessionReadyMessages(sessionID, lastTs, func(readyMsg messages.Message) {
+			builderMap.IterateSessionReadyMessages(sessionID, lastTs, func(readyMsg messages.Message) {
-			// 	producer.Produce(TOPIC_TRIGGER, sessionID, messages.Encode(readyMsg))
+				producer.Produce(TOPIC_TRIGGER, sessionID, messages.Encode(readyMsg))
-			// })
+			})
 		},
 	)
 	consumer.DisableAutoCommit()
@@ -57,6 +56,9 @@ func main() {
 			os.Exit(0)
 		case <- tick:
 			builderMap.IterateReadyMessages(time.Now().UnixNano()/1e6, func(sessionID uint64, readyMsg messages.Message) {
+				if _, ok := readyMsg.(*messages.SessionEnd); ok {
+					log.Printf("ENDSOME %v", sessionID)
+				}
 				producer.Produce(TOPIC_TRIGGER, sessionID, messages.Encode(readyMsg))
 			})
 			// TODO: why exactly do we need Flush here and not in any other place?
@@ -64,7 +66,7 @@ func main() {
 			consumer.CommitBack(intervals.EVENTS_BACK_COMMIT_GAP)
 		default:
 			if err := consumer.ConsumeNext(); err != nil {
-				log.Fatalf("Error on consuming: %v", err)
+				log.Printf("Error on consuming: %v", err)
 			}
 		}
 	}