RLP gateway logging, error channel, configurable retry. change es channel size to 10000 (#239)

Author: rockb1017

.circleci/ci_nozzle_manifest.yml

@@ -7,12 +7,12 @@ applications:
     cmd: splunk-firehose-nozzle
     env:
       GOPACKAGENAME: main
-      API_ENDPOINT:
-      CLIENT_ID:
-      CLIENT_SECRET:
-      SPLUNK_HOST:
-      SPLUNK_TOKEN:
-      SPLUNK_INDEX:
+      API_ENDPOINT: 
+      CLIENT_ID: 
+      CLIENT_SECRET: 
+      SPLUNK_HOST: 
+      SPLUNK_TOKEN: 
+      SPLUNK_INDEX: main
       SKIP_SSL_VALIDATION_CF: true
       SKIP_SSL_VALIDATION_SPLUNK: true
       JOB_NAME: splunk-nozzle

.circleci/config.yml

@@ -141,11 +141,6 @@ workflows:
       - deploy-nozzle:
           requires:
             - build
-          filters:
-            branches:
-              only:
-                - develop
-                - master
 #      - tile-builder:
 #          requires:
 #            - deploy-nozzle
@@ -155,11 +150,7 @@ workflows:
       - execute_tests:
           requires:
             - deploy-nozzle
-          filters:
-            branches:
-              only:
-                - develop
-                - master
+
 
 #      - execute_perf_tests:
 #          requires:

.gitignore

@@ -18,3 +18,5 @@ testing/integration/venv/
 testing/integration/config/local.ini
 *.pyc
 testing/integration/config/env.json
+testing/manual_perf/data_gen_manifest.yml
+

cache/boltdb.go

@@ -78,13 +78,13 @@ func (c *Boltdb) Open() error {
 	// Open bolt db
 	db, err := bolt.Open(c.config.Path, 0600, &bolt.Options{Timeout: 5 * time.Second})
 	if err != nil {
-		c.config.Logger.Error("Fail to open boltdb: ", err)
+		c.config.Logger.Error("Failed to open boltdb: ", err)
 		return err
 	}
 	c.appdb = db
 
 	if err := c.createBucket(); err != nil {
-		c.config.Logger.Error("Fail to create bucket: ", err)
+		c.config.Logger.Error("Failed to create bucket: ", err)
 		return err
 	}
 

eventsource/firehose.go

@@ -1,18 +1,24 @@
 package eventsource
 
 import (
-	"code.cloudfoundry.org/go-loggregator"
+	"code.cloudfoundry.org/go-loggregator/v8"
+	"code.cloudfoundry.org/lager"
 	"github.com/cloudfoundry/sonde-go/events"
+	"log"
 	"net/http"
 	"time"
 )
 
 // FirehoseConfig struct with 4 fields of different types.
 type FirehoseConfig struct {
-	KeepAlive      time.Duration
-	SkipSSL        bool
-	Endpoint       string
-	SubscriptionID string
+	KeepAlive          time.Duration
+	SkipSSL            bool
+	Endpoint           string
+	SubscriptionID     string
+	GatewayErrChanAddr *chan error
+	GatewayLoggerAddr  *log.Logger
+	GatewayMaxRetries  int
+	Logger             lager.Logger
 }
 
 // Doer is used to make HTTP requests to the RLP Gateway.
@@ -32,6 +38,9 @@ func NewFirehose(tokenClient doer, config *FirehoseConfig) *Firehose {
 	c := loggregator.NewRLPGatewayClient(
 		config.Endpoint,
 		loggregator.WithRLPGatewayHTTPClient(tokenClient),
+		loggregator.WithRLPGatewayClientLogger(config.GatewayLoggerAddr),
+		loggregator.WithRLPGatewayErrChan(*config.GatewayErrChanAddr),
+		loggregator.WithRLPGatewayMaxRetries(config.GatewayMaxRetries),
 	)
 
 	f := &Firehose{
@@ -55,5 +64,5 @@ func (f *Firehose) Close() error {
 
 // Read reads envelope stream
 func (f *Firehose) Read() <-chan *events.Envelope {
-	return f.v2.Firehose(f.config.SubscriptionID)
+	return f.v2.Firehose(f.config)
 }

eventsource/v2adapter.go

@@ -1,9 +1,9 @@
 package eventsource
 
 import (
-	"code.cloudfoundry.org/go-loggregator"
-	"code.cloudfoundry.org/go-loggregator/conversion"
-	"code.cloudfoundry.org/go-loggregator/rpc/loggregator_v2"
+	"code.cloudfoundry.org/go-loggregator/v8"
+	"code.cloudfoundry.org/go-loggregator/v8/conversion"
+	"code.cloudfoundry.org/go-loggregator/v8/rpc/loggregator_v2"
 	"context"
 	"github.com/cloudfoundry/sonde-go/events"
 )
@@ -27,11 +27,12 @@ func NewV2Adapter(s Streamer) V2Adapter {
 }
 
 // Firehose returns only selected event stream
-func (a V2Adapter) Firehose(subscriptionID string) chan *events.Envelope {
+func (a V2Adapter) Firehose(config *FirehoseConfig) chan *events.Envelope {
 	ctx := context.Background()
-
+	var v1msgs = make(chan *events.Envelope, 10000)
+	var v2msgs = make(chan *loggregator_v2.Envelope, 10000)
 	es := a.streamer.Stream(ctx, &loggregator_v2.EgressBatchRequest{
-		ShardId: subscriptionID,
+		ShardId: config.SubscriptionID,
 		Selectors: []*loggregator_v2.Selector{
 			{
 				Message: &loggregator_v2.Selector_Log{
@@ -61,17 +62,23 @@ func (a V2Adapter) Firehose(subscriptionID string) chan *events.Envelope {
 		},
 	})
 
-	var msgs = make(chan *events.Envelope, 100)
 	go func() {
 		for ctx.Err() == nil {
 			for _, e := range es() {
-				// ToV1 converts v2 envelopes down to v1 envelopes.
-				for _, v1e := range conversion.ToV1(e) {
-					msgs <- v1e
-				}
+				v2msgs <- e
+			}
+		}
+	}()
+
+	go func() {
+		for ctx.Err() == nil {
+			e := <-v2msgs
+			//// ToV1 converts v2 envelopes down to v1 envelopes.
+			for _, v1e := range conversion.ToV1(e) {
+				v1msgs <- v1e
 			}
 		}
 	}()
 
-	return msgs
+	return v1msgs
 }

eventsource/v2adapter_test.go

@@ -1,9 +1,9 @@
 package eventsource_test
 
 import (
-	"code.cloudfoundry.org/go-loggregator"
-	"code.cloudfoundry.org/go-loggregator/conversion"
-	"code.cloudfoundry.org/go-loggregator/rpc/loggregator_v2"
+	"code.cloudfoundry.org/go-loggregator/v8"
+	"code.cloudfoundry.org/go-loggregator/v8/conversion"
+	"code.cloudfoundry.org/go-loggregator/v8/rpc/loggregator_v2"
 	"context"
 
 	. "github.com/onsi/ginkgo"
@@ -29,9 +29,11 @@ var _ = Describe("V2adapter", func() {
 
 		stubStreamer := newStubStreamer()
 		stubStreamer.envs = []*loggregator_v2.Envelope{v2Env}
-
+		config := &eventsource.FirehoseConfig{
+			SubscriptionID: "test-subscription",
+		}
 		firehoseAdapter := eventsource.NewV2Adapter(stubStreamer)
-		messages := firehoseAdapter.Firehose("test-subscription")
+		messages := firehoseAdapter.Firehose(config)
 
 		expected := conversion.ToV1(v2Env)
 		Eventually(messages).Should(Receive(Equal(expected[0])))

eventwriter/splunk.go

@@ -2,15 +2,14 @@ package eventwriter
 
 import (
 	"bytes"
+	"code.cloudfoundry.org/cfhttp"
+	"code.cloudfoundry.org/lager"
 	"crypto/tls"
 	"encoding/json"
 	"errors"
 	"fmt"
 	"io/ioutil"
 	"net/http"
-
-	"code.cloudfoundry.org/cfhttp"
-	"code.cloudfoundry.org/lager"
 )
 
 type SplunkConfig struct {

glide.lock

@@ -31,8 +31,8 @@ func main() {
 		logger.Info("Apps are not being cached. When apps are not cached, the org and space caching TTL is ineffective")
 	}
 
-	splunkNozzle := splunknozzle.NewSplunkFirehoseNozzle(config)
-	err := splunkNozzle.Run(shutdownChan, logger)
+	splunkNozzle := splunknozzle.NewSplunkFirehoseNozzle(config, logger)
+	err := splunkNozzle.Run(shutdownChan)
 	if err != nil {
 		logger.Error("Failed to run splunk-firehose-nozzle", err)
 	}