Fix the error count for subscription requests for apollo telemetry

.changesets/fix_bnjjj_fix_subscription_metrics.md

@@ -0,0 +1,7 @@
+### Fix the error count for subscription requests for apollo telemetry ([PR #3500](https://github.com/apollographql/router/pull/3500))
+
+Count subscription requests only if the feature is enabled.
+
+The router would previously count subscription requests regardless of whether the feature is enabled or not. This changeset will only count subscription requests if the feature has been enabled.
+
+By [@bnjjj](https://github.com/bnjjj) in https://github.com/apollographql/router/pull/3500

apollo-router/src/plugin/test/mock/canned.rs

@@ -114,6 +114,16 @@ pub(crate) fn reviews_subgraph() -> MockSubgraph {
                         ]
                     }
                 }}
+        ),
+        (
+            json! {{
+                    "query": "subscription{reviewAdded{body}}",
+                }},
+            json! {{
+                "errors": [{
+                    "message": "subscription is not enabled"
+                }]
+            }}
         )
     ].into_iter().map(|(query, response)| (serde_json::from_value(query).unwrap(), serde_json::from_value(response).unwrap())).collect();
     MockSubgraph::new(review_mocks)

apollo-router/src/plugins/telemetry/metrics/apollo.rs

@@ -68,6 +68,7 @@ mod test {
     use super::*;
     use crate::plugin::Plugin;
     use crate::plugin::PluginInit;
+    use crate::plugins::subscription;
     use crate::plugins::telemetry::apollo;
     use crate::plugins::telemetry::apollo::default_buffer_size;
     use crate::plugins::telemetry::apollo::ENDPOINT_DEFAULT;
@@ -107,7 +108,7 @@ mod test {
     #[tokio::test(flavor = "multi_thread")]
     async fn apollo_metrics_single_operation() -> Result<(), BoxError> {
         let query = "query {topProducts{name}}";
-        let results = get_metrics_for_request(query, None, None).await?;
+        let results = get_metrics_for_request(query, None, None, false).await?;
         let mut settings = insta::Settings::clone_current();
         settings.set_sort_maps(true);
         settings.add_redaction("[].request_id", "[REDACTED]");
@@ -124,7 +125,24 @@ mod test {
         let _ = context
             .insert(OPERATION_KIND, OperationKind::Subscription)
             .unwrap();
-        let results = get_metrics_for_request(query, None, Some(context)).await?;
+        let results = get_metrics_for_request(query, None, Some(context), true).await?;
+        let mut settings = insta::Settings::clone_current();
+        settings.set_sort_maps(true);
+        settings.add_redaction("[].request_id", "[REDACTED]");
+        settings.bind(|| {
+            insta::assert_json_snapshot!(results);
+        });
+        Ok(())
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn apollo_metrics_for_subscription_error() -> Result<(), BoxError> {
+        let query = "subscription{reviewAdded{body}}";
+        let context = Context::new();
+        let _ = context
+            .insert(OPERATION_KIND, OperationKind::Subscription)
+            .unwrap();
+        let results = get_metrics_for_request(query, None, Some(context), true).await?;
         let mut settings = insta::Settings::clone_current();
         settings.set_sort_maps(true);
         settings.add_redaction("[].request_id", "[REDACTED]");
@@ -137,7 +155,7 @@ mod test {
     #[tokio::test(flavor = "multi_thread")]
     async fn apollo_metrics_multiple_operations() -> Result<(), BoxError> {
         let query = "query {topProducts{name}} query {topProducts{name}}";
-        let results = get_metrics_for_request(query, None, None).await?;
+        let results = get_metrics_for_request(query, None, None, false).await?;
         let mut settings = insta::Settings::clone_current();
         settings.set_sort_maps(true);
         settings.add_redaction("[].request_id", "[REDACTED]");
@@ -150,7 +168,7 @@ mod test {
     #[tokio::test(flavor = "multi_thread")]
     async fn apollo_metrics_parse_failure() -> Result<(), BoxError> {
         let query = "garbage";
-        let results = get_metrics_for_request(query, None, None).await?;
+        let results = get_metrics_for_request(query, None, None, false).await?;
         let mut settings = insta::Settings::clone_current();
         settings.set_sort_maps(true);
         settings.add_redaction("[].request_id", "[REDACTED]");
@@ -163,7 +181,7 @@ mod test {
     #[tokio::test(flavor = "multi_thread")]
     async fn apollo_metrics_unknown_operation() -> Result<(), BoxError> {
         let query = "query {topProducts{name}}";
-        let results = get_metrics_for_request(query, Some("UNKNOWN"), None).await?;
+        let results = get_metrics_for_request(query, Some("UNKNOWN"), None, false).await?;
         let mut settings = insta::Settings::clone_current();
         settings.set_sort_maps(true);
         settings.add_redaction("[].request_id", "[REDACTED]");
@@ -174,7 +192,7 @@ mod test {
     #[tokio::test(flavor = "multi_thread")]
     async fn apollo_metrics_validation_failure() -> Result<(), BoxError> {
         let query = "query {topProducts{unknown}}";
-        let results = get_metrics_for_request(query, None, None).await?;
+        let results = get_metrics_for_request(query, None, None, false).await?;
         let mut settings = insta::Settings::clone_current();
         settings.set_sort_maps(true);
         settings.add_redaction("[].request_id", "[REDACTED]");
@@ -190,7 +208,7 @@ mod test {
         let query = "query {topProducts{name}}";
         let context = Context::new();
         context.insert(STUDIO_EXCLUDE, true)?;
-        let results = get_metrics_for_request(query, None, Some(context)).await?;
+        let results = get_metrics_for_request(query, None, Some(context), false).await?;
         let mut settings = insta::Settings::clone_current();
         settings.set_sort_maps(true);
         settings.add_redaction("[].request_id", "[REDACTED]");
@@ -205,27 +223,31 @@ mod test {
         query: &str,
         operation_name: Option<&str>,
         context: Option<Context>,
+        is_subscription: bool,
     ) -> Result<Vec<SingleStatsReport>, BoxError> {
         let _ = tracing_subscriber::fmt::try_init();
         let mut plugin = create_plugin().await?;
         // Replace the apollo metrics sender so we can test metrics collection.
         let (tx, rx) = futures::channel::mpsc::channel(100);
         plugin.apollo_metrics_sender = Sender::Apollo(tx);
+        let mut request_builder = SupergraphRequest::fake_builder()
+            .header("name_header", "test_client")
+            .header("version_header", "1.0-test")
+            .query(query)
+            .and_operation_name(operation_name)
+            .and_context(context);
+        if is_subscription {
+            request_builder = request_builder.header(
+                "accept",
+                "multipart/mixed; boundary=graphql; subscriptionSpec=1.0",
+            );
+        }
         TestHarness::builder()
             .extra_plugin(plugin)
+            .extra_plugin(create_subscription_plugin().await?)
             .build_router()
             .await?
-            .oneshot(
-                SupergraphRequest::fake_builder()
-                    .header("name_header", "test_client")
-                    .header("version_header", "1.0-test")
-                    .query(query)
-                    .and_operation_name(operation_name)
-                    .and_context(context)
-                    .build()?
-                    .try_into()
-                    .unwrap(),
-            )
+            .oneshot(request_builder.build()?.try_into().unwrap())
             .await
             .unwrap()
             .next_response()
@@ -278,4 +300,12 @@ mod test {
         ))
         .await
     }
+
+    async fn create_subscription_plugin() -> Result<subscription::Subscription, BoxError> {
+        subscription::Subscription::new(PluginInit::fake_new(
+            subscription::SubscriptionConfig::default(),
+            Default::default(),
+        ))
+        .await
+    }
 }

apollo-router/src/plugins/telemetry/metrics/snapshots/apollo_router__plugins__telemetry__metrics__apollo__test__apollo_metrics_for_subscription_error.snap

@@ -0,0 +1,61 @@
+---
+source: apollo-router/src/plugins/telemetry/metrics/apollo.rs
+expression: results
+---
+[
+  {
+    "request_id": "[REDACTED]",
+    "stats": {
+      "# -\nsubscription{reviewAdded{body}}": {
+        "stats_with_context": {
+          "context": {
+            "client_name": "test_client",
+            "client_version": "1.0-test",
+            "operation_type": "subscription",
+            "operation_subtype": "subscription-request"
+          },
+          "query_latency_stats": {
+            "latency": {
+              "secs": 0,
+              "nanos": 100000000
+            },
+            "cache_hit": false,
+            "persisted_query_hit": null,
+            "cache_latency": null,
+            "root_error_stats": {
+              "children": {},
+              "errors_count": 0,
+              "requests_with_errors_count": 0
+            },
+            "has_errors": true,
+            "public_cache_ttl_latency": null,
+            "private_cache_ttl_latency": null,
+            "registered_operation": false,
+            "forbidden_operation": false,
+            "without_field_instrumentation": false
+          },
+          "per_type_stat": {}
+        },
+        "referenced_fields_by_type": {
+          "Review": {
+            "field_names": [
+              "body"
+            ],
+            "is_interface": false
+          },
+          "Subscription": {
+            "field_names": [
+              "reviewAdded"
+            ],
+            "is_interface": false
+          }
+        }
+      }
+    },
+    "licensed_operation_count_by_type": {
+      "type": "subscription",
+      "subtype": "subscription-request",
+      "licensed_operation_count": 1
+    }
+  }
+]

apollo-router/src/plugins/telemetry/mod.rs

@@ -1145,13 +1145,15 @@ impl Telemetry {
                 Err(e)
             }
             Ok(router_response) => {
-                let mut has_errors = !router_response.response.status().is_success();
-                if operation_kind == OperationKind::Subscription {
+                let http_status_is_success = router_response.response.status().is_success();
+
+                // Only send the subscription-request metric if it's an http status in error because we won't always enter the stream after.
+                if operation_kind == OperationKind::Subscription && !http_status_is_success {
                     Self::update_apollo_metrics(
                         ctx,
                         field_level_instrumentation_ratio,
                         sender.clone(),
-                        has_errors,
+                        true,
                         start.elapsed(),
                         operation_kind,
                         Some(OperationSubType::SubscriptionRequest),
@@ -1164,14 +1166,25 @@ impl Telemetry {
                     response_stream
                         .enumerate()
                         .map(move |(idx, response)| {
-                            if !response.errors.is_empty() {
-                                has_errors = true;
-                            }
+                            let has_errors = !response.errors.is_empty();
 
                             if !matches!(sender, Sender::Noop) {
                                 if operation_kind == OperationKind::Subscription {
-                                    // Don't send for the first empty response because it's a heartbeat
-                                    if idx != 0 {
+                                    // The first empty response is always a heartbeat except if it's an error
+                                    if idx == 0 {
+                                        // Don't count for subscription-request if http status was in error because it has been counted before
+                                        if http_status_is_success {
+                                            Self::update_apollo_metrics(
+                                                &ctx,
+                                                field_level_instrumentation_ratio,
+                                                sender.clone(),
+                                                has_errors,
+                                                start.elapsed(),
+                                                operation_kind,
+                                                Some(OperationSubType::SubscriptionRequest),
+                                            );
+                                        }
+                                    } else {
                                         // Only for subscription events
                                         Self::update_apollo_metrics(
                                             &ctx,

apollo-router/src/services/router.rs

@@ -250,7 +250,7 @@ impl Response {
     }
 }
 
-#[derive(Clone, Default)]
+#[derive(Clone, Default, Debug)]
 pub(crate) struct ClientRequestAccepts {
     pub(crate) multipart_defer: bool,
     pub(crate) multipart_subscription: bool,

apollo-router/src/services/supergraph_service.rs

@@ -230,9 +230,8 @@ async fn service_call(
                 .cloned()
                 .unwrap_or_default();
             let mut subscription_tx = None;
-            if (is_deferred || is_subscription)
-                && !accepts_multipart_defer
-                && !accepts_multipart_subscription
+            if (is_deferred && !accepts_multipart_defer)
+                || (is_subscription && !accepts_multipart_subscription)
             {
                 let (error_message, error_code) = if is_deferred {
                     (String::from("the router received a query with the @defer directive but the client does not accept multipart/mixed HTTP responses. To enable @defer support, add the HTTP header 'Accept: multipart/mixed; deferSpec=20220824'"), "DEFER_BAD_HEADER")

apollo-router/src/uplink/license_enforcement.rs

@@ -155,7 +155,8 @@ impl LicenseEnforcementReport {
                 .name("Subgraph entity caching")
                 .build(),
             ConfigurationRestriction::builder()
-                .path("$.subscription")
+                .path("$.subscription.enabled")
+                .value(true)
                 .name("Federated subscriptions")
                 .build(),
             // Per-operation limits are restricted but parser limits like `parser_max_recursion`

apollo-router/src/uplink/snapshots/apollo_router__uplink__license_enforcement__test__restricted_features_via_config.snap

@@ -25,7 +25,7 @@ Configuration yaml:
   .traffic_shaping..experimental_entity_caching
 
 * Federated subscriptions
-  .subscription
+  .subscription.enabled
 
 * Operation depth limiting
   .limits.max_depth

apollo-router/testing_schema.graphql

@@ -32,6 +32,7 @@ type Mutation {
 
 type Subscription {
   userWasCreated: User @join__field(graph: ACCOUNTS)
+  reviewAdded: Review @join__field(graph: REVIEWS)
 }
 
 type Product