<commit_before>package signalfx

import (
	"bufio"
	"encoding/json"
	"fmt"
	"io"
	"net"
	"net/http"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"code.google.com/p/goprotobuf/proto"
	log "github.com/Sirupsen/logrus"
	"github.com/cep21/gohelpers/structdefaults"
	"github.com/cep21/gohelpers/workarounds"
	"github.com/codegangsta/negroni"
	"github.com/gorilla/mux"
	"github.com/signalfuse/com_signalfuse_metrics_protobuf"
	"github.com/signalfuse/signalfxproxy/config"
	"github.com/signalfuse/signalfxproxy/datapoint"
	"github.com/signalfuse/signalfxproxy/protocol/collectd"
	"github.com/signalfuse/signalfxproxy/reqcounter"
	"github.com/signalfuse/signalfxproxy/stats"
)

type listenerServer struct {
	name     string
	listener net.Listener

	statKeepers []stats.Keeper
}

func (streamer *listenerServer) Stats() []datapoint.Datapoint {
	return stats.CombineStats(streamer.statKeepers)
}

func (streamer *listenerServer) Close() {
	streamer.listener.Close()
}

// MericTypeGetter is an old metric interface that returns the type of a metric name
type MericTypeGetter interface {
	GetMetricTypeFromMap(metricName string) com_signalfuse_metrics_protobuf.MetricType
}

// ErrorReader are datapoint streamers that read from a HTTP request and return errors if
// the stream is invalid
type ErrorReader interface {
	Read(req *http.Request) error
}

// ErrorTrackerHandler behaves like a http handler, but tracks error returns from a ErrorReader
type ErrorTrackerHandler struct {
	TotalErrors int64
	reader      ErrorReader
}

// Stats returns the number of calls to AddDatapoint
func (e *ErrorTrackerHandler) Stats(dimensions map[string]string) []datapoint.Datapoint {
	ret := []datapoint.Datapoint{}
	ret = append(
		ret,
		datapoint.NewOnHostDatapointDimensions(
			"total_errors",
			datapoint.NewIntValue(e.TotalErrors),
			com_signalfuse_metrics_protobuf.MetricType_CUMULATIVE_COUNTER,
			dimensions))
	return ret
}

func (e *ErrorTrackerHandler) ServeHTTP(rw http.ResponseWriter, req *http.Request) {
	if err := e.reader.Read(req); err != nil {
		atomic.AddInt64(&e.TotalErrors, 1)
		rw.WriteHeader(http.StatusBadRequest)
		rw.Write([]byte(err.Error()))
		return
	}
	rw.Write([]byte(`"OK"`))
}

type protobufDecoderV1 struct {
	typeGetter       MericTypeGetter
	datapointTracker datapoint.Tracker
}

var errInvalidProtobuf = fmt.Errorf("Invalid protocol buffer sent")
var errProtobufTooLarge = fmt.Errorf("Protobuf structure too large")
var errInvalidProtobufVarint = fmt.Errorf("Invalid protobuf varint")

func (decoder *protobufDecoderV1) Read(req *http.Request) error {
	body := req.Body
	bufferedBody := bufio.NewReaderSize(body, 32768)
	for {
		log.WithField("body", body).Debug("Starting protobuf loop")
		buf, err := bufferedBody.Peek(1)
		if err == io.EOF {
			log.Debug("EOF")
			return nil
		}
		buf, err = bufferedBody.Peek(4) // should be big enough for any varint

		if err != nil {
			log.WithField("err", err).Info("peek error")
			return err
		}
		log.Debug("Decoding varint")
		num, bytesRead := proto.DecodeVarint(buf)
		log.WithField("num", num).WithField("bytesRead", bytesRead).Debug("Decode results")
		log.WithFields(log.Fields{"num": num, "bytesRead": bytesRead}).Debug("Decoding result")
		if bytesRead == 0 {
			// Invalid varint?
			return errInvalidProtobufVarint
		}
		if num > 32768 {
			// Sanity check
			return errProtobufTooLarge
		}
		// Get the varint out
		buf = make([]byte, bytesRead)
		io.ReadFull(bufferedBody, buf)

		// Get the structure out
		buf = make([]byte, num)
		_, err = io.ReadFull(bufferedBody, buf)
		if err != nil {
			return fmt.Errorf("unable to fully read protobuf message: %s", err)
		}
		var msg com_signalfuse_metrics_protobuf.DataPoint
		err = proto.Unmarshal(buf, &msg)
		if err != nil {
			return err
		}
		if msg.Metric == nil || msg.Value == nil {
			return errInvalidProtobuf
		}
		mt := decoder.typeGetter.GetMetricTypeFromMap(msg.GetMetric())
		dp := NewProtobufDataPointWithType(&msg, mt)
		log.WithField("dp", dp).Debug("Adding a point")
		decoder.datapointTracker.AddDatapoint(dp)
	}
}

