Merge pull request #5727 from TheThingsNetwork/fix/lr-fhss-rnd-fixes

Miscellaneous LR-FHSS fixes

api/api.md

@@ -5902,7 +5902,7 @@ Only the components for which the keys were meant, will have the key-encryption-
 | Field | Type | Label | Description |
 | ----- | ---- | ----- | ----------- |
 | `modulation_type` | [`uint32`](#uint32) |  |  |
-| `operating_channel_width` | [`uint32`](#uint32) |  | Operating Channel Width (kHz). |
+| `operating_channel_width` | [`uint32`](#uint32) |  | Operating Channel Width (Hz). |
 | `coding_rate` | [`string`](#string) |  |  |
 
 ### <a name="ttn.lorawan.v3.LoRaDataRate">Message `LoRaDataRate`</a>

api/api.swagger.json

@@ -22117,7 +22117,7 @@
         "operating_channel_width": {
           "type": "integer",
           "format": "int64",
-          "description": "Operating Channel Width (kHz)."
+          "description": "Operating Channel Width (Hz)."
         },
         "coding_rate": {
           "type": "string"

api/lorawan.proto

@@ -410,7 +410,7 @@ message FSKDataRate {
 message LRFHSSDataRate {
   option (thethings.flags.message) = { select: true, set: false };
   uint32 modulation_type = 1;
-  // Operating Channel Width (kHz).
+  // Operating Channel Width (Hz).
   uint32 operating_channel_width = 2;
   string coding_rate = 3;
 }

config/messages.json

@@ -4562,6 +4562,15 @@
       "file": "firewall.go"
     }
   },
+  "error:pkg/gatewayserver/io/udp:packet_type": {
+    "translations": {
+      "en": "invalid packet type"
+    },
+    "description": {
+      "package": "pkg/gatewayserver/io/udp",
+      "file": "udp.go"
+    }
+  },
   "error:pkg/gatewayserver/io/udp:rate_exceeded": {
     "translations": {
       "en": "gateway traffic exceeded allowed rate"

pkg/gatewayserver/io/udp/observability.go

@@ -0,0 +1,113 @@
+// Copyright © 2022 The Things Network Foundation, The Things Industries B.V.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package udp
+
+import (
+	"context"
+	"encoding/json"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"go.thethings.network/lorawan-stack/v3/pkg/errors"
+	"go.thethings.network/lorawan-stack/v3/pkg/metrics"
+	encoding "go.thethings.network/lorawan-stack/v3/pkg/ttnpb/udp"
+)
+
+const subsystem = "gs_io_udp"
+
+var udpMetrics = &messageMetrics{
+	messageReceived: prometheus.NewCounter(
+		prometheus.CounterOpts{
+			Subsystem: subsystem,
+			Name:      "message_received_total",
+			Help:      "Total number of received UDP messages",
+		},
+	),
+	messageForwarded: prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Subsystem: subsystem,
+			Name:      "message_forwarded_total",
+			Help:      "Total number of forwarded UDP messages",
+		},
+		[]string{"type"},
+	),
+	messageDropped: prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Subsystem: subsystem,
+			Name:      "message_dropped_total",
+			Help:      "Total number of dropped UDP messages",
+		},
+		[]string{"error"},
+	),
+
+	unmarshalTypeErrors: prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Subsystem: subsystem,
+			Name:      "unmarshal_type_errors_total",
+			Help:      "Total number of unmarshal type errors",
+		},
+		[]string{"value", "field"},
+	),
+}
+
+func init() {
+	metrics.MustRegister(udpMetrics)
+}
+
+type messageMetrics struct {
+	messageReceived  prometheus.Counter
+	messageForwarded *prometheus.CounterVec
+	messageDropped   *prometheus.CounterVec
+
+	unmarshalTypeErrors *prometheus.CounterVec
+}
+
+// Describe implements prometheus.Collector.
+func (m messageMetrics) Describe(ch chan<- *prometheus.Desc) {
+	m.messageReceived.Describe(ch)
+	m.messageForwarded.Describe(ch)
+	m.messageDropped.Describe(ch)
+
+	m.unmarshalTypeErrors.Describe(ch)
+}
+
+// Collect implements prometheus.Collector.
+func (m messageMetrics) Collect(ch chan<- prometheus.Metric) {
+	m.messageReceived.Collect(ch)
+	m.messageForwarded.Collect(ch)
+	m.messageDropped.Collect(ch)
+
+	m.unmarshalTypeErrors.Collect(ch)
+}
+
+func registerMessageReceived(_ context.Context) {
+	udpMetrics.messageReceived.Inc()
+}
+
+func registerMessageForwarded(_ context.Context, tp encoding.PacketType) {
+	udpMetrics.messageForwarded.WithLabelValues(tp.String()).Inc()
+}
+
+func registerMessageDropped(_ context.Context, err error) {
+	errorLabel := "unknown"
+	if ttnErr, ok := errors.From(err); ok {
+		errorLabel = ttnErr.FullName()
+	} else if jsonErr := (&json.SyntaxError{}); errors.Is(err, jsonErr) {
+		errorLabel = "encoding/json:syntax"
+	} else if jsonErr := (&json.UnmarshalTypeError{}); errors.Is(err, jsonErr) {
+		errorLabel = "encoding/json:unmarshal_type"
+		udpMetrics.unmarshalTypeErrors.WithLabelValues(jsonErr.Value, jsonErr.Field).Inc()
+	}
+	udpMetrics.messageDropped.WithLabelValues(errorLabel).Inc()
+}

