notifier testing - improved notifier queue system

.dockerignore

@@ -9,4 +9,6 @@ source/rice-box.go
 vendor
 servers
 dev
+*.lock
+*.toml
 !build/alpine-linux-amd64

Dockerfile

@@ -1,4 +1,5 @@
-FROM hunterlong/statup:base-v0.59
+ARG VERSION
+FROM hunterlong/statup:base-v${VERSION}
 MAINTAINER "Hunter Long (https://github.com/hunterlong)"
 
 # Locked version of Statup for 'latest' Docker tag

Makefile

@@ -119,11 +119,13 @@ build-alpine: compile
 
 # build :latest docker tag
 docker-build-latest: docker-build-base
-	docker build -t hunterlong/statup:latest --no-cache -f Dockerfile .
+	docker build --build-arg VERSION=$(VERSION) -t hunterlong/statup:latest --no-cache -f Dockerfile .
+	docker tag hunterlong/statup:latest hunterlong/statup:latest-v$(VERSION)
 
 # build :dev docker tag
 docker-build-dev: docker-build-base
-	docker build -t hunterlong/statup:dev --no-cache -f dev/Dockerfile-dev .
+	docker build --build-arg VERSION=$(VERSION) -t hunterlong/statup:dev --no-cache -f dev/Dockerfile-dev .
+	docker tag hunterlong/statup:dev hunterlong/statup:dev-v$(VERSION)
 
 # build :base and base-v{VERSION} docker tag
 docker-build-base: clean
@@ -150,9 +152,16 @@ docker-run-dev: docker-build-dev
 docker-run-cypress: docker-build-cypress
 	docker run -t hunterlong/statup:cypress
 
+# push the :base and :base-v{VERSION} tag to Docker hub
+docker-push-base:
+	docker tag hunterlong/statup:base hunterlong/statup:base-v$(VERSION)
+	docker push hunterlong/statup:base
+	docker push hunterlong/statup:base-v$(VERSION)
+
 # push the :dev tag to Docker hub
 docker-push-dev:
 	docker push hunterlong/statup:dev
+	docker push hunterlong/statup:dev-v$(VERSION)
 
 # push the :cypress tag to Docker hub
 docker-push-cypress:
@@ -161,12 +170,7 @@ docker-push-cypress:
 # push the :latest tag to Docker hub
 docker-push-latest:
 	docker push hunterlong/statup:latest
-
+	docker push hunterlong/statup:latest-v$(VERSION)
-# push the :base and :base-v{VERSION} tag to Docker hub
-docker-push-base: docker-build-base
-	docker tag hunterlong/statup:base hunterlong/statup:base-v$(VERSION)
-	docker push hunterlong/statup:base
-	docker push hunterlong/statup:base-v$(VERSION)
 
 # create Postgres, and MySQL instance using Docker (used for testing)
 databases:

core/notifier/example_test.go

