xds: Add header mutations library

xds/src/main/java/io/grpc/xds/internal/grpcservice/CachedChannelManager.java

@@ -0,0 +1,138 @@
+/*
+ * Copyright 2026 The 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.
+ */
+
+package io.grpc.xds.internal.grpcservice;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+import com.google.auto.value.AutoValue;
+import com.google.common.annotations.VisibleForTesting;
+import io.grpc.ManagedChannel;
+import io.grpc.xds.client.ConfiguredChannelCredentials.ChannelCredsConfig;
+import io.grpc.xds.internal.grpcservice.GrpcServiceConfig.GoogleGrpcConfig;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Function;
+
+/**
+ * Concrete class managing the lifecycle of a single ManagedChannel for a GrpcServiceConfig.
+ */
+public class CachedChannelManager {
+  private final Function<GrpcServiceConfig, ManagedChannel> channelCreator;
+  private final Object lock = new Object();
+
+  private final AtomicReference<ChannelHolder> channelHolder = new AtomicReference<>();
+  private boolean closed;
+
+  /**
+   * Default constructor for production that creates a channel using the config's target and
+   * credentials.
+   */
+  public CachedChannelManager() {
+    this(config -> {
+      GoogleGrpcConfig googleGrpc = config.googleGrpc();
+      return io.grpc.Grpc.newChannelBuilder(googleGrpc.target(),
+          googleGrpc.configuredChannelCredentials().channelCredentials()).build();
+    });
+  }
+
+  /**
+   * Constructor for testing to inject a channel creator.
+   */
+  @VisibleForTesting
+  public CachedChannelManager(Function<GrpcServiceConfig, ManagedChannel> channelCreator) {
+    this.channelCreator = checkNotNull(channelCreator, "channelCreator");
+  }
+
+  /**
+   * Returns a ManagedChannel for the given configuration. If the target or credentials config
+   * changes, the old channel is shut down and a new one is created.
+   */
+  public ManagedChannel getChannel(GrpcServiceConfig config) {
+    GoogleGrpcConfig googleGrpc = config.googleGrpc();
+    ChannelKey newChannelKey = ChannelKey.of(
+        googleGrpc.target(),
+        googleGrpc.configuredChannelCredentials().channelCredsConfig());
+
+    // 1. Fast path: Lock-free read
+    ChannelHolder holder = channelHolder.get();
+    if (holder != null && holder.channelKey().equals(newChannelKey)) {
+      return holder.channel();
+    }
+
+    ManagedChannel oldChannel = null;
+    ManagedChannel newChannel;
+
+    // 2. Slow path: Update with locking
+    synchronized (lock) {
+      if (closed) {
+        throw new IllegalStateException("CachedChannelManager is closed");
+      }
+      holder = channelHolder.get(); // Double check
+      if (holder != null && holder.channelKey().equals(newChannelKey)) {
+        return holder.channel();
+      }
+
+      // 3. Create inside lock to avoid creation storms
+      newChannel = channelCreator.apply(config);
+      ChannelHolder newHolder = ChannelHolder.create(newChannelKey, newChannel);
+
+      if (holder != null) {
+        oldChannel = holder.channel();
+      }
+      channelHolder.set(newHolder);
+    }
+
+    // 4. Shutdown outside lock
+    if (oldChannel != null) {
+      oldChannel.shutdown();
+    }
+
+    return newChannel;
+  }
+
+  /** Removes underlying resources on shutdown. */
+  public void close() {
+    synchronized (lock) {
+      closed = true;
+      ChannelHolder holder = channelHolder.getAndSet(null);
+      if (holder != null) {
+        holder.channel().shutdown();
+      }
+    }
+  }
+
+  @AutoValue
+  abstract static class ChannelKey {
+    static ChannelKey of(String target, ChannelCredsConfig credentialsConfig) {
+      return new AutoValue_CachedChannelManager_ChannelKey(target, credentialsConfig);
+    }
+
+    abstract String target();
+
+    abstract ChannelCredsConfig channelCredsConfig();
+  }
+
+  @AutoValue
+  abstract static class ChannelHolder {
+    static ChannelHolder create(ChannelKey channelKey, ManagedChannel channel) {
+      return new AutoValue_CachedChannelManager_ChannelHolder(channelKey, channel);
+    }
+
+    abstract ChannelKey channelKey();
+
+    abstract ManagedChannel channel();
+  }
+}

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

