Forward buffer (#549)

* push up forward output

* Refactored flush buffer

* Updated buffer close test for forward output

* Added closing of test server in forward tests

Signed-off-by: Corbin Phelps <[email protected]>

Co-authored-by: jmwilliams89 <[email protected]>
Co-authored-by: Corbin Phelps <[email protected]>

operator/builtin/output/forward/forward.go

@@ -8,6 +8,7 @@ import (
 	"io/ioutil"
 	"net/http"
 	"sync"
+	"time"
 
 	"github.com/observiq/stanza/v2/operator/buffer"
 	"github.com/observiq/stanza/v2/operator/flusher"
@@ -102,11 +103,40 @@ func (f *ForwardOutput) Stop() error {
 	f.cancel()
 	f.wg.Wait()
 	f.flusher.Stop()
-	// TODO handle buffer close entries
-	_, err := f.buffer.Close()
+
+	entries, err := f.buffer.Close()
+	if err != nil {
+		f.Errorf("Failed to retreive entries")
+		return err
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
+	defer cancel()
+
+	err = f.sendEntries(ctx, entries)
 	return err
 }
 
+// sendEntries sends entries using the configured client
+func (f *ForwardOutput) sendEntries(ctx context.Context, entries []*entry.Entry) error {
+	req, err := f.createRequest(ctx, entries)
+	if err != nil {
+		f.Errorf("Failed to create request", zap.Error(err))
+		return err
+	}
+
+	res, err := f.client.Do(req)
+	if err != nil {
+		return otelerrors.Wrap(err, "send request")
+	}
+
+	if err := f.handleResponse(res); err != nil {
+		return err
+	}
+
+	return nil
+}
+
 // Process adds an entry to the outputs buffer
 func (f *ForwardOutput) Process(ctx context.Context, entry *entry.Entry) error {
 	return f.buffer.Add(ctx, entry)
@@ -136,22 +166,7 @@ func (f *ForwardOutput) feedFlusher(ctx context.Context) {
 		}
 
 		f.flusher.Do(ctx, func(flushCtx context.Context) error {
-			req, err := f.createRequest(flushCtx, entries)
-			if err != nil {
-				f.Errorf("Failed to create request", zap.Error(err))
-				return nil
-			}
-
-			res, err := f.client.Do(req)
-			if err != nil {
-				return otelerrors.Wrap(err, "send request")
-			}
-
-			if err := f.handleResponse(res); err != nil {
-				return err
-			}
-
-			return nil
+			return f.sendEntries(flushCtx, entries)
 		})
 	}
 }

operator/builtin/output/forward/forward_test.go

@@ -23,6 +23,7 @@ func TestForwardOutput(t *testing.T) {
 		body, _ := ioutil.ReadAll(req.Body)
 		received <- body
 	}))
+	defer srv.Close()
 
 	cfg := NewForwardOutputConfig("test")
 	memoryCfg := buffer.NewMemoryBufferConfig()
@@ -59,3 +60,32 @@ func TestForwardOutput(t *testing.T) {
 		require.Equal(t, newEntry.Resource, e.Resource)
 	}
 }
+
+func TestFlushBufferOnClose(t *testing.T) {
+	cfg := NewForwardOutputConfig("test")
+	received := make(chan []byte, 1)
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+		body, _ := ioutil.ReadAll(req.Body)
+		received <- body
+	}))
+	defer srv.Close()
+
+	cfg.Address = srv.URL
+	ops, err := cfg.Build((testutil.NewBuildContext(t)))
+	require.NoError(t, err)
+
+	forwardOutput, ok := ops[0].(*ForwardOutput)
+	require.True(t, ok)
+
+	newEntry := entry.New()
+	newEntry.Body = "test"
+	newEntry.Timestamp = newEntry.Timestamp.Round(time.Second)
+	err = forwardOutput.Process(forwardOutput.ctx, newEntry)
+	require.NoError(t, err)
+
+	err = forwardOutput.Stop()
+	require.NoError(t, err)
+
+	response := <-received
+	require.Contains(t, string(response), `"body":"test"`)
+}