feature: Publish Minimal nats event early and several fixes

Author: Gleez Technologies

Dockerfile

@@ -1,59 +0,0 @@
-############################
-# STEP 1 build executable binary
-############################
-FROM golang:1.14 AS builder
-
-ADD . /go/src/github.com/sandeepone/mysql-manticore
-
-RUN cd /go/src/github.com/sandeepone/mysql-manticore \
- && COMMIT_SHA=$(git rev-parse --short HEAD) \
- && CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -ldflags "-s -w  \
-    -X main.version=0.8.0 \
-    -X main.revision=${COMMIT_SHA}" \
-    -a -tags netgo -installsuffix netgo -o mysql-manticore cmd/mysql-manticore/main.go
-
-############################
-# STEP 2 build a certs image
-############################
-
-# Alpine certs
-FROM alpine:3.11 as alpine
-
-RUN apk update && apk add --no-cache ca-certificates tzdata && update-ca-certificates
-
-# Create appuser
-ENV USER=appuser
-ENV UID=10001
-
-# See https://stackoverflow.com/a/55757473/12429735
-RUN adduser \
-    --disabled-password \
-    --gecos "" \
-    --home "/nonexistent" \
-    --shell "/sbin/nologin" \
-    --no-create-home \
-    --uid "${UID}" \
-    "${USER}"
-
-############################
-# STEP 3 build a release image
-############################
-FROM scratch
-MAINTAINER Sandeep Sangamreddi <[email protected]>
-
-# Import from builder.
-COPY --from=alpine /usr/share/zoneinfo /usr/share/zoneinfo
-COPY --from=alpine /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/
-COPY --from=alpine /etc/passwd /etc/passwd
-COPY --from=alpine /etc/group /etc/group
-
-COPY --from=builder /go/src/github.com/sandeepone/mysql-manticore/mysql-manticore /usr/bin/
-COPY etc/river.toml /etc/mysql-manticore/river.toml
-COPY etc/dict /etc/mysql-manticore/dict
-
-EXPOSE 8080
-
-# Use an unprivileged user.
-USER appuser:appuser
-
-ENTRYPOINT ["mysql-manticore", "-config", "/etc/mysql-manticore/river.toml"]

river/config.go

