Merge pull request #7080 from TheThingsNetwork/feature/upgrade-thrott…

…led-usage Upgrade throttled usage
TheThingsNetwork · May 15, 2024 · 2457908 · 2457908
2 parents 5d83cbd + b6ce6e2
commit 2457908
Show file tree

Hide file tree

Showing 3 changed files with 24 additions and 10 deletions.
diff --git a/pkg/applicationserver/observability.go b/pkg/applicationserver/observability.go
@@ -308,7 +308,7 @@ func registerDropUp(ctx context.Context, msg *ttnpb.ApplicationUp, err error) {
 func registerUplinkLatency(ctx context.Context, msg *ttnpb.ApplicationUplink) {
 	asMetrics.nsAsUplinkLatency.WithLabelValues(ctx).Observe(time.Since(*ttnpb.StdTime(msg.ReceivedAt)).Seconds())
 	for _, meta := range msg.RxMetadata {
-		if stdTime := ttnpb.StdTime(meta.Time); meta.Time != nil && !stdTime.IsZero() {
+		if stdTime := ttnpb.StdTime(meta.ReceivedAt); stdTime != nil && !stdTime.IsZero() {
 			asMetrics.gtwAsUplinkLatency.WithLabelValues(ctx).Observe(time.Since(*stdTime).Seconds())
 		}
 	}

diff --git a/pkg/ratelimit/rate_limit.go b/pkg/ratelimit/rate_limit.go
@@ -16,6 +16,7 @@ package ratelimit
 
 import (
 	"context"
+	"fmt"
 
 	"github.com/throttled/throttled/v2"
 	"go.thethings.network/lorawan-stack/v3/pkg/log"
@@ -40,18 +41,20 @@ func (*NoopRateLimiter) RateLimit(Resource) (bool, Result) {
 
 type rateLimiter struct {
 	ctx     context.Context
-	limiter throttled.RateLimiter
+	limiter throttled.RateLimiterCtx
 }
 
 // RateLimit implements ratelimit.Interface.
 func (l *rateLimiter) RateLimit(resource Resource) (bool, Result) {
-	ok, result, err := l.limiter.RateLimit(resource.Key(), 1)
+	ok, result, err := l.limiter.RateLimitCtx(l.ctx, resource.Key(), 1)
 	if err != nil {
-		// NOTE: The memstore.MemStore implementation does not fail.
-		log.FromContext(l.ctx).Error("Rate limiter failed")
-		return true, Result{}
+		log.FromContext(l.ctx).
+			WithError(err).
+			WithFields(logFields(resource)).
+			Error("Rate limiter failed")
+		// NOTE: The missing return here is intentional - the underlying implementation
+		// will still return a valid ok and result even on error.
 	}
-
 	return ok, Result{
 		Limit:      result.Limit,
 		Remaining:  result.Remaining,
@@ -94,3 +97,14 @@ func Require(limiter Interface, resource Resource) error {
 type RateLimitKeyer interface {
 	RateLimitKey() string
 }
+
+func logFields(resource Resource) log.Fielder {
+	fields := append(
+		make([]any, 0, 2+2*len(resource.Classes())),
+		"resource_key", resource.Key(),
+	)
+	for i, c := range resource.Classes() {
+		fields = append(fields, fmt.Sprintf("resource_class_%d", i), c)
+	}
+	return log.Fields(fields...)
+}
diff --git a/pkg/ratelimit/resource.go b/pkg/ratelimit/resource.go
@@ -115,7 +115,7 @@ func grpcStreamAcceptResource(ctx context.Context, fullMethod string) Resource {
 // grpcStreamUpResource represents client messages for a gRPC server stream.
 func grpcStreamUpResource(ctx context.Context, fullMethod string) Resource {
 	return &resource{
-		key:     fmt.Sprintf("grpc:stream:up:%s:streamID:%s", fullMethod, events.NewCorrelationID()),
+		key:     fmt.Sprintf("grpc:stream:up:%s:id:%s", fullMethod, events.NewCorrelationID()),
 		classes: []string{fmt.Sprintf("grpc:stream:up:%s", fullMethod), "grpc:stream:up"},
 	}
 }
@@ -140,10 +140,10 @@ func GatewayAcceptMQTTConnectionResource(remoteAddr string) Resource {
 	}
 }
 
-// GatewayUDPTrafficResource represents UDP gateway traffic from a remote IP address.
+// GatewayUDPTrafficResource represents UDP gateway traffic from a remote address.
 func GatewayUDPTrafficResource(addr *net.UDPAddr) Resource {
 	return &resource{
-		key:     fmt.Sprintf("gs:up:udp:ip:%s", addr.IP.String()),
+		key:     fmt.Sprintf("gs:up:udp:addr:%s", addr.String()),
 		classes: []string{"gs:up:udp", "gs:up"},
 	}
 }