Restore protocol/signalfx.SetupChain (#144)

* Restore protocol/signalfx.SetupChain

It's still being used in the signalfx-forwarder monitor

* Update protocol/zipper/zipper.go

Co-authored-by: Curtis Robert <[email protected]>

---------

Co-authored-by: Curtis Robert <[email protected]>

Author: dmitryax

Makefile

@@ -36,7 +36,7 @@ lint: ; $(info $(M) running linting on $(CURDIR))
 
 ARGS                                        = -race -timeout=60s -failfast
 COVERAGE_MODE                               = atomic
-REQUIRED_COVERAGE                           = 93.0
+REQUIRED_COVERAGE                           = 92.0
 COVERAGE_DIR                                = $(CURDIR)/coverage.$(shell date -u +"%Y_%m_%dT%H_%M_%SZ")
 COVERAGE_P_FILE                             = $(COVERAGE_DIR)/coverage/parallel/coverage.out
 COVERAGE_S_FILE                             = $(COVERAGE_DIR)/coverage/serialized/coverage.out

protocol/signalfx/datapoint.go

@@ -24,9 +24,11 @@ import (
 	"github.com/signalfx/golib/v3/errors"
 	"github.com/signalfx/golib/v3/log"
 	"github.com/signalfx/golib/v3/sfxclient"
+	"github.com/signalfx/golib/v3/web"
 	"github.com/signalfx/ingest-protocols/logkey"
 	"github.com/signalfx/ingest-protocols/protocol"
 	signalfxformat "github.com/signalfx/ingest-protocols/protocol/signalfx/format"
+	"github.com/signalfx/ingest-protocols/protocol/zipper"
 )
 
 // MericTypeGetter is an old metric interface that returns the type of a metric name
@@ -262,3 +264,70 @@ func SetupJSONV1Paths(r *mux.Router, handler http.Handler) {
 	SetupJSONByPaths(r, handler, "/datapoint")
 	SetupJSONByPaths(r, handler, "/v1/datapoint")
 }
+
+// ErrorTrackerHandler behaves like a http handler, but tracks error returns from a ErrorReader
+type ErrorTrackerHandler struct {
+	TotalErrors int64
+	reader      ErrorReader
+	Logger      log.Logger
+}
+
+// Datapoints gets TotalErrors stats
+func (e *ErrorTrackerHandler) Datapoints() []*datapoint.Datapoint {
+	return []*datapoint.Datapoint{
+		sfxclient.Cumulative("total_errors", nil, atomic.LoadInt64(&e.TotalErrors)),
+	}
+}
+
+func addTokenToContext(ctx context.Context, req *http.Request) context.Context {
+	head := req.Header.Get(sfxclient.TokenHeaderName)
+	if head != "" {
+		ctx = context.WithValue(ctx, sfxclient.TokenHeaderName, head)
+	}
+	return ctx
+}
+
+// ServeHTTPC will serve the wrapped ErrorReader and return the error (if any) to rw if ErrorReader
+// fails
+func (e *ErrorTrackerHandler) ServeHTTPC(ctx context.Context, rw http.ResponseWriter, req *http.Request) {
+	ctx = addTokenToContext(ctx, req)
+	if err := e.reader.Read(ctx, req); err != nil {
+		atomic.AddInt64(&e.TotalErrors, 1)
+		rw.WriteHeader(http.StatusBadRequest)
+		_, err = rw.Write([]byte(err.Error()))
+		log.IfErr(e.Logger, err)
+		return
+	}
+	_, err := rw.Write([]byte(`"OK"`))
+	log.IfErr(e.Logger, err)
+}
+
+// SetupChain wraps the reader returned by getReader in an http.Handler along
+// with some middleware that calculates internal metrics about requests.
+func SetupChain(ctx context.Context, sink Sink, chainType string, getReader func(Sink) ErrorReader, httpChain web.NextConstructor, logger log.Logger, counter *dpsink.Counter, moreConstructors ...web.Constructor) (http.Handler, sfxclient.Collector) {
+	zippers := zipper.NewZipper()
+
+	ucount := UnifyNextSinkWrap(counter)
+	finalSink := FromChain(sink, NextWrap(ucount))
+	errReader := getReader(finalSink)
+	errorTracker := ErrorTrackerHandler{
+		reader: errReader,
+		Logger: logger,
+	}
+	metricTracking := web.RequestCounter{}
+	handler := web.NewHandler(ctx, &errorTracker).Add(web.NextHTTP(metricTracking.ServeHTTP)).Add(httpChain)
+	for _, c := range moreConstructors {
+		handler.Add(c)
+	}
+	st := &sfxclient.WithDimensions{
+		Collector: sfxclient.NewMultiCollector(
+			&metricTracking,
+			&errorTracker,
+			zippers,
+		),
+		Dimensions: map[string]string{
+			"protocol": "sfx_" + chainType,
+		},
+	}
+	return zippers.GzipHandler(handler), st
+}

protocol/zipper/zipper.go

@@ -0,0 +1,86 @@
+package zipper
+
+import (
+	"bytes"
+	"compress/gzip"
+	"io"
+	"net/http"
+	"sync"
+	"sync/atomic"
+
+	"github.com/signalfx/golib/v3/datapoint"
+	"github.com/signalfx/golib/v3/log"
+	"github.com/signalfx/golib/v3/sfxclient"
+)
+
+// ReadZipper creates a Pool that contains previously used Readers and can create new ones if we run out.
+type ReadZipper struct {
+	zippers   sync.Pool
+	Log       log.Logger
+	NewCount  int64
+	HitCount  int64
+	MissCount int64
+	ErrCount  int64
+}
+
+// GzipHandler transparently decodes your possibly gzipped request
+func (z *ReadZipper) GzipHandler(h http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		var err error
+		if r.Header.Get("Content-Encoding") == "gzip" {
+			gzi := z.zippers.Get()
+			if gzi != nil {
+				gz := gzi.(*gzip.Reader)
+				// put it back
+				defer z.zippers.Put(gz)
+				err = gz.Reset(r.Body)
+				if err == nil {
+					defer log.IfErr(z.Log, gz.Close())
+					// nasty? could construct another object but seems expensive
+					r.Body = gz
+					atomic.AddInt64(&z.HitCount, 1)
+					h.ServeHTTP(w, r)
+					return
+				}
+			}
+		}
+		if err != nil {
+			atomic.AddInt64(&z.ErrCount, 1)
+			w.WriteHeader(http.StatusBadRequest)
+			_, err = w.Write([]byte("error handling gzip compressed request " + err.Error()))
+			log.IfErr(z.Log, err)
+			return
+		}
+		atomic.AddInt64(&z.MissCount, 1)
+		h.ServeHTTP(w, r)
+	})
+}
+
+// Datapoints implements Collector interface and returns metrics
+func (z *ReadZipper) Datapoints() []*datapoint.Datapoint {
+	return []*datapoint.Datapoint{
+		sfxclient.CumulativeP("zipper.hitCount", nil, &z.HitCount),
+		sfxclient.CumulativeP("zipper.missCount", nil, &z.MissCount),
+		sfxclient.CumulativeP("zipper.newCount", nil, &z.NewCount),
+		sfxclient.CumulativeP("zipper.errCount", nil, &z.ErrCount),
+	}
+}
+
+// NewZipper gives you a ReadZipper
+func NewZipper() *ReadZipper {
+	return newZipper(gzip.NewReader)
+}
+
+func newZipper(zipperFunc func(r io.Reader) (*gzip.Reader, error)) *ReadZipper {
+	z := &ReadZipper{}
+	z.zippers = sync.Pool{New: func() interface{} {
+		atomic.AddInt64(&z.NewCount, 1)
+		// This is just the header of an empty gzip, unlike NewWriter, it can't pass in nil to empty bytes
+		g, err := zipperFunc(bytes.NewBuffer([]byte{31, 139, 8, 0, 0, 0, 0, 0, 0, 255}))
+		if err == nil {
+			return g
+		}
+		return nil
+	}}
+	return z
+}

