Improve efficiency in event source monitoring and AddAppInfo param clean up (#254)

* clean event source and restructure AddAppInfo param

Author: rockb1017

.circleci/ci_nozzle_manifest.yml

@@ -18,7 +18,7 @@ applications:
       JOB_NAME: splunk-nozzle
       JOB_INDEX: -1
       JOB_HOST: localhost
-      ADD_APP_INFO: true
+      ADD_APP_INFO: AppName,OrgName,OrgGuid,SpaceName,SpaceGuid
       IGNORE_MISSING_APP: true
       MISSING_APP_CACHE_INVALIDATE_TTL: 3600s
       APP_CACHE_INVALIDATE_TTL: 86440s

events/events.go

@@ -20,13 +20,21 @@ type Event struct {
 
 type Config struct {
 	SelectedEvents string
-	AddAppName   bool
+	AddAppName     bool
 	AddOrgName     bool
 	AddOrgGuid     bool
 	AddSpaceName   bool
 	AddSpaceGuid   bool
 }
 
+var AppMetadata = []string{
+	"AppName",
+	"OrgName",
+	"OrgGuid",
+	"SpaceName",
+	"SpaceGuid",
+}
+
 func HttpStartStop(msg *events.Envelope) *Event {
 	httpStartStop := msg.GetHttpStartStop()
 
@@ -258,3 +266,7 @@ func ParseExtraFields(extraEventsString string) (map[string]string, error) {
 	}
 	return extraEvents, nil
 }
+
+func AuthorizedMetadata() string {
+	return strings.Join(AppMetadata, ", ")
+}

eventsource/v2adapter.go

@@ -33,7 +33,7 @@ func NewV2Adapter(s Streamer) V2Adapter {
 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)
+	var v2msgs = make(chan []*loggregator_v2.Envelope, 10000)
 	es := a.streamer.Stream(ctx, &loggregator_v2.EgressBatchRequest{
 		ShardId: config.SubscriptionID,
 		Selectors: []*loggregator_v2.Selector{
@@ -67,9 +67,8 @@ func (a V2Adapter) Firehose(config *FirehoseConfig) chan *events.Envelope {
 
 	go func() {
 		for ctx.Err() == nil {
-			for _, e := range es() {
-				v2msgs <- e
-			}
+			e := es()
+			v2msgs <- e
 		}
 	}()
 
@@ -81,27 +80,30 @@ func (a V2Adapter) Firehose(config *FirehoseConfig) chan *events.Envelope {
 			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
+				case eArray := <-v2msgs:
+					atomic.AddUint64(&receivedCount, uint64(len(eArray)))
+					for _, e := range eArray {
+						//// 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
+				eArray := <-v2msgs
+				for _, e := range eArray {
+					for _, v1e := range conversion.ToV1(e) {
+						v1msgs <- v1e
+					}
 				}
 			}
 		}

splunknozzle/config.go

@@ -31,20 +31,13 @@ type Config struct {
 	SubscriptionID string        `json:"subscription-id"`
 	KeepAlive      time.Duration `json:"keep-alive"`
 
-	AddAppInfo         bool          `json:"add-app-info"`
+	AddAppInfo         string        `json:"add-app-info"`
 	IgnoreMissingApps  bool          `json:"ignore-missing-apps"`
 	MissingAppCacheTTL time.Duration `json:"missing-app-cache-ttl"`
 	AppCacheTTL        time.Duration `json:"app-cache-ttl"`
 	OrgSpaceCacheTTL   time.Duration `json:"org-space-cache-ttl"`
 	AppLimits          int           `json:"app-limits"`
 
-	// Add configuration to select interested fields from app info.
-	AddAppName   bool `json:"add-app-name"`
-	AddOrgName   bool `json:"add-org-name"`
-	AddOrgGuid   bool `json:"add-org-guid"`
-	AddSpaceName bool `json:"add-space-name"`
-	AddSpaceGuid bool `json:"add-space-guid"`
-
 	BoltDBPath   string `json:"boltdb-path"`
 	WantedEvents string `json:"wanted-events"`
 	ExtraFields  string `json:"extra-fields"`
@@ -111,18 +104,8 @@ func NewConfigFromCmdFlags(version, branch, commit, buildos string) *Config {
 	kingpin.Flag("firehose-keep-alive", "Keep Alive duration for the firehose consumer").
 		OverrideDefaultFromEnvar("FIREHOSE_KEEP_ALIVE").Default("25s").DurationVar(&c.KeepAlive)
 
-	kingpin.Flag("add-app-info", "Query API to fetch app details").
-		OverrideDefaultFromEnvar("ADD_APP_INFO").Default("false").BoolVar(&c.AddAppInfo)
-	kingpin.Flag("add-app-name", "Add app name from app cache").
-		OverrideDefaultFromEnvar("ADD_APP_NAME").Default("true").BoolVar(&c.AddAppName)
-	kingpin.Flag("add-org-name", "Add org name from app cache").
-		OverrideDefaultFromEnvar("ADD_ORG_NAME").Default("true").BoolVar(&c.AddOrgName)
-	kingpin.Flag("add-org-guid", "Add org guid from app cache").
-		OverrideDefaultFromEnvar("ADD_ORG_GUID").Default("true").BoolVar(&c.AddOrgGuid)
-	kingpin.Flag("add-space-name", "Add space name from app cache").
-		OverrideDefaultFromEnvar("ADD_SPACE_NAME").Default("true").BoolVar(&c.AddSpaceName)
-	kingpin.Flag("add-space-guid", "Add space guid from app cache").
-		OverrideDefaultFromEnvar("ADD_SPACE_GUID").Default("true").BoolVar(&c.AddSpaceGuid)
+	kingpin.Flag("add-app-info", fmt.Sprintf("Comma separated list of app metadata to enrich event. Valid options are %s", events.AuthorizedMetadata())).
+		OverrideDefaultFromEnvar("ADD_APP_INFO").Default("").StringVar(&c.AddAppInfo)
 
 	kingpin.Flag("ignore-missing-app", "If app is missing, stop repeatedly querying app info from PCF").
 		OverrideDefaultFromEnvar("IGNORE_MISSING_APP").Default("true").BoolVar(&c.IgnoreMissingApps)

splunknozzle/config_test.go

@@ -47,7 +47,7 @@ var _ = Describe("Config", func() {
 			os.Setenv("FIREHOSE_SUBSCRIPTION_ID", "my-nozzle")
 			os.Setenv("FIREHOSE_KEEP_ALIVE", "42s")
 
-			os.Setenv("ADD_APP_INFO", "true")
+			os.Setenv("ADD_APP_INFO", "AppName")
 			os.Setenv("IGNORE_MISSING_APP", "true")
 			os.Setenv("MISSING_APP_CACHE_INVALIDATE_TTL", "100s")
 			os.Setenv("APP_CACHE_INVALIDATE_TTL", "100s")
@@ -89,7 +89,7 @@ var _ = Describe("Config", func() {
 			Expect(c.SubscriptionID).To(Equal("my-nozzle"))
 			Expect(c.KeepAlive).To(Equal(42 * time.Second))
 
-			Expect(c.AddAppInfo).To(BeTrue())
+			Expect(c.AddAppInfo).To(Equal("AppName"))
 			Expect(c.IgnoreMissingApps).To(BeTrue())
 			Expect(c.MissingAppCacheTTL).To(Equal(100 * time.Second))
 			Expect(c.AppCacheTTL).To(Equal(100 * time.Second))
@@ -128,7 +128,7 @@ var _ = Describe("Config", func() {
 			Expect(c.SubscriptionID).To(Equal("splunk-firehose"))
 			Expect(c.KeepAlive).To(Equal(25 * time.Second))
 
-			Expect(c.AddAppInfo).To(BeFalse())
+			Expect(c.AddAppInfo).To(Equal(""))
 			Expect(c.IgnoreMissingApps).To(BeTrue())
 			Expect(c.MissingAppCacheTTL).To(Equal(0 * time.Second))
 			Expect(c.AppCacheTTL).To(Equal(0 * time.Second))
@@ -179,7 +179,7 @@ var _ = Describe("Config", func() {
 				"--skip-ssl-validation-splunk",
 				"--subscription-id=my-nozzlec",
 				"--firehose-keep-alive=24s",
-				"--add-app-info",
+				"--add-app-info=OrgName",
 				"--ignore-missing-app",
 				"--missing-app-cache-invalidate-ttl=15s",
 				"--app-cache-invalidate-ttl=15s",
@@ -222,7 +222,7 @@ var _ = Describe("Config", func() {
 			Expect(c.SubscriptionID).To(Equal("my-nozzlec"))
 			Expect(c.KeepAlive).To(Equal(24 * time.Second))
 
-			Expect(c.AddAppInfo).To(BeTrue())
+			Expect(c.AddAppInfo).To(Equal("OrgName"))
 			Expect(c.IgnoreMissingApps).To(BeTrue())
 			Expect(c.MissingAppCacheTTL).To(Equal(15 * time.Second))
 			Expect(c.AppCacheTTL).To(Equal(15 * time.Second))

splunknozzle/nozzle.go

@@ -39,13 +39,14 @@ func NewSplunkFirehoseNozzle(config *Config, logger lager.Logger) *SplunkFirehos
 
 // EventRouter creates EventRouter object and setup routes for interested events
 func (s *SplunkFirehoseNozzle) EventRouter(cache cache.Cache, eventSink eventsink.Sink) (eventrouter.Router, error) {
+	LowerAddAppInfo := strings.ToLower(s.config.AddAppInfo)
 	config := &eventrouter.Config{
 		SelectedEvents: s.config.WantedEvents,
-		AddAppName:     s.config.AddAppName,
-		AddOrgName:     s.config.AddOrgName,
-		AddOrgGuid:     s.config.AddOrgGuid,
-		AddSpaceName:   s.config.AddSpaceName,
-		AddSpaceGuid:   s.config.AddSpaceGuid,
+		AddAppName:     strings.Contains(LowerAddAppInfo, "appname"),
+		AddOrgName:     strings.Contains(LowerAddAppInfo, "orgname"),
+		AddOrgGuid:     strings.Contains(LowerAddAppInfo, "orgguid"),
+		AddSpaceName:   strings.Contains(LowerAddAppInfo, "spacename"),
+		AddSpaceGuid:   strings.Contains(LowerAddAppInfo, "spaceguid"),
 	}
 	return eventrouter.New(cache, eventSink, config)
 }
@@ -66,7 +67,7 @@ func (s *SplunkFirehoseNozzle) PCFClient() (*cfclient.Client, error) {
 
 // AppCache creates in-memory cache or boltDB cache
 func (s *SplunkFirehoseNozzle) AppCache(client cache.AppClient) (cache.Cache, error) {
-	if s.config.AddAppInfo {
+	if s.config.AddAppInfo != "" {
 		c := cache.BoltdbConfig{
 			Path:               s.config.BoltDBPath,
 			IgnoreMissingApps:  s.config.IgnoreMissingApps,

splunknozzle/nozzle_test.go

@@ -34,7 +34,7 @@ func newConfig() *Config {
 		SubscriptionID: "splunk-sub",
 		KeepAlive:      time.Second * 25,
 
-		AddAppInfo:         true,
+		AddAppInfo:         "AppName,OrgName,OrgGuid,SpaceName,SpaceGuid",
 		IgnoreMissingApps:  true,
 		MissingAppCacheTTL: time.Second * 30,
 		AppCacheTTL:        time.Second * 30,
@@ -106,7 +106,7 @@ var _ = Describe("SplunkFirehoseNozzle", func() {
 		_, err := noz.AppCache(client)
 		Ω(err).ShouldNot(HaveOccurred())
 
-		config.AddAppInfo = false
+		config.AddAppInfo = ""
 		_, err = noz.AppCache(client)
 		Ω(err).ShouldNot(HaveOccurred())
 	})
@@ -146,7 +146,7 @@ var _ = Describe("SplunkFirehoseNozzle", func() {
 	})
 
 	It("Run with cloudcontroller", func() {
-		config.AddAppInfo = false
+		config.AddAppInfo = ""
 		port := 9911
 		cc := testing.NewCloudControllerMock(port)
 		started := make(chan struct{})