pkg/gatewayserver/io/udp/udp.go

@@ -33,6 +33,7 @@ import (
 	"go.thethings.network/lorawan-stack/v3/pkg/types"
 	"go.thethings.network/lorawan-stack/v3/pkg/unique"
 	"go.thethings.network/lorawan-stack/v3/pkg/workerpool"
+	"golang.org/x/exp/slices"
 )
 
 type srv struct {
@@ -96,6 +97,8 @@ func Serve(ctx context.Context, server io.Server, conn *net.UDPConn, config Conf
 	return s.read(wp)
 }
 
+var errPacketType = errors.DefineInvalidArgument("packet_type", "invalid packet type")
+
 func (s *srv) read(wp workerpool.WorkerPool[encoding.Packet]) error {
 	var buf [65507]byte
 	for {
@@ -110,38 +113,45 @@ func (s *srv) read(wp workerpool.WorkerPool[encoding.Packet]) error {
 		ctx := log.NewContextWithField(s.ctx, "remote_addr", addr.String())
 		logger := log.FromContext(ctx)
 
+		registerMessageReceived(ctx)
 		if err := ratelimit.Require(s.server.RateLimiter(), ratelimit.GatewayUDPTrafficResource(addr)); err != nil {
 			if ratelimit.Require(s.limitLogs, ratelimit.NewCustomResource(addr.IP.String())) == nil {
 				logger.WithError(err).Warn("Drop packet")
 			}
+			registerMessageDropped(ctx, err)
 			continue
 		}
 
-		packetBuf := make([]byte, n)
-		copy(packetBuf, buf[:])
+		packetBuf := slices.Clone(buf[:n])
 
 		packet := encoding.Packet{
 			GatewayAddr: addr,
 			ReceivedAt:  now,
 		}
 		if err := packet.UnmarshalBinary(packetBuf); err != nil {
 			logger.WithError(err).Debug("Failed to unmarshal packet")
+			registerMessageDropped(ctx, err)
 			continue
 		}
 		switch packet.PacketType {
 		case encoding.PullData, encoding.PushData, encoding.TxAck:
 		default:
 			logger.WithField("packet_type", packet.PacketType).Debug("Invalid packet type for uplink")
+			registerMessageDropped(ctx, errPacketType)
 			continue
 		}
 		if packet.GatewayEUI == nil {
 			logger.Debug("No gateway EUI in uplink message")
+			registerMessageDropped(ctx, errNoEUI)
 			continue
 		}
 
 		if err := wp.Publish(ctx, packet); err != nil {
 			logger.WithError(err).Warn("UDP packet publishing failed")
+			registerMessageDropped(ctx, err)
+			continue
 		}
+		registerMessageForwarded(ctx, packet.PacketType)
 	}
 }
 

pkg/gatewayserver/observability.go

@@ -379,7 +379,7 @@ func registerSuccessDownlink(ctx context.Context, gtw *ttnpb.Gateway, protocol s
 }
 
 func registerFailDownlink(ctx context.Context, gtw *ttnpb.Gateway, txAck *ttnpb.TxAcknowledgment, protocol string) {
-	events.Publish(evtTxFailureDown.NewWithIdentifiersAndData(ctx, gtw, txAck.Result))
+	events.Publish(evtTxFailureDown.NewWithIdentifiersAndData(ctx, gtw, txAck))
 	gsMetrics.downlinkTxFailed.WithLabelValues(ctx, protocol, txAck.Result.String()).Inc()
 }
 

pkg/networkserver/mac/adr_param_setup_test.go

@@ -214,7 +214,7 @@ func TestHandleADRParamSetupAns(t *testing.T) {
 			Events: events.Builders{
 				EvtReceiveADRParamSetupAnswer,
 			},
-			Error: ErrRequestNotFound,
+			Error: ErrRequestNotFound.WithAttributes("cid", ttnpb.MACCommandIdentifier_CID_ADR_PARAM_SETUP),
 		},
 		{
 			Name: "limit 32768, delay 1024",

pkg/networkserver/mac/beacon_freq_test.go

@@ -148,7 +148,7 @@ func TestHandleBeaconFreqAns(t *testing.T) {
 					FrequencyAck: true,
 				})),
 			},
