Merge pull request #1952 from b0b3rt/grpc-js_compression_support

grpc-js: Allow per-channel request compression from the client and decompression from the server

Author: murgatroid99

packages/grpc-js/package.json

@@ -52,13 +52,15 @@
     "test": "gulp test",
     "check": "gts check src/**/*.ts",
     "fix": "gts fix src/*.ts",
-    "pretest": "npm run generate-types && npm run compile",
+    "pretest": "npm run generate-types && npm run generate-test-types && npm run compile",
     "posttest": "npm run check && madge -c ./build/src",
-    "generate-types": "proto-loader-gen-types --keepCase --longs String --enums String --defaults --oneofs --includeComments --includeDirs proto/ -O src/generated/ --grpcLib ../index channelz.proto"
+    "generate-types": "proto-loader-gen-types --keepCase --longs String --enums String --defaults --oneofs --includeComments --includeDirs proto/ --include-dirs test/fixtures/ -O src/generated/ --grpcLib ../index channelz.proto",
+    "generate-test-types": "proto-loader-gen-types --keepCase --longs String --enums String --defaults --oneofs --includeComments --include-dirs test/fixtures/ -O test/generated/ --grpcLib ../../src/index test_service.proto"
   },
   "dependencies": {
     "@grpc/proto-loader": "^0.6.4",
-    "@types/node": ">=12.12.47"
+    "@types/node": ">=12.12.47",
+    "@types/semver": "^7.3.9"
   },
   "files": [
     "src/**/*.ts",

packages/grpc-js/src/channel-options.ts

@@ -15,6 +15,8 @@
  *
  */
 
+import { CompressionAlgorithms } from './compression-algorithms';
+
 /**
  * An interface that contains options used when initializing a Channel instance.
  */
@@ -36,6 +38,7 @@ export interface ChannelOptions {
   'grpc.enable_http_proxy'?: number;
   'grpc.http_connect_target'?: string;
   'grpc.http_connect_creds'?: string;
+  'grpc.default_compression_algorithm'?: CompressionAlgorithms;
   'grpc.enable_channelz'?: number;
   'grpc-node.max_session_memory'?: number;
   // eslint-disable-next-line @typescript-eslint/no-explicit-any

packages/grpc-js/src/channel.ts

@@ -45,7 +45,6 @@ import { MaxMessageSizeFilterFactory } from './max-message-size-filter';
 import { mapProxyName } from './http_proxy';
 import { GrpcUri, parseUri, uriToString } from './uri-parser';
 import { ServerSurfaceCall } from './server-call';
-import { SurfaceCall } from './call';
 import { Filter } from './filter';
 
 import { ConnectivityState } from './connectivity-state';
@@ -333,7 +332,7 @@ export class ChannelImplementation implements Channel {
       new CallCredentialsFilterFactory(this),
       new DeadlineFilterFactory(this),
       new MaxMessageSizeFilterFactory(this.options),
-      new CompressionFilterFactory(this),
+      new CompressionFilterFactory(this, this.options),
     ]);
     this.trace('Channel constructed with options ' + JSON.stringify(options, undefined, 2));
   }

packages/grpc-js/src/compression-algorithms.ts

@@ -0,0 +1,22 @@
+/*
+ * Copyright 2021 gRPC authors.
+ *
+ * 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.
+ *
+ */
+
+export enum CompressionAlgorithms {
+  identity = 0,
+  deflate = 1,
+  gzip = 2
+};

packages/grpc-js/src/compression-filter.ts

@@ -17,11 +17,25 @@
 
 import * as zlib from 'zlib';
 
-import { Call, WriteFlags, WriteObject } from './call-stream';
+import { Call, WriteObject, WriteFlags } from './call-stream';
 import { Channel } from './channel';
+import { ChannelOptions } from './channel-options';
+import { CompressionAlgorithms } from './compression-algorithms';
+import { LogVerbosity } from './constants';
 import { BaseFilter, Filter, FilterFactory } from './filter';
+import * as logging from './logging';
 import { Metadata, MetadataValue } from './metadata';
 