@@ -104,8 +104,8 @@ type Config struct {
 
 	StatAddr string `toml:"stat_addr"`
 
-	NatsEnabled bool   `toml:"nats_enabled"`
 	NatsAddr    string `toml:"nats_addr"`
+	NatsEnabled bool   `toml:"nats_enabled"`
 
 	ServerID        uint32       `toml:"server_id"`
 	Flavor          string       `toml:"flavor"`
@@ -135,6 +135,8 @@ type Config struct {
 	SkipRebuild       bool `toml:"skip_rebuild"`
 	SkipUploadIndex   bool `toml:"skip_upload_index"`
 	SkipReloadRtIndex bool `toml:"skip_reload_rt_index"`
+
+	TaoMap map[string]string `toml:"tao_type"`
 }
 
 // IndexConfigField field of an index as it's seen in the indexer config

river/cron_service.go

@@ -33,7 +33,7 @@ func (s *CronService) Serve() {
 
 	s.cron = cron.New(
 		cron.WithSeconds(),
-		cron.WithLogger(cron.VerbosePrintfLogger(stdLogger)),
+		cron.WithLogger(cron.PrintfLogger(stdLogger)),
 	)
 
 	cfg := s.riverInstance.c.MaintenanceConfig

river/nats_service.go

@@ -3,6 +3,7 @@ package river
 import (
 	"context"
 	"fmt"
+	"strconv"
 	"sync"
 	"time"
 
@@ -17,6 +18,9 @@ type Message struct {
 	// Id is an unique identifier of action.
 	Id uint64
 
+	// Type id of the event
+	Type uint32
+
 	// event name: users/posts/comments etc
 	Name string
 
@@ -44,6 +48,8 @@ type NatsService struct {
 	sphm          sync.Mutex
 	wg            sync.WaitGroup
 
+	seq uint64
+
 	// NATS connection types
 	nc *nats.Conn
 	ec *nats.EncodedConn
@@ -83,8 +89,9 @@ func (s *NatsService) SyncLoop(ctx context.Context) {
 
 	for {
 		if s.ec == nil {
+			s.log.Debug("SyncLoop Nats not connected, so drain channel")
 			<-sync2Nats
-			break
+			continue
 		}
 
 		select {
@@ -93,20 +100,33 @@ func (s *NatsService) SyncLoop(ctx context.Context) {
 		case <-s.ctx.Done():
 			s.log.Info("SyncLoop Closed")
 			break
+		default:
+			s.log.Debug("SyncLoop No activity")
 		}
 	}
+
+	close(sync2Nats)
+	s.log.Info("SyncLoop exited")
 }
 
 func (s *NatsService) Publish(v *Message) {
 	if s.ec == nil {
 		return
 	}
 
-	sub := fmt.Sprintf("binlog.%s.%s", v.Name, v.Action)
-	s.log.Debugf("Publishing event %s: %d", sub, v.Id)
+	cfg := s.riverInstance.c
+	typ := strconv.FormatUint(uint64(v.Type), 10)
+
+	// try to get name from tao map
+	if name, ok := cfg.TaoMap[typ]; ok {
+		v.Name = name
+	}
+
+	sub := fmt.Sprintf("binlog.%s.%s.%d", v.Name, v.Action, s.nextSeqNo())
+	s.log.Infof("Publishing event [%s] for [%d]", sub, v.Id)
 
 	if err := s.ec.Publish(sub, v); err != nil {
-		s.log.Errorf("Error publishing event %d: %v", v.Id, err)
+		s.log.Errorf("Error publishing event [%s] for [%d]: %v", sub, v.Id, err)
 	}
 
 	// // Sends a PING and wait for a PONG from the server, up to the given timeout.
@@ -123,6 +143,12 @@ func NewNatsService(r *River) *NatsService {
 	return &s
 }
 
+func (s *NatsService) nextSeqNo() uint64 {
+	s.seq++
+
+	return s.seq
+}
+
 func (s *NatsService) connect() {
 	s.sphm.Lock()
 	defer s.sphm.Unlock()
@@ -192,11 +218,23 @@ func NewMessage(id uint64) *Message {
 	}
 }
 
-func PublishRowToNats(doc TableRowChange, rule IngestRule) {
+func PublishRowToNats(id uint64, typ uint32, action string, table string) {
+	ev := NewMessage(id)
+	ev.Type = typ
+	ev.Name = table
+	ev.Action = action
+	ev.Metadata["tableName"] = table
+
+	sync2Nats <- ev
+}
+
+func PublishDocToNats(doc TableRowChange, rule IngestRule) {
 	ev := NewMessage(doc.DocID)
-	ev.Name = doc.Index
+	ev.Type = uint32(rule.JsonTypeValue)
+	ev.Name = doc.TableName
 	ev.Action = doc.Action
 
+	ev.Metadata["index"] = doc.Index
 	ev.Metadata["tableName"] = doc.TableName
 	ev.Metadata["timeStamp"] = doc.TS.String()
 

river/river.go

@@ -226,7 +226,7 @@ func (r *River) run() error {
 		r.cronToken = &t
 	}
 
-	if r.natsToken == nil && r.c.NatsEnabled {
+	if r.natsToken == nil {
 		t := r.sup.Add(NewNatsService(r))
 		r.natsToken = &t
 	}

river/rule.go

@@ -30,8 +30,8 @@ type IngestRule struct {
 	Index          string              `toml:"index"`
 	ColumnMap      map[string][]string `toml:"column_map"`
 	JsonColumnName string              `toml:"json_column_name"`
-	JsonTypeName   string              `toml:"json_type_name"`  // optional used for tao_objects type column ex: user or post or comment
-	JsonTypeValue  int                 `toml:"json_type_value"` // optional used for tao_objects type column ex: 1 or 3 or 4
+	JsonTypeName   string              `toml:"json_type_name"`  // optional used for tao_objects type column name ex: type
+	JsonTypeValue  int                 `toml:"json_type_value"` // optional used for tao_objects type column value ex: 1 or 3 or 4
 	timeProvider   func() time.Time
 
 	// MySQL table information
@@ -158,6 +158,7 @@ func (r *IngestRule) Apply(e *canal.RowsEvent) ([]TableRowChange, error) {
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
+
 	switch e.Action {
 	case canal.InsertAction:
 		return r.makeInsertChangeSet(e, idColNo)
@@ -172,7 +173,9 @@ func (r *IngestRule) Apply(e *canal.RowsEvent) ([]TableRowChange, error) {
 // ApplyRuleSet converts row event to document changeset and sends to the channel c
 func ApplyRuleSet(ruleSet []IngestRule, e *canal.RowsEvent, c chan interface{}) (uint64, error) {
 	var docCount uint64
-	var ts = time.Now().UTC()
+
+	// Publish Minimal Row Event - without payload
+	processRowEventForNats(e)
 
 	for ruleID := range ruleSet {
 		rule := ruleSet[ruleID]
@@ -181,37 +184,22 @@ func ApplyRuleSet(ruleSet []IngestRule, e *canal.RowsEvent, c chan interface{})
 			return docCount, errors.Trace(err)
 		}
 
-		// Log skipped event
-		if docs == nil && rule.JsonTypeValue > 0 {
-			log.Infof(
-				"[row event skipped] ruleId=%d index=%s table=%s type!=%d action=%s rows=%d timeStamp=%s",
-				ruleID,
-				rule.Index,
-				e.Table,
-				rule.JsonTypeValue,
-				e.Action,
-				rowCount(e),
-				ts.String(),
-			)
-		}
-
 		if docs != nil {
 			log.Infof(
-				"[row event] ruleId=%d index=%s table=%s type=%d action=%s rows=%d docs=%v timeStamp=%s",
+				"[rule event] ruleId=%d index=%s table=%s type=%d action=%s rows=%d docs=%v",
 				ruleID,
 				rule.Index,
 				e.Table,
 				rule.JsonTypeValue,
 				e.Action,
 				rowCount(e),
 				DocIDList(docs),
-				ts.String(),
 			)
+
 			for _, doc := range docs {
 				c <- doc
 				docCount++
-
-				PublishRowToNats(doc, rule)
+				// PublishDocToNats(doc, rule)
 			}
 		}
 	}
@@ -427,3 +415,49 @@ func isJsonRowTypeValid(e *canal.RowsEvent, typColNo, typColVal int) (bool, erro
 
 	return false, nil
 }
+
+func processRowEventForNats(e *canal.RowsEvent) {
+	idColNo, _ := fieldIndex(e, "id")
+	typColNo, _ := fieldIndex(e, "type")
+
+	if e.Action == "update" {
+		for i := 0; i < len(e.Rows)/2; i++ {
+			var id uint64
+			var typ uint32
+			newRow := e.Rows[i*2+1]
+
+			if idColNo >= 0 {
+				id, _ = util.CoerceToUint64(newRow[idColNo])
+			}
+
+			if typColNo >= 0 {
+				typ, _ = util.CoerceToUint32(newRow[typColNo])
+			}
+
+			if id > 0 && typ > 0 {
+				PublishRowToNats(id, typ, e.Action, e.Table.String())
+				log.Debugf("[row event] table=%s type=%d action=%s id=%d", e.Table, typ, e.Action, id)
+			}
+		}
+	}
+
+	if e.Action == "insert" || e.Action == "delete" {
+		for _, row := range e.Rows {
+			var id uint64
+			var typ uint32
+
+			if idColNo >= 0 {
+				id, _ = util.CoerceToUint64(row[idColNo])
+			}
+
+			if typColNo >= 0 {
+				typ, _ = util.CoerceToUint32(row[typColNo])
+			}
+
+			if id > 0 && typ > 0 {
+				PublishRowToNats(id, typ, e.Action, e.Table.String())
+				log.Debugf("[row event] table=%s type=%d action=%s id=%d", e.Table, typ, e.Action, id)
+			}
+		}
+	}
+}

sphinx/sphinx.go

@@ -300,7 +300,7 @@ func (c *SphConn) reconnect() error {
 	}
 
 	// addr := c.RemoteAddr().String()
-	log.Infof("[sphinx-reconnect] %s", c.addr)
+	log.Debugf("[sphinx-reconnect] %s", c.addr)
 	conn, err := client.Connect(c.addr, "", "", "")
 	if err != nil {
 		return errors.Trace(err)