phares3757/gatus
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: Break core package into multiple packages under config/endpoint (#759)

Browse Source 
* refactor: Partially break core package into dns, result and ssh packages

* refactor: Move core package to config/endpoint

* refactor: Fix warning about overlapping imported package name with endpoint variable

* refactor: Rename EndpointStatus to Status

* refactor: Merge result pkg back into endpoint pkg, because it makes more sense

* refactor: Rename parameter r to result in Condition.evaluate

* refactor: Rename parameter r to result

* refactor: Revert accidental change to endpoint.TypeDNS

* refactor: Rename parameter r to result

* refactor: Merge util package into endpoint package

* refactor: Rename parameter r to result
This commit is contained in:
TwiN
2024-05-09 22:56:16 -04:00
committed by GitHub
parent 4397dcb5fc
commit 9d151fcdb4
104 changed files with 1216 additions and 1211 deletions
Download Patch File Download Diff File Expand all files Collapse all files

105
storage/store/sql/sql.go
View File

@ -9,10 +9,9 @@ import (
"strings"
"time"
"github.com/TwiN/gatus/v5/core"
"github.com/TwiN/gatus/v5/config/endpoint"
"github.com/TwiN/gatus/v5/storage/store/common"
"github.com/TwiN/gatus/v5/storage/store/common/paging"
"github.com/TwiN/gatus/v5/util"
"github.com/TwiN/gocache/v2"
_ "github.com/lib/pq"
_ "modernc.org/sqlite"
@ -100,9 +99,9 @@ func (s *Store) createSchema() error {
return s.createPostgresSchema()
}
// GetAllEndpointStatuses returns all monitored core.EndpointStatus
// with a subset of core.Result defined by the page and pageSize parameters
func (s *Store) GetAllEndpointStatuses(params *paging.EndpointStatusParams) ([]*core.EndpointStatus, error) {
// GetAllEndpointStatuses returns all monitored endpoint.Status
// with a subset of endpoint.Result defined by the page and pageSize parameters
func (s *Store) GetAllEndpointStatuses(params *paging.EndpointStatusParams) ([]*endpoint.Status, error) {
tx, err := s.db.Begin()
if err != nil {
return nil, err
@ -112,7 +111,7 @@ func (s *Store) GetAllEndpointStatuses(params *paging.EndpointStatusParams) ([]*
_ = tx.Rollback()
return nil, err
}
endpointStatuses := make([]*core.EndpointStatus, 0, len(keys))
endpointStatuses := make([]*endpoint.Status, 0, len(keys))
for _, key := range keys {
endpointStatus, err := s.getEndpointStatusByKey(tx, key, params)
if err != nil {
@ -127,12 +126,12 @@ func (s *Store) GetAllEndpointStatuses(params *paging.EndpointStatusParams) ([]*
}
// GetEndpointStatus returns the endpoint status for a given endpoint name in the given group
func (s *Store) GetEndpointStatus(groupName, endpointName string, params *paging.EndpointStatusParams) (*core.EndpointStatus, error) {
return s.GetEndpointStatusByKey(util.ConvertGroupAndEndpointNameToKey(groupName, endpointName), params)
func (s *Store) GetEndpointStatus(groupName, endpointName string, params *paging.EndpointStatusParams) (*endpoint.Status, error) {
return s.GetEndpointStatusByKey(endpoint.ConvertGroupAndEndpointNameToKey(groupName, endpointName), params)
}
// GetEndpointStatusByKey returns the endpoint status for a given key
func (s *Store) GetEndpointStatusByKey(key string, params *paging.EndpointStatusParams) (*core.EndpointStatus, error) {
func (s *Store) GetEndpointStatusByKey(key string, params *paging.EndpointStatusParams) (*endpoint.Status, error) {
tx, err := s.db.Begin()
if err != nil {
return nil, err
@ -224,30 +223,30 @@ func (s *Store) GetHourlyAverageResponseTimeByKey(key string, from, to time.Time
}
// Insert adds the observed result for the specified endpoint into the store
func (s *Store) Insert(endpoint *core.Endpoint, result *core.Result) error {
func (s *Store) Insert(ep *endpoint.Endpoint, result *endpoint.Result) error {
tx, err := s.db.Begin()
if err != nil {
return err
}
endpointID, err := s.getEndpointID(tx, endpoint)
endpointID, err := s.getEndpointID(tx, ep)
if err != nil {
if errors.Is(err, common.ErrEndpointNotFound) {
// Endpoint doesn't exist in the database, insert it
if endpointID, err = s.insertEndpoint(tx, endpoint); err != nil {
if endpointID, err = s.insertEndpoint(tx, ep); err != nil {
_ = tx.Rollback()
log.Printf("[sql.Insert] Failed to create endpoint with group=%s; endpoint=%s: %s", endpoint.Group, endpoint.Name, err.Error())
log.Printf("[sql.Insert] Failed to create endpoint with group=%s; endpoint=%s: %s", ep.Group, ep.Name, err.Error())
return err
}
} else {
_ = tx.Rollback()
log.Printf("[sql.Insert] Failed to retrieve id of endpoint with group=%s; endpoint=%s: %s", endpoint.Group, endpoint.Name, err.Error())
log.Printf("[sql.Insert] Failed to retrieve id of endpoint with group=%s; endpoint=%s: %s", ep.Group, ep.Name, err.Error())
return err
}
}
// 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 endpoint can only be 1 if the only existing event is
// 1. There is only 1 event. The total number of events for an endpoint 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 endpoint went from healthy to unhealthy or
@ -256,38 +255,38 @@ func (s *Store) Insert(endpoint *core.Endpoint, result *core.Result) error {
numberOfEvents, err := s.getNumberOfEventsByEndpointID(tx, endpointID)
if err != nil {
// Silently fail
log.Printf("[sql.Insert] Failed to retrieve total number of events for group=%s; endpoint=%s: %s", endpoint.Group, endpoint.Name, err.Error())
log.Printf("[sql.Insert] Failed to retrieve total number of events for group=%s; endpoint=%s: %s", ep.Group, ep.Name, err.Error())
}
if numberOfEvents == 0 {
// There's no events yet, which means we need to add the EventStart and the first healthy/unhealthy event
err = s.insertEndpointEvent(tx, endpointID, &core.Event{
Type: core.EventStart,
err = s.insertEndpointEvent(tx, endpointID, &endpoint.Event{
Type: endpoint.EventStart,
Timestamp: result.Timestamp.Add(-50 * time.Millisecond),
})
if err != nil {
// Silently fail
log.Printf("[sql.Insert] Failed to insert event=%s for group=%s; endpoint=%s: %s", core.EventStart, endpoint.Group, endpoint.Name, err.Error())
log.Printf("[sql.Insert] Failed to insert event=%s for group=%s; endpoint=%s: %s", endpoint.EventStart, ep.Group, ep.Name, err.Error())
}
event := core.NewEventFromResult(result)
event := endpoint.NewEventFromResult(result)
if err = s.insertEndpointEvent(tx, endpointID, event); err != nil {
// Silently fail
log.Printf("[sql.Insert] Failed to insert event=%s for group=%s; endpoint=%s: %s", event.Type, endpoint.Group, endpoint.Name, err.Error())
log.Printf("[sql.Insert] Failed to insert event=%s for group=%s; endpoint=%s: %s", event.Type, ep.Group, ep.Name, err.Error())
}
} else {
// Get the success value of the previous result
var lastResultSuccess bool
if lastResultSuccess, err = s.getLastEndpointResultSuccessValue(tx, endpointID); err != nil {
log.Printf("[sql.Insert] Failed to retrieve outcome of previous result for group=%s; endpoint=%s: %s", endpoint.Group, endpoint.Name, err.Error())
log.Printf("[sql.Insert] Failed to retrieve outcome of previous result for group=%s; endpoint=%s: %s", ep.Group, ep.Name, 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 endpoint 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 := core.NewEventFromResult(result)
event := endpoint.NewEventFromResult(result)
if err = s.insertEndpointEvent(tx, endpointID, event); err != nil {
// Silently fail
log.Printf("[sql.Insert] Failed to insert event=%s for group=%s; endpoint=%s: %s", event.Type, endpoint.Group, endpoint.Name, err.Error())
log.Printf("[sql.Insert] Failed to insert event=%s for group=%s; endpoint=%s: %s", event.Type, ep.Group, ep.Name, err.Error())
}
}
}
@ -296,45 +295,45 @@ func (s *Store) Insert(endpoint *core.Endpoint, result *core.Result) error {
// (since we're only deleting MaximumNumberOfEvents at a time instead of 1)
if numberOfEvents > eventsCleanUpThreshold {
if err = s.deleteOldEndpointEvents(tx, endpointID); err != nil {
log.Printf("[sql.Insert] Failed to delete old events for group=%s; endpoint=%s: %s", endpoint.Group, endpoint.Name, err.Error())
log.Printf("[sql.Insert] Failed to delete old events for group=%s; endpoint=%s: %s", ep.Group, ep.Name, err.Error())
}
}
}
// Second, we need to insert the result.
if err = s.insertEndpointResult(tx, endpointID, result); err != nil {
log.Printf("[sql.Insert] Failed to insert result for group=%s; endpoint=%s: %s", endpoint.Group, endpoint.Name, err.Error())
log.Printf("[sql.Insert] Failed to insert result for group=%s; endpoint=%s: %s", ep.Group, ep.Name, err.Error())
_ = tx.Rollback() // If we can't insert the result, we'll rollback now since there's no point continuing
return err
}
// Clean up old results
numberOfResults, err := s.getNumberOfResultsByEndpointID(tx, endpointID)
if err != nil {
log.Printf("[sql.Insert] Failed to retrieve total number of results for group=%s; endpoint=%s: %s", endpoint.Group, endpoint.Name, err.Error())
log.Printf("[sql.Insert] Failed to retrieve total number of results for group=%s; endpoint=%s: %s", ep.Group, ep.Name, err.Error())
} else {
if numberOfResults > resultsCleanUpThreshold {
if err = s.deleteOldEndpointResults(tx, endpointID); err != nil {
log.Printf("[sql.Insert] Failed to delete old results for group=%s; endpoint=%s: %s", endpoint.Group, endpoint.Name, err.Error())
log.Printf("[sql.Insert] Failed to delete old results for group=%s; endpoint=%s: %s", ep.Group, ep.Name, err.Error())
}
}
}
// Finally, we need to insert the uptime data.
// Because the uptime data significantly outlives the results, we can't rely on the results for determining the uptime
if err = s.updateEndpointUptime(tx, endpointID, result); err != nil {
log.Printf("[sql.Insert] Failed to update uptime for group=%s; endpoint=%s: %s", endpoint.Group, endpoint.Name, err.Error())
log.Printf("[sql.Insert] Failed to update uptime for group=%s; endpoint=%s: %s", ep.Group, ep.Name, err.Error())
}
// Clean up old uptime entries
ageOfOldestUptimeEntry, err := s.getAgeOfOldestEndpointUptimeEntry(tx, endpointID)
if err != nil {
log.Printf("[sql.Insert] Failed to retrieve oldest endpoint uptime entry for group=%s; endpoint=%s: %s", endpoint.Group, endpoint.Name, err.Error())
log.Printf("[sql.Insert] Failed to retrieve oldest endpoint uptime entry for group=%s; endpoint=%s: %s", ep.Group, ep.Name, err.Error())
} else {
if ageOfOldestUptimeEntry > uptimeCleanUpThreshold {
if err = s.deleteOldUptimeEntries(tx, endpointID, time.Now().Add(-(uptimeRetention + time.Hour))); err != nil {
log.Printf("[sql.Insert] Failed to delete old uptime entries for group=%s; endpoint=%s: %s", endpoint.Group, endpoint.Name, err.Error())
log.Printf("[sql.Insert] Failed to delete old uptime entries for group=%s; endpoint=%s: %s", ep.Group, ep.Name, err.Error())
}
}
}
if s.writeThroughCache != nil {
cacheKeysToRefresh := s.writeThroughCache.GetKeysByPattern(endpoint.Key()+"*", 0)
cacheKeysToRefresh := s.writeThroughCache.GetKeysByPattern(ep.Key()+"*", 0)
for _, cacheKey := range cacheKeysToRefresh {
s.writeThroughCache.Delete(cacheKey)
endpointKey, params, err := extractKeyAndParamsFromCacheKey(cacheKey)
@ -405,14 +404,14 @@ func (s *Store) Close() {
}
// insertEndpoint inserts an endpoint in the store and returns the generated id of said endpoint
func (s *Store) insertEndpoint(tx *sql.Tx, endpoint *core.Endpoint) (int64, error) {
//log.Printf("[sql.insertEndpoint] Inserting endpoint with group=%s and name=%s", endpoint.Group, endpoint.Name)
func (s *Store) insertEndpoint(tx *sql.Tx, ep *endpoint.Endpoint) (int64, error) {
//log.Printf("[sql.insertEndpoint] Inserting endpoint with group=%s and name=%s", ep.Group, ep.Name)
var id int64
err := tx.QueryRow(
"INSERT INTO endpoints (endpoint_key, endpoint_name, endpoint_group) VALUES ($1, $2, $3) RETURNING endpoint_id",
endpoint.Key(),
endpoint.Name,
endpoint.Group,
ep.Key(),
ep.Name,
ep.Group,
).Scan(&id)
if err != nil {
return 0, err
@ -421,7 +420,7 @@ func (s *Store) insertEndpoint(tx *sql.Tx, endpoint *core.Endpoint) (int64, erro
}
// insertEndpointEvent inserts en event in the store
func (s *Store) insertEndpointEvent(tx *sql.Tx, endpointID int64, event *core.Event) error {
func (s *Store) insertEndpointEvent(tx *sql.Tx, endpointID int64, event *endpoint.Event) error {
_, err := tx.Exec(
"INSERT INTO endpoint_events (endpoint_id, event_type, event_timestamp) VALUES ($1, $2, $3)",
endpointID,
@ -435,7 +434,7 @@ func (s *Store) insertEndpointEvent(tx *sql.Tx, endpointID int64, event *core.Ev
}
// insertEndpointResult inserts a result in the store
func (s *Store) insertEndpointResult(tx *sql.Tx, endpointID int64, result *core.Result) error {
func (s *Store) insertEndpointResult(tx *sql.Tx, endpointID int64, result *endpoint.Result) error {
var endpointResultID int64
err := tx.QueryRow(
`
@ -462,7 +461,7 @@ func (s *Store) insertEndpointResult(tx *sql.Tx, endpointID int64, result *core.
return s.insertConditionResults(tx, endpointResultID, result.ConditionResults)
}
func (s *Store) insertConditionResults(tx *sql.Tx, endpointResultID int64, conditionResults []*core.ConditionResult) error {
func (s *Store) insertConditionResults(tx *sql.Tx, endpointResultID int64, conditionResults []*endpoint.ConditionResult) error {
var err error
for _, cr := range conditionResults {
_, err = tx.Exec("INSERT INTO endpoint_result_conditions (endpoint_result_id, condition, success) VALUES ($1, $2, $3)",
@ -477,7 +476,7 @@ func (s *Store) insertConditionResults(tx *sql.Tx, endpointResultID int64, condi
return nil
}
func (s *Store) updateEndpointUptime(tx *sql.Tx, endpointID int64, result *core.Result) error {
func (s *Store) updateEndpointUptime(tx *sql.Tx, endpointID int64, result *endpoint.Result) error {
unixTimestampFlooredAtHour := result.Timestamp.Truncate(time.Hour).Unix()
var successfulExecutions int
if result.Success {
@ -514,12 +513,12 @@ func (s *Store) getAllEndpointKeys(tx *sql.Tx) (keys []string, err error) {
return
}
func (s *Store) getEndpointStatusByKey(tx *sql.Tx, key string, parameters *paging.EndpointStatusParams) (*core.EndpointStatus, error) {
func (s *Store) getEndpointStatusByKey(tx *sql.Tx, key string, parameters *paging.EndpointStatusParams) (*endpoint.Status, error) {
var cacheKey string
if s.writeThroughCache != nil {
cacheKey = generateCacheKey(key, parameters)
if cachedEndpointStatus, exists := s.writeThroughCache.Get(cacheKey); exists {
if castedCachedEndpointStatus, ok := cachedEndpointStatus.(*core.EndpointStatus); ok {
if castedCachedEndpointStatus, ok := cachedEndpointStatus.(*endpoint.Status); ok {
return castedCachedEndpointStatus, nil
}
}
@ -528,7 +527,7 @@ func (s *Store) getEndpointStatusByKey(tx *sql.Tx, key string, parameters *pagin
if err != nil {
return nil, err
}
endpointStatus := core.NewEndpointStatus(group, endpointName)
endpointStatus := endpoint.NewStatus(group, endpointName)
if parameters.EventsPageSize > 0 {
if endpointStatus.Events, err = s.getEndpointEventsByEndpointID(tx, endpointID, parameters.EventsPage, parameters.EventsPageSize); err != nil {
log.Printf("[sql.getEndpointStatusByKey] Failed to retrieve events for key=%s: %s", key, err.Error())
@ -564,7 +563,7 @@ func (s *Store) getEndpointIDGroupAndNameByKey(tx *sql.Tx, key string) (id int64
return
}
func (s *Store) getEndpointEventsByEndpointID(tx *sql.Tx, endpointID int64, page, pageSize int) (events []*core.Event, err error) {
func (s *Store) getEndpointEventsByEndpointID(tx *sql.Tx, endpointID int64, page, pageSize int) (events []*endpoint.Event, err error) {
rows, err := tx.Query(
`
SELECT event_type, event_timestamp
@ -581,14 +580,14 @@ func (s *Store) getEndpointEventsByEndpointID(tx *sql.Tx, endpointID int64, page
return nil, err
}
for rows.Next() {
event := &core.Event{}
event := &endpoint.Event{}
_ = rows.Scan(&event.Type, &event.Timestamp)
events = append(events, event)
}
return
}
func (s *Store) getEndpointResultsByEndpointID(tx *sql.Tx, endpointID int64, page, pageSize int) (results []*core.Result, err error) {
func (s *Store) getEndpointResultsByEndpointID(tx *sql.Tx, endpointID int64, page, pageSize int) (results []*endpoint.Result, err error) {
rows, err := tx.Query(
`
SELECT endpoint_result_id, success, errors, connected, status, dns_rcode, certificate_expiration, domain_expiration, hostname, ip, duration, timestamp
@ -604,9 +603,9 @@ func (s *Store) getEndpointResultsByEndpointID(tx *sql.Tx, endpointID int64, pag
if err != nil {
return nil, err
}
idResultMap := make(map[int64]*core.Result)
idResultMap := make(map[int64]*endpoint.Result)
for rows.Next() {
result := &core.Result{}
result := &endpoint.Result{}
var id int64
var joinedErrors string
err = rows.Scan(&id, &result.Success, &joinedErrors, &result.Connected, &result.HTTPStatus, &result.DNSRCode, &result.CertificateExpiration, &result.DomainExpiration, &result.Hostname, &result.IP, &result.Duration, &result.Timestamp)
@ -618,7 +617,7 @@ func (s *Store) getEndpointResultsByEndpointID(tx *sql.Tx, endpointID int64, pag
result.Errors = strings.Split(joinedErrors, arraySeparator)
}
// This is faster than using a subselect
results = append([]*core.Result{result}, results...)
results = append([]*endpoint.Result{result}, results...)
idResultMap[id] = result
}
if len(idResultMap) == 0 {
@ -643,7 +642,7 @@ func (s *Store) getEndpointResultsByEndpointID(tx *sql.Tx, endpointID int64, pag
}
defer rows.Close() // explicitly defer the close in case an error happens during the scan
for rows.Next() {
conditionResult := &core.ConditionResult{}
conditionResult := &endpoint.ConditionResult{}
var endpointResultID int64
if err = rows.Scan(&endpointResultID, &conditionResult.Condition, &conditionResult.Success); err != nil {
return
@ -734,9 +733,9 @@ func (s *Store) getEndpointHourlyAverageResponseTimes(tx *sql.Tx, endpointID int
return hourlyAverageResponseTimes, nil
}
func (s *Store) getEndpointID(tx *sql.Tx, endpoint *core.Endpoint) (int64, error) {
func (s *Store) getEndpointID(tx *sql.Tx, ep *endpoint.Endpoint) (int64, error) {
var id int64
err := tx.QueryRow("SELECT endpoint_id FROM endpoints WHERE endpoint_key = $1", endpoint.Key()).Scan(&id)
err := tx.QueryRow("SELECT endpoint_id FROM endpoints WHERE endpoint_key = $1", ep.Key()).Scan(&id)
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
return 0, common.ErrEndpointNotFound