Merged in feature/SCFF-21 (pull request #3)

Feature/SCFF-21

Author: sklava

eventRouting/eventRouting_test.go

@@ -4,6 +4,7 @@ import (
 	. "bitbucket.org/mcplusa-ondemand/firehouse-to-sumologic/caching/cachingfakes"
 	. "bitbucket.org/mcplusa-ondemand/firehouse-to-sumologic/eventRouting"
 	. "bitbucket.org/mcplusa-ondemand/firehouse-to-sumologic/logging/loggingfakes"
+	. "bitbucket.org/mcplusa-ondemand/firehouse-to-sumologic/sumoCFFirehose/sumoLog4gofakes" //**
 	. "github.com/cloudfoundry/sonde-go/events"
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
@@ -16,7 +17,8 @@ var _ = Describe("Events", func() {
 	BeforeEach(func() {
 		logging := new(FakeLogging)
 		caching := new(FakeCaching)
-		eventRouting = NewEventRouting(caching, logging)
+		sLAppender := new(FakeSumoLog4go)
+		eventRouting = NewEventRouting(caching, logging, sLAppender)
 		eventRouting.SetupEventRouting("")
 
 	})

eventRouting/eventrouting.go

@@ -2,33 +2,37 @@ package eventRouting
 
 import (
 	"fmt"
-	"github.com/Sirupsen/logrus"
-	"bitbucket.org/mcplusa-ondemand/firehouse-to-sumologic/caching"
-	fevents "bitbucket.org/mcplusa-ondemand/firehouse-to-sumologic/events"
-	"bitbucket.org/mcplusa-ondemand/firehouse-to-sumologic/extrafields"
-	"bitbucket.org/mcplusa-ondemand/firehouse-to-sumologic/logging"
-	"github.com/cloudfoundry/sonde-go/events"
 	"os"
 	"sort"
 	"strings"
 	"sync"
 	"time"
+
+	"bitbucket.org/mcplusa-ondemand/firehouse-to-sumologic/caching"
+	fevents "bitbucket.org/mcplusa-ondemand/firehouse-to-sumologic/events"
+	"bitbucket.org/mcplusa-ondemand/firehouse-to-sumologic/extrafields"
+	"bitbucket.org/mcplusa-ondemand/firehouse-to-sumologic/logging"
+	"bitbucket.org/mcplusa-ondemand/firehouse-to-sumologic/sumoCFFirehose" //**
+	"github.com/Sirupsen/logrus"
+	"github.com/cloudfoundry/sonde-go/events"
 )
 
 type EventRouting struct {
 	CachingClient       caching.Caching
 	selectedEvents      map[string]bool
 	selectedEventsCount map[string]uint64
 	mutex               *sync.Mutex
+	sLAppender          sumoCFFirehose.SumoCFFirehose //**
 	log                 logging.Logging
 	ExtraFields         map[string]string
 }
 
-func NewEventRouting(caching caching.Caching, logging logging.Logging) *EventRouting {
+func NewEventRouting(caching caching.Caching, logging logging.Logging, sLAppender sumoCFFirehose.SumoCFFirehose) *EventRouting {
 	return &EventRouting{
 		CachingClient:       caching,
 		selectedEvents:      make(map[string]bool),
 		selectedEventsCount: make(map[string]uint64),
+		sLAppender:          sLAppender, //**
 		log:                 logging,
 		mutex:               &sync.Mutex{},
 		ExtraFields:         make(map[string]string),
@@ -76,7 +80,8 @@ func (e *EventRouting) RouteEvent(msg *events.Envelope) {
 		if ignored, hasIgnoredField := event.Fields["cf_ignored_app"]; ignored == true && hasIgnoredField {
 			e.selectedEventsCount["ignored_app_message"]++
 		} else {
-			e.log.ShipEvents(event.Fields, event.Msg)
+			e.sLAppender.AppendLogs(event.Fields)     //**
+			e.log.ShipEvents(event.Fields, event.Msg) // here we have to change the method for the one on sumoLogicAppender
 			e.selectedEventsCount[eventType.String()]++
 
 		}

events/events.go

@@ -2,9 +2,10 @@ package events
 
 import (
 	"fmt"
-	"github.com/Sirupsen/logrus"
+
 	"bitbucket.org/mcplusa-ondemand/firehouse-to-sumologic/caching"
 	"bitbucket.org/mcplusa-ondemand/firehouse-to-sumologic/utils"
+	"github.com/Sirupsen/logrus"
 	"github.com/cloudfoundry/sonde-go/events"
 )
 

firehoseclient/firehoseclient.go

@@ -2,13 +2,14 @@ package firehoseclient
 
 import (
 	"crypto/tls"
+	"time"
+
 	"bitbucket.org/mcplusa-ondemand/firehouse-to-sumologic/eventRouting"
 	"bitbucket.org/mcplusa-ondemand/firehouse-to-sumologic/logging"
 	"github.com/cloudfoundry-community/go-cfclient"
 	"github.com/cloudfoundry/noaa/consumer"
 	"github.com/cloudfoundry/sonde-go/events"
 	"github.com/gorilla/websocket"
-	"time"
 )
 
 type FirehoseNozzle struct {

main.go

@@ -4,10 +4,12 @@ import (
 	"fmt"
 	"log"
 	"os"
+
 	"bitbucket.org/mcplusa-ondemand/firehouse-to-sumologic/caching"
 	"bitbucket.org/mcplusa-ondemand/firehouse-to-sumologic/eventRouting"
 	"bitbucket.org/mcplusa-ondemand/firehouse-to-sumologic/firehoseclient"
 	"bitbucket.org/mcplusa-ondemand/firehouse-to-sumologic/logging"
+	"bitbucket.org/mcplusa-ondemand/firehouse-to-sumologic/sumoCFFirehose"
 	"github.com/cloudfoundry-community/go-cfclient"
 	"github.com/pkg/profile"
 	"gopkg.in/alecthomas/kingpin.v2"
@@ -17,8 +19,8 @@ var (
 	debug              = kingpin.Flag("debug", "Enable debug mode. This disables forwarding to syslog").Default("false").OverrideDefaultFromEnvar("DEBUG").Bool()
 	apiEndpoint        = "https://api.bosh-lite.com" //kingpin.Flag("api-endpoint", "Api endpoint address. For bosh-lite installation of CF: https://api.10.244.0.34.xip.io").OverrideDefaultFromEnvar("API_ENDPOINT").Required().String()
 	dopplerEndpoint    = kingpin.Flag("doppler-endpoint", "Overwrite default doppler endpoint return by /v2/info").OverrideDefaultFromEnvar("DOPPLER_ENDPOINT").String()
-	syslogServer       = "192.168.33.10:514"//kingpin.Flag("syslog-server", "Syslog server.").OverrideDefaultFromEnvar("SYSLOG_ENDPOINT").String()
-	syslogProtocol     = "udp"//kingpin.Flag("syslog-protocol", "Syslog protocol (tcp/udp).").Default("tcp").OverrideDefaultFromEnvar("SYSLOG_PROTOCOL").String()
+	syslogServer       = "192.168.33.10:514" //kingpin.Flag("syslog-server", "Syslog server.").OverrideDefaultFromEnvar("SYSLOG_ENDPOINT").String()
+	syslogProtocol     = "udp"               //kingpin.Flag("syslog-protocol", "Syslog protocol (tcp/udp).").Default("tcp").OverrideDefaultFromEnvar("SYSLOG_PROTOCOL").String()
 	subscriptionId     = kingpin.Flag("subscription-id", "Id for the subscription.").Default("firehose").OverrideDefaultFromEnvar("FIREHOSE_SUBSCRIPTION_ID").String()
 	user               = kingpin.Flag("user", "Admin user.").Default("admin").OverrideDefaultFromEnvar("FIREHOSE_USER").String()
 	password           = kingpin.Flag("password", "Admin password.").Default("admin").OverrideDefaultFromEnvar("FIREHOSE_PASSWORD").String()
@@ -43,14 +45,12 @@ func main() {
 	kingpin.Version(version)
 	kingpin.Parse()
 
-	//Setup Logging
+	//Setup Logging <-- this loggingClient has to be removed when sumoLog4go Library is working
 	loggingClient := logging.NewLogging(syslogServer, syslogProtocol, *logFormatterType, *debug)
-	//loggingClient, err := syslog.Dial(syslogProtocol, syslogServer, syslog.LOG_ERR, "demotag") 
-         //defer loggingClient.Close()
-         //if err != nil {
-         //        log.Fatal("error")
-         //}
-	logging.LogStd(fmt.Sprintf("Starting firehose-to-syslog %s ", version), true)
+	//Setup Loggin with sumoLog4go library
+	loggingClientSumo := sumoCFFirehose.NewSumoLogicAppender("http://httpbin.org/post", 1000)
+
+	logging.LogStd(fmt.Sprintf("Starting firehose-to-sumo %s ", version), true)
 
 	if *modeProf != "" {
 		switch *modeProf {
@@ -77,12 +77,6 @@ func main() {
 		cfClient.Endpoint.DopplerEndpoint = *dopplerEndpoint
 	}
 
-	logging.LogStd(fmt.Sprintf("Login with '%s' user", *user), true)
-  	logging.LogStd(fmt.Sprintf("using '%s' as user", c.Username), true)
-  	logging.LogStd(fmt.Sprintf("using '%s' as password", *password), true)
-
-	logging.LogStd(fmt.Sprintf("Using %s as doppler endpoint", cfClient.Endpoint.DopplerEndpoint), true)
-
 	//Creating Caching
 	var cachingClient caching.Caching
 	if caching.IsNeeded(*wantedEvents) {
@@ -91,7 +85,7 @@ func main() {
 		cachingClient = caching.NewCachingEmpty()
 	}
 	//Creating Events
-	events := eventRouting.NewEventRouting(cachingClient, loggingClient)
+	events := eventRouting.NewEventRouting(cachingClient, loggingClient, loggingClientSumo)
 	err := events.SetupEventRouting(*wantedEvents)
 	if err != nil {
 		log.Fatal("Error setting up event routing: ", err)
@@ -124,12 +118,10 @@ func main() {
 		IdleTimeoutSeconds:     *keepAlive,
 		FirehoseSubscriptionID: *subscriptionId,
 	}
-	logging.LogStd(fmt.Sprintf("connect logging '%s'",loggingClient.Connect()), true)
-	logging.LogStd(fmt.Sprintf("using '%s' as syslogServer", syslogServer), true)
-	logging.LogStd(fmt.Sprintf("using '%s' as syslogServer",  *debug), true)
-	if loggingClient.Connect() || *debug {
 
-		logging.LogStd("Connected to Syslog Server! Connecting to Firehose...", true)
+	if /*loggingClientSumo.Connect() ||*/ *debug {
+
+		logging.LogStd("Connected to Server! Connecting to Firehose...", true)
 		firehoseClient := firehoseclient.NewFirehoseNozzle(cfClient, events, firehoseConfig)
 		err = firehoseClient.Start()
 		if err != nil {

sumoCFFirehose/sumoCFFirehose.go

@@ -0,0 +1,6 @@
+package sumoCFFirehose
+
+type SumoCFFirehose interface {
+	Connect() bool
+	AppendLogs(map[string]interface{})
+}

sumoCFFirehose/sumoLog4gofakes/fake_SumoLog4go.go

@@ -0,0 +1,98 @@
+// This file was generated by counterfeiter
+package sumoLog4gofakes
+
+import (
+	"sync"
+
+	"bitbucket.org/mcplusa-ondemand/firehouse-to-sumologic/sumoCFFirehose"
+)
+
+type FakeSumoLog4go struct {
+	ConnectStub        func() bool
+	connectMutex       sync.RWMutex
+	connectArgsForCall []struct{}
+	connectReturns     struct {
+		result1 bool
+	}
+	AppendLogsStub        func(map[string]interface{})
+	AppendLogsMutex       sync.RWMutex
+	AppendLogsArgsForCall []struct {
+		arg1 map[string]interface{}
+	}
+	invocations      map[string][][]interface{}
+	invocationsMutex sync.RWMutex
+}
+
+func (fake *FakeSumoLog4go) Connect() bool {
+	fake.connectMutex.Lock()
+	fake.connectArgsForCall = append(fake.connectArgsForCall, struct{}{})
+	fake.recordInvocation("Connect", []interface{}{})
+	fake.connectMutex.Unlock()
+	if fake.ConnectStub != nil {
+		return fake.ConnectStub()
+	} else {
+		return fake.connectReturns.result1
+	}
+}
+
+func (fake *FakeSumoLog4go) ConnectCallCount() int {
+	fake.connectMutex.RLock()
+	defer fake.connectMutex.RUnlock()
+	return len(fake.connectArgsForCall)
+}
+
+func (fake *FakeSumoLog4go) ConnectReturns(result1 bool) {
+	fake.ConnectStub = nil
+	fake.connectReturns = struct {
+		result1 bool
+	}{result1}
+}
+
+func (fake *FakeSumoLog4go) AppendLogs(arg1 map[string]interface{}) {
+	fake.AppendLogsMutex.Lock()
+	fake.AppendLogsArgsForCall = append(fake.AppendLogsArgsForCall, struct {
+		arg1 map[string]interface{}
+		//arg2 string
+	}{arg1 /*, arg2*/})
+	fake.recordInvocation("ShipEvents", []interface{}{arg1 /*, arg2*/})
+	fake.AppendLogsMutex.Unlock()
+	if fake.AppendLogsStub != nil {
+		fake.AppendLogsStub(arg1 /*, arg2*/)
+	}
+}
+
+func (fake *FakeSumoLog4go) ShipEventsCallCount() int {
+	fake.AppendLogsMutex.RLock()
+	defer fake.AppendLogsMutex.RUnlock()
+	return len(fake.AppendLogsArgsForCall)
+}
+
+func (fake *FakeSumoLog4go) ShipEventsArgsForCall(i int) map[string]interface{} {
+	fake.AppendLogsMutex.RLock()
+	defer fake.AppendLogsMutex.RUnlock()
+	return fake.AppendLogsArgsForCall[i].arg1
+}
+
+func (fake *FakeSumoLog4go) Invocations() map[string][][]interface{} {
+	fake.invocationsMutex.RLock()
+	defer fake.invocationsMutex.RUnlock()
+	fake.connectMutex.RLock()
+	defer fake.connectMutex.RUnlock()
+	fake.AppendLogsMutex.RLock()
+	defer fake.AppendLogsMutex.RUnlock()
+	return fake.invocations
+}
+
+func (fake *FakeSumoLog4go) recordInvocation(key string, args []interface{}) {
+	fake.invocationsMutex.Lock()
+	defer fake.invocationsMutex.Unlock()
+	if fake.invocations == nil {
+		fake.invocations = map[string][][]interface{}{}
+	}
+	if fake.invocations[key] == nil {
+		fake.invocations[key] = [][]interface{}{}
+	}
+	fake.invocations[key] = append(fake.invocations[key], args)
+}
+
+var _ sumoCFFirehose.SumoCFFirehose = new(FakeSumoLog4go)

sumoCFFirehose/sumoLogicAppender.go

@@ -0,0 +1,81 @@
+package sumoCFFirehose
+
+import (
+	"bytes"
+	"fmt"
+	"net"
+	"net/http"
+	"time"
+)
+
+type SumoLogicAppender struct {
+	url               string
+	connectionTimeout int //10000
+	httpClient        http.Client
+}
+
+func NewSumoLogicAppender(urlValue string, connectionTimeoutValue int) *SumoLogicAppender {
+	return &SumoLogicAppender{
+		url:               urlValue,
+		connectionTimeout: connectionTimeoutValue,
+		httpClient:        http.Client{Timeout: time.Duration(connectionTimeoutValue * int(time.Millisecond))},
+	}
+}
+
+func (s *SumoLogicAppender) Connect() bool {
+	success := false
+	if s.url != "" {
+		conn, err := net.Dial("tcp", s.url)
+		if err != nil {
+			fmt.Printf(fmt.Sprintf("Unable to connect to sumo server [%s]!\n", s.url), err.Error())
+		} else {
+
+			fmt.Printf(fmt.Sprintf("Connected to [%s]!\n", s.url), false)
+			success = true
+			defer conn.Close()
+		}
+	}
+
+	return success
+}
+
+func (s *SumoLogicAppender) AppendLogs(Event map[string]interface{}) {
+	//adding the message to the map
+	if Event == nil {
+		return
+	}
+
+	if Event["msg"] == nil {
+		return
+	}
+
+	if Event["msg"] == "" {
+		return
+	}
+
+	Message := ""
+	Message = Event["msg"].(string) + "\n"
+
+	s.SendToSumo(Message)
+
+}
+
+func (s *SumoLogicAppender) SendToSumo(log string) {
+	request, err := http.NewRequest("POST", s.url, bytes.NewBufferString(log))
+	if err != nil {
+		fmt.Printf("http.NewRequest() error: %v\n", err)
+		return
+	}
+	request.Header.Add("content-type", "application/json")
+	//request.SetBasicAuth("admin", "admin")
+	response, err := s.httpClient.Do(request)
+
+	if err != nil {
+		fmt.Printf("http.Do() error: %v\n", err)
+		return
+	} else {
+		fmt.Println("Do(Request) successful")
+	}
+	defer response.Body.Close()
+
+}