Skip to content

Commit 62f40df

Browse files
ckcdckcd
authored
Add additional status properties for pub/sub metrics (dapr#7652)
* add additional status properties for pub/sub metrics Signed-off-by: Kun Chang <[email protected]> * fix integration test Signed-off-by: Kun Chang <[email protected]> --------- Signed-off-by: Kun Chang <[email protected]>
1 parent e4feaa6 commit 62f40df

File tree

8 files changed

+58
-39
lines changed

8 files changed

+58
-39
lines changed

‎pkg/diagnostics/component_monitoring.go‎

Lines changed: 8 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -167,8 +167,8 @@ func (c *componentMetrics) Init(appID, namespace string) error {
167167
c.namespace = namespace
168168

169169
return view.Register(
170-
diagUtils.NewMeasureView(c.pubsubIngressLatency, []tag.Key{appIDKey, componentKey, namespaceKey, processStatusKey, topicKey}, defaultLatencyDistribution),
171-
diagUtils.NewMeasureView(c.pubsubIngressCount, []tag.Key{appIDKey, componentKey, namespaceKey, processStatusKey, topicKey}, view.Count()),
170+
diagUtils.NewMeasureView(c.pubsubIngressLatency, []tag.Key{appIDKey, componentKey, namespaceKey, processStatusKey, topicKey, statusKey}, defaultLatencyDistribution),
171+
diagUtils.NewMeasureView(c.pubsubIngressCount, []tag.Key{appIDKey, componentKey, namespaceKey, processStatusKey, topicKey, statusKey}, view.Count()),
172172
diagUtils.NewMeasureView(c.bulkPubsubIngressLatency, []tag.Key{appIDKey, componentKey, namespaceKey, processStatusKey, topicKey}, defaultLatencyDistribution),
173173
diagUtils.NewMeasureView(c.bulkPubsubIngressCount, []tag.Key{appIDKey, componentKey, namespaceKey, processStatusKey, topicKey}, view.Count()),
174174
diagUtils.NewMeasureView(c.bulkPubsubEventIngressCount, []tag.Key{appIDKey, componentKey, namespaceKey, processStatusKey, topicKey}, view.Count()),
@@ -192,17 +192,20 @@ func (c *componentMetrics) Init(appID, namespace string) error {
192192
}
193193

194194
// PubsubIngressEvent records the metrics for a pub/sub ingress event.
195-
func (c *componentMetrics) PubsubIngressEvent(ctx context.Context, component, processStatus, topic string, elapsed float64) {
195+
func (c *componentMetrics) PubsubIngressEvent(ctx context.Context, component, processStatus, status, topic string, elapsed float64) {
196196
if c.enabled {
197+
if status == "" {
198+
status = processStatus
199+
}
197200
stats.RecordWithTags(
198201
ctx,
199-
diagUtils.WithTags(c.pubsubIngressCount.Name(), appIDKey, c.appID, componentKey, component, namespaceKey, c.namespace, processStatusKey, processStatus, topicKey, topic),
202+
diagUtils.WithTags(c.pubsubIngressCount.Name(), appIDKey, c.appID, componentKey, component, namespaceKey, c.namespace, processStatusKey, processStatus, statusKey, status, topicKey, topic),
200203
c.pubsubIngressCount.M(1))
201204

202205
if elapsed > 0 {
203206
stats.RecordWithTags(
204207
ctx,
205-
diagUtils.WithTags(c.pubsubIngressLatency.Name(), appIDKey, c.appID, componentKey, component, namespaceKey, c.namespace, processStatusKey, processStatus, topicKey, topic),
208+
diagUtils.WithTags(c.pubsubIngressLatency.Name(), appIDKey, c.appID, componentKey, component, namespaceKey, c.namespace, processStatusKey, processStatus, statusKey, status, topicKey, topic),
206209
c.pubsubIngressLatency.M(elapsed))
207210
}
208211
}

‎pkg/diagnostics/component_monitoring_test.go‎

Lines changed: 18 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -21,10 +21,26 @@ func componentsMetrics() *componentMetrics {
2121
}
2222

2323
func TestPubSub(t *testing.T) {
24+
t.Run("record drop by app or sidecar", func(t *testing.T) {
25+
c := componentsMetrics()
26+
27+
c.PubsubIngressEvent(context.Background(), componentName, "drop", "success", "A", 1)
28+
c.PubsubIngressEvent(context.Background(), componentName, "drop", "drop", "A", 1)
29+
30+
viewData, _ := view.RetrieveData("component/pubsub_ingress/count")
31+
v := view.Find("component/pubsub_ingress/count")
32+
33+
allTagsPresent(t, v, viewData[0].Tags)
34+
35+
assert.Len(t, viewData, 2)
36+
assert.Equal(t, int64(1), viewData[0].Data.(*view.CountData).Value)
37+
assert.Equal(t, int64(1), viewData[1].Data.(*view.CountData).Value)
38+
})
39+
2440
t.Run("record ingress count", func(t *testing.T) {
2541
c := componentsMetrics()
2642

27-
c.PubsubIngressEvent(context.Background(), componentName, "retry", "A", 0)
43+
c.PubsubIngressEvent(context.Background(), componentName, "retry", "retry", "A", 0)
2844

2945
viewData, _ := view.RetrieveData("component/pubsub_ingress/count")
3046
v := view.Find("component/pubsub_ingress/count")
@@ -35,7 +51,7 @@ func TestPubSub(t *testing.T) {
3551
t.Run("record ingress latency", func(t *testing.T) {
3652
c := componentsMetrics()
3753

38-
c.PubsubIngressEvent(context.Background(), componentName, "retry", "A", 1)
54+
c.PubsubIngressEvent(context.Background(), componentName, "retry", "", "A", 1)
3955

4056
viewData, _ := view.RetrieveData("component/pubsub_ingress/latencies")
4157
v := view.Find("component/pubsub_ingress/latencies")

‎pkg/runtime/processor/pubsub/publish.go‎

Lines changed: 18 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -121,7 +121,7 @@ func (p *pubsub) publishMessageHTTP(ctx context.Context, msg *subscribedMessage)
121121
elapsed := diag.ElapsedSince(start)
122122

123123
if err != nil {
124-
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, msg.pubsub, strings.ToLower(string(contribpubsub.Retry)), msg.topic, elapsed)
124+
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, msg.pubsub, strings.ToLower(string(contribpubsub.Retry)), "", msg.topic, elapsed)
125125
return fmt.Errorf("error returned from app channel while sending pub/sub event to app: %w", rterrors.NewRetriable(err))
126126
}
127127
defer resp.Close()
@@ -145,7 +145,7 @@ func (p *pubsub) publishMessageHTTP(ctx context.Context, msg *subscribedMessage)
145145
} else {
146146
log.Debugf("skipping status check due to error parsing result from pub/sub event %v: %s", cloudEvent[contribpubsub.IDField], err)
147147
}
148-
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, msg.pubsub, strings.ToLower(string(contribpubsub.Success)), msg.topic, elapsed)
148+
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, msg.pubsub, strings.ToLower(string(contribpubsub.Success)), "", msg.topic, elapsed)
149149
return nil
150150
}
151151

@@ -154,19 +154,19 @@ func (p *pubsub) publishMessageHTTP(ctx context.Context, msg *subscribedMessage)
154154
// Consider empty status field as success
155155
fallthrough
156156
case contribpubsub.Success:
157-
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, msg.pubsub, strings.ToLower(string(contribpubsub.Success)), msg.topic, elapsed)
157+
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, msg.pubsub, strings.ToLower(string(contribpubsub.Success)), "", msg.topic, elapsed)
158158
return nil
159159
case contribpubsub.Retry:
160-
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, msg.pubsub, strings.ToLower(string(contribpubsub.Retry)), msg.topic, elapsed)
160+
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, msg.pubsub, strings.ToLower(string(contribpubsub.Retry)), "", msg.topic, elapsed)
161161
// TODO: add retry error info
162162
return fmt.Errorf("RETRY status returned from app while processing pub/sub event %v: %w", cloudEvent[contribpubsub.IDField], rterrors.NewRetriable(nil))
163163
case contribpubsub.Drop:
164-
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, msg.pubsub, strings.ToLower(string(contribpubsub.Drop)), msg.topic, elapsed)
164+
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, msg.pubsub, strings.ToLower(string(contribpubsub.Drop)), strings.ToLower(string(contribpubsub.Success)), msg.topic, elapsed)
165165
log.Warnf("DROP status returned from app while processing pub/sub event %v", cloudEvent[contribpubsub.IDField])
166166
return rtpubsub.ErrMessageDropped
167167
}
168168
// Consider unknown status field as error and retry
169-
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, msg.pubsub, strings.ToLower(string(contribpubsub.Retry)), msg.topic, elapsed)
169+
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, msg.pubsub, strings.ToLower(string(contribpubsub.Retry)), "", msg.topic, elapsed)
170170
return fmt.Errorf("unknown status returned from app while processing pub/sub event %v, status: %v, err: %w", cloudEvent[contribpubsub.IDField], appResponse.Status, rterrors.NewRetriable(nil))
171171
}
172172

