Merge pull request #476 from xmidt-org/feat/sink-interface

Feat/sink interface

Author: maurafortino

caduceus_type.go

@@ -12,15 +12,6 @@ import (
 
 // Below is the struct we're using to contain the data from a provided config file
 // TODO: Try to figure out how to make bucket ranges configurable
-type CaduceusConfig struct {
-	AuthHeader       []string
-	NumWorkerThreads int
-	JobQueueSize     int
-	Sink             SinkConfig
-	JWTValidators    []JWTValidator
-	AllowInsecureTLS bool
-}
-
 type SinkConfig struct {
 	// The number of workers to assign to each SinkSender created.
 	NumWorkersPerSender int

http.go

@@ -58,19 +58,19 @@ type HandlerTelemetry struct {
 
 func (sh *ServerHandler) ServeHTTP(response http.ResponseWriter, request *http.Request) {
 	eventType := unknownEventType
-	log := sallust.Get(request.Context())
+	logger := sallust.Get(request.Context())
 	// find time difference, add to metric after function finishes
 	defer func(s time.Time) {
 		sh.recordQueueLatencyToHistogram(s, eventType)
 	}(sh.now())
 
-	log.Info("Receiving incoming request...")
+	logger.Info("Receiving incoming request...")
 
 	if len(request.Header["Content-Type"]) != 1 || request.Header["Content-Type"][0] != "application/msgpack" {
 		//return a 415
 		response.WriteHeader(http.StatusUnsupportedMediaType)
 		response.Write([]byte("Invalid Content-Type header(s). Expected application/msgpack. \n"))
-		log.Debug("Invalid Content-Type header(s). Expected application/msgpack. \n")
+		logger.Debug("Invalid Content-Type header(s). Expected application/msgpack. \n")
 		return
 	}
 
@@ -81,7 +81,7 @@ func (sh *ServerHandler) ServeHTTP(response http.ResponseWriter, request *http.R
 		// return a 503
 		response.WriteHeader(http.StatusServiceUnavailable)
 		response.Write([]byte("Incoming queue is full.\n"))
-		log.Debug("Incoming queue is full.\n")
+		logger.Debug("Incoming queue is full.\n")
 		return
 	}
 
@@ -91,14 +91,14 @@ func (sh *ServerHandler) ServeHTTP(response http.ResponseWriter, request *http.R
 	payload, err := io.ReadAll(request.Body)
 	if err != nil {
 		sh.telemetry.errorRequests.Add(1.0)
-		log.Error("Unable to retrieve the request body.", zap.Error(err))
+		logger.Error("Unable to retrieve the request body.", zap.Error(err))
 		response.WriteHeader(http.StatusBadRequest)
 		return
 	}
 
 	if len(payload) == 0 {
 		sh.telemetry.emptyRequests.Add(1.0)
-		log.Error("Empty payload.")
+		logger.Error("Empty payload.")
 		response.WriteHeader(http.StatusBadRequest)
 		response.Write([]byte("Empty payload.\n"))
 		return
@@ -114,10 +114,10 @@ func (sh *ServerHandler) ServeHTTP(response http.ResponseWriter, request *http.R
 		response.WriteHeader(http.StatusBadRequest)
 		if err != nil {
 			response.Write([]byte("Invalid payload format.\n"))
-			log.Debug("Invalid payload format.")
+			logger.Debug("Invalid payload format.")
 		} else {
 			response.Write([]byte("Invalid MessageType.\n"))
-			log.Debug("Invalid MessageType.")
+			logger.Debug("Invalid MessageType.")
 		}
 		return
 	}
@@ -128,7 +128,7 @@ func (sh *ServerHandler) ServeHTTP(response http.ResponseWriter, request *http.R
 		sh.telemetry.invalidCount.Add(1.0)
 		response.WriteHeader(http.StatusBadRequest)
 		response.Write([]byte("Strings must be UTF-8.\n"))
-		log.Debug("Strings must be UTF-8.")
+		logger.Debug("Strings must be UTF-8.")
 		return
 	}
 	eventType = msg.FindEventStringSubMatch()
@@ -139,7 +139,7 @@ func (sh *ServerHandler) ServeHTTP(response http.ResponseWriter, request *http.R
 	response.WriteHeader(http.StatusAccepted)
 	response.Write([]byte("Request placed on to queue.\n"))
 
-	log.Debug("event passed to senders.", zap.Any("event", msg))
+	logger.Debug("event passed to senders.", zap.Any("event", msg))
 }
 
 func (sh *ServerHandler) recordQueueLatencyToHistogram(startTime time.Time, eventType string) {

listenerStub.go

@@ -11,7 +11,6 @@ import (
 // This is a stub for the webhook and kafka listeners. This will be removed once the webhook-schema configuration is approved
 type Listener interface {
 	GetId() string
-	GetAddress() string
 	GetPartnerIds() []string
 	GetUntil() time.Time
 }
@@ -208,21 +207,13 @@ func (v1 *ListenerV1) GetPartnerIds() []string {
 	return v1.PartnerIds
 }
 
-func (v1 *ListenerV1) GetAddress() string {
-	return v1.Registration.Address
-}
-
 func (v1 *ListenerV1) GetUntil() time.Time {
 	return v1.Registration.Until
 }
 func (v2 *ListenerV2) GetId() string {
 	return v2.Registration.CanonicalName
 }
 
-func (v2 *ListenerV2) GetAddress() string {
-	return v2.Registration.Address
-}
-
 func (v2 *ListenerV2) GetPartnerIds() []string {
 	return v2.PartnerIds
 }

webhooks.go renamed to matcher.go

@@ -23,11 +23,16 @@ func (c *ClientMock) Do(req *http.Request) (*http.Response, error) {
 	return &http.Response{}, nil
 }
 
-type WebhookI interface {
-	CheckMsg(*wrp.Message) error
+// move to subpackage and change to Interface
+type Matcher interface {
+	IsMatch(*wrp.Message) bool
+
+	//TODO: not sure if this will be functionality of all webhooks or just v1
+	//leaving for now - will make changes if running into roadblock with this
 	getUrls() *ring.Ring
 }
-type WebhookV1 struct {
+
+type MatcherV1 struct {
 	events  []*regexp.Regexp
 	matcher []*regexp.Regexp
 	urls    *ring.Ring
@@ -41,10 +46,10 @@ type CommonWebhook struct {
 
 // Update applies user configurable values for the outbound sender when a
 // webhook is registered
-func (w1 *WebhookV1) Update(l ListenerV1) error {
+func (m1 *MatcherV1) Update(l ListenerV1) error {
 
 	//TODO: don't believe the logger for webhook is being set anywhere just yet
-	w1.logger = w1.logger.With(zap.String("webhook.address", l.Registration.Address))
+	m1.logger = m1.logger.With(zap.String("webhook.address", l.Registration.Address))
 
 	if l.Registration.FailureURL != "" {
 		_, err := url.ParseRequestURI(l.Registration.FailureURL)
@@ -86,54 +91,54 @@ func (w1 *WebhookV1) Update(l ListenerV1) error {
 	for i := 0; i < urlCount; i++ {
 		_, err := url.Parse(l.Registration.Config.AlternativeURLs[i])
 		if err != nil {
-			w1.logger.Error("failed to update url", zap.Any("url", l.Registration.Config.AlternativeURLs[i]), zap.Error(err))
+			m1.logger.Error("failed to update url", zap.Any("url", l.Registration.Config.AlternativeURLs[i]), zap.Error(err))
 			return err
 		}
 	}
 
 	// write/update sink sender
-	w1.mutex.Lock()
-	defer w1.mutex.Unlock()
+	m1.mutex.Lock()
+	defer m1.mutex.Unlock()
 
-	w1.events = events
+	m1.events = events
 
 	//TODO: need to figure out how to set this
 
 	// if matcher list is empty set it nil for Queue() logic
-	w1.matcher = nil
+	m1.matcher = nil
 	if 0 < len(matcher) {
-		w1.matcher = matcher
+		m1.matcher = matcher
 	}
 
 	if urlCount == 0 {
-		w1.urls = ring.New(1)
-		w1.urls.Value = l.Registration.Config.ReceiverURL
+		m1.urls = ring.New(1)
+		m1.urls.Value = l.Registration.Config.ReceiverURL
 	} else {
 		ring := ring.New(urlCount)
 		for i := 0; i < urlCount; i++ {
 			ring.Value = l.Registration.Config.AlternativeURLs[i]
 			ring = ring.Next()
 		}
-		w1.urls = ring
+		m1.urls = ring
 	}
 
 	// Randomize where we start so all the instances don't synchronize
 	rand := rand.New(rand.NewSource(time.Now().UnixNano()))
-	offset := rand.Intn(w1.urls.Len())
+	offset := rand.Intn(m1.urls.Len())
 	for 0 < offset {
-		w1.urls = w1.urls.Next()
+		m1.urls = m1.urls.Next()
 		offset--
 	}
 
 	return nil
 
 }
 
-func (w1 *WebhookV1) CheckMsg(msg *wrp.Message) (err error) {
-	w1.mutex.RLock()
-	events := w1.events
-	matcher := w1.matcher
-	w1.mutex.RUnlock()
+func (m1 *MatcherV1) IsMatch(msg *wrp.Message) bool {
+	m1.mutex.RLock()
+	events := m1.events
+	matcher := m1.matcher
+	m1.mutex.RUnlock()
 
 	var (
 		matchEvent  bool
@@ -146,9 +151,8 @@ func (w1 *WebhookV1) CheckMsg(msg *wrp.Message) (err error) {
 		}
 	}
 	if !matchEvent {
-		w1.logger.Debug("destination regex doesn't match", zap.String("event.dest", msg.Destination))
-		//TODO: return an error here?
-		return
+		m1.logger.Debug("destination regex doesn't match", zap.String("event.dest", msg.Destination))
+		return false
 	}
 
 	if matcher != nil {
@@ -162,37 +166,17 @@ func (w1 *WebhookV1) CheckMsg(msg *wrp.Message) (err error) {
 	}
 
 	if !matchDevice {
-		w1.logger.Debug("device regex doesn't match", zap.String("event.source", msg.Source))
-		//TODO: return an error here?
-		return
+		m1.logger.Debug("device regex doesn't match", zap.String("event.source", msg.Source))
+		return false
 	}
-	return
+	return true
 }
 
-func (w1 *WebhookV1) getUrls() (urls *ring.Ring) {
-	urls = w1.urls
+func (m1 *MatcherV1) getUrls() (urls *ring.Ring) {
+	urls = m1.urls
 	// Move to the next URL to try 1st the next time.
 	// This is okay because we run a single dispatcher and it's the
 	// only one updating this field.
-	w1.urls = w1.urls.Next()
+	m1.urls = m1.urls.Next()
 	return
 }
-
-type WebhookV2 struct {
-	//nolint:staticcheck
-	placeholder string
-	CommonWebhook
-}
-
-func (w2 *WebhookV2) Update(l ListenerV2) error {
-
-	return nil
-}
-
-func (w2 *WebhookV2) CheckMsg(msg *wrp.Message) error {
-	return nil
-}
-
-func (w2 *WebhookV2) getUrls() *ring.Ring {
-	return nil
-}