feat: add http/Error to error Channel making Error.Header accessible

Author: karel-rehor

api/http/error.go

@@ -27,6 +27,8 @@ func (e *Error) Error() string {
 		return e.Err.Error()
 	case e.Code != "" && e.Message != "":
 		return fmt.Sprintf("%s: %s", e.Code, e.Message)
+	case e.Message != "":
+		return e.Message
 	default:
 		return "Unexpected status code " + strconv.Itoa(e.StatusCode)
 	}
@@ -39,6 +41,23 @@ func (e *Error) Unwrap() error {
 	return nil
 }
 
+// HeaderToString generates a string value from the Header property.  Useful in logging.
+func (e *Error) HeaderToString(selected []string) string {
+	headerString := ""
+	if len(selected) == 0 {
+		for key := range e.Header {
+			headerString += fmt.Sprintf("%s: %s\r\n", key, e.Header[key])
+		}
+	} else {
+		for _, candidate := range selected {
+			if e.Header.Get(candidate) != "" {
+				headerString += fmt.Sprintf("%s: %s\n", candidate, e.Header.Get(candidate))
+			}
+		}
+	}
+	return headerString
+}
+
 // NewError returns newly created Error initialised with nested error and default values
 func NewError(err error) *Error {
 	return &Error{

api/http/service.go

@@ -16,7 +16,6 @@ package http
 import (
 	"context"
 	"encoding/json"
-	"fmt"
 	"io"
 	"mime"
 	"net/http"
@@ -114,25 +113,7 @@ func (s *service) DoHTTPRequest(req *http.Request, requestCallback RequestCallba
 	}
 
 	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
-		logHeaders := ""
-		httpErr := s.parseHTTPError(resp)
-		for _, candidate := range []string{
-			"date",
-			"trace-id",
-			"trace-sampled",
-			"X-Influxdb-Build",
-			"X-Influxdb-Request-ID",
-			"X-Influxdb-Version",
-		} {
-			if httpErr.Header.Get(candidate) != "" {
-				logHeaders += fmt.Sprintf("%s: %s\n", candidate, httpErr.Header.Get(candidate))
-			}
-		}
-		log.Errorf("http status code: %d, %s\nSelect Response Headers:\n%s",
-			resp.StatusCode,
-			httpErr.Error(),
-			logHeaders)
-		return httpErr
+		return s.parseHTTPError(resp)
 	}
 	if responseCallback != nil {
 		err := responseCallback(resp)

api/write_test.go

@@ -8,7 +8,10 @@ import (
 	"fmt"
 	"io"
 	"math"
+	ihttp "net/http"
+	"net/http/httptest"
 	"runtime"
+	"strconv"
 	"strings"
 	"sync"
 	"testing"
@@ -265,3 +268,47 @@ func TestFlushWithRetries(t *testing.T) {
 	// two remained
 	assert.Equal(t, 2, len(service.Lines()))
 }
+
+func TestWriteApiErrorHeaders(t *testing.T) {
+	calls := 0
+	var mu sync.Mutex
+	server := httptest.NewServer(ihttp.HandlerFunc(func(w ihttp.ResponseWriter, r *ihttp.Request) {
+		mu.Lock()
+		defer mu.Unlock()
+		calls++
+		w.Header().Set("X-Test-Val1", "Not All Correct")
+		w.Header().Set("X-Test-Val2", "Atlas LV-3B")
+		w.Header().Set("X-Call-Count", strconv.Itoa(calls))
+		w.WriteHeader(ihttp.StatusBadRequest)
+		_, _ = w.Write([]byte(`{ "code": "bad request", "message": "test header" }`))
+	}))
+	defer server.Close()
+	svc := http.NewService(server.URL, "my-token", http.DefaultOptions())
+	writeAPI := NewWriteAPI("my-org", "my-bucket", svc, write.DefaultOptions().SetBatchSize(5))
+	defer writeAPI.Close()
+	errCh := writeAPI.Errors()
+	var wg sync.WaitGroup
+	var recErr error
+	wg.Add(1)
+	go func() {
+		for i := 0; i < 3; i++ {
+			recErr = <-errCh
+			assert.NotNil(t, recErr, "errCh should not run out of values")
+			assert.Len(t, recErr.(*http.Error).Header, 6)
+			assert.NotEqual(t, "", recErr.(*http.Error).Header.Get("Date"))
+			assert.NotEqual(t, "", recErr.(*http.Error).Header.Get("Content-Length"))
+			assert.NotEqual(t, "", recErr.(*http.Error).Header.Get("Content-Type"))
+			assert.Equal(t, strconv.Itoa(i+1), recErr.(*http.Error).Header.Get("X-Call-Count"))
+			assert.Equal(t, "Not All Correct", recErr.(*http.Error).Header.Get("X-Test-Val1"))
+			assert.Equal(t, "Atlas LV-3B", recErr.(*http.Error).Header.Get("X-Test-Val2"))
+		}
+		wg.Done()
+	}()
+	points := test.GenPoints(15)
+	for i := 0; i < 15; i++ {
+		writeAPI.WritePoint(points[i])
+	}
+	writeAPI.waitForFlushing()
+	wg.Wait()
+	assert.Equal(t, calls, 3)
+}

internal/write/service.go

@@ -164,6 +164,7 @@ func (w *Service) HandleWrite(ctx context.Context, batch *Batch) error {
 		if batchToWrite != nil {
 			perror := w.WriteBatch(ctx, batchToWrite)
 			if perror != nil {
+				// fmt.Printf("DEBUG perror type %s\n", reflect.TypeOf(perror))
 				if isIgnorableError(perror) {
 					log.Warnf("Write error: %s", perror.Error())
 				} else {
@@ -196,9 +197,30 @@ func (w *Service) HandleWrite(ctx context.Context, batch *Batch) error {
 						w.retryAttempts++
 						log.Debugf("Write proc: next wait for write is %dms\n", w.retryDelay)
 					} else {
-						log.Errorf("Write error: %s\n", perror.Error())
+						logMessage := fmt.Sprintf("Write error: %s", perror.Error())
+						logHeaders := perror.HeaderToString([]string{
+							"date",
+							"trace-id",
+							"trace-sampled",
+							"X-Influxdb-Build",
+							"X-Influxdb-Request-ID",
+							"X-Influxdb-Version",
+						})
+						if len(logHeaders) > 0 {
+							logMessage += fmt.Sprintf("\nSelect Response Headers:\n%s", logHeaders)
+						}
+						log.Error(logMessage)
+					}
+					return &http2.Error{
+						StatusCode: int(perror.StatusCode),
+						Code:       perror.Code,
+						Message: fmt.Errorf(
+							"write failed (attempts %d): %w", batchToWrite.RetryAttempts, perror,
+						).Error(),
+						Err:        perror.Err,
+						RetryAfter: perror.RetryAfter,
+						Header:     perror.Header,
 					}
-					return fmt.Errorf("write failed (attempts %d): %w", batchToWrite.RetryAttempts, perror)
 				}
 			}
 

internal/write/service_test.go

@@ -337,10 +337,13 @@ func TestMaxRetryTime(t *testing.T) {
 	b := NewBatch("2\n", exp)
 	// First batch will  be checked against maxRetryTime and it will expire. New batch will fail and it will added to retry queue
 	err = srv.HandleWrite(ctx, b)
+	//fmt.Printf("DEBUG err %v\n", err)
+	//fmt.Printf("DEBUG err %v\n", err)
 	require.NotNil(t, err)
 	// 1st Batch expires and writing 2nd trows error
 	assert.Equal(t, "write failed (attempts 1): Unexpected status code 429", err.Error())
 	assert.Equal(t, 1, srv.retryQueue.list.Len())
+	// fmt.Printf("DEBUG Header len: %d\n", len(err.(*http.Error).Header))
 
 	//wait until remaining accumulated retryDelay has passed, because there hasn't been a successful write yet
 	<-time.After(time.Until(srv.lastWriteAttempt.Add(time.Millisecond * time.Duration(srv.retryDelay))))
@@ -702,3 +705,20 @@ func TestIgnoreErrors(t *testing.T) {
 	err = srv.HandleWrite(ctx, b)
 	assert.Error(t, err)
 }
+
+func TestHttpErrorHeaders(t *testing.T) {
+	server := httptest.NewServer(ihttp.HandlerFunc(func(w ihttp.ResponseWriter, r *ihttp.Request) {
+		w.Header().Set("X-Test-Val1", "Not All Correct")
+		w.Header().Set("X-Test-Val2", "Atlas LV-3B")
+		w.WriteHeader(ihttp.StatusBadRequest)
+		_, _ = w.Write([]byte(`{ "code": "bad request", "message": "test header" }`))
+	}))
+	defer server.Close()
+	svc := NewService("my-org", "my-bucket", http.NewService(server.URL, "", http.DefaultOptions()),
+		write.DefaultOptions())
+	err := svc.HandleWrite(context.Background(), NewBatch("1", 20))
+	assert.Error(t, err)
+	assert.Equal(t, "400 Bad Request: write failed (attempts 0): 400 Bad Request: { \"code\": \"bad request\", \"message\": \"test header\" }", err.Error())
+	assert.Equal(t, "Not All Correct", err.(*http.Error).Header.Get("X-Test-Val1"))
+	assert.Equal(t, "Atlas LV-3B", err.(*http.Error).Header.Get("X-Test-Val2"))
+}