Skip to content

Commit bed112a

Browse files
tchaptchap
committed
Remove fatal logging in executable code
All affected functions should now return an error. ErrorS is now being used instead of Fatal.
1 parent d5f5507 commit bed112a

File tree

3 files changed

+187
-96
lines changed

3 files changed

+187
-96
lines changed

Diff for: ‎pkg/ext-proc/main.go

+107-53
Original file line numberDiff line numberDiff line change
@@ -6,9 +6,11 @@ import (
66
"fmt"
77
"net"
88
"net/http"
9+
"os"
910
"strconv"
1011

1112
"github.com/prometheus/client_golang/prometheus/promhttp"
13+
"golang.org/x/sync/errgroup"
1214
"google.golang.org/grpc"
1315
healthPb "google.golang.org/grpc/health/grpc_health_v1"
1416
"inference.networking.x-k8s.io/gateway-api-inference-extension/api/v1alpha1"
@@ -72,17 +74,25 @@ func init() {
7274
}
7375

7476
func main() {
77+
if err := run(); err != nil {
78+
os.Exit(1)
79+
}
80+
}
81+
82+
func run() error {
7583
klog.InitFlags(nil)
7684
flag.Parse()
7785

7886
ctrl.SetLogger(klog.TODO())
7987
cfg, err := ctrl.GetConfig()
8088
if err != nil {
81-
klog.Fatalf("Failed to get rest config: %v", err)
89+
klog.ErrorS(err, "Failed to get rest config")
90+
return err
8291
}
8392
// Validate flags
8493
if err := validateFlags(); err != nil {
85-
klog.Fatalf("Failed to validate flags: %v", err)
94+
klog.ErrorS(err, "Failed to validate flags")
95+
return err
8696
}
8797

8898
// Print all flag values
@@ -113,96 +123,140 @@ func main() {
113123
}
114124
datastore.SetClient(k8sClient)
115125

116-
// Start health and ext-proc servers in goroutines
117-
healthSvr := startHealthServer(datastore, *grpcHealthPort)
118-
extProcSvr := serverRunner.Start(&vllm.PodMetricsClientImpl{})
119-
// Start metrics handler
120-
metricsSvr := startMetricsHandler(*metricsPort, cfg)
126+
if err := serverRunner.Setup(); err != nil {
127+
klog.ErrorS(err, "Failed to setup server runner")
128+
return err
129+
}
121130

122-
// Start manager, blocking
123-
serverRunner.StartManager()
131+
// Start processing signals and init the group to manage goroutines.
132+
g, ctx := errgroup.WithContext(ctrl.SetupSignalHandler())
124133

125-
// Gracefully shutdown servers
126-
if healthSvr != nil {
127-
klog.Info("Health server shutting down")
128-
healthSvr.GracefulStop()
129-
}
130-
if extProcSvr != nil {
131-
klog.Info("Ext-proc server shutting down")
132-
extProcSvr.GracefulStop()
133-
}
134-
if metricsSvr != nil {
135-
klog.Info("Metrics server shutting down")
136-
if err := metricsSvr.Shutdown(context.Background()); err != nil {
137-
klog.Infof("Metrics server Shutdown: %v", err)
138-
}
139-
}
134+
// Start health server.
135+
startHealthServer(ctx, g, datastore, *grpcHealthPort)
136+
137+
// Start ext-proc server.
138+
g.Go(func() error {
139+
return serverRunner.Start(ctx, &vllm.PodMetricsClientImpl{})
140+
})
141+
142+
// Start metrics handler.
143+
startMetricsHandler(ctx, g, *metricsPort, cfg)
140144

141-
klog.Info("All components shutdown")
145+
// Start manager.
146+
g.Go(func() error {
147+
return serverRunner.StartManager(ctx)
148+
})
149+
150+
err = g.Wait()
151+
klog.InfoS("All components terminated")
152+
return err
142153
}
143154

144-
// startHealthServer starts the gRPC health probe server in a goroutine.
145-
func startHealthServer(ds *backend.K8sDatastore, port int) *grpc.Server {
146-
svr := grpc.NewServer()
147-
healthPb.RegisterHealthServer(svr, &healthServer{datastore: ds})
155+
// startHealthServer starts the gRPC health probe server using the given errgroup.
156+
func startHealthServer(ctx context.Context, g *errgroup.Group, ds *backend.K8sDatastore, port int) {
157+
g.Go(func() error {
158+
klog.InfoS("Health server starting...")
148159

149-
go func() {
160+
// Start listening.
150161
lis, err := net.Listen("tcp", fmt.Sprintf(":%d", port))
151162
if err != nil {
152-
klog.Fatalf("Health server failed to listen: %v", err)
163+
klog.ErrorS(err, "Health server failed to listen")
164+
return err
153165
}
154-
klog.Infof("Health server listening on port: %d", port)
155166

156-
// Blocking and will return when shutdown is complete.
167+
klog.InfoS("Health server listening", "port", port)
168+
169+
svr := grpc.NewServer()
170+
healthPb.RegisterHealthServer(svr, &healthServer{datastore: ds})
171+
172+
// Shutdown on context closed.
173+
g.Go(func() error {
174+
<-ctx.Done()
175+
klog.InfoS("Health server shutting down...")
176+
svr.GracefulStop()
177+
return nil
178+
})
179+
180+
// Keep serving until terminated.
157181
if err := svr.Serve(lis); err != nil && err != grpc.ErrServerStopped {
158-
klog.Fatalf("Health server failed: %v", err)
182+
klog.ErrorS(err, "Health server failed")
183+
return err
159184
}
160-
klog.Info("Health server shutting down")
161-
}()
162-
return svr
185+
klog.InfoS("Health server terminated")
186+
return nil
187+
})
163188
}
164189

165-
func startMetricsHandler(port int, cfg *rest.Config) *http.Server {
166-
metrics.Register()
190+
// startMetricsHandler starts the metrics HTTP handler using the given errgroup.
191+
func startMetricsHandler(ctx context.Context, g *errgroup.Group, port int, cfg *rest.Config) {
192+
g.Go(func() error {
193+
metrics.Register()
194+
klog.InfoS("Metrics HTTP handler starting...")
195+
196+
// Start listening.
197+
lis, err := net.Listen("tcp", fmt.Sprintf(":%d", port))
198+
if err != nil {
199+
klog.ErrorS(err, "Metrics HTTP handler failed to listen")
200+
return err
201+
}
167202

168-
var svr *http.Server
169-
go func() {
170-
klog.Info("Starting metrics HTTP handler ...")
203+
klog.InfoS("Metrics HTTP handler listening", "port", port)
204+
205+
// Init HTTP server.
206+
h, err := metricsHandlerWithAuthenticationAndAuthorization(cfg)
207+
if err != nil {
208+
return err
209+
}
171210

172211
mux := http.NewServeMux()
173-
mux.Handle(defaultMetricsEndpoint, metricsHandlerWithAuthenticationAndAuthorization(cfg))
212+
mux.Handle(defaultMetricsEndpoint, h)
174213

175-
svr = &http.Server{
214+
svr := &http.Server{
176215
Addr: net.JoinHostPort("", strconv.Itoa(port)),
177216
Handler: mux,
178217
}
179-
if err := svr.ListenAndServe(); err != http.ErrServerClosed {
180-
klog.Fatalf("failed to start metrics HTTP handler: %v", err)
218+
219+
// Shutdown on interrupt.
220+
g.Go(func() error {
221+
<-ctx.Done()
222+
klog.InfoS("Metrics HTTP handler shutting down...")
223+
_ = svr.Shutdown(context.Background())
224+
return nil
225+
})
226+
227+
// Keep serving until terminated.
228+
if err := svr.Serve(lis); err != http.ErrServerClosed {
229+
klog.ErrorS(err, "Metrics HTTP handler failed")
230+
return err
181231
}
182-
}()
183-
return svr
232+
klog.InfoS("Metrics HTTP handler terminated")
233+
return nil
234+
})
184235
}
185236

186-
func metricsHandlerWithAuthenticationAndAuthorization(cfg *rest.Config) http.Handler {
237+
func metricsHandlerWithAuthenticationAndAuthorization(cfg *rest.Config) (http.Handler, error) {
187238
h := promhttp.HandlerFor(
188239
legacyregistry.DefaultGatherer,
189240
promhttp.HandlerOpts{},
190241
)
191242
httpClient, err := rest.HTTPClientFor(cfg)
192243
if err != nil {
193-
klog.Fatalf("failed to create http client for metrics auth: %v", err)
244+
klog.ErrorS(err, "Failed to create http client for metrics auth")
245+
return nil, err
194246
}
195247

196248
filter, err := filters.WithAuthenticationAndAuthorization(cfg, httpClient)
197249
if err != nil {
198-
klog.Fatalf("failed to create metrics filter for auth: %v", err)
250+
klog.ErrorS(err, "Failed to create metrics filter for auth")
251+
return nil, err
199252
}
200253
metricsLogger := klog.LoggerWithValues(klog.NewKlogr(), "path", defaultMetricsEndpoint)
201254
metricsAuthHandler, err := filter(metricsLogger, h)
202255
if err != nil {
203-
klog.Fatalf("failed to create metrics auth handler: %v", err)
256+
klog.ErrorS(err, "Failed to create metrics auth handler")
257+
return nil, err
204258
}
205-
return metricsAuthHandler
259+
return metricsAuthHandler, nil
206260
}
207261

208262
func validateFlags() error {

Diff for: ‎pkg/ext-proc/server/runserver.go

+64-34
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,8 @@
11
package server
22

33
import (
4+
"context"
5+
"errors"
46
"fmt"
57
"net"
68
"time"
@@ -53,12 +55,12 @@ func NewDefaultExtProcServerRunner() *ExtProcServerRunner {
5355
}
5456
}
5557

56-
// Setup creates the reconcilers for pools and models and starts the manager.
57-
func (r *ExtProcServerRunner) Setup() {
58+
// Setup creates the reconcilers for pools, models, and endpointSlices and starts the manager.
59+
func (r *ExtProcServerRunner) Setup() error {
5860
// Create a new manager to manage controllers
5961
mgr, err := ctrl.NewManager(r.Config, ctrl.Options{Scheme: r.Scheme})
6062
if err != nil {
61-
klog.Fatalf("Failed to create controller manager: %v", err)
63+
return fmt.Errorf("failed to create controller manager: %w", err)
6264
}
6365
r.Manager = mgr
6466

@@ -73,7 +75,7 @@ func (r *ExtProcServerRunner) Setup() {
7375
},
7476
Record: mgr.GetEventRecorderFor("InferencePool"),
7577
}).SetupWithManager(mgr); err != nil {
76-
klog.Fatalf("Failed setting up InferencePoolReconciler: %v", err)
78+
return fmt.Errorf("failed setting up InferencePoolReconciler: %w", err)
7779
}
7880

7981
if err := (&backend.InferenceModelReconciler{
@@ -86,52 +88,80 @@ func (r *ExtProcServerRunner) Setup() {
8688
},
8789
Record: mgr.GetEventRecorderFor("InferenceModel"),
8890
}).SetupWithManager(mgr); err != nil {
89-
klog.Fatalf("Failed setting up InferenceModelReconciler: %v", err)
91+
return fmt.Errorf("failed setting up InferenceModelReconciler: %w", err)
9092
}
93+
return nil
9194
}
9295

93-
// Start starts the Envoy external processor server in a goroutine.
96+
// Start starts the Envoy external processor server and blocks
97+
// until the context is canceled or an error encountered.
9498
func (r *ExtProcServerRunner) Start(
99+
ctx context.Context,
95100
podMetricsClient backend.PodMetricsClient,
96-
) *grpc.Server {
101+
) error {
102+
klog.InfoS("Ext-proc server starting...")
103+
104+
// Start listening.
97105
svr := grpc.NewServer()
106+
lis, err := net.Listen("tcp", fmt.Sprintf(":%d", r.GrpcPort))
107+
if err != nil {
108+
klog.ErrorS(err, "Ext-proc server failed to listen", "port", r.GrpcPort)
109+
return err
110+
}
111+
// The listener will be closed by the server,
112+
// but the function may also return earlier on error.
113+
defer lis.Close()
98114

99-
go func() {
100-
lis, err := net.Listen("tcp", fmt.Sprintf(":%d", r.GrpcPort))
101-
if err != nil {
102-
klog.Fatalf("Ext-proc server failed to listen: %v", err)
103-
}
104-
klog.Infof("Ext-proc server listening on port: %d", r.GrpcPort)
115+
klog.InfoS("Ext-proc server listening", "port", r.GrpcPort)
105116

106-
// Initialize backend provider
107-
pp := backend.NewProvider(podMetricsClient, r.Datastore)
108-
if err := pp.Init(r.RefreshPodsInterval, r.RefreshMetricsInterval); err != nil {
109-
klog.Fatalf("Failed to initialize backend provider: %v", err)
110-
}
117+
// Initialize backend provider
118+
pp := backend.NewProvider(podMetricsClient, r.Datastore)
119+
if err := pp.Init(r.RefreshPodsInterval, r.RefreshMetricsInterval); err != nil {
120+
klog.ErrorS(err, "Failed to initialize backend provider")
121+
return err
122+
}
111123

112-
// Register ext_proc handlers
113-
extProcPb.RegisterExternalProcessorServer(
114-
svr,
115-
handlers.NewServer(pp, scheduling.NewScheduler(pp), r.TargetEndpointKey, r.Datastore),
116-
)
124+
// Register ext_proc handlers
125+
extProcPb.RegisterExternalProcessorServer(
126+
svr,
127+
handlers.NewServer(pp, scheduling.NewScheduler(pp), r.TargetEndpointKey, r.Datastore),
128+
)
117129

118-
// Blocking and will return when shutdown is complete.
119-
if err := svr.Serve(lis); err != nil && err != grpc.ErrServerStopped {
120-
klog.Fatalf("Ext-proc server failed: %v", err)
130+
// Terminate the server on context closed.
131+
// Make sure the goroutine does not leak.
132+
doneCh := make(chan struct{})
133+
defer close(doneCh)
134+
go func() {
135+
select {
136+
case <-ctx.Done():
137+
klog.InfoS("Ext-proc server shutting down...")
138+
svr.GracefulStop()
139+
case <-doneCh:
121140
}
122-
klog.Info("Ext-proc server shutting down")
123141
}()
124-
return svr
142+
143+
// Block until terminated.
144+
if err := svr.Serve(lis); err != nil && err != grpc.ErrServerStopped {
145+
klog.ErrorS(err, "Ext-proc server failed")
146+
return err
147+
}
148+
klog.InfoS("Ext-proc server terminated")
149+
return nil
125150
}
126151

127-
func (r *ExtProcServerRunner) StartManager() {
152+
func (r *ExtProcServerRunner) StartManager(ctx context.Context) error {
128153
if r.Manager == nil {
129-
klog.Fatalf("Runner has no manager setup to run: %v", r)
154+
err := errors.New("runner manager is not set")
155+
klog.ErrorS(err, "Runner has no manager setup to run")
156+
return err
130157
}
158+
131159
// Start the controller manager. Blocking and will return when shutdown is complete.
132-
klog.Infof("Starting controller manager")
133-
if err := r.Manager.Start(ctrl.SetupSignalHandler()); err != nil {
134-
klog.Fatalf("Error starting controller manager: %v", err)
160+
klog.InfoS("Controller manager starting...")
161+
if err := r.Manager.Start(ctx); err != nil {
162+
klog.ErrorS(err, "Error starting controller manager")
163+
return err
135164
}
136-
klog.Info("Controller manager shutting down")
165+
klog.InfoS("Controller manager terminated")
166+
return nil
137167
}

0 commit comments

Comments
 (0)