@@ -176,14 +176,14 @@ func (p *pubsub) publishMessageHTTP(ctx context.Context, msg *subscribedMessage)
176176
// When adding/removing an error here, check if that is also applicable to GRPC since there is a mapping between HTTP and GRPC errors:
177177
// https://cloud.google.com/apis/design/errors#handling_errors
178178
log.Errorf("non-retriable error returned from app while processing pub/sub event %v: %s. status code returned: %v", cloudEvent[contribpubsub.IDField], body, statusCode)
179-
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, msg.pubsub, strings.ToLower(string(contribpubsub.Drop)), msg.topic, elapsed)
179+
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, msg.pubsub, strings.ToLower(string(contribpubsub.Drop)), "", msg.topic, elapsed)
180180
return nil
181181
}
182182

183183
// Every error from now on is a retriable error.
184184
errMsg := fmt.Sprintf("retriable error returned from app while processing pub/sub event %v, topic: %v, body: %s. status code returned: %v", cloudEvent[contribpubsub.IDField], cloudEvent[contribpubsub.TopicField], body, statusCode)
185185
log.Warnf(errMsg)
186-
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, msg.pubsub, strings.ToLower(string(contribpubsub.Retry)), msg.topic, elapsed)
186+
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, msg.pubsub, strings.ToLower(string(contribpubsub.Retry)), "", msg.topic, elapsed)
187187
return rterrors.NewRetriable(errors.New(errMsg))
188188
}
189189