@@ -33,7 +33,8 @@ var example = &Example{&Notification{
 	Description: "Example Notifier",
 	Author:      "Hunter Long",
 	AuthorUrl:   "https://github.com/hunterlong",
-	Delay:       time.Duration(5 * time.Second),
+	Delay:       time.Duration(1 * time.Second),
+	Limits:      7,
 	Form: []NotificationForm{{
 		Type:        "text",
 		Title:       "Host",

core/notifier/notifiers.go

@@ -56,6 +56,7 @@ type Notification struct {
 	AuthorUrl   string             `gorm:"-" json:"-"`
 	Delay       time.Duration      `gorm:"-" json:"-"`
 	Queue       []interface{}      `gorm:"-" json:"-"`
+	Running     chan bool          `gorm:"-" json:"-"`
 }
 
 type NotificationForm struct {
@@ -81,10 +82,6 @@ func modelDb(n *Notification) *gorm.DB {
 	return db.Model(&Notification{}).Where("method = ?", n.Method).Find(n)
 }
 
-func toNotification(n Notifier) *Notification {
-	return n.Select()
-}
-
 // SetDB is called by core to inject the database for a notifier to use
 func SetDB(d *gorm.DB) {
 	db = d
@@ -156,7 +153,7 @@ func (n *Notification) removeQueue(msg interface{}) interface{} {
 }
 
 // Log will record a new notification into memory and will show the logs on the settings page
-func (n *Notification) Log(msg interface{}) {
+func (n *Notification) makeLog(msg interface{}) {
 	log := &NotificationLog{
 		Message:   normalizeType(msg),
 		Time:      utils.Timestamp(time.Now()),
@@ -192,9 +189,14 @@ func SelectNotification(n Notifier) (*Notification, error) {
 }
 
 // Update will update the notification into the database
-func (n *Notification) Update() (*Notification, error) {
+func Update(n Notifier, notif *Notification) (*Notification, error) {
-	err := db.Model(&Notification{}).Update(n)
+	err := db.Model(&Notification{}).Update(notif)
-	return n, err.Error
+	if notif.Enabled {
+		notif.close()
+		notif.start()
+		go Queue(n)
+	}
+	return notif, err.Error
 }
 
 // insertDatabase will create a new record into the database for the notifier
@@ -208,18 +210,18 @@ func insertDatabase(n *Notification) (int64, error) {
 }
 
 // SelectNotifier returns the Notification struct from the database
-func SelectNotifier(method string) (*Notification, error) {
+func SelectNotifier(method string) (*Notification, Notifier, error) {
 	for _, comm := range AllCommunications {
 		n, ok := comm.(Notifier)
 		if !ok {
-			return nil, errors.New(fmt.Sprintf("incorrect notification type: %v", reflect.TypeOf(n).String()))
+			return nil, nil, errors.New(fmt.Sprintf("incorrect notification type: %v", reflect.TypeOf(n).String()))
 		}
 		notifier := n.Select()
 		if notifier.Method == method {
-			return notifier, nil
+			return notifier, comm.(Notifier), nil
 		}
 	}
-	return nil, nil
+	return nil, nil, nil
 }
 
 // Init accepts the Notifier interface to initialize the notifier
@@ -228,7 +230,7 @@ func Init(n Notifier) (*Notification, error) {
 	var notify *Notification
 	if err == nil {
 		notify, _ = SelectNotification(n)
-		notify.Form = toNotification(n).Form
+		notify.Form = n.Select().Form
 	}
 	return notify, err
 }
@@ -236,48 +238,49 @@ func Init(n Notifier) (*Notification, error) {
 func startAllNotifiers() {
 	for _, comm := range AllCommunications {
 		if isType(comm, new(Notifier)) {
-			if toNotification(comm.(Notifier)).Enabled {
+			notify := comm.(Notifier)
-				go runQue(comm.(Notifier))
+			if notify.Select().Enabled {
+				notify.Select().close()
+				notify.Select().start()
+				go Queue(notify)
 			}
 		}
 	}
 }
 
-func runQue(n Notifier) {
+func Queue(n Notifier) {
-	for {
 	notification := n.Select()
+	rateLimit := notification.Delay
+
+CheckNotifier:
+	for {
+		select {
+		case <-notification.Running:
+			break CheckNotifier
+		case <-time.After(rateLimit):
+			notification = n.Select()
 			if len(notification.Queue) > 0 {
-			for _, msg := range notification.Queue {
 				if notification.WithinLimits() {
+					msg := notification.Queue[0]
 					err := n.Send(msg)
 					if err != nil {
 						utils.Log(2, fmt.Sprintf("notifier %v had an error: %v", notification.Method, err))
 					}
-					notification.Log(msg)
+					notification.makeLog(msg)
+					notification.Queue = notification.Queue[1:]
+					rateLimit = notification.Delay
 				}
 			}
 		}
-		time.Sleep(notification.Delay)
+		continue
 	}
 }
 
-func RunQue(n Notifier) error {
-	notifier := n.Select()
-	if len(notifier.Queue) == 0 {
-		return nil
-	}
-	queMsg := notifier.Queue[0]
-	err := n.Send(queMsg)
-	notifier.Log(queMsg)
-	notifier.Queue = notifier.Queue[1:]
-	return err
-}
-
 // install will check the database for the notification, if its not inserted it will insert a new record for it
 func install(n Notifier) error {
 	inDb := isInDatabase(n.Select())
 	if !inDb {
-		_, err := insertDatabase(toNotification(n))
+		_, err := insertDatabase(n.Select())
 		if err != nil {
 			utils.Log(3, err)
 			return err
@@ -297,9 +300,9 @@ func (f *Notification) LastSent() time.Duration {
 }
 
 // SentLastHour returns the amount of sent notifications within the last hour
-func (f *Notification) SentLastHour() int {
+func (f *Notification) SentLastMinute() int {
 	sent := 0
-	hourAgo := time.Now().Add(-1 * time.Hour)
+	hourAgo := time.Now().Add(-1 * time.Minute)
 	for _, v := range f.Logs() {
 		lastTime := time.Time(v.Time)
 		if lastTime.After(hourAgo) {
@@ -309,11 +312,6 @@ func (f *Notification) SentLastHour() int {
 	return sent
 }
 
-// Limit returns the limits on how many notifications can be sent in 1 hour
-func (f *Notification) Limit() int {
-	return f.Limits
-}
-
 // GetValue returns the database value of a accept DbField value.
 func (n *Notification) GetValue(dbField string) string {
 	dbField = strings.ToLower(dbField)
@@ -356,19 +354,48 @@ func isEnabled(n interface{}) bool {
 }
 
 func inLimits(n interface{}) bool {
-	notifier := toNotification(n.(Notifier))
+	notifier := n.(Notifier).Select()
 	return notifier.WithinLimits()
 }
 
 func (notify *Notification) WithinLimits() bool {
-	if notify.SentLastHour() >= notify.Limit() {
+	if notify.SentLastMinute() >= notify.Limits {
 		return false
 	}
 	if notify.Delay.Seconds() == 0 {
-		notify.Delay = time.Duration(2 * time.Second)
+		notify.Delay = time.Duration(500 * time.Millisecond)
 	}
-	if notify.LastSent().Seconds() >= notify.Delay.Seconds() {
+	if notify.LastSent().Seconds() == 0 {
+		return true
+	}
+	if notify.Delay.Seconds() >= notify.LastSent().Seconds() {
 		return false
 	}
 	return true
 }
+
+func (n *Notification) ResetQueue() {
+	n.Queue = nil
+}
+
+func (n *Notification) start() {
+	n.Running = make(chan bool)
+}
+
+func (n *Notification) close() {
+	if n.IsRunning() {
+		close(n.Running)
+	}
+}
+
+func (n *Notification) IsRunning() bool {
+	if n.Running == nil {
+		return false
+	}
+	select {
+	case <-n.Running:
+		return false
+	default:
+		return true
+	}
+}

core/notifier/notifiers_test.go

@@ -91,6 +91,7 @@ func TestSelectNotification(t *testing.T) {
 	assert.Nil(t, err)
 	assert.Equal(t, "example", notifier.Method)
 	assert.False(t, notifier.Enabled)
+	assert.False(t, notifier.IsRunning())
 }
 
 func TestAddQueue(t *testing.T) {
@@ -118,8 +119,8 @@ func TestNotification_Update(t *testing.T) {
 	notifier.Var2 = "var2_is_here"
 	notifier.ApiKey = "USBdu82HDiiuw9327yGYDGw"
 	notifier.ApiSecret = "PQopncow929hUIDHGwiud"
-	notifier.Limits = 15
+	notifier.Limits = 10
-	_, err = notifier.Update()
+	_, err = Update(example, notifier)
 	assert.Nil(t, err)
 
 	selected, err := SelectNotification(example)
@@ -130,20 +131,26 @@ func TestNotification_Update(t *testing.T) {
 	assert.Equal(t, "USBdu82HDiiuw9327yGYDGw", selected.GetValue("api_key"))
 	assert.Equal(t, "USBdu82HDiiuw9327yGYDGw", example.ApiKey)
 	assert.False(t, selected.Enabled)
+	assert.False(t, selected.IsRunning())
 }
 
 func TestEnableNotification(t *testing.T) {
 	notifier, err := SelectNotification(example)
 	notifier.Enabled = true
-	updated, err := notifier.Update()
+	updated, err := Update(example, notifier)
 	assert.Nil(t, err)
 	assert.True(t, updated.Enabled)
+	assert.True(t, updated.IsRunning())
 }
 
 func TestIsEnabled(t *testing.T) {
 	assert.True(t, isEnabled(example))
 }
 
+func TestIsRunning(t *testing.T) {
+	assert.True(t, example.IsRunning())
+}
+
 func TestLastSent(t *testing.T) {
 	notifier, err := SelectNotification(example)
 	assert.Nil(t, err)
@@ -153,8 +160,7 @@ func TestLastSent(t *testing.T) {
 func TestWithinLimits(t *testing.T) {
 	notifier, err := SelectNotification(example)
 	assert.Nil(t, err)
-	assert.Equal(t, 15, notifier.Limit())
+	assert.Equal(t, 10, notifier.Limits)
-	assert.Equal(t, 15, notifier.Limits)
 	assert.True(t, inLimits(example))
 }
 
@@ -165,25 +171,6 @@ func TestNotification_GetValue(t *testing.T) {
 	assert.Equal(t, "http://demo.statup.io/api", val)
 }
 
-func TestRunQue(t *testing.T) {
-	assert.Nil(t, RunQue(example))
-	assert.Equal(t, 4, len(example.Queue))
-	assert.Nil(t, RunQue(example))
-	assert.Equal(t, 3, len(example.Queue))
-	assert.Nil(t, RunQue(example))
-	assert.Equal(t, 2, len(example.Queue))
-
-	time.Sleep(2 * time.Second)
-	assert.True(t, example.LastSent().Seconds() >= float64(2))
-
-	assert.Nil(t, RunQue(example))
-	assert.Equal(t, 1, len(example.Queue))
-	assert.Nil(t, RunQue(example))
-	assert.Equal(t, 0, len(example.Queue))
-	assert.Nil(t, RunQue(example))
-	assert.Equal(t, 0, len(example.Queue))
-}
-
 func TestOnSave(t *testing.T) {
 	err := example.OnSave()
 	assert.Equal(t, "onsave triggered", err.Error())
@@ -191,54 +178,64 @@ func TestOnSave(t *testing.T) {
 
 func TestOnSuccess(t *testing.T) {
 	OnSuccess(service)
-	assert.Equal(t, 2, len(example.Queue))
+	assert.Equal(t, 7, len(example.Queue))
 }
 
 func TestOnFailure(t *testing.T) {
 	OnFailure(service, failure)
-	assert.Equal(t, 3, len(example.Queue))
+	assert.Equal(t, 8, len(example.Queue))
 }
 
 func TestOnNewService(t *testing.T) {
 	OnNewService(service)
-	assert.Equal(t, 4, len(example.Queue))
+	assert.Equal(t, 9, len(example.Queue))
 }
 
 func TestOnUpdatedService(t *testing.T) {
 	OnUpdatedService(service)
-	assert.Equal(t, 5, len(example.Queue))
+	assert.Equal(t, 10, len(example.Queue))
 }
 
 func TestOnDeletedService(t *testing.T) {
 	OnDeletedService(service)
-	assert.Equal(t, 6, len(example.Queue))
+	assert.Equal(t, 11, len(example.Queue))
 }
 
 func TestOnNewUser(t *testing.T) {
 	OnNewUser(user)
-	assert.Equal(t, 7, len(example.Queue))
+	assert.Equal(t, 12, len(example.Queue))
 }
 
 func TestOnUpdatedUser(t *testing.T) {
 	OnUpdatedUser(user)
-	assert.Equal(t, 8, len(example.Queue))
+	assert.Equal(t, 13, len(example.Queue))
 }
 
 func TestOnDeletedUser(t *testing.T) {
 	OnDeletedUser(user)
-	assert.Equal(t, 9, len(example.Queue))
+	assert.Equal(t, 14, len(example.Queue))
 }
 
 func TestOnUpdatedCore(t *testing.T) {
 	OnUpdatedCore(core)
-	assert.Equal(t, 10, len(example.Queue))
+	assert.Equal(t, 15, len(example.Queue))
 }
 
 func TestOnUpdatedNotifier(t *testing.T) {
 	OnUpdatedNotifier(example.Select())
-	assert.Equal(t, 11, len(example.Queue))
+	assert.Equal(t, 16, len(example.Queue))
 }
 
-func TestRunAllQueue(t *testing.T) {
+func TestRunAllQueueAndStop(t *testing.T) {
-	//runQue(example)
+	assert.True(t, example.IsRunning())
+	assert.Equal(t, 16, len(example.Queue))
+	go Queue(example)
+	assert.Equal(t, 16, len(example.Queue))
+	time.Sleep(15 * time.Second)
+	assert.Equal(t, 6, len(example.Queue))
+	time.Sleep(1 * time.Second)
+	assert.Equal(t, 6, len(example.Queue))
+	example.close()
+	assert.False(t, example.IsRunning())
+	assert.Equal(t, 6, len(example.Queue))
 }

core/services.go

@@ -165,9 +165,6 @@ func GroupDataBy(column string, id int64, tm time.Time, increment string) string
 	return sql
 }
 
-type graphObject struct {
-}
-
 func (s *Service) GraphDataRaw() []*DateScan {
 	var d []*DateScan
 	since := time.Now().Add(time.Hour*-24 + time.Minute*0 + time.Second*0)

core/services_test.go

@@ -117,7 +117,7 @@ func TestServiceOnline24Hours(t *testing.T) {
 	service2 := SelectService(5)
 	assert.Equal(t, float32(100), service2.OnlineSince(since))
 	service3 := SelectService(14)
-	assert.Equal(t, float32(49.69), service3.OnlineSince(since))
+	assert.True(t, service3.OnlineSince(since) > float32(49))
 }
 
 func TestServiceSmallText(t *testing.T) {
@@ -135,7 +135,7 @@ func TestServiceAvgUptime(t *testing.T) {
 	service3 := SelectService(13)
 	assert.Equal(t, "100", service3.AvgUptime(since))
 	service4 := SelectService(15)
-	assert.Equal(t, "49.69", service4.AvgUptime(since))
+	assert.NotEqual(t, "0", service4.AvgUptime(since))
 }
 
 func TestServiceHits(t *testing.T) {

dev/Dockerfile-dev

@@ -1,3 +1,4 @@
+ARG VERSION
 FROM golang:1.10.3-alpine
 MAINTAINER "Hunter Long (https://github.com/hunterlong)"
 # Statup 'test' image for running a full test using the production environment
@@ -13,7 +14,7 @@ RUN curl -L -s https://assets.statup.io/sass -o /usr/local/bin/sass && \
 WORKDIR /go/src/github.com/hunterlong/statup
 ADD . /go/src/github.com/hunterlong/statup
 
-ENV VERSION=$(VERSION)
+ENV VERSION=${VERSION}
 ENV IS_DOCKER=true
 
 RUN make dev-deps

handlers/handlers_test.go

@@ -18,6 +18,7 @@ package handlers
 import (
 	"fmt"
 	"github.com/hunterlong/statup/core"
+	"github.com/hunterlong/statup/core/notifier"
 	_ "github.com/hunterlong/statup/notifiers"
 	"github.com/hunterlong/statup/source"
 	"github.com/hunterlong/statup/utils"
@@ -500,6 +501,10 @@ func TestPrometheusHandler(t *testing.T) {
 }
 
 func TestSaveNotificationHandler(t *testing.T) {
+	notification, _, err := notifier.SelectNotifier("email")
+	assert.Nil(t, err)
+	assert.False(t, notification.IsRunning())
+
 	form := url.Values{}
 	form.Add("enable", "on")
 	form.Add("host", "smtp.emailer.com")
@@ -518,6 +523,9 @@ func TestSaveNotificationHandler(t *testing.T) {
 	Router().ServeHTTP(rr, req)
 	assert.Equal(t, 303, rr.Code)
 	assert.True(t, isRouteAuthenticated(req))
+	notification, _, err = notifier.SelectNotifier("email")
+	assert.Nil(t, err)
+	assert.True(t, notification.IsRunning())
 }
 
 func TestViewNotificationSettingsHandler(t *testing.T) {

handlers/settings.go

@@ -144,7 +144,7 @@ func SaveNotificationHandler(w http.ResponseWriter, r *http.Request) {
 	apiSecret := form.Get("api_secret")
 	limits := int(utils.StringInt(form.Get("limits")))
 
-	notifer, err := notifier.SelectNotifier(method)
+	notifer, notif, err := notifier.SelectNotifier(method)
 	if err != nil {
 		utils.Log(3, fmt.Sprintf("issue saving notifier %v: %v", method, err))
 		ExecuteResponse(w, r, "settings.html", core.CoreApp, "/settings")
@@ -179,7 +179,7 @@ func SaveNotificationHandler(w http.ResponseWriter, r *http.Request) {
 		notifer.Limits = limits
 	}
 	notifer.Enabled = enabled == "on"
-	_, err = notifer.Update()
+	_, err = notifier.Update(notif, notifer)
 	if err != nil {
 		utils.Log(3, fmt.Sprintf("issue updating notifier: %v", err))
 	}