Premature stream termination is not propagated

[carlbraganza]

What happened:
Premature termination of the gRPC stream (either due to client cancellation or sidecar detected error) must be propagated to the CSI driver by cancelling the context used to create the stream.

For example, we have logic that looks like this:

func (s *Server) GetMetadataAllocated(req *api.GetMetadataAllocatedRequest, stream api.SnapshotMetadata_GetMetadataAllocatedServer) error {
        ...
        ctx := stream.Context()
	csiStream, err := csi.NewSnapshotMetadataClient(s.csiConnection()).GetMetadataAllocated(ctx, csiReq)
	for {
		csiResp, err := csiStream.Recv()

		if err := clientStream.Send(clientResp); err != nil {
			return s.statusPassOrWrapError(err, codes.Internal, msgInternalFailedToSendResponseFmt, err)
		}
	}
}

A failure to send to the client unwinds in the sidecar, but does not cancel the downstream CSI driver - it would block on its Send call until it times out (if it had set a timeout).

What you expected to happen:

We should be doing the equivalent of this:

func (s *Server) GetMetadataAllocated(req *api.GetMetadataAllocatedRequest, stream api.SnapshotMetadata_GetMetadataAllocatedServer) error {
        ...
        ctx, cancelFn := context.WithCancel(stream.Context())
        defer cancelFn()

	csiStream, err := csi.NewSnapshotMetadataClient(s.csiConnection()).GetMetadataAllocated(ctx, csiReq)
	for {
		csiResp, err := csiStream.Recv()

		if err := clientStream.Send(clientResp); err != nil {
			return s.statusPassOrWrapError(err, codes.Internal, msgInternalFailedToSendResponseFmt, err)
		}
	}
}

Here the deferred cancel function would force cancellation of the CSI driver's stream either when a local error is detected or if the sidecar's client cancelled its stream.

How to reproduce it:

Anything else we need to know?:

Environment:

• Driver version:
• Kubernetes version (use kubectl version):
• OS (e.g. from /etc/os-release):
• Kernel (e.g. uname -a):
• Install tools:
• Others:

[carlbraganza]

/assign @carlbraganza