protocol/zipper/zipper_test.go

@@ -0,0 +1,68 @@
+package zipper
+
+import (
+	"bytes"
+	"compress/gzip"
+	"context"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/signalfx/golib/v3/errors"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+type foo struct{}
+
+func (f *foo) ServeHTTP(rw http.ResponseWriter, req *http.Request) {
+	buf := new(bytes.Buffer)
+	if _, err := buf.ReadFrom(req.Body); err == nil {
+		if buf.String() == "OK" {
+			rw.WriteHeader(http.StatusOK)
+		}
+	}
+	rw.WriteHeader(http.StatusBadRequest)
+}
+
+func TestZipper(t *testing.T) {
+	zippers := NewZipper()
+	badZippers := newZipper(func(io.Reader) (*gzip.Reader, error) {
+		return new(gzip.Reader), errors.New("nope")
+	})
+	f := new(foo)
+	zipped := new(bytes.Buffer)
+	w := gzip.NewWriter(zipped)
+	_, err := w.Write([]byte("OK"))
+	require.NoError(t, err)
+	require.NoError(t, w.Close())
+	tests := []struct {
+		zipper  *ReadZipper
+		name    string
+		data    []byte
+		status  int
+		headers map[string]string
+	}{
+		{zippers, "test non gzipped", []byte("OK"), http.StatusOK, map[string]string{}},
+		{zippers, "test gzipped", zipped.Bytes(), http.StatusOK, map[string]string{"Content-Encoding": "gzip"}},
+		{zippers, "test gzipped but bad", zipped.Bytes()[:5], http.StatusBadRequest, map[string]string{"Content-Encoding": "gzip"}},
+		{badZippers, "test gzipped failure", zipped.Bytes(), http.StatusBadRequest, map[string]string{"Content-Encoding": "gzip"}},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			req, err := http.NewRequestWithContext(context.Background(), "GET", "/health-check", bytes.NewBuffer(test.data))
+			for k, v := range test.headers {
+				req.Header.Set(k, v)
+			}
+			require.NoError(t, err)
+			rr := httptest.NewRecorder()
+			g := test.zipper.GzipHandler(f)
+			g.ServeHTTP(rr, req)
+			assert.Equal(t, test.status, rr.Code)
+		})
+	}
+	t.Run("check datapoints", func(t *testing.T) {
+		assert.Equal(t, 4, len(zippers.Datapoints()))
+	})
+}