#136: Start working on database persistence

storage/store/database/database.go

@@ -0,0 +1,451 @@
+package database
+
+import (
+	"database/sql"
+	"errors"
+	"log"
+
+	"github.com/TwinProduction/gatus/core"
+	"github.com/TwinProduction/gatus/util"
+	_ "modernc.org/sqlite"
+)
+
+var (
+	// ErrFilePathNotSpecified is the error returned when path parameter passed in NewStore is blank
+	ErrFilePathNotSpecified = errors.New("file path cannot be empty")
+
+	// ErrDatabaseDriverNotSpecified is the error returned when the driver parameter passed in NewStore is blank
+	ErrDatabaseDriverNotSpecified = errors.New("database driver cannot be empty")
+
+	errServiceNotFoundInDatabase = errors.New("service does not exist in database")
+	errNoRowsReturned            = errors.New("expected a row to be returned, but none was")
+)
+
+// Store that leverages a database
+type Store struct {
+	driver, file string
+
+	db *sql.DB
+}
+
+// NewStore initializes the database and creates the schema if it doesn't already exist in the file specified
+func NewStore(driver, path string) (*Store, error) {
+	if len(driver) == 0 {
+		return nil, ErrDatabaseDriverNotSpecified
+	}
+	if len(path) == 0 {
+		return nil, ErrFilePathNotSpecified
+	}
+	store := &Store{driver: driver, file: path}
+	var err error
+	if store.db, err = sql.Open(driver, path); err != nil {
+		return nil, err
+	}
+	if err = store.createSchema(); err != nil {
+		_ = store.db.Close()
+		return nil, err
+	}
+	return store, nil
+}
+
+// createSchema creates the schema required to perform all database operations.
+func (s *Store) createSchema() error {
+	_, err := s.db.Exec(`
+		CREATE TABLE IF NOT EXISTS service (
+		    service_id    INTEGER PRIMARY KEY,
+			service_key   TEXT UNIQUE,
+			service_name  TEXT,
+			service_group TEXT,
+		    UNIQUE(service_name, service_group)
+		)
+	`)
+	if err != nil {
+		return err
+	}
+	_, err = s.db.Exec(`
+		CREATE TABLE IF NOT EXISTS service_event (
+		    service_event_id   INTEGER PRIMARY KEY,
+		    service_id         INTEGER REFERENCES service(id),
+			event_type         TEXT,
+			event_timestamp    TIMESTAMP
+		)
+	`)
+	if err != nil {
+		return err
+	}
+	_, err = s.db.Exec(`
+		CREATE TABLE IF NOT EXISTS service_result (
+		    service_result_id      INTEGER PRIMARY KEY,
+		    service_id             INTEGER REFERENCES service(id),
+			success                INTEGER,
+		    connected              INTEGER,
+		    status                 INTEGER,
+		    dns_rcode              TEXT,
+		    certificate_expiration INTEGER,
+		    hostname               TEXT,
+		    ip                     TEXT,
+		    duration               INTEGER,
+		    timestamp              TIMESTAMP
+		)
+	`)
+	if err != nil {
+		return err
+	}
+	_, err = s.db.Exec(`
+		CREATE TABLE IF NOT EXISTS service_result_condition (
+		    service_result_condition_id  INTEGER PRIMARY KEY,
+		    service_result_id            INTEGER REFERENCES service_result(service_result_id),
+		    condition                    TEXT,
+		    success                      INTEGER
+		)
+	`)
+	return err
+}
+
+// TODO: add parameter event and uptime & only fetch them if necessary
+
+func (s *Store) GetAllServiceStatusesWithResultPagination(page, pageSize int) map[string]*core.ServiceStatus {
+	panic("implement me")
+}
+
+// GetServiceStatus returns the service status for a given service name in the given group
+func (s *Store) GetServiceStatus(groupName, serviceName string) *core.ServiceStatus {
+	return s.GetServiceStatusByKey(util.ConvertGroupAndServiceToKey(groupName, serviceName))
+}
+
+// GetServiceStatusByKey returns the service status for a given key
+func (s *Store) GetServiceStatusByKey(key string) *core.ServiceStatus {
+	serviceID, serviceName, serviceGroup, err := s.getServiceIDNameAndGroupByKey(key)
+	if err != nil {
+		return nil
+	}
+	serviceStatus := &core.ServiceStatus{
+		Name:    serviceName,
+		Group:   serviceGroup,
+		Key:     key,
+		Results: nil,
+		Events:  nil,
+		Uptime:  nil,
+	}
+	if serviceStatus.Events, err = s.getEventsByServiceID(serviceID); err != nil {
+		log.Printf("[database][GetServiceStatusByKey] Failed to retrieve events for key=%s: %s", key, err.Error())
+	}
+	serviceStatus.Results, err = s.getResultsByServiceID(serviceID)
+	return serviceStatus
+}
+
+func (s *Store) getServiceIDNameAndGroupByKey(key string) (id int64, group, name string, err error) {
+	rows, err := s.db.Query("SELECT service_id, service_name, service_group FROM service WHERE service_key = $1 LIMIT 1", key)
+	if err != nil {
+		return 0, "", "", err
+	}
+	for rows.Next() {
+		_ = rows.Scan(&id, &name, &group)
+	}
+	if id == 0 {
+		return 0, "", "", errServiceNotFoundInDatabase
+	}
+	return
+}
+
+func (s *Store) getEventsByServiceID(serviceID int64) (events []*core.Event, err error) {
+	rows, err := s.db.Query("SELECT event_type, event_timestamp FROM service_event WHERE service_id = $1 ORDER BY event_timestamp DESC LIMIT $2", serviceID, core.MaximumNumberOfEvents)
+	if err != nil {
+		return nil, err
+	}
+	for rows.Next() {
+		event := &core.Event{}
+		_ = rows.Scan(&event.Type, &event.Timestamp)
+		events = append(events, event)
+	}
+	_ = rows.Close()
+	return
+}
+
+func (s *Store) getResultsByServiceID(serviceID int64) (results []*core.Result, err error) {
+	rows, err := s.db.Query(`
+			SELECT service_result_id, success, connected, status, dns_rcode, certificate_expiration, hostname, ip, duration, timestamp
+			FROM service_result
+			WHERE service_id = $1
+			ORDER BY timestamp DESC
+			LIMIT $2`,
+		serviceID,
+		core.MaximumNumberOfResults,
+	)
+	if err != nil {
+		return nil, err
+	}
+	idConditionResultsMap := make(map[int64]*[]*core.ConditionResult)
+	for rows.Next() {
+		result := &core.Result{}
+		var id int64
+		_ = rows.Scan(&id, &result.Success, &result.Connected, &result.HTTPStatus, &result.DNSRCode, &result.CertificateExpiration, &result.Hostname, &result.IP, &result.Duration, &result.Timestamp)
+		results = append(results, result)
+		idConditionResultsMap[id] = &result.ConditionResults
+	}
+	_ = rows.Close()
+	// Get the conditionResults
+	transaction, err := s.db.Begin()
+	if err != nil {
+		return
+	}
+	for serviceResultID, conditionResults := range idConditionResultsMap {
+		rows, err = transaction.Query(`
+			SELECT service_result_id, condition, success
+			FROM service_result_condition
+			WHERE service_result_id = $1`,
+			serviceResultID,
+		)
+		if err != nil {
+			_ = transaction.Rollback()
+			return
+		}
+		for rows.Next() {
+			var id int64
+			conditionResult := &core.ConditionResult{}
+			_ = rows.Scan(&id, &conditionResult.Condition, &conditionResult.Success)
+			*conditionResults = append(*conditionResults, conditionResult)
+			// XXX: profile this ^
+		}
+		_ = rows.Close()
+	}
+	err = transaction.Commit()
+	if err != nil {
+		_ = transaction.Rollback()
+		return
+	}
+	return
+}
+
+// Insert adds the observed result for the specified service into the store
+func (s *Store) Insert(service *core.Service, result *core.Result) {
+	//start := time.Now()
+	serviceID, err := s.getServiceID(service)
+	if err != nil {
+		if err == errServiceNotFoundInDatabase {
+			// Service doesn't exist in the database, insert it
+			if serviceID, err = s.insertService(service); err != nil {
+				return // failed to insert service
+			}
+		} else {
+			return
+		}
+	}
+	// First, we need to check if we need to insert a new event.
+	//
+	// A new event must be added if either of the following cases happen:
+	// 1. There is only 1 event. The total number of events for a service can only be 1 if the only existing event is
+	//    of type EventStart, in which case we will have to create a new event of type EventHealthy or EventUnhealthy
+	//    based on result.Success.
+	// 2. The lastResult.Success != result.Success. This implies that the service went from healthy to unhealthy or
+	//    vice-versa, in which case we will have to create a new event of type EventHealthy or EventUnhealthy
+	//	  based on result.Success.
+	numberOfEvents, err := s.getNumberOfEventsByServiceID(serviceID)
+	if err != nil {
+		return
+	}
+	//log.Printf("there are currently %d events", numberOfEvents)
+	if numberOfEvents == 0 {
+		// There's no events yet, which means we need to add the EventStart and the first healthy/unhealthy event
+		err = s.insertEvent(serviceID, &core.Event{
+			Type:      core.EventStart,
+			Timestamp: result.Timestamp.Add(-result.Duration),
+		})
+		if err != nil {
+			// Silently fail
+			log.Printf("[database][Insert] Failed to insert event=%s for group=%s; service=%s: %s", core.EventStart, service.Name, service.Group, err.Error())
+		}
+		event := generateEventBasedOnResult(result)
+		err = s.insertEvent(serviceID, event)
+		if err != nil {
+			// Silently fail
+			log.Printf("[database][Insert] Failed to insert event=%s for group=%s; service=%s: %s", event.Type, service.Name, service.Group, err.Error())
+		}
+	} else {
+		// Get the success value of the previous result
+		var lastResultSuccess bool
+		lastResultSuccess, err = s.getLastServiceResultSuccessValue(serviceID)
+		if err != nil {
+			log.Printf("[database][Insert] Failed to retrieve outcome of previous result for group=%s; service=%s: %s", service.Name, service.Group, err.Error())
+		} else {
+			// If we managed to retrieve the outcome of the previous result, we'll compare it with the new result.
+			// If the final outcome (success or failure) of the previous and the new result aren't the same, it means
+			// that the service either went from Healthy to Unhealthy or Unhealthy -> Healthy, therefore, we'll add
+			// an event to mark the change in state
+			if lastResultSuccess != result.Success {
+				event := generateEventBasedOnResult(result)
+				err = s.insertEvent(serviceID, event)
+				if err != nil {
+					// Silently fail
+					log.Printf("[database][Insert] Failed to insert event=%s for group=%s; service=%s: %s", event.Type, service.Name, service.Group, err.Error())
+				}
+			}
+		}
+	}
+	// Second, we need to insert the result.
+	err = s.insertResult(serviceID, result)
+	if err != nil {
+		// Silently fail
+		log.Printf("[database][Insert] Failed to insert result for group=%s; service=%s: %s", service.Name, service.Group, err.Error())
+	}
+	//log.Printf("[database][Insert] Successfully inserted result in duration=%dms", time.Since(start).Milliseconds())
+	return
+}
+
+func (s *Store) getServiceID(service *core.Service) (int64, error) {
+	rows, err := s.db.Query(
+		"SELECT service_id FROM service WHERE service_name = $1 AND service_group = $2",
+		service.Name,
+		service.Group,
+	)
+	if err != nil {
+		return 0, err
+	}
+	var id int64
+	var found bool
+	for rows.Next() {
+		_ = rows.Scan(&id)
+		found = true
+		break
+	}
+	_ = rows.Close()
+	if !found {
+		return 0, errServiceNotFoundInDatabase
+	}
+	return id, nil
+}
+
+// insertService inserts a service in the store and returns the generated id of said service
+func (s *Store) insertService(service *core.Service) (int64, error) {
+	result, err := s.db.Exec(
+		"INSERT INTO service (service_key, service_name, service_group) VALUES ($1, $2, $3)",
+		service.Key(),
+		service.Name,
+		service.Group,
+	)
+	if err != nil {
+		return 0, err
+	}
+	return result.LastInsertId()
+}
+
+func (s *Store) getNumberOfEventsByServiceID(serviceID int64) (int64, error) {
+	rows, err := s.db.Query("SELECT COUNT(1) FROM service_event WHERE service_id = $1", serviceID)
+	if err != nil {
+		return 0, err
+	}
+	var numberOfEvents int64
+	var found bool
+	for rows.Next() {
+		_ = rows.Scan(&numberOfEvents)
+		found = true
+		break
+	}
+	_ = rows.Close()
+	if !found {
+		return 0, errNoRowsReturned
+	}
+	return numberOfEvents, nil
+}
+
+// insertService inserts a service in the store and returns the generated id of said service
+func (s *Store) insertEvent(serviceID int64, event *core.Event) error {
+	_, err := s.db.Exec(
+		"INSERT INTO service_event (service_id, event_type, event_timestamp) VALUES ($1, $2, $3)",
+		serviceID,
+		event.Type,
+		event.Timestamp,
+	)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (s *Store) getLastServiceResultSuccessValue(serviceID int64) (bool, error) {
+	rows, err := s.db.Query("SELECT success FROM service_result WHERE service_id = $1 ORDER BY timestamp LIMIT 1", serviceID)
+	if err != nil {
+		return false, err
+	}
+	var success bool
+	var found bool
+	for rows.Next() {
+		_ = rows.Scan(&success)
+		found = true
+		break
+	}
+	_ = rows.Close()
+	if !found {
+		return false, errNoRowsReturned
+	}
+	return success, nil
+}
+
+// insertService inserts a service in the store and returns the generated id of said service
+func (s *Store) insertResult(serviceID int64, result *core.Result) error {
+	res, err := s.db.Exec(`
+			INSERT INTO service_result (service_id, success, connected, status, dns_rcode, certificate_expiration, hostname, ip, duration, timestamp)
+			VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)`,
+		serviceID,
+		result.Success,
+		result.Connected,
+		result.HTTPStatus,
+		result.DNSRCode,
+		result.CertificateExpiration,
+		result.Hostname,
+		result.IP,
+		result.Duration,
+		result.Timestamp,
+	)
+	if err != nil {
+		return err
+	}
+	serviceResultID, err := res.LastInsertId()
+	if err != nil {
+		return err
+	}
+	return s.insertConditionResults(serviceResultID, result.ConditionResults)
+}
+
+func (s *Store) insertConditionResults(serviceResultID int64, conditionResults []*core.ConditionResult) error {
+	transaction, err := s.db.Begin()
+	if err != nil {
+		return err
+	}
+	for _, cr := range conditionResults {
+		_, err = transaction.Exec("INSERT INTO service_result_condition (service_result_id, condition, success) VALUES ($1, $2, $3)",
+			serviceResultID,
+			cr.Condition,
+			cr.Success,
+		)
+		if err != nil {
+			_ = transaction.Rollback()
+			return err
+		}
+	}
+	err = transaction.Commit()
+	if err != nil {
+		_ = transaction.Rollback()
+		return err
+	}
+	return nil
+}
+
+// DeleteAllServiceStatusesNotInKeys removes all rows owned by a service whose key is not within the keys provided
+func (s *Store) DeleteAllServiceStatusesNotInKeys(keys []string) int {
+	panic("implement me")
+}
+
+// Clear deletes everything from the store
+func (s *Store) Clear() {
+	panic("implement me")
+}
+
+// Save does nothing, because this store is immediately persistent.
+func (s *Store) Save() error {
+	return nil
+}
+
+func (s *Store) Close() {
+	_ = s.db.Close()
+}

storage/store/database/database_test.go

@@ -0,0 +1,152 @@
+package database
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/TwinProduction/gatus/core"
+)
+
+var (
+	firstCondition  = core.Condition("[STATUS] == 200")
+	secondCondition = core.Condition("[RESPONSE_TIME] < 500")
+	thirdCondition  = core.Condition("[CERTIFICATE_EXPIRATION] < 72h")
+
+	timestamp = time.Now()
+
+	testService = core.Service{
+		Name:                    "name",
+		Group:                   "group",
+		URL:                     "https://example.org/what/ever",
+		Method:                  "GET",
+		Body:                    "body",
+		Interval:                30 * time.Second,
+		Conditions:              []*core.Condition{&firstCondition, &secondCondition, &thirdCondition},
+		Alerts:                  nil,
+		Insecure:                false,
+		NumberOfFailuresInARow:  0,
+		NumberOfSuccessesInARow: 0,
+	}
+	testSuccessfulResult = core.Result{
+		Hostname:              "example.org",
+		IP:                    "127.0.0.1",
+		HTTPStatus:            200,
+		Errors:                nil,
+		Connected:             true,
+		Success:               true,
+		Timestamp:             timestamp,
+		Duration:              150 * time.Millisecond,
+		CertificateExpiration: 10 * time.Hour,
+		ConditionResults: []*core.ConditionResult{
+			{
+				Condition: "[STATUS] == 200",
+				Success:   true,
+			},
+			{
+				Condition: "[RESPONSE_TIME] < 500",
+				Success:   true,
+			},
+			{
+				Condition: "[CERTIFICATE_EXPIRATION] < 72h",
+				Success:   true,
+			},
+		},
+	}
+	testUnsuccessfulResult = core.Result{
+		Hostname:              "example.org",
+		IP:                    "127.0.0.1",
+		HTTPStatus:            200,
+		Errors:                []string{"error-1", "error-2"},
+		Connected:             true,
+		Success:               false,
+		Timestamp:             timestamp,
+		Duration:              750 * time.Millisecond,
+		CertificateExpiration: 10 * time.Hour,
+		ConditionResults: []*core.ConditionResult{
+			{
+				Condition: "[STATUS] == 200",
+				Success:   true,
+			},
+			{
+				Condition: "[RESPONSE_TIME] < 500",
+				Success:   false,
+			},
+			{
+				Condition: "[CERTIFICATE_EXPIRATION] < 72h",
+				Success:   false,
+			},
+		},
+	}
+)
+
+func TestNewStore(t *testing.T) {
+	if _, err := NewStore("", "TestNewStore.db"); err != ErrDatabaseDriverNotSpecified {
+		t.Error("expected error due to blank driver parameter")
+	}
+	if _, err := NewStore("sqlite", ""); err != ErrFilePathNotSpecified {
+		t.Error("expected error due to blank path parameter")
+	}
+	if store, err := NewStore("sqlite", t.TempDir()+"/TestNewStore.db"); err != nil {
+		t.Error("shouldn't have returned any error, got", err.Error())
+	} else {
+		_ = store.db.Close()
+	}
+}
+
+func TestStore_Insert(t *testing.T) {
+	store, _ := NewStore("sqlite", t.TempDir()+"/TestStore_Insert.db")
+	defer store.db.Close()
+	//store.Insert(&testService, &testSuccessfulResult)
+	//
+	//actual := store.GetServiceStatusByKey(testService.Key())
+	//t.Fatal(actual)
+
+	store.Insert(&testService, &testSuccessfulResult)
+	store.Insert(&testService, &testUnsuccessfulResult)
+
+	key := fmt.Sprintf("%s_%s", testService.Group, testService.Name)
+	serviceStatus := store.GetServiceStatusByKey(key)
+	if serviceStatus == nil {
+		t.Fatalf("Store should've had key '%s', but didn't", key)
+	}
+	if len(serviceStatus.Results) != 2 {
+		t.Fatalf("Service '%s' should've had 2 results, but actually returned %d", serviceStatus.Name, len(serviceStatus.Results))
+	}
+	for i, r := range serviceStatus.Results {
+		expectedResult := store.GetServiceStatus(testService.Group, testService.Name).Results[i]
+		if r.HTTPStatus != expectedResult.HTTPStatus {
+			t.Errorf("Result at index %d should've had a HTTPStatus of %d, but was actually %d", i, expectedResult.HTTPStatus, r.HTTPStatus)
+		}
+		if r.DNSRCode != expectedResult.DNSRCode {
+			t.Errorf("Result at index %d should've had a DNSRCode of %s, but was actually %s", i, expectedResult.DNSRCode, r.DNSRCode)
+		}
+		if r.Hostname != expectedResult.Hostname {
+			t.Errorf("Result at index %d should've had a Hostname of %s, but was actually %s", i, expectedResult.Hostname, r.Hostname)
+		}
+		if r.IP != expectedResult.IP {
+			t.Errorf("Result at index %d should've had a IP of %s, but was actually %s", i, expectedResult.IP, r.IP)
+		}
+		if r.Connected != expectedResult.Connected {
+			t.Errorf("Result at index %d should've had a Connected value of %t, but was actually %t", i, expectedResult.Connected, r.Connected)
+		}
+		if r.Duration != expectedResult.Duration {
+			t.Errorf("Result at index %d should've had a Duration of %s, but was actually %s", i, expectedResult.Duration.String(), r.Duration.String())
+		}
+		if len(r.Errors) != len(expectedResult.Errors) {
+			t.Errorf("Result at index %d should've had %d errors, but actually had %d errors", i, len(expectedResult.Errors), len(r.Errors))
+		}
+		if len(r.ConditionResults) != len(expectedResult.ConditionResults) {
+			t.Errorf("Result at index %d should've had %d ConditionResults, but actually had %d ConditionResults", i, len(expectedResult.ConditionResults), len(r.ConditionResults))
+		}
+		if r.Success != expectedResult.Success {
+			t.Errorf("Result at index %d should've had a Success of %t, but was actually %t", i, expectedResult.Success, r.Success)
+		}
+		if r.Timestamp != expectedResult.Timestamp {
+			t.Errorf("Result at index %d should've had a Timestamp of %s, but was actually %s", i, expectedResult.Timestamp.String(), r.Timestamp.String())
+		}
+		if r.CertificateExpiration != expectedResult.CertificateExpiration {
+			t.Errorf("Result at index %d should've had a CertificateExpiration of %s, but was actually %s", i, expectedResult.CertificateExpiration.String(), r.CertificateExpiration.String())
+		}
+	}
+}

