feat: Create more granular OpenTracing spans

Makefile

@@ -28,7 +28,6 @@ create-cassandra-table:
 	@docker exec $(CASSANDRA_CONTAINER) cqlsh -e "$$(cat scripts/create.cql)";
 	@echo 'Done'
 
-# make setup/mongo MONGODB_HOST=mongodb://localhost:27017 or make MONGODB_HOST=mongodb://localhost:27017 setup/mongo
 setup/mongo: 
 	go run scripts/setup_mongo_messages-index.go
 

README.md

@@ -42,7 +42,7 @@ Make sure you have Go installed on your machine.
 
 You also need to have access to running instances of Cassandra and Mongo.
 
-## Running the application
+### Running the application
 
 If you want to run the application locally you can do so by running
 
@@ -57,7 +57,7 @@ You may need to change the configurations to point to your MQTT, Cassandra
 and Mongo servers, or you can use the provided containers, they can be run
 by executing `make run-containers`
 
-## Running the tests
+### Running the tests
 
 The project is integrated with Github Actions and uses docker to run the needed services.
 
@@ -91,3 +91,22 @@ httpAuth:
       username: user
       password: pass
 ```
+## Observability
+
+### Logs
+
+There is a middleware using the [zap](https://github.com/uber-go/zap) package to log requests with the JSON zap encoder.
+
+All other pieces of code use the [op](https://github.com/op/go-logging) package with a custom format.
+
+⚠️ Only logs formatted as JSON are usually exported to a logging platform, i.e., only the aforementioned middleware is logging correctly and the team should fix this issue.
+
+### Traces
+
+OpenTracing is used to report traces. The client implementation used is Jaeger's.
+
+⚠️ OpenTracing is deprecated, as well as Jaeger's client package. That is, the team should replace OpenTracing with OpenTelemetry and Jaeger's client package with OpenTelemetry's agnostic client package.
+
+## Metrics
+
+Metrics are exported using [Datadog's statsd](https://github.com/DataDog/datadog-go) client package.

app/app.go

@@ -163,7 +163,6 @@ func (app *App) configureJaeger() {
 			cfg.Sampler.Type = jaeger.SamplerTypeProbabilistic
 		}
 	}
-	logger.Logger.Info("Jaeger Config: ", cfg)
 	if _, err := cfg.InitGlobalTracer(""); err != nil {
 		logger.Logger.Error("Failed to initialize Jaeger.", err)
 	} else {
@@ -208,7 +207,6 @@ func (app *App) configureApplication() {
 	a.Use(NewLoggerMiddleware(zap.New(
 		zap.NewJSONEncoder(),
 	)).Serve)
-	a.Use(NewJaegerMiddleware())
 	a.Use(NewSentryMiddleware().Serve)
 	a.Use(VersionMiddleware)
 	a.Use(NewRecoveryMiddleware(app.OnErrorHandler).Serve)

app/helpers.go

@@ -82,7 +82,11 @@ func findAuthorizedTopics(ctx context.Context, username string, topics []string)
 		return cursor.All(ctx, &searchResults)
 	}
 	search := func() error {
-		return mongoclient.GetCollection("mqtt_acl", query)
+		mongoCollection, err := mongoclient.GetCollection(ctx, "mqtt_acl")
+		if err != nil {
+			return err
+		}
+		return query(mongoCollection)
 	}
 	err := search()
 	return searchResults, err

app/histories.go

@@ -36,7 +36,15 @@ func HistoriesHandler(app *App) func(c echo.Context) error {
 			collection := app.Defaults.MongoMessagesCollection
 
 			for _, topic := range authorizedTopics {
-				topicMessages := mongoclient.GetMessages(c, topic, from, limit, collection)
+				topicMessages := mongoclient.GetMessages(
+					c,
+					mongoclient.QueryParameters{
+						Topic:      topic,
+						From:       from,
+						Limit:      limit,
+						Collection: collection,
+					},
+				)
 				messages = append(messages, topicMessages...)
 			}
 			return c.JSON(http.StatusOK, messages)

app/histories_test.go

@@ -176,7 +176,10 @@ func TestHistoriesHandler(t *testing.T) {
 					return err
 				}
 
-				err := mongoclient.GetCollection("mqtt_acl", query)
+				mongoCollection, err := mongoclient.GetCollection(ctx, "mqtt_acl")
+				Expect(err).To(BeNil())
+
+				err = query(mongoCollection)
 				Expect(err).To(BeNil())
 
 				testMessage := models.Message{

app/histories_v2.go

@@ -36,7 +36,15 @@ func HistoriesV2Handler(app *App) func(c echo.Context) error {
 		collection := app.Defaults.MongoMessagesCollection
 
 		for _, topic := range authorizedTopics {
-			topicMessages := mongoclient.GetMessagesV2(c, topic, from, limit, collection)
+			topicMessages := mongoclient.GetMessagesV2(
+				c,
+				mongoclient.QueryParameters{
+					Topic:      topic,
+					From:       from,
+					Limit:      limit,
+					Collection: collection,
+				},
+			)
 			messages = append(messages, topicMessages...)
 		}
 

app/history.go

@@ -30,7 +30,15 @@ func HistoryHandler(app *App) func(c echo.Context) error {
 
 		if app.Defaults.MongoEnabled {
 			collection := app.Defaults.MongoMessagesCollection
-			messages := mongoclient.GetMessages(c, topic, from, limit, collection)
+			messages := mongoclient.GetMessages(
+				c,
+				mongoclient.QueryParameters{
+					Topic:      topic,
+					From:       from,
+					Limit:      limit,
+					Collection: collection,
+				},
+			)
 			return c.JSON(http.StatusOK, messages)
 		}
 

app/history_v2.go

@@ -33,7 +33,16 @@ func HistoryV2Handler(app *App) func(c echo.Context) error {
 
 		messages := make([]*models.MessageV2, 0)
 		collection := app.Defaults.MongoMessagesCollection
-		messages = mongoclient.GetMessagesV2WithParameter(c, topic, from, limit, collection, isBlocked)
+		messages = mongoclient.GetMessagesV2WithParameter(
+			c,
+			mongoclient.QueryParameters{
+				Topic:      topic,
+				From:       from,
+				Limit:      limit,
+				Collection: collection,
+				IsBlocked:  isBlocked,
+			},
+		)
 
 		if len(messages) > 0 {
 			gameId := messages[0].GameId

app/history_v2_player_support.go

@@ -36,7 +36,18 @@ func HistoriesV2PSHandler(app *App) func(c echo.Context) error {
 
 		messages := make([]*models.MessageV2, 0)
 		collection := app.Defaults.MongoMessagesCollection
-		messages = mongoclient.GetMessagesPlayerSupportV2WithParameter(c, topic, from, to, limit, collection, isBlocked, playerId)
+		messages = mongoclient.GetMessagesPlayerSupportV2WithParameter(
+			c,
+			mongoclient.QueryParameters{
+				Topic:      topic,
+				From:       from,
+				To:         to,
+				Limit:      limit,
+				Collection: collection,
+				IsBlocked:  isBlocked,
+				PlayerID:   playerId,
+			},
+		)
 
 		return c.JSON(http.StatusOK, messages)
 	}

app/middleware.go

@@ -2,17 +2,12 @@ package app
 
 import (
 	"fmt"
-	"net/http"
 	"runtime/debug"
 	"time"
 
 	"github.com/getsentry/raven-go"
 	"github.com/labstack/echo"
-	"github.com/labstack/echo/engine"
-	"github.com/opentracing/opentracing-go"
-	"github.com/opentracing/opentracing-go/log"
 	"github.com/topfreegames/extensions/middleware"
-	"github.com/topfreegames/mqtt-history/logger"
 	"github.com/uber-go/zap"
 )
 
@@ -236,78 +231,3 @@ func NewResponseTimeMetricsMiddleware(ddStatsD *middleware.DogStatsD) *ResponseT
 		DDStatsD: ddStatsD,
 	}
 }
-
-// NewJaegerMiddleware create a new middleware to instrument traces.
-func NewJaegerMiddleware() echo.MiddlewareFunc {
-	return func(next echo.HandlerFunc) echo.HandlerFunc {
-		return func(c echo.Context) error {
-			tracer := opentracing.GlobalTracer()
-
-			request := c.Request()
-			method := request.Method()
-			url := request.URL()
-			header := getCarrier(request)
-			parent, err := tracer.Extract(opentracing.HTTPHeaders, header)
-			if err != nil && err != opentracing.ErrSpanContextNotFound {
-				logger.Logger.Errorf(
-					"Could no extract parent trace from incoming request. Method: %s. Path: %s.",
-					method,
-					url.Path(),
-					err,
-				)
-			}
-
-			operationName := fmt.Sprintf("HTTP %s %s", method, c.Path())
-			reference := opentracing.ChildOf(parent)
-			tags := opentracing.Tags{
-				"http.method":   method,
-				"http.host":     request.Host(),
-				"http.pathname": url.Path(),
-				"http.query":    url.QueryString(),
-				"span.kind":     "server",
-			}
-			span := opentracing.StartSpan(operationName, reference, tags)
-			defer span.Finish()
-			defer func(span opentracing.Span) {
-				if err, ok := recover().(error); ok {
-					span.SetTag("error", true)
-					span.LogFields(
-						log.String("event", "error"),
-						log.String("message", "Panic serving request."),
-						log.Error(err),
-					)
-					panic(err)
-				}
-			}(span)
-
-			ctx := c.StdContext()
-			ctx = opentracing.ContextWithSpan(ctx, span)
-			c.SetStdContext(ctx)
-
-			err = next(c)
-			if err != nil {
-				span.SetTag("error", true)
-				span.LogFields(
-					log.String("event", "error"),
-					log.String("message", "Error serving request."),
-					log.Error(err),
-				)
-			}
-
-			response := c.Response()
-			statusCode := response.Status()
-			span.SetTag("http.status_code", statusCode)
-			return err
-		}
-	}
-}
-
-func getCarrier(request engine.Request) opentracing.HTTPHeadersCarrier {
-	original := request.Header()
-	copy := make(http.Header)
-	for _, key := range original.Keys() {
-		value := original.Get(key)
-		copy.Set(key, value)
-	}
-	return opentracing.HTTPHeadersCarrier(copy)
-}