Skip to content

Commit ac9f5fc

Browse files
mabdhmabdh
authored
feat(telemetry): add telemetry to worker (#161)
* feat(telemetry): add telemetry to worker * refactor(telemetry): additional span attributes arg * refactor(telemetry): span key name * feat(telemetry): add gauge message queue time
1 parent a2bf367 commit ac9f5fc

File tree

11 files changed

+64
-40
lines changed

11 files changed

+64
-40
lines changed

‎cli/worker.go

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@ import (
1111
"github.com/odpf/siren/core/notification"
1212
"github.com/odpf/siren/pkg/pgc"
1313
"github.com/odpf/siren/pkg/secret"
14+
"github.com/odpf/siren/pkg/telemetry"
1415
"github.com/odpf/siren/pkg/worker"
1516
"github.com/odpf/siren/plugins/queues"
1617
"github.com/odpf/siren/plugins/queues/postgresq"
@@ -133,6 +134,8 @@ func workerStartNotificationDLQHandlerCommand() *cobra.Command {
133134
func StartNotificationHandlerWorker(ctx context.Context, cfg config.Config, cancelWorkerChan chan struct{}) error {
134135
logger := initLogger(cfg.Log)
135136

137+
telemetry.Init(ctx, cfg.Telemetry, logger)
138+
136139
dbClient, err := db.New(cfg.DB)
137140
if err != nil {
138141
return err
@@ -182,6 +185,8 @@ func StartNotificationHandlerWorker(ctx context.Context, cfg config.Config, canc
182185
func StartNotificationDLQHandlerWorker(ctx context.Context, cfg config.Config, cancelWorkerChan chan struct{}) error {
183186
logger := initLogger(cfg.Log)
184187

188+
telemetry.Init(ctx, cfg.Telemetry, logger)
189+
185190
dbClient, err := db.New(cfg.DB)
186191
if err != nil {
187192
return err

‎core/notification/handler.go

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ import (
99
"github.com/odpf/siren/pkg/errors"
1010
"github.com/odpf/siren/pkg/telemetry"
1111
"go.opencensus.io/tag"
12+
"go.opencensus.io/trace"
1213
)
1314

1415
const (
@@ -85,11 +86,15 @@ func (h *Handler) Process(ctx context.Context, runAt time.Time) error {
8586
if len(receiverTypes) == 0 {
8687
return errors.New("no receiver type plugin registered, skipping dequeue")
8788
} else {
88-
ctx, span := h.messagingTracer.StartSpan(ctx, "batch_dequeue", nil)
89+
ctx, span := h.messagingTracer.StartSpan(ctx, "batch_dequeue", trace.StringAttribute("messaging.handler_id", h.identifier))
8990
defer span.End()
9091

9192
if err := h.q.Dequeue(ctx, receiverTypes, h.batchSize, h.MessageHandler); err != nil {
9293
if !errors.Is(err, ErrNoMessage) {
94+
span.SetStatus(trace.Status{
95+
Code: trace.StatusCodeUnknown,
96+
Message: err.Error(),
97+
})
9398
return fmt.Errorf("dequeue failed on handler with id %s: %w", h.identifier, err)
9499
}
95100
}
@@ -100,6 +105,10 @@ func (h *Handler) Process(ctx context.Context, runAt time.Time) error {
100105
// MessageHandler is a function to handler dequeued message
101106
func (h *Handler) MessageHandler(ctx context.Context, messages []Message) error {
102107
for _, message := range messages {
108+
109+
telemetry.GaugeMillisecond(ctx, telemetry.MetricNotificationMessageQueueTime, time.Since(message.UpdatedAt).Milliseconds(),
110+
tag.Upsert(telemetry.TagReceiverType, message.ReceiverType))
111+
103112
notifier, err := h.getNotifierPlugin(message.ReceiverType)
104113
if err != nil {
105114
return err

‎core/notification/message.go

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -93,6 +93,7 @@ func (m *Message) Initialize(
9393

9494
m.ReceiverType = receiverType
9595
m.Configs = notificationConfigs
96+
9697
details := make(map[string]interface{})
9798
for k, v := range n.Labels {
9899
details[k] = v

‎core/notification/service.go

Lines changed: 9 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@ import (
1010
"github.com/odpf/siren/pkg/errors"
1111
"github.com/odpf/siren/pkg/telemetry"
1212
"go.opencensus.io/tag"
13+
"go.opencensus.io/trace"
1314
"gopkg.in/yaml.v3"
1415
)
1516

@@ -71,10 +72,10 @@ func (ns *NotificationService) DispatchToReceiver(ctx context.Context, n Notific
7172
return err
7273
}
7374

74-
ctx, span := ns.messagingTracer.StartSpan(ctx, "prepare_enqueue", map[string]string{
75-
"messages.notification_id": n.ID,
76-
"messages.routing_method": RoutingMethodReceiver.String(),
77-
})
75+
ctx, span := ns.messagingTracer.StartSpan(ctx, "prepare_enqueue",
76+
trace.StringAttribute("messaging.notification_id", n.ID),
77+
trace.StringAttribute("messaging.routing_method", RoutingMethodReceiver.String()),
78+
)
7879
defer span.End()
7980

8081
notifierPlugin, err := ns.getNotifierPlugin(rcv.Type)
@@ -125,10 +126,10 @@ func (ns *NotificationService) DispatchToSubscribers(ctx context.Context, n Noti
125126
return errors.ErrInvalid.WithMsgf("not matching any subscription")
126127
}
127128

128-
ctx, span := ns.messagingTracer.StartSpan(ctx, "prepare_enqueue", map[string]string{
129-
"messages.notification_id": n.ID,
130-
"messages.routing_method": RoutingMethodSubscribers.String(),
131-
})
129+
ctx, span := ns.messagingTracer.StartSpan(ctx, "prepare_enqueue",
130+
trace.StringAttribute("messaging.notification_id", n.ID),
131+
trace.StringAttribute("messaging.routing_method", RoutingMethodSubscribers.String()),
132+
)
132133
defer span.End()
133134

134135
var messages = make([]Message, 0)

‎docs/docs/reference/server_configuration.md

Lines changed: 0 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -46,9 +46,6 @@ telemetry:
4646
# OpenCensus.
4747
enable_newrelic: <bool> | default=false
4848

49-
# new relic app name, if left empty, app name will be service_name
50-
newrelic_app_name: <string> | default=""
51-
5249
# newrelic_api_key must be a valid NewRelic License key.
5350
newrelic_api_key: <string> | default="____LICENSE_STRING_OF_40_CHARACTERS_____"
5451

‎pkg/pgc/client.go

Lines changed: 15 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -88,9 +88,9 @@ func Migrate(cfg db.Config) error {
8888
}
8989

9090
func (c *Client) QueryRowxContext(ctx context.Context, op string, tableName string, query string, args ...interface{}) *sqlx.Row {
91-
ctx, span := c.postgresTracer.StartSpan(ctx, op, tableName, map[string]string{
92-
"db.statement": query,
93-
})
91+
ctx, span := c.postgresTracer.StartSpan(ctx, op, tableName,
92+
trace.StringAttribute("db.statement", query),
93+
)
9494
defer span.End()
9595
sqlxRow := c.GetDB(ctx).QueryRowxContext(ctx, query, args...)
9696
if sqlxRow.Err() != nil {
@@ -103,9 +103,9 @@ func (c *Client) QueryRowxContext(ctx context.Context, op string, tableName stri
103103
}
104104

105105
func (c *Client) QueryxContext(ctx context.Context, op string, tableName string, query string, args ...interface{}) (*sqlx.Rows, error) {
106-
ctx, span := c.postgresTracer.StartSpan(ctx, op, tableName, map[string]string{
107-
"db.statement": query,
108-
})
106+
ctx, span := c.postgresTracer.StartSpan(ctx, op, tableName,
107+
trace.StringAttribute("db.statement", query),
108+
)
109109
defer span.End()
110110
sqlxRow, err := c.GetDB(ctx).QueryxContext(ctx, query, args...)
111111
if err != nil {
@@ -118,9 +118,9 @@ func (c *Client) QueryxContext(ctx context.Context, op string, tableName string,
118118
}
119119

120120
func (c *Client) GetContext(ctx context.Context, op string, tableName string, dest interface{}, query string, args ...interface{}) error {
121-
ctx, span := c.postgresTracer.StartSpan(ctx, op, tableName, map[string]string{
122-
"db.statement": query,
123-
})
121+
ctx, span := c.postgresTracer.StartSpan(ctx, op, tableName,
122+
trace.StringAttribute("db.statement", query),
123+
)
124124
defer span.End()
125125

126126
if err := c.GetDB(ctx).QueryRowxContext(ctx, query, args...).StructScan(dest); err != nil {
@@ -135,9 +135,9 @@ func (c *Client) GetContext(ctx context.Context, op string, tableName string, de
135135
}
136136

137137
func (c *Client) ExecContext(ctx context.Context, op string, tableName string, query string, args ...interface{}) (sql.Result, error) {
138-
ctx, span := c.postgresTracer.StartSpan(ctx, op, tableName, map[string]string{
139-
"db.statement": query,
140-
})
138+
ctx, span := c.postgresTracer.StartSpan(ctx, op, tableName,
139+
trace.StringAttribute("db.statement", query),
140+
)
141141
defer span.End()
142142

143143
res, err := c.db.ExecContext(ctx, query, args...)
@@ -153,9 +153,9 @@ func (c *Client) ExecContext(ctx context.Context, op string, tableName string, q
153153
}
154154

155155
func (c *Client) NamedExecContext(ctx context.Context, op string, tableName string, query string, arg interface{}) (sql.Result, error) {
156-
ctx, span := c.postgresTracer.StartSpan(ctx, op, tableName, map[string]string{
157-
"db.statement": query,
158-
})
156+
ctx, span := c.postgresTracer.StartSpan(ctx, op, tableName,
157+
trace.StringAttribute("db.statement", query),
158+
)
159159
defer span.End()
160160

161161
res, err := c.db.NamedExecContext(ctx, query, arg)

‎pkg/telemetry/application.go

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,8 @@ var (
1414
TagRoutingMethod = tag.MustNewKey("routing_method")
1515
TagMessageStatus = tag.MustNewKey("status")
1616

17+
MetricNotificationMessageQueueTime = stats.Int64("notification.message.queue.time", "time of message from enqueued to be picked up", stats.UnitMilliseconds)
18+
1719
MetricNotificationMessageEnqueue = stats.Int64("notification.message.enqueue", "enqueued notification messages", stats.UnitDimensionless)
1820
MetricNotificationMessagePending = stats.Int64("notification.message.pending", "processed notification messages", stats.UnitDimensionless)
1921
MetricNotificationMessageFailed = stats.Int64("notification.message.failed", "failed to publish notification messages", stats.UnitDimensionless)
@@ -29,6 +31,13 @@ var (
2931

3032
func setupApplicationViews() error {
3133
return view.Register(
34+
&view.View{
35+
Name: MetricNotificationMessageQueueTime.Name(),
36+
Description: MetricNotificationMessageQueueTime.Description(),
37+
TagKeys: []tag.Key{TagReceiverType},
38+
Measure: MetricNotificationMessageQueueTime,
39+
Aggregation: view.Distribution(),
40+
},
3241
&view.View{
3342
Name: MetricNotificationMessageEnqueue.Name(),
3443
Description: MetricNotificationMessageEnqueue.Description(),

‎pkg/telemetry/messaging.go

Lines changed: 2 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,7 @@ func NewMessagingTracer(queueSystem string) *MessagingTracer {
1717
}
1818
}
1919

20-
func (msg MessagingTracer) StartSpan(ctx context.Context, op string, spanAttributes map[string]string) (context.Context, *trace.Span) {
20+
func (msg MessagingTracer) StartSpan(ctx context.Context, op string, spanAttributes ...trace.Attribute) (context.Context, *trace.Span) {
2121
// Refer https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/trace/semantic_conventions/messaging.md
2222
ctx, span := trace.StartSpan(ctx, fmt.Sprintf("notification_queue %s", op), trace.WithSpanKind(trace.SpanKindClient))
2323

@@ -28,9 +28,7 @@ func (msg MessagingTracer) StartSpan(ctx context.Context, op string, spanAttribu
2828
trace.StringAttribute("messaging.operation", op),
2929
}
3030

31-
for k, v := range spanAttributes {
32-
traceAttributes = append(traceAttributes, trace.StringAttribute(k, v))
33-
}
31+
traceAttributes = append(traceAttributes, spanAttributes...)
3432

3533
span.AddAttributes(
3634
traceAttributes...,

‎pkg/telemetry/opencensus.go

Lines changed: 1 addition & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -37,11 +37,7 @@ func setupOpenCensus(ctx context.Context, mux *http.ServeMux, cfg Config) error
3737
}
3838

3939
if cfg.EnableNewrelic {
40-
nrAppName := cfg.ServiceName
41-
if cfg.NewRelicAppName != "" {
42-
nrAppName = cfg.NewRelicAppName
43-
}
44-
exporter, err := nrcensus.NewExporter(nrAppName, cfg.NewRelicAPIKey)
40+
exporter, err := nrcensus.NewExporter(cfg.ServiceName, cfg.NewRelicAPIKey)
4541
if err != nil {
4642
return err
4743
}

‎pkg/telemetry/postgres.go

Lines changed: 2 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,7 @@ func NewPostgresTracer(url string) (*PostgresTracer, error) {
3131
}, err
3232
}
3333

34-
func (d PostgresTracer) StartSpan(ctx context.Context, op string, tableName string, spanAttributes map[string]string) (context.Context, *trace.Span) {
34+
func (d PostgresTracer) StartSpan(ctx context.Context, op string, tableName string, spanAttributes ...trace.Attribute) (context.Context, *trace.Span) {
3535
// Refer https://github.com/open-telemetry/opentelemetry-specification/blob/master/specification/trace/semantic_conventions/database.md
3636
ctx, span := trace.StartSpan(ctx, fmt.Sprintf("%s %s.%s", op, d.dbName, tableName), trace.WithSpanKind(trace.SpanKindClient))
3737

@@ -45,9 +45,7 @@ func (d PostgresTracer) StartSpan(ctx context.Context, op string, tableName stri
4545
trace.StringAttribute("db.sql.table", tableName),
4646
}
4747

48-
for k, v := range spanAttributes {
49-
traceAttributes = append(traceAttributes, trace.StringAttribute(k, v))
50-
}
48+
traceAttributes = append(traceAttributes, spanAttributes...)
5149

5250
span.AddAttributes(
5351
traceAttributes...,

‎pkg/telemetry/telemetry.go

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -41,3 +41,13 @@ func IncrementInt64Counter(ctx context.Context, si64 *stats.Int64Measure, tagMut
4141

4242
stats.Record(counterCtx, si64.M(1))
4343
}
44+
45+
func GaugeMillisecond(ctx context.Context, si64 *stats.Int64Measure, value int64, tagMutator ...tag.Mutator) {
46+
counterCtx := ctx
47+
48+
if tagMutator != nil {
49+
counterCtx, _ = tag.New(ctx, tagMutator...)
50+
}
51+
52+
stats.Record(counterCtx, si64.M(value))
53+
}

0 commit comments

Comments
 (0)