SCFF-21 added changes related to the HTTP post methods

Author: nmillalaf

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/sumoLog4go/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/sumoLog4go" //**
+	"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          sumoLog4go.SumoLog4go //**
 	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 sumoLog4go.SumoLog4go) *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, event.Msg) //**
+			e.log.ShipEvents(event.Fields, event.Msg)        // here we have to change the method for the one on sumoLogicAppender
 			e.selectedEventsCount[eventType.String()]++
 
 		}

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/sumoLog4go"
 	"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()
@@ -45,11 +47,12 @@ func main() {
 
 	//Setup Logging
 	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")
-         //}
+	loggingClientSumo := sumoLog4go.NewSumoLogicAppender("http://httpbin.org/post", 1000)
+	//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)
 
 	if *modeProf != "" {
@@ -78,8 +81,8 @@ func main() {
 	}
 
 	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 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)
 
@@ -91,7 +94,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 +127,12 @@ 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)
+	//	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)
+		logging.LogStd("Connected to Server! Connecting to Firehose...", true)
 		firehoseClient := firehoseclient.NewFirehoseNozzle(cfClient, events, firehoseConfig)
 		err = firehoseClient.Start()
 		if err != nil {

sumo-log4go/sumoLogicAppender.go

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

sumoLog4go/sumoLog4go.go

@@ -0,0 +1,99 @@
+// This file was generated by counterfeiter
+package sumoLog4gofakes
+
+import (
+	"sync"
+
+	"bitbucket.org/mcplusa-ondemand/firehouse-to-sumologic/sumolog4go"
+)
+
+type FakeSumoLog4go struct {
+	ConnectStub        func() bool
+	connectMutex       sync.RWMutex
+	connectArgsForCall []struct{}
+	connectReturns     struct {
+		result1 bool
+	}
+	AppendLogsStub        func(map[string]interface{}, string)
+	AppendLogsMutex       sync.RWMutex
+	AppendLogsArgsForCall []struct {
+		arg1 map[string]interface{}
+		arg2 string
+	}
+	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{}, arg2 string) {
+	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{}, string) {
+	fake.AppendLogsMutex.RLock()
+	defer fake.AppendLogsMutex.RUnlock()
+	return fake.AppendLogsArgsForCall[i].arg1, fake.AppendLogsArgsForCall[i].arg2
+}
+
+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 _ sumoLog4go.SumoLog4go = new(FakeSumoLog4go)

sumoLog4go/sumoLog4gofakes/fake_SumoLog4go.go

@@ -0,0 +1,77 @@
+package sumoLog4go
+
+import (
+	"bytes"
+	"encoding/json"
+	"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)
+		fmt.Printf(fmt.Sprintf("Unable to connect to sumo server [%s]!\n", s.url), err.Error())
+		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{}, Message string) {
+	//adding the message to the map
+	Event["msg"] = Message
+	jsonEvent, err := json.Marshal(Event)
+	if err == nil {
+		fmt.Println("-----here are the logs to send to sumo-------")
+
+		//fmt.Println(string(jsonEvent)) //**
+		s.SendToSumo(jsonEvent)
+		//fmt.Println("---------------------------------------------")
+	}
+
+}
+
+func (s *SumoLogicAppender) SendToSumo(log []byte) {
+	request, err := http.NewRequest("POST", s.url, bytes.NewBuffer(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
+		//consume the body if you want to re-use the connection
+	} else {
+		fmt.Println("Do(Request) successful")
+	}
+	fmt.Println(response)
+	defer response.Body.Close()
+
+}