-
Notifications
You must be signed in to change notification settings - Fork 665
/
forwarder.go
63 lines (53 loc) · 1.89 KB
/
forwarder.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
// Copyright (c) Abstract Machines
// SPDX-License-Identifier: Apache-2.0
package tracing
import (
"context"
"fmt"
"github.com/absmach/magistrala/internal/server"
"github.com/absmach/magistrala/mqtt"
"github.com/absmach/magistrala/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 {
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)
}