Downgrade various logs to DEBUG level (backport #2557) (#2558)

The Trillium handler for errors was unconditionally logging them at
`WARN`, generating large volumes of logging for problems that aren't
Janus' fault, like reports whose timestamps are in the future. Now, we
check whether the respones indicates a server error before bothing to
log it (keeping in mind that client errors will still appear in
metrics).

Additionally, we've made liberal use of the [`tracing::instrument`][1]
attribute macro to decorate various functions and methods with tracing
spans, and in particular its `err` argument for logging anytime a
function that returns `Result<T, E>` returns an `Err`. Mostly, logging
those errors is handled at a higher level -- say, in a Trillium handler
or in the top-level loop of `aggregation_job_driver` -- so logging them
at level `ERROR` at the function call itself isn't that helpful. This
commit further qualifies the `err` argument so that those errors are now
logged at `DEBUG`, so we can opt back into them should they prove
useful.

[1]: https://docs.rs/tracing/latest/tracing/attr.instrument.html

aggregator/src/aggregator.rs

@@ -90,7 +90,7 @@ use std::{
     time::{Duration as StdDuration, Instant},
 };
 use tokio::{sync::Mutex, try_join};
-use tracing::{debug, info, trace_span, warn};
+use tracing::{debug, info, trace_span, warn, Level};
 use url::Url;
 
 pub mod accumulator;
@@ -654,7 +654,7 @@ impl<C: Clock> Aggregator<C> {
     }
 
     /// Opts in or out of a taskprov task.
-    #[tracing::instrument(skip(self, aggregator_auth_token), err)]
+    #[tracing::instrument(skip(self, aggregator_auth_token), err(level = Level::DEBUG))]
     async fn taskprov_opt_in(
         &self,
         peer_role: &Role,
@@ -727,7 +727,7 @@ impl<C: Clock> Aggregator<C> {
     /// Validate and authorize a taskprov request. Returns values necessary for determining whether
     /// we can opt into the task. This function might return an opt-out error for conditions that
     /// are relevant for all DAP workflows (e.g. task expiration).
-    #[tracing::instrument(skip(self, aggregator_auth_token), err)]
+    #[tracing::instrument(skip(self, aggregator_auth_token), err(level = Level::DEBUG))]
     async fn taskprov_authorize_request(
         &self,
         peer_role: &Role,
@@ -1227,7 +1227,7 @@ macro_rules! vdaf_ops_dispatch {
 }
 
 impl VdafOps {
-    #[tracing::instrument(skip_all, fields(task_id = ?task.id()), err)]
+    #[tracing::instrument(skip_all, fields(task_id = ?task.id()), err(level = Level::DEBUG))]
     async fn handle_upload<C: Clock>(
         &self,
         clock: &C,
@@ -1278,7 +1278,7 @@ impl VdafOps {
     #[tracing::instrument(
         skip(self, datastore, global_hpke_keypairs, aggregate_step_failure_counter, task, req_bytes),
         fields(task_id = ?task.id()),
-        err
+        err(level = Level::DEBUG)
     )]
     async fn handle_aggregate_init<C: Clock>(
         &self,
@@ -1329,7 +1329,7 @@ impl VdafOps {
     #[tracing::instrument(
         skip(self, datastore, aggregate_step_failure_counter, task, req, request_hash),
         fields(task_id = ?task.id()),
-        err
+        err(level = Level::DEBUG)
     )]
     async fn handle_aggregate_continue<C: Clock>(
         &self,
@@ -2290,7 +2290,7 @@ impl VdafOps {
     #[tracing::instrument(
         skip(self, datastore, task, collection_req_bytes),
         fields(task_id = ?task.id()),
-        err
+        err(level = Level::DEBUG)
     )]
     async fn handle_create_collection_job<C: Clock>(
         &self,
@@ -2592,7 +2592,7 @@ impl VdafOps {
     /// Handle GET requests to the leader's `tasks/{task-id}/collection_jobs/{collection-job-id}`
     /// endpoint. The return value is an encoded `CollectResp<Q>`.
     /// <https://www.ietf.org/archive/id/draft-ietf-ppm-dap-07.html#name-collecting-results>
-    #[tracing::instrument(skip(self, datastore, task), fields(task_id = ?task.id()), err)]
+    #[tracing::instrument(skip(self, datastore, task), fields(task_id = ?task.id()), err(level = Level::DEBUG))]
     async fn handle_get_collection_job<C: Clock>(
         &self,
         datastore: &Datastore<C>,
@@ -2758,7 +2758,7 @@ impl VdafOps {
         }
     }
 
-    #[tracing::instrument(skip(self, datastore, task), fields(task_id = ?task.id()), err)]
+    #[tracing::instrument(skip(self, datastore, task), fields(task_id = ?task.id()), err(level = Level::DEBUG))]
     async fn handle_delete_collection_job<C: Clock>(
         &self,
         datastore: &Datastore<C>,
@@ -2842,7 +2842,7 @@ impl VdafOps {
     #[tracing::instrument(
         skip(self, datastore, clock, task, req_bytes),
         fields(task_id = ?task.id()),
-        err
+        err(level = Level::DEBUG)
     )]
     async fn handle_aggregate_share<C: Clock>(
         &self,
@@ -3140,7 +3140,7 @@ fn empty_batch_aggregations<
         http_request_duration_histogram,
     ),
     fields(url = %url),
-    err,
+    err(level = Level::DEBUG),
 )]
 async fn send_request_to_helper<T: Encode>(
     http_client: &Client,

aggregator/src/aggregator/http_handlers.rs

@@ -41,8 +41,7 @@ impl Handler for Error {
     async fn run(&self, mut conn: Conn) -> Conn {
         let error_code = self.error_code();
         conn.set_state(ErrorCode(error_code));
-        warn!(error_code, error=?self, "Error handling endpoint");
-        match self {
+        let conn = match self {
             Error::InvalidConfiguration(_) => conn.with_status(Status::InternalServerError),
             Error::MessageDecode(_) => conn
                 .with_problem_document(&ProblemDocument::new_dap(DapProblemType::InvalidMessage)),
@@ -132,7 +131,13 @@ impl Handler for Error {
                 &ProblemDocument::new_dap(DapProblemType::InvalidTask).with_task_id(task_id),
             ),
             Error::DifferentialPrivacy(_) => conn.with_status(Status::InternalServerError),
+        };
+
+        if matches!(conn.status(), Some(status) if status.is_server_error()) {
+            warn!(error_code, error=?self, "Error handling endpoint");
         }
+
+        conn
     }
 }