Release 1.2.4 (#308)

* added 'drop_warn_threshold' config in tile (#300)

* Upgrading the OAuth dependency to resolve the auth issue

* Added a troubleshooting item for the auth issue with special characters

* Drain the response buffer to reuse the connection with HEC

* Changed eventwriter to write events on stdout in debug mode

* Added a test for debug mode

* Added/updated testcases to increase coverage

* updated eventsink tests

* fixed race condition in eventsink test

* Updated the nozzle version set in HEC req header

* Corrected version in test

* Updated version in readme

Co-authored-by: harshit-splunk <[email protected]>

Author: kashyap-splunk

README.md

@@ -398,6 +398,10 @@ Each instance of the Splunk Firehose Nozzle will run with a randomly generated U
 index=main | stats count as total_events, min(nozzle-event-counter) as min_number, max(nozzle-event-counter) as max_number by uuid | eval event_number =  max_number - min_number | eval success_percentage = total_events/event_number*100 | stats max(success_percentage) by uuid
 </pre>
 
+### 6. Authentication is not working even if correct CF Client ID/secret is configured: (applicable in v1.2.3)
+
+Due to a known issue in an indirect dependency (an OAuth library), if the client secret has any special characters (eg. *!#$&@^) then it will not work. For now, user has to configure a client secret without any of this characters. Once the library in question is updated in the next release it will work even with the special characters.
+
 #### Searching Events
 
 Here are two short Splunk queries to start exploring some of the Cloud Foundry events in Splunk.
@@ -442,7 +446,7 @@ $ chmod +x tools/nozzle.sh
 Build project:
 
 ```
-$ make VERSION=1.2.3
+$ make VERSION=1.2.4
 ```
 
 Run tests with [Ginkgo](http://onsi.github.io/ginkgo/)

cache/cache_test.go

@@ -32,6 +32,7 @@ var _ = Describe("Cache", func() {
 			MissingAppCacheTTL: missingAppCacheTTL,
 			OrgSpaceCacheTTL:   orgSpaceCacheTTL,
 			Logger:             lager.NewLogger("test"),
+			AppLimits:          n,
 		}
 
 		client *testing.AppClientMock = nil
@@ -104,6 +105,22 @@ var _ = Describe("Cache", func() {
 		})
 	})
 
+	Context("When orphan app is requested", func() {
+
+		It("Should found app in cache", func() {
+			app_guid := "orphan_app_id"
+			client.CreateApp(app_guid, "orphan_space_id")
+			Ω(cache.GetApp(app_guid)).NotTo(Equal(nil))
+			client.DeleteApp(app_guid)
+			cache.ManuallyInvalidateCaches()
+
+			app, err := cache.GetApp(app_guid)
+			Ω(err).ShouldNot(HaveOccurred())
+			Expect(app).NotTo(Equal(nil))
+			Expect(app.Guid).To(Equal(app_guid))
+		})
+	})
+
 	Context("Cache invalidation", func() {
 		BeforeEach(func() {
 			// close the cache created in the outer BeforeEach

eventsink/splunk_test.go

@@ -1,6 +1,8 @@
 package eventsink_test
 
 import (
+	"fmt"
+	"os"
 	"strconv"
 	"time"
 
@@ -32,6 +34,7 @@ var _ = Describe("Splunk", func() {
 
 		memSink *testing.MemorySinkMock
 		sink    *eventsink.Splunk
+		config  *eventsink.SplunkConfig
 
 		event      map[string]interface{}
 		logger     lager.Logger
@@ -68,18 +71,63 @@ var _ = Describe("Splunk", func() {
 		mockClient2 = &testing.EventWriterMock{}
 
 		logger = lager.NewLogger("test")
-		config := &eventsink.SplunkConfig{
-			FlushInterval: time.Millisecond,
-			QueueSize:     1000,
-			BatchSize:     1,
-			Retries:       1,
-			Hostname:      "localhost",
-			ExtraFields:   map[string]string{"env": "dev", "test": "field"},
-			UUID:          "0a956421-f2e1-4215-9d88-d15633bb3023",
-			Logger:        logger,
+		config = &eventsink.SplunkConfig{
+			FlushInterval:     time.Millisecond,
+			QueueSize:         1000,
+			BatchSize:         1,
+			Retries:           1,
+			Hostname:          "localhost",
+			ExtraFields:       map[string]string{"env": "dev", "test": "field"},
+			UUID:              "0a956421-f2e1-4215-9d88-d15633bb3023",
+			Logger:            logger,
+			DropWarnThreshold: 1000,
 		}
 		sink = eventsink.NewSplunk([]eventwriter.Writer{mockClient, mockClient2}, config)
 	})
+	Context("When LogStatus is executed", func() {
+		BeforeEach(func() {
+			config.StatusMonitorInterval = time.Second * 1
+			flushInterval := time.Second * 2
+			config.FlushInterval = flushInterval
+			file, _ := os.OpenFile("lager.log", os.O_CREATE|os.O_RDWR, 0600)
+			loggerSink := lager.NewReconfigurableSink(lager.NewWriterSink(file, lager.DEBUG), lager.DEBUG)
+			myLogger := lager.NewLogger("LogStatus")
+			myLogger.RegisterSink(loggerSink)
+			config.Logger = myLogger
+			defer file.Close()
+			go sink.LogStatus()
+			// low pressure
+			for i := 0; i < int(float64(config.QueueSize)*0.12); i++ {
+				sink.Write(make(map[string]interface{}), fmt.Sprintf("event %d", i))
+			}
+			// medium pressure
+			time.Sleep(flushInterval)
+			for i := 0; i < int(float64(config.QueueSize)*0.40); i++ {
+				sink.Write(make(map[string]interface{}), fmt.Sprintf("event %d", i))
+			}
+			time.Sleep(flushInterval)
+			// high pressure
+			for i := 0; i < int(float64(config.QueueSize)*0.40); i++ {
+				sink.Write(make(map[string]interface{}), fmt.Sprintf("event %d", i))
+			}
+			time.Sleep(flushInterval)
+			// too high pressure
+			for i := 0; i < int(float64(config.QueueSize)*0.08); i++ {
+				sink.Write(make(map[string]interface{}), fmt.Sprintf("event %d", i))
+			}
+			time.Sleep(flushInterval)
+		})
+
+		It("tests pressure status", func() {
+			data, _ := os.ReadFile("lager.log")
+			log := string(data)
+			Expect(log).Should(ContainSubstring("status\":\"too high"))
+			Expect(log).Should(ContainSubstring("status\":\"high"))
+			Expect(log).Should(ContainSubstring("status\":\"medium"))
+			Expect(log).Should(ContainSubstring("status\":\"low"))
+			os.Remove("lager.log")
+		})
+	})
 
 	It("sends events to client", func() {
 		eventType = events.Envelope_Error

eventwriter/splunk.go

@@ -6,6 +6,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
 	"io/ioutil"
 	"net/http"
 
@@ -19,6 +20,7 @@ type SplunkConfig struct {
 	Index   string
 	Fields  map[string]string
 	SkipSSL bool
+	Debug   bool
 
 	Logger lager.Logger
 }
@@ -70,9 +72,14 @@ func (s *splunkClient) Write(events []map[string]interface{}) (error, uint64) {
 			)
 		}
 	}
-	bodyBytes := bodyBuffer.Bytes()
 
-	return s.send(&bodyBytes), count
+	if s.config.Debug {
+		bodyString := bodyBuffer.String()
+		return s.dump(bodyString), count
+	} else {
+		bodyBytes := bodyBuffer.Bytes()
+		return s.send(&bodyBytes), count
+	}
 }
 
 func (s *splunkClient) send(postBody *[]byte) error {
@@ -87,7 +94,7 @@ func (s *splunkClient) send(postBody *[]byte) error {
 	//Add app headers for HEC telemetry
 	//Todo: update static values with appName and appVersion variables
 	req.Header.Set("__splunk_app_name", "Splunk Firehose Nozzle")
-	req.Header.Set("__splunk_app_version", "1.2.2")
+	req.Header.Set("__splunk_app_version", "1.2.4")
 
 	resp, err := s.httpClient.Do(req)
 	if err != nil {
@@ -98,7 +105,17 @@ func (s *splunkClient) send(postBody *[]byte) error {
 	if resp.StatusCode > 299 {
 		responseBody, _ := ioutil.ReadAll(resp.Body)
 		return errors.New(fmt.Sprintf("Non-ok response code [%d] from splunk: %s", resp.StatusCode, responseBody))
+	} else {
+		//Draining the response buffer, so that the same connection can be reused the next time
+		io.Copy(ioutil.Discard, resp.Body)
 	}
 
 	return nil
 }
+
+//To dump the event on stdout instead of Splunk, in case of 'debug' mode
+func (s *splunkClient) dump(eventString string) error {
+	fmt.Println(string(eventString))
+
+	return nil
+}

eventwriter/splunk_test.go

@@ -104,7 +104,7 @@ var _ = Describe("Splunk", func() {
 		})
 
 		It("sets app appVersion", func() {
-			appVersion := "1.2.2"
+			appVersion := "1.2.4"
 
 			client := NewSplunk(config)
 			events := []map[string]interface{}{}
@@ -209,6 +209,15 @@ var _ = Describe("Splunk", func() {
 			Expect(err).To(BeNil())
 			Expect(capturedRequest.URL.Path).To(Equal("/services/collector"))
 		})
+
+		It("Writes to stdout in debug without error", func() {
+			config.Debug = true
+			client := NewSplunk(config)
+			events := []map[string]interface{}{}
+			err, _ := client.Write(events)
+
+			Expect(err).To(BeNil())
+		})
 	})
 
 	It("returns error on bad splunk host", func() {

go.mod

@@ -11,7 +11,7 @@ require (
 	github.com/cloudfoundry/sonde-go v0.0.0-20160804000546-81c3f6be579c
 	github.com/gogo/protobuf v1.3.2
 	github.com/google/uuid v0.0.0-20170728174318-281f560d28af
-	github.com/gorilla/websocket v1.0.1-0.20160802133203-a69d25be2fe2
+	github.com/gorilla/websocket v1.4.2
 	github.com/mailru/easyjson v0.0.0-20160816214844-3d7eb5818bd5
 	github.com/onsi/ginkgo v1.6.0
 	github.com/onsi/gomega v1.4.3
@@ -32,18 +32,19 @@ require (
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/elazarl/goproxy v0.0.0-20210801061803-8e322dfb79c4 // indirect
 	github.com/elazarl/goproxy/ext v0.0.0-20210801061803-8e322dfb79c4 // indirect
-	github.com/golang/protobuf v1.3.2 // indirect
+	github.com/golang/protobuf v1.4.2 // indirect
 	github.com/hpcloud/tail v1.0.0 // indirect
 	github.com/mattn/go-isatty v0.0.14 // indirect
 	github.com/nu7hatch/gouuid v0.0.0-20131221200532-179d4d0c4d8d // indirect
 	github.com/pkg/errors v0.8.1 // indirect
 	github.com/sergi/go-diff v1.2.0 // indirect
 	github.com/stretchr/testify v1.7.0 // indirect
 	golang.org/x/net v0.0.0-20210520170846-37e1c6afe023 // indirect
-	golang.org/x/oauth2 v0.0.0-20190130055435-99b60b757ec1 // indirect
+	golang.org/x/oauth2 v0.0.0-20211005180243-6b3c2da341f1 // indirect
 	golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c // indirect
 	golang.org/x/text v0.3.6 // indirect
-	google.golang.org/appengine v1.4.0 // indirect
+	google.golang.org/appengine v1.6.6 // indirect
+	google.golang.org/protobuf v1.25.0 // indirect
 	gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 // indirect
 	gopkg.in/fsnotify.v1 v1.4.7 // indirect
 	gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 // indirect