diff --git a/.github/workflows/pr-labels.yml b/.github/workflows/pr-labels.yml index ddeafcc46d..a288fc372d 100644 --- a/.github/workflows/pr-labels.yml +++ b/.github/workflows/pr-labels.yml @@ -6,7 +6,7 @@ jobs: runs-on: ubuntu-24.04 timeout-minutes: 5 steps: - - uses: actions/labeler@v6 + - uses: actions/labeler@v7 if: ${{ github.event.pull_request.head.repo.full_name == github.repository && !startsWith(github.actor, 'dependabot') }} with: repo-token: "${{ secrets.GITHUB_TOKEN }}" diff --git a/CHANGELOG.md b/CHANGELOG.md index b459342ae3..0d1427de2e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,9 +11,13 @@ For details about compatibility between different releases, see the **Commitment ### Added +- `gs_gateways_disconnected_total` metric, counting gateway disconnections by protocol and by the error the connection was disconnected with. This makes disconnection reasons (such as gateways disappearing without a close handshake, or missing too many pongs) observable as a rate, instead of only through logs. + ### Changed - In the Semtech UDP Packet Forwarder protocol, `PUSH_ACK` and `PULL_ACK` are only sent after the gateway has connected to the Gateway Server and the gateway's `PUSH_DATA` or `PULL_DATA` respectively has been accepted. +- Don't log a `Task failed` warning in GS for every task attached to a gateway connection when that connection is closed. The tasks that run for the lifetime of a gateway connection now stop without an error when the connection is closed, and the disconnection is logged once, as `Disconnected`, including the reason. This removes several duplicate warnings per gateway disconnection, which were particularly noisy for gateways on unreliable backhaul. +- Websocket close errors on the LoRa Basics Station frontend (such as `websocket: close 1006 (abnormal closure): unexpected EOF`, which is what a gateway disappearing without a close handshake looks like) are now reported as the defined error `pkg/gatewayserver/io/semtechws:websocket_closed`, with the close code as an attribute and the original error as the cause. As a result, the `gs.gateway.disconnect` event for these disconnections now carries structured error details instead of a plain string; consumers that parse the event data should expect the `ErrorDetails` format. ### Deprecated diff --git a/config/messages.json b/config/messages.json index 88e2d764e5..fbfa9f16ae 100644 --- a/config/messages.json +++ b/config/messages.json @@ -5273,6 +5273,15 @@ "file": "ws.go" } }, + "error:pkg/gatewayserver/io/semtechws:websocket_closed": { + "translations": { + "en": "websocket closed with code `{code}`" + }, + "description": { + "package": "pkg/gatewayserver/io/semtechws", + "file": "ws.go" + } + }, "error:pkg/gatewayserver/io/ttigw:downlink_channel_mixed_bandwidths": { "translations": { "en": "downlink channel `{channel}` has mixed bandwidths `{bandwidth_low}` and `{bandwidth_high}` Hz" diff --git a/pkg/gatewayserver/gatewayserver.go b/pkg/gatewayserver/gatewayserver.go index 69b7512025..4948bf2487 100644 --- a/pkg/gatewayserver/gatewayserver.go +++ b/pkg/gatewayserver/gatewayserver.go @@ -694,12 +694,19 @@ func (gs *GatewayServer) Connect( for name, handler := range gs.upstreamHandlers { connCtx := log.NewContextWithField(conn.Context(), "upstream_handler", name) - handler := handler gs.StartTask(&task.Config{ Context: connCtx, ID: fmt.Sprintf("%s_connect_gateway_%s", name, ids.GatewayId), Func: func(ctx context.Context) error { - return handler.ConnectGateway(ctx, ids, conn) + err := handler.ConnectGateway(ctx, ids, conn) + if err != nil && errors.Is(err, ctx.Err()) { + // Expected stop — the handler returned the error that the context was + // canceled with, so the gateway is disconnected. The reason is reported + // when the connection is torn down, not in this task. Any other error is + // a genuine failure and is reported, even when the context is done. + return nil + } + return err }, Done: wg.Done, Restart: task.RestartOnFailure, @@ -764,7 +771,9 @@ func (gs *GatewayServer) startDisconnectOnChangeTask(conn connectionEntry) { d := random.Jitter(gs.config.FetchGatewayInterval, gs.config.FetchGatewayJitter) select { case <-ctx.Done(): - return ctx.Err() + // Expected stop — the context is done, gateway is disconnected. The reason + // is reported when the connection is torn down, not in this task. + return nil case <-time.After(d): } @@ -991,7 +1000,7 @@ func (gs *GatewayServer) handleUpstream(ctx context.Context, conn connectionEntr defer func() { gs.connections.Delete(unique.ID(ctx, gtw.GetIds())) registerGatewayDisconnect(ctx, gtw.GetIds(), protocol, ctx.Err()) - logger.Info("Disconnected") + logger.WithError(ctx.Err()).Info("Disconnected") }() hosts := make([]*upstreamHost, 0, len(gs.upstreamHandlers)) diff --git a/pkg/gatewayserver/io/semtechws/ws.go b/pkg/gatewayserver/io/semtechws/ws.go index 204cd81bb6..dea9b2ceeb 100644 --- a/pkg/gatewayserver/io/semtechws/ws.go +++ b/pkg/gatewayserver/io/semtechws/ws.go @@ -58,8 +58,18 @@ var ( errGatewayID = errors.DefineInvalidArgument("invalid_gateway_id", "invalid gateway ID `{id}`") errNoAuthProvided = errors.DefineUnauthenticated("no_auth_provided", "no auth provided `{uid}`") errMissedTooManyPongs = errors.Define("missed_too_many_pongs", "gateway missed too many pongs") + errWebsocketClosed = errors.DefineAborted("websocket_closed", "websocket closed with code `{code}`") ) +// disconnectError converts err into the error that the connection is disconnected with. +func disconnectError(err error) error { + var closeErr *websocket.CloseError + if errors.As(err, &closeErr) { + return errWebsocketClosed.WithCause(err).WithAttributes("code", closeErr.Code) + } + return err +} + type srv struct { ctx context.Context server io.Server @@ -288,7 +298,7 @@ func (s *srv) handleTraffic(w http.ResponseWriter, r *http.Request) (err error) span.RecordError(err) span.SetStatus(codes.Error, "handle traffic failed") } - conn.Disconnect(err) + conn.Disconnect(disconnectError(err)) err = nil // Errors are sent over the websocket connection that is established by this point. }() @@ -346,7 +356,7 @@ func (s *srv) handleTraffic(w http.ResponseWriter, r *http.Request) (err error) defer ws.Close() defer func() { if err != nil { - conn.Disconnect(err) + conn.Disconnect(disconnectError(err)) } }() for { diff --git a/pkg/gatewayserver/io/semtechws/ws_internal_test.go b/pkg/gatewayserver/io/semtechws/ws_internal_test.go new file mode 100644 index 0000000000..046186868a --- /dev/null +++ b/pkg/gatewayserver/io/semtechws/ws_internal_test.go @@ -0,0 +1,87 @@ +// Copyright © 2026 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 semtechws + +import ( + "fmt" + "io" + "testing" + + "github.com/gorilla/websocket" + "github.com/smarty/assertions" + "go.thethings.network/lorawan-stack/v3/pkg/errors" + "go.thethings.network/lorawan-stack/v3/pkg/util/test/assertions/should" +) + +func TestDisconnectError(t *testing.T) { + t.Parallel() + + for _, tc := range []struct { + Name string + Err error + // Code is the expected close code, nil if the error is expected to pass through. + Code any + }{ + { + // This is what the gateway disappearing without a close handshake looks like. + Name: "AbnormalClosure", + Err: &websocket.CloseError{Code: websocket.CloseAbnormalClosure, Text: io.ErrUnexpectedEOF.Error()}, + Code: websocket.CloseAbnormalClosure, + }, + { + Name: "GoingAway", + Err: &websocket.CloseError{Code: websocket.CloseGoingAway}, + Code: websocket.CloseGoingAway, + }, + { + Name: "WrappedCloseError", + Err: fmt.Errorf("read: %w", + &websocket.CloseError{Code: websocket.CloseNoStatusReceived}, + ), + Code: websocket.CloseNoStatusReceived, + }, + { + Name: "DefinedError", + Err: errMissedTooManyPongs.New(), + }, + { + Name: "OtherError", + Err: io.ErrUnexpectedEOF, + }, + } { + t.Run(tc.Name, func(t *testing.T) { + t.Parallel() + a := assertions.New(t) + + err := disconnectError(tc.Err) + if tc.Code == nil { + a.So(err, should.Equal, tc.Err) + return + } + if !a.So(errors.IsAborted(err), should.BeTrue) { + t.FailNow() + } + ttnErr, ok := errors.From(err) + if !a.So(ok, should.BeTrue) { + t.FailNow() + } + // The error must be classifiable, so that it can be used as a metric label. + a.So(ttnErr.FullName(), should.Equal, "pkg/gatewayserver/io/semtechws:websocket_closed") + a.So(ttnErr.Attributes()["code"], should.Equal, tc.Code) + // The original error must be preserved for diagnostics. + a.So(errors.Is(err, tc.Err), should.BeTrue) + }) + } +} diff --git a/pkg/gatewayserver/observability.go b/pkg/gatewayserver/observability.go index 49fb120601..2c21004266 100644 --- a/pkg/gatewayserver/observability.go +++ b/pkg/gatewayserver/observability.go @@ -136,6 +136,14 @@ var gsMetrics = &messageMetrics{ }, []string{protocol}, ), + gatewaysDisconnected: metrics.NewContextualCounterVec( + prometheus.CounterOpts{ + Subsystem: subsystem, + Name: "gateways_disconnected_total", + Help: "Total number of gateway disconnections", + }, + []string{protocol, "error"}, + ), statusReceived: metrics.NewContextualCounterVec( prometheus.CounterOpts{ Subsystem: subsystem, @@ -256,6 +264,7 @@ func init() { type messageMetrics struct { gatewaysConnected *metrics.ContextualGaugeVec + gatewaysDisconnected *metrics.ContextualCounterVec statusReceived *metrics.ContextualCounterVec statusForwarded *metrics.ContextualCounterVec statusDropped *metrics.ContextualCounterVec @@ -274,6 +283,7 @@ type messageMetrics struct { func (m messageMetrics) Describe(ch chan<- *prometheus.Desc) { m.gatewaysConnected.Describe(ch) + m.gatewaysDisconnected.Describe(ch) m.statusReceived.Describe(ch) m.statusForwarded.Describe(ch) m.statusDropped.Describe(ch) @@ -292,6 +302,7 @@ func (m messageMetrics) Describe(ch chan<- *prometheus.Desc) { func (m messageMetrics) Collect(ch chan<- prometheus.Metric) { m.gatewaysConnected.Collect(ch) + m.gatewaysDisconnected.Collect(ch) m.statusReceived.Collect(ch) m.statusForwarded.Collect(ch) m.statusDropped.Collect(ch) @@ -320,6 +331,11 @@ func registerGatewayConnect( func registerGatewayDisconnect(ctx context.Context, ids *ttnpb.GatewayIdentifiers, protocol string, err error) { events.Publish(evtGatewayDisconnect.NewWithIdentifiersAndData(ctx, ids, err)) gsMetrics.gatewaysConnected.WithLabelValues(ctx, protocol).Dec() + errorLabel := unknown + if ttnErr, ok := errors.From(err); ok { + errorLabel = ttnErr.FullName() + } + gsMetrics.gatewaysDisconnected.WithLabelValues(ctx, protocol, errorLabel).Inc() } func registerGatewayConnectionStats(ctx context.Context, ids *ttnpb.GatewayIdentifiers, stats *ttnpb.GatewayConnectionStats) { diff --git a/pkg/task/task_test.go b/pkg/task/task_test.go new file mode 100644 index 0000000000..792fb42959 --- /dev/null +++ b/pkg/task/task_test.go @@ -0,0 +1,158 @@ +// Copyright © 2026 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 task_test + +import ( + "context" + "io" + "testing" + "time" + + "github.com/smarty/assertions" + "go.thethings.network/lorawan-stack/v3/pkg/errorcontext" + "go.thethings.network/lorawan-stack/v3/pkg/errors" + "go.thethings.network/lorawan-stack/v3/pkg/log" + "go.thethings.network/lorawan-stack/v3/pkg/log/handler/memory" + "go.thethings.network/lorawan-stack/v3/pkg/task" + "go.thethings.network/lorawan-stack/v3/pkg/util/test" + "go.thethings.network/lorawan-stack/v3/pkg/util/test/assertions/should" +) + +var errTest = errors.DefineAborted("test", "test error") + +func TestDefaultStartTaskLogging(t *testing.T) { + t.Parallel() + + for _, tc := range []struct { + Name string + // Context returns the context the task is started with. + Context func(context.Context) context.Context + // Func is the task function. + Func task.Func + // Message is the expected log message, empty if nothing is expected to be logged. + Message string + Level log.Level + }{ + { + Name: "Failure", + Context: func(ctx context.Context) context.Context { return ctx }, + Func: func(context.Context) error { return errTest.New() }, + Message: "Task failed", + Level: log.WarnLevel, + }, + { + Name: "Success", + Context: func(ctx context.Context) context.Context { return ctx }, + Func: func(context.Context) error { return nil }, + }, + { + Name: "EOF", + Context: func(ctx context.Context) context.Context { return ctx }, + Func: func(context.Context) error { return io.EOF }, + }, + { + Name: "ContextCanceled", + Context: func(ctx context.Context) context.Context { + ctx, cancel := context.WithCancel(ctx) + cancel() + return ctx + }, + Func: func(ctx context.Context) error { return ctx.Err() }, + }, + { + // Contexts from pkg/errorcontext return the cancelation cause instead of + // context.Canceled, and the task runner cannot tell that apart from a genuine + // failure. Tasks that stop because their context is done must return nil instead of + // the context error, or every cancelation is reported as a failure. + Name: "ErrorContextCanceled", + Context: func(ctx context.Context) context.Context { + ctx, cancel := errorcontext.New(ctx) + cancel(errTest.New()) + return ctx + }, + Func: func(ctx context.Context) error { return ctx.Err() }, + Message: "Task failed", + Level: log.WarnLevel, + }, + } { + t.Run(tc.Name, func(t *testing.T) { + t.Parallel() + a := assertions.New(t) + + handler := memory.New() + ctx := log.NewContext(test.Context(), log.NewLogger(handler, log.WithLevel(log.DebugLevel))) + + done := make(chan struct{}) + task.DefaultStartTask(&task.Config{ + Context: tc.Context(ctx), + ID: "test", + Func: tc.Func, + Done: func() { close(done) }, + Restart: task.RestartNever, + }) + + select { + case <-done: + case <-time.After(test.Delay << 10): + t.Fatal("Timed out waiting for the task to finish") + } + + var entries []log.Entry + for _, entry := range handler.Entries { + if entry.Message() == tc.Message { + entries = append(entries, entry) + } + } + if tc.Message == "" { + a.So(handler.Entries, should.BeEmpty) + return + } + if !a.So(entries, should.HaveLength, 1) { + t.FailNow() + } + a.So(entries[0].Level(), should.Equal, tc.Level) + }) + } +} + +// TestDefaultStartTaskRestartOnFailure ensures that a task whose context is done is not restarted, +// even though it stopped with an error. +func TestDefaultStartTaskRestartOnFailure(t *testing.T) { + t.Parallel() + a := assertions.New(t) + + ctx, cancel := errorcontext.New(test.Context()) + cancel(errTest.New()) + + invocations := 0 + done := make(chan struct{}) + task.DefaultStartTask(&task.Config{ + Context: ctx, + ID: "test", + Func: func(ctx context.Context) error { + invocations++ + return ctx.Err() + }, + Done: func() { close(done) }, + Restart: task.RestartOnFailure, + }) + + select { + case <-done: + case <-time.After(test.Delay << 10): + t.Fatal("Timed out waiting for the task to finish") + } + a.So(invocations, should.Equal, 1) +}