Add better shutdown handling for tessera (#432)

Before this change. it was possible for Tessera to get in the way of the
process shutting down in two ways:

1) The context in the Appender did not respond to signals, so if it ran
   into an irrecoverable error in `integrateEntriesJob`, it would retry
   indefinitely and the process would not end shutdown.
2) The appender Shutdown function also runs in an indefinite loop, and
   if Tessera got into a corrupt state where it could never finish
   appending the sequenced entries, the shutdown loop would never
   terminate.

This change adds a new timeout parameter specifically for Tessera to
give it a maximum time to finish processing entries. In the event it is
stuck looping trying to process an unprocessable entry, it will now
terminate in response to SIGINT/SIGTERM.

The default timeout for the server timeout, which is shared by the HTTP
and gRPC servers, is reduced to 20s, which in a correctly functioning
system with a checkpoint period of 10s should be long enough to finish
processing all in-flight requests. The new tessera timeout is 30s. This
means that in a badly behaving system where tessera did not integrate
entries in time to send the response to clients, it still has 30s to try
to finish committing entries to the log and empty the Spanner queue.

In the event that tessera encounters an irrecoverable error and cannot
finish processing entries, the maximum time to shut down is 50s by
default. This is reduced from 60s (server idle timeout) + indefinite
(tessera loop). The Pod termination grace period should be extended to
50-60s accordingly.

Signed-off-by: Colleen Murphy <[email protected]>

Author: cmurphy

cmd/rekor-server/app/serve.go

@@ -157,7 +157,7 @@ var serveCmd = &cobra.Command{
 			server.NewHTTPConfig(
 				server.WithHTTPPort(viper.GetInt("http-port")),
 				server.WithHTTPHost(viper.GetString("http-address")),
-				server.WithHTTPTimeout(viper.GetDuration("timeout")),
+				server.WithHTTPTimeout(viper.GetDuration("server-timeout")),
 				server.WithHTTPMaxRequestBodySize(viper.GetInt("max-request-body-size")),
 				server.WithHTTPMetricsPort(viper.GetInt("http-metrics-port")),
 				server.WithHTTPTLSCredentials(viper.GetString("http-tls-cert-file"), viper.GetString("http-tls-key-file")),
@@ -166,11 +166,12 @@ var serveCmd = &cobra.Command{
 			server.NewGRPCConfig(
 				server.WithGRPCPort(viper.GetInt("grpc-port")),
 				server.WithGRPCHost(viper.GetString("grpc-address")),
-				server.WithGRPCTimeout(viper.GetDuration("timeout")),
+				server.WithGRPCTimeout(viper.GetDuration("server-timeout")),
 				server.WithGRPCMaxMessageSize(viper.GetInt("max-request-body-size")),
 				server.WithGRPCLogLevel(logLevel, viper.GetBool("request-response-logging")),
 				server.WithTLSCredentials(viper.GetString("grpc-tls-cert-file"), viper.GetString("grpc-tls-key-file")),
 			),
+			viper.GetDuration("tlog-timeout"),
 			rekorServer,
 			shutdownFn,
 		)
@@ -185,7 +186,7 @@ func init() {
 	serveCmd.Flags().Int("http-metrics-port", 2112, "HTTP port to bind metrics to")
 	serveCmd.Flags().Int("grpc-port", 3001, "GRPC port to bind to")
 	serveCmd.Flags().String("grpc-address", "127.0.0.1", "GRPC address to bind to")
-	serveCmd.Flags().Duration("timeout", 60*time.Second, "timeout")
+	serveCmd.Flags().Duration("server-timeout", 20*time.Second, "timeout settings for gRPC and HTTP connections")
 	serveCmd.Flags().Int("max-request-body-size", 4*1024*1024, "maximum request body size in bytes")
 	serveCmd.Flags().String("log-level", "info", "log level for the process. options are [debug, info, warn, error]")
 	serveCmd.Flags().Bool("request-response-logging", false, "enables logging of request and response content; log-level must be 'debug' for this to take effect")
@@ -218,6 +219,7 @@ func init() {
 	serveCmd.Flags().Duration("batch-max-age", tessera.DefaultBatchMaxAge, "the maximum amount of time a batch of entries will wait before being sent to the sequencer")
 	serveCmd.Flags().Duration("checkpoint-interval", tessera.DefaultCheckpointInterval, "the frequency at which a checkpoint will be published")
 	serveCmd.Flags().Uint("pushback-max-outstanding", tessera.DefaultPushbackMaxOutstanding, "the maximum number of 'in-flight' add requests")
+	serveCmd.Flags().Duration("tlog-timeout", 30*time.Second, "timeout for terminating the tiles log queue")
 
 	// antispam configs
 	serveCmd.Flags().Bool("persistent-antispam", false, "whether to enable persistent antispam measures; only available for GCP storage backend and not supported by the Spanner storage emulator")

internal/server/serve.go

@@ -19,10 +19,11 @@ import (
 	"log/slog"
 	"os"
 	"sync"
+	"time"
 )
 
 // Serve starts the grpc server and its http proxy.
-func Serve(ctx context.Context, hc *HTTPConfig, gc *GRPCConfig, s rekorServer, tesseraShutdownFn func(context.Context) error) {
+func Serve(ctx context.Context, hc *HTTPConfig, gc *GRPCConfig, tesseraTimeout time.Duration, s rekorServer, tesseraShutdownFn func(context.Context) error) {
 	var wg sync.WaitGroup
 
 	if hc.port == 0 || gc.port == 0 {
@@ -46,7 +47,9 @@ func Serve(ctx context.Context, hc *HTTPConfig, gc *GRPCConfig, s rekorServer, t
 	wg.Wait()
 
 	slog.Info("shutting down Tessera sequencer")
-	if err := tesseraShutdownFn(ctx); err != nil {
+	tesseraCtx, cancel := context.WithTimeout(ctx, tesseraTimeout)
+	defer cancel()
+	if err := tesseraShutdownFn(tesseraCtx); err != nil {
 		slog.Error("error shutting down Tessera", "error", err)
 	}
 	slog.Info("stopped Tessera sequencer")

internal/server/serve_test.go

@@ -32,7 +32,7 @@ func TestServe(t *testing.T) {
 	}
 	go func() {
 		pid.Store(uint64(syscall.Getpid())) // Process IDs are positive ints
-		Serve(context.Background(), NewHTTPConfig(), NewGRPCConfig(), nil, shutdownFn)
+		Serve(context.Background(), NewHTTPConfig(), NewGRPCConfig(), 1*time.Second, nil, shutdownFn)
 		wg.Done()
 	}()
 	// One for Serve returning, one for shutdown function being invoked

internal/server/service_mock.go

@@ -68,7 +68,7 @@ func (ms *MockServer) Start(t *testing.T) {
 	// Start the server
 	ms.wg = &sync.WaitGroup{}
 	go func() {
-		Serve(context.Background(), ms.hc, ms.gc, s, shutdownFn)
+		Serve(context.Background(), ms.hc, ms.gc, 1*time.Second, s, shutdownFn)
 		ms.wg.Done()
 	}()
 	ms.wg.Add(1)
@@ -116,7 +116,7 @@ func (ms *MockServer) StartTLS(t *testing.T) {
 	// Start the server
 	ms.wg = &sync.WaitGroup{}
 	go func() {
-		Serve(context.Background(), ms.hc, ms.gc, s, shutdownFn)
+		Serve(context.Background(), ms.hc, ms.gc, 1*time.Second, s, shutdownFn)
 		ms.wg.Done()
 	}()
 	ms.wg.Add(1)