mirror of
https://github.com/mainflux/mainflux.git
synced 2025-04-29 13:49:28 +08:00
38992085bd
* Initial Commit: Sync Env Veriables With Docker Deployment Signed-off-by: rodneyosodo <blackd0t@protonmail.com> * Sync Env Vars With Master Signed-off-by: rodneyosodo <blackd0t@protonmail.com> * Initial Commit: Add Tags to Database and Message Bus Signed-off-by: rodneyosodo <blackd0t@protonmail.com> * Format Address Well Signed-off-by: rodneyosodo <blackd0t@protonmail.com> * Propagate Context Signed-off-by: rodneyosodo <blackd0t@protonmail.com> * Update PostgresSQL spans Signed-off-by: rodneyosodo <blackd0t@protonmail.com> * Update Message Bus Spans Signed-off-by: rodneyosodo <blackd0t@protonmail.com> * Add Tracing To MQTT Adapter Signed-off-by: rodneyosodo <blackd0t@protonmail.com> * Add Span Tags to HTTP Signed-off-by: rodneyosodo <blackd0t@protonmail.com> * Combine Tracing and PubSub Signed-off-by: rodneyosodo <blackd0t@protonmail.com> * Fix Error After Rebase Signed-off-by: rodneyosodo <blackd0t@protonmail.com> * Reorder Server Config Signed-off-by: rodneyosodo <blackd0t@protonmail.com> * Seperate Tracing Signed-off-by: rodneyosodo <blackd0t@protonmail.com> * shorten span names Signed-off-by: rodneyosodo <blackd0t@protonmail.com> * Fix Issue After Rebase Signed-off-by: rodneyosodo <blackd0t@protonmail.com> --------- Signed-off-by: rodneyosodo <blackd0t@protonmail.com>
64 lines
1.9 KiB
Go
64 lines
1.9 KiB
Go
// Copyright (c) Mainflux
|
|
// SPDX-License-Identifier: Apache-2.0
|
|
|
|
package tracing
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
|
|
"github.com/mainflux/mainflux/internal/server"
|
|
"github.com/mainflux/mainflux/mqtt"
|
|
"github.com/mainflux/mainflux/pkg/messaging"
|
|
"go.opentelemetry.io/otel/attribute"
|
|
"go.opentelemetry.io/otel/trace"
|
|
)
|
|
|
|
const forwardOP = "process"
|
|
|
|
var _ mqtt.Forwarder = (*forwarderMiddleware)(nil)
|
|
|
|
type forwarderMiddleware struct {
|
|
topic string
|
|
forwarder mqtt.Forwarder
|
|
tracer trace.Tracer
|
|
host server.Config
|
|
}
|
|
|
|
// New creates new mqtt forwarder tracing middleware.
|
|
func New(config server.Config, tracer trace.Tracer, forwarder mqtt.Forwarder, topic string) mqtt.Forwarder {
|
|
return &forwarderMiddleware{
|
|
forwarder: forwarder,
|
|
tracer: tracer,
|
|
topic: topic,
|
|
host: config,
|
|
}
|
|
}
|
|
|
|
// Forward traces mqtt forward operations.
|
|
func (fm *forwarderMiddleware) Forward(ctx context.Context, id string, sub messaging.Subscriber, pub messaging.Publisher) error {
|
|
var subject = fmt.Sprintf("channels.%s.messages", fm.topic)
|
|
spanName := fmt.Sprintf("%s %s", subject, forwardOP)
|
|
|
|
ctx, span := fm.tracer.Start(ctx,
|
|
spanName,
|
|
trace.WithAttributes(
|
|
attribute.String("messaging.system", "mqtt"),
|
|
attribute.Bool("messaging.destination.anonymous", false),
|
|
attribute.String("messaging.destination.template", "channels/{channelID}/messages/*"),
|
|
attribute.Bool("messaging.destination.temporary", true),
|
|
attribute.String("network.protocol.name", "mqtt"),
|
|
attribute.String("network.protocol.version", "3.1.1"),
|
|
attribute.String("network.transport", "tcp"),
|
|
attribute.String("network.type", "ipv4"),
|
|
attribute.String("messaging.operation", forwardOP),
|
|
attribute.String("messaging.client_id", id),
|
|
attribute.String("server.address", fm.host.Host),
|
|
attribute.String("server.socket.port", fm.host.Port),
|
|
),
|
|
)
|
|
defer span.End()
|
|
|
|
return fm.forwarder.Forward(ctx, id, sub, pub)
|
|
}
|