@@ -206,14 +206,14 @@ func (p *pubsub) publishMessageGRPC(ctx context.Context, msg *subscribedMessage)
206206
decoded, decodeErr := base64.StdEncoding.DecodeString(dataAsString)
207207
if decodeErr != nil {
208208
log.Debugf("unable to base64 decode cloudEvent field data_base64: %s", decodeErr)
209-
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, msg.pubsub, strings.ToLower(string(contribpubsub.Retry)), msg.topic, 0)
209+
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, msg.pubsub, strings.ToLower(string(contribpubsub.Retry)), "", msg.topic, 0)
210210

211211
return fmt.Errorf("error returned from app while processing pub/sub event: %w", rterrors.NewRetriable(decodeErr))
212212
}
213213

214214
envelope.Data = decoded
215215
} else {
216-
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, msg.pubsub, strings.ToLower(string(contribpubsub.Retry)), msg.topic, 0)
216+
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, msg.pubsub, strings.ToLower(string(contribpubsub.Retry)), "", msg.topic, 0)
217217
return fmt.Errorf("error returned from app while processing pub/sub event: %w", rterrors.NewRetriable(errUnexpectedEnvelopeData))
218218
}
219219
} else if data, ok := cloudEvent[contribpubsub.DataField]; ok && data != nil {
@@ -226,7 +226,7 @@ func (p *pubsub) publishMessageGRPC(ctx context.Context, msg *subscribedMessage)
226226
case []byte:
227227
envelope.Data = v
228228
default:
229-
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, msg.pubsub, strings.ToLower(string(contribpubsub.Retry)), msg.topic, 0)
229+
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, msg.pubsub, strings.ToLower(string(contribpubsub.Retry)), "", msg.topic, 0)
230230
return fmt.Errorf("error returned from app while processing pub/sub event: %w", rterrors.NewRetriable(errUnexpectedEnvelopeData))
231231
}
232232
} else if contenttype.IsJSONContentType(envelope.GetDataContentType()) || contenttype.IsCloudEventContentType(envelope.GetDataContentType()) {
@@ -257,7 +257,7 @@ func (p *pubsub) publishMessageGRPC(ctx context.Context, msg *subscribedMessage)
257257

258258
extensions, extensionsErr := extractCloudEventExtensions(cloudEvent)
259259
if extensionsErr != nil {
260-
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, msg.pubsub, strings.ToLower(string(contribpubsub.Retry)), msg.topic, 0)
260+
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, msg.pubsub, strings.ToLower(string(contribpubsub.Retry)), "", msg.topic, 0)
261261
return extensionsErr
262262
}
263263
envelope.Extensions = extensions
@@ -286,14 +286,14 @@ func (p *pubsub) publishMessageGRPC(ctx context.Context, msg *subscribedMessage)
286286
if hasErrStatus && (errStatus.Code() == codes.Unimplemented) {
287287
// DROP
288288
log.Warnf("non-retriable error returned from app while processing pub/sub event %v: %s", cloudEvent[contribpubsub.IDField], err)
289-
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, msg.pubsub, strings.ToLower(string(contribpubsub.Drop)), msg.topic, elapsed)
289+
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, msg.pubsub, strings.ToLower(string(contribpubsub.Drop)), "", msg.topic, elapsed)
290290

