opencensus: drive rsp future to establish conn (#588)

In the `futures` 0.1 version of the OpenCensus module, the `tower-grpc`
response future for the client-streaming RPC was never polled, because
we did not need the server response for that RPC. At the time, this was
fine, as `tower-grpc` still established the connection and ensured
messages were sent to the server. However, after updating to `tonic`,
this was no longer the case --- the response future is now responsible
for establishing the connection. This means spans are no longer sent to
the OpenCensus collector.

Due to an issue where the cloud integration tests were silently failing
to run, we missed this regression.

This PR fixes the regression by ensuring that we drive the response
future.

I'd like to rewrite this code to use async-await rather than the manual
state machine, but this change fixes the regression, so I thought I'd go
ahead and open a PR, and refactor in a separate branch.

Signed-off-by: Eliza Weisman <[email protected]>

Author: hawkw

linkerd/opencensus/src/lib.rs

@@ -1,5 +1,5 @@
 #![deny(warnings, rust_2018_idioms)]
-use futures::{ready, Stream};
+use futures::{ready, FutureExt, Stream};
 use http_body::Body as HttpBody;
 use linkerd2_error::Error;
 use linkerd2_stack::NewService;
@@ -45,16 +45,17 @@ enum State {
         // Node data should only be sent on the first message of a streaming
         // request.
         node: Option<Node>,
-        // We hold the response future, but never poll it.
-        _rsp: Pin<
-            Box<
-                dyn Future<
-                        Output = Result<
-                            grpc::Response<Streaming<ExportTraceServiceResponse>>,
-                            grpc::Status,
-                        >,
-                    > + Send
-                    + 'static,
+        rsp: Option<
+            Pin<
+                Box<
+                    dyn Future<
+                            Output = Result<
+                                grpc::Response<Streaming<ExportTraceServiceResponse>>,
+                                grpc::Status,
+                            >,
+                        > + Send
+                        + 'static,
+                >,
             >,
         >,
         metrics: Registry,
@@ -177,30 +178,47 @@ where
                     let req = grpc::Request::new(request_rx);
                     trace!("Establishing new TraceService::export request");
                     this.metrics.start_stream();
-                    let _rsp = Box::pin(async move { svc.export(req).await });
+                    let rsp = Box::pin(async move { svc.export(req).await });
                     State::Sending {
                         sender: request_tx,
                         node: Some(this.node.clone()),
-                        _rsp,
+                        rsp: Some(rsp),
                         metrics: this.metrics.clone(),
                     }
                 }
                 State::Sending {
                     ref mut sender,
                     ref mut node,
                     ref mut metrics,
+                    ref mut rsp,
                     ..
                 } => {
-                    match ready!(Self::poll_send_spans(
-                        this.spans,
-                        sender,
-                        node,
-                        *this.max_batch_size,
-                        metrics,
-                        cx,
-                    )) {
-                        Ok(()) => return Poll::Ready(()),
-                        Err(()) => State::Idle,
+                    let mut idle = false;
+                    if let Some(mut f) = rsp.take() {
+                        match f.poll_unpin(cx) {
+                            Poll::Ready(Ok(_)) => {}
+                            Poll::Pending => *rsp = Some(f),
+                            Poll::Ready(Err(error)) => {
+                                tracing::debug!(%error, "response future failed, sending a new request");
+                                idle = true;
+                            }
+                        }
+                    }
+                    // this is gross...let's just see if it works.
+                    if idle {
+                        State::Idle
+                    } else {
+                        match ready!(Self::poll_send_spans(
+                            this.spans,
+                            sender,
+                            node,
+                            *this.max_batch_size,
+                            metrics,
+                            cx,
+                        )) {
+                            Ok(()) => return Poll::Ready(()),
+                            Err(()) => State::Idle,
+                        }
                     }
                 }
             };