xds: Add header mutations library (grpc#12494)

This commit introduces a library for handling header mutations as
specified by the xDS protocol. This library provides the core
functionality for modifying request and response headers based on a set
of rules.

The main components of this library are:
- `HeaderMutator`: Applies header mutations to `Metadata` objects.
- `HeaderMutationFilter`: Filters header mutations based on a set of
configurable rules, such as disallowing mutations of system headers.
- `HeaderMutations`: A value class that represents the set of mutations
to be applied to request and response headers.
- `HeaderMutationDisallowedException`: An exception that is thrown when
a disallowed header mutation is attempted.

This commit also includes comprehensive unit tests for the new library.

Author: sauravzg

xds/src/main/java/io/grpc/xds/internal/headermutations/HeaderMutationFilter.java

@@ -62,10 +62,14 @@ private <T> ImmutableList<T> filterCollection(Collection<T> items,
       throws HeaderMutationDisallowedException {
     ImmutableList.Builder<T> allowed = ImmutableList.builder();
     for (T item : items) {
-      if (isIgnoredPredicate.test(item)) {
-        continue;
-      }
-      if (isAllowedPredicate.test(item)) {
+      boolean isIgnored = isIgnoredPredicate.test(item);
+      boolean isAllowed = isAllowedPredicate.test(item);
+
+      // TODO(sauravzg): The specification is ambiguous regarding whether system headers
+      // should be silently ignored or trigger an error when disallowIsError is enabled.
+      // We default to triggering errors matching Envoy's implementation.
+      // Ref: https://github.com/grpc/proposal/pull/481#discussion_r3124453674
+      if (!isIgnored && isAllowed) {
         allowed.add(item);
       } else if (disallowIsError()) {
         throw new HeaderMutationDisallowedException("Header mutation disallowed");
@@ -97,8 +101,9 @@ private boolean isHeaderMutationAllowed(String headerName,
         && rules.disallowExpression().get().matcher(headerName).matches()) {
       return false;
     }
-    if (rules.allowExpression().isPresent()) {
-      return rules.allowExpression().get().matcher(headerName).matches();
+    if (rules.allowExpression().isPresent()
+        && rules.allowExpression().get().matcher(headerName).matches()) {
+      return true;
     }
     return !rules.disallowAll();
   }

xds/src/main/java/io/grpc/xds/internal/headermutations/HeaderMutator.java

@@ -23,8 +23,8 @@
 import java.util.logging.Logger;
 
 /**
- * The HeaderMutator class is an implementation of the HeaderMutator interface. It provides methods
- * to apply header mutations to a given set of headers based on a given set of rules.
+ * The HeaderMutator provides methods to apply header mutations to a given set of headers based on a
+ * given set of rules.
  */
 public class HeaderMutator {
 
@@ -71,23 +71,29 @@ private void updateHeader(final HeaderValueOption option, Metadata mutableHeader
 
     if (header.key().endsWith(Metadata.BINARY_HEADER_SUFFIX)) {
       if (header.rawValue().isPresent()) {
-        updateHeader(action, Metadata.Key.of(header.key(), Metadata.BINARY_BYTE_MARSHALLER),
-            header.rawValue().get().toByteArray(), mutableHeaders, keepEmptyValue);
+        byte[] value = header.rawValue().get().toByteArray();
+        if (value.length > 0 || keepEmptyValue) {
+          updateHeader(action, Metadata.Key.of(header.key(), Metadata.BINARY_BYTE_MARSHALLER),
+                  value, mutableHeaders);
+        }
       } else {
         logger.fine("Missing binary rawValue for header: " + header.key());
       }
     } else {
       if (header.value().isPresent()) {
-        updateHeader(action, Metadata.Key.of(header.key(), Metadata.ASCII_STRING_MARSHALLER),
-            header.value().get(), mutableHeaders, keepEmptyValue);
+        String value = header.value().get();
+        if (!value.isEmpty() || keepEmptyValue) {
+          updateHeader(action, Metadata.Key.of(header.key(), Metadata.ASCII_STRING_MARSHALLER),
+                  value, mutableHeaders);
+        }
       } else {
         logger.fine("Missing value for header: " + header.key());
       }
     }
   }
 
   private <T> void updateHeader(final HeaderAppendAction action, final Metadata.Key<T> key,
-      final T value, Metadata mutableHeaders, boolean keepEmptyValue) {
+      final T value, Metadata mutableHeaders) {
     switch (action) {
       case APPEND_IF_EXISTS_OR_ADD:
         mutableHeaders.put(key, value);
@@ -112,33 +118,6 @@ private <T> void updateHeader(final HeaderAppendAction action, final Metadata.Ke
         // Should be unreachable unless there's a proto schema mismatch.
         logger.fine("Unknown HeaderAppendAction: " + action);
     }
-
-    if (!keepEmptyValue) {
-      checkAndRemoveEmpty(key, mutableHeaders);
-    }
-  }
-
-  private <T> void checkAndRemoveEmpty(Metadata.Key<T> key, Metadata headers) {
-    Iterable<T> values = headers.getAll(key);
-    if (values == null) {
-      return;
-    }
-    boolean allEmpty = true;
-    for (T val : values) {
-      if (val instanceof String) {
-        if (!((String) val).isEmpty()) {
-          allEmpty = false;
-          break;
-        }
-      } else if (val instanceof byte[]) {
-        if (((byte[]) val).length > 0) {
-          allEmpty = false;
-          break;
-        }
-      }
-    }
-    if (allEmpty) {
-      headers.discardAll(key);
-    }
   }
 }
+