fix unhandled error from ingester (#61)

* fix unhandled error from ingester

Signed-off-by: Nishu Bharti <nishu.bhart1@ibm.com>

* minor update

Signed-off-by: Nishu Bharti <nishu.bhart1@ibm.com>

* add unit test

Signed-off-by: Nishu Bharti <nishu.bhart1@ibm.com>

* minor fix

Signed-off-by: Nishu Bharti <nishu.bhart1@ibm.com>

---------

Signed-off-by: Nishu Bharti <nishu.bhart1@ibm.com>
Co-authored-by: Nishu Bharti <nishu.bhart1@ibm.com>

Author: nishubharti

fluent/client/ws_client.go

@@ -28,7 +28,10 @@ import (
 	"bytes"
 	"crypto/tls"
 	"errors"
+	"fmt"
+	"io"
 	"net/http"
+	"strconv"
 	"sync"
 
 	"github.com/IBM/fluent-forward-go/fluent/client/ws"
@@ -112,7 +115,14 @@ func (wcf *DefaultWSConnectionFactory) New() (ext.Conn, error) {
 
 	conn, resp, err := dialer.Dial(wcf.URL, header)
 	if resp != nil && resp.Body != nil {
-		// TODO: dump response, which is second return value from Dial
+		bodyBytes, readErr := io.ReadAll(resp.Body)
+		if readErr == nil {
+			bodyString := string(bodyBytes)
+			if resp.StatusCode >= 300 {
+				err = fmt.Errorf("%s. %s:%s", err, strconv.Itoa(resp.StatusCode), bodyString)
+			}
+		}
+
 		resp.Body.Close()
 	}
 

fluent/client/ws_client_test.go

@@ -61,12 +61,12 @@ var _ = Describe("IAMAuthInfo", func() {
 
 var _ = Describe("DefaultWSConnectionFactory", func() {
 	var (
-		svr    *httptest.Server
-		happy  chan struct{}
-		useTLS bool
+		svr               *httptest.Server
+		ch                chan struct{}
+		useTLS, testError bool
 	)
 
-	newHandler := func(happy chan struct{}) http.Handler {
+	happyHandler := func(ch chan struct{}) http.Handler {
 		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 			defer GinkgoRecover()
 			svrOpts := ws.ConnectionOptions{}
@@ -82,19 +82,27 @@ var _ = Describe("DefaultWSConnectionFactory", func() {
 				Fail("broke")
 			}
 
-			happy <- struct{}{}
+			ch <- struct{}{}
 
 			svrConnection.Close()
 		})
 	}
 
+	sadHandler := func(ch chan struct{}) http.Handler {
+		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			http.Error(w, "broken test", http.StatusInternalServerError)
+		})
+	}
+
 	JustBeforeEach(func() {
-		happy = make(chan struct{})
+		ch = make(chan struct{})
 
 		if useTLS {
-			svr = httptest.NewTLSServer(newHandler(happy))
+			svr = httptest.NewTLSServer(happyHandler(ch))
+		} else if testError {
+			svr = httptest.NewTLSServer(sadHandler(ch))
 		} else {
-			svr = httptest.NewServer(newHandler(happy))
+			svr = httptest.NewServer(happyHandler(ch))
 		}
 
 		time.Sleep(5 * time.Millisecond)
@@ -118,10 +126,36 @@ var _ = Describe("DefaultWSConnectionFactory", func() {
 		})
 
 		Expect(cli.Connect()).ToNot(HaveOccurred())
-		Eventually(happy).Should(Receive())
+		Eventually(ch).Should(Receive())
 		Expect(cli.Disconnect()).ToNot(HaveOccurred())
 	})
 
+	When("sends wrong url, expects error", func() {
+
+		BeforeEach(func() {
+			testError = true
+		})
+
+		It("returns error", func() {
+			u := "ws" + strings.TrimPrefix(svr.URL, "http")
+
+			cli := fclient.NewWS(client.WSConnectionOptions{
+				Factory: &client.DefaultWSConnectionFactory{
+					URL: u + "/test",
+					TLSConfig: &tls.Config{
+						InsecureSkipVerify: true,
+					},
+					AuthInfo: NewIAMAuthInfo("oi"),
+				},
+			})
+
+			err := cli.Connect()
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("websocket: bad handshake. 500:broken test"))
+		})
+
+	})
+
 	When("the factory is configured for TLS", func() {
 		BeforeEach(func() {
 			useTLS = true
@@ -204,22 +238,6 @@ var _ = Describe("WSClient", func() {
 			})
 		})
 
-		When("the factory returns an error", func() {
-			var (
-				connectionError error
-			)
-
-			JustBeforeEach(func() {
-				connectionError = errors.New("Nope")
-				factory.NewReturns(nil, connectionError)
-			})
-
-			It("Returns an error", func() {
-				err := client.Connect()
-				Expect(err).To(HaveOccurred())
-				Expect(err).To(BeIdenticalTo(connectionError))
-			})
-		})
 	})
 
 	Describe("Disconnect", func() {