type jsonDecoderV1 struct {
	typeGetter       MericTypeGetter
	datapointTracker datapoint.Tracker
}

func (decoder *jsonDecoderV1) Read(req *http.Request) error {
	dec := json.NewDecoder(req.Body)
	for {
		var d JSONDatapointV1
		if err := dec.Decode(&d); err == io.EOF {
			break
		} else if err != nil {
			return err
		} else {
			log.WithField("dp", d).Debug("Got a new point")
			if d.Metric == "" {
				log.WithField("dp", d).Debug("Invalid Datapoint")
				continue
			}
			mt := decoder.typeGetter.GetMetricTypeFromMap(d.Metric)
			dp := datapoint.NewRelativeTime(d.Metric, map[string]string{"sf_source": d.Source}, datapoint.NewFloatValue(d.Value), mt, 0)
			decoder.datapointTracker.AddDatapoint(dp)
		}
	}
	return nil
}

type protobufDecoderV2 struct {
	datapointTracker datapoint.Tracker
}

var errInvalidContentLength = fmt.Errorf("Invalid Content Length")

func (decoder *protobufDecoderV2) Read(req *http.Request) error {
	return DecodeProtobufV2(req, &decoder.datapointTracker)
}

// DecodeProtobufV2 will take a request of signalfx's V2 protocol buffers and forward them to datapointTracker
func DecodeProtobufV2(req *http.Request, datapointTracker datapoint.Adder) error {
	if req.ContentLength == -1 {
		return errInvalidContentLength
	}

	buf := make([]byte, req.ContentLength)
	readLen, err := io.ReadFull(req.Body, buf)
	if err != nil {
		log.WithField("err", err).WithField("len", readLen).WithField("content-len", req.ContentLength).Warn("Unable to fully read from buffer")
		return err
	}
	var msg com_signalfuse_metrics_protobuf.DataPointUploadMessage
	err = proto.Unmarshal(buf, &msg)
	if err != nil {
		return err
	}
	for _, protoDb := range msg.GetDatapoints() {
		dp := NewProtobufDataPointWithType(protoDb, com_signalfuse_metrics_protobuf.MetricType_GAUGE)
		datapointTracker.AddDatapoint(dp)
	}
	return nil
}

