Merge pull request #404 from influxdata/feat/httpErrorHeaders

feat: HTTP headers in Error type

Author: karel-rehor

CHANGELOG.md

@@ -1,4 +1,10 @@
-## 2.14.0 [Unreleased]
+## 2.14 [unreleased]
+
+### Features
+
+- [#404](https://github.com/influxdata/influxdb-client-go/pull/404) Expose HTTP response headers in the Error type to aid analysis and debugging of error results.  Add selected response headers to the error log.
+  
+ Also, unified errors returned by WriteAPI, which now always returns `http.Error`    
 
 ### Fixes
 - [#403](https://github.com/influxdata/influxdb-client-go/pull/403) Custom checks de/serialization to allow calling server Check API

api/examples_test.go

@@ -11,6 +11,7 @@ import (
 	"time"
 
 	"github.com/influxdata/influxdb-client-go/v2/api"
+	apiHttp "github.com/influxdata/influxdb-client-go/v2/api/http"
 	"github.com/influxdata/influxdb-client-go/v2/api/write"
 	"github.com/influxdata/influxdb-client-go/v2/domain"
 	influxdb2 "github.com/influxdata/influxdb-client-go/v2/internal/examples"
@@ -123,6 +124,7 @@ func ExampleWriteAPI_errors() {
 	go func() {
 		for err := range errorsCh {
 			fmt.Printf("write error: %s\n", err.Error())
+			fmt.Printf("trace-id: %s\n", err.(*apiHttp.Error).Header.Get("Trace-ID"))
 		}
 	}()
 	// write some points

api/http/error.go

@@ -6,6 +6,7 @@ package http
 
 import (
 	"fmt"
+	"net/http"
 	"strconv"
 )
 
@@ -16,6 +17,7 @@ type Error struct {
 	Message    string
 	Err        error
 	RetryAfter uint
+	Header     http.Header
 }
 
 // Error fulfils error interface
@@ -37,6 +39,25 @@ 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 {
+			k := http.CanonicalHeaderKey(key)
+			headerString += fmt.Sprintf("%s: %s\r\n", k, e.Header.Get(k))
+		}
+	} else {
+		for _, candidate := range selected {
+			c := http.CanonicalHeaderKey(candidate)
+			if e.Header.Get(c) != "" {
+				headerString += fmt.Sprintf("%s: %s\n", c, e.Header.Get(c))
+			}
+		}
+	}
+	return headerString
+}
+
 // NewError returns newly created Error initialised with nested error and default values
 func NewError(err error) *Error {
 	return &Error{
@@ -45,5 +66,6 @@ func NewError(err error) *Error {
 		Message:    "",
 		Err:        err,
 		RetryAfter: 0,
+		Header:     http.Header{},
 	}
 }

api/http/error_test.go

@@ -0,0 +1,105 @@
+// Copyright 2020-2024 InfluxData, Inc. All rights reserved.
+// Use of this source code is governed by MIT
+// license that can be found in the LICENSE file.
+
+package http
+
+import (
+	"fmt"
+	ihttp "net/http"
+
+	"github.com/stretchr/testify/assert"
+
+	"testing"
+)
+
+func TestWriteErrorHeaderToString(t *testing.T) {
+	header := ihttp.Header{
+		"Date":           []string{"2024-08-07T12:00:00.009"},
+		"Content-Length": []string{"12"},
+		"Content-Type":   []string{"application/json", "encoding UTF-8"},
+		"X-Test-Value1":  []string{"SaturnV"},
+		"X-Test-Value2":  []string{"Apollo11"},
+		"Retry-After":    []string{"2044"},
+		"Trace-Id":       []string{"123456789ABCDEF0"},
+	}
+
+	err := Error{
+		StatusCode: ihttp.StatusBadRequest,
+		Code:       "bad request",
+		Message:    "this is just a test",
+		Err:        nil,
+		RetryAfter: 2044,
+		Header:     header,
+	}
+
+	fullString := err.HeaderToString([]string{})
+
+	// write order is not guaranteed
+	assert.Contains(t, fullString, "Date: 2024-08-07T12:00:00.009")
+	assert.Contains(t, fullString, "Content-Length: 12")
+	assert.Contains(t, fullString, "Content-Type: application/json")
+	assert.Contains(t, fullString, "X-Test-Value1: SaturnV")
+	assert.Contains(t, fullString, "X-Test-Value2: Apollo11")
+	assert.Contains(t, fullString, "Retry-After: 2044")
+	assert.Contains(t, fullString, "Trace-Id: 123456789ABCDEF0")
+
+	filterString := err.HeaderToString([]string{"date", "trace-id", "x-test-value1", "x-test-value2"})
+
+	// write order will follow filter arguments
+	assert.Equal(t, filterString,
+		"Date: 2024-08-07T12:00:00.009\nTrace-Id: 123456789ABCDEF0\nX-Test-Value1: SaturnV\nX-Test-Value2: Apollo11\n",
+	)
+	assert.NotContains(t, filterString, "Content-Type: application/json")
+	assert.NotContains(t, filterString, "Retry-After: 2044")
+}
+
+func TestErrorIfaceError(t *testing.T) {
+	tests := []struct {
+		name       string
+		statusCode int
+		err        error
+		code       string
+		message    string
+		expected   string
+	}{
+		{name: "TestNestedErrorNotNilCode0Message0",
+			statusCode: 418,
+			err:        fmt.Errorf("original test message"),
+			code:       "",
+			message:    "",
+			expected:   "original test message"},
+		{name: "TestNestedErrorNotNilCodeXMessageX",
+			statusCode: 418,
+			err:        fmt.Errorf("original test message"),
+			code:       "bad request",
+			message:    "is this a teapot?",
+			expected:   "original test message"},
+		{name: "TestNestedErrorNilCodeXMessageX",
+			statusCode: 418,
+			err:        nil,
+			code:       "bad request",
+			message:    "is this a teapot?",
+			expected:   "bad request: is this a teapot?"},
+		{name: "TestNesterErrorNilCodeXMessage0",
+			statusCode: 418,
+			err:        nil,
+			code:       "I'm a teapot",
+			message:    "",
+			expected:   "Unexpected status code 418"},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			err := Error{
+				StatusCode: test.statusCode,
+				Code:       test.code,
+				Message:    test.message,
+				Err:        test.err,
+				RetryAfter: 0,
+				Header:     ihttp.Header{},
+			}
+			assert.Equal(t, test.expected, err.Error())
+		})
+	}
+}

api/http/service.go

@@ -151,6 +151,7 @@ func (s *service) parseHTTPError(r *http.Response) *Error {
 
 	perror := NewError(nil)
 	perror.StatusCode = r.StatusCode
+	perror.Header = r.Header
 
 	if v := r.Header.Get("Retry-After"); v != "" {
 		r, err := strconv.ParseUint(v, 10, 32)

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)
+}

client_e2e_test.go

@@ -18,6 +18,7 @@ import (
 	"time"
 
 	influxdb2 "github.com/influxdata/influxdb-client-go/v2"
+	"github.com/influxdata/influxdb-client-go/v2/api/http"
 	"github.com/influxdata/influxdb-client-go/v2/domain"
 	"github.com/influxdata/influxdb-client-go/v2/internal/test"
 	"github.com/influxdata/influxdb-client-go/v2/log"
@@ -368,3 +369,16 @@ func TestWriteCustomBatch(t *testing.T) {
 	}
 	assert.Equal(t, 10, l)
 }
+
+func TestHttpHeadersInError(t *testing.T) {
+	client := influxdb2.NewClientWithOptions(serverURL, authToken, influxdb2.DefaultOptions().SetLogLevel(0))
+	err := client.WriteAPIBlocking("my-org", "my-bucket").WriteRecord(context.Background(), "asdf")
+	assert.Error(t, err)
+	assert.Len(t, err.(*http.Error).Header, 6)
+	assert.NotEqual(t, err.(*http.Error).Header.Get("Date"), "")
+	assert.NotEqual(t, err.(*http.Error).Header.Get("Content-Length"), "")
+	assert.NotEqual(t, err.(*http.Error).Header.Get("Content-Type"), "")
+	assert.NotEqual(t, err.(*http.Error).Header.Get("X-Platform-Error-Code"), "")
+	assert.Contains(t, err.(*http.Error).Header.Get("X-Influxdb-Version"), "v")
+	assert.Equal(t, err.(*http.Error).Header.Get("X-Influxdb-Build"), "OSS")
+}

internal/write/service.go

@@ -196,9 +196,24 @@ 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("\nSelected Response Headers:\n%s", logHeaders)
+						}
+						log.Error(logMessage)
 					}
-					return fmt.Errorf("write failed (attempts %d): %w", batchToWrite.RetryAttempts, perror)
+					log.Errorf("Write failed (retry attempts %d): Status Code %d",
+						batchToWrite.RetryAttempts,
+						perror.StatusCode)
+					return perror
 				}
 			}
 

internal/write/service_test.go

@@ -339,7 +339,7 @@ func TestMaxRetryTime(t *testing.T) {
 	err = srv.HandleWrite(ctx, b)
 	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, "Unexpected status code 429", err.Error())
 	assert.Equal(t, 1, srv.retryQueue.list.Len())
 
 	//wait until remaining accumulated retryDelay has passed, because there hasn't been a successful write yet
@@ -702,3 +702,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: { \"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"))
+}