-			Error: ErrRequestNotFound,
+			Error: ErrRequestNotFound.WithAttributes("cid", ttnpb.MACCommandIdentifier_CID_BEACON_FREQ),
 		},
 		{
 			Name: "nack/no request",
@@ -168,7 +168,7 @@ func TestHandleBeaconFreqAns(t *testing.T) {
 			Events: events.Builders{
 				EvtReceiveBeaconFreqReject.With(events.WithData(&ttnpb.MACCommand_BeaconFreqAns{})),
 			},
-			Error: ErrRequestNotFound,
+			Error: ErrRequestNotFound.WithAttributes("cid", ttnpb.MACCommandIdentifier_CID_BEACON_FREQ),
 		},
 		{
 			Name: "ack/valid request",

pkg/networkserver/mac/dev_status_test.go

@@ -224,7 +224,7 @@ func TestHandleDevStatusAns(t *testing.T) {
 					Margin:  4,
 				})),
 			},
-			Error: ErrRequestNotFound,
+			Error: ErrRequestNotFound.WithAttributes("cid", ttnpb.MACCommandIdentifier_CID_DEV_STATUS),
 		},
 		{
 			Name: "battery 42%/margin 4",

pkg/networkserver/mac/dl_channel_test.go

@@ -384,7 +384,7 @@ func TestHandleDLChannelAns(t *testing.T) {
 					ChannelIndexAck: true,
 				})),
 			},
-			Error: ErrRequestNotFound,
+			Error: ErrRequestNotFound.WithAttributes("cid", ttnpb.MACCommandIdentifier_CID_DL_CHANNEL),
 		},
 		{
 			Name: "frequency nack/channel index ack/no request",
@@ -408,7 +408,7 @@ func TestHandleDLChannelAns(t *testing.T) {
 					ChannelIndexAck: true,
 				})),
 			},
