HTTP-Server: Graceful shutdown (#97223) (#97645)

Co-authored-by: Kibana Machine <42973632+kibanamachine@users.noreply.github.com>

Co-authored-by: Alejandro Fernández Haro <alejandro.haro@elastic.co>

packages/kbn-cli-dev-mode/src/base_path_proxy_server.test.ts

@@ -8,6 +8,7 @@
 
 import { Server } from '@hapi/hapi';
 import { EMPTY } from 'rxjs';
+import moment from 'moment';
 import supertest from 'supertest';
 import {
   getServerOptions,
@@ -35,6 +36,7 @@ describe('BasePathProxyServer', () => {
     config = {
       host: '127.0.0.1',
       port: 10012,
+      shutdownTimeout: moment.duration(30, 'seconds'),
       keepaliveTimeout: 1000,
       socketTimeout: 1000,
       cors: {

packages/kbn-cli-dev-mode/src/cli_dev_mode.test.ts

@@ -108,7 +108,7 @@ it('passes correct args to sub-classes', () => {
             "bar",
             "baz",
           ],
-          "gracefulTimeout": 5000,
+          "gracefulTimeout": 30000,
           "log": <TestLog>,
           "mapLogLine": [Function],
           "script": <absolute path>/scripts/kibana,

packages/kbn-cli-dev-mode/src/cli_dev_mode.ts

@@ -44,7 +44,7 @@ Rx.merge(
   .subscribe(exitSignal$);
 
 // timeout where the server is allowed to exit gracefully
-const GRACEFUL_TIMEOUT = 5000;
+const GRACEFUL_TIMEOUT = 30000;
 
 export type SomeCliArgs = Pick<
   CliArgs,

packages/kbn-cli-dev-mode/src/config/http_config.ts

@@ -8,6 +8,7 @@
 
 import { ByteSizeValue, schema, TypeOf } from '@kbn/config-schema';
 import { ICorsConfig, IHttpConfig, ISslConfig, SslConfig, sslSchema } from '@kbn/server-http-tools';
+import { Duration } from 'moment';
 
 export const httpConfigSchema = schema.object(
   {
@@ -22,6 +23,7 @@ export const httpConfigSchema = schema.object(
     maxPayload: schema.byteSize({
       defaultValue: '1048576b',
     }),
+    shutdownTimeout: schema.duration({ defaultValue: '30s' }),
     keepaliveTimeout: schema.number({
       defaultValue: 120000,
     }),
@@ -47,6 +49,7 @@ export class HttpConfig implements IHttpConfig {
   host: string;
   port: number;
   maxPayload: ByteSizeValue;
+  shutdownTimeout: Duration;
   keepaliveTimeout: number;
   socketTimeout: number;
   cors: ICorsConfig;
@@ -57,6 +60,7 @@ export class HttpConfig implements IHttpConfig {
     this.host = rawConfig.host;
     this.port = rawConfig.port;
     this.maxPayload = rawConfig.maxPayload;
+    this.shutdownTimeout = rawConfig.shutdownTimeout;
     this.keepaliveTimeout = rawConfig.keepaliveTimeout;
     this.socketTimeout = rawConfig.socketTimeout;
     this.cors = rawConfig.cors;

packages/kbn-cli-dev-mode/src/dev_server.ts

@@ -103,7 +103,7 @@ export class DevServer {
   /**
    * Run the Kibana server
    *
-   * The observable will error if the child process failes to spawn for some reason, but if
+   * The observable will error if the child process fails to spawn for some reason, but if
    * the child process is successfully spawned then the server will be run until it completes
    * and restart when the watcher indicates it should. In order to restart the server as
    * quickly as possible we kill it with SIGKILL and spawn the process again.

packages/kbn-server-http-tools/src/get_server_options.test.ts

@@ -6,6 +6,7 @@
  * Side Public License, v 1.
  */
 
+import moment from 'moment';
 import { ByteSizeValue } from '@kbn/config-schema';
 import { getServerOptions } from './get_server_options';
 import { IHttpConfig } from './types';
@@ -24,6 +25,7 @@ const createConfig = (parts: Partial<IHttpConfig>): IHttpConfig => ({
   port: 5601,
   socketTimeout: 120000,
   keepaliveTimeout: 120000,
+  shutdownTimeout: moment.duration(30, 'seconds'),
   maxPayload: ByteSizeValue.parse('1048576b'),
   ...parts,
   cors: {

packages/kbn-server-http-tools/src/types.ts

@@ -7,6 +7,7 @@
  */
 
 import { ByteSizeValue } from '@kbn/config-schema';
+import type { Duration } from 'moment';
 
 export interface IHttpConfig {
   host: string;
@@ -16,6 +17,7 @@ export interface IHttpConfig {
   socketTimeout: number;
   cors: ICorsConfig;
   ssl: ISslConfig;
+  shutdownTimeout: Duration;
 }
 
 export interface ICorsConfig {

src/core/server/http/__snapshots__/http_config.test.ts.snap

@@ -65,6 +65,7 @@ Object {
     "strictTransportSecurity": null,
     "xContentTypeOptions": "nosniff",
   },
+  "shutdownTimeout": "PT30S",
   "socketTimeout": 120000,
   "ssl": Object {
     "cipherSuites": Array [

src/core/server/http/http_config.test.ts

@@ -109,6 +109,35 @@ test('can specify max payload as string', () => {
   expect(configValue.maxPayload.getValueInBytes()).toBe(2 * 1024 * 1024);
 });
 
+describe('shutdownTimeout', () => {
+  test('can specify a valid shutdownTimeout', () => {
+    const configValue = config.schema.validate({ shutdownTimeout: '5s' });
+    expect(configValue.shutdownTimeout.asMilliseconds()).toBe(5000);
+  });
+
+  test('can specify a valid shutdownTimeout (lower-edge of 1 second)', () => {
+    const configValue = config.schema.validate({ shutdownTimeout: '1s' });
+    expect(configValue.shutdownTimeout.asMilliseconds()).toBe(1000);
+  });
+
+  test('can specify a valid shutdownTimeout (upper-edge of 2 minutes)', () => {
+    const configValue = config.schema.validate({ shutdownTimeout: '2m' });
+    expect(configValue.shutdownTimeout.asMilliseconds()).toBe(120000);
+  });
+
+  test('should error if below 1s', () => {
+    expect(() => config.schema.validate({ shutdownTimeout: '100ms' })).toThrow(
+      '[shutdownTimeout]: the value should be between 1 second and 2 minutes'
+    );
+  });
+
+  test('should error if over 2 minutes', () => {
+    expect(() => config.schema.validate({ shutdownTimeout: '3m' })).toThrow(
+      '[shutdownTimeout]: the value should be between 1 second and 2 minutes'
+    );
+  });
+});
+
 describe('basePath', () => {
   test('throws if missing prepended slash', () => {
     const httpSchema = config.schema;

src/core/server/http/http_config.ts

@@ -11,6 +11,7 @@ import { IHttpConfig, SslConfig, sslSchema } from '@kbn/server-http-tools';
 import { hostname } from 'os';
 import url from 'url';
 
+import type { Duration } from 'moment';
 import { ServiceConfigDescriptor } from '../internal_types';
 import { CspConfigType, CspConfig, ICspConfig } from '../csp';
 import { ExternalUrlConfig, IExternalUrlConfig } from '../external_url';
@@ -35,6 +36,15 @@ const configSchema = schema.object(
         validate: match(validBasePathRegex, "must start with a slash, don't end with one"),
       })
     ),
+    shutdownTimeout: schema.duration({
+      defaultValue: '30s',
+      validate: (duration) => {
+        const durationMs = duration.asMilliseconds();
+        if (durationMs < 1000 || durationMs > 2 * 60 * 1000) {
+          return 'the value should be between 1 second and 2 minutes';
+        }
+      },
+    }),
     cors: schema.object(
       {
         enabled: schema.boolean({ defaultValue: false }),
@@ -183,6 +193,7 @@ export class HttpConfig implements IHttpConfig {
   public externalUrl: IExternalUrlConfig;
   public xsrf: { disableProtection: boolean; allowlist: string[] };
   public requestId: { allowFromAnyIp: boolean; ipAllowlist: string[] };
+  public shutdownTimeout: Duration;
 
   /**
    * @internal
@@ -228,6 +239,7 @@ export class HttpConfig implements IHttpConfig {
     this.externalUrl = rawExternalUrlConfig;
     this.xsrf = rawHttpConfig.xsrf;
     this.requestId = rawHttpConfig.requestId;
+    this.shutdownTimeout = rawHttpConfig.shutdownTimeout;
   }
 }
 

src/core/server/http/http_server.test.ts

@@ -26,6 +26,8 @@ import { HttpServer } from './http_server';
 import { Readable } from 'stream';
 import { RequestHandlerContext } from 'kibana/server';
 import { KBN_CERT_PATH, KBN_KEY_PATH } from '@kbn/dev-utils';
+import moment from 'moment';
+import { of } from 'rxjs';
 
 const cookieOptions = {
   name: 'sid',
@@ -65,6 +67,7 @@ beforeEach(() => {
     cors: {
       enabled: false,
     },
+    shutdownTimeout: moment.duration(500, 'ms'),
   } as any;
 
   configWithSSL = {
@@ -79,7 +82,7 @@ beforeEach(() => {
     },
   } as HttpConfig;
 
-  server = new HttpServer(loggingService, 'tests');
+  server = new HttpServer(loggingService, 'tests', of(config.shutdownTimeout));
 });
 
 afterEach(async () => {
@@ -1431,3 +1434,79 @@ describe('setup contract', () => {
     });
   });
 });
+
+describe('Graceful shutdown', () => {
+  let shutdownTimeout: number;
+  let innerServerListener: Server;
+
+  beforeEach(async () => {
+    shutdownTimeout = config.shutdownTimeout.asMilliseconds();
+    const { registerRouter, server: innerServer } = await server.setup(config);
+    innerServerListener = innerServer.listener;
+
+    const router = new Router('', logger, enhanceWithContext);
+    router.post(
+      {
+        path: '/',
+        validate: false,
+        options: { body: { accepts: 'application/json' } },
+      },
+      async (context, req, res) => {
+        // It takes to resolve the same period of the shutdownTimeout.
+        // Since we'll trigger the stop a few ms after, it should have time to finish
+        await new Promise((resolve) => setTimeout(resolve, shutdownTimeout));
+        return res.ok({ body: { ok: 1 } });
+      }
+    );
+    registerRouter(router);
+
+    await server.start();
+  });
+
+  test('any ongoing requests should be resolved with `connection: close`', async () => {
+    const [response] = await Promise.all([
+      // Trigger a request that should hold the server from stopping until fulfilled
+      supertest(innerServerListener).post('/'),
+      // Stop the server while the request is in progress
+      (async () => {
+        await new Promise((resolve) => setTimeout(resolve, shutdownTimeout / 3));
+        await server.stop();
+      })(),
+    ]);
+
+    expect(response.status).toBe(200);
+    expect(response.body).toStrictEqual({ ok: 1 });
+    // The server is about to be closed, we need to ask connections to close on their end (stop their keep-alive policies)
+    expect(response.header.connection).toBe('close');
+  });
+
+  test('any requests triggered while stopping should be rejected with 503', async () => {
+    const [, , response] = await Promise.all([
+      // Trigger a request that should hold the server from stopping until fulfilled (otherwise the server will stop straight away)
+      supertest(innerServerListener).post('/'),
+      // Stop the server while the request is in progress
+      (async () => {
+        await new Promise((resolve) => setTimeout(resolve, shutdownTimeout / 3));
+        await server.stop();
+      })(),
+      // Trigger a new request while shutting down (should be rejected)
+      (async () => {
+        await new Promise((resolve) => setTimeout(resolve, (2 * shutdownTimeout) / 3));
+        return supertest(innerServerListener).post('/');
+      })(),
+    ]);
+    expect(response.status).toBe(503);
+    expect(response.body).toStrictEqual({
+      statusCode: 503,
+      error: 'Service Unavailable',
+      message: 'Kibana is shutting down and not accepting new incoming requests',
+    });
+    expect(response.header.connection).toBe('close');
+  });
+
+  test('when no ongoing connections, the server should stop without waiting any longer', async () => {
+    const preStop = Date.now();
+    await server.stop();
+    expect(Date.now() - preStop).toBeLessThan(shutdownTimeout);
+  });
+});

src/core/server/http/http_server.ts

@@ -17,6 +17,9 @@ import {
   getRequestId,
 } from '@kbn/server-http-tools';
 
+import type { Duration } from 'moment';
+import { Observable } from 'rxjs';
+import { take } from 'rxjs/operators';
 import { Logger, LoggerFactory } from '../logging';
 import { HttpConfig } from './http_config';
 import { adoptToHapiAuthFormat, AuthenticationHandler } from './lifecycle/auth';
@@ -80,6 +83,7 @@ export class HttpServer {
   private authRegistered = false;
   private cookieSessionStorageCreated = false;
   private handleServerResponseEvent?: (req: Request) => void;
+  private stopping = false;
   private stopped = false;
 
   private readonly log: Logger;
@@ -87,7 +91,11 @@ export class HttpServer {
   private readonly authRequestHeaders: AuthHeadersStorage;
   private readonly authResponseHeaders: AuthHeadersStorage;
 
-  constructor(private readonly logger: LoggerFactory, private readonly name: string) {
+  constructor(
+    private readonly logger: LoggerFactory,
+    private readonly name: string,
+    private readonly shutdownTimeout$: Observable<Duration>
+  ) {
     this.authState = new AuthStateStorage(() => this.authRegistered);
     this.authRequestHeaders = new AuthHeadersStorage();
     this.authResponseHeaders = new AuthHeadersStorage();
@@ -118,6 +126,7 @@ export class HttpServer {
     this.setupConditionalCompression(config);
     this.setupResponseLogging();
     this.setupRequestStateAssignment(config);
+    this.setupGracefulShutdownHandlers();
 
     return {
       registerRouter: this.registerRouter.bind(this),
@@ -153,7 +162,7 @@ export class HttpServer {
     if (this.server === undefined) {
       throw new Error('Http server is not setup up yet');
     }
-    if (this.stopped) {
+    if (this.stopping || this.stopped) {
       this.log.warn(`start called after stop`);
       return;
     }
@@ -213,19 +222,29 @@ export class HttpServer {
   }
 
   public async stop() {
-    this.stopped = true;
+    this.stopping = true;
     if (this.server === undefined) {
+      this.stopping = false;
+      this.stopped = true;
       return;
     }
 
     const hasStarted = this.server.info.started > 0;
     if (hasStarted) {
       this.log.debug('stopping http server');
+
+      const shutdownTimeout = await this.shutdownTimeout$.pipe(take(1)).toPromise();
+      await this.server.stop({ timeout: shutdownTimeout.asMilliseconds() });
+
+      this.log.debug(`http server stopped`);
+
+      // Removing the listener after stopping so we don't leave any pending requests unhandled
       if (this.handleServerResponseEvent) {
         this.server.events.removeListener('response', this.handleServerResponseEvent);
       }
-      await this.server.stop();
     }
+    this.stopping = false;
+    this.stopped = true;
   }
 
   private getAuthOption(
@@ -246,6 +265,18 @@ export class HttpServer {
     }
   }
 
+  private setupGracefulShutdownHandlers() {
+    this.registerOnPreRouting((request, response, toolkit) => {
+      if (this.stopping || this.stopped) {
+        return response.customError({
+          statusCode: 503,
+          body: { message: 'Kibana is shutting down and not accepting new incoming requests' },
+        });
+      }
+      return toolkit.next();
+    });
+  }
+
   private setupBasePathRewrite(config: HttpConfig, basePathService: BasePath) {
     if (config.basePath === undefined || !config.rewriteBasePath) {
       return;
@@ -266,7 +297,7 @@ export class HttpServer {
     if (this.server === undefined) {
       throw new Error('Server is not created yet');
     }
-    if (this.stopped) {
+    if (this.stopping || this.stopped) {
       this.log.warn(`setupConditionalCompression called after stop`);
     }
 
@@ -296,7 +327,7 @@ export class HttpServer {
     if (this.server === undefined) {
       throw new Error('Server is not created yet');
     }
-    if (this.stopped) {
+    if (this.stopping || this.stopped) {
       this.log.warn(`setupResponseLogging called after stop`);
     }
 
@@ -325,7 +356,7 @@ export class HttpServer {
     if (this.server === undefined) {
       throw new Error('Server is not created yet');
     }
-    if (this.stopped) {
+    if (this.stopping || this.stopped) {
       this.log.warn(`registerOnPreAuth called after stop`);
     }
 
@@ -336,7 +367,7 @@ export class HttpServer {
     if (this.server === undefined) {
       throw new Error('Server is not created yet');
     }
-    if (this.stopped) {
+    if (this.stopping || this.stopped) {
       this.log.warn(`registerOnPostAuth called after stop`);
     }
 
@@ -347,7 +378,7 @@ export class HttpServer {
     if (this.server === undefined) {
       throw new Error('Server is not created yet');
     }
-    if (this.stopped) {
+    if (this.stopping || this.stopped) {
       this.log.warn(`registerOnPreRouting called after stop`);
     }
 
@@ -358,7 +389,7 @@ export class HttpServer {
     if (this.server === undefined) {
       throw new Error('Server is not created yet');
     }
-    if (this.stopped) {
+    if (this.stopping || this.stopped) {
       this.log.warn(`registerOnPreResponse called after stop`);
     }
 
@@ -372,7 +403,7 @@ export class HttpServer {
     if (this.server === undefined) {
       throw new Error('Server is not created yet');
     }
-    if (this.stopped) {
+    if (this.stopping || this.stopped) {
       this.log.warn(`createCookieSessionStorageFactory called after stop`);
     }
     if (this.cookieSessionStorageCreated) {
@@ -392,7 +423,7 @@ export class HttpServer {
     if (this.server === undefined) {
       throw new Error('Server is not created yet');
     }
-    if (this.stopped) {
+    if (this.stopping || this.stopped) {
       this.log.warn(`registerAuth called after stop`);
     }
     if (this.authRegistered) {
@@ -438,7 +469,7 @@ export class HttpServer {
     if (this.server === undefined) {
       throw new Error('Http server is not setup up yet');
     }
-    if (this.stopped) {
+    if (this.stopping || this.stopped) {
       this.log.warn(`registerStaticDir called after stop`);
     }
 

src/core/server/http/http_service.ts

@@ -6,7 +6,7 @@
  * Side Public License, v 1.
  */
 
-import { Observable, Subscription, combineLatest } from 'rxjs';
+import { Observable, Subscription, combineLatest, of } from 'rxjs';
 import { first, map } from 'rxjs/operators';
 import { Server } from '@hapi/hapi';
 import { pick } from '@kbn/std';
@@ -69,7 +69,8 @@ export class HttpService
       configService.atPath<CspConfigType>(cspConfig.path),
       configService.atPath<ExternalUrlConfigType>(externalUrlConfig.path),
     ]).pipe(map(([http, csp, externalUrl]) => new HttpConfig(http, csp, externalUrl)));
-    this.httpServer = new HttpServer(logger, 'Kibana');
+    const shutdownTimeout$ = this.config$.pipe(map(({ shutdownTimeout }) => shutdownTimeout));
+    this.httpServer = new HttpServer(logger, 'Kibana', shutdownTimeout$);
     this.httpsRedirectServer = new HttpsRedirectServer(logger.get('http', 'redirect', 'server'));
   }
 
@@ -167,7 +168,7 @@ export class HttpService
       return;
     }
 
-    this.configSubscription.unsubscribe();
+    this.configSubscription?.unsubscribe();
     this.configSubscription = undefined;
 
     if (this.notReadyServer) {
@@ -179,7 +180,7 @@ export class HttpService
 
   private async runNotReadyServer(config: HttpConfig) {
     this.log.debug('starting NotReady server');
-    const httpServer = new HttpServer(this.logger, 'NotReady');
+    const httpServer = new HttpServer(this.logger, 'NotReady', of(config.shutdownTimeout));
     const { server } = await httpServer.setup(config);
     this.notReadyServer = server;
     // use hapi server while KibanaResponseFactory doesn't allow specifying custom headers

src/core/server/http/integration_tests/lifecycle_handlers.test.ts

@@ -7,6 +7,7 @@
  */
 
 import supertest from 'supertest';
+import moment from 'moment';
 import { BehaviorSubject } from 'rxjs';
 import { ByteSizeValue } from '@kbn/config-schema';
 
@@ -44,6 +45,7 @@ describe('core lifecycle handlers', () => {
         return new BehaviorSubject({
           hosts: ['localhost'],
           maxPayload: new ByteSizeValue(1024),
+          shutdownTimeout: moment.duration(30, 'seconds'),
           autoListen: true,
           ssl: {
             enabled: false,

src/core/server/http/test_utils.ts

@@ -7,6 +7,7 @@
  */
 
 import { BehaviorSubject } from 'rxjs';
+import moment from 'moment';
 import { REPO_ROOT } from '@kbn/dev-utils';
 import { ByteSizeValue } from '@kbn/config-schema';
 import { Env } from '../config';
@@ -44,6 +45,7 @@ configService.atPath.mockImplementation((path) => {
         allowFromAnyIp: true,
         ipAllowlist: [],
       },
+      shutdownTimeout: moment.duration(30, 'seconds'),
       keepaliveTimeout: 120_000,
       socketTimeout: 120_000,
     } as any);

src/core/server/server.ts

@@ -271,10 +271,10 @@ export class Server {
     this.log.debug('stopping server');
 
     await this.legacy.stop();
+    await this.http.stop(); // HTTP server has to stop before savedObjects and ES clients are closed to be able to gracefully attempt to resolve any pending requests
     await this.plugins.stop();
     await this.savedObjects.stop();
     await this.elasticsearch.stop();
-    await this.http.stop();
     await this.uiSettings.stop();
     await this.rendering.stop();
     await this.metrics.stop();

src/setup_node_env/ensure_node_preserve_symlinks.js

@@ -99,6 +99,13 @@
     return 0;
   };
 
+  // Since we are using `stdio: inherit`, the child process will receive
+  // the `SIGINT` and `SIGTERM` from the terminal.
+  // However, we want the parent process not to exit until the child does.
+  // Adding the following handlers achieves that.
+  process.on('SIGINT', function () {});
+  process.on('SIGTERM', function () {});
+
   var spawnResult = cp.spawnSync(nodeArgv[0], nodeArgs.concat(restArgs), { stdio: 'inherit' });
   process.exit(getExitCodeFromSpawnResult(spawnResult));
 })();