Page not found
@@ -53,6 +62,16 @@
{/if}
+{#if appConfig.version}
+
+{/if}
+
diff --git a/client/web/src/pages/Host.svelte b/client/web/src/pages/Host.svelte
index 2d38aac..309aab4 100644
--- a/client/web/src/pages/Host.svelte
+++ b/client/web/src/pages/Host.svelte
@@ -4,7 +4,8 @@
import { api, diskFullSoonDays, type DiskForecast, type HostDetail } from '../lib/api';
import { hostSections, loadChart, loadCores, type ChartSpec } from '../lib/charts';
import { appConfig } from '../lib/config.svelte';
- import { formatAgo, formatBytes, formatDays, formatDuration, formatMiB, formatPercent } from '../lib/format';
+ import { formatAgo, formatBytes, formatDate, formatDays, formatDuration, formatMiB, formatPercent } from '../lib/format';
+ import { projectDisks } from '../lib/projection';
import { poll } from '../lib/poll';
import { hostPath, location, navigate } from '../lib/router.svelte';
import { rangeQuery, resolveRange } from '../lib/timerange';
@@ -124,6 +125,9 @@
return !!chart?.data && chart.data.labels.length > 0;
}
+ // follows the disk space chart's series, so each disk keeps its color
+ const projection = $derived(projectDisks(Object.values(forecasts), now, charts['disk_used']?.data?.labels ?? []));
+
const snapshot = $derived(detail?.snapshot);
const system = $derived(snapshot?.System);
const cpu = $derived(snapshot?.ProcUsage);
@@ -142,6 +146,7 @@
{#if cpu?.Model}
{cpu.Model}{/if}
{#if cpu?.NoOfCores}
{cpu.NoOfCores} CPUs{/if}
{#if snapshot?.Memory.Total}
{formatMiB(snapshot.Memory.Total)} memory{/if}
+ {#if snapshot?.AgentVersion}
agent {snapshot.AgentVersion}{/if}
{:else if detailError}
{detailError === 'no data found' ? 'This host has not sent any data yet.' : `Could not load host: ${detailError}`}
{/if}
@@ -179,6 +184,38 @@
onpick={pickTime}
marker={processesAt}
/>
+ {#if spec.key === 'disk_used' && projection}
+
+ {#snippet table()}
+
+
+ | Mount | Used | Growth a day | Full in | Full on |
|---|
+
+
+ {#each projection.disks as disk (disk.device + disk.mount)}
+
+ | {disk.mount} |
+ {formatPercent(disk.usedPct)} |
+ {formatBytes(disk.bytesPerDay)} |
+ {formatDays(disk.daysToFull ?? 0)} |
+ {formatDate(now + (disk.daysToFull ?? 0) * 86400)} |
+
+ {/each}
+
+
+ {/snippet}
+
+ {/if}
{/each}
{#if section.title === 'CPU' && showCores && cores.length === 0}
diff --git a/collector/alerts.go b/collector/alerts.go
index ef15281..e095860 100644
--- a/collector/alerts.go
+++ b/collector/alerts.go
@@ -1,16 +1,11 @@
package main
import (
- "bytes"
- "crypto/tls"
- "crypto/x509"
+ "cmp"
"fmt"
"log"
- "net/http"
- "os"
"strconv"
"strings"
- "sync"
"time"
"github.com/dhamith93/SyMon/internal/alertapi"
@@ -21,7 +16,6 @@ import (
"github.com/dhamith93/SyMon/internal/monitor"
"github.com/dhamith93/SyMon/internal/store"
"github.com/dhamith93/SyMon/internal/transport"
- "github.com/dhamith93/SyMon/pkg/memdb"
)
// alertClient is created once in handleAlerts and shared by all alert checks
@@ -35,171 +29,36 @@ func handleAlerts(alertConfigs []alerts.AlertConfig, config *config.Collector, s
defer conn.Close()
alertClient = alertapi.NewAlertServiceClient(conn)
- ticker := time.NewTicker(15 * time.Second)
- endpointTicker := time.NewTicker(time.Duration(config.EndpointCheckInterval) * time.Second)
- quit := make(chan struct{})
- var wg sync.WaitGroup
- incidentTracker := memdb.CreateDatabase("incident_tracker")
- rules := newEvaluator(st, sendAlert)
-
- delta := 1
if config.EndpointMonitoringEnabled {
- delta = 2
+ logger.Log("info", "starting endpoint monitor")
+ go runEndpointChecks(alertConfigs, time.Duration(config.EndpointCheckInterval)*time.Second, st)
}
- wg.Add(delta)
logger.Log("info", "starting alert checker")
- go func() {
- for {
- select {
- case <-ticker.C:
- for _, alert := range alertConfigs {
- if alert.MetricName == "endpoint" {
- continue
- }
- for _, server := range alert.Servers {
- ctx, cancel := transport.Context()
- if err := rules.evaluate(ctx, &alert, server); err != nil {
- logger.Log("error", "alert "+alert.Name+" on "+server+": "+err.Error())
- }
- cancel()
- }
- }
- case <-quit:
- ticker.Stop()
- return
+ rules := newEvaluator(st, sendAlert)
+ ticker := time.NewTicker(15 * time.Second)
+ defer ticker.Stop()
+ for range ticker.C {
+ for _, alert := range alertConfigs {
+ // endpoint rules are checked from the collector, not on a host
+ hosts := alert.Servers
+ if alert.MetricName == monitor.ENDPOINT {
+ hosts = []string{""}
}
- }
- }()
-
- if config.EndpointMonitoringEnabled {
- logger.Log("info", "starting endpoint monitor")
- err := incidentTracker.Create(
- "endpoint_monitor",
- memdb.Col{Name: "url", Type: memdb.String},
- memdb.Col{Name: "method", Type: memdb.String},
- memdb.Col{Name: "expected", Type: memdb.Int},
- memdb.Col{Name: "actual", Type: memdb.Int},
- memdb.Col{Name: "time", Type: memdb.Int64},
- memdb.Col{Name: "failed", Type: memdb.Bool},
- memdb.Col{Name: "error", Type: memdb.String},
- memdb.Col{Name: "alerted", Type: memdb.Bool},
- )
- if err != nil {
- logger.Log("error", "memdb: "+err.Error())
- } else {
- go func() {
- for {
- select {
- case <-endpointTicker.C:
- for _, alert := range alertConfigs {
- if alert.MetricName == "endpoint" {
- checkEndpoint(&alert, &incidentTracker, config)
- }
- }
- case <-quit:
- ticker.Stop()
- return
- }
- }
- }()
- }
- }
- wg.Wait()
- fmt.Println("Exiting")
-}
-
-func checkEndpoint(alert *alerts.AlertConfig, incidentTracker *memdb.Database, config *config.Collector) {
- var (
- res *http.Response
- err error
- )
- customCACertUsed := len(strings.TrimSpace(alert.CustomCACert)) > 0
- client := &http.Client{}
-
- if customCACertUsed {
- caCert, err := os.ReadFile(alert.CustomCACert)
- if err != nil {
- logger.Log("error", err.Error())
- return
- }
- caCertPool := x509.NewCertPool()
- caCertPool.AppendCertsFromPEM(caCert)
-
- client = &http.Client{
- Transport: &http.Transport{
- TLSClientConfig: &tls.Config{
- RootCAs: caCertPool,
- },
- },
- }
- }
-
- method := strings.ToUpper(alert.Method)
-
- if method == alerts.ENDPOINT_METHOD_GET {
- res, err = client.Get(alert.Endpoint)
- }
- if method == alerts.ENDPOINT_METHOD_POST {
- body := []byte(alert.POSTBody)
- bodyReader := bytes.NewReader(body)
- res, err = client.Post(alert.Endpoint, alert.POSTContentType, bodyReader)
- }
-
- timeNow := time.Now().Unix()
- failed := false
- errMsg := ""
-
- if err != nil {
- logger.Log("error", err.Error())
- errMsg = err.Error()
- failed = true
- }
-
- statusCode := -1
-
- if !failed {
- defer res.Body.Close()
- statusCode = res.StatusCode
- }
-
- existingRecord := incidentTracker.Tables["endpoint_monitor"].Where("url", "==", alert.Endpoint).And("method", "==", method).And("expected", "==", alert.ExpectedHTTPCode)
-
- if existingRecord.RowCount > 0 {
- diff := timeNow - existingRecord.Rows[0].Columns["time"].Int64Val
- existingActual := existingRecord.Rows[0].Columns["actual"].IntVal
- alerted := existingRecord.Rows[0].Columns["alerted"].BoolVal
-
- existingRecord.Update("actual", statusCode)
- existingRecord.Update("failed", failed)
- existingRecord.Update("error", errMsg)
-
- if statusCode != alert.ExpectedHTTPCode {
- if alert.ExpectedHTTPCode != existingActual && !alerted {
- if diff > int64(alert.TriggerIntveral) {
- alertToSend := buildEndpointAlert(alert, statusCode, errMsg, false, timeNow)
- sendAlert(alertToSend)
- existingRecord.Update("alerted", true)
+ for _, server := range hosts {
+ ctx, cancel := transport.Context()
+ if err := rules.evaluate(ctx, &alert, server); err != nil {
+ logger.Log("error", "alert "+alert.Name+" on "+cmp.Or(server, alert.Endpoint)+": "+err.Error())
}
+ cancel()
}
- } else {
- if alert.ExpectedHTTPCode != existingActual && alerted {
- alertToSend := buildEndpointAlert(alert, statusCode, errMsg, true, timeNow)
- sendAlert(alertToSend)
- existingRecord.Update("alerted", false)
- }
- existingRecord.Update("time", timeNow)
}
-
- } else {
- incidentTracker.Tables["endpoint_monitor"].Insert(
- "url, method, expected, actual, time, failed, error, alerted",
- alert.Endpoint, method, alert.ExpectedHTTPCode, statusCode, timeNow, failed, errMsg, false,
- )
}
}
-func buildEndpointAlert(alert *alerts.AlertConfig, actualHTTPCode int, errMsg string, resolved bool, unixtime int64) *alertapi.Alert {
+// buildEndpointAlert fills an endpoint rule's template. id is the alert's
+// row id, and actualHTTPCode is 0 when there was no response.
+func buildEndpointAlert(alert *alerts.AlertConfig, id int64, actualHTTPCode int, errMsg string, resolved bool, unixtime int64) *alertapi.Alert {
subject := "[Resolved] "
status := 0
if !resolved {
@@ -217,7 +76,7 @@ func buildEndpointAlert(alert *alerts.AlertConfig, actualHTTPCode int, errMsg st
"{metricName}",
alert.MetricName,
"{expected}",
- strconv.Itoa(alert.ExpectedHTTPCode),
+ strconv.Itoa(expectedCode(alert)),
"{actual}",
strconv.Itoa(actualHTTPCode),
"{timestamp}",
@@ -231,6 +90,7 @@ func buildEndpointAlert(alert *alerts.AlertConfig, actualHTTPCode int, errMsg st
return &alertapi.Alert{
ServerName: alert.Endpoint,
MetricName: alert.MetricName,
+ LogId: id,
Status: int32(status),
Subject: subject,
Content: content,
diff --git a/collector/endpoints.go b/collector/endpoints.go
new file mode 100644
index 0000000..3930726
--- /dev/null
+++ b/collector/endpoints.go
@@ -0,0 +1,124 @@
+package main
+
+import (
+ "context"
+ "crypto/tls"
+ "crypto/x509"
+ "errors"
+ "io"
+ "net/http"
+ "os"
+ "strings"
+ "sync"
+ "time"
+
+ "github.com/dhamith93/SyMon/internal/alerts"
+ "github.com/dhamith93/SyMon/internal/logger"
+ "github.com/dhamith93/SyMon/internal/monitor"
+ "github.com/dhamith93/SyMon/internal/store"
+ "github.com/dhamith93/SyMon/internal/transport"
+)
+
+// endpointTimeout is how long a check waits for the response headers
+const endpointTimeout = 30 * time.Second
+
+// runEndpointChecks checks every endpoint rule now and then every interval,
+// and stores the results. The rule loop raises the alerts from them.
+func runEndpointChecks(rules []alerts.AlertConfig, interval time.Duration, st *store.Store) {
+ checkAll := func() {
+ var wg sync.WaitGroup
+ for i := range rules {
+ rule := &rules[i]
+ if rule.MetricName != monitor.ENDPOINT {
+ continue
+ }
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ check := checkEndpoint(context.Background(), rule, time.Now())
+ ctx, cancel := transport.Context()
+ defer cancel()
+ if err := st.SaveEndpointCheck(ctx, check); err != nil {
+ logger.Log("error", "cannot save the check of "+rule.Name+": "+err.Error())
+ }
+ }()
+ }
+ wg.Wait()
+ }
+
+ checkAll()
+ ticker := time.NewTicker(interval)
+ defer ticker.Stop()
+ for range ticker.C {
+ checkAll()
+ }
+}
+
+// expectedCode is the status code an endpoint rule wants, 200 when unset
+func expectedCode(rule *alerts.AlertConfig) int {
+ if rule.ExpectedHTTPCode == 0 {
+ return http.StatusOK
+ }
+ return rule.ExpectedHTTPCode
+}
+
+// checkEndpoint makes one request for an endpoint rule. A request that
+// fails has status code 0 and the error.
+func checkEndpoint(ctx context.Context, rule *alerts.AlertConfig, now time.Time) store.EndpointCheck {
+ method := strings.ToUpper(strings.TrimSpace(rule.Method))
+ if method == "" {
+ method = http.MethodGet
+ }
+ check := store.EndpointCheck{Time: now, Name: rule.Name, URL: rule.Endpoint, Method: method}
+
+ client, err := endpointClient(rule)
+ if err != nil {
+ check.Error = err.Error()
+ return check
+ }
+ // checks are minutes apart, so connections are not worth keeping
+ defer client.CloseIdleConnections()
+
+ var body io.Reader
+ if method == http.MethodPost {
+ body = strings.NewReader(rule.POSTBody)
+ }
+ request, err := http.NewRequestWithContext(ctx, method, rule.Endpoint, body)
+ if err != nil {
+ check.Error = err.Error()
+ return check
+ }
+ if method == http.MethodPost && rule.POSTContentType != "" {
+ request.Header.Set("Content-Type", rule.POSTContentType)
+ }
+
+ started := time.Now()
+ response, err := client.Do(request)
+ if err != nil {
+ check.Error = err.Error()
+ return check
+ }
+ check.Latency = time.Since(started)
+ response.Body.Close()
+ check.StatusCode = response.StatusCode
+ check.OK = response.StatusCode == expectedCode(rule)
+ return check
+}
+
+// endpointClient trusts the rule's CA certificate as well, when it has one
+func endpointClient(rule *alerts.AlertConfig) (*http.Client, error) {
+ client := &http.Client{Timeout: endpointTimeout}
+ if strings.TrimSpace(rule.CustomCACert) == "" {
+ return client, nil
+ }
+ pem, err := os.ReadFile(rule.CustomCACert)
+ if err != nil {
+ return nil, err
+ }
+ pool := x509.NewCertPool()
+ if !pool.AppendCertsFromPEM(pem) {
+ return nil, errors.New("no certificates in " + rule.CustomCACert)
+ }
+ client.Transport = &http.Transport{TLSClientConfig: &tls.Config{RootCAs: pool}}
+ return client, nil
+}
diff --git a/collector/endpoints_test.go b/collector/endpoints_test.go
new file mode 100644
index 0000000..c8ac6bb
--- /dev/null
+++ b/collector/endpoints_test.go
@@ -0,0 +1,87 @@
+package main
+
+import (
+ "context"
+ "encoding/pem"
+ "io"
+ "net/http"
+ "net/http/httptest"
+ "os"
+ "path/filepath"
+ "testing"
+ "time"
+
+ "github.com/dhamith93/SyMon/internal/alerts"
+ "github.com/dhamith93/SyMon/internal/monitor"
+)
+
+func TestCheckEndpoint(t *testing.T) {
+ var gotBody, gotType string
+ server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ switch r.URL.Path {
+ case "/down":
+ w.WriteHeader(http.StatusServiceUnavailable)
+ case "/post":
+ body, _ := io.ReadAll(r.Body)
+ gotBody, gotType = string(body), r.Header.Get("Content-Type")
+ w.WriteHeader(http.StatusCreated)
+ }
+ }))
+ defer server.Close()
+ now := time.Unix(1700000000, 0)
+
+ rule := &alerts.AlertConfig{Name: "api", MetricName: monitor.ENDPOINT, Endpoint: server.URL + "/"}
+ check := checkEndpoint(context.Background(), rule, now)
+ // the expected code is 200 when the rule does not say
+ if !check.OK || check.StatusCode != 200 || check.Method != "GET" || check.Latency <= 0 || !check.Time.Equal(now) || check.Error != "" {
+ t.Errorf("expected an OK check, got %+v", check)
+ }
+
+ rule.Endpoint = server.URL + "/down"
+ if check := checkEndpoint(context.Background(), rule, now); check.OK || check.StatusCode != 503 {
+ t.Errorf("expected a failed check with 503, got %+v", check)
+ }
+
+ post := &alerts.AlertConfig{Name: "post", Endpoint: server.URL + "/post", Method: "post", ExpectedHTTPCode: 201,
+ POSTBody: `{"ping":true}`, POSTContentType: "application/json"}
+ if check := checkEndpoint(context.Background(), post, now); !check.OK || check.Method != "POST" {
+ t.Errorf("expected an OK POST, got %+v", check)
+ }
+ if gotBody != `{"ping":true}` || gotType != "application/json" {
+ t.Errorf("unexpected request body %q and type %q", gotBody, gotType)
+ }
+
+ // nothing listens there any more
+ closed := httptest.NewServer(http.NotFoundHandler())
+ closed.Close()
+ rule.Endpoint = closed.URL
+ if check := checkEndpoint(context.Background(), rule, now); check.OK || check.StatusCode != 0 || check.Latency != 0 || check.Error == "" {
+ t.Errorf("expected a check without a response, got %+v", check)
+ }
+}
+
+func TestCheckEndpointWithCustomCA(t *testing.T) {
+ server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}))
+ defer server.Close()
+ rule := &alerts.AlertConfig{Name: "tls", Endpoint: server.URL}
+
+ // the test server's certificate is not trusted by default
+ if check := checkEndpoint(context.Background(), rule, time.Now()); check.OK || check.Error == "" {
+ t.Errorf("expected an untrusted certificate to fail, got %+v", check)
+ }
+
+ caPath := filepath.Join(t.TempDir(), "ca.pem")
+ caPEM := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: server.Certificate().Raw})
+ if err := os.WriteFile(caPath, caPEM, 0600); err != nil {
+ t.Fatal(err)
+ }
+ rule.CustomCACert = caPath
+ if check := checkEndpoint(context.Background(), rule, time.Now()); !check.OK {
+ t.Errorf("expected the custom CA to be trusted, got %+v", check)
+ }
+
+ rule.CustomCACert = filepath.Join(t.TempDir(), "missing.pem")
+ if check := checkEndpoint(context.Background(), rule, time.Now()); check.OK || check.Error == "" {
+ t.Errorf("expected a missing CA file to fail the check, got %+v", check)
+ }
+}
diff --git a/collector/evaluator.go b/collector/evaluator.go
index ad7a170..6b96007 100644
--- a/collector/evaluator.go
+++ b/collector/evaluator.go
@@ -18,6 +18,7 @@ type alertStore interface {
LatestValue(ctx context.Context, host string, metric string, target string, isCustom bool) (float64, time.Time, error)
LastSeen(ctx context.Context, host string) (time.Time, error)
DaysUntilFull(ctx context.Context, host string, device string) (float64, error)
+ LatestEndpointCheck(ctx context.Context, name string) (store.EndpointCheck, error)
OpenAlert(ctx context.Context, host string, rule string, metric string, target string) (*store.Alert, error)
CreateAlert(ctx context.Context, alert store.Alert) (int64, error)
UpdateAlert(ctx context.Context, id int64, severity int, value float64, at time.Time) error
@@ -36,6 +37,9 @@ type evaluator struct {
pending map[string]pendingStatus
// checked is the newest sample time seen, so each sample counts once
checked map[string]time.Time
+ // checkErrors is the error of each endpoint rule's newest check, for
+ // its alert message
+ checkErrors map[string]string
}
type pendingStatus struct {
@@ -45,21 +49,25 @@ type pendingStatus struct {
func newEvaluator(st alertStore, send func(*alertapi.Alert)) *evaluator {
return &evaluator{
- store: st,
- send: send,
- now: time.Now,
- pending: map[string]pendingStatus{},
- checked: map[string]time.Time{},
+ store: st,
+ send: send,
+ now: time.Now,
+ pending: map[string]pendingStatus{},
+ checked: map[string]time.Time{},
+ checkErrors: map[string]string{},
}
}
-// ruleTarget is the disk or service a rule watches, empty for other metrics
+// ruleTarget is the disk, service or URL a rule watches, empty for other
+// metrics
func ruleTarget(rule *alerts.AlertConfig) string {
switch rule.MetricName {
case monitor.DISKS, monitor.DISK_FORECAST:
return rule.Disk
case monitor.SERVICES:
return rule.Service
+ case monitor.ENDPOINT:
+ return rule.Endpoint
}
return ""
}
@@ -94,8 +102,17 @@ func (e *evaluator) evaluate(ctx context.Context, rule *alerts.AlertConfig, host
// latest returns the value a rule checks. For ping it is the number of
// seconds since the host was last heard from, for a disk forecast the days
-// until the disk is full.
+// until the disk is full, and for an endpoint the status code of its newest
+// check.
func (e *evaluator) latest(ctx context.Context, rule *alerts.AlertConfig, host string, target string) (float64, time.Time, error) {
+ if rule.MetricName == monitor.ENDPOINT {
+ check, err := e.store.LatestEndpointCheck(ctx, rule.Name)
+ if err != nil {
+ return 0, time.Time{}, err
+ }
+ e.checkErrors[rule.Name] = check.Error
+ return float64(check.StatusCode), check.Time, nil
+ }
if rule.MetricName == monitor.DISK_FORECAST {
days, err := e.store.DaysUntilFull(ctx, host, target)
return days, e.now(), err
@@ -116,7 +133,14 @@ func (e *evaluator) latest(ctx context.Context, rule *alerts.AlertConfig, host s
// ruleStatus compares a value with a rule. Ping rules have no thresholds,
// a host is critical once it has been silent longer than the trigger interval.
+// An endpoint is critical while it answers with another status code.
func ruleStatus(rule *alerts.AlertConfig, value float64) alertstatus.StatusType {
+ if rule.MetricName == monitor.ENDPOINT {
+ if int(value) != expectedCode(rule) {
+ return alertstatus.Critical
+ }
+ return alertstatus.Normal
+ }
if rule.MetricName == monitor.PING {
if value > float64(rule.TriggerIntveral) {
return alertstatus.Critical
@@ -140,7 +164,7 @@ func (e *evaluator) updateOpen(ctx context.Context, rule *alerts.AlertConfig, ho
return err
}
delete(e.pending, key)
- e.send(buildAlertToSend(host, rule, alertStatus(rule, host, open.ID, status, value, at)))
+ e.send(e.message(rule, host, open.ID, status, value, at))
return nil
}
@@ -152,7 +176,7 @@ func (e *evaluator) updateOpen(ctx context.Context, rule *alerts.AlertConfig, ho
if err := e.store.UpdateAlert(ctx, open.ID, int(status), value, at); err != nil {
return err
}
- e.send(buildAlertToSend(host, rule, alertStatus(rule, host, open.ID, status, value, at)))
+ e.send(e.message(rule, host, open.ID, status, value, at))
return nil
}
@@ -183,10 +207,19 @@ func (e *evaluator) openIfBreached(ctx context.Context, rule *alerts.AlertConfig
return err
}
delete(e.pending, key)
- e.send(buildAlertToSend(host, rule, alertStatus(rule, host, id, status, value, at)))
+ e.send(e.message(rule, host, id, status, value, at))
return nil
}
+// message builds what is sent to the alert processor. Endpoint rules have
+// their own template fields, like the status code and the error.
+func (e *evaluator) message(rule *alerts.AlertConfig, host string, id int64, status alertstatus.StatusType, value float64, at time.Time) *alertapi.Alert {
+ if rule.MetricName == monitor.ENDPOINT {
+ return buildEndpointAlert(rule, id, int(value), e.checkErrors[rule.Name], status == alertstatus.Normal, at.Unix())
+ }
+ return buildAlertToSend(host, rule, alertStatus(rule, host, id, status, value, at))
+}
+
func triggerInterval(rule *alerts.AlertConfig) time.Duration {
return time.Duration(rule.TriggerIntveral) * time.Second
}
diff --git a/collector/evaluator_test.go b/collector/evaluator_test.go
index 0287a4b..c3cb959 100644
--- a/collector/evaluator_test.go
+++ b/collector/evaluator_test.go
@@ -19,7 +19,9 @@ type fakeStore struct {
lastSeen time.Time
// daysUntilFull is keyed by device, a missing device has no forecast yet
daysUntilFull map[string]float64
- alerts []store.Alert
+ // check is the newest endpoint check, none while its time is zero
+ check store.EndpointCheck
+ alerts []store.Alert
}
func (f *fakeStore) LatestValue(ctx context.Context, host string, metric string, target string, isCustom bool) (float64, time.Time, error) {
@@ -41,6 +43,13 @@ func (f *fakeStore) DaysUntilFull(ctx context.Context, host string, device strin
return days, nil
}
+func (f *fakeStore) LatestEndpointCheck(ctx context.Context, name string) (store.EndpointCheck, error) {
+ if f.check.Time.IsZero() {
+ return store.EndpointCheck{}, store.ErrNotFound
+ }
+ return f.check, nil
+}
+
func (f *fakeStore) OpenAlert(ctx context.Context, host string, rule string, metric string, target string) (*store.Alert, error) {
for i := range f.alerts {
a := &f.alerts[i]
@@ -265,3 +274,54 @@ func TestDiskForecastAlert(t *testing.T) {
t.Errorf("expected the alert to resolve, got %+v", sent)
}
}
+
+func TestEndpointAlert(t *testing.T) {
+ fake := &fakeStore{}
+ var sent []*alertapi.Alert
+ e := newEvaluator(fake, func(a *alertapi.Alert) { sent = append(sent, a) })
+ rule := alerts.AlertConfig{
+ Name: "API up",
+ MetricName: monitor.ENDPOINT,
+ Endpoint: "https://api.example.com/health",
+ TriggerIntveral: 60,
+ Template: "{subject} expected {expected} got {actual}: {error}",
+ }
+ start := time.Unix(1700000000, 0)
+
+ // a check every 2 minutes, evaluated every 15s, so each is seen more than once
+ check := func(minutes int, statusCode int, errMsg string) {
+ t.Helper()
+ fake.check = store.EndpointCheck{Time: start.Add(time.Duration(minutes) * time.Minute), Name: rule.Name, StatusCode: statusCode, Error: errMsg}
+ for i := 0; i < 2; i++ {
+ if err := e.evaluate(context.Background(), &rule, ""); err != nil {
+ t.Fatal(err)
+ }
+ }
+ }
+
+ // nothing checked yet
+ if err := e.evaluate(context.Background(), &rule, ""); err != nil {
+ t.Fatal(err)
+ }
+ check(0, 200, "")
+ check(2, 0, "connection refused")
+ if len(sent) != 0 {
+ t.Fatalf("expected no alert after one failed check, got %+v", sent)
+ }
+ check(4, 0, "connection refused")
+ if len(sent) != 1 || sent[0].Status != int32(alertstatus.Critical) || sent[0].LogId != 1 || sent[0].ServerName != rule.Endpoint {
+ t.Fatalf("expected a critical alert for the endpoint, got %+v", sent)
+ }
+ if sent[0].Content != "[Critical] endpoint check failed on https://api.example.com/health expected 200 got 0: connection refused" {
+ t.Errorf("unexpected message %q", sent[0].Content)
+ }
+ if len(fake.alerts) != 1 || fake.alerts[0].Host != "" || fake.alerts[0].Target != rule.Endpoint {
+ t.Errorf("expected an alert without a host on the URL, got %+v", fake.alerts)
+ }
+
+ check(6, 200, "")
+ check(8, 200, "")
+ if len(sent) != 2 || !sent[1].Resolved || sent[1].LogId != 1 || fake.alerts[0].ResolvedAt == nil {
+ t.Errorf("expected the alert to resolve, got %+v", sent)
+ }
+}
diff --git a/collector/main.go b/collector/main.go
index 4633690..56ba61c 100644
--- a/collector/main.go
+++ b/collector/main.go
@@ -18,6 +18,7 @@ import (
"github.com/dhamith93/SyMon/internal/logger"
"github.com/dhamith93/SyMon/internal/store"
"github.com/dhamith93/SyMon/internal/transport"
+ "github.com/dhamith93/SyMon/internal/version"
)
func main() {
@@ -30,7 +31,12 @@ func main() {
tokenUses := flag.Int("uses", 1, "With -enroll-token: how many hosts the token can enroll")
tokenTTL := flag.Duration("ttl", time.Hour, "With -enroll-token: how long the token is valid")
envFile := flag.String("env", config.DefaultEnvFile("collector"), "Settings file with KEY=value lines, loaded if it exists")
+ versionPtr := flag.Bool("version", false, "Print the version and exit")
flag.Parse()
+ if *versionPtr {
+ fmt.Println("SyMon collector " + version.String())
+ return
+ }
if err := config.LoadEnvFile(*envFile); err != nil {
log.Fatal(err)
}
@@ -93,6 +99,7 @@ func main() {
log.Fatal(err)
}
api.RegisterMonitorDataServiceServer(grpcServer, &api.Server{Store: st})
+ logger.Log("info", "collector "+version.String()+" started on :"+config.Port)
if err := grpcServer.Serve(lis); err != nil {
log.Fatalf("failed to serve: %s", err)
}
diff --git a/docs/install.md b/docs/install.md
index 3d7b420..3e068a1 100644
--- a/docs/install.md
+++ b/docs/install.md
@@ -216,12 +216,14 @@ A rule looks like this:
| `disk_forecast` | days until the disk is full, at its growth over the last week | `Disk`, the device |
| `services` | a service from the service list. `Op` `inactive` alerts when it stops, `active` when it runs | `Service`, the name from the service list |
| `ping` | host silent for longer than `TriggerIntveral` seconds | |
-| `endpoint` | an HTTP check from the collector | `Endpoint`, `Method`, `ExpectedHTTPCode`, `POSTBody`, `POSTContentType` |
+| `endpoint` | an HTTP check from the collector, no `Servers` | `Endpoint`, `Method`, `ExpectedHTTPCode`, `POSTBody`, `POSTContentType`, `CustomCACert` |
| any name, with `"IsCustom": true` | a custom metric | |
-`Op` is one of `>`, `<`, `>=`, `<=`, `==` or `!=`. A value has to stay past a threshold for `TriggerIntveral` seconds before the alert opens, and back to normal for as long before it resolves. Endpoint checks need `SYMON_ENABLE_ENDPOINT_MONITORING=true` on the collector.
+`Op` is one of `>`, `<`, `>=`, `<=`, `==` or `!=`. A value has to stay past a threshold for `TriggerIntveral` seconds before the alert opens, and back to normal for as long before it resolves.
-`disk_forecast` rules use `Op` `<`, for example a warning under 14 days and critical under 3. A forecast needs a day of history and steady growth, so a disk that fills and empties, like one with rotating logs, gets none. A disk that is not filling up counts as 365 days. The dashboard shows the forecast in the host's disk table, and on the hosts page when a disk fills up within 30 days.
+`endpoint` rules need `SYMON_ENABLE_ENDPOINT_MONITORING=true` on the collector, which then requests each `Endpoint` every `SYMON_ENDPOINT_CHECK_INTERVAL` seconds (60 by default) and waits up to 30 seconds for an answer. A check passes when the status code is `ExpectedHTTPCode`, 200 if left out. The alert opens once checks have failed for `TriggerIntveral` seconds and resolves once they have passed for as long. Endpoint alerts belong to no host, and every check is kept as long as the 1 hour averages, for the response time and uptime charts on the dashboard's Endpoints page.
+
+`disk_forecast` rules use `Op` `<`, for example a warning under 14 days and critical under 3. A forecast needs a day of history and steady growth, so a disk that fills and empties, like one with rotating logs, gets none. A disk that is not filling up counts as 365 days. The dashboard shows the forecast in the host's disk table and on a chart of where each filling disk is headed, and on the hosts page when a disk fills up within 30 days.
The collector reads the rules when it starts, so restart it after editing them.
@@ -259,7 +261,7 @@ scrape_configs:
- targets: ["symon.example.lan:8080"]
```
-Every value has a `host` label. Disks, interfaces, sensors, services and containers have their own labels too. `symon_up` is 0 for a host that stopped reporting, and its other values are left out until it reports again. Like the rest of the dashboard, `/metrics` has no login, so keep it behind the same reverse proxy or firewall.
+Every value has a `host` label. Disks, interfaces, sensors, services and containers have their own labels too. Custom metrics sent in the last 2 days show as `symon_custom_metric` with `name` and `unit` labels, and `symon_custom_metric_timestamp_seconds` says when each was sent, so you can alert when a job stops reporting. `symon_up` is 0 for a host that stopped reporting, and its other values are left out until it reports again. Like the rest of the dashboard, `/metrics` has no login, so keep it behind the same reverse proxy or firewall. To switch it off, add `SYMON_CLIENT_METRICS_ENABLED=false` to `/etc/symon/client.env` and restart the dashboard.
## Upgrades
@@ -276,6 +278,8 @@ sudo systemctl start symon_collector symon_client
The collector updates the database schema by itself when it starts. Back up the database first (see below) if you want a way back, because schema changes are not undone by going back to an older build.
+**Checking versions.** Every binary prints its version with `-version`, for example `/opt/symon/collector_linux_x86_64/collector_linux_x86_64 -version`, and logs it when it starts. The dashboard footer shows its own version, and the collector's too when they differ. Each host's page shows the version of its agent, so you can see which hosts still need the upgrade below.
+
**Hosts.** Run the install command again on each host. No token is needed: the script sees the host is already enrolled, replaces the agent with the build the dashboard now serves, and keeps the key and the settings.
```sh
diff --git a/internal/alertapi/alertapi.go b/internal/alertapi/alertapi.go
index 84205cd..270a762 100644
--- a/internal/alertapi/alertapi.go
+++ b/internal/alertapi/alertapi.go
@@ -23,7 +23,7 @@ func (s *Server) HandleAlerts(ctx context.Context, in *Alert) (*Response, error)
sendEmail := in.Email
sendSlack := in.Slack
- if in.MetricName == monitor.DISKS {
+ if in.MetricName == monitor.DISKS || in.MetricName == monitor.DISK_FORECAST {
metricName = in.Disk
}
if in.MetricName == monitor.SERVICES {
diff --git a/internal/alerts/alert.go b/internal/alerts/alert.go
index ca0893f..b7cee71 100644
--- a/internal/alerts/alert.go
+++ b/internal/alerts/alert.go
@@ -5,9 +5,6 @@ import (
"os"
)
-const ENDPOINT_METHOD_GET string = "GET"
-const ENDPOINT_METHOD_POST string = "POST"
-
type AlertConfig struct {
Name string
Description string
diff --git a/internal/api/api.go b/internal/api/api.go
index 8b3cf70..594a114 100644
--- a/internal/api/api.go
+++ b/internal/api/api.go
@@ -11,6 +11,7 @@ import (
"github.com/dhamith93/SyMon/internal/logger"
"github.com/dhamith93/SyMon/internal/monitor"
"github.com/dhamith93/SyMon/internal/store"
+ "github.com/dhamith93/SyMon/internal/version"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
)
@@ -226,6 +227,10 @@ func (s *Server) Snapshots(ctx context.Context, in *Void) (*SnapshotList, error)
if err != nil {
return nil, toStatus(err)
}
+ customValues, err := s.Store.LatestCustomValues(ctx)
+ if err != nil {
+ return nil, toStatus(err)
+ }
now := time.Now()
list := &SnapshotList{}
for _, latest := range hosts {
@@ -237,6 +242,15 @@ func (s *Server) Snapshots(ctx context.Context, in *Void) (*SnapshotList, error)
Up: isUp(latest.LastSeen, now),
})
}
+ for _, custom := range customValues {
+ list.CustomMetrics = append(list.CustomMetrics, &CustomValue{
+ Host: custom.Host,
+ Name: custom.Name,
+ Unit: custom.Unit,
+ Value: custom.Value,
+ Time: custom.Time.Unix(),
+ })
+ }
return list, nil
}
@@ -254,7 +268,11 @@ func (s *Server) QuerySeries(ctx context.Context, in *SeriesRequest) (*SeriesRes
return nil, toStatus(err)
}
- response := &SeriesResponse{Metric: in.Metric, Source: result.Source, StepSeconds: int64(result.Step.Seconds())}
+ return seriesResponse(in.Metric, result), nil
+}
+
+func seriesResponse(metric string, result store.SeriesResult) *SeriesResponse {
+ response := &SeriesResponse{Metric: metric, Source: result.Source, StepSeconds: int64(result.Step.Seconds())}
for _, series := range result.Series {
points := make([]*Point, 0, len(series.Points))
for _, point := range series.Points {
@@ -262,7 +280,48 @@ func (s *Server) QuerySeries(ctx context.Context, in *SeriesRequest) (*SeriesRes
}
response.Series = append(response.Series, &Series{Label: series.Label, Points: points})
}
- return response, nil
+ return response
+}
+
+func (s *Server) Endpoints(ctx context.Context, in *EndpointsRequest) (*EndpointList, error) {
+ summaries, err := s.Store.Endpoints(ctx, time.Unix(in.From, 0), time.Unix(in.To, 0))
+ if err != nil {
+ return nil, toStatus(err)
+ }
+ list := &EndpointList{}
+ for _, summary := range summaries {
+ latest := summary.Latest
+ list.Endpoints = append(list.Endpoints, &EndpointStatus{
+ Name: latest.Name,
+ Url: latest.URL,
+ Method: latest.Method,
+ Time: latest.Time.Unix(),
+ Ok: latest.OK,
+ StatusCode: int32(latest.StatusCode),
+ LatencyMs: milliseconds(latest.Latency),
+ Error: latest.Error,
+ Checks: int32(summary.Checks),
+ UptimePct: summary.UptimePct,
+ AvgLatencyMs: milliseconds(summary.AvgLatency),
+ })
+ }
+ return list, nil
+}
+
+func (s *Server) EndpointSeries(ctx context.Context, in *EndpointSeriesRequest) (*SeriesResponse, error) {
+ result, err := s.Store.EndpointSeries(ctx, in.Name, in.Metric, time.Unix(in.From, 0), time.Unix(in.To, 0), int(in.MaxPoints))
+ if err != nil {
+ return nil, toStatus(err)
+ }
+ return seriesResponse(in.Metric, result), nil
+}
+
+func (s *Server) Version(ctx context.Context, in *Void) (*Message, error) {
+ return &Message{Body: version.String()}, nil
+}
+
+func milliseconds(d time.Duration) float64 {
+ return float64(d) / float64(time.Millisecond)
}
func (s *Server) Processes(ctx context.Context, in *ProcessesRequest) (*ProcessesResponse, error) {
diff --git a/internal/api/api.pb.go b/internal/api/api.pb.go
index 7ff53b4..a48af09 100644
--- a/internal/api/api.pb.go
+++ b/internal/api/api.pb.go
@@ -1628,16 +1628,95 @@ func (x *ProcessUsageList) GetProcesses() []*ProcessUsage {
return nil
}
-type SnapshotList struct {
+// The newest value of one of a host's custom metrics
+type CustomValue struct {
state protoimpl.MessageState `protogen:"open.v1"`
- Hosts []*HostSnapshot `protobuf:"bytes,1,rep,name=hosts,proto3" json:"hosts,omitempty"`
+ Host string `protobuf:"bytes,1,opt,name=host,proto3" json:"host,omitempty"`
+ Name string `protobuf:"bytes,2,opt,name=name,proto3" json:"name,omitempty"`
+ Unit string `protobuf:"bytes,3,opt,name=unit,proto3" json:"unit,omitempty"`
+ Value float64 `protobuf:"fixed64,4,opt,name=value,proto3" json:"value,omitempty"`
+ Time int64 `protobuf:"varint,5,opt,name=time,proto3" json:"time,omitempty"`
+ unknownFields protoimpl.UnknownFields
+ sizeCache protoimpl.SizeCache
+}
+
+func (x *CustomValue) Reset() {
+ *x = CustomValue{}
+ mi := &file_api_api_proto_msgTypes[25]
+ ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+ ms.StoreMessageInfo(mi)
+}
+
+func (x *CustomValue) String() string {
+ return protoimpl.X.MessageStringOf(x)
+}
+
+func (*CustomValue) ProtoMessage() {}
+
+func (x *CustomValue) ProtoReflect() protoreflect.Message {
+ mi := &file_api_api_proto_msgTypes[25]
+ if x != nil {
+ ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+ if ms.LoadMessageInfo() == nil {
+ ms.StoreMessageInfo(mi)
+ }
+ return ms
+ }
+ return mi.MessageOf(x)
+}
+
+// Deprecated: Use CustomValue.ProtoReflect.Descriptor instead.
+func (*CustomValue) Descriptor() ([]byte, []int) {
+ return file_api_api_proto_rawDescGZIP(), []int{25}
+}
+
+func (x *CustomValue) GetHost() string {
+ if x != nil {
+ return x.Host
+ }
+ return ""
+}
+
+func (x *CustomValue) GetName() string {
+ if x != nil {
+ return x.Name
+ }
+ return ""
+}
+
+func (x *CustomValue) GetUnit() string {
+ if x != nil {
+ return x.Unit
+ }
+ return ""
+}
+
+func (x *CustomValue) GetValue() float64 {
+ if x != nil {
+ return x.Value
+ }
+ return 0
+}
+
+func (x *CustomValue) GetTime() int64 {
+ if x != nil {
+ return x.Time
+ }
+ return 0
+}
+
+type SnapshotList struct {
+ state protoimpl.MessageState `protogen:"open.v1"`
+ Hosts []*HostSnapshot `protobuf:"bytes,1,rep,name=hosts,proto3" json:"hosts,omitempty"`
+ // custom metrics sent in the last 2 days
+ CustomMetrics []*CustomValue `protobuf:"bytes,2,rep,name=customMetrics,proto3" json:"customMetrics,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *SnapshotList) Reset() {
*x = SnapshotList{}
- mi := &file_api_api_proto_msgTypes[25]
+ mi := &file_api_api_proto_msgTypes[26]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@@ -1649,7 +1728,7 @@ func (x *SnapshotList) String() string {
func (*SnapshotList) ProtoMessage() {}
func (x *SnapshotList) ProtoReflect() protoreflect.Message {
- mi := &file_api_api_proto_msgTypes[25]
+ mi := &file_api_api_proto_msgTypes[26]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@@ -1662,7 +1741,7 @@ func (x *SnapshotList) ProtoReflect() protoreflect.Message {
// Deprecated: Use SnapshotList.ProtoReflect.Descriptor instead.
func (*SnapshotList) Descriptor() ([]byte, []int) {
- return file_api_api_proto_rawDescGZIP(), []int{25}
+ return file_api_api_proto_rawDescGZIP(), []int{26}
}
func (x *SnapshotList) GetHosts() []*HostSnapshot {
@@ -1672,6 +1751,314 @@ func (x *SnapshotList) GetHosts() []*HostSnapshot {
return nil
}
+func (x *SnapshotList) GetCustomMetrics() []*CustomValue {
+ if x != nil {
+ return x.CustomMetrics
+ }
+ return nil
+}
+
+type EndpointsRequest struct {
+ state protoimpl.MessageState `protogen:"open.v1"`
+ From int64 `protobuf:"varint,1,opt,name=from,proto3" json:"from,omitempty"`
+ To int64 `protobuf:"varint,2,opt,name=to,proto3" json:"to,omitempty"`
+ unknownFields protoimpl.UnknownFields
+ sizeCache protoimpl.SizeCache
+}
+
+func (x *EndpointsRequest) Reset() {
+ *x = EndpointsRequest{}
+ mi := &file_api_api_proto_msgTypes[27]
+ ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+ ms.StoreMessageInfo(mi)
+}
+
+func (x *EndpointsRequest) String() string {
+ return protoimpl.X.MessageStringOf(x)
+}
+
+func (*EndpointsRequest) ProtoMessage() {}
+
+func (x *EndpointsRequest) ProtoReflect() protoreflect.Message {
+ mi := &file_api_api_proto_msgTypes[27]
+ if x != nil {
+ ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+ if ms.LoadMessageInfo() == nil {
+ ms.StoreMessageInfo(mi)
+ }
+ return ms
+ }
+ return mi.MessageOf(x)
+}
+
+// Deprecated: Use EndpointsRequest.ProtoReflect.Descriptor instead.
+func (*EndpointsRequest) Descriptor() ([]byte, []int) {
+ return file_api_api_proto_rawDescGZIP(), []int{27}
+}
+
+func (x *EndpointsRequest) GetFrom() int64 {
+ if x != nil {
+ return x.From
+ }
+ return 0
+}
+
+func (x *EndpointsRequest) GetTo() int64 {
+ if x != nil {
+ return x.To
+ }
+ return 0
+}
+
+// An endpoint's newest check up to the end of the range, and how it did
+// over the range. name is the endpoint rule's name.
+type EndpointStatus struct {
+ state protoimpl.MessageState `protogen:"open.v1"`
+ Name string `protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"`
+ Url string `protobuf:"bytes,2,opt,name=url,proto3" json:"url,omitempty"`
+ Method string `protobuf:"bytes,3,opt,name=method,proto3" json:"method,omitempty"`
+ Time int64 `protobuf:"varint,4,opt,name=time,proto3" json:"time,omitempty"`
+ Ok bool `protobuf:"varint,5,opt,name=ok,proto3" json:"ok,omitempty"`
+ // 0 when there was no response
+ StatusCode int32 `protobuf:"varint,6,opt,name=statusCode,proto3" json:"statusCode,omitempty"`
+ LatencyMs float64 `protobuf:"fixed64,7,opt,name=latencyMs,proto3" json:"latencyMs,omitempty"`
+ Error string `protobuf:"bytes,8,opt,name=error,proto3" json:"error,omitempty"`
+ Checks int32 `protobuf:"varint,9,opt,name=checks,proto3" json:"checks,omitempty"`
+ UptimePct float64 `protobuf:"fixed64,10,opt,name=uptimePct,proto3" json:"uptimePct,omitempty"`
+ // over the checks that got a response
+ AvgLatencyMs float64 `protobuf:"fixed64,11,opt,name=avgLatencyMs,proto3" json:"avgLatencyMs,omitempty"`
+ unknownFields protoimpl.UnknownFields
+ sizeCache protoimpl.SizeCache
+}
+
+func (x *EndpointStatus) Reset() {
+ *x = EndpointStatus{}
+ mi := &file_api_api_proto_msgTypes[28]
+ ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+ ms.StoreMessageInfo(mi)
+}
+
+func (x *EndpointStatus) String() string {
+ return protoimpl.X.MessageStringOf(x)
+}
+
+func (*EndpointStatus) ProtoMessage() {}
+
+func (x *EndpointStatus) ProtoReflect() protoreflect.Message {
+ mi := &file_api_api_proto_msgTypes[28]
+ if x != nil {
+ ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+ if ms.LoadMessageInfo() == nil {
+ ms.StoreMessageInfo(mi)
+ }
+ return ms
+ }
+ return mi.MessageOf(x)
+}
+
+// Deprecated: Use EndpointStatus.ProtoReflect.Descriptor instead.
+func (*EndpointStatus) Descriptor() ([]byte, []int) {
+ return file_api_api_proto_rawDescGZIP(), []int{28}
+}
+
+func (x *EndpointStatus) GetName() string {
+ if x != nil {
+ return x.Name
+ }
+ return ""
+}
+
+func (x *EndpointStatus) GetUrl() string {
+ if x != nil {
+ return x.Url
+ }
+ return ""
+}
+
+func (x *EndpointStatus) GetMethod() string {
+ if x != nil {
+ return x.Method
+ }
+ return ""
+}
+
+func (x *EndpointStatus) GetTime() int64 {
+ if x != nil {
+ return x.Time
+ }
+ return 0
+}
+
+func (x *EndpointStatus) GetOk() bool {
+ if x != nil {
+ return x.Ok
+ }
+ return false
+}
+
+func (x *EndpointStatus) GetStatusCode() int32 {
+ if x != nil {
+ return x.StatusCode
+ }
+ return 0
+}
+
+func (x *EndpointStatus) GetLatencyMs() float64 {
+ if x != nil {
+ return x.LatencyMs
+ }
+ return 0
+}
+
+func (x *EndpointStatus) GetError() string {
+ if x != nil {
+ return x.Error
+ }
+ return ""
+}
+
+func (x *EndpointStatus) GetChecks() int32 {
+ if x != nil {
+ return x.Checks
+ }
+ return 0
+}
+
+func (x *EndpointStatus) GetUptimePct() float64 {
+ if x != nil {
+ return x.UptimePct
+ }
+ return 0
+}
+
+func (x *EndpointStatus) GetAvgLatencyMs() float64 {
+ if x != nil {
+ return x.AvgLatencyMs
+ }
+ return 0
+}
+
+type EndpointList struct {
+ state protoimpl.MessageState `protogen:"open.v1"`
+ Endpoints []*EndpointStatus `protobuf:"bytes,1,rep,name=endpoints,proto3" json:"endpoints,omitempty"`
+ unknownFields protoimpl.UnknownFields
+ sizeCache protoimpl.SizeCache
+}
+
+func (x *EndpointList) Reset() {
+ *x = EndpointList{}
+ mi := &file_api_api_proto_msgTypes[29]
+ ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+ ms.StoreMessageInfo(mi)
+}
+
+func (x *EndpointList) String() string {
+ return protoimpl.X.MessageStringOf(x)
+}
+
+func (*EndpointList) ProtoMessage() {}
+
+func (x *EndpointList) ProtoReflect() protoreflect.Message {
+ mi := &file_api_api_proto_msgTypes[29]
+ if x != nil {
+ ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+ if ms.LoadMessageInfo() == nil {
+ ms.StoreMessageInfo(mi)
+ }
+ return ms
+ }
+ return mi.MessageOf(x)
+}
+
+// Deprecated: Use EndpointList.ProtoReflect.Descriptor instead.
+func (*EndpointList) Descriptor() ([]byte, []int) {
+ return file_api_api_proto_rawDescGZIP(), []int{29}
+}
+
+func (x *EndpointList) GetEndpoints() []*EndpointStatus {
+ if x != nil {
+ return x.Endpoints
+ }
+ return nil
+}
+
+type EndpointSeriesRequest struct {
+ state protoimpl.MessageState `protogen:"open.v1"`
+ Name string `protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"`
+ // latency or availability
+ Metric string `protobuf:"bytes,2,opt,name=metric,proto3" json:"metric,omitempty"`
+ From int64 `protobuf:"varint,3,opt,name=from,proto3" json:"from,omitempty"`
+ To int64 `protobuf:"varint,4,opt,name=to,proto3" json:"to,omitempty"`
+ MaxPoints int32 `protobuf:"varint,5,opt,name=maxPoints,proto3" json:"maxPoints,omitempty"`
+ unknownFields protoimpl.UnknownFields
+ sizeCache protoimpl.SizeCache
+}
+
+func (x *EndpointSeriesRequest) Reset() {
+ *x = EndpointSeriesRequest{}
+ mi := &file_api_api_proto_msgTypes[30]
+ ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+ ms.StoreMessageInfo(mi)
+}
+
+func (x *EndpointSeriesRequest) String() string {
+ return protoimpl.X.MessageStringOf(x)
+}
+
+func (*EndpointSeriesRequest) ProtoMessage() {}
+
+func (x *EndpointSeriesRequest) ProtoReflect() protoreflect.Message {
+ mi := &file_api_api_proto_msgTypes[30]
+ if x != nil {
+ ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+ if ms.LoadMessageInfo() == nil {
+ ms.StoreMessageInfo(mi)
+ }
+ return ms
+ }
+ return mi.MessageOf(x)
+}
+
+// Deprecated: Use EndpointSeriesRequest.ProtoReflect.Descriptor instead.
+func (*EndpointSeriesRequest) Descriptor() ([]byte, []int) {
+ return file_api_api_proto_rawDescGZIP(), []int{30}
+}
+
+func (x *EndpointSeriesRequest) GetName() string {
+ if x != nil {
+ return x.Name
+ }
+ return ""
+}
+
+func (x *EndpointSeriesRequest) GetMetric() string {
+ if x != nil {
+ return x.Metric
+ }
+ return ""
+}
+
+func (x *EndpointSeriesRequest) GetFrom() int64 {
+ if x != nil {
+ return x.From
+ }
+ return 0
+}
+
+func (x *EndpointSeriesRequest) GetTo() int64 {
+ if x != nil {
+ return x.To
+ }
+ return 0
+}
+
+func (x *EndpointSeriesRequest) GetMaxPoints() int32 {
+ if x != nil {
+ return x.MaxPoints
+ }
+ return 0
+}
+
var File_api_api_proto protoreflect.FileDescriptor
const file_api_api_proto_rawDesc = "" +
@@ -1803,9 +2190,42 @@ const file_api_api_proto_rawDesc = "" +
"\x10ProcessUsageList\x12\x1c\n" +
"\tsnapshots\x18\x01 \x01(\x05R\tsnapshots\x12\x1c\n" +
"\tfirstTime\x18\x02 \x01(\x03R\tfirstTime\x12/\n" +
- "\tprocesses\x18\x03 \x03(\v2\x11.api.ProcessUsageR\tprocesses\"7\n" +
+ "\tprocesses\x18\x03 \x03(\v2\x11.api.ProcessUsageR\tprocesses\"s\n" +
+ "\vCustomValue\x12\x12\n" +
+ "\x04host\x18\x01 \x01(\tR\x04host\x12\x12\n" +
+ "\x04name\x18\x02 \x01(\tR\x04name\x12\x12\n" +
+ "\x04unit\x18\x03 \x01(\tR\x04unit\x12\x14\n" +
+ "\x05value\x18\x04 \x01(\x01R\x05value\x12\x12\n" +
+ "\x04time\x18\x05 \x01(\x03R\x04time\"o\n" +
"\fSnapshotList\x12'\n" +
- "\x05hosts\x18\x01 \x03(\v2\x11.api.HostSnapshotR\x05hosts2\x82\x06\n" +
+ "\x05hosts\x18\x01 \x03(\v2\x11.api.HostSnapshotR\x05hosts\x126\n" +
+ "\rcustomMetrics\x18\x02 \x03(\v2\x10.api.CustomValueR\rcustomMetrics\"6\n" +
+ "\x10EndpointsRequest\x12\x12\n" +
+ "\x04from\x18\x01 \x01(\x03R\x04from\x12\x0e\n" +
+ "\x02to\x18\x02 \x01(\x03R\x02to\"\xa0\x02\n" +
+ "\x0eEndpointStatus\x12\x12\n" +
+ "\x04name\x18\x01 \x01(\tR\x04name\x12\x10\n" +
+ "\x03url\x18\x02 \x01(\tR\x03url\x12\x16\n" +
+ "\x06method\x18\x03 \x01(\tR\x06method\x12\x12\n" +
+ "\x04time\x18\x04 \x01(\x03R\x04time\x12\x0e\n" +
+ "\x02ok\x18\x05 \x01(\bR\x02ok\x12\x1e\n" +
+ "\n" +
+ "statusCode\x18\x06 \x01(\x05R\n" +
+ "statusCode\x12\x1c\n" +
+ "\tlatencyMs\x18\a \x01(\x01R\tlatencyMs\x12\x14\n" +
+ "\x05error\x18\b \x01(\tR\x05error\x12\x16\n" +
+ "\x06checks\x18\t \x01(\x05R\x06checks\x12\x1c\n" +
+ "\tuptimePct\x18\n" +
+ " \x01(\x01R\tuptimePct\x12\"\n" +
+ "\favgLatencyMs\x18\v \x01(\x01R\favgLatencyMs\"A\n" +
+ "\fEndpointList\x121\n" +
+ "\tendpoints\x18\x01 \x03(\v2\x13.api.EndpointStatusR\tendpoints\"\x85\x01\n" +
+ "\x15EndpointSeriesRequest\x12\x12\n" +
+ "\x04name\x18\x01 \x01(\tR\x04name\x12\x16\n" +
+ "\x06metric\x18\x02 \x01(\tR\x06metric\x12\x12\n" +
+ "\x04from\x18\x03 \x01(\x03R\x04from\x12\x0e\n" +
+ "\x02to\x18\x04 \x01(\x03R\x02to\x12\x1c\n" +
+ "\tmaxPoints\x18\x05 \x01(\x05R\tmaxPoints2\xa6\a\n" +
"\x12MonitorDataService\x123\n" +
"\x06Enroll\x12\x12.api.EnrollRequest\x1a\x13.api.EnrollResponse\"\x00\x12-\n" +
"\n" +
@@ -1821,7 +2241,10 @@ const file_api_api_proto_rawDesc = "" +
"\x06Alerts\x12\x12.api.AlertsRequest\x1a\x0e.api.AlertList\"\x00\x12:\n" +
"\rDiskForecasts\x12\x10.api.HostRequest\x1a\x15.api.DiskForecastList\"\x00\x12A\n" +
"\fProcessUsage\x12\x18.api.ProcessUsageRequest\x1a\x15.api.ProcessUsageList\"\x00\x12+\n" +
- "\tSnapshots\x12\t.api.Void\x1a\x11.api.SnapshotList\"\x00B)Z'github.com/dhamith93/SyMon/internal/apib\x06proto3"
+ "\tSnapshots\x12\t.api.Void\x1a\x11.api.SnapshotList\"\x00\x127\n" +
+ "\tEndpoints\x12\x15.api.EndpointsRequest\x1a\x11.api.EndpointList\"\x00\x12C\n" +
+ "\x0eEndpointSeries\x12\x1a.api.EndpointSeriesRequest\x1a\x13.api.SeriesResponse\"\x00\x12$\n" +
+ "\aVersion\x12\t.api.Void\x1a\f.api.Message\"\x00B)Z'github.com/dhamith93/SyMon/internal/apib\x06proto3"
var (
file_api_api_proto_rawDescOnce sync.Once
@@ -1835,34 +2258,39 @@ func file_api_api_proto_rawDescGZIP() []byte {
return file_api_api_proto_rawDescData
}
-var file_api_api_proto_msgTypes = make([]protoimpl.MessageInfo, 26)
+var file_api_api_proto_msgTypes = make([]protoimpl.MessageInfo, 31)
var file_api_api_proto_goTypes = []any{
- (*Void)(nil), // 0: api.Void
- (*Message)(nil), // 1: api.Message
- (*ServerInfo)(nil), // 2: api.ServerInfo
- (*MonitorData)(nil), // 3: api.MonitorData
- (*EnrollRequest)(nil), // 4: api.EnrollRequest
- (*EnrollResponse)(nil), // 5: api.EnrollResponse
- (*HostSummary)(nil), // 6: api.HostSummary
- (*FleetSummary)(nil), // 7: api.FleetSummary
- (*HostRequest)(nil), // 8: api.HostRequest
- (*HostSnapshot)(nil), // 9: api.HostSnapshot
- (*SeriesRequest)(nil), // 10: api.SeriesRequest
- (*Point)(nil), // 11: api.Point
- (*Series)(nil), // 12: api.Series
- (*SeriesResponse)(nil), // 13: api.SeriesResponse
- (*ProcessesRequest)(nil), // 14: api.ProcessesRequest
- (*ProcessesResponse)(nil), // 15: api.ProcessesResponse
- (*NameList)(nil), // 16: api.NameList
- (*AlertsRequest)(nil), // 17: api.AlertsRequest
- (*AlertRecord)(nil), // 18: api.AlertRecord
- (*AlertList)(nil), // 19: api.AlertList
- (*DiskForecast)(nil), // 20: api.DiskForecast
- (*DiskForecastList)(nil), // 21: api.DiskForecastList
- (*ProcessUsageRequest)(nil), // 22: api.ProcessUsageRequest
- (*ProcessUsage)(nil), // 23: api.ProcessUsage
- (*ProcessUsageList)(nil), // 24: api.ProcessUsageList
- (*SnapshotList)(nil), // 25: api.SnapshotList
+ (*Void)(nil), // 0: api.Void
+ (*Message)(nil), // 1: api.Message
+ (*ServerInfo)(nil), // 2: api.ServerInfo
+ (*MonitorData)(nil), // 3: api.MonitorData
+ (*EnrollRequest)(nil), // 4: api.EnrollRequest
+ (*EnrollResponse)(nil), // 5: api.EnrollResponse
+ (*HostSummary)(nil), // 6: api.HostSummary
+ (*FleetSummary)(nil), // 7: api.FleetSummary
+ (*HostRequest)(nil), // 8: api.HostRequest
+ (*HostSnapshot)(nil), // 9: api.HostSnapshot
+ (*SeriesRequest)(nil), // 10: api.SeriesRequest
+ (*Point)(nil), // 11: api.Point
+ (*Series)(nil), // 12: api.Series
+ (*SeriesResponse)(nil), // 13: api.SeriesResponse
+ (*ProcessesRequest)(nil), // 14: api.ProcessesRequest
+ (*ProcessesResponse)(nil), // 15: api.ProcessesResponse
+ (*NameList)(nil), // 16: api.NameList
+ (*AlertsRequest)(nil), // 17: api.AlertsRequest
+ (*AlertRecord)(nil), // 18: api.AlertRecord
+ (*AlertList)(nil), // 19: api.AlertList
+ (*DiskForecast)(nil), // 20: api.DiskForecast
+ (*DiskForecastList)(nil), // 21: api.DiskForecastList
+ (*ProcessUsageRequest)(nil), // 22: api.ProcessUsageRequest
+ (*ProcessUsage)(nil), // 23: api.ProcessUsage
+ (*ProcessUsageList)(nil), // 24: api.ProcessUsageList
+ (*CustomValue)(nil), // 25: api.CustomValue
+ (*SnapshotList)(nil), // 26: api.SnapshotList
+ (*EndpointsRequest)(nil), // 27: api.EndpointsRequest
+ (*EndpointStatus)(nil), // 28: api.EndpointStatus
+ (*EndpointList)(nil), // 29: api.EndpointList
+ (*EndpointSeriesRequest)(nil), // 30: api.EndpointSeriesRequest
}
var file_api_api_proto_depIdxs = []int32{
6, // 0: api.FleetSummary.hosts:type_name -> api.HostSummary
@@ -1872,39 +2300,47 @@ var file_api_api_proto_depIdxs = []int32{
20, // 4: api.DiskForecastList.disks:type_name -> api.DiskForecast
23, // 5: api.ProcessUsageList.processes:type_name -> api.ProcessUsage
9, // 6: api.SnapshotList.hosts:type_name -> api.HostSnapshot
- 4, // 7: api.MonitorDataService.Enroll:input_type -> api.EnrollRequest
- 2, // 8: api.MonitorDataService.HandlePing:input_type -> api.ServerInfo
- 2, // 9: api.MonitorDataService.InitAgent:input_type -> api.ServerInfo
- 3, // 10: api.MonitorDataService.HandleMonitorData:input_type -> api.MonitorData
- 3, // 11: api.MonitorDataService.HandleCustomMonitorData:input_type -> api.MonitorData
- 0, // 12: api.MonitorDataService.Fleet:input_type -> api.Void
- 8, // 13: api.MonitorDataService.Snapshot:input_type -> api.HostRequest
- 10, // 14: api.MonitorDataService.QuerySeries:input_type -> api.SeriesRequest
- 14, // 15: api.MonitorDataService.Processes:input_type -> api.ProcessesRequest
- 8, // 16: api.MonitorDataService.CustomMetricNames:input_type -> api.HostRequest
- 17, // 17: api.MonitorDataService.Alerts:input_type -> api.AlertsRequest
- 8, // 18: api.MonitorDataService.DiskForecasts:input_type -> api.HostRequest
- 22, // 19: api.MonitorDataService.ProcessUsage:input_type -> api.ProcessUsageRequest
- 0, // 20: api.MonitorDataService.Snapshots:input_type -> api.Void
- 5, // 21: api.MonitorDataService.Enroll:output_type -> api.EnrollResponse
- 1, // 22: api.MonitorDataService.HandlePing:output_type -> api.Message
- 1, // 23: api.MonitorDataService.InitAgent:output_type -> api.Message
- 1, // 24: api.MonitorDataService.HandleMonitorData:output_type -> api.Message
- 1, // 25: api.MonitorDataService.HandleCustomMonitorData:output_type -> api.Message
- 7, // 26: api.MonitorDataService.Fleet:output_type -> api.FleetSummary
- 9, // 27: api.MonitorDataService.Snapshot:output_type -> api.HostSnapshot
- 13, // 28: api.MonitorDataService.QuerySeries:output_type -> api.SeriesResponse
- 15, // 29: api.MonitorDataService.Processes:output_type -> api.ProcessesResponse
- 16, // 30: api.MonitorDataService.CustomMetricNames:output_type -> api.NameList
- 19, // 31: api.MonitorDataService.Alerts:output_type -> api.AlertList
- 21, // 32: api.MonitorDataService.DiskForecasts:output_type -> api.DiskForecastList
- 24, // 33: api.MonitorDataService.ProcessUsage:output_type -> api.ProcessUsageList
- 25, // 34: api.MonitorDataService.Snapshots:output_type -> api.SnapshotList
- 21, // [21:35] is the sub-list for method output_type
- 7, // [7:21] is the sub-list for method input_type
- 7, // [7:7] is the sub-list for extension type_name
- 7, // [7:7] is the sub-list for extension extendee
- 0, // [0:7] is the sub-list for field type_name
+ 25, // 7: api.SnapshotList.customMetrics:type_name -> api.CustomValue
+ 28, // 8: api.EndpointList.endpoints:type_name -> api.EndpointStatus
+ 4, // 9: api.MonitorDataService.Enroll:input_type -> api.EnrollRequest
+ 2, // 10: api.MonitorDataService.HandlePing:input_type -> api.ServerInfo
+ 2, // 11: api.MonitorDataService.InitAgent:input_type -> api.ServerInfo
+ 3, // 12: api.MonitorDataService.HandleMonitorData:input_type -> api.MonitorData
+ 3, // 13: api.MonitorDataService.HandleCustomMonitorData:input_type -> api.MonitorData
+ 0, // 14: api.MonitorDataService.Fleet:input_type -> api.Void
+ 8, // 15: api.MonitorDataService.Snapshot:input_type -> api.HostRequest
+ 10, // 16: api.MonitorDataService.QuerySeries:input_type -> api.SeriesRequest
+ 14, // 17: api.MonitorDataService.Processes:input_type -> api.ProcessesRequest
+ 8, // 18: api.MonitorDataService.CustomMetricNames:input_type -> api.HostRequest
+ 17, // 19: api.MonitorDataService.Alerts:input_type -> api.AlertsRequest
+ 8, // 20: api.MonitorDataService.DiskForecasts:input_type -> api.HostRequest
+ 22, // 21: api.MonitorDataService.ProcessUsage:input_type -> api.ProcessUsageRequest
+ 0, // 22: api.MonitorDataService.Snapshots:input_type -> api.Void
+ 27, // 23: api.MonitorDataService.Endpoints:input_type -> api.EndpointsRequest
+ 30, // 24: api.MonitorDataService.EndpointSeries:input_type -> api.EndpointSeriesRequest
+ 0, // 25: api.MonitorDataService.Version:input_type -> api.Void
+ 5, // 26: api.MonitorDataService.Enroll:output_type -> api.EnrollResponse
+ 1, // 27: api.MonitorDataService.HandlePing:output_type -> api.Message
+ 1, // 28: api.MonitorDataService.InitAgent:output_type -> api.Message
+ 1, // 29: api.MonitorDataService.HandleMonitorData:output_type -> api.Message
+ 1, // 30: api.MonitorDataService.HandleCustomMonitorData:output_type -> api.Message
+ 7, // 31: api.MonitorDataService.Fleet:output_type -> api.FleetSummary
+ 9, // 32: api.MonitorDataService.Snapshot:output_type -> api.HostSnapshot
+ 13, // 33: api.MonitorDataService.QuerySeries:output_type -> api.SeriesResponse
+ 15, // 34: api.MonitorDataService.Processes:output_type -> api.ProcessesResponse
+ 16, // 35: api.MonitorDataService.CustomMetricNames:output_type -> api.NameList
+ 19, // 36: api.MonitorDataService.Alerts:output_type -> api.AlertList
+ 21, // 37: api.MonitorDataService.DiskForecasts:output_type -> api.DiskForecastList
+ 24, // 38: api.MonitorDataService.ProcessUsage:output_type -> api.ProcessUsageList
+ 26, // 39: api.MonitorDataService.Snapshots:output_type -> api.SnapshotList
+ 29, // 40: api.MonitorDataService.Endpoints:output_type -> api.EndpointList
+ 13, // 41: api.MonitorDataService.EndpointSeries:output_type -> api.SeriesResponse
+ 1, // 42: api.MonitorDataService.Version:output_type -> api.Message
+ 26, // [26:43] is the sub-list for method output_type
+ 9, // [9:26] is the sub-list for method input_type
+ 9, // [9:9] is the sub-list for extension type_name
+ 9, // [9:9] is the sub-list for extension extendee
+ 0, // [0:9] is the sub-list for field type_name
}
func init() { file_api_api_proto_init() }
@@ -1920,7 +2356,7 @@ func file_api_api_proto_init() {
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_api_api_proto_rawDesc), len(file_api_api_proto_rawDesc)),
NumEnums: 0,
- NumMessages: 26,
+ NumMessages: 31,
NumExtensions: 0,
NumServices: 1,
},
diff --git a/internal/api/api.proto b/internal/api/api.proto
index c8f283d..ae0a74c 100644
--- a/internal/api/api.proto
+++ b/internal/api/api.proto
@@ -186,8 +186,55 @@ message ProcessUsageList {
repeated ProcessUsage processes = 3;
}
+// The newest value of one of a host's custom metrics
+message CustomValue {
+ string host = 1;
+ string name = 2;
+ string unit = 3;
+ double value = 4;
+ int64 time = 5;
+}
+
message SnapshotList {
repeated HostSnapshot hosts = 1;
+ // custom metrics sent in the last 2 days
+ repeated CustomValue customMetrics = 2;
+}
+
+message EndpointsRequest {
+ int64 from = 1;
+ int64 to = 2;
+}
+
+// An endpoint's newest check up to the end of the range, and how it did
+// over the range. name is the endpoint rule's name.
+message EndpointStatus {
+ string name = 1;
+ string url = 2;
+ string method = 3;
+ int64 time = 4;
+ bool ok = 5;
+ // 0 when there was no response
+ int32 statusCode = 6;
+ double latencyMs = 7;
+ string error = 8;
+ int32 checks = 9;
+ double uptimePct = 10;
+ // over the checks that got a response
+ double avgLatencyMs = 11;
+}
+
+message EndpointList {
+ repeated EndpointStatus endpoints = 1;
+}
+
+message EndpointSeriesRequest {
+ string name = 1;
+ // latency or availability
+ string metric = 2;
+ int64 from = 3;
+ int64 to = 4;
+ int32 maxPoints = 5;
}
service MonitorDataService {
@@ -209,4 +256,8 @@ service MonitorDataService {
rpc ProcessUsage(ProcessUsageRequest) returns (ProcessUsageList) {}
// every host's latest snapshot, for the metrics endpoint
rpc Snapshots(Void) returns (SnapshotList) {}
+ rpc Endpoints(EndpointsRequest) returns (EndpointList) {}
+ rpc EndpointSeries(EndpointSeriesRequest) returns (SeriesResponse) {}
+ // the collector's build, in the body
+ rpc Version(Void) returns (Message) {}
}
diff --git a/internal/api/api_grpc.pb.go b/internal/api/api_grpc.pb.go
index 80b5f5f..a82f772 100644
--- a/internal/api/api_grpc.pb.go
+++ b/internal/api/api_grpc.pb.go
@@ -33,6 +33,9 @@ const (
MonitorDataService_DiskForecasts_FullMethodName = "/api.MonitorDataService/DiskForecasts"
MonitorDataService_ProcessUsage_FullMethodName = "/api.MonitorDataService/ProcessUsage"
MonitorDataService_Snapshots_FullMethodName = "/api.MonitorDataService/Snapshots"
+ MonitorDataService_Endpoints_FullMethodName = "/api.MonitorDataService/Endpoints"
+ MonitorDataService_EndpointSeries_FullMethodName = "/api.MonitorDataService/EndpointSeries"
+ MonitorDataService_Version_FullMethodName = "/api.MonitorDataService/Version"
)
// MonitorDataServiceClient is the client API for MonitorDataService service.
@@ -56,6 +59,10 @@ type MonitorDataServiceClient interface {
ProcessUsage(ctx context.Context, in *ProcessUsageRequest, opts ...grpc.CallOption) (*ProcessUsageList, error)
// every host's latest snapshot, for the metrics endpoint
Snapshots(ctx context.Context, in *Void, opts ...grpc.CallOption) (*SnapshotList, error)
+ Endpoints(ctx context.Context, in *EndpointsRequest, opts ...grpc.CallOption) (*EndpointList, error)
+ EndpointSeries(ctx context.Context, in *EndpointSeriesRequest, opts ...grpc.CallOption) (*SeriesResponse, error)
+ // the collector's build, in the body
+ Version(ctx context.Context, in *Void, opts ...grpc.CallOption) (*Message, error)
}
type monitorDataServiceClient struct {
@@ -206,6 +213,36 @@ func (c *monitorDataServiceClient) Snapshots(ctx context.Context, in *Void, opts
return out, nil
}
+func (c *monitorDataServiceClient) Endpoints(ctx context.Context, in *EndpointsRequest, opts ...grpc.CallOption) (*EndpointList, error) {
+ cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+ out := new(EndpointList)
+ err := c.cc.Invoke(ctx, MonitorDataService_Endpoints_FullMethodName, in, out, cOpts...)
+ if err != nil {
+ return nil, err
+ }
+ return out, nil
+}
+
+func (c *monitorDataServiceClient) EndpointSeries(ctx context.Context, in *EndpointSeriesRequest, opts ...grpc.CallOption) (*SeriesResponse, error) {
+ cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+ out := new(SeriesResponse)
+ err := c.cc.Invoke(ctx, MonitorDataService_EndpointSeries_FullMethodName, in, out, cOpts...)
+ if err != nil {
+ return nil, err
+ }
+ return out, nil
+}
+
+func (c *monitorDataServiceClient) Version(ctx context.Context, in *Void, opts ...grpc.CallOption) (*Message, error) {
+ cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+ out := new(Message)
+ err := c.cc.Invoke(ctx, MonitorDataService_Version_FullMethodName, in, out, cOpts...)
+ if err != nil {
+ return nil, err
+ }
+ return out, nil
+}
+
// MonitorDataServiceServer is the server API for MonitorDataService service.
// All implementations must embed UnimplementedMonitorDataServiceServer
// for forward compatibility.
@@ -227,6 +264,10 @@ type MonitorDataServiceServer interface {
ProcessUsage(context.Context, *ProcessUsageRequest) (*ProcessUsageList, error)
// every host's latest snapshot, for the metrics endpoint
Snapshots(context.Context, *Void) (*SnapshotList, error)
+ Endpoints(context.Context, *EndpointsRequest) (*EndpointList, error)
+ EndpointSeries(context.Context, *EndpointSeriesRequest) (*SeriesResponse, error)
+ // the collector's build, in the body
+ Version(context.Context, *Void) (*Message, error)
mustEmbedUnimplementedMonitorDataServiceServer()
}
@@ -279,6 +320,15 @@ func (UnimplementedMonitorDataServiceServer) ProcessUsage(context.Context, *Proc
func (UnimplementedMonitorDataServiceServer) Snapshots(context.Context, *Void) (*SnapshotList, error) {
return nil, status.Errorf(codes.Unimplemented, "method Snapshots not implemented")
}
+func (UnimplementedMonitorDataServiceServer) Endpoints(context.Context, *EndpointsRequest) (*EndpointList, error) {
+ return nil, status.Errorf(codes.Unimplemented, "method Endpoints not implemented")
+}
+func (UnimplementedMonitorDataServiceServer) EndpointSeries(context.Context, *EndpointSeriesRequest) (*SeriesResponse, error) {
+ return nil, status.Errorf(codes.Unimplemented, "method EndpointSeries not implemented")
+}
+func (UnimplementedMonitorDataServiceServer) Version(context.Context, *Void) (*Message, error) {
+ return nil, status.Errorf(codes.Unimplemented, "method Version not implemented")
+}
func (UnimplementedMonitorDataServiceServer) mustEmbedUnimplementedMonitorDataServiceServer() {}
func (UnimplementedMonitorDataServiceServer) testEmbeddedByValue() {}
@@ -552,6 +602,60 @@ func _MonitorDataService_Snapshots_Handler(srv interface{}, ctx context.Context,
return interceptor(ctx, in, info, handler)
}
+func _MonitorDataService_Endpoints_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+ in := new(EndpointsRequest)
+ if err := dec(in); err != nil {
+ return nil, err
+ }
+ if interceptor == nil {
+ return srv.(MonitorDataServiceServer).Endpoints(ctx, in)
+ }
+ info := &grpc.UnaryServerInfo{
+ Server: srv,
+ FullMethod: MonitorDataService_Endpoints_FullMethodName,
+ }
+ handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+ return srv.(MonitorDataServiceServer).Endpoints(ctx, req.(*EndpointsRequest))
+ }
+ return interceptor(ctx, in, info, handler)
+}
+
+func _MonitorDataService_EndpointSeries_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+ in := new(EndpointSeriesRequest)
+ if err := dec(in); err != nil {
+ return nil, err
+ }
+ if interceptor == nil {
+ return srv.(MonitorDataServiceServer).EndpointSeries(ctx, in)
+ }
+ info := &grpc.UnaryServerInfo{
+ Server: srv,
+ FullMethod: MonitorDataService_EndpointSeries_FullMethodName,
+ }
+ handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+ return srv.(MonitorDataServiceServer).EndpointSeries(ctx, req.(*EndpointSeriesRequest))
+ }
+ return interceptor(ctx, in, info, handler)
+}
+
+func _MonitorDataService_Version_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+ in := new(Void)
+ if err := dec(in); err != nil {
+ return nil, err
+ }
+ if interceptor == nil {
+ return srv.(MonitorDataServiceServer).Version(ctx, in)
+ }
+ info := &grpc.UnaryServerInfo{
+ Server: srv,
+ FullMethod: MonitorDataService_Version_FullMethodName,
+ }
+ handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+ return srv.(MonitorDataServiceServer).Version(ctx, req.(*Void))
+ }
+ return interceptor(ctx, in, info, handler)
+}
+
// MonitorDataService_ServiceDesc is the grpc.ServiceDesc for MonitorDataService service.
// It's only intended for direct use with grpc.RegisterService,
// and not to be introspected or modified (even as a copy)
@@ -615,6 +719,18 @@ var MonitorDataService_ServiceDesc = grpc.ServiceDesc{
MethodName: "Snapshots",
Handler: _MonitorDataService_Snapshots_Handler,
},
+ {
+ MethodName: "Endpoints",
+ Handler: _MonitorDataService_Endpoints_Handler,
+ },
+ {
+ MethodName: "EndpointSeries",
+ Handler: _MonitorDataService_EndpointSeries_Handler,
+ },
+ {
+ MethodName: "Version",
+ Handler: _MonitorDataService_Version_Handler,
+ },
},
Streams: []grpc.StreamDesc{},
Metadata: "api/api.proto",
diff --git a/internal/config/config_handler.go b/internal/config/config_handler.go
index 5bf018d..3ab33ce 100644
--- a/internal/config/config_handler.go
+++ b/internal/config/config_handler.go
@@ -50,6 +50,8 @@ type Client struct {
// collector. Empty means CollectorEndpoint, with localhost replaced by
// the dashboard's host name.
AgentCollectorEndpoint string
+ // MetricsEnabled serves /metrics for Prometheus, on unless set to false
+ MetricsEnabled bool
}
type AlertProcessor struct {
@@ -209,6 +211,7 @@ func GetClient() Client {
RefreshSeconds: refreshSeconds,
DownloadsDir: downloadsDir,
AgentCollectorEndpoint: os.Getenv("SYMON_CLIENT_AGENT_COLLECTOR_ENDPOINT"),
+ MetricsEnabled: strings.ToUpper(os.Getenv("SYMON_CLIENT_METRICS_ENABLED")) != "FALSE",
}
}
diff --git a/internal/config/config_handler_test.go b/internal/config/config_handler_test.go
index b346135..b388543 100644
--- a/internal/config/config_handler_test.go
+++ b/internal/config/config_handler_test.go
@@ -78,3 +78,12 @@ func TestAgentDefaults(t *testing.T) {
t.Errorf("unexpected default key path %q", agent.AgentKeyPath)
}
}
+
+func TestClientMetricsSetting(t *testing.T) {
+ for value, want := range map[string]bool{"": true, "true": true, "false": false, "FALSE": false} {
+ t.Setenv("SYMON_CLIENT_METRICS_ENABLED", value)
+ if got := config.GetClient().MetricsEnabled; got != want {
+ t.Errorf("SYMON_CLIENT_METRICS_ENABLED=%q: got %v, want %v", value, got, want)
+ }
+ }
+}
diff --git a/internal/monitor/monitor.go b/internal/monitor/monitor.go
index 8551e17..5a47101 100644
--- a/internal/monitor/monitor.go
+++ b/internal/monitor/monitor.go
@@ -9,6 +9,7 @@ import (
"github.com/dhamith93/SyMon/internal/config"
"github.com/dhamith93/SyMon/internal/logger"
+ "github.com/dhamith93/SyMon/internal/version"
"github.com/dhamith93/systats"
)
@@ -30,8 +31,12 @@ const (
CONTAINERS string = "containers"
)
-// DISK_FORECAST is an alert metric, the days until a disk is full
-const DISK_FORECAST string = "disk_forecast"
+// alert metrics that are not in a snapshot. DISK_FORECAST is the days until
+// a disk is full, ENDPOINT an HTTP check the collector runs.
+const (
+ DISK_FORECAST string = "disk_forecast"
+ ENDPOINT string = "endpoint"
+)
// optional collectors, SYMON_DISABLED_COLLECTORS can switch these off
const (
@@ -72,6 +77,10 @@ type MonitorData struct {
Pressure *Pressure `json:",omitempty"`
Temperatures []Temperature `json:",omitempty"`
Containers []Container `json:",omitempty"`
+
+ // AgentVersion is the build of the agent that sent the snapshot, empty
+ // from agents older than it
+ AgentVersion string `json:",omitempty"`
}
// Collector gathers a MonitorData snapshot on every tick. It keeps the
@@ -145,6 +154,8 @@ func (c *Collector) Collect(ctx context.Context) MonitorData {
Networks: c.networks(),
Services: c.services(ctx, unixTime),
Processes: c.processes(ctx),
+
+ AgentVersion: version.String(),
}
if c.enabled[CollectorDiskIO] {
diff --git a/internal/store/alerts.go b/internal/store/alerts.go
index fadb849..c510977 100644
--- a/internal/store/alerts.go
+++ b/internal/store/alerts.go
@@ -11,11 +11,13 @@ import (
)
type Alert struct {
- ID int64
+ ID int64
+ // Host is empty for endpoint alerts, which belong to no host
Host string
Rule string
Metric string
- // Target is the disk, service or custom metric name, empty otherwise
+ // Target is the disk, service, custom metric name or endpoint URL, empty
+ // otherwise
Target string
// Severity is 1 for warning and 2 for critical
Severity int
@@ -69,7 +71,7 @@ func (s *Store) LatestValue(ctx context.Context, host string, metric string, tar
return *value, at, nil
}
-const alertColumns = "a.id, h.name, a.rule, a.metric, a.target, a.severity, a.value, a.started_at, a.updated_at, a.resolved_at"
+const alertColumns = "a.id, coalesce(h.name, ''), a.rule, a.metric, a.target, a.severity, a.value, a.started_at, a.updated_at, a.resolved_at"
func scanAlert(row pgx.Row) (Alert, error) {
var alert Alert
@@ -83,11 +85,11 @@ func scanAlert(row pgx.Row) (Alert, error) {
}
// OpenAlert returns the unresolved alert for a host, rule and target, or
-// nil if there is none
+// nil if there is none. An empty host finds endpoint alerts.
func (s *Store) OpenAlert(ctx context.Context, host string, rule string, metric string, target string) (*Alert, error) {
row := s.pool.QueryRow(ctx, `SELECT `+alertColumns+`
- FROM alerts a JOIN hosts h ON h.id = a.host_id
- WHERE h.name = $1 AND a.rule = $2 AND a.metric = $3 AND a.target = $4 AND a.resolved_at IS NULL`,
+ FROM alerts a LEFT JOIN hosts h ON h.id = a.host_id
+ WHERE coalesce(h.name, '') = $1 AND a.rule = $2 AND a.metric = $3 AND a.target = $4 AND a.resolved_at IS NULL`,
host, rule, metric, target)
alert, err := scanAlert(row)
if errors.Is(err, pgx.ErrNoRows) {
@@ -101,12 +103,16 @@ func (s *Store) OpenAlert(ctx context.Context, host string, rule string, metric
// CreateAlert opens an alert and returns its id
func (s *Store) CreateAlert(ctx context.Context, alert Alert) (int64, error) {
- hostID, err := s.hostID(ctx, alert.Host)
- if err != nil {
- return 0, err
+ var hostID *int64
+ if alert.Host != "" {
+ id, err := s.hostID(ctx, alert.Host)
+ if err != nil {
+ return 0, err
+ }
+ hostID = &id
}
var id int64
- err = s.pool.QueryRow(ctx, `
+ err := s.pool.QueryRow(ctx, `
INSERT INTO alerts (host_id, rule, metric, target, severity, value, started_at, updated_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, $7)
RETURNING id`,
@@ -139,7 +145,7 @@ type AlertFilter struct {
// Alerts lists alerts, newest first
func (s *Store) Alerts(ctx context.Context, filter AlertFilter) ([]Alert, error) {
- sql := `SELECT ` + alertColumns + ` FROM alerts a JOIN hosts h ON h.id = a.host_id WHERE true`
+ sql := `SELECT ` + alertColumns + ` FROM alerts a LEFT JOIN hosts h ON h.id = a.host_id WHERE true`
args := []any{}
if filter.Host != "" {
args = append(args, filter.Host)
diff --git a/internal/store/endpoints.go b/internal/store/endpoints.go
new file mode 100644
index 0000000..a1f468b
--- /dev/null
+++ b/internal/store/endpoints.go
@@ -0,0 +1,184 @@
+package store
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "time"
+
+ "github.com/jackc/pgx/v5"
+)
+
+// EndpointCheck is one HTTP check the collector ran for an endpoint rule
+type EndpointCheck struct {
+ Time time.Time
+ // Name is the rule's name
+ Name string
+ URL string
+ Method string
+ // StatusCode is 0 when there was no response
+ StatusCode int
+ // OK is true when the response had the expected status code
+ OK bool
+ // Latency is the time until the response headers arrived, 0 when there
+ // was no response
+ Latency time.Duration
+ Error string
+}
+
+func (s *Store) SaveEndpointCheck(ctx context.Context, check EndpointCheck) error {
+ var statusCode *int
+ var latencyMs *float64
+ if check.StatusCode != 0 {
+ ms := float64(check.Latency) / float64(time.Millisecond)
+ statusCode, latencyMs = &check.StatusCode, &ms
+ }
+ _, err := s.pool.Exec(ctx, `
+ INSERT INTO endpoint_checks (time, name, url, method, status_code, ok, latency_ms, error)
+ VALUES ($1, $2, $3, $4, $5, $6, $7, $8)`,
+ check.Time, check.Name, check.URL, check.Method, statusCode, check.OK, latencyMs, check.Error)
+ return err
+}
+
+const endpointCheckColumns = "time, name, url, method, status_code, ok, latency_ms, error"
+
+func scanEndpointCheck(row pgx.Row) (EndpointCheck, error) {
+ var check EndpointCheck
+ var statusCode *int
+ var latencyMs *float64
+ err := row.Scan(&check.Time, &check.Name, &check.URL, &check.Method, &statusCode, &check.OK, &latencyMs, &check.Error)
+ if statusCode != nil {
+ check.StatusCode = *statusCode
+ }
+ if latencyMs != nil {
+ check.Latency = time.Duration(*latencyMs * float64(time.Millisecond))
+ }
+ return check, err
+}
+
+// LatestEndpointCheck returns the newest check of an endpoint rule
+func (s *Store) LatestEndpointCheck(ctx context.Context, name string) (EndpointCheck, error) {
+ row := s.pool.QueryRow(ctx, `SELECT `+endpointCheckColumns+` FROM endpoint_checks
+ WHERE name = $1 ORDER BY time DESC LIMIT 1`, name)
+ check, err := scanEndpointCheck(row)
+ if errors.Is(err, pgx.ErrNoRows) {
+ return EndpointCheck{}, ErrNotFound
+ }
+ return check, err
+}
+
+// EndpointSummary is an endpoint's newest check up to the end of a range,
+// and how it did over the range
+type EndpointSummary struct {
+ Latest EndpointCheck
+ Checks int
+ // UptimePct is the share of checks that were OK
+ UptimePct float64
+ // AvgLatency is over the checks that got a response
+ AvgLatency time.Duration
+}
+
+// Endpoints summarizes every endpoint checked within the range
+func (s *Store) Endpoints(ctx context.Context, from time.Time, to time.Time) ([]EndpointSummary, error) {
+ if !to.After(from) {
+ return nil, fmt.Errorf("%w: from must be before to", ErrInvalid)
+ }
+ rows, err := s.pool.Query(ctx, `
+ WITH ranged AS (
+ SELECT name, count(*) AS checks, avg(ok::int) * 100 AS uptime, avg(latency_ms) AS latency
+ FROM endpoint_checks
+ WHERE time >= $1 AND time < $2
+ GROUP BY name
+ )
+ SELECT r.checks, r.uptime, r.latency, l.time, l.name, l.url, l.method, l.status_code, l.ok, l.latency_ms, l.error
+ FROM ranged r
+ CROSS JOIN LATERAL (
+ SELECT `+endpointCheckColumns+` FROM endpoint_checks
+ WHERE name = r.name AND time < $2
+ ORDER BY time DESC LIMIT 1
+ ) l
+ ORDER BY r.name`, from, to)
+ if err != nil {
+ return nil, err
+ }
+ defer rows.Close()
+
+ summaries := []EndpointSummary{}
+ for rows.Next() {
+ var summary EndpointSummary
+ var latencyMs *float64
+ var statusCode *int
+ var lastLatencyMs *float64
+ latest := &summary.Latest
+ if err := rows.Scan(&summary.Checks, &summary.UptimePct, &latencyMs, &latest.Time, &latest.Name, &latest.URL, &latest.Method,
+ &statusCode, &latest.OK, &lastLatencyMs, &latest.Error); err != nil {
+ return nil, err
+ }
+ if latencyMs != nil {
+ summary.AvgLatency = time.Duration(*latencyMs * float64(time.Millisecond))
+ }
+ if statusCode != nil {
+ latest.StatusCode = *statusCode
+ }
+ if lastLatencyMs != nil {
+ latest.Latency = time.Duration(*lastLatencyMs * float64(time.Millisecond))
+ }
+ summaries = append(summaries, summary)
+ }
+ return summaries, rows.Err()
+}
+
+// endpointSeriesColumns are the metrics EndpointSeries charts
+var endpointSeriesColumns = map[string]string{
+ // milliseconds until the response headers, over checks that got one
+ "latency": "avg(latency_ms)",
+ // share of checks that were OK
+ "availability": "avg(ok::int) * 100",
+}
+
+// EndpointSeries returns an endpoint's latency or availability over a range,
+// with at most maxPoints buckets
+func (s *Store) EndpointSeries(ctx context.Context, name string, metric string, from time.Time, to time.Time, maxPoints int) (SeriesResult, error) {
+ column, ok := endpointSeriesColumns[metric]
+ if !ok {
+ return SeriesResult{}, fmt.Errorf("%w: unknown endpoint metric %q", ErrInvalid, metric)
+ }
+ if !to.After(from) {
+ return SeriesResult{}, fmt.Errorf("%w: from must be before to", ErrInvalid)
+ }
+ if maxPoints <= 0 {
+ maxPoints = defaultMaxPoints
+ }
+ // rounded like pickSource does for host metrics
+ step := (to.Sub(from) / time.Duration(maxPoints)).Truncate(time.Second) + time.Second
+
+ // column comes from endpointSeriesColumns, never from the caller
+ rows, err := s.pool.Query(ctx, `
+ SELECT time_bucket($1::interval, time) AS t, `+column+`
+ FROM endpoint_checks
+ WHERE name = $2 AND time >= $3 AND time < $4
+ GROUP BY t ORDER BY t`, fmt.Sprintf("%d seconds", int64(step.Seconds())), name, from, to)
+ if err != nil {
+ return SeriesResult{}, err
+ }
+ defer rows.Close()
+
+ result := SeriesResult{Source: "raw", Step: step, Series: []Series{}}
+ series := Series{Label: name}
+ for rows.Next() {
+ var at time.Time
+ var value *float64
+ if err := rows.Scan(&at, &value); err != nil {
+ return SeriesResult{}, err
+ }
+ // no response in the whole bucket, so no latency
+ if value == nil {
+ continue
+ }
+ series.Points = append(series.Points, Point{Time: at, Value: *value})
+ }
+ if len(series.Points) > 0 {
+ result.Series = append(result.Series, series)
+ }
+ return result, rows.Err()
+}
diff --git a/internal/store/endpoints_test.go b/internal/store/endpoints_test.go
new file mode 100644
index 0000000..c4295ee
--- /dev/null
+++ b/internal/store/endpoints_test.go
@@ -0,0 +1,116 @@
+package store
+
+import (
+ "context"
+ "errors"
+ "testing"
+ "time"
+)
+
+func TestEndpointChecks(t *testing.T) {
+ st := testStore(t)
+ ctx := context.Background()
+
+ if _, err := st.LatestEndpointCheck(ctx, "api"); !errors.Is(err, ErrNotFound) {
+ t.Errorf("expected ErrNotFound before the first check, got %v", err)
+ }
+
+ // api: a check every 2 minutes for an hour, down for the last 3
+ start := time.Now().Add(-time.Hour).Truncate(time.Minute)
+ for i := 0; i < 30; i++ {
+ check := EndpointCheck{Time: start.Add(time.Duration(i) * 2 * time.Minute), Name: "api", URL: "https://api.example.com/health", Method: "GET",
+ StatusCode: 200, OK: true, Latency: 100 * time.Millisecond}
+ switch {
+ case i >= 28:
+ check.StatusCode, check.OK, check.Latency, check.Error = 0, false, 0, "connection refused"
+ case i == 27:
+ check.StatusCode, check.OK, check.Latency = 503, false, 400*time.Millisecond
+ }
+ if err := st.SaveEndpointCheck(ctx, check); err != nil {
+ t.Fatal(err)
+ }
+ }
+ end := start.Add(time.Hour)
+
+ latest, err := st.LatestEndpointCheck(ctx, "api")
+ if err != nil || latest.StatusCode != 0 || latest.OK || latest.Error != "connection refused" || !latest.Time.Equal(start.Add(58*time.Minute)) {
+ t.Errorf("unexpected latest check %+v %v", latest, err)
+ }
+
+ summaries, err := st.Endpoints(ctx, start, end)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(summaries) != 1 {
+ t.Fatalf("expected one endpoint, got %+v", summaries)
+ }
+ summary := summaries[0]
+ // 27 of 30 OK, and 28 responses averaging (27*100 + 400) / 28 ms
+ if summary.Checks != 30 || summary.UptimePct != 90 || summary.Latest.URL != "https://api.example.com/health" {
+ t.Errorf("unexpected summary %+v", summary)
+ }
+ wantMs := (27.0*100 + 400) / 28
+ if want := time.Duration(wantMs * float64(time.Millisecond)); summary.AvgLatency.Round(time.Microsecond) != want.Round(time.Microsecond) {
+ t.Errorf("expected an average of %v, got %v", want, summary.AvgLatency)
+ }
+ // the status at the end of an earlier range is the check before it
+ earlier, err := st.Endpoints(ctx, start, start.Add(30*time.Minute))
+ if err != nil || len(earlier) != 1 || !earlier[0].Latest.OK || earlier[0].Checks != 15 {
+ t.Errorf("unexpected earlier summary %+v %v", earlier, err)
+ }
+
+ // buckets of a few seconds, so each check is its own point
+ availability, err := st.EndpointSeries(ctx, "api", "availability", start, end, 0)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(availability.Series) != 1 || len(availability.Series[0].Points) != 30 {
+ t.Fatalf("expected 30 points, got %+v", availability)
+ }
+ if points := availability.Series[0].Points; points[26].Value != 100 || points[27].Value != 0 || points[29].Value != 0 {
+ t.Errorf("expected the last 3 checks to fail, got %+v", points[26:])
+ }
+ // the checks without a response have no latency
+ latency, err := st.EndpointSeries(ctx, "api", "latency", start, end, 0)
+ if err != nil || len(latency.Series[0].Points) != 28 || latency.Series[0].Points[0].Value != 100 || latency.Series[0].Points[27].Value != 400 {
+ t.Errorf("expected 28 latencies ending in 400ms, got %+v %v", latency, err)
+ }
+ if _, err := st.EndpointSeries(ctx, "api", "nope", start, end, 0); !errors.Is(err, ErrInvalid) {
+ t.Errorf("expected ErrInvalid for an unknown metric, got %v", err)
+ }
+}
+
+func TestAlertWithoutHost(t *testing.T) {
+ st := testStore(t)
+ ctx := context.Background()
+ if err := st.AddHost(ctx, "web1", "UTC"); err != nil {
+ t.Fatal(err)
+ }
+ at := time.Now().Truncate(time.Second)
+ id, err := st.CreateAlert(ctx, Alert{Rule: "API up", Metric: "endpoint", Target: "https://api.example.com", Severity: 2, StartedAt: at})
+ if err != nil {
+ t.Fatal(err)
+ }
+ // a second open alert for the same endpoint rule is refused, like for hosts
+ if _, err := st.CreateAlert(ctx, Alert{Rule: "API up", Metric: "endpoint", Target: "https://api.example.com", Severity: 2, StartedAt: at}); err == nil {
+ t.Error("expected a second open alert to be refused")
+ }
+ if _, err := st.CreateAlert(ctx, Alert{Host: "web1", Rule: "API up", Metric: "endpoint", Target: "https://api.example.com", Severity: 2, StartedAt: at}); err != nil {
+ t.Errorf("expected a host's alert with the same rule to be separate, got %v", err)
+ }
+
+ open, err := st.OpenAlert(ctx, "", "API up", "endpoint", "https://api.example.com")
+ if err != nil || open == nil || open.ID != id || open.Host != "" {
+ t.Fatalf("expected the open endpoint alert, got %+v %v", open, err)
+ }
+ alerts, err := st.Alerts(ctx, AlertFilter{OpenOnly: true})
+ if err != nil || len(alerts) != 2 {
+ t.Errorf("expected both alerts in the list, got %+v %v", alerts, err)
+ }
+ if err := st.ResolveAlert(ctx, id, 200, at.Add(time.Minute)); err != nil {
+ t.Fatal(err)
+ }
+ if open, err := st.OpenAlert(ctx, "", "API up", "endpoint", "https://api.example.com"); err != nil || open != nil {
+ t.Errorf("expected no open endpoint alert after resolving, got %+v %v", open, err)
+ }
+}
diff --git a/internal/store/migrations/006_endpoint_checks.sql b/internal/store/migrations/006_endpoint_checks.sql
new file mode 100644
index 0000000..c463f77
--- /dev/null
+++ b/internal/store/migrations/006_endpoint_checks.sql
@@ -0,0 +1,26 @@
+-- HTTP checks the collector runs for endpoint alert rules. name is the
+-- rule's name. status_code and latency_ms are null when there was no
+-- response. Kept as long as the hourly rollups, since a check every few
+-- minutes is little data.
+
+CREATE TABLE endpoint_checks (
+ time timestamptz NOT NULL,
+ name text NOT NULL,
+ url text NOT NULL,
+ method text NOT NULL,
+ status_code integer,
+ ok boolean NOT NULL,
+ latency_ms double precision,
+ error text NOT NULL DEFAULT ''
+);
+SELECT create_hypertable('endpoint_checks', by_range('time', INTERVAL '7 days'));
+CREATE INDEX ON endpoint_checks (name, time DESC);
+
+ALTER TABLE endpoint_checks SET (timescaledb.compress, timescaledb.compress_segmentby = 'name');
+SELECT add_compression_policy('endpoint_checks', INTERVAL '2 days');
+
+-- endpoint alerts belong to no host. NULLS NOT DISTINCT keeps at most one
+-- open alert per endpoint rule too.
+ALTER TABLE alerts ALTER COLUMN host_id DROP NOT NULL;
+DROP INDEX alerts_one_open;
+CREATE UNIQUE INDEX alerts_one_open ON alerts (host_id, rule, metric, target) NULLS NOT DISTINCT WHERE resolved_at IS NULL;
diff --git a/internal/store/query.go b/internal/store/query.go
index c0ce6c1..ea28eb1 100644
--- a/internal/store/query.go
+++ b/internal/store/query.go
@@ -266,6 +266,51 @@ func (s *Store) ProcessUsage(ctx context.Context, host string, from time.Time, t
return result, rows.Err()
}
+// CustomValue is the newest value of one of a host's custom metrics
+type CustomValue struct {
+ Host string
+ Name string
+ Unit string
+ Value float64
+ Time time.Time
+}
+
+// customValueWindow is how far back LatestCustomValues looks. Two days keeps
+// daily jobs in and only reads chunks that are not compressed yet.
+const customValueWindow = 48 * time.Hour
+
+// LatestCustomValues returns the newest value of every custom metric sent
+// within customValueWindow
+func (s *Store) LatestCustomValues(ctx context.Context) ([]CustomValue, error) {
+ since := time.Now().Add(-customValueWindow)
+ // the hourly rollup finds the metric names cheaply, then the index on
+ // (host_id, name, time) finds each one's newest value
+ rows, err := s.pool.Query(ctx, `
+ SELECT h.name, k.name, c.unit, c.value, c.time
+ FROM (SELECT DISTINCT host_id, name FROM custom_metrics_1h WHERE bucket > $1::timestamptz - INTERVAL '1 hour') k
+ JOIN hosts h ON h.id = k.host_id
+ CROSS JOIN LATERAL (
+ SELECT unit, value, time FROM custom_metrics
+ WHERE host_id = k.host_id AND name = k.name AND time >= $1 AND value IS NOT NULL
+ ORDER BY time DESC LIMIT 1
+ ) c
+ ORDER BY h.name, k.name`, since)
+ if err != nil {
+ return nil, err
+ }
+ defer rows.Close()
+
+ values := []CustomValue{}
+ for rows.Next() {
+ var value CustomValue
+ if err := rows.Scan(&value.Host, &value.Name, &value.Unit, &value.Value, &value.Time); err != nil {
+ return nil, err
+ }
+ values = append(values, value)
+ }
+ return values, rows.Err()
+}
+
// CustomMetricNames lists the custom metrics a host has sent within the
// hourly retention
func (s *Store) CustomMetricNames(ctx context.Context, host string) ([]string, error) {
diff --git a/internal/store/store.go b/internal/store/store.go
index c021696..dff27b5 100644
--- a/internal/store/store.go
+++ b/internal/store/store.go
@@ -119,6 +119,8 @@ var (
rawTables = []string{"host_metrics", "cpu_core_metrics", "disk_metrics", "disk_io_metrics", "net_metrics", "temp_metrics", "custom_metrics", "service_status", "process_snapshots", "container_metrics"}
minuteTables = []string{"host_metrics_1m", "disk_metrics_1m", "disk_io_metrics_1m", "net_metrics_1m", "temp_metrics_1m", "custom_metrics_1m", "container_metrics_1m"}
hourTables = []string{"host_metrics_1h", "disk_metrics_1h", "disk_io_metrics_1h", "net_metrics_1h", "temp_metrics_1h", "custom_metrics_1h", "container_metrics_1h"}
+ // raw data with no rollups, kept as long as the hourly rollups
+ longTables = []string{"endpoint_checks"}
)
// ApplyRetention replaces the retention policies with the configured ones,
@@ -131,6 +133,7 @@ func (s *Store) ApplyRetention(ctx context.Context) error {
{rawTables, s.retention.Raw},
{minuteTables, s.retention.Minute},
{hourTables, s.retention.Hour},
+ {longTables, s.retention.Hour},
}
for _, group := range groups {
for _, table := range group.tables {
diff --git a/internal/store/store_test.go b/internal/store/store_test.go
index 2c373f0..7b916a7 100644
--- a/internal/store/store_test.go
+++ b/internal/store/store_test.go
@@ -477,8 +477,8 @@ func TestPolicies(t *testing.T) {
}
want := map[string]int{
- "policy_retention": len(rawTables) + len(minuteTables) + len(hourTables),
- "policy_compression": len(rawTables),
+ "policy_retention": len(rawTables) + len(minuteTables) + len(hourTables) + len(longTables),
+ "policy_compression": len(rawTables) + len(longTables),
"policy_refresh_continuous_aggregate": len(minuteTables) + len(hourTables),
}
for name, count := range want {
@@ -637,3 +637,42 @@ func TestLatestSnapshots(t *testing.T) {
t.Errorf("expected web1's snapshot, got %+v", hosts[1])
}
}
+
+func TestLatestCustomValues(t *testing.T) {
+ st := testStore(t)
+ ctx := context.Background()
+ if err := st.AddHost(ctx, "web1", "UTC"); err != nil {
+ t.Fatal(err)
+ }
+ now := time.Now().Truncate(time.Second)
+ send := func(name string, value string, at time.Time) {
+ t.Helper()
+ metric := &monitor.CustomMetric{Name: name, Unit: "jobs", Value: value, ServerId: "web1", Time: strconv.FormatInt(at.Unix(), 10)}
+ if err := st.SaveCustomMetric(ctx, metric); err != nil {
+ t.Fatal(err)
+ }
+ }
+ // stale was last sent 3 days ago, so it is left out
+ send("stale", "1", now.Add(-72*time.Hour))
+ send("queue", "10", now.Add(-2*time.Hour))
+ send("queue", "12", now.Add(-time.Minute))
+ send("workers", "4", now.Add(-26*time.Hour))
+ refreshRollups(t, st, "custom_metrics_1m", "custom_metrics_1h")
+
+ values, err := st.LatestCustomValues(ctx)
+ if err != nil {
+ t.Fatal(err)
+ }
+ want := []CustomValue{
+ {Host: "web1", Name: "queue", Unit: "jobs", Value: 12, Time: now.Add(-time.Minute)},
+ {Host: "web1", Name: "workers", Unit: "jobs", Value: 4, Time: now.Add(-26 * time.Hour)},
+ }
+ if len(values) != len(want) {
+ t.Fatalf("got %+v, want %+v", values, want)
+ }
+ for i := range want {
+ if values[i].Host != want[i].Host || values[i].Name != want[i].Name || values[i].Value != want[i].Value || !values[i].Time.Equal(want[i].Time) {
+ t.Errorf("got %+v, want %+v", values[i], want[i])
+ }
+ }
+}
diff --git a/internal/version/version.go b/internal/version/version.go
new file mode 100644
index 0000000..81b1cf0
--- /dev/null
+++ b/internal/version/version.go
@@ -0,0 +1,46 @@
+// Package version is the SyMon version a binary was built from
+package version
+
+import "runtime/debug"
+
+// Version is set at build time, for example
+//
+// go build -ldflags "-X github.com/dhamith93/SyMon/internal/version.Version=v3.1.0"
+//
+// The Makefile sets it from git describe.
+var Version = ""
+
+// String returns Version. A binary built without it, like with a plain go
+// build, reports the git commit it was built from, and "dev" without one.
+func String() string {
+ if Version != "" {
+ return Version
+ }
+ info, ok := debug.ReadBuildInfo()
+ if !ok {
+ return "dev"
+ }
+ return fromBuildInfo(info.Settings)
+}
+
+func fromBuildInfo(settings []debug.BuildSetting) string {
+ revision, modified := "", false
+ for _, setting := range settings {
+ switch setting.Key {
+ case "vcs.revision":
+ revision = setting.Value
+ case "vcs.modified":
+ modified = setting.Value == "true"
+ }
+ }
+ if revision == "" {
+ return "dev"
+ }
+ if len(revision) > 7 {
+ revision = revision[:7]
+ }
+ if modified {
+ revision += "-dirty"
+ }
+ return revision
+}
diff --git a/internal/version/version_test.go b/internal/version/version_test.go
new file mode 100644
index 0000000..938fb30
--- /dev/null
+++ b/internal/version/version_test.go
@@ -0,0 +1,31 @@
+package version
+
+import (
+ "runtime/debug"
+ "testing"
+)
+
+func TestString(t *testing.T) {
+ Version = "v3.1.0"
+ t.Cleanup(func() { Version = "" })
+ if got := String(); got != "v3.1.0" {
+ t.Errorf("expected the version set at build time, got %q", got)
+ }
+}
+
+func TestFromBuildInfo(t *testing.T) {
+ commit := debug.BuildSetting{Key: "vcs.revision", Value: "2519821d4b5c6e0f1a2b3c4d5e6f7a8b9c0d1e2f"}
+ tests := []struct {
+ settings []debug.BuildSetting
+ want string
+ }{
+ {nil, "dev"},
+ {[]debug.BuildSetting{commit}, "2519821"},
+ {[]debug.BuildSetting{commit, {Key: "vcs.modified", Value: "true"}}, "2519821-dirty"},
+ }
+ for _, tt := range tests {
+ if got := fromBuildInfo(tt.settings); got != tt.want {
+ t.Errorf("%v: got %q, want %q", tt.settings, got, tt.want)
+ }
+ }
+}