perf: Queries issued by the /histories endpoint run concurrently

topfreegames · Oct 6, 2022 · fb016af · fb016af
1 parent 657c260
commit fb016af
Show file tree

Hide file tree

Showing 5 changed files with 89 additions and 42 deletions.
diff --git a/app/app.go b/app/app.go
@@ -26,7 +26,6 @@ import (
 	"github.com/topfreegames/mqtt-history/models"
 	"github.com/uber-go/zap"
 
-	"github.com/uber/jaeger-client-go"
 	"github.com/uber/jaeger-client-go/config"
 )
 
@@ -159,9 +158,6 @@ func (app *App) configureJaeger() {
 		if cfg.ServiceName == "" {
 			cfg.ServiceName = "mqtt-history"
 		}
-		if cfg.Sampler.Type == "" {
-			cfg.Sampler.Type = jaeger.SamplerTypeProbabilistic
-		}
 	}
 	if _, err := cfg.InitGlobalTracer(""); err != nil {
 		logger.Logger.Error("Failed to initialize Jaeger.", err)

diff --git a/app/histories.go b/app/histories.go
@@ -2,6 +2,7 @@ package app
 
 import (
 	"net/http"
+	"sync"
 
 	"github.com/topfreegames/mqtt-history/mongoclient"
 
@@ -30,23 +31,51 @@ func HistoriesHandler(app *App) func(c echo.Context) error {
 			return c.String(echo.ErrUnauthorized.Code, echo.ErrUnauthorized.Message)
 		}
 
-		// retrieve messages
 		messages := make([]*models.Message, 0)
 		if app.Defaults.MongoEnabled {
 			collection := app.Defaults.MongoMessagesCollection
-
+			var wg sync.WaitGroup
+			var mu sync.Mutex
+			// guarantees ordering in responses payload
+			topicsMessagesMap := make(map[string][]*models.MessageV2, len(authorizedTopics))
+			for _, topic := range authorizedTopics {
+				wg.Add(1)
+				go func(topicsMessagesMap map[string][]*models.MessageV2, topic string) {
+					topicMessages := mongoclient.GetMessagesV2(
+						c,
+						mongoclient.QueryParameters{
+							Topic:      topic,
+							From:       from,
+							Limit:      limit,
+							Collection: collection,
+						},
+					)
+					mu.Lock()
+					topicsMessagesMap[topic] = topicMessages
+					mu.Unlock()
+					wg.Done()
+				}(topicsMessagesMap, topic)
+			}
+			wg.Wait()
+			var gameID string
+			// guarantees ordering in responses payload
 			for _, topic := range authorizedTopics {
-				topicMessages := mongoclient.GetMessages(
-					c,
-					mongoclient.QueryParameters{
-						Topic:      topic,
-						From:       from,
-						Limit:      limit,
-						Collection: collection,
-					},
-				)
+				topicMessages := make([]*models.Message, len(topicsMessagesMap[topic]))
+				for idx, topicMessageV2 := range topicsMessagesMap[topic] {
+					topicMessages[idx] = mongoclient.ConvertMessageV2ToMessage(topicMessageV2)
+				}
 				messages = append(messages, topicMessages...)
+				if gameID != "" && len(topicsMessagesMap[topic]) > 0 {
+					gameID = topicsMessagesMap[topic][0].GameId
+				}
 			}
+
+			if gameID != "" {
+				if metricTagsMap, ok := c.Get("metricTagsMap").(map[string]interface{}); ok {
+					metricTagsMap["gameID"] = gameID
+				}
+			}
+
 			return c.JSON(http.StatusOK, messages)
 		}
 

diff --git a/app/histories_v2.go b/app/histories_v2.go
@@ -2,6 +2,7 @@ package app
 
 import (
 	"net/http"
+	"sync"
 
 	"github.com/topfreegames/mqtt-history/logger"
 	"github.com/topfreegames/mqtt-history/mongoclient"
@@ -35,17 +36,32 @@ func HistoriesV2Handler(app *App) func(c echo.Context) error {
 		messages := make([]*models.MessageV2, 0)
 		collection := app.Defaults.MongoMessagesCollection
 
+		var wg sync.WaitGroup
+		var mu sync.Mutex
+		// guarantees ordering in responses payload
+		topicsMessagesMap := make(map[string][]*models.MessageV2, len(authorizedTopics))
 		for _, topic := range authorizedTopics {
-			topicMessages := mongoclient.GetMessagesV2(
-				c,
-				mongoclient.QueryParameters{
-					Topic:      topic,
-					From:       from,
-					Limit:      limit,
-					Collection: collection,
-				},
-			)
-			messages = append(messages, topicMessages...)
+			wg.Add(1)
+			go func(topicsMessagesMap map[string][]*models.MessageV2, topic string) {
+				topicMessages := mongoclient.GetMessagesV2(
+					c,
+					mongoclient.QueryParameters{
+						Topic:      topic,
+						From:       from,
+						Limit:      limit,
+						Collection: collection,
+					},
+				)
+				mu.Lock()
+				topicsMessagesMap[topic] = topicMessages
+				mu.Unlock()
+				wg.Done()
+			}(topicsMessagesMap, topic)
+		}
+		wg.Wait()
+		// guarantees ordering in responses payload
+		for _, topic := range authorizedTopics {
+			messages = append(messages, topicsMessagesMap[topic]...)
 		}
 
 		if len(messages) > 0 {

diff --git a/mongoclient/get_messages.go b/mongoclient/get_messages.go
@@ -18,6 +18,7 @@ import (
 	"go.mongodb.org/mongo-driver/bson"
 
 	"github.com/opentracing/opentracing-go"
+	"github.com/opentracing/opentracing-go/ext"
 	"github.com/opentracing/opentracing-go/log"
 	"go.mongodb.org/mongo-driver/mongo"
 	"go.mongodb.org/mongo-driver/mongo/options"
@@ -59,21 +60,24 @@ func GetMessages(ctx context.Context, queryParameters QueryParameters) []*models
 	searchResults := GetMessagesV2(ctx, queryParameters)
 	messages := make([]*models.Message, 0)
 	for _, result := range searchResults {
-		payload := result.Payload
-		bytes, _ := json.Marshal(payload)
-
-		finalStr := string(bytes)
-		message := &models.Message{
-			Timestamp: time.Unix(result.Timestamp, 0),
-			Payload:   finalStr,
-			Topic:     queryParameters.Topic,
-		}
-		messages = append(messages, message)
+		messages = append(messages, ConvertMessageV2ToMessage(result))
 	}
 
 	return messages
 }
 
+func ConvertMessageV2ToMessage(messagev2 *models.MessageV2) *models.Message {
+	payload := messagev2.Payload
+	bytes, _ := json.Marshal(payload)
+
+	finalStr := string(bytes)
+	return &models.Message{
+		Timestamp: time.Unix(messagev2.Timestamp, 0),
+		Payload:   finalStr,
+		Topic:     messagev2.Topic,
+	}
+}
+
 func convertRawMessageToModelMessage(rawMessage MongoMessage) (*models.MessageV2, error) {
 	playerIdAsString, err := convertPlayerIdToString(rawMessage.PlayerId)
 	if err != nil {
@@ -190,8 +194,10 @@ func getMessagesPlayerSupportFromCollection(
 		ctx,
 		"get_messages_player_support_from_collection",
 		opentracing.Tags{
-			"db.statement": statement,
-			"db.type":      "mongo",
+			string(ext.DBStatement): statement,
+			string(ext.DBType):      "mongo",
+			string(ext.DBInstance):  database,
+			string(ext.DBUser):      user,
 		},
 	)
 	defer span.Finish()

diff --git a/mongoclient/get_messages_v2.go b/mongoclient/get_messages_v2.go
@@ -4,6 +4,7 @@ import (
 	"context"
 
 	"github.com/opentracing/opentracing-go"
+	"github.com/opentracing/opentracing-go/ext"
 	"github.com/opentracing/opentracing-go/log"
 	"github.com/topfreegames/mqtt-history/logger"
 	"github.com/topfreegames/mqtt-history/models"
@@ -82,11 +83,10 @@ func getMessagesFromCollection(
 		ctx,
 		"get_messages_from_collection",
 		opentracing.Tags{
-			"span.kind":    "client",
-			"db.statement": statement,
-			"db.type":      "mongo",
-			"db.instance":  database,
-			"db.user":      user,
+			string(ext.DBStatement): statement,
+			string(ext.DBType):      "mongo",
+			string(ext.DBInstance):  database,
+			string(ext.DBUser):      user,
 		},
 	)
 	defer span.Finish()