-			Error: ErrRequestNotFound,
+			Error: ErrRequestNotFound.WithAttributes("cid", ttnpb.MACCommandIdentifier_CID_DL_CHANNEL),
 		},
 		{
 			Name: "frequency nack/channel index nack/valid request/no rejections",

pkg/networkserver/mac/duty_cycle_test.go

@@ -106,7 +106,7 @@ func TestHandleDutyCycleAns(t *testing.T) {
 			Events: events.Builders{
 				EvtReceiveDutyCycleAnswer,
 			},
-			Error: ErrRequestNotFound,
+			Error: ErrRequestNotFound.WithAttributes("cid", ttnpb.MACCommandIdentifier_CID_DUTY_CYCLE),
 		},
 		{
 			Name: "2048",

pkg/networkserver/mac/errors.go

@@ -19,6 +19,8 @@ import (
 )
 
 var (
-	ErrRequestNotFound = errors.DefineInvalidArgument("request_not_found", "MAC response received, but corresponding request not found")
-	ErrNoPayload       = errors.DefineInvalidArgument("no_payload", "no message payload specified")
+	ErrRequestNotFound = errors.DefineInvalidArgument(
+		"request_not_found", "MAC response received, but corresponding request not found", "cid",
+	)
+	ErrNoPayload = errors.DefineInvalidArgument("no_payload", "no message payload specified")
 )

pkg/networkserver/mac/link_adr.go

@@ -141,11 +141,6 @@ func generateLinkADRReq(ctx context.Context, dev *ttnpb.EndDevice, phy *band.Ban
 			"desired_adr_data_rate_index", dev.MacState.DesiredParameters.AdrDataRateIndex,
 		}
 		switch {
-		case dev.MacState.DesiredParameters.AdrDataRateIndex > phy.MaxADRDataRateIndex:
-			return linkADRReqParameters{}, false, internal.ErrCorruptedMACState.
-				WithAttributes(append(attributes,
-					"phy_max_adr_data_rate_index", phy.MaxADRDataRateIndex,
-				)...)
 		case dev.MacState.DesiredParameters.AdrDataRateIndex < minDataRateIndex:
 			return linkADRReqParameters{}, false, internal.ErrCorruptedMACState.
 				WithAttributes(append(attributes,

pkg/networkserver/mac/link_adr_test.go

@@ -481,7 +481,7 @@ func TestHandleLinkADRAns(t *testing.T) {
 					TxPowerIndexAck:  true,
 				})),
 			},
-			Error: ErrRequestNotFound,
+			Error: ErrRequestNotFound.WithAttributes("cid", ttnpb.MACCommandIdentifier_CID_LINK_ADR),
 		},
 		{
 			Name: "1 request/all ack",

pkg/networkserver/mac/mac.go

@@ -1,4 +1,4 @@
-// Copyright © 2019 The Things Network Foundation, The Things Industries B.V.
+// Copyright © 2022 The Things Network Foundation, The Things Industries B.V.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -69,7 +69,7 @@ func handleMACResponse(
 	if allowMissing {
 		return cmds, nil
 	}
-	return cmds, ErrRequestNotFound.New()
+	return cmds, ErrRequestNotFound.WithAttributes("cid", cid)
 }
 
 // handleMACResponse searches for first MAC command block in cmds with CID equal to cid and calls f for each found value as argument.
@@ -79,7 +79,8 @@ func handleMACResponse(
 func handleMACResponseBlock(
 	cid ttnpb.MACCommandIdentifier,
 	allowMissing bool,
-	f func(*ttnpb.MACCommand) error, cmds ...*ttnpb.MACCommand,
+	f func(*ttnpb.MACCommand) error,
+	cmds ...*ttnpb.MACCommand,
 ) ([]*ttnpb.MACCommand, error) {
 	first := -1
 	last := -1
@@ -90,12 +91,14 @@ outer:
 
 		switch {
 		case first >= 0 && cmd.Cid != cid:
+			last--
 			break outer
 		case first < 0 && cmd.Cid != cid:
 			continue
 		case first < 0:
 			first = i
 		}
+
 		if err := f(cmd); err != nil {
 			return cmds, err
 		}
@@ -104,7 +107,7 @@ outer:
 	case first < 0 && allowMissing:
 		return cmds, nil
 	case first < 0 && !allowMissing:
-		return cmds, ErrRequestNotFound.New()
+		return cmds, ErrRequestNotFound.WithAttributes("cid", cid)
 	default:
 		return append(cmds[:first], cmds[last+1:]...), nil
 	}