connectors: add warnings when using plugins

This commit adds a few informative warnings when using connectors
with certain plugins which are currently incompatible. These
warnings should better inform router operators of certain features
which aren't yet supported when used in tandem with connectors.

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,102 @@ 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) {
+    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...
+    let incompatible_prefixes = [
+        ("apq", ["subgraph"].as_slice()),
+        ("authentication", &["subgraph"]),
+        ("batching", &["subgraph"]),
+        ("coprocessor", &["subgraph"]),
+        ("headers", &[]),
+        (
+            "telemetry",
+            &["exporters", "metrics", "common", "attributes", "subgraph"],
+        ),
+        ("preview_entity_cache", &["subgraph"]),
+        ("telemetry", &["apollo", "errors", "subgraph"]),
+        ("traffic_shaping", &[]),
+    ];
+
+    for (plugin_name, prefix) in incompatible_prefixes {
+        // Plugin configuration is only populated if the user has specified it,
+        // so we can skip any that are missing.
+        let Some(plugin_config) = config.apollo_plugins.plugins.get(plugin_name) else {
+            continue;
+        };
+
+        // Drill into the prefix
+        let Some(prefixed_config) = prefix
+            .iter()
+            .try_fold(plugin_config, |acc, next| acc.get(next))
+        else {
+            continue;
+        };
+
+        // Check if any of the connector enabled subgraphs are targeted
+        let incompatible_subgraphs = if prefixed_config.get("all").is_some() {
+            // If all is configured, then all connector-enabled subgraphs are affected.
+            &connector_enabled_subgraphs
+        } else if let Some(subgraphs) = prefixed_config.get("subgraphs") {
+            // Otherwise, we'll need to do a set intersection between the list of connector-enabled
+            // subgraphs and configured subgraphs to see which, if any, are affected.
+            let configured = subgraphs
+                .as_object()
+                .map(|o| o.keys().collect())
+                .unwrap_or(HashSet::new());
+
+            &configured
+                .intersection(&connector_enabled_subgraphs)
+                .copied()
+                .collect()
+        } else {
+            &HashSet::new()
+        };
+
+        if !incompatible_subgraphs.is_empty() {
+            tracing::warn!(
+                subgraphs = incompatible_subgraphs.iter().join(","),
+                "plugin `{plugin_name}` is enabled for connector-enabled subgraphs, which is not yet supported. See https://go.apollo.dev/INSERT_DOCS_LINK for more info"
+            );
+        }
+    }
+
+    // Lastly, 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 in incompatible_plugins {
+        if config.apollo_plugins.plugins.get(plugin).is_some() {
+            tracing::warn!(
+                subgraphs = connector_enabled_subgraphs.iter().join(","),
+                "plugin `{plugin}` is enabled for connector-enabled subgraphs, which is not yet supported. See https://go.apollo.dev/INSERT_DOCS_LINK for more info"
+            );
+        }
+    }
+}

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

@@ -0,0 +1,33 @@
+# A valid router configuration with every incompatible
+# router feature turned on for testing warnings.
+
+apq:
+  subgraph:
+    subgraphs:
+      connectors:
+        enabled: true
+coprocessor:
+  url: http://127.0.0.1:8081
+  subgraph:
+    all:
+      request:
+        headers: true
+headers:
+  all:
+    request:
+      - propagate:
+          matching: ^upstream-header-.*
+      - remove:
+          named: "x-legacy-account-id"
+telemetry:
+  apollo:
+    errors:
+      subgraph:
+        all:
+          send: true
+          redact: false
+traffic_shaping:
+  subgraphs:
+    connectors:
+      deduplicate_query: false
+      compression: gzip

apollo-router/tests/fixtures/connectors/quickstart.graphql

