Rewind the request body for retries

client/internal/httpsender.go

@@ -26,6 +26,23 @@
 const headerContentEncoding = "Content-Encoding"
 const encodingTypeGZip = "gzip"
 
+type requestWrapper struct {
+	*http.Request
+
+	bodyReader func() io.ReadCloser
+}
+
+func bodyReader(buf []byte) func() io.ReadCloser {
+	return func() io.ReadCloser {
+		return io.NopCloser(bytes.NewReader(buf))
+	}
+}
+
+func (r *requestWrapper) rewind(ctx context.Context) {
+	r.Body = r.bodyReader()
+	r.Request = r.Request.WithContext(ctx)
+}
+
 // HTTPSender allows scheduling messages to send. Once run, it will loop through
 // a request/response cycle for each message to send and will process all received
 // responses using a receivedProcessor. If there are no pending messages to send
@@ -156,7 +173,8 @@
 		select {
 		case <-timer.C:
 			{
-				resp, err := h.client.Do(req)
+				req.rewind(ctx)
+				resp, err := h.client.Do(req.Request)
 				if err == nil {
 					switch resp.StatusCode {
 					case http.StatusOK:
@@ -198,7 +216,7 @@
 	return interval
 }
 
-func (h *HTTPSender) prepareRequest(ctx context.Context) (*http.Request, error) {
+func (h *HTTPSender) prepareRequest(ctx context.Context) (*requestWrapper, error) {
 	msgToSend := h.nextMessage.PopPending()
 	if msgToSend == nil || proto.Equal(msgToSend, &protobufs.AgentToServer{}) {
 		// There is no pending message or the message is empty.
@@ -211,7 +229,11 @@
 		return nil, err
 	}
 
-	var body io.Reader
+	r, err := http.NewRequestWithContext(ctx, OpAMPPlainHTTPMethod, h.url, nil)
+	if err != nil {
+		return nil, err
+	}
+	req := requestWrapper{Request: r}
 
 	if h.compressionEnabled {
 		var buf bytes.Buffer
@@ -224,17 +246,16 @@
 			h.logger.Errorf("Failed to close the writer: %v", err)
 			return nil, err
 		}
-		body = &buf
+		req.bodyReader = bodyReader(buf.Bytes())
 	} else {
-		body = bytes.NewReader(data)
+		req.bodyReader = bodyReader(data)
 	}
-	req, err := http.NewRequestWithContext(ctx, OpAMPPlainHTTPMethod, h.url, body)
 	if err != nil {
 		return nil, err
 	}
 
 	req.Header = h.requestHeader
-	return req, nil
+	return &req, nil
 }
 
 func (h *HTTPSender) receiveResponse(ctx context.Context, resp *http.Response) {

client/internal/httpsender_test.go

@@ -3,13 +3,18 @@ package internal
 import (
 	"context"
 	"crypto/tls"
+	"io"
+	"net"
 	"net/http"
+	"net/http/httptest"
+	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/open-telemetry/opamp-go/client/types"
 	sharedinternal "github.com/open-telemetry/opamp-go/internal"
+	"github.com/open-telemetry/opamp-go/internal/testhelpers"
 	"github.com/open-telemetry/opamp-go/protobufs"
 	"github.com/stretchr/testify/assert"
 )
@@ -98,3 +103,72 @@ EKTcWGekdmdDPsHloRNtsiCa697B2O9IFA==
 
 	return cert, nil
 }
+
+func TestHTTPSenderRetryForFailedRequests(t *testing.T) {
+
+	srv, m := newMockServer(t)
+	address := testhelpers.GetAvailableLocalAddress()
+	var connectionAttempts int64
+
+	var buf []byte
+	srv.OnRequest = func(w http.ResponseWriter, r *http.Request) {
+		attempt := atomic.AddInt64(&connectionAttempts, 1)
+		if attempt == 1 {
+			hj, ok := w.(http.Hijacker)
+			if !ok {
+				t.Error("server doesn't support hijacking")
+				return
+			}
+			conn, _, err := hj.Hijack()
+			if err != nil {
+				t.Error(err)
+				return
+			}
+			conn.Close()
+		} else {
+			buf, _ = io.ReadAll(r.Body)
+			w.WriteHeader(http.StatusOK)
+		}
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	url := "http://" + address
+	sender := NewHTTPSender(&sharedinternal.NopLogger{})
+	sender.NextMessage().Update(func(msg *protobufs.AgentToServer) {
+		msg.AgentDescription = &protobufs.AgentDescription{
+			IdentifyingAttributes: []*protobufs.KeyValue{{
+				Key: "service.name",
+				Value: &protobufs.AnyValue{
+					Value: &protobufs.AnyValue_StringValue{StringValue: "test-service"},
+				},
+			}},
+		}
+	})
+	sender.callbacks = types.CallbacksStruct{
+		OnConnectFunc: func() {
+		},
+		OnConnectFailedFunc: func(_ error) {
+		},
+	}
+	sender.url = url
+	var wg sync.WaitGroup
+	wg.Add(2)
+	go func() {
+		sender.sendRequestWithRetries(ctx)
+		wg.Done()
+	}()
+	go func() {
+		l, err := net.Listen("tcp", address)
+		assert.NoError(t, err)
+		ts := httptest.NewUnstartedServer(m)
+		ts.Listener.Close()
+		ts.Listener = l
+		ts.Start()
+		srv.srv = ts
+		wg.Done()
+	}()
+	wg.Wait()
+	assert.True(t, len(buf) > 0)
+	assert.Contains(t, string(buf), "test-service")
+	cancel()
+	srv.Close()
+}