+const isCompressionAlgorithmKey = (key: number): key is CompressionAlgorithms => {
+  return typeof key === 'number' && typeof CompressionAlgorithms[key] === 'string';
+}
+
+type CompressionAlgorithm = keyof typeof CompressionAlgorithms;
+
+type SharedCompressionFilterConfig = {
+  serverSupportedEncodingHeader?: string;
+};
+
 abstract class CompressionHandler {
   protected abstract compressMessage(message: Buffer): Promise<Buffer>;
   protected abstract decompressMessage(data: Buffer): Promise<Buffer>;
@@ -167,10 +181,45 @@ function getCompressionHandler(compressionName: string): CompressionHandler {
 export class CompressionFilter extends BaseFilter implements Filter {
   private sendCompression: CompressionHandler = new IdentityHandler();
   private receiveCompression: CompressionHandler = new IdentityHandler();
+  private currentCompressionAlgorithm: CompressionAlgorithm = 'identity';
+
+  constructor(channelOptions: ChannelOptions, private sharedFilterConfig: SharedCompressionFilterConfig) {
+    super();
+
+    const compressionAlgorithmKey = channelOptions['grpc.default_compression_algorithm'];
+    if (compressionAlgorithmKey !== undefined) {
+      if (isCompressionAlgorithmKey(compressionAlgorithmKey)) {
+        const clientSelectedEncoding = CompressionAlgorithms[compressionAlgorithmKey] as CompressionAlgorithm;
+        const serverSupportedEncodings = sharedFilterConfig.serverSupportedEncodingHeader?.split(',');
+        /**
+         * There are two possible situations here:
+         * 1) We don't have any info yet from the server about what compression it supports
+         *    In that case we should just use what the client tells us to use
+         * 2) We've previously received a response from the server including a grpc-accept-encoding header
+         *    In that case we only want to use the encoding chosen by the client if the server supports it
+         */
+        if (!serverSupportedEncodings || serverSupportedEncodings.includes(clientSelectedEncoding)) {
+          this.currentCompressionAlgorithm = clientSelectedEncoding;
+          this.sendCompression = getCompressionHandler(this.currentCompressionAlgorithm);
+        }
+      } else {
+        logging.log(LogVerbosity.ERROR, `Invalid value provided for grpc.default_compression_algorithm option: ${compressionAlgorithmKey}`);
+      }
+    }
+  }
+
   async sendMetadata(metadata: Promise<Metadata>): Promise<Metadata> {
     const headers: Metadata = await metadata;
     headers.set('grpc-accept-encoding', 'identity,deflate,gzip');
     headers.set('accept-encoding', 'identity');
+
+    // No need to send the header if it's "identity" -  behavior is identical; save the bandwidth
+    if (this.currentCompressionAlgorithm === 'identity') {
+      headers.remove('grpc-encoding');
+    } else {
+      headers.set('grpc-encoding', this.currentCompressionAlgorithm);
+    }
+
     return headers;
   }
 
@@ -183,6 +232,19 @@ export class CompressionFilter extends BaseFilter implements Filter {
       }
     }
     metadata.remove('grpc-encoding');
+
+    /* Check to see if the compression we're using to send messages is supported by the server
+     * If not, reset the sendCompression filter and have it use the default IdentityHandler */
+    const serverSupportedEncodingsHeader = metadata.get('grpc-accept-encoding')[0] as string | undefined;
+    if (serverSupportedEncodingsHeader) {
+      this.sharedFilterConfig.serverSupportedEncodingHeader = serverSupportedEncodingsHeader;
+      const serverSupportedEncodings = serverSupportedEncodingsHeader.split(',');
+
+      if (!serverSupportedEncodings.includes(this.currentCompressionAlgorithm)) {
+        this.sendCompression = new IdentityHandler();
+        this.currentCompressionAlgorithm = 'identity';
+      }
+    }
     metadata.remove('grpc-accept-encoding');
     return metadata;
   }
@@ -192,10 +254,13 @@ export class CompressionFilter extends BaseFilter implements Filter {
      * and the output is a framed and possibly compressed message. For this
      * reason, this filter should be at the bottom of the filter stack */
     const resolvedMessage: WriteObject = await message;
-    const compress =
-      resolvedMessage.flags === undefined
-        ? false
-        : (resolvedMessage.flags & WriteFlags.NoCompress) === 0;
+    let compress: boolean;
+    if (this.sendCompression instanceof IdentityHandler) {
+      compress = false;
+    } else {
+      compress = ((resolvedMessage.flags ?? 0) & WriteFlags.NoCompress) === 0;
+    }
+
     return {
       message: await this.sendCompression.writeMessage(
         resolvedMessage.message,
@@ -216,8 +281,9 @@ export class CompressionFilter extends BaseFilter implements Filter {
 
 export class CompressionFilterFactory
   implements FilterFactory<CompressionFilter> {
-  constructor(private readonly channel: Channel) {}
+    private sharedFilterConfig: SharedCompressionFilterConfig = {};
+  constructor(private readonly channel: Channel, private readonly options: ChannelOptions) {}
   createFilter(callStream: Call): CompressionFilter {
-    return new CompressionFilter();
+    return new CompressionFilter(this.options, this.sharedFilterConfig);
   }
 }

packages/grpc-js/src/index.ts

@@ -25,6 +25,7 @@ import {
 import { CallCredentials, OAuth2Client } from './call-credentials';
 import { Deadline, StatusObject } from './call-stream';
 import { Channel, ChannelImplementation } from './channel';
+import { CompressionAlgorithms } from './compression-algorithms';
 import { ConnectivityState } from './connectivity-state';
 import { ChannelCredentials } from './channel-credentials';
 import {
@@ -126,6 +127,7 @@ export {
   Status as status,
   ConnectivityState as connectivityState,
   Propagate as propagate,
+  CompressionAlgorithms as compressionAlgorithms
   // TODO: Other constants as well
 };