291291
return nil
292292
}
293293

294294
err = fmt.Errorf("error returned from app while processing pub/sub event %v: %w", cloudEvent[contribpubsub.IDField], rterrors.NewRetriable(err))
295295
log.Debug(err)
296-
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, msg.pubsub, strings.ToLower(string(contribpubsub.Retry)), msg.topic, elapsed)
296+
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, msg.pubsub, strings.ToLower(string(contribpubsub.Retry)), "", msg.topic, elapsed)
297297

298298
// on error from application, return error for redelivery of event
299299
return err
@@ -303,21 +303,21 @@ func (p *pubsub) publishMessageGRPC(ctx context.Context, msg *subscribedMessage)
303303
case runtimev1.TopicEventResponse_SUCCESS: //nolint:nosnakecase
304304
// on uninitialized status, this is the case it defaults to as an uninitialized status defaults to 0 which is
305305
// success from protobuf definition
306-
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, msg.pubsub, strings.ToLower(string(contribpubsub.Success)), msg.topic, elapsed)
306+
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, msg.pubsub, strings.ToLower(string(contribpubsub.Success)), "", msg.topic, elapsed)
307307
return nil
308308
case runtimev1.TopicEventResponse_RETRY: //nolint:nosnakecase
309-
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, msg.pubsub, strings.ToLower(string(contribpubsub.Retry)), msg.topic, elapsed)
309+
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, msg.pubsub, strings.ToLower(string(contribpubsub.Retry)), "", msg.topic, elapsed)
310310
// TODO: add retry error info
311311
return fmt.Errorf("RETRY status returned from app while processing pub/sub event %v: %w", cloudEvent[contribpubsub.IDField], rterrors.NewRetriable(nil))
312312
case runtimev1.TopicEventResponse_DROP: //nolint:nosnakecase
313313
log.Warnf("DROP status returned from app while processing pub/sub event %v", cloudEvent[contribpubsub.IDField])
314-
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, msg.pubsub, strings.ToLower(string(contribpubsub.Drop)), msg.topic, elapsed)
314+
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, msg.pubsub, strings.ToLower(string(contribpubsub.Drop)), strings.ToLower(string(contribpubsub.Success)), msg.topic, elapsed)
315315

316316
return rtpubsub.ErrMessageDropped
317317
}
318318

319319
// Consider unknown status field as error and retry
320-
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, msg.pubsub, strings.ToLower(string(contribpubsub.Retry)), msg.topic, elapsed)
320+
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, msg.pubsub, strings.ToLower(string(contribpubsub.Retry)), "", msg.topic, elapsed)
321321
return fmt.Errorf("unknown status returned from app while processing pub/sub event %v, status: %v, err: %w", cloudEvent[contribpubsub.IDField], res.GetStatus(), rterrors.NewRetriable(nil))
322322
}
323323

‎pkg/runtime/processor/pubsub/topics.go‎

Lines changed: 10 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -94,11 +94,11 @@ func (p *pubsub) subscribeTopic(name, topic string, route compstore.TopicRouteEl
9494
if route.DeadLetterTopic != "" {
9595
if dlqErr := p.sendToDeadLetter(ctx, name, msg, route.DeadLetterTopic); dlqErr == nil {
9696
// dlq has been configured and message is successfully sent to dlq.
97-
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, name, strings.ToLower(string(contribpubsub.Drop)), msgTopic, 0)
97+
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, name, strings.ToLower(string(contribpubsub.Drop)), "", msgTopic, 0)
9898
return nil
9999
}
100100
}
101-
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, name, strings.ToLower(string(contribpubsub.Retry)), msgTopic, 0)
101+
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, name, strings.ToLower(string(contribpubsub.Retry)), "", msgTopic, 0)
102102
return err
103103
}
104104