@@ -0,0 +1,158 @@
+schema
+    @link(url: "https://specs.apollo.dev/link/v1.0")
+    @link(url: "https://specs.apollo.dev/join/v0.5", for: EXECUTION)
+    @link(url: "https://specs.apollo.dev/connect/v0.1", for: EXECUTION)
+    @join__directive(
+        graphs: [CONNECTORS]
+        name: "link"
+        args: {
+            url: "https://specs.apollo.dev/connect/v0.1"
+            import: ["@connect", "@source"]
+        }
+    )
+    @join__directive(
+        graphs: [CONNECTORS]
+        name: "source"
+        args: {
+            name: "jsonPlaceholder"
+            http: { baseURL: "https://jsonplaceholder.typicode.com/" }
+        }
+    ) {
+    query: Query
+}
+
+directive @join__directive(
+    graphs: [join__Graph!]
+    name: String!
+    args: join__DirectiveArguments
+) repeatable on SCHEMA | OBJECT | INTERFACE | FIELD_DEFINITION
+
+directive @join__enumValue(graph: join__Graph!) repeatable on ENUM_VALUE
+
+directive @join__field(
+    graph: join__Graph
+    requires: join__FieldSet
+    provides: join__FieldSet
+    type: String
+    external: Boolean
+    override: String
+    usedOverridden: Boolean
+    overrideLabel: String
+    contextArguments: [join__ContextArgument!]
+) repeatable on FIELD_DEFINITION | INPUT_FIELD_DEFINITION
+
+directive @join__graph(name: String!, url: String!) on ENUM_VALUE
+
+directive @join__implements(
+    graph: join__Graph!
+    interface: String!
+) repeatable on OBJECT | INTERFACE
+
+directive @join__type(
+    graph: join__Graph!
+    key: join__FieldSet
+    extension: Boolean! = false
+    resolvable: Boolean! = true
+    isInterfaceObject: Boolean! = false
+) repeatable on OBJECT | INTERFACE | UNION | ENUM | INPUT_OBJECT | SCALAR
+
+directive @join__unionMember(
+    graph: join__Graph!
+    member: String!
+) repeatable on UNION
+
+directive @link(
+    url: String
+    as: String
+    for: link__Purpose
+    import: [link__Import]
+) repeatable on SCHEMA
+
+input join__ContextArgument {
+    name: String!
+    type: String!
+    context: String!
+    selection: join__FieldValue!
+}
+
+scalar join__DirectiveArguments
+
+scalar join__FieldSet
+
+scalar join__FieldValue
+
+enum join__Graph {
+    CONNECTORS @join__graph(name: "connectors", url: "none")
+}
+
+scalar link__Import
+
+enum link__Purpose {
+    """
+    `SECURITY` features provide metadata necessary to securely resolve fields.
+    """
+    SECURITY
+
+    """
+    `EXECUTION` features provide metadata necessary for operation execution.
+    """
+    EXECUTION
+}
+
+type Post @join__type(graph: CONNECTORS) {
+    id: ID!
+    body: String
+    title: String
+    author: User
+}
+
+type Query @join__type(graph: CONNECTORS) {
+    posts: [Post]
+        @join__directive(
+            graphs: [CONNECTORS]
+            name: "connect"
+            args: {
+                source: "jsonPlaceholder"
+                http: { GET: "/posts" }
+                selection: "id\ntitle\nbody\nauthor: { id: userId }"
+            }
+        )
+    post(id: ID!): Post
+        @join__directive(
+            graphs: [CONNECTORS]
+            name: "connect"
+            args: {
+                source: "jsonPlaceholder"
+                http: { GET: "/posts/{$args.id}" }
+                selection: "id\ntitle\nbody\nauthor: { id: userId }"
+                entity: true
+            }
+        )
+    user(id: ID!): User
+        @join__directive(
+            graphs: [CONNECTORS]
+            name: "connect"
+            args: {
+                source: "jsonPlaceholder"
+                http: { GET: "/users/{$args.id}" }
+                selection: "id\nname\nusername"
+                entity: true
+            }
+        )
+}
+
+type User @join__type(graph: CONNECTORS) {
+    id: ID!
+    name: String
+    username: String
+    posts: [Post]
+        @join__directive(
+            graphs: [CONNECTORS]
+            name: "connect"
+            args: {
+                source: "jsonPlaceholder"
+                http: { GET: "/users/{$this.id}/posts" }
+                selection: "id\ntitle\nbody"
+            }
+        )
+}

apollo-router/tests/integration/connectors.rs

@@ -0,0 +1,41 @@
+use std::path::PathBuf;
+
+use tower::BoxError;
+
+use crate::integration::IntegrationTest;
+
+const INCOMPATIBLE_PLUGINS_CONFIG: &str =
+    include_str!("../fixtures/connectors/incompatible.router.yaml");
+
+#[tokio::test(flavor = "multi_thread")]
+async fn test_incompatible_plugin_warnings() -> Result<(), BoxError> {
+    // Ensure that we have the test keys before running
+    // Note: The [IntegrationTest] ensures that these test credentials get
+    // set before running the router.
+    if std::env::var("TEST_APOLLO_KEY").is_err() || std::env::var("TEST_APOLLO_GRAPH_REF").is_err()
+    {
+        return Ok(());
+    };
+
+    let mut router = IntegrationTest::builder()
+        .config(INCOMPATIBLE_PLUGINS_CONFIG)
+        .supergraph(PathBuf::from_iter([
+            "tests",
+            "fixtures",
+            "connectors",
+            "quickstart.graphql",
+        ]))
+        .build()
+        .await;
+
+    router.start().await;
+
+    // Make sure that we have the warnings we expect
+    let plugins = ["coprocessor", "headers", "telemetry", "traffic_shaping"];
+    for plugin in plugins {
+        let msg = format!("plugin `{plugin}` is enabled for connector-enabled subgraphs, which is not yet supported");
+        router.assert_log_contains(&msg).await;
+    }
+
+    Ok(())
+}

apollo-router/tests/integration/mod.rs

@@ -3,6 +3,7 @@ mod batching;
 pub(crate) mod common;
 pub(crate) use common::IntegrationTest;
 
+mod connectors;
 mod coprocessor;
 mod docs;
 mod file_upload;