Data monitor (#242)

* data reliability monitoring

Co-authored-by: Shubham Jain <[email protected]>

Author: rockb1017

Makefile

@@ -65,6 +65,8 @@ testall: test vet race cov
 test:
 	@go test ${PKGS}
 
+testv:
+	@go test -v ${PKGS}
 # Run "short" unit tests
 test-short:
 	@go test -short ${PKGS}

eventsink/splunk.go

@@ -6,9 +6,8 @@ import (
 	"strconv"
 	"strings"
 	"sync"
-	"time"
-
 	"sync/atomic"
+	"time"
 
 	"code.cloudfoundry.org/lager"
 	"github.com/cloudfoundry-community/splunk-firehose-nozzle/eventwriter"
@@ -18,25 +17,27 @@ import (
 const SPLUNK_HEC_FIELDS_SUPPORT_VERSION = "6.4"
 
 type SplunkConfig struct {
-	FlushInterval  time.Duration
-	QueueSize      int // consumer queue buffer size
-	BatchSize      int
-	Retries        int // No of retries to post events to HEC before dropping events
-	Hostname       string
-	Version        string
-	SubscriptionID string
-	ExtraFields    map[string]string
-	TraceLogging   bool
-	UUID           string
-	Logger         lager.Logger
+	FlushInterval         time.Duration
+	QueueSize             int // consumer queue buffer size
+	BatchSize             int
+	Retries               int // No of retries to post events to HEC before dropping events
+	Hostname              string
+	Version               string
+	SubscriptionID        string
+	ExtraFields           map[string]string
+	TraceLogging          bool
+	UUID                  string
+	Logger                lager.Logger
+	StatusMonitorInterval time.Duration
 }
 
 type Splunk struct {
-	writers    []eventwriter.Writer
-	config     *SplunkConfig
-	events     chan map[string]interface{}
-	wg         sync.WaitGroup
-	eventCount uint64
+	writers       []eventwriter.Writer
+	config        *SplunkConfig
+	events        chan map[string]interface{}
+	wg            sync.WaitGroup
+	eventCount    uint64
+	sentCountChan chan uint64
 
 	// cached IP
 	ip string
@@ -47,11 +48,12 @@ func NewSplunk(writers []eventwriter.Writer, config *SplunkConfig) *Splunk {
 	config.Hostname = hostname
 
 	return &Splunk{
-		writers:    writers,
-		config:     config,
-		events:     make(chan map[string]interface{}, config.QueueSize),
-		ip:         ip,
-		eventCount: 0,
+		writers:       writers,
+		config:        config,
+		events:        make(chan map[string]interface{}, config.QueueSize),
+		ip:            ip,
+		eventCount:    0,
+		sentCountChan: make(chan uint64, 100),
 	}
 }
 
@@ -74,7 +76,6 @@ func (s *Splunk) Write(fields map[string]interface{}, msg string) error {
 	if len(msg) > 0 {
 		fields["msg"] = msg
 	}
-
 	s.events <- fields
 	return nil
 }
@@ -121,10 +122,14 @@ func (s *Splunk) indexEvents(writer eventwriter.Writer, batch []map[string]inter
 	}
 	var err error
 	for i := 0; i < s.config.Retries; i++ {
-		err = writer.Write(batch)
+		err, sentCount := writer.Write(batch)
 		if err == nil {
+			if s.config.StatusMonitorInterval > time.Second*0 {
+				s.sentCountChan <- sentCount
+			}
 			return nil
 		}
+		// add length of batch to counter
 		s.config.Logger.Error("Unable to talk to Splunk", err, lager.Data{"Retry attempt": i + 1})
 		time.Sleep(getRetryInterval(i))
 	}
@@ -213,6 +218,35 @@ func (s *Splunk) Log(message lager.LogFormat) {
 	s.writers[len(s.writers)-1].Write(events)
 }
 
+func (s *Splunk) LogStatus() {
+	timer := time.NewTimer(s.config.StatusMonitorInterval)
+	var sent uint64 = 0
+	for {
+		select {
+		case <-timer.C:
+			percent := float64(len(s.events)) / float64(s.config.QueueSize) * 100.0
+			status := "low"
+			switch {
+			case percent > 99.9:
+				status = "too high"
+			case percent > 90:
+				status = "high"
+			case percent > 50:
+				status = "medium"
+			}
+			s.config.Logger.Info("Memory_Queue_Pressure", lager.Data{"events_in_consumer_queue": len(s.events), "percentage": int(percent), "status": status})
+			s.config.Logger.Info("Event_Count", lager.Data{"event_count_sent": sent})
+			sent = 0
+			timer.Reset(s.config.StatusMonitorInterval)
+		default:
+		}
+		select {
+		case sentCount := <-s.sentCountChan:
+			atomic.AddUint64(&sent, sentCount)
+		default:
+		}
+	}
+}
 func getRetryInterval(attempt int) time.Duration {
 	// algorithm taken from https://en.wikipedia.org/wiki/Exponential_backoff
 	timeInSec := 5 + (0.5 * (math.Exp2(float64(attempt)) - 1.0))

eventsource/firehose.go

@@ -11,14 +11,15 @@ import (
 
 // FirehoseConfig struct with 4 fields of different types.
 type FirehoseConfig struct {
-	KeepAlive          time.Duration
-	SkipSSL            bool
-	Endpoint           string
-	SubscriptionID     string
-	GatewayErrChanAddr *chan error
-	GatewayLoggerAddr  *log.Logger
-	GatewayMaxRetries  int
-	Logger             lager.Logger
+	KeepAlive             time.Duration
+	SkipSSL               bool
+	Endpoint              string
+	SubscriptionID        string
+	GatewayErrChanAddr    *chan error
+	GatewayLoggerAddr     *log.Logger
+	GatewayMaxRetries     int
+	StatusMonitorInterval time.Duration
+	Logger                lager.Logger
 }
 
 // Doer is used to make HTTP requests to the RLP Gateway.

eventsource/v2adapter.go

@@ -4,8 +4,11 @@ import (
 	"code.cloudfoundry.org/go-loggregator/v8"
 	"code.cloudfoundry.org/go-loggregator/v8/conversion"
 	"code.cloudfoundry.org/go-loggregator/v8/rpc/loggregator_v2"
+	"code.cloudfoundry.org/lager"
 	"context"
 	"github.com/cloudfoundry/sonde-go/events"
+	"sync/atomic"
+	"time"
 )
 
 // Streamer implements Stream which returns a new EnvelopeStream for the given context and request.
@@ -71,11 +74,35 @@ func (a V2Adapter) Firehose(config *FirehoseConfig) chan *events.Envelope {
 	}()
 
 	go func() {
-		for ctx.Err() == nil {
-			e := <-v2msgs
-			//// ToV1 converts v2 envelopes down to v1 envelopes.
-			for _, v1e := range conversion.ToV1(e) {
-				v1msgs <- v1e
+		var receivedCount uint64 = 0
+
+		if config.StatusMonitorInterval > time.Second*0 {
+			timer := time.NewTimer(config.StatusMonitorInterval)
+			for ctx.Err() == nil {
+				select {
+				case <-timer.C:
+					config.Logger.Info("Data_Flow_Monitoring", lager.Data{"events_pre_processing": len(v2msgs), "events_in_process": len(v1msgs)})
+					config.Logger.Info("Event_Count", lager.Data{"event_count_received": receivedCount})
+					timer.Reset(config.StatusMonitorInterval)
+					receivedCount = 0
+				default:
+				}
+				select {
+				case e := <-v2msgs:
+					atomic.AddUint64(&receivedCount, 1)
+					//// ToV1 converts v2 envelopes down to v1 envelopes.
+					for _, v1e := range conversion.ToV1(e) {
+						v1msgs <- v1e
+					}
+				default:
+				}
+			}
+		} else {
+			for ctx.Err() == nil {
+				e := <-v2msgs
+				for _, v1e := range conversion.ToV1(e) {
+					v1msgs <- v1e
+				}
 			}
 		}
 	}()

eventsource/v2adapter_test.go

@@ -4,6 +4,7 @@ import (
 	"code.cloudfoundry.org/go-loggregator/v8"
 	"code.cloudfoundry.org/go-loggregator/v8/conversion"
 	"code.cloudfoundry.org/go-loggregator/v8/rpc/loggregator_v2"
+	"code.cloudfoundry.org/lager"
 	"context"
 
 	. "github.com/onsi/ginkgo"
@@ -30,7 +31,9 @@ var _ = Describe("V2adapter", func() {
 		stubStreamer := newStubStreamer()
 		stubStreamer.envs = []*loggregator_v2.Envelope{v2Env}
 		config := &eventsource.FirehoseConfig{
-			SubscriptionID: "test-subscription",
+			SubscriptionID:        "test-subscription",
+			StatusMonitorInterval: time.Second * 10,
+			Logger:                lager.NewLogger("test"),
 		}
 		firehoseAdapter := eventsource.NewV2Adapter(stubStreamer)
 		messages := firehoseAdapter.Firehose(config)

eventwriter/splunk.go

@@ -2,14 +2,16 @@ package eventwriter
 
 import (
 	"bytes"
-	"code.cloudfoundry.org/cfhttp"
-	"code.cloudfoundry.org/lager"
 	"crypto/tls"
 	"encoding/json"
 	"errors"
 	"fmt"
 	"io/ioutil"
 	"net/http"
+	"sync/atomic"
+
+	"code.cloudfoundry.org/cfhttp"
+	"code.cloudfoundry.org/lager"
 )
 
 type SplunkConfig struct {
@@ -40,10 +42,10 @@ func NewSplunk(config *SplunkConfig) Writer {
 	}
 }
 
-func (s *splunkClient) Write(events []map[string]interface{}) error {
+func (s *splunkClient) Write(events []map[string]interface{}) (error, uint64) {
 	bodyBuffer := new(bytes.Buffer)
+	var count uint64 = 0
 	for i, event := range events {
-
 		if event["event"].(map[string]interface{})["info_splunk_index"] != nil {
 			event["index"] = event["event"].(map[string]interface{})["info_splunk_index"]
 		} else if s.config.Index != "" {
@@ -57,6 +59,7 @@ func (s *splunkClient) Write(events []map[string]interface{}) error {
 		eventJson, err := json.Marshal(event)
 		if err == nil {
 			bodyBuffer.Write(eventJson)
+			atomic.AddUint64(&count, 1)
 			if i < len(events)-1 {
 				bodyBuffer.Write([]byte("\n\n"))
 			}
@@ -70,7 +73,7 @@ func (s *splunkClient) Write(events []map[string]interface{}) error {
 	}
 	bodyBytes := bodyBuffer.Bytes()
 
-	return s.send(&bodyBytes)
+	return s.send(&bodyBytes), count
 }
 
 func (s *splunkClient) send(postBody *[]byte) error {
@@ -93,6 +96,5 @@ func (s *splunkClient) send(postBody *[]byte) error {
 		responseBody, _ := ioutil.ReadAll(resp.Body)
 		return errors.New(fmt.Sprintf("Non-ok response code [%d] from splunk: %s", resp.StatusCode, responseBody))
 	}
-
 	return nil
 }

eventwriter/splunk_test.go

@@ -65,7 +65,7 @@ var _ = Describe("Splunk", func() {
 
 			client := NewSplunk(config)
 			events := []map[string]interface{}{}
-			err := client.Write(events)
+			err, _ := client.Write(events)
 
 			Expect(err).To(BeNil())
 			Expect(capturedRequest).NotTo(BeNil())
@@ -79,7 +79,7 @@ var _ = Describe("Splunk", func() {
 		It("sets content type to json", func() {
 			client := NewSplunk(config)
 			events := []map[string]interface{}{}
-			err := client.Write(events)
+			err, _ := client.Write(events)
 
 			Expect(err).To(BeNil())
 			Expect(capturedRequest).NotTo(BeNil())
@@ -101,10 +101,11 @@ var _ = Describe("Splunk", func() {
 			}}
 
 			events := []map[string]interface{}{event1, event2, event3}
-			err := client.Write(events)
+			err, sentCount := client.Write(events)
 
 			Expect(err).To(BeNil())
 			Expect(capturedRequest).NotTo(BeNil())
+			Expect(sentCount).To(Equal(uint64(3)))
 
 			expectedPayload := strings.TrimSpace(`
 {"event":{"greeting":"hello world"}}
@@ -127,7 +128,7 @@ var _ = Describe("Splunk", func() {
 			}}
 
 			events := []map[string]interface{}{event1, event2}
-			err := client.Write(events)
+			err, _ := client.Write(events)
 
 			Expect(err).To(BeNil())
 			Expect(capturedRequest).NotTo(BeNil())
@@ -156,7 +157,7 @@ var _ = Describe("Splunk", func() {
 			}}
 
 			events := []map[string]interface{}{event1, event2}
-			err := client.Write(events)
+			err, _ := client.Write(events)
 
 			Expect(err).To(BeNil())
 			Expect(capturedRequest).NotTo(BeNil())
@@ -173,7 +174,7 @@ var _ = Describe("Splunk", func() {
 		It("Writes to correct endpoint", func() {
 			client := NewSplunk(config)
 			events := []map[string]interface{}{}
-			err := client.Write(events)
+			err, _ := client.Write(events)
 
 			Expect(err).To(BeNil())
 			Expect(capturedRequest.URL.Path).To(Equal("/services/collector"))
@@ -184,7 +185,7 @@ var _ = Describe("Splunk", func() {
 		config.Host = ":"
 		client := NewSplunk(config)
 		events := []map[string]interface{}{}
-		err := client.Write(events)
+		err, _ := client.Write(events)
 
 		Expect(err).NotTo(BeNil())
 		Expect(err.Error()).To(ContainSubstring("protocol"))
@@ -199,7 +200,7 @@ var _ = Describe("Splunk", func() {
 		config.Host = testServer.URL
 		client := NewSplunk(config)
 		events := []map[string]interface{}{}
-		err := client.Write(events)
+		err, _ := client.Write(events)
 
 		Expect(err).NotTo(BeNil())
 		Expect(err.Error()).To(ContainSubstring("500"))
@@ -209,7 +210,7 @@ var _ = Describe("Splunk", func() {
 		config.Host = "foo://example.com"
 		client := NewSplunk(config)
 		events := []map[string]interface{}{}
-		err := client.Write(events)
+		err, _ := client.Write(events)
 
 		Expect(err).NotTo(BeNil())
 		Expect(err.Error()).To(ContainSubstring("foo"))

eventwriter/writer.go

@@ -1,5 +1,5 @@
 package eventwriter
 
 type Writer interface {
-	Write([]map[string]interface{}) error
+	Write([]map[string]interface{}) (error, uint64)
 }