diff --git a/src/compute-plane-services/worker-utils/service/BUILD.bazel b/src/compute-plane-services/worker-utils/service/BUILD.bazel index e334f3abc..bf374b42d 100644 --- a/src/compute-plane-services/worker-utils/service/BUILD.bazel +++ b/src/compute-plane-services/worker-utils/service/BUILD.bazel @@ -2,7 +2,10 @@ load("@rules_go//go:def.bzl", "go_library", "go_test") go_library( name = "service", - srcs = ["service.go"], + srcs = [ + "service.go", + "shutdown.go", + ], importpath = "github.com/NVIDIA/nvcf/src/compute-plane-services/worker-utils/service", visibility = ["//visibility:public"], deps = [ @@ -28,6 +31,7 @@ go_test( "main_test.go", "service_command_test.go", "service_test.go", + "shutdown_test.go", ], embed = [":service"], # keep: binds the upstream-hardcoded fixed ports (asset 8001, S3 8002, NVCF diff --git a/src/compute-plane-services/worker-utils/service/service.go b/src/compute-plane-services/worker-utils/service/service.go index f069e5ba8..c6a9f496b 100644 --- a/src/compute-plane-services/worker-utils/service/service.go +++ b/src/compute-plane-services/worker-utils/service/service.go @@ -58,7 +58,7 @@ func Run() { } } err := NewRootCommand(context.Background(), logger).Execute() - if err != nil && err.Error() != "received signal interrupt" { + if isFatalRunError(err) { utils.ExitReason(err) zap.S().Panic(err) } diff --git a/src/compute-plane-services/worker-utils/service/shutdown.go b/src/compute-plane-services/worker-utils/service/shutdown.go new file mode 100644 index 000000000..3173db712 --- /dev/null +++ b/src/compute-plane-services/worker-utils/service/shutdown.go @@ -0,0 +1,26 @@ +/* +SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +SPDX-License-Identifier: Apache-2.0 + +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 service + +import "github.com/NVIDIA/nvcf/src/compute-plane-services/worker-utils/worker" + +// isFatalRunError reports whether a terminal error from the root command is a +// genuine failure rather than the expected response to a shutdown signal. +func isFatalRunError(err error) bool { + return err != nil && !worker.IsShutdownSignalError(err) +} diff --git a/src/compute-plane-services/worker-utils/service/shutdown_test.go b/src/compute-plane-services/worker-utils/service/shutdown_test.go new file mode 100644 index 000000000..794aa75c1 --- /dev/null +++ b/src/compute-plane-services/worker-utils/service/shutdown_test.go @@ -0,0 +1,54 @@ +/* +SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +SPDX-License-Identifier: Apache-2.0 + +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 service + +import ( + "errors" + "testing" +) + +// TestIsFatalRunError_SIGTERM is the regression test for Run() panicking on a +// normal container stop. Kubernetes stops containers with SIGTERM, which the +// servers run group surfaces as "received signal terminated"; only the SIGINT +// wording used to be excused, so every graceful shutdown panicked. +func TestIsFatalRunError_SIGTERM(t *testing.T) { + if isFatalRunError(errors.New("received signal terminated")) { + t.Fatal("SIGTERM shutdown must not be treated as a fatal run error") + } +} + +func TestIsFatalRunError(t *testing.T) { + tests := []struct { + name string + err error + want bool + }{ + {name: "clean exit", err: nil, want: false}, + {name: "sigterm", err: errors.New("received signal terminated"), want: false}, + {name: "sigint", err: errors.New("received signal interrupt"), want: false}, + {name: "startup failure", err: errors.New("inference container not ready"), want: true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := isFatalRunError(tt.err); got != tt.want { + t.Fatalf("isFatalRunError(%v) = %v, want %v", tt.err, got, tt.want) + } + }) + } +} diff --git a/src/compute-plane-services/worker-utils/worker/BUILD.bazel b/src/compute-plane-services/worker-utils/worker/BUILD.bazel index 41c99508c..924dafdfc 100644 --- a/src/compute-plane-services/worker-utils/worker/BUILD.bazel +++ b/src/compute-plane-services/worker-utils/worker/BUILD.bazel @@ -6,6 +6,7 @@ go_library( "cancel.go", "large.go", "refresh.go", + "shutdown.go", "stateful.go", "work.go", "worker.go", @@ -67,6 +68,7 @@ go_test( "large_helpers_test.go", "main_test.go", "newworker_test.go", + "shutdown_test.go", "work_helpers_test.go", "work_test.go", "worker_extra_test.go", diff --git a/src/compute-plane-services/worker-utils/worker/shutdown.go b/src/compute-plane-services/worker-utils/worker/shutdown.go new file mode 100644 index 000000000..899dedf2c --- /dev/null +++ b/src/compute-plane-services/worker-utils/worker/shutdown.go @@ -0,0 +1,67 @@ +/* +SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +SPDX-License-Identifier: Apache-2.0 + +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 worker + +import ( + "os" + "strings" + "syscall" +) + +// signalErrorPrefix is how the nvkit servers run group formats the terminal +// error it returns when a shutdown signal arrives. Keep in sync with the +// signal actor in pkg/nvkit/servers/grpc.go, which returns +// fmt.Errorf("received signal %s", sig). +const signalErrorPrefix = "received signal " + +// shutdownSignals are the signals that run group installs a handler for. +// SIGTERM is how Kubernetes asks a container to stop, so it is the signal the +// worker actually sees in production; SIGINT only shows up in local runs. +var shutdownSignals = []os.Signal{syscall.SIGINT, syscall.SIGTERM} + +// IsShutdownSignalError reports whether err is the run group reporting a +// graceful shutdown signal rather than a server failure. +// +// The servers package models a shutdown signal as a terminal error, so an +// expected stop and a genuine crash both reach callers as ordinary errors and +// can only be told apart by inspecting the message. Matching is derived from +// the signal names rather than one hard-coded string: comparing against the +// SIGINT wording alone classified every SIGTERM, and therefore every normal pod +// termination, as a crash. +func IsShutdownSignalError(err error) bool { + if err == nil { + return false + } + + msg := err.Error() + for _, sig := range shutdownSignals { + if strings.Contains(msg, signalErrorPrefix+sig.String()) { + return true + } + } + return false +} + +// isFatalServerError reports whether an error returned by the worker's server +// run group is a genuine failure worth crashing the process for. +// +// A shutdown signal is never fatal. The shutdown context is still consulted so +// that a real server error raced against an in-progress shutdown stays quiet. +func (w *NVCFWorker) isFatalServerError(err error) bool { + return err != nil && !IsShutdownSignalError(err) && w.shutdownCtx.Err() == nil +} diff --git a/src/compute-plane-services/worker-utils/worker/shutdown_test.go b/src/compute-plane-services/worker-utils/worker/shutdown_test.go new file mode 100644 index 000000000..32d7f06ab --- /dev/null +++ b/src/compute-plane-services/worker-utils/worker/shutdown_test.go @@ -0,0 +1,107 @@ +/* +SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +SPDX-License-Identifier: Apache-2.0 + +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 worker + +import ( + "context" + "errors" + "fmt" + "testing" +) + +// sigtermRunError is the exact error the nvkit servers run group returns when +// the process receives SIGTERM, which is how Kubernetes always asks a container +// to stop. Reproduced verbatim from a production utils crash. +const sigtermRunError = "received signal terminated" + +// sigintRunError is the SIGINT equivalent. +const sigintRunError = "received signal interrupt" + +func TestIsShutdownSignalError(t *testing.T) { + tests := []struct { + name string + err error + want bool + }{ + {name: "sigterm", err: errors.New(sigtermRunError), want: true}, + {name: "sigint", err: errors.New(sigintRunError), want: true}, + { + name: "wrapped sigterm", + err: fmt.Errorf("internal error: %w", errors.New(sigtermRunError)), + want: true, + }, + {name: "unrelated failure", err: errors.New("listen tcp :9191: address already in use"), want: false}, + {name: "signal-shaped but not a shutdown signal", err: errors.New("received signal killed"), want: false}, + {name: "nil", err: nil, want: false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := IsShutdownSignalError(tt.err); got != tt.want { + t.Fatalf("IsShutdownSignalError(%v) = %v, want %v", tt.err, got, tt.want) + } + }) + } +} + +// TestIsFatalServerError_SIGTERMBeforeShutdownCancel is the regression test for +// the utils panic at worker.go "internal error: received signal terminated". +// The server run group always returns a non-nil error on SIGTERM, so when the +// shutdown context has not been cancelled yet the old guard classified a normal +// pod termination as a crash. +func TestIsFatalServerError_SIGTERMBeforeShutdownCancel(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + + w := &NVCFWorker{shutdownCtx: ctx} + + if w.isFatalServerError(errors.New(sigtermRunError)) { + t.Fatal("SIGTERM with a live shutdown context must not be treated as a fatal server error") + } +} + +func TestIsFatalServerError(t *testing.T) { + tests := []struct { + name string + err error + cancelBeforeCall bool + want bool + }{ + {name: "no error", err: nil, want: false}, + {name: "sigterm", err: errors.New(sigtermRunError), want: false}, + {name: "sigint", err: errors.New(sigintRunError), want: false}, + {name: "real failure during shutdown", err: errors.New("boom"), cancelBeforeCall: true, want: false}, + {name: "real failure while running", err: errors.New("boom"), want: true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + if tt.cancelBeforeCall { + cancel() + } + + w := &NVCFWorker{shutdownCtx: ctx} + + if got := w.isFatalServerError(tt.err); got != tt.want { + t.Fatalf("isFatalServerError(%v) = %v, want %v", tt.err, got, tt.want) + } + }) + } +} diff --git a/src/compute-plane-services/worker-utils/worker/worker.go b/src/compute-plane-services/worker-utils/worker/worker.go index 6041afe1f..561299741 100644 --- a/src/compute-plane-services/worker-utils/worker/worker.go +++ b/src/compute-plane-services/worker-utils/worker/worker.go @@ -402,7 +402,7 @@ func (w *NVCFWorker) Run(withHttpServer bool) error { go func() { // health server + framework config err := w.server.Run() - if err != nil && w.shutdownCtx.Err() == nil { + if w.isFatalServerError(err) { err = types.NewInternalError(err) utils.ExitReason(err) zap.S().Panic(err)