From 67f3beb924f417dfdf1f8ec062d14b18fb9923f4 Mon Sep 17 00:00:00 2001 From: Dhamith Hewamullage Date: Tue, 29 Sep 2026 21:35:27 +0530 Subject: [PATCH 1/7] let the metrics endpoint be switched off --- README.MD | 2 +- client/.env-example | 2 ++ client/internal/server/metrics.go | 4 ++++ client/internal/server/metrics_test.go | 10 ++++++++++ client/internal/server/server.go | 2 ++ client/internal/server/server_test.go | 2 +- docs/install.md | 2 +- internal/config/config_handler.go | 3 +++ internal/config/config_handler_test.go | 9 +++++++++ 9 files changed, 33 insertions(+), 3 deletions(-) diff --git a/README.MD b/README.MD index 848b296..7d27c26 100644 --- a/README.MD +++ b/README.MD @@ -115,4 +115,4 @@ The Client exposes a JSON API under `/api/v1`. Times are unix seconds. Errors re * `GET /api/v1/alerts?host=&open=1&from=&to=` * Alerts, newest first. `open=1` leaves out resolved ones -The Client also serves `GET /metrics`, every host's latest values in the Prometheus text format. See [Prometheus and Grafana](docs/install.md#prometheus-and-grafana). +The Client also serves `GET /metrics`, every host's latest values in the Prometheus text format, unless `SYMON_CLIENT_METRICS_ENABLED=false`. See [Prometheus and Grafana](docs/install.md#prometheus-and-grafana). diff --git a/client/.env-example b/client/.env-example index a6a9f0d..c9db332 100644 --- a/client/.env-example +++ b/client/.env-example @@ -8,6 +8,8 @@ export SYMON_CLIENT_DOWNLOADS_DIR=downloads # how agents on other hosts reach the collector, when the dashboard's host # name with the collector's port does not work export SYMON_CLIENT_AGENT_COLLECTOR_ENDPOINT= +# /metrics for Prometheus, on unless this is false +export SYMON_CLIENT_METRICS_ENABLED= export SYMON_CLIENT_LOG_FILE_ENABLED= export SYMON_CLIENT_LOG_FILE_PATH= export SYMON_KEY='' diff --git a/client/internal/server/metrics.go b/client/internal/server/metrics.go index 3dc4ec4..6982ac8 100644 --- a/client/internal/server/metrics.go +++ b/client/internal/server/metrics.go @@ -16,6 +16,10 @@ import ( // getMetrics serves every host's latest values in the Prometheus text // format, so Prometheus can scrape the dashboard func (s *server) getMetrics(w http.ResponseWriter, r *http.Request) { + if !s.metricsEnabled { + http.Error(w, "metrics are switched off with SYMON_CLIENT_METRICS_ENABLED=false", http.StatusNotFound) + return + } response, err := s.collector.Snapshots(r.Context(), &api.Void{}) if err != nil { writeGRPCError(w, "snapshots", err) diff --git a/client/internal/server/metrics_test.go b/client/internal/server/metrics_test.go index c953e76..10a6d28 100644 --- a/client/internal/server/metrics_test.go +++ b/client/internal/server/metrics_test.go @@ -3,6 +3,7 @@ package server import ( "context" "encoding/json" + "net/http" "net/http/httptest" "strings" "testing" @@ -70,6 +71,15 @@ func TestMetrics(t *testing.T) { } } +func TestMetricsSwitchedOff(t *testing.T) { + s, _ := newTestServer(t, nil) + s.metricsEnabled = false + code, body, _ := get(t, s, "/metrics") + if code != http.StatusNotFound || !strings.Contains(body, "SYMON_CLIENT_METRICS_ENABLED") { + t.Errorf("unexpected response %d: %s", code, body) + } +} + func TestMetricsWriter(t *testing.T) { w := newMetricsWriter() w.gauge("a", "First.", 1, "name", `quote " backslash \ newline`+"\n") diff --git a/client/internal/server/server.go b/client/internal/server/server.go index 184fe5d..09c0c0b 100644 --- a/client/internal/server/server.go +++ b/client/internal/server/server.go @@ -31,6 +31,7 @@ type server struct { collectorEndpoint string agentCollector string downloadsDir string + metricsEnabled bool } // Run starts the server on the given address, like ":8080" @@ -50,6 +51,7 @@ func Run(address string) { collectorEndpoint: config.CollectorEndpoint, agentCollector: config.AgentCollectorEndpoint, downloadsDir: config.DownloadsDir, + metricsEnabled: config.MetricsEnabled, } httpServer := &http.Server{ Addr: address, diff --git a/client/internal/server/server_test.go b/client/internal/server/server_test.go index 1d03129..c193a86 100644 --- a/client/internal/server/server_test.go +++ b/client/internal/server/server_test.go @@ -103,7 +103,7 @@ func newTestServer(t *testing.T, files fstest.MapFS) (*server, *fakeCollector) { t.Fatal(err) } t.Cleanup(func() { conn.Close() }) - return &server{collector: api.NewMonitorDataServiceClient(conn), refreshSeconds: 15, files: files}, fake + return &server{collector: api.NewMonitorDataServiceClient(conn), refreshSeconds: 15, files: files, metricsEnabled: true}, fake } // get calls the server and returns the status, the body and what was logged diff --git a/docs/install.md b/docs/install.md index 3d7b420..6cb0b47 100644 --- a/docs/install.md +++ b/docs/install.md @@ -259,7 +259,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. `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 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) + } + } +} From 261634685709f7e128d2e5cb13414a14bad7f8db Mon Sep 17 00:00:00 2001 From: Dhamith Hewamullage Date: Tue, 29 Sep 2026 21:37:21 +0530 Subject: [PATCH 2/7] add custom metrics to /metrics --- client/internal/server/metrics.go | 11 ++ client/internal/server/metrics_test.go | 20 ++- docs/install.md | 2 +- internal/api/api.go | 13 ++ internal/api/api.pb.go | 181 +++++++++++++++++++------ internal/api/api.proto | 11 ++ internal/store/query.go | 45 ++++++ internal/store/store_test.go | 39 ++++++ 8 files changed, 272 insertions(+), 50 deletions(-) diff --git a/client/internal/server/metrics.go b/client/internal/server/metrics.go index 6982ac8..595e2d2 100644 --- a/client/internal/server/metrics.go +++ b/client/internal/server/metrics.go @@ -27,7 +27,9 @@ func (s *server) getMetrics(w http.ResponseWriter, r *http.Request) { } metrics := newMetricsWriter() + up := map[string]bool{} for _, host := range response.Hosts { + up[host.Host] = host.Up metrics.gauge("symon_up", "1 when the host reported within the last minute.", boolValue(host.Up), "host", host.Host) if host.LastSeen > 0 { metrics.gauge("symon_last_seen_timestamp_seconds", "When the host was last heard from.", float64(host.LastSeen), "host", host.Host) @@ -44,6 +46,15 @@ func (s *server) getMetrics(w http.ResponseWriter, r *http.Request) { } addHostMetrics(metrics, host.Host, &data) } + for _, custom := range response.CustomMetrics { + if !up[custom.Host] { + continue + } + metrics.gauge("symon_custom_metric", "The newest value of a custom metric sent in the last 2 days.", custom.Value, + "host", custom.Host, "name", custom.Name, "unit", custom.Unit) + metrics.gauge("symon_custom_metric_timestamp_seconds", "When the custom metric was last sent.", float64(custom.Time), + "host", custom.Host, "name", custom.Name) + } w.Header().Set("Content-Type", "text/plain; version=0.0.4; charset=utf-8") if err := metrics.writeTo(w); err != nil { diff --git a/client/internal/server/metrics_test.go b/client/internal/server/metrics_test.go index 10a6d28..3e7acf6 100644 --- a/client/internal/server/metrics_test.go +++ b/client/internal/server/metrics_test.go @@ -31,11 +31,17 @@ func (f *fakeCollector) Snapshots(ctx context.Context, in *api.Void) (*api.Snaps if err != nil { return nil, err } - return &api.SnapshotList{Hosts: []*api.HostSnapshot{ - {Host: "web1", Up: true, LastSeen: 1700000000, Time: 1700000000, SnapshotJson: string(snapshot)}, - {Host: "db1", LastSeen: 1690000000, Time: 1690000000, SnapshotJson: string(snapshot)}, - {Host: "new1", Up: true, LastSeen: 1700000000}, - }}, nil + return &api.SnapshotList{ + Hosts: []*api.HostSnapshot{ + {Host: "web1", Up: true, LastSeen: 1700000000, Time: 1700000000, SnapshotJson: string(snapshot)}, + {Host: "db1", LastSeen: 1690000000, Time: 1690000000, SnapshotJson: string(snapshot)}, + {Host: "new1", Up: true, LastSeen: 1700000000}, + }, + CustomMetrics: []*api.CustomValue{ + {Host: "web1", Name: "queue-length", Unit: "jobs", Value: 42, Time: 1699999000}, + {Host: "db1", Name: "backup-age", Unit: "hours", Value: 30, Time: 1689999000}, + }, + }, nil } func TestMetrics(t *testing.T) { @@ -56,13 +62,15 @@ func TestMetrics(t *testing.T) { "# TYPE symon_network_receive_bytes_total counter\n" + `symon_network_receive_bytes_total{host="web1",iface="eth0"} 1000`, `symon_service_up{host="web1",service="nginx"} 1`, `symon_container_receive_bytes_per_second{host="web1",container="web",project="shop"} 2048`, + `symon_custom_metric{host="web1",name="queue-length",unit="jobs"} 42`, + `symon_custom_metric_timestamp_seconds{host="web1",name="queue-length"} 1.699999e+09`, } { if !strings.Contains(body, line+"\n") { t.Errorf("expected %q in:\n%s", line, body) } } // db1 stopped reporting, so its old values are left out - if strings.Contains(body, `{host="db1",`) || strings.Contains(body, `symon_cpu_usage_percent{host="db1"}`) { + if strings.Contains(body, `{host="db1",`) || strings.Contains(body, `symon_cpu_usage_percent{host="db1"}`) || strings.Contains(body, "backup-age") { t.Errorf("expected only up and last seen for db1:\n%s", body) } // web1 has no transmit rate, as if it were on the host network diff --git a/docs/install.md b/docs/install.md index 6cb0b47..5f8383d 100644 --- a/docs/install.md +++ b/docs/install.md @@ -259,7 +259,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. To switch it off, add `SYMON_CLIENT_METRICS_ENABLED=false` to `/etc/symon/client.env` and restart the dashboard. +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 diff --git a/internal/api/api.go b/internal/api/api.go index 8b3cf70..4da8c54 100644 --- a/internal/api/api.go +++ b/internal/api/api.go @@ -226,6 +226,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 +241,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 } diff --git a/internal/api/api.pb.go b/internal/api/api.pb.go index 7ff53b4..9357809 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,13 @@ func (x *SnapshotList) GetHosts() []*HostSnapshot { return nil } +func (x *SnapshotList) GetCustomMetrics() []*CustomValue { + if x != nil { + return x.CustomMetrics + } + return nil +} + var File_api_api_proto protoreflect.FileDescriptor const file_api_api_proto_rawDesc = "" + @@ -1803,9 +1889,16 @@ 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\rcustomMetrics2\x82\x06\n" + "\x12MonitorDataService\x123\n" + "\x06Enroll\x12\x12.api.EnrollRequest\x1a\x13.api.EnrollResponse\"\x00\x12-\n" + "\n" + @@ -1835,7 +1928,7 @@ 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, 27) var file_api_api_proto_goTypes = []any{ (*Void)(nil), // 0: api.Void (*Message)(nil), // 1: api.Message @@ -1862,7 +1955,8 @@ var file_api_api_proto_goTypes = []any{ (*ProcessUsageRequest)(nil), // 22: api.ProcessUsageRequest (*ProcessUsage)(nil), // 23: api.ProcessUsage (*ProcessUsageList)(nil), // 24: api.ProcessUsageList - (*SnapshotList)(nil), // 25: api.SnapshotList + (*CustomValue)(nil), // 25: api.CustomValue + (*SnapshotList)(nil), // 26: api.SnapshotList } var file_api_api_proto_depIdxs = []int32{ 6, // 0: api.FleetSummary.hosts:type_name -> api.HostSummary @@ -1872,39 +1966,40 @@ 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 + 4, // 8: api.MonitorDataService.Enroll:input_type -> api.EnrollRequest + 2, // 9: api.MonitorDataService.HandlePing:input_type -> api.ServerInfo + 2, // 10: api.MonitorDataService.InitAgent:input_type -> api.ServerInfo + 3, // 11: api.MonitorDataService.HandleMonitorData:input_type -> api.MonitorData + 3, // 12: api.MonitorDataService.HandleCustomMonitorData:input_type -> api.MonitorData + 0, // 13: api.MonitorDataService.Fleet:input_type -> api.Void + 8, // 14: api.MonitorDataService.Snapshot:input_type -> api.HostRequest + 10, // 15: api.MonitorDataService.QuerySeries:input_type -> api.SeriesRequest + 14, // 16: api.MonitorDataService.Processes:input_type -> api.ProcessesRequest + 8, // 17: api.MonitorDataService.CustomMetricNames:input_type -> api.HostRequest + 17, // 18: api.MonitorDataService.Alerts:input_type -> api.AlertsRequest + 8, // 19: api.MonitorDataService.DiskForecasts:input_type -> api.HostRequest + 22, // 20: api.MonitorDataService.ProcessUsage:input_type -> api.ProcessUsageRequest + 0, // 21: api.MonitorDataService.Snapshots:input_type -> api.Void + 5, // 22: api.MonitorDataService.Enroll:output_type -> api.EnrollResponse + 1, // 23: api.MonitorDataService.HandlePing:output_type -> api.Message + 1, // 24: api.MonitorDataService.InitAgent:output_type -> api.Message + 1, // 25: api.MonitorDataService.HandleMonitorData:output_type -> api.Message + 1, // 26: api.MonitorDataService.HandleCustomMonitorData:output_type -> api.Message + 7, // 27: api.MonitorDataService.Fleet:output_type -> api.FleetSummary + 9, // 28: api.MonitorDataService.Snapshot:output_type -> api.HostSnapshot + 13, // 29: api.MonitorDataService.QuerySeries:output_type -> api.SeriesResponse + 15, // 30: api.MonitorDataService.Processes:output_type -> api.ProcessesResponse + 16, // 31: api.MonitorDataService.CustomMetricNames:output_type -> api.NameList + 19, // 32: api.MonitorDataService.Alerts:output_type -> api.AlertList + 21, // 33: api.MonitorDataService.DiskForecasts:output_type -> api.DiskForecastList + 24, // 34: api.MonitorDataService.ProcessUsage:output_type -> api.ProcessUsageList + 26, // 35: api.MonitorDataService.Snapshots:output_type -> api.SnapshotList + 22, // [22:36] is the sub-list for method output_type + 8, // [8:22] is the sub-list for method input_type + 8, // [8:8] is the sub-list for extension type_name + 8, // [8:8] is the sub-list for extension extendee + 0, // [0:8] is the sub-list for field type_name } func init() { file_api_api_proto_init() } @@ -1920,7 +2015,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: 27, NumExtensions: 0, NumServices: 1, }, diff --git a/internal/api/api.proto b/internal/api/api.proto index c8f283d..979ed24 100644 --- a/internal/api/api.proto +++ b/internal/api/api.proto @@ -186,8 +186,19 @@ 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; } service MonitorDataService { 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_test.go b/internal/store/store_test.go index 2c373f0..9acc99a 100644 --- a/internal/store/store_test.go +++ b/internal/store/store_test.go @@ -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]) + } + } +} From 81c66c5259cb47af758c0d053e7ab3042ccb3f27 Mon Sep 17 00:00:00 2001 From: Dhamith Hewamullage Date: Tue, 29 Sep 2026 21:40:19 +0530 Subject: [PATCH 3/7] chart where filling disks are headed --- README.MD | 2 +- client/web/src/components/ChartCard.svelte | 13 +++-- client/web/src/components/TimeChart.svelte | 14 ++++-- client/web/src/lib/align.ts | 3 ++ client/web/src/lib/format.ts | 5 ++ client/web/src/lib/projection.test.ts | 49 +++++++++++++++++++ client/web/src/lib/projection.ts | 57 ++++++++++++++++++++++ client/web/src/pages/Host.svelte | 38 ++++++++++++++- docs/install.md | 2 +- 9 files changed, 171 insertions(+), 12 deletions(-) create mode 100644 client/web/src/lib/projection.test.ts create mode 100644 client/web/src/lib/projection.ts diff --git a/README.MD b/README.MD index 7d27c26..ec1157c 100644 --- a/README.MD +++ b/README.MD @@ -11,7 +11,7 @@ SyMon is a self-hosted monitoring tool for Linux servers, home labs and Raspberr **Hosts** - CPU, overall and per core, load average, memory and swap - Disk space, disk IO and busy time, network traffic, TCP connections -- When each disk will be full, at the rate it grew over the last week +- When each disk will be full, at the rate it grew over the last week, with a chart of where it is headed - Pressure stall information and temperature sensors - The top processes by CPU and memory at any point in time, and the programs that used the most over any range - Whether chosen systemd services are running diff --git a/client/web/src/components/ChartCard.svelte b/client/web/src/components/ChartCard.svelte index f157862..8c90d05 100644 --- a/client/web/src/components/ChartCard.svelte +++ b/client/web/src/components/ChartCard.svelte @@ -21,13 +21,16 @@ onpick?: (time: number) => void; // a picked time to mark on the chart, 0 for none marker?: number; + // dashed lines, for values that are projected + projection?: boolean; // replaces the line chart, like the per core heatmap body?: Snippet; - // rows for the table view when body replaces the chart + // rows for the table view, instead of one row per time table?: Snippet; } - let { title, unit, data, loading = false, error = '', yMax, syncKey, from, to, area = false, note, onzoom, onpick, marker = 0, body, table }: Props = $props(); + let { title, unit, data, loading = false, error = '', yMax, syncKey, from, to, area = false, note, onzoom, onpick, marker = 0, projection = false, body, table }: Props = + $props(); let showTable = $state(false); const colors = ['--series-1', '--series-2', '--series-3', '--series-4', '--series-5', '--series-6', '--series-7', '--series-8']; @@ -59,7 +62,7 @@ {#if data && data.labels.length > 1 && !showTable}
    {#each data.labels as label, i (label)} -
  • {label}
    • +
  • {label}
    • {/each}
{/if} @@ -73,7 +76,7 @@

No data in this time range.

{:else if showTable}
- {#if body && table} + {#if table} {@render table()} {:else if data} @@ -100,7 +103,7 @@ {:else if body} {@render body()} {:else if data} - + {/if} {#if data && data.hidden.length > 0} diff --git a/client/web/src/components/TimeChart.svelte b/client/web/src/components/TimeChart.svelte index 4864b4c..8e74915 100644 --- a/client/web/src/components/TimeChart.svelte +++ b/client/web/src/components/TimeChart.svelte @@ -23,9 +23,12 @@ onpick?: (time: number) => void; // a picked time to mark with a line, 0 for none marker?: number; + // dashed lines with dots at their ends, for values that are projected + projection?: boolean; } - let { data, unit, yMax, syncKey, from, to, area = false, height = 180, onzoom, onpick, marker = 0 }: Props = $props(); + let { data, unit, yMax, syncKey, from, to, area = false, height = 180, onzoom, onpick, marker = 0, projection = false }: Props = + $props(); let wrapper: HTMLDivElement; // uPlot owns this element, Svelte owns the tooltip next to it @@ -45,8 +48,10 @@ const font = '12px system-ui, -apple-system, "Segoe UI", sans-serif'; + // one color per series, from its palette slot function seriesColors(): string[] { - return Array.from({ length: 8 }, (_, i) => cssVar(`--series-${i + 1}`)); + const palette = Array.from({ length: 8 }, (_, i) => cssVar(`--series-${i + 1}`)); + return data.labels.map((_, i) => palette[data.colors?.[i] ?? i]); } function plotData(d: Aligned): uPlot.AlignedData { @@ -92,7 +97,8 @@ stroke: colors[i], width: 2, fill: single && area ? colors[i] + '1a' : undefined, - points: { show: false }, + dash: projection ? [6, 4] : undefined, + points: projection ? { show: true, size: 8, fill: colors[i], stroke: colors[i] } : { show: false }, })), ], hooks: { @@ -169,7 +175,7 @@ // rebuild when the series or the theme change, otherwise swap the data in $effect(() => { - const key = [theme.resolved, unit, yMax, height, area, ...data.labels].join('|'); + const key = [theme.resolved, unit, yMax, height, area, projection, ...data.labels, ...(data.colors ?? [])].join('|'); // a new range only needs the x scale to move, not a rebuild void from; void to; diff --git a/client/web/src/lib/align.ts b/client/web/src/lib/align.ts index 4ebf9e8..05ec20e 100644 --- a/client/web/src/lib/align.ts +++ b/client/web/src/lib/align.ts @@ -16,6 +16,9 @@ export interface Aligned { values: Value[][]; // labels left out because of the series cap hidden: string[]; + // palette slot of each series, so a series keeps the color it has on + // another chart. By default a series takes the slot of its position. + colors?: number[]; } // the categorical palette has 8 slots, more series than that do not get diff --git a/client/web/src/lib/format.ts b/client/web/src/lib/format.ts index 27b27ee..2fd8fa8 100644 --- a/client/web/src/lib/format.ts +++ b/client/web/src/lib/format.ts @@ -75,6 +75,11 @@ export function formatDays(days: number): string { return `about ${Math.round(days / 30)} months`; } +export function formatDate(unixSeconds: number): string { + if (!unixSeconds) return '–'; + return new Date(unixSeconds * 1000).toLocaleDateString(undefined, { year: 'numeric', month: 'short', day: 'numeric' }); +} + export function formatDateTime(unixSeconds: number): string { if (!unixSeconds) return '–'; return new Date(unixSeconds * 1000).toLocaleString(undefined, { diff --git a/client/web/src/lib/projection.test.ts b/client/web/src/lib/projection.test.ts new file mode 100644 index 0000000..89bccb5 --- /dev/null +++ b/client/web/src/lib/projection.test.ts @@ -0,0 +1,49 @@ +import { describe, expect, it } from 'vitest'; +import type { DiskForecast } from './api'; +import { projectDisks } from './projection'; + +const now = 1_700_000_000; +const day = 86400; + +function disk(mount: string, usedPct: number, pctPerDay: number, daysToFull: number | null): DiskForecast { + return { device: `/dev/${mount}`, mount, usedPct, pctPerDay, bytesPerDay: 1e9, daysToFull }; +} + +describe('projectDisks', () => { + it('leaves out disks that are not filling up', () => { + expect(projectDisks([disk('/', 40, 0, null)], now, ['/'])).toBeNull(); + }); + + it('runs each line from now to the day the disk is full', () => { + const projection = projectDisks([disk('/data', 60, 2, 20), disk('/', 40, 0, null)], now, ['/', '/data'])!; + expect(projection.from).toBe(now); + // a bit past the 20 days, so the end of the line shows + expect(projection.to).toBe(now + 24 * day); + expect(projection.data.labels).toEqual(['/data']); + expect(projection.data.times).toEqual([now, now + 20 * day]); + expect(projection.data.values).toEqual([[60, 100]]); + // /data keeps its color from the disk space chart, where it is second + expect(projection.data.colors).toEqual([1]); + }); + + it('stops lines at the end of the chart', () => { + const projection = projectDisks([disk('/', 50, 0.5, 100), disk('/var', 80, 4, 5)], now, ['/', '/var'])!; + expect(projection.to).toBe(now + 90 * day); + expect(projection.data.times).toEqual([now, now + 5 * day, now + 90 * day]); + expect(projection.data.values).toEqual([ + [50, undefined, 95], + [80, 100, undefined], + ]); + }); + + it('shows at least a week', () => { + const projection = projectDisks([disk('/', 99, 2, 0.5)], now, ['/'])!; + expect(projection.to).toBe(now + 7 * day); + }); + + it('gives a disk the disk chart does not show a free color', () => { + const projection = projectDisks([disk('/new', 10, 1, 90), disk('/', 50, 1, 50)], now, ['/'])!; + expect(projection.data.labels).toEqual(['/', '/new']); + expect(projection.data.colors).toEqual([0, 1]); + }); +}); diff --git a/client/web/src/lib/projection.ts b/client/web/src/lib/projection.ts new file mode 100644 index 0000000..0d1a566 --- /dev/null +++ b/client/web/src/lib/projection.ts @@ -0,0 +1,57 @@ +import { maxSeries, type Aligned, type Value } from './align'; +import type { DiskForecast } from './api'; + +const day = 86400; + +export interface Projection { + data: Aligned; + from: number; + to: number; + // the filling disks, in the order of the chart's series + disks: DiskForecast[]; +} + +// Where each filling disk is headed at its growth over the last week. Each +// line runs from now to the day the disk is full, or to the end of the +// chart. The chart spans a bit past the last disk to fill, 7 to 90 days. +// mounts is the series order of the disk space chart, so a disk keeps its +// color there. +export function projectDisks(forecasts: DiskForecast[], now: number, mounts: string[]): Projection | null { + const rank = (mount: string) => { + const i = mounts.indexOf(mount); + return i < 0 ? mounts.length : i; + }; + const disks = forecasts + .filter((f) => f.daysToFull !== null) + .sort((a, b) => rank(a.mount) - rank(b.mount) || a.mount.localeCompare(b.mount)) + .slice(0, maxSeries); + if (disks.length === 0) return null; + + const longest = Math.max(...disks.map((d) => d.daysToFull ?? 0)); + const horizon = Math.min(90, Math.max(7, Math.ceil(longest * 1.2))); + const ends = disks.map((d) => now + Math.round(Math.min(d.daysToFull ?? 0, horizon) * day)); + const times = [...new Set([now, ...ends])].sort((a, b) => a - b); + + // undefined between the two ends, so each line joins straight across + const values: Value[][] = disks.map((d, i) => + times.map((time) => { + if (time === now) return d.usedPct; + if (time === ends[i]) return Math.min(100, d.usedPct + (d.pctPerDay * (ends[i] - now)) / day); + return undefined; + }), + ); + // a mount the disk chart does not show gets the first free color + const taken = new Set(disks.map((d) => mounts.indexOf(d.mount)).filter((i) => i >= 0 && i < maxSeries)); + const free = [...Array(maxSeries).keys()].filter((i) => !taken.has(i)); + const colors = disks.map((d) => { + const i = mounts.indexOf(d.mount); + return i >= 0 && i < maxSeries ? i : (free.shift() ?? 0); + }); + + return { + data: { times, labels: disks.map((d) => d.mount), values, hidden: [], colors }, + from: now, + to: now + horizon * day, + disks, + }; +} diff --git a/client/web/src/pages/Host.svelte b/client/web/src/pages/Host.svelte index 2d38aac..6862bfe 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); @@ -179,6 +183,38 @@ onpick={pickTime} marker={processesAt} /> + {#if spec.key === 'disk_used' && projection} + + {#snippet table()} +
+ + + + + {#each projection.disks as disk (disk.device + disk.mount)} + + + + + + + + {/each} + +
MountUsedGrowth a dayFull inFull on
{disk.mount}{formatPercent(disk.usedPct)}{formatBytes(disk.bytesPerDay)}{formatDays(disk.daysToFull ?? 0)}{formatDate(now + (disk.daysToFull ?? 0) * 86400)}
+ {/snippet} + + {/if} {/each} {#if section.title === 'CPU' && showCores && cores.length === 0} diff --git a/docs/install.md b/docs/install.md index 5f8383d..e8fa97a 100644 --- a/docs/install.md +++ b/docs/install.md @@ -221,7 +221,7 @@ A rule looks like this: `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. -`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. +`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. From f73616d69e9eee05e5a5fa39696157a43eef7192 Mon Sep 17 00:00:00 2001 From: Dhamith Hewamullage Date: Tue, 29 Sep 2026 21:41:33 +0530 Subject: [PATCH 4/7] keep disk forecast alerts apart per disk in the alert processor --- internal/alertapi/alertapi.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 { From b3531b5575cac036856b90518f49709b7d04bc26 Mon Sep 17 00:00:00 2001 From: Dhamith Hewamullage Date: Tue, 29 Sep 2026 21:51:30 +0530 Subject: [PATCH 5/7] store endpoint checks and raise their alerts like other rules --- client/internal/server/server.go | 77 +++ client/internal/server/server_test.go | 34 ++ collector/alerts.go | 184 +------ collector/endpoints.go | 124 +++++ collector/endpoints_test.go | 87 ++++ collector/evaluator.go | 53 +- collector/evaluator_test.go | 62 ++- internal/alerts/alert.go | 3 - internal/api/api.go | 45 +- internal/api/api.pb.go | 466 +++++++++++++++--- internal/api/api.proto | 38 ++ internal/api/api_grpc.pb.go | 76 +++ internal/monitor/monitor.go | 8 +- internal/store/alerts.go | 28 +- internal/store/endpoints.go | 184 +++++++ internal/store/endpoints_test.go | 116 +++++ .../store/migrations/006_endpoint_checks.sql | 26 + internal/store/store.go | 3 + internal/store/store_test.go | 4 +- 19 files changed, 1361 insertions(+), 257 deletions(-) create mode 100644 collector/endpoints.go create mode 100644 collector/endpoints_test.go create mode 100644 internal/store/endpoints.go create mode 100644 internal/store/endpoints_test.go create mode 100644 internal/store/migrations/006_endpoint_checks.sql diff --git a/client/internal/server/server.go b/client/internal/server/server.go index 09c0c0b..fee21ab 100644 --- a/client/internal/server/server.go +++ b/client/internal/server/server.go @@ -73,6 +73,8 @@ func (s *server) routes() http.Handler { mux.HandleFunc("GET /api/v1/hosts/{host}/custom-metrics", s.getCustomMetrics) mux.HandleFunc("GET /api/v1/hosts/{host}/disk-forecasts", s.getDiskForecasts) mux.HandleFunc("GET /api/v1/alerts", s.getAlerts) + mux.HandleFunc("GET /api/v1/endpoints", s.getEndpoints) + mux.HandleFunc("GET /api/v1/endpoints/series", s.getEndpointSeries) mux.HandleFunc("/api/", func(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusNotFound, "no such endpoint") }) @@ -183,7 +185,10 @@ func (s *server) getSeries(w http.ResponseWriter, r *http.Request) { writeGRPCError(w, query.Get("metric")+" for "+host, err) return } + writeSeries(w, response) +} +func writeSeries(w http.ResponseWriter, response *api.SeriesResponse) { out := make([]series, 0, len(response.Series)) for _, s := range response.Series { points := make([][2]float64, 0, len(s.Points)) @@ -265,6 +270,78 @@ func (s *server) getCustomMetrics(w http.ResponseWriter, r *http.Request) { writeJSON(w, map[string]any{"names": nonNil(names.Names)}) } +type endpointStatus struct { + Name string `json:"name"` + URL string `json:"url"` + Method string `json:"method"` + // the newest check + Time int64 `json:"time"` + OK bool `json:"ok"` + StatusCode int32 `json:"statusCode"` + LatencyMs float64 `json:"latencyMs"` + Error string `json:"error"` + // over the range + Checks int32 `json:"checks"` + UptimePct float64 `json:"uptimePct"` + AvgLatencyMs float64 `json:"avgLatencyMs"` +} + +// getEndpoints lists the endpoints checked within the range +func (s *server) getEndpoints(w http.ResponseWriter, r *http.Request) { + query := r.URL.Query() + from, to, err := timeRange(query.Get("from"), query.Get("to")) + if err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + response, err := s.collector.Endpoints(r.Context(), &api.EndpointsRequest{From: from, To: to}) + if err != nil { + writeGRPCError(w, "endpoints", err) + return + } + endpoints := make([]endpointStatus, 0, len(response.Endpoints)) + for _, e := range response.Endpoints { + endpoints = append(endpoints, endpointStatus{ + Name: e.Name, + URL: e.Url, + Method: e.Method, + Time: e.Time, + OK: e.Ok, + StatusCode: e.StatusCode, + LatencyMs: e.LatencyMs, + Error: e.Error, + Checks: e.Checks, + UptimePct: e.UptimePct, + AvgLatencyMs: e.AvgLatencyMs, + }) + } + writeJSON(w, map[string]any{"endpoints": endpoints}) +} + +// getEndpointSeries returns an endpoint's latency or availability over time, +// in the same shape as a host's series +func (s *server) getEndpointSeries(w http.ResponseWriter, r *http.Request) { + query := r.URL.Query() + from, to, err := timeRange(query.Get("from"), query.Get("to")) + if err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + maxPoints, _ := strconv.Atoi(query.Get("maxPoints")) + response, err := s.collector.EndpointSeries(r.Context(), &api.EndpointSeriesRequest{ + Name: query.Get("name"), + Metric: query.Get("metric"), + From: from, + To: to, + MaxPoints: int32(maxPoints), + }) + if err != nil { + writeGRPCError(w, query.Get("metric")+" for endpoint "+query.Get("name"), err) + return + } + writeSeries(w, response) +} + type diskForecast struct { Device string `json:"device"` Mount string `json:"mount"` diff --git a/client/internal/server/server_test.go b/client/internal/server/server_test.go index c193a86..51dc74d 100644 --- a/client/internal/server/server_test.go +++ b/client/internal/server/server_test.go @@ -73,6 +73,21 @@ func (f *fakeCollector) ProcessUsage(ctx context.Context, in *api.ProcessUsageRe }}, nil } +func (f *fakeCollector) Endpoints(ctx context.Context, in *api.EndpointsRequest) (*api.EndpointList, error) { + return &api.EndpointList{Endpoints: []*api.EndpointStatus{ + {Name: "api", Url: "https://api.example.com", Method: "GET", Time: in.To, StatusCode: 0, Error: "connection refused", Checks: 30, UptimePct: 90, AvgLatencyMs: 110.5}, + }}, nil +} + +func (f *fakeCollector) EndpointSeries(ctx context.Context, in *api.EndpointSeriesRequest) (*api.SeriesResponse, error) { + if in.Metric != "latency" { + return nil, status.Error(codes.InvalidArgument, "invalid request: unknown endpoint metric") + } + return &api.SeriesResponse{Metric: in.Metric, Source: "raw", StepSeconds: 4, Series: []*api.Series{ + {Label: in.Name, Points: []*api.Point{{Time: 1700000000, Value: 100}}}, + }}, nil +} + func floatPtr(v float64) *float64 { return &v } @@ -149,6 +164,25 @@ func TestProcessUsage(t *testing.T) { } } +func TestEndpoints(t *testing.T) { + s, _ := newTestServer(t, nil) + code, body, _ := get(t, s, "/api/v1/endpoints?from=1700000000&to=1700003600") + want := `{"endpoints":[{"name":"api","url":"https://api.example.com","method":"GET","time":1700003600,"ok":false,` + + `"statusCode":0,"latencyMs":0,"error":"connection refused","checks":30,"uptimePct":90,"avgLatencyMs":110.5}]}` + if code != 200 || strings.TrimSpace(body) != want { + t.Errorf("unexpected response %d: %s", code, body) + } + + code, body, _ = get(t, s, "/api/v1/endpoints/series?name=api&metric=latency&from=1700000000&to=1700003600") + want = `{"metric":"latency","series":[{"label":"api","points":[[1700000000,100]]}],"source":"raw","stepSeconds":4}` + if code != 200 || strings.TrimSpace(body) != want { + t.Errorf("unexpected response %d: %s", code, body) + } + if code, _, _ := get(t, s, "/api/v1/endpoints/series?name=api&metric=nope"); code != http.StatusBadRequest { + t.Errorf("expected 400 for an unknown metric, got %d", code) + } +} + func TestDiskForecasts(t *testing.T) { s, _ := newTestServer(t, nil) code, body, _ := get(t, s, "/api/v1/hosts/web1/disk-forecasts") 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/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 4da8c54..15da3c5 100644 --- a/internal/api/api.go +++ b/internal/api/api.go @@ -267,7 +267,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 { @@ -275,7 +279,44 @@ 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 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 9357809..2248a62 100644 --- a/internal/api/api.pb.go +++ b/internal/api/api.pb.go @@ -1758,6 +1758,307 @@ func (x *SnapshotList) GetCustomMetrics() []*CustomValue { 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 = "" + @@ -1898,7 +2199,33 @@ const file_api_api_proto_rawDesc = "" + "\x04time\x18\x05 \x01(\x03R\x04time\"o\n" + "\fSnapshotList\x12'\n" + "\x05hosts\x18\x01 \x03(\v2\x11.api.HostSnapshotR\x05hosts\x126\n" + - "\rcustomMetrics\x18\x02 \x03(\v2\x10.api.CustomValueR\rcustomMetrics2\x82\x06\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\x80\a\n" + "\x12MonitorDataService\x123\n" + "\x06Enroll\x12\x12.api.EnrollRequest\x1a\x13.api.EnrollResponse\"\x00\x12-\n" + "\n" + @@ -1914,7 +2241,9 @@ 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\"\x00B)Z'github.com/dhamith93/SyMon/internal/apib\x06proto3" var ( file_api_api_proto_rawDescOnce sync.Once @@ -1928,35 +2257,39 @@ func file_api_api_proto_rawDescGZIP() []byte { return file_api_api_proto_rawDescData } -var file_api_api_proto_msgTypes = make([]protoimpl.MessageInfo, 27) +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 - (*CustomValue)(nil), // 25: api.CustomValue - (*SnapshotList)(nil), // 26: 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 @@ -1967,39 +2300,44 @@ var file_api_api_proto_depIdxs = []int32{ 23, // 5: api.ProcessUsageList.processes:type_name -> api.ProcessUsage 9, // 6: api.SnapshotList.hosts:type_name -> api.HostSnapshot 25, // 7: api.SnapshotList.customMetrics:type_name -> api.CustomValue - 4, // 8: api.MonitorDataService.Enroll:input_type -> api.EnrollRequest - 2, // 9: api.MonitorDataService.HandlePing:input_type -> api.ServerInfo - 2, // 10: api.MonitorDataService.InitAgent:input_type -> api.ServerInfo - 3, // 11: api.MonitorDataService.HandleMonitorData:input_type -> api.MonitorData - 3, // 12: api.MonitorDataService.HandleCustomMonitorData:input_type -> api.MonitorData - 0, // 13: api.MonitorDataService.Fleet:input_type -> api.Void - 8, // 14: api.MonitorDataService.Snapshot:input_type -> api.HostRequest - 10, // 15: api.MonitorDataService.QuerySeries:input_type -> api.SeriesRequest - 14, // 16: api.MonitorDataService.Processes:input_type -> api.ProcessesRequest - 8, // 17: api.MonitorDataService.CustomMetricNames:input_type -> api.HostRequest - 17, // 18: api.MonitorDataService.Alerts:input_type -> api.AlertsRequest - 8, // 19: api.MonitorDataService.DiskForecasts:input_type -> api.HostRequest - 22, // 20: api.MonitorDataService.ProcessUsage:input_type -> api.ProcessUsageRequest - 0, // 21: api.MonitorDataService.Snapshots:input_type -> api.Void - 5, // 22: api.MonitorDataService.Enroll:output_type -> api.EnrollResponse - 1, // 23: api.MonitorDataService.HandlePing:output_type -> api.Message - 1, // 24: api.MonitorDataService.InitAgent:output_type -> api.Message - 1, // 25: api.MonitorDataService.HandleMonitorData:output_type -> api.Message - 1, // 26: api.MonitorDataService.HandleCustomMonitorData:output_type -> api.Message - 7, // 27: api.MonitorDataService.Fleet:output_type -> api.FleetSummary - 9, // 28: api.MonitorDataService.Snapshot:output_type -> api.HostSnapshot - 13, // 29: api.MonitorDataService.QuerySeries:output_type -> api.SeriesResponse - 15, // 30: api.MonitorDataService.Processes:output_type -> api.ProcessesResponse - 16, // 31: api.MonitorDataService.CustomMetricNames:output_type -> api.NameList - 19, // 32: api.MonitorDataService.Alerts:output_type -> api.AlertList - 21, // 33: api.MonitorDataService.DiskForecasts:output_type -> api.DiskForecastList - 24, // 34: api.MonitorDataService.ProcessUsage:output_type -> api.ProcessUsageList - 26, // 35: api.MonitorDataService.Snapshots:output_type -> api.SnapshotList - 22, // [22:36] is the sub-list for method output_type - 8, // [8:22] is the sub-list for method input_type - 8, // [8:8] is the sub-list for extension type_name - 8, // [8:8] is the sub-list for extension extendee - 0, // [0:8] is the sub-list for field type_name + 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 + 5, // 25: api.MonitorDataService.Enroll:output_type -> api.EnrollResponse + 1, // 26: api.MonitorDataService.HandlePing:output_type -> api.Message + 1, // 27: api.MonitorDataService.InitAgent:output_type -> api.Message + 1, // 28: api.MonitorDataService.HandleMonitorData:output_type -> api.Message + 1, // 29: api.MonitorDataService.HandleCustomMonitorData:output_type -> api.Message + 7, // 30: api.MonitorDataService.Fleet:output_type -> api.FleetSummary + 9, // 31: api.MonitorDataService.Snapshot:output_type -> api.HostSnapshot + 13, // 32: api.MonitorDataService.QuerySeries:output_type -> api.SeriesResponse + 15, // 33: api.MonitorDataService.Processes:output_type -> api.ProcessesResponse + 16, // 34: api.MonitorDataService.CustomMetricNames:output_type -> api.NameList + 19, // 35: api.MonitorDataService.Alerts:output_type -> api.AlertList + 21, // 36: api.MonitorDataService.DiskForecasts:output_type -> api.DiskForecastList + 24, // 37: api.MonitorDataService.ProcessUsage:output_type -> api.ProcessUsageList + 26, // 38: api.MonitorDataService.Snapshots:output_type -> api.SnapshotList + 29, // 39: api.MonitorDataService.Endpoints:output_type -> api.EndpointList + 13, // 40: api.MonitorDataService.EndpointSeries:output_type -> api.SeriesResponse + 25, // [25:41] is the sub-list for method output_type + 9, // [9:25] 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() } @@ -2015,7 +2353,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: 27, + NumMessages: 31, NumExtensions: 0, NumServices: 1, }, diff --git a/internal/api/api.proto b/internal/api/api.proto index 979ed24..8f3ef37 100644 --- a/internal/api/api.proto +++ b/internal/api/api.proto @@ -201,6 +201,42 @@ message SnapshotList { 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 { // agent rpc Enroll(EnrollRequest) returns (EnrollResponse) {} @@ -220,4 +256,6 @@ 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) {} } diff --git a/internal/api/api_grpc.pb.go b/internal/api/api_grpc.pb.go index 80b5f5f..3b13a91 100644 --- a/internal/api/api_grpc.pb.go +++ b/internal/api/api_grpc.pb.go @@ -33,6 +33,8 @@ 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" ) // MonitorDataServiceClient is the client API for MonitorDataService service. @@ -56,6 +58,8 @@ 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) } type monitorDataServiceClient struct { @@ -206,6 +210,26 @@ 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 +} + // MonitorDataServiceServer is the server API for MonitorDataService service. // All implementations must embed UnimplementedMonitorDataServiceServer // for forward compatibility. @@ -227,6 +251,8 @@ 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) mustEmbedUnimplementedMonitorDataServiceServer() } @@ -279,6 +305,12 @@ 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) mustEmbedUnimplementedMonitorDataServiceServer() {} func (UnimplementedMonitorDataServiceServer) testEmbeddedByValue() {} @@ -552,6 +584,42 @@ 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) +} + // 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 +683,14 @@ var MonitorDataService_ServiceDesc = grpc.ServiceDesc{ MethodName: "Snapshots", Handler: _MonitorDataService_Snapshots_Handler, }, + { + MethodName: "Endpoints", + Handler: _MonitorDataService_Endpoints_Handler, + }, + { + MethodName: "EndpointSeries", + Handler: _MonitorDataService_EndpointSeries_Handler, + }, }, Streams: []grpc.StreamDesc{}, Metadata: "api/api.proto", diff --git a/internal/monitor/monitor.go b/internal/monitor/monitor.go index 8551e17..2ca2094 100644 --- a/internal/monitor/monitor.go +++ b/internal/monitor/monitor.go @@ -30,8 +30,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 ( 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/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 9acc99a..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 { From 2519821c3d13f08e4bfe9b12b8f1e98febb60f7d Mon Sep 17 00:00:00 2001 From: Dhamith Hewamullage Date: Tue, 29 Sep 2026 21:51:30 +0530 Subject: [PATCH 6/7] add an endpoints page --- README.MD | 9 +- client/web/src/App.svelte | 8 +- client/web/src/lib/api.ts | 20 +++ client/web/src/lib/format.test.ts | 13 +- client/web/src/lib/format.ts | 16 +- client/web/src/lib/router.svelte.ts | 2 + client/web/src/pages/Alerts.svelte | 8 +- client/web/src/pages/Endpoints.svelte | 217 ++++++++++++++++++++++++++ docs/install.md | 6 +- 9 files changed, 292 insertions(+), 7 deletions(-) create mode 100644 client/web/src/pages/Endpoints.svelte diff --git a/README.MD b/README.MD index ec1157c..361e938 100644 --- a/README.MD +++ b/README.MD @@ -20,6 +20,9 @@ SyMon is a self-hosted monitoring tool for Linux servers, home labs and Raspberr - Every running container on a host: Docker, Podman, containerd, LXC and systemd-nspawn - CPU, memory, network and disk IO per container, grouped by Compose project +**Endpoints** +- HTTP checks of any URL from the server, with response time and uptime history + **Custom metrics** - Send any number from a script or cron job and get a chart for it @@ -113,6 +116,10 @@ The Client exposes a JSON API under `/api/v1`. Times are unix seconds. Errors re * `GET /api/v1/hosts/{host}/custom-metrics` * Names of the host's custom metrics * `GET /api/v1/alerts?host=&open=1&from=&to=` - * Alerts, newest first. `open=1` leaves out resolved ones + * Alerts, newest first. `open=1` leaves out resolved ones. Endpoint alerts have an empty `host` +* `GET /api/v1/endpoints?from=&to=` + * Every endpoint checked within the range: its newest check, the number of checks, the share that passed and the average response time +* `GET /api/v1/endpoints/series?name=&metric=latency&from=&to=` + * An endpoint's response time in milliseconds (`latency`) or share of passing checks (`availability`) over time, in the same shape as a host's series The Client also serves `GET /metrics`, every host's latest values in the Prometheus text format, unless `SYMON_CLIENT_METRICS_ENABLED=false`. See [Prometheus and Grafana](docs/install.md#prometheus-and-grafana). diff --git a/client/web/src/App.svelte b/client/web/src/App.svelte index 0a59aae..119f0c4 100644 --- a/client/web/src/App.svelte +++ b/client/web/src/App.svelte @@ -3,11 +3,14 @@ import { setTheme, theme, type ThemeChoice } from './lib/theme.svelte'; import Alerts from './pages/Alerts.svelte'; import CustomMetrics from './pages/CustomMetrics.svelte'; + import Endpoints from './pages/Endpoints.svelte'; import Fleet from './pages/Fleet.svelte'; import Host from './pages/Host.svelte'; const page = $derived(match(location.path)); const onAlerts = $derived(page.name === 'alerts'); + const onEndpoints = $derived(page.name === 'endpoints'); + const onHosts = $derived(!onAlerts && !onEndpoints); @@ -22,7 +25,8 @@ SyMon