storage/store/database/util.go

@@ -0,0 +1,16 @@
+package database
+
+import "github.com/TwinProduction/gatus/core"
+
+func generateEventBasedOnResult(result *core.Result) *core.Event {
+	var eventType core.EventType
+	if result.Success {
+		eventType = core.EventHealthy
+	} else {
+		eventType = core.EventUnhealthy
+	}
+	return &core.Event{
+		Type:      eventType,
+		Timestamp: result.Timestamp,
+	}
+}

storage/store/store.go

@@ -2,6 +2,7 @@ package store
 
 import (
 	"github.com/TwinProduction/gatus/core"
+	"github.com/TwinProduction/gatus/storage/store/database"
 	"github.com/TwinProduction/gatus/storage/store/memory"
 )
 
@@ -35,4 +36,5 @@ type Store interface {
 var (
 	// Validate interface implementation on compile
 	_ Store = (*memory.Store)(nil)
+	_ Store = (*database.Store)(nil)
 )

storage/store/store_bench_test.go

@@ -5,6 +5,7 @@ import (
 	"time"
 
 	"github.com/TwinProduction/gatus/core"
+	"github.com/TwinProduction/gatus/storage/store/database"
 	"github.com/TwinProduction/gatus/storage/store/memory"
 )
 
@@ -112,6 +113,11 @@ func BenchmarkStore_Insert(b *testing.B) {
 	if err != nil {
 		b.Fatal("failed to create store:", err.Error())
 	}
+	databaseStore, err := database.NewStore("sqlite", b.TempDir()+"/BenchmarkStore_Insert.db")
+	if err != nil {
+		b.Fatal("failed to create store:", err.Error())
+	}
+	defer databaseStore.Close()
 	type Scenario struct {
 		Name  string
 		Store Store
@@ -121,14 +127,18 @@ func BenchmarkStore_Insert(b *testing.B) {
 			Name:  "memory",
 			Store: memoryStore,
 		},
+		{
+			Name:  "database",
+			Store: databaseStore,
+		},
 	}
 	for _, scenario := range scenarios {
 		b.Run(scenario.Name, func(b *testing.B) {
 			for n := 0; n < b.N; n++ {
 				if n%100 == 0 {
-					scenario.Store.Insert(&testService, &testSuccessfulResult)
-				} else {
 					scenario.Store.Insert(&testService, &testUnsuccessfulResult)
+				} else {
+					scenario.Store.Insert(&testService, &testSuccessfulResult)
 				}
 			}
 			b.ReportAllocs()