@@ -112,11 +112,11 @@ func (p *pubsub) subscribeTopic(name, topic string, route compstore.TopicRouteEl
112112
if route.DeadLetterTopic != "" {
113113
if dlqErr := p.sendToDeadLetter(ctx, name, msg, route.DeadLetterTopic); dlqErr == nil {
114114
// dlq has been configured and message is successfully sent to dlq.
115-
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, name, strings.ToLower(string(contribpubsub.Drop)), msgTopic, 0)
115+
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, name, strings.ToLower(string(contribpubsub.Drop)), "", msgTopic, 0)
116116
return nil
117117
}
118118
}
119-
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, name, strings.ToLower(string(contribpubsub.Retry)), msgTopic, 0)
119+
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, name, strings.ToLower(string(contribpubsub.Retry)), "", msgTopic, 0)
120120
return err
121121
}
122122
} else {
@@ -126,18 +126,18 @@ func (p *pubsub) subscribeTopic(name, topic string, route compstore.TopicRouteEl
126126
if route.DeadLetterTopic != "" {
127127
if dlqErr := p.sendToDeadLetter(ctx, name, msg, route.DeadLetterTopic); dlqErr == nil {
128128
// dlq has been configured and message is successfully sent to dlq.
129-
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, name, strings.ToLower(string(contribpubsub.Drop)), msgTopic, 0)
129+
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, name, strings.ToLower(string(contribpubsub.Drop)), "", msgTopic, 0)
130130
return nil
131131
}
132132
}
133-
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, name, strings.ToLower(string(contribpubsub.Retry)), msgTopic, 0)
133+
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, name, strings.ToLower(string(contribpubsub.Retry)), "", msgTopic, 0)
134134
return err
135135
}
136136
}
137137

138138
if contribpubsub.HasExpired(cloudEvent) {
139139
log.Warnf("dropping expired pub/sub event %v as of %v", cloudEvent[contribpubsub.IDField], cloudEvent[contribpubsub.ExpirationField])
140-
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, name, strings.ToLower(string(contribpubsub.Drop)), msgTopic, 0)
140+
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, name, strings.ToLower(string(contribpubsub.Drop)), "", msgTopic, 0)
141141

142142
if route.DeadLetterTopic != "" {
143143
_ = p.sendToDeadLetter(ctx, name, msg, route.DeadLetterTopic)
@@ -151,17 +151,17 @@ func (p *pubsub) subscribeTopic(name, topic string, route compstore.TopicRouteEl
151151
if route.DeadLetterTopic != "" {
152152
if dlqErr := p.sendToDeadLetter(ctx, name, msg, route.DeadLetterTopic); dlqErr == nil {
153153
// dlq has been configured and message is successfully sent to dlq.
154-
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, name, strings.ToLower(string(contribpubsub.Drop)), msgTopic, 0)
154+
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, name, strings.ToLower(string(contribpubsub.Drop)), "", msgTopic, 0)
155155
return nil
156156
}
157157
}
158-
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, name, strings.ToLower(string(contribpubsub.Retry)), msgTopic, 0)
158+
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, name, strings.ToLower(string(contribpubsub.Retry)), "", msgTopic, 0)
159159
return err
160160
}
161161
if !shouldProcess {
162162
// The event does not match any route specified so ignore it.
163163
log.Debugf("no matching route for event %v in pubsub %s and topic %s; skipping", cloudEvent[contribpubsub.IDField], name, msgTopic)
164-
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, name, strings.ToLower(string(contribpubsub.Drop)), msgTopic, 0)
164+
diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, name, strings.ToLower(string(contribpubsub.Drop)), strings.ToLower(string(contribpubsub.Success)), msgTopic, 0)
165165
if route.DeadLetterTopic != "" {
166166
_ = p.sendToDeadLetter(ctx, name, msg, route.DeadLetterTopic)
167167
}

‎tests/integration/suite/daprd/hotreload/operator/subscriptions/inflight.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -143,7 +143,7 @@ func (i *inflight) Run(t *testing.T, ctx context.Context) {
143143
}, time.Second*5, time.Millisecond*10)
144144

145145
egressMetric := fmt.Sprintf("dapr_component_pubsub_egress_count|app_id:%s|component:pubsub|namespace:|success:true|topic:a", i.daprd.AppID())
146-
ingressMetric := fmt.Sprintf("dapr_component_pubsub_ingress_count|app_id:%s|component:pubsub|namespace:|process_status:success|topic:a", i.daprd.AppID())
146+
ingressMetric := fmt.Sprintf("dapr_component_pubsub_ingress_count|app_id:%s|component:pubsub|namespace:|process_status:success|status:success|topic:a", i.daprd.AppID())
147147
metrics := i.daprd.Metrics(t, ctx)
148148
assert.Equal(t, 1, int(metrics[egressMetric]))
149149
assert.Equal(t, 0, int(metrics[ingressMetric]))

0 commit comments

Comments
 (0)