fix(outputs.influxdb_v2): Fix panic and API error handling

Author: srebhan

plugins/outputs/influxdb_v2/http.go

@@ -26,16 +26,17 @@ import (
 )
 
 type APIError struct {
-	StatusCode  int
-	Title       string
-	Description string
+	Err        error
+	StatusCode int
+	Retryable  bool
 }
 
 func (e APIError) Error() string {
-	if e.Description != "" {
-		return fmt.Sprintf("%s: %s", e.Title, e.Description)
-	}
-	return e.Title
+	return e.Err.Error()
+}
+
+func (e APIError) Unwrap() error {
+	return e.Err
 }
 
 const (
@@ -185,7 +186,7 @@ func (c *httpClient) Write(ctx context.Context, metrics []telegraf.Metric) error
 			}
 
 			batches[bucket] = append(batches[bucket], metric)
-			batchIndices[c.bucket] = append(batchIndices[c.bucket], i)
+			batchIndices[bucket] = append(batchIndices[bucket], i)
 		}
 	}
 
@@ -201,10 +202,14 @@ func (c *httpClient) Write(ctx context.Context, metrics []telegraf.Metric) error
 		var apiErr *APIError
 		if errors.As(err, &apiErr) {
 			if apiErr.StatusCode == http.StatusRequestEntityTooLarge {
+				// TODO: Need a testcase to verify rejected metrics are not retried...
 				return c.splitAndWriteBatch(ctx, c.bucket, metrics)
 			}
 			wErr.Err = err
-			wErr.MetricsReject = append(wErr.MetricsReject, batchIndices[bucket]...)
+			if !apiErr.Retryable {
+				wErr.MetricsReject = append(wErr.MetricsReject, batchIndices[bucket]...)
+			}
+			// TODO: Clarify if we should continue here to try the remaining buckets?
 			return &wErr
 		}
 
@@ -301,21 +306,19 @@ func (c *httpClient) writeBatch(ctx context.Context, bucket string, metrics []te
 	}
 
 	// We got an error and now try to decode further
+	var desc string
 	writeResp := &genericRespError{}
-	err = json.NewDecoder(resp.Body).Decode(writeResp)
-	desc := writeResp.Error()
-	if err != nil {
-		desc = resp.Status
+	if json.NewDecoder(resp.Body).Decode(writeResp) == nil {
+		desc = writeResp.Error()
 	}
 
 	switch resp.StatusCode {
 	// request was too large, send back to try again
 	case http.StatusRequestEntityTooLarge:
 		c.log.Errorf("Failed to write metric to %s, request was too large (413)", bucket)
 		return &APIError{
-			StatusCode:  resp.StatusCode,
-			Title:       resp.Status,
-			Description: desc,
+			Err:        fmt.Errorf("%s: %s", resp.Status, desc),
+			StatusCode: resp.StatusCode,
 		}
 	case
 		// request was malformed:
@@ -325,14 +328,10 @@ func (c *httpClient) writeBatch(ctx context.Context, bucket string, metrics []te
 		http.StatusUnprocessableEntity,
 		http.StatusNotAcceptable:
 
-		// Clients should *not* repeat the request and the metrics should be dropped.
-		rejected := make([]int, 0, len(metrics))
-		for i := range len(metrics) {
-			rejected = append(rejected, i)
-		}
-		return &internal.PartialWriteError{
-			Err:           fmt.Errorf("failed to write metric to %s (will be dropped: %s): %s", bucket, resp.Status, desc),
-			MetricsReject: rejected,
+		// Clients should *not* repeat the request and the metrics should be rejected.
+		return &APIError{
+			Err:        fmt.Errorf("failed to write metric to %s (will be dropped: %s): %s", bucket, resp.Status, desc),
+			StatusCode: resp.StatusCode,
 		}
 	case http.StatusUnauthorized, http.StatusForbidden:
 		return fmt.Errorf("failed to write metric to %s (%s): %s", bucket, resp.Status, desc)
@@ -351,13 +350,9 @@ func (c *httpClient) writeBatch(ctx context.Context, bucket string, metrics []te
 	// if it's any other 4xx code, the client should not retry as it's the client's mistake.
 	// retrying will not make the request magically work.
 	if len(resp.Status) > 0 && resp.Status[0] == '4' {
-		rejected := make([]int, 0, len(metrics))
-		for i := range len(metrics) {
-			rejected = append(rejected, i)
-		}
-		return &internal.PartialWriteError{
-			Err:           fmt.Errorf("failed to write metric to %s (will be dropped: %s): %s", bucket, resp.Status, desc),
-			MetricsReject: rejected,
+		return &APIError{
+			Err:        fmt.Errorf("failed to write metric to %s (will be dropped: %s): %s", bucket, resp.Status, desc),
+			StatusCode: resp.StatusCode,
 		}
 	}
 
@@ -367,10 +362,14 @@ func (c *httpClient) writeBatch(ctx context.Context, bucket string, metrics []te
 		desc = fmt.Sprintf("%s; %s", desc, xErr)
 	}
 
+	if desc != "" {
+		desc = ": " + desc
+	}
+
 	return &APIError{
-		StatusCode:  resp.StatusCode,
-		Title:       resp.Status,
-		Description: desc,
+		Err:        fmt.Errorf("failed to write metric to bucket %q: %s%s", bucket, resp.Status, desc),
+		StatusCode: resp.StatusCode,
+		Retryable:  true,
 	}
 }
 

plugins/outputs/influxdb_v2/influxdb_v2.go

@@ -199,11 +199,11 @@ func (i *InfluxDB) Write(metrics []telegraf.Metric) error {
 	for _, n := range rand.Perm(len(i.clients)) {
 		client := i.clients[n]
 		if err := client.Write(ctx, metrics); err != nil {
+			i.Log.Errorf("When writing to [%s]: %v", client.url, err)
 			var werr *internal.PartialWriteError
 			if errors.As(err, &werr) || errors.Is(err, internal.ErrSizeLimitReached) {
 				return err
 			}
-			i.Log.Errorf("When writing to [%s]: %v", client.url, err)
 			continue
 		}
 		return nil