connnectors: feature warnings on incompatible plugins

apollo-router/src/plugins/connectors/configuration.rs

@@ -1,8 +1,10 @@
 use std::collections::HashMap;
+use std::collections::HashSet;
 use std::sync::Arc;
 
 use apollo_federation::sources::connect::expand::Connectors;
 use apollo_federation::sources::connect::CustomConfiguration;
+use itertools::Itertools as _;
 use schemars::JsonSchema;
 use serde::Deserialize;
 use serde::Serialize;
@@ -66,8 +68,15 @@ pub(crate) struct SourceConfiguration {
 }
 
 /// Modifies connectors with values from the configuration
-pub(crate) fn apply_config(config: &Configuration, mut connectors: Connectors) -> Connectors {
-    let Some(config) = config.apollo_plugins.plugins.get(PLUGIN_NAME) else {
+pub(crate) fn apply_config(
+    router_config: &Configuration,
+    mut connectors: Connectors,
+) -> Connectors {
+    // Enabling connectors might end up interfering with other router features, so we insert warnings
+    // into the logs for any incompatibilites found.
+    warn_incompatible_plugins(router_config, &connectors);
+
+    let Some(config) = router_config.apollo_plugins.plugins.get(PLUGIN_NAME) else {
         return connectors;
     };
     let Ok(config) = serde_json::from_value::<ConnectorsConfig>(config.clone()) else {
@@ -96,3 +105,237 @@ pub(crate) fn apply_config(config: &Configuration, mut connectors: Connectors) -
     }
     connectors
 }
+
+/// Warn about possible incompatibilities with other router features / plugins.
+///
+/// Connectors do not currently work with some of the existing router
+/// features, so we need to inform the user when those features are
+/// detected as being enabled.
+fn warn_incompatible_plugins(config: &Configuration, connectors: &Connectors) {
+    /// Generate a consistent warning message for a specified plugin
+    fn msg(plugin_name: &str) -> String {
+        format!("plugin `{}` is enabled for connector-enabled subgraphs, which is currently unsupported. See https://go.apollo.dev/connectors/incompat for more info", plugin_name)
+    }
+
+    let connector_enabled_subgraphs: HashSet<&String> = connectors
+        .by_service_name
+        .values()
+        .map(|v| &v.id.subgraph_name)
+        .collect();
+
+    // If we don't have any connector-enabled subgraphs, then no need to warn
+    if connector_enabled_subgraphs.is_empty() {
+        return;
+    }
+
+    // Plugins tend to have a few forms of specifying which subgraph to target:
+    // - A catch-all `all`
+    // - By the subgraph's actual name under a `subgraphs` key
+    // In either case, the configuration will have a prefix which then ends in the
+    // target subgraph, so we keep track of all of those prefixes which aren't
+    // supported by connector-enabled subgraphs.
+    //
+    // TODO: Each of these config options come from a corresponding plugin which
+    // is identified by its name. These are currently hardcoded here, so it'd be
+    // nice to extract them from the plugins themselves...
+    //
+    // Note: Some of these also allow for enabling / disabling (and overriding
+    // global options), so we need to know if that is the case when collecting
+    // which subgraphs might trigger incompatibilities.
+    struct IncompatiblePlugin {
+        enabled: bool,
+        /// If the configuration allows for overriding on a subgraph-level whether
+        /// to enable or disable a feature.
+        subgraph_can_override: bool,
+        /// The name of the plugin
+        plugin: &'static str,
+        /// The set of keys needed to drill through to reach the subgraph
+        /// configuration.
+        drill_prefix: &'static [&'static str],
+    }
+    let incompatible_prefixes = [
+        IncompatiblePlugin {
+            enabled: config.apq.enabled,
+            subgraph_can_override: true,
+            plugin: "apq",
+            drill_prefix: &["subgraph"],
+        },
+        IncompatiblePlugin {
+            enabled: true,
+            subgraph_can_override: false,
+            plugin: "authentication",
+            drill_prefix: &["subgraph"],
+        },
+        IncompatiblePlugin {
+            enabled: config.batching.enabled,
+            subgraph_can_override: true,
+            plugin: "batching",
+            drill_prefix: &["subgraph"],
+        },
+        IncompatiblePlugin {
+            enabled: true,
+            subgraph_can_override: false,
+            plugin: "coprocessor",
+            drill_prefix: &["subgraph"],
+        },
+        IncompatiblePlugin {
+            enabled: true,
+            subgraph_can_override: false,
+            plugin: "headers",
+            drill_prefix: &[],
+        },
+        IncompatiblePlugin {
+            enabled: config
+                .apollo_plugins
+                .plugins
+                .get("preview_entity_cache")
+                .and_then(|p| p.get("enabled"))
+                .and_then(serde_json::Value::as_bool)
+                .unwrap_or_default(),
+            subgraph_can_override: true,
+            plugin: "preview_entity_cache",
+            drill_prefix: &["subgraph"],
+        },
+        IncompatiblePlugin {
+            enabled: true,
+            subgraph_can_override: false,
+            plugin: "telemetry",
+            drill_prefix: &["apollo", "errors", "subgraph"],
+        },
+        IncompatiblePlugin {
+            enabled: true,
+            subgraph_can_override: false,
+            plugin: "telemetry",
+            drill_prefix: &["exporters", "metrics", "common", "attributes", "subgraph"],
+        },
+        IncompatiblePlugin {
+            enabled: true,
+            subgraph_can_override: false,
+            plugin: "tls",
+            drill_prefix: &["subgraph"],
+        },
+        IncompatiblePlugin {
+            enabled: true,
+            subgraph_can_override: false,
+            plugin: "traffic_shaping",
+            drill_prefix: &[],
+        },
+    ];
+
+    // Note: Some of the incopmpatible plugins are hoisted up into individual
+    // properties on the config object, so we operate on the actual yaml to
+    // consolidate how we handle core features vs arbitrary plugins.
+    //
+    // Note: Execution of this entire chain of validation methods shouldn't happen
+    // if the configuration is invalid, but we add a debug log just in case.
+    let Some(raw_config) = config
+        .validated_yaml
+        .as_ref()
+        .and_then(serde_json::Value::as_object)
+    else {
+        tracing::debug!("configuration is invalid, skipping connector incompatibility checks...");
+        return;
+    };
+
+    for IncompatiblePlugin {
+        enabled,
+        subgraph_can_override,
+        plugin,
+        drill_prefix,
+    } in incompatible_prefixes
+    {
+        // If the plugin is not enabled, no need to process it
+        if !enabled {
+            continue;
+        }
+
+        // Plugin configuration is only populated if the user has specified it,
+        // so we can skip any that are missing.
+        let Some(plugin_config) = raw_config.get(plugin) else {
+            continue;
+        };
+
+        // Drill into the prefix
+        let Some(prefixed_config) = drill_prefix
+            .iter()
+            .try_fold(plugin_config, |acc, next| acc.get(next))
+        else {
+            continue;
+        };
+
+        // Grab all of the configured subgraphs for this plugin
+        // Note: If the plugin supports overriding on a per-subgraph level, then
+        // we'll need to partition based on an enabled flag per subgraph.
+        let empty = serde_json::Map::new();
+        let configured = prefixed_config
+            .get("subgraphs")
+            .and_then(serde_json::Value::as_object)
+            .unwrap_or(&empty);
+
+        let (enabled, disabled): (HashSet<&String>, HashSet<&String>) =
+            configured.iter().partition_map(|(name, subgraph_conf)| {
+                if subgraph_can_override {
+                    let enabled = subgraph_conf
+                        .get("enabled")
+                        .and_then(serde_json::Value::as_bool)
+                        .unwrap_or_default();
+                    match enabled {
+                        true => itertools::Either::Left(name),
+                        false => itertools::Either::Right(name),
+                    }
+                } else {
+                    itertools::Either::Left(name)
+                }
+            });
+
+        // If a plugin allows for overriding enablement, then it also allows
+        // for enabling / disabling on `all`. Otherwise, just the presence of the
+        // `all` key is enough to signal that all should be targeted.
+        let all_enabled = if subgraph_can_override {
+            prefixed_config
+                .get("all")
+                .and_then(|a| a.get("enabled"))
+                .and_then(serde_json::Value::as_bool)
+                .unwrap_or_default()
+        } else {
+            prefixed_config.get("all").is_some()
+        };
+
+        // Now actually calculate which are incompatible
+        // Note: We need to collect here because we need to know if the iterator
+        // is empty or not when printing the warning message.
+        let incompatible = if all_enabled {
+            // If all are enabled, then we can subtract out those which are disabled explicitly
+            connector_enabled_subgraphs
+                .difference(&disabled)
+                .copied()
+                .collect::<HashSet<&String>>()
+        } else {
+            // Otherwise, then we only care about those explicitly enabled
+            enabled
+                .intersection(&connector_enabled_subgraphs)
+                .copied()
+                .collect::<HashSet<&String>>()
+        };
+
+        if !incompatible.is_empty() {
+            tracing::warn!(
+                subgraphs = incompatible.iter().join(","),
+                message = msg(plugin)
+            );
+        }
+    }
+
+    // There are a few plugins which influence all subgraphs, regardless
+    // of configuration, so we warn about these statically here if we have
+    // any connector-enabled subgraphs.
+    let incompatible_plugins = ["demand_control", "rhai"];
+    for plugin_name in incompatible_plugins {
+        if config.apollo_plugins.plugins.get(plugin_name).is_some() {
+            tracing::warn!(
+                subgraphs = connector_enabled_subgraphs.iter().join(","),
+                message = msg(plugin_name)
+            );
+        }
+    }
+}

apollo-router/tests/fixtures/connectors/incompatible.router.yaml

@@ -0,0 +1,71 @@
+# A valid router configuration with every incompatible
+# router feature turned on for testing warnings.
+
+apq:
+  subgraph:
+    all:
+      enabled: true
+authentication:
+  subgraph:
+    all:
+      aws_sig_v4:
+        default_chain:
+          profile_name: "my-test-profile"
+          region: "us-east-1"
+          service_name: "lambda"
+          assume_role:
+            role_arn: "test-arn"
+            session_name: "test-session"
+            external_id: "test-id"
+batching:
+  enabled: true
+  mode: batch_http_link
+  subgraph:
+    all:
+      enabled: false
+    subgraphs:
+      connectors:
+        enabled: true
+coprocessor:
+  url: http://127.0.0.1:8081
+  subgraph:
+    all:
+      request:
+        headers: true
+demand_control:
+  enabled: true
+  mode: measure
+  strategy:
+    static_estimated:
+      list_size: 10
+      max: 1000
+headers:
+  all:
+    request:
+      - propagate:
+          matching: ^upstream-header-.*
+      - remove:
+          named: "x-legacy-account-id"
+preview_entity_cache:
+  enabled: true
+  subgraph:
+    all:
+      enabled: true
+rhai:
+  scripts: "/rhai/scripts/directory"
+telemetry:
+  apollo:
+    errors:
+      subgraph:
+        all:
+          send: true
+          redact: false
+tls:
+  subgraph:
+    all:
+      certificate_authorities: "${file./nonsense/path.crt}"
+traffic_shaping:
+  subgraphs:
+    connectors:
+      deduplicate_query: false
+      compression: gzip