Send metrics to Dynatrace in chunks of 1000

exporter/dynatraceexporter/metrics_exporter.go

@@ -22,6 +22,7 @@ import (
 	"io/ioutil"
 	"net/http"
 	"strings"
+	"time"
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/consumer/consumererror"
@@ -125,10 +126,39 @@ func (e *exporter) serializeMetrics(md pdata.Metrics) ([]string, int) {
 	return lines, dropped
 }
 
+var lastLog int64 = 0
+
 // send sends a serialized metric batch to Dynatrace.
 // Returns the number of lines rejected by Dynatrace.
 // An error indicates all lines were dropped regardless of the returned number.
 func (e *exporter) send(ctx context.Context, lines []string) (int, error) {
+	if now := time.Now().Unix(); len(lines) > 1000 && now-lastLog > 60 {
+		e.logger.Warn("Batch too large. Sending in chunks of 1000 metrics. If any chunk fails, previous chunks in the batch could be retried by the batch processor. Please set send_batch_max_size to 1000 or less. Suppressing this log for 60 seconds.")
+		lastLog = time.Now().Unix()
+	}
+
+	rejected := 0
+	for i := 0; i < len(lines); i += 1000 {
+		end := i + 1000
+
+		if end > len(lines) {
+			end = len(lines)
+		}
+
+		batchRejected, err := e.sendBatch(ctx, lines[i:end])
+		rejected += batchRejected
+		if err != nil {
+			return rejected, err
+		}
+	}
+
+	return rejected, nil
+}
+
+// send sends a serialized metric batch to Dynatrace.
+// Returns the number of lines rejected by Dynatrace.
+// An error indicates all lines were dropped regardless of the returned number.
+func (e *exporter) sendBatch(ctx context.Context, lines []string) (int, error) {
 	message := strings.Join(lines, "\n")
 	e.logger.Debug("Sending lines to Dynatrace\n" + message)
 	req, err := http.NewRequestWithContext(ctx, "POST", e.cfg.Endpoint, bytes.NewBufferString(message))

exporter/dynatraceexporter/metrics_exporter_test.go

@@ -17,6 +17,7 @@ package dynatraceexporter
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"io/ioutil"
 	"net/http"
 	"net/http/httptest"
@@ -281,7 +282,7 @@ func Test_exporter_send_BadRequest(t *testing.T) {
 		},
 		client: ts.Client(),
 	}
-	invalid, err := e.send(context.Background(), []string{})
+	invalid, err := e.send(context.Background(), []string{""})
 	if invalid != 10 {
 		t.Errorf("Expected 10 lines to be reported invalid")
 		return
@@ -310,7 +311,7 @@ func Test_exporter_send_Unauthorized(t *testing.T) {
 		},
 		client: ts.Client(),
 	}
-	_, err := e.send(context.Background(), []string{})
+	_, err := e.send(context.Background(), []string{""})
 	if !consumererror.IsPermanent(err) {
 		t.Errorf("Expected error to be permanent %v", err)
 		return
@@ -335,7 +336,7 @@ func Test_exporter_send_TooLarge(t *testing.T) {
 		},
 		client: ts.Client(),
 	}
-	_, err := e.send(context.Background(), []string{})
+	_, err := e.send(context.Background(), []string{""})
 	if !consumererror.IsPermanent(err) {
 		t.Errorf("Expected error to be permanent %v", err)
 		return
@@ -363,7 +364,7 @@ func Test_exporter_send_NotFound(t *testing.T) {
 		},
 		client: ts.Client(),
 	}
-	_, err := e.send(context.Background(), []string{})
+	_, err := e.send(context.Background(), []string{""})
 	if !consumererror.IsPermanent(err) {
 		t.Errorf("Expected error to be permanent %v", err)
 		return
@@ -374,6 +375,52 @@ func Test_exporter_send_NotFound(t *testing.T) {
 	}
 }
 
+func Test_exporter_send_chunking(t *testing.T) {
+	sentChunks := 0
+
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusBadRequest)
+		body, _ := json.Marshal(metricsResponse{
+			Ok:      0,
+			Invalid: 1,
+		})
+		w.Write(body)
+		sentChunks++
+	}))
+	defer ts.Close()
+
+	e := &exporter{
+		logger: zap.NewNop(),
+		cfg: &config.Config{
+			HTTPClientSettings: confighttp.HTTPClientSettings{Endpoint: ts.URL},
+		},
+		client: ts.Client(),
+	}
+
+	batch := make([]string, 1001)
+
+	for i := 0; i < 1001; i++ {
+		batch[i] = fmt.Sprintf("%d", i)
+	}
+
+	invalid, err := e.send(context.Background(), batch)
+	if sentChunks != 2 {
+		t.Errorf("Expected batch to be sent in 2 chunks")
+	}
+	if invalid != 2 {
+		t.Errorf("Expected 2 lines to be reported invalid")
+		return
+	}
+	if consumererror.IsPermanent(err) {
+		t.Errorf("Expected error to not be permanent %v", err)
+		return
+	}
+	if e.isDisabled {
+		t.Error("Expected exporter to not be disabled")
+		return
+	}
+}
+
 func Test_exporter_PushMetricsData_Error(t *testing.T) {
 	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		w.WriteHeader(200)