type jsonDecoderV2 struct {
	datapointTracker datapoint.Tracker<commit_msg>Add doc for future improvements<commit_after>package signalfx

import (
	"bufio"
	"encoding/json"
	"fmt"
	"io"
	"net"
	"net/http"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"code.google.com/p/goprotobuf/proto"
	log "github.com/Sirupsen/logrus"
	"github.com/cep21/gohelpers/structdefaults"
	"github.com/cep21/gohelpers/workarounds"
	"github.com/codegangsta/negroni"
	"github.com/gorilla/mux"
	"github.com/signalfuse/com_signalfuse_metrics_protobuf"
	"github.com/signalfuse/signalfxproxy/config"
	"github.com/signalfuse/signalfxproxy/datapoint"
	"github.com/signalfuse/signalfxproxy/protocol/collectd"
	"github.com/signalfuse/signalfxproxy/reqcounter"
	"github.com/signalfuse/signalfxproxy/stats"
)

type listenerServer struct {
	name     string
	listener net.Listener

	statKeepers []stats.Keeper
}

func (streamer *listenerServer) Stats() []datapoint.Datapoint {
	return stats.CombineStats(streamer.statKeepers)
}

func (streamer *listenerServer) Close() {
	streamer.listener.Close()
}

// MericTypeGetter is an old metric interface that returns the type of a metric name
type MericTypeGetter interface {
	GetMetricTypeFromMap(metricName string) com_signalfuse_metrics_protobuf.MetricType
}

// ErrorReader are datapoint streamers that read from a HTTP request and return errors if
// the stream is invalid
type ErrorReader interface {
	Read(req *http.Request) error
}

// ErrorTrackerHandler behaves like a http handler, but tracks error returns from a ErrorReader
type ErrorTrackerHandler struct {
	TotalErrors int64
	reader      ErrorReader
}

// Stats returns the number of calls to AddDatapoint
func (e *ErrorTrackerHandler) Stats(dimensions map[string]string) []datapoint.Datapoint {
	ret := []datapoint.Datapoint{}
	ret = append(
		ret,
		datapoint.NewOnHostDatapointDimensions(
			"total_errors",
			datapoint.NewIntValue(e.TotalErrors),
			com_signalfuse_metrics_protobuf.MetricType_CUMULATIVE_COUNTER,
			dimensions))
	return ret
}

func (e *ErrorTrackerHandler) ServeHTTP(rw http.ResponseWriter, req *http.Request) {
	if err := e.reader.Read(req); err != nil {
		atomic.AddInt64(&e.TotalErrors, 1)
		rw.WriteHeader(http.StatusBadRequest)
		rw.Write([]byte(err.Error()))
		return
	}
	rw.Write([]byte(`"OK"`))
}

type protobufDecoderV1 struct {
	typeGetter       MericTypeGetter
	datapointTracker datapoint.Tracker
}

var errInvalidProtobuf = fmt.Errorf("Invalid protocol buffer sent")
var errProtobufTooLarge = fmt.Errorf("Protobuf structure too large")
var errInvalidProtobufVarint = fmt.Errorf("Invalid protobuf varint")

func (decoder *protobufDecoderV1) Read(req *http.Request) error {
	body := req.Body
	bufferedBody := bufio.NewReaderSize(body, 32768)
	for {
		log.WithField("body", body).Debug("Starting protobuf loop")
		buf, err := bufferedBody.Peek(1)
		if err == io.EOF {
			log.Debug("EOF")
			return nil
		}
		buf, err = bufferedBody.Peek(4) // should be big enough for any varint

		if err != nil {
			log.WithField("err", err).Info("peek error")
			return err
		}
		log.Debug("Decoding varint")
		num, bytesRead := proto.DecodeVarint(buf)
		log.WithField("num", num).WithField("bytesRead", bytesRead).Debug("Decode results")
		log.WithFields(log.Fields{"num": num, "bytesRead": bytesRead}).Debug("Decoding result")
		if bytesRead == 0 {
			// Invalid varint?
			return errInvalidProtobufVarint
		}
		if num > 32768 {
			// Sanity check
			return errProtobufTooLarge
		}
		// Get the varint out
		buf = make([]byte, bytesRead)
		io.ReadFull(bufferedBody, buf)

		// Get the structure out
		buf = make([]byte, num)
		_, err = io.ReadFull(bufferedBody, buf)
		if err != nil {
			return fmt.Errorf("unable to fully read protobuf message: %s", err)
		}
		var msg com_signalfuse_metrics_protobuf.DataPoint
		err = proto.Unmarshal(buf, &msg)
		if err != nil {
			return err
		}
		if msg.Metric == nil || msg.Value == nil {
			return errInvalidProtobuf
		}
		mt := decoder.typeGetter.GetMetricTypeFromMap(msg.GetMetric())
		dp := NewProtobufDataPointWithType(&msg, mt)
		log.WithField("dp", dp).Debug("Adding a point")
		decoder.datapointTracker.AddDatapoint(dp)
	}
}

type jsonDecoderV1 struct {
	typeGetter       MericTypeGetter
	datapointTracker datapoint.Tracker
}

func (decoder *jsonDecoderV1) Read(req *http.Request) error {
	dec := json.NewDecoder(req.Body)
	for {
		var d JSONDatapointV1
		if err := dec.Decode(&d); err == io.EOF {
			break
		} else if err != nil {
			return err
		} else {
			log.WithField("dp", d).Debug("Got a new point")
			if d.Metric == "" {
				log.WithField("dp", d).Debug("Invalid Datapoint")
				continue
			}
			mt := decoder.typeGetter.GetMetricTypeFromMap(d.Metric)
			dp := datapoint.NewRelativeTime(d.Metric, map[string]string{"sf_source": d.Source}, datapoint.NewFloatValue(d.Value), mt, 0)
			decoder.datapointTracker.AddDatapoint(dp)
		}
	}
	return nil
}

type protobufDecoderV2 struct {
	datapointTracker datapoint.Tracker
}

var errInvalidContentLength = fmt.Errorf("Invalid Content Length")

func (decoder *protobufDecoderV2) Read(req *http.Request) error {
	return DecodeProtobufV2(req, &decoder.datapointTracker)
}

// DecodeProtobufV2 will take a request of signalfx's V2 protocol buffers and forward them to datapointTracker
func DecodeProtobufV2(req *http.Request, datapointTracker datapoint.Adder) error {
	if req.ContentLength == -1 {
		return errInvalidContentLength
	}

	// TODO: Source of memory creation.  Maybe pass buf in?
	buf := make([]byte, req.ContentLength)
	readLen, err := io.ReadFull(req.Body, buf)
	if err != nil {
		log.WithField("err", err).WithField("len", readLen).WithField("content-len", req.ContentLength).Warn("Unable to fully read from buffer")
		return err
	}
	var msg com_signalfuse_metrics_protobuf.DataPointUploadMessage
	err = proto.Unmarshal(buf, &msg)
	if err != nil {
		return err
	}
	for _, protoDb := range msg.GetDatapoints() {
		dp := NewProtobufDataPointWithType(protoDb, com_signalfuse_metrics_protobuf.MetricType_GAUGE)
		datapointTracker.AddDatapoint(dp)
	}
	return nil
}

type jsonDecoderV2 struct {
	datapointTracker datapoint.Tracker
{'fraction_non_alphanumeric': 0.11588054238416105, 'fraction_numerical': 0.005989518342899925, 'mean_word_length': 4.692234848484849, 'pattern_counts': {'":': 6, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3731317', 'n_tokens_mistral': 4349, 'n_tokens_neox': 4074, 'n_words': 924}