@@ -0,0 +1,32 @@
+/*
+ * Copyright 2024 The 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.
+ */
+
+package io.grpc.xds.internal.headermutations;
+
+import io.grpc.Status;
+import io.grpc.StatusException;
+
+/**
+ * Exception thrown when a header mutation is disallowed.
+ */
+public final class HeaderMutationDisallowedException extends StatusException {
+
+  private static final long serialVersionUID = 1L;
+
+  public HeaderMutationDisallowedException(String message) {
+    super(Status.INTERNAL.withDescription(message));
+  }
+}

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

@@ -0,0 +1,109 @@
+/*
+ * Copyright 2025 The 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.
+ */
+
+package io.grpc.xds.internal.headermutations;
+
+import com.google.common.collect.ImmutableList;
+import io.grpc.xds.internal.grpcservice.HeaderValueValidationUtils;
+import java.util.Collection;
+import java.util.Optional;
+import java.util.function.Predicate;
+
+/**
+ * The HeaderMutationFilter class is responsible for filtering header mutations based on a given set
+ * of rules.
+ */
+public class HeaderMutationFilter {
+  private final Optional<HeaderMutationRulesConfig> mutationRules;
+
+
+
+  public HeaderMutationFilter(Optional<HeaderMutationRulesConfig> mutationRules) {
+    this.mutationRules = mutationRules;
+  }
+
+  /**
+   * Filters the given header mutations based on the configured rules and returns the allowed
+   * mutations.
+   *
+   * @param mutations The header mutations to filter
+   * @return The allowed header mutations.
+   * @throws HeaderMutationDisallowedException if a disallowed mutation is encountered and the rules
+   *         specify that this should be an error.
+   */
+  public HeaderMutations filter(HeaderMutations mutations)
+      throws HeaderMutationDisallowedException {
+    ImmutableList<HeaderValueOption> allowedHeaders =
+        filterCollection(mutations.headers(), this::isDisallowed, this::isHeaderMutationAllowed);
+    ImmutableList<String> allowedHeadersToRemove =
+        filterCollection(mutations.headersToRemove(), this::isDisallowed,
+                this::isHeaderMutationAllowed);
+    return HeaderMutations.create(allowedHeaders, allowedHeadersToRemove);
+  }
+
+  /**
+   * A generic helper to filter a collection based on a predicate.
+   */
+  private <T> ImmutableList<T> filterCollection(Collection<T> items,
+      Predicate<T> isIgnoredPredicate, Predicate<T> isAllowedPredicate)
+      throws HeaderMutationDisallowedException {
+    ImmutableList.Builder<T> allowed = ImmutableList.builder();
+    for (T item : items) {
+      if (isIgnoredPredicate.test(item)) {
+        continue;
+      }
+      if (isAllowedPredicate.test(item)) {
+        allowed.add(item);
+      } else if (disallowIsError()) {
+        throw new HeaderMutationDisallowedException("Header mutation disallowed");
+      }
+    }
+    return allowed.build();
+  }
+
+  private boolean isDisallowed(String key) {
+    return HeaderValueValidationUtils.isDisallowed(key);
+  }
+
+  private boolean isDisallowed(HeaderValueOption option) {
+    return HeaderValueValidationUtils.isDisallowed(option.header());
+  }
+
+  private boolean isHeaderMutationAllowed(HeaderValueOption option) {
+    return isHeaderMutationAllowed(option.header().key());
+  }
+
+  private boolean isHeaderMutationAllowed(String headerName) {
+    return mutationRules.map(rules -> isHeaderMutationAllowed(headerName, rules))
+        .orElse(true);
+  }
+
+  private boolean isHeaderMutationAllowed(String headerName,
+          HeaderMutationRulesConfig rules) {
+    if (rules.disallowExpression().isPresent()
+        && rules.disallowExpression().get().matcher(headerName).matches()) {
+      return false;
+    }
+    if (rules.allowExpression().isPresent()) {
+      return rules.allowExpression().get().matcher(headerName).matches();
+    }
+    return !rules.disallowAll();
+  }
+
+  private boolean disallowIsError() {
+    return mutationRules.map(HeaderMutationRulesConfig::disallowIsError).orElse(false);
+  }
+}

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

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2025 The 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.
+ */
+
+package io.grpc.xds.internal.headermutations;
+
+import com.google.auto.value.AutoValue;
+import com.google.common.collect.ImmutableList;
+
+/** A collection of header mutations. */
+@AutoValue
+public abstract class HeaderMutations {
+
+  public static HeaderMutations create(ImmutableList<HeaderValueOption> headers,
+      ImmutableList<String> headersToRemove) {
+    return new AutoValue_HeaderMutations(headers, headersToRemove);
+  }
+
+  public abstract ImmutableList<HeaderValueOption> headers();
+
+  public abstract ImmutableList<String> headersToRemove();
+}