Propagate errors in rlp log client streaming

Propogates errors received from the rlp via gRPC when sending requests.
Now that we don't block on dialing the server, we need to handle errors
at the sending requests level.

Signed-off-by: Andrew Crump <[email protected]>

Author: ctlong

src/rlp-gateway/internal/ingress/log_client.go

@@ -33,13 +33,14 @@ func NewLogClient(creds credentials.TransportCredentials, logsProviderAddr strin
 }
 
 // Stream opens a new stream on the log client.
-func (c *LogClient) Stream(ctx context.Context, req *loggregator_v2.EgressBatchRequest) web.Receiver {
+func (c *LogClient) Stream(ctx context.Context, req *loggregator_v2.EgressBatchRequest) (web.Receiver, error) {
 	receiver, err := c.c.BatchedReceiver(ctx, req)
 	if err != nil {
 		log.Printf("failed to open stream from logs provider: %s", err)
+		return nil, err
 	}
 
-	return receiver.Recv
+	return receiver.Recv, nil
 }
 
 func (c *LogClient) Close() error {

src/rlp-gateway/internal/web/handler.go

@@ -14,7 +14,7 @@ type Receiver func() (*loggregator_v2.EnvelopeBatch, error)
 // LogsProvder defines the interface for opening streams to the
 // logs provider
 type LogsProvider interface {
-	Stream(ctx context.Context, req *loggregator_v2.EgressBatchRequest) Receiver
+	Stream(ctx context.Context, req *loggregator_v2.EgressBatchRequest) (Receiver, error)
 }
 
 // Handler defines a struct for servering http endpoints

src/rlp-gateway/internal/web/json_error.go

@@ -13,6 +13,7 @@ var (
 	errCounterNamePresentButEmpty = newJSONError(http.StatusBadRequest, "missing_counter_name", "counter.name is invalid without value")
 	errGaugeNamePresentButEmpty   = newJSONError(http.StatusBadRequest, "missing_gauge_name", "gauge.name is invalid without value")
 	errStreamingUnsupported       = newJSONError(http.StatusInternalServerError, "streaming_unsupported", "request does not support streaming")
+	errStreamingUnavailable       = newJSONError(http.StatusServiceUnavailable, "streaming_unavailable", "streaming is temporarily unavailable")
 	errNotFound                   = newJSONError(http.StatusNotFound, "not_found", "resource not found")
 )
 

src/rlp-gateway/internal/web/read.go

@@ -47,13 +47,7 @@ func ReadHandler(
 			return
 		}
 
-		w.Header().Set("Content-Type", "text/event-stream")
-		w.Header().Set("Cache-Control", "no-cache")
-		w.Header().Set("Connection", "keep-alive")
-
-		flusher.Flush()
-
-		recv := lp.Stream(
+		recv, err := lp.Stream(
 			ctx,
 			&loggregator_v2.EgressBatchRequest{
 				ShardId:           query.Get("shard_id"),
@@ -62,6 +56,16 @@ func ReadHandler(
 				Selectors:         s,
 			},
 		)
+		if err != nil {
+			errStreamingUnavailable.Write(w)
+			return
+		}
+
+		w.Header().Set("Content-Type", "text/event-stream")
+		w.Header().Set("Cache-Control", "no-cache")
+		w.Header().Set("Connection", "keep-alive")
+
+		flusher.Flush()
 
 		data := make(chan *loggregator_v2.EnvelopeBatch)
 		errs := make(chan error, 1)

src/rlp-gateway/internal/web/read_test.go

@@ -27,7 +27,7 @@ var _ = Describe("Read", func() {
 
 	BeforeEach(func() {
 		lp = newStubLogsProvider()
-		lp._batchResponse = &loggregator_v2.EnvelopeBatch{
+		lp.resp = &loggregator_v2.EnvelopeBatch{
 			Batch: []*loggregator_v2.Envelope{
 				{
 					SourceId: "source-id-a",
@@ -150,7 +150,7 @@ var _ = Describe("Read", func() {
 	})
 
 	It("contains zero values for gauge metrics", func() {
-		lp._batchResponse = &loggregator_v2.EnvelopeBatch{
+		lp.resp = &loggregator_v2.EnvelopeBatch{
 			Batch: []*loggregator_v2.Envelope{
 				{
 					SourceId: "source-id-a",
@@ -310,9 +310,28 @@ var _ = Describe("Read", func() {
 		}).Should(Equal(io.EOF))
 	})
 
+	It("returns service unavailable when unable to stream from the logs provider", func() {
+		lp.streamErr = errors.New("streaming unavailable")
+		req, err := http.NewRequest(http.MethodGet, server.URL+"/v2/read?log", nil)
+		Expect(err).ToNot(HaveOccurred())
+
+		req = req.WithContext(ctx)
+
+		resp, err := server.Client().Do(req)
+		Expect(err).ToNot(HaveOccurred())
+		body, err := io.ReadAll(resp.Body)
+		Expect(err).ToNot(HaveOccurred())
+
+		Expect(resp.StatusCode).To(Equal(http.StatusServiceUnavailable))
+		Expect(body).To(MatchJSON(`{
+			"error": "streaming_unavailable",
+			"message": "streaming is temporarily unavailable"
+		}`))
+	})
+
 	It("closes the SSE stream if the envelope stream returns any error", func() {
-		lp._batchResponse = nil
-		lp._errorResponse = errors.New("an error")
+		lp.resp = nil
+		lp.respErr = errors.New("an error")
 
 		req, err := http.NewRequest(http.MethodGet, server.URL+"/v2/read?log", nil)
 		Expect(err).ToNot(HaveOccurred())
@@ -379,38 +398,43 @@ var _ = Describe("Read", func() {
 })
 
 type stubLogsProvider struct {
-	mu             sync.Mutex
-	_requests      []*loggregator_v2.EgressBatchRequest
-	_batchResponse *loggregator_v2.EnvelopeBatch
-	_errorResponse error
-	block          bool
+	mu        sync.Mutex
+	reqs      []*loggregator_v2.EgressBatchRequest
+	resp      *loggregator_v2.EnvelopeBatch
+	respErr   error
+	block     bool
+	streamErr error
 }
 
 func newStubLogsProvider() *stubLogsProvider {
 	return &stubLogsProvider{}
 }
 
-func (s *stubLogsProvider) Stream(ctx context.Context, req *loggregator_v2.EgressBatchRequest) web.Receiver {
+func (s *stubLogsProvider) Stream(ctx context.Context, req *loggregator_v2.EgressBatchRequest) (web.Receiver, error) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
-	s._requests = append(s._requests, req)
+	s.reqs = append(s.reqs, req)
+
+	if s.streamErr != nil {
+		return nil, s.streamErr
+	}
 
 	return func() (*loggregator_v2.EnvelopeBatch, error) {
 		if s.block {
 			var block chan int
 			<-block
 		}
 
-		return s._batchResponse, s._errorResponse
-	}
+		return s.resp, s.respErr
+	}, nil
 }
 
 func (s *stubLogsProvider) requests() []*loggregator_v2.EgressBatchRequest {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
-	result := make([]*loggregator_v2.EgressBatchRequest, len(s._requests))
-	copy(result, s._requests)
+	result := make([]*loggregator_v2.EgressBatchRequest, len(s.reqs))
+	copy(result, s.reqs)
 
 	return result
 }