Merge pull request #2266 from TheThingsNetwork/release/v3.6.3

Release v3.6.3

CHANGELOG.md

@@ -19,6 +19,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Security
 
+## [3.6.3] (2020-03-30)
+
+### Fixed
+
+-  Limited throughput in upstream handlers in Gateway Server when one gateway's upstream handler is busy.
+
 ## [3.6.2] (2020-03-19)
 
 ### Fixed
@@ -678,7 +684,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 NOTE: These links should respect backports. See https://github.com/TheThingsNetwork/lorawan-stack/pull/1444/files#r333379706.
 -->
 
-[unreleased]: https://github.com/TheThingsNetwork/lorawan-stack/compare/v3.6.2...HEAD
+[unreleased]: https://github.com/TheThingsNetwork/lorawan-stack/compare/v3.6.3...HEAD
+[3.6.3]: https://github.com/TheThingsNetwork/lorawan-stack/compare/v3.6.2...v3.6.3
 [3.6.2]: https://github.com/TheThingsNetwork/lorawan-stack/compare/v3.6.1...v3.6.2
 [3.6.1]: https://github.com/TheThingsNetwork/lorawan-stack/compare/v3.6.0...v3.6.1
 [3.6.0]: https://github.com/TheThingsNetwork/lorawan-stack/compare/v3.5.3...v3.6.0

cmd/internal/shared/init.go

@@ -21,7 +21,7 @@ import (
 )
 
 // Initialize global packages.
