HTTP-Server: Graceful shutdown

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/http_config.test.ts

@@ -108,6 +108,33 @@ 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' })
+    ).toThrowErrorMatchingSnapshot();
+  });
+
+  test('should error if over 2 minutes', () => {
+    expect(() => config.schema.validate({ shutdownTimeout: '3m' })).toThrowErrorMatchingSnapshot();
+  });
+});
+
 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';
@@ -31,6 +32,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 }),
@@ -182,6 +192,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
@@ -217,6 +228,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 { of } from 'rxjs';
+import moment from 'moment';
 
 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,96 @@ 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: req.route });
+      }
+    );
+    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({
+      method: 'post',
+      path: '/',
+      options: {
+        authRequired: true,
+        xsrfRequired: true,
+        tags: [],
+        timeout: {
+          payload: 10000,
+        },
+        body: {
+          parse: true, // hapi populates the default
+          maxBytes: 1024, // hapi populates the default
+          accepts: ['application/json'],
+          output: 'data',
+        },
+      },
+    });
+    // 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 type { Observable } from 'rxjs';
+import { take } from 'rxjs/operators';
 import { Logger, LoggerFactory } from '../logging';
 import { HttpConfig } from './http_config';
 import { adoptToHapiAuthFormat, AuthenticationHandler } from './lifecycle/auth';
@@ -87,7 +90,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 +125,7 @@ export class HttpServer {
     this.setupConditionalCompression(config);
     this.setupResponseLogging();
     this.setupRequestStateAssignment(config);
+    this.setupGracefulShutdownHandlers();
 
     return {
       registerRouter: this.registerRouter.bind(this),
@@ -221,10 +229,16 @@ export class HttpServer {
     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();
     }
   }
 
@@ -246,6 +260,18 @@ export class HttpServer {
     }
   }
 
+  private setupGracefulShutdownHandlers() {
+    this.registerOnPreRouting((request, response, toolkit) => {
+      if (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;