-func Initialize(ctx context.Context, config config.ServiceBase) error {
+func Initialize(ctx context.Context, config *config.ServiceBase) error {
 	// Fallback to the default Redis configuration for the cache system
 	if config.Cache.Redis.IsZero() {
 		config.Cache.Redis = config.Redis
@@ -31,7 +31,7 @@ func Initialize(ctx context.Context, config config.ServiceBase) error {
 		config.Events.Redis = config.Redis
 	}
 
-	if err := InitializeEvents(ctx, config); err != nil {
+	if err := InitializeEvents(ctx, *config); err != nil {
 		return err
 	}
 	return nil

cmd/ttn-lw-cli/commands/simulate.go

@@ -95,7 +95,7 @@ func (m *simulateMetadataParams) setDefaults() error {
 		m.SpreadingFactor, m.Bandwidth = lora.SpreadingFactor, lora.Bandwidth
 	} else if m.DataRateIndex == 0 {
 		for i, dr := range phy.DataRates {
-			if dr.Rate.GetLoRa().SpreadingFactor == m.SpreadingFactor && dr.Rate.GetLoRa().Bandwidth == m.Bandwidth {
+			if lora := dr.Rate.GetLoRa(); lora != nil && lora.SpreadingFactor == m.SpreadingFactor && lora.Bandwidth == m.Bandwidth {
 				m.DataRateIndex = uint32(i)
 				break
 			}

cmd/ttn-lw-stack/commands/root.go

@@ -84,7 +84,7 @@ var (
 			ctx = log.NewContext(ctx, logger)
 
 			// initialize shared packages
-			if err := shared.Initialize(ctx, config.ServiceBase); err != nil {
+			if err := shared.Initialize(ctx, &config.ServiceBase); err != nil {
 				return err
 			}
 

config/messages.json

@@ -3392,6 +3392,15 @@
       "file": "format_protobufv2.go"
     }
   },
+  "error:pkg/gatewayserver/io/mqtt:mqtt_frontend_recovered": {
+    "translations": {
+      "en": "internal server error"
+    },
+    "description": {
+      "package": "pkg/gatewayserver/io/mqtt",
+      "file": "mqtt.go"
+    }
+  },
   "error:pkg/gatewayserver/io/mqtt:not_authorized": {
     "translations": {
       "en": "not authorized"
@@ -3491,6 +3500,15 @@
       "file": "firewall_ratelimit.go"
     }
   },
+  "error:pkg/gatewayserver/io/udp:udp_frontend_recovered": {
+    "translations": {
+      "en": "internal server error"
+    },
+    "description": {
+      "package": "pkg/gatewayserver/io/udp",
+      "file": "udp.go"
+    }
+  },
   "error:pkg/gatewayserver/io:buffer_full": {
     "translations": {
       "en": "buffer is full"
@@ -3770,6 +3788,15 @@
       "file": "gatewayserver.go"
     }
   },
+  "error:pkg/gatewayserver:handler_recovered": {
+    "translations": {
+      "en": "internal server error"
+    },
+    "description": {
+      "package": "pkg/gatewayserver",
+      "file": "gatewayserver.go"
+    }
+  },
   "error:pkg/gatewayserver:host_handle": {
     "translations": {
       "en": "host `{host}` failed to handle message"

doc/config.toml

@@ -27,7 +27,7 @@ pygmentsUseClasses = true
   keywords = []
   github_repository = "https://github.com/TheThingsNetwork/lorawan-stack"
   github_repository_edit = "https://github.com/TheThingsNetwork/lorawan-stack/edit/master/doc/content"
-  version = "v3.6.2"
+  version = "v3.6.3"
 
 [markup]
   [markup.goldmark]

doc/themes/the-things-stack/package.json

@@ -1,6 +1,6 @@
 {
   "name": "hugo-theme-the-things-stack",
-  "version": "3.6.2",
+  "version": "3.6.3",
   "private": true,
   "description": "Hugo Theme for The Things Stack",
   "dependencies": {

package.json

@@ -1,6 +1,6 @@
 {
   "name": "ttn-stack",
-  "version": "3.6.2",
+  "version": "3.6.3",
   "description": "The Things Stack",
   "main": "index.js",
   "repository": "https://github.com/TheThingsNetwork/lorawan-stack.git",

pkg/gatewayserver/gatewayserver.go

@@ -20,6 +20,8 @@ import (
 	"fmt"
 	"net"
 	"net/http"
+	"os"
+	"runtime/debug"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -480,8 +482,6 @@ var (
 	maxUpstreamHandlers = int32(1 << 5)
 	// upstreamHandlerIdleTimeout is the duration after which an idle upstream handler stops to save resources.
 	upstreamHandlerIdleTimeout = (1 << 7) * time.Millisecond
-	// upstreamHandlerBusyTimeout is the duration after traffic gets dropped if all upstream handlers are busy.
-	upstreamHandlerBusyTimeout = (1 << 6) * time.Millisecond
 )
 
 type upstreamHost struct {
@@ -511,6 +511,7 @@ func (gs *GatewayServer) handleUpstream(conn connectionEntry) {
 	}()
 
 	handleFn := func(host *upstreamHost) {
+		defer recoverHandler(ctx)
 		defer host.handleWg.Done()
 		defer atomic.AddInt32(&host.handlers, -1)
 		for {
@@ -579,7 +580,7 @@ func (gs *GatewayServer) handleUpstream(conn connectionEntry) {
 			}
 			return false
 		}
-		hosts = append(hosts, &upstreamHost{
+		host := &upstreamHost{
 			name: name,
 			handler: func(ids *ttnpb.EndDeviceIdentifiers) upstream.Handler {
 				if ids != nil && ids.DevAddr != nil && !passDevAddr(handler.GetDevAddrPrefixes(), *ids.DevAddr) {
@@ -588,10 +589,8 @@ func (gs *GatewayServer) handleUpstream(conn connectionEntry) {
 				return handler
 			},
 			handleCh: make(chan upstreamItem),
-		})
-	}
-
-	for _, host := range hosts {
+		}
+		hosts = append(hosts, host)
 		defer host.handleWg.Wait()
 	}
 
@@ -632,17 +631,17 @@ func (gs *GatewayServer) handleUpstream(conn connectionEntry) {
 					atomic.AddInt32(&host.handlers, 1)
 					host.handleWg.Add(1)
 					go handleFn(host)
+					go func(host *upstreamHost) {
+						host.handleCh <- item
+					}(host)
+					continue
 				}
-				select {
-				case host.handleCh <- item:
-				case <-time.After(upstreamHandlerBusyTimeout):
-					logger.WithField("name", host.name).Warn("Upstream handler busy, drop message")
-					switch msg := val.(type) {
-					case *ttnpb.UplinkMessage:
-						registerFailUplink(ctx, conn.Gateway(), msg, host.name)
-					case *ttnpb.GatewayStatus:
-						registerFailStatus(ctx, conn.Gateway(), msg, host.name)
-					}
+				logger.WithField("name", host.name).Warn("Upstream handler busy, drop message")
+				switch msg := val.(type) {
+				case *ttnpb.UplinkMessage:
+					registerFailUplink(ctx, conn.Gateway(), msg, host.name)
+				case *ttnpb.GatewayStatus:
+					registerFailStatus(ctx, conn.Gateway(), msg, host.name)
 				}
 			}
 		}
@@ -756,3 +755,21 @@ func (gs *GatewayServer) GetMQTTConfig(ctx context.Context) (*config.MQTT, error
 	}
 	return &config.MQTT, nil
 }
+
+var errHandlerRecovered = errors.DefineInternal("handler_recovered", "internal server error")
+
+func recoverHandler(ctx context.Context) error {
+	if p := recover(); p != nil {
+		fmt.Fprintln(os.Stderr, p)
+		os.Stderr.Write(debug.Stack())
+		var err error
+		if pErr, ok := p.(error); ok {
+			err = errHandlerRecovered.WithCause(pErr)
+		} else {
+			err = errHandlerRecovered.WithAttributes("panic", p)
+		}
+		log.FromContext(ctx).WithError(err).Error("Handler failed")
+		return err
+	}
+	return nil
+}

pkg/gatewayserver/gatewayserver_internal_test.go

@@ -14,10 +14,4 @@
 
 package gatewayserver
 
-var (
-	ErrSchedule = errSchedule
-)
-
-func init() {
-	maxUpstreamHandlers = 1
-}
+var ErrSchedule = errSchedule

pkg/gatewayserver/gatewayserver_test.go

@@ -671,7 +671,7 @@ func TestGatewayServer(t *testing.T) {
 					for _, tc := range []struct {
 						Name     string
 						Up       *ttnpb.GatewayUp
-						Forwards []int // Indices of uplink messages in Up that are being forwarded.
+						Forwards []uint32 // Timestamps of uplink messages in Up that are being forwarded.
 					}{
 						{
 							Name: "GatewayStatus",
@@ -705,12 +705,12 @@ func TestGatewayServer(t *testing.T) {
 											},
 											CodingRate: "4/5",
 											Frequency:  867900000,
-											Timestamp:  4242000,
+											Timestamp:  100,
 										},
 										RxMetadata: []*ttnpb.RxMetadata{
 											{
 												GatewayIdentifiers: ids,
-												Timestamp:          4242000,
+												Timestamp:          100,
 												RSSI:               -69,
 												ChannelRSSI:        -69,
 												SNR:                11,
@@ -721,7 +721,7 @@ func TestGatewayServer(t *testing.T) {
 									},
 								},
 							},
-							Forwards: []int{0},
+							Forwards: []uint32{100},
 						},
 						{
 							Name: "OneValidFSK",
@@ -737,12 +737,12 @@ func TestGatewayServer(t *testing.T) {
 												},
 											},
 											Frequency: 867900000,
-											Timestamp: 4242000,
+											Timestamp: 100,
 										},
 										RxMetadata: []*ttnpb.RxMetadata{
 											{
 												GatewayIdentifiers: ids,
-												Timestamp:          4242000,
+												Timestamp:          100,
 												RSSI:               -69,
 												ChannelRSSI:        -69,
 												SNR:                11,
@@ -753,7 +753,7 @@ func TestGatewayServer(t *testing.T) {
 									},
 								},
 							},
-							Forwards: []int{0},
+							Forwards: []uint32{100},
 						},
 						{
 							Name: "OneGarbageWithStatus",
@@ -771,12 +771,12 @@ func TestGatewayServer(t *testing.T) {
 											},
 											CodingRate: "4/5",
 											Frequency:  868500000,
-											Timestamp:  1234560000,
+											Timestamp:  100,
 										},
 										RxMetadata: []*ttnpb.RxMetadata{
 											{
 												GatewayIdentifiers: ids,
-												Timestamp:          1234560000,
+												Timestamp:          100,
 												RSSI:               -112,
 												ChannelRSSI:        -112,
 												SNR:                2,
@@ -797,12 +797,12 @@ func TestGatewayServer(t *testing.T) {
 											},
 											CodingRate: "4/5",
 											Frequency:  868100000,
-											Timestamp:  4242000,
+											Timestamp:  200,
 										},
 										RxMetadata: []*ttnpb.RxMetadata{
 											{
 												GatewayIdentifiers: ids,
-												Timestamp:          4242000,
+												Timestamp:          200,
 												RSSI:               -69,
 												ChannelRSSI:        -69,
 												SNR:                11,
@@ -823,12 +823,12 @@ func TestGatewayServer(t *testing.T) {
 											},
 											CodingRate: "4/5",
 											Frequency:  867700000,
-											Timestamp:  2424000,
+											Timestamp:  300,
 										},
 										RxMetadata: []*ttnpb.RxMetadata{
 											{
 												GatewayIdentifiers: ids,
-												Timestamp:          2424000,
+												Timestamp:          300,
 												RSSI:               -36,
 												ChannelRSSI:        -36,
 												SNR:                5,
@@ -845,7 +845,7 @@ func TestGatewayServer(t *testing.T) {
 									Time: time.Unix(4242424, 0),
 								},
 							},
-							Forwards: []int{1, 2},
+							Forwards: []uint32{200, 300},
 						},
 					} {
 						t.Run(tc.Name, func(t *testing.T) {
@@ -879,10 +879,27 @@ func TestGatewayServer(t *testing.T) {
 								uplinkCount += len(tc.Up.UplinkMessages)
 							}
 
-							for _, msgIdx := range tc.Forwards {
+							notSeen := make(map[uint32]struct{})
+							for _, t := range tc.Forwards {
+								notSeen[t] = struct{}{}
+							}
+							for len(notSeen) > 0 {
 								select {
 								case msg := <-ns.Up():
-									expected := tc.Up.UplinkMessages[msgIdx]
+									var expected *ttnpb.UplinkMessage
+									for _, up := range tc.Up.UplinkMessages {
+										if ts := up.Settings.Timestamp; ts == msg.Settings.Timestamp {
+											if _, ok := notSeen[ts]; !ok {
+												t.Fatalf("Not expecting message %v", msg)
+											}
+											expected = up
+											delete(notSeen, ts)
+											break
+										}
+									}
+									if expected == nil {
+										t.Fatalf("Received unexpected message")
+									}
 									a.So(time.Since(msg.ReceivedAt), should.BeLessThan, timeout)
 									a.So(msg.Settings, should.Resemble, expected.Settings)
 									for _, md := range msg.RxMetadata {