diff --git a/apps/checker/checker/dns.go b/apps/checker/checker/dns.go index 4ef2b07f..b2383c2c 100644 --- a/apps/checker/checker/dns.go +++ b/apps/checker/checker/dns.go @@ -9,20 +9,24 @@ import ( "github.com/rs/zerolog/log" ) - type DnsResponse struct { A []string `json:"a,omitempty"` AAAA []string `json:"aaaa,omitempty"` CNAME string `json:"cname,omitempty"` MX []string `json:"mx,omitempty"` - NS []string `json:"ns,omitempty"` - TXT []string `json:"txt,omitempty"` + NS []string `json:"ns,omitempty"` + TXT []string `json:"txt,omitempty"` } +// resolver is net.DefaultResolver reached through its context-aware methods: +// the package-level net.Lookup* helpers take no context, so a stalled resolver +// would block a probe past its timeout and hold up the monitor's next run. +var resolver = net.DefaultResolver + func Dns(ctx context.Context, host string) (*DnsResponse, error) { - logger:= log.Ctx(ctx).With().Str("monitor", host).Logger() + logger := log.Ctx(ctx).With().Str("monitor", host).Logger() - ips, err := net.LookupIP(host) + ips, err := resolver.LookupIP(ctx, "ip", host) if err != nil { logger.Error().Err(err).Msg("DNS IP lookup failed") return nil, fmt.Errorf("failed to lookup IPs: %w", err) @@ -38,20 +42,26 @@ func Dns(ctx context.Context, host string) (*DnsResponse, error) { AAAA = append(AAAA, ip.String()) } } - CNAME,err := lookupCNAME(host) + CNAME, err := lookupCNAME(ctx, host) if err != nil { logger.Error().Err(err).Msg("DNS CNAME record lookup failed") return nil, fmt.Errorf("failed to lookup CNAME record: %w", err) } - MXRecords := lookupMX(host) + MXRecords := lookupMX(ctx, host) - NS,err := lookupNS(host) + NS, err := lookupNS(ctx, host) if err != nil { logger.Error().Err(err).Msg("DNS NS record lookup failed") return nil, fmt.Errorf("failed to lookup NS record: %w", err) } - TXT := lookupTXT(host) + TXT := lookupTXT(ctx, host) + // MX and TXT tolerate lookup errors, but an expired deadline means those + // records are unknown rather than absent — don't report that as a success. + if err := ctx.Err(); err != nil { + logger.Error().Err(err).Msg("DNS lookup did not complete before the deadline") + return nil, fmt.Errorf("DNS lookup for %s did not complete: %w", host, err) + } response := &DnsResponse{ A: A, @@ -59,16 +69,27 @@ func Dns(ctx context.Context, host string) (*DnsResponse, error) { CNAME: CNAME, MX: MXRecords, NS: NS, - TXT: TXT, + TXT: TXT, } return response, nil } +// FormatDNSRecords flattens a lookup into the per-record-type map shape that +// both the public handler and the private-location probe report. +func FormatDNSRecords(result *DnsResponse) map[string][]string { + return map[string][]string{ + "A": append([]string{}, result.A...), + "AAAA": append([]string{}, result.AAAA...), + "CNAME": {result.CNAME}, + "MX": append([]string{}, result.MX...), + "NS": append([]string{}, result.NS...), + "TXT": append([]string{}, result.TXT...), + } +} - -func lookupCNAME(domain string) (string, error) { - cname, err := net.LookupCNAME(domain) +func lookupCNAME(ctx context.Context, domain string) (string, error) { + cname, err := resolver.LookupCNAME(ctx, domain) if err != nil { return "", err } @@ -76,10 +97,9 @@ func lookupCNAME(domain string) (string, error) { return cname, nil } -func lookupMX(domain string) ([]string) { +func lookupMX(ctx context.Context, domain string) []string { mx := []string{} - mxRecords,_ := net.LookupMX(domain) - + mxRecords, _ := resolver.LookupMX(ctx, domain) for _, r := range mxRecords { mx = append(mx, fmt.Sprintf("%s:%d", r.Host, r.Pref)) @@ -87,14 +107,14 @@ func lookupMX(domain string) ([]string) { return mx } -func lookupNS(domain string) ([]string, error) { +func lookupNS(ctx context.Context, domain string) ([]string, error) { hosts := []string{} isSubdomain := isSubdomain(domain) if isSubdomain { return hosts, nil } - nsRecords, err := net.LookupNS(domain) + nsRecords, err := resolver.LookupNS(ctx, domain) if err != nil { return nil, err } @@ -105,9 +125,9 @@ func lookupNS(domain string) ([]string, error) { return hosts, nil } -func lookupTXT(domain string) ([]string) { +func lookupTXT(ctx context.Context, domain string) []string { records := []string{} - txtRecords, err := net.LookupTXT(domain) + txtRecords, err := resolver.LookupTXT(ctx, domain) if err != nil { return nil } @@ -118,7 +138,6 @@ func lookupTXT(domain string) ([]string) { return records } - func isSubdomain(domain string) bool { parent := strings.Split(domain, ".") if len(parent) < 3 { diff --git a/apps/checker/checker/dns_test.go b/apps/checker/checker/dns_test.go index 5b352d03..d8d34fca 100644 --- a/apps/checker/checker/dns_test.go +++ b/apps/checker/checker/dns_test.go @@ -1,12 +1,15 @@ package checker_test import ( + "context" + "errors" + "net" "testing" + "time" "github.com/openstatushq/openstatus/apps/checker/checker" ) - func TestPingDNS(t *testing.T) { ctx := t.Context() data, err := checker.Dns(ctx, "openstat.us") @@ -17,3 +20,57 @@ func TestPingDNS(t *testing.T) { t.Errorf("Dns() A records = %v", data.A) } } + +// A stalled resolver must not outlive the deadline: the probe runs as a +// single-instance job, so a hung lookup blocks the monitor's next runs. +func TestDns_HonoursContextDeadline(t *testing.T) { + useBlackholeResolver(t) + + ctx, cancel := context.WithTimeout(context.Background(), 250*time.Millisecond) + defer cancel() + + done := make(chan error, 1) + go func() { + _, err := checker.Dns(ctx, "openstatus.dev") + done <- err + }() + + select { + case err := <-done: + if err == nil { + t.Fatal("expected a lookup error against a blackholed resolver") + } + case <-time.After(10 * time.Second): + t.Fatal("Dns() ignored the context deadline and is still blocked") + } +} + +func TestDns_ReturnsErrorOnAlreadyCancelledContext(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + if _, err := checker.Dns(ctx, "openstatus.dev"); !errors.Is(err, context.Canceled) { + t.Errorf("expected a context.Canceled error, got %v", err) + } +} + +// Points the package resolver at a UDP socket that accepts queries and never +// answers, so lookups hang until the context stops them. +func useBlackholeResolver(t *testing.T) { + t.Helper() + + conn, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)}) + if err != nil { + t.Fatalf("failed to open blackhole resolver: %v", err) + } + t.Cleanup(func() { conn.Close() }) + + addr := conn.LocalAddr().String() + checker.SetResolverForTest(t, &net.Resolver{ + PreferGo: true, + Dial: func(ctx context.Context, network, _ string) (net.Conn, error) { + var d net.Dialer + return d.DialContext(ctx, network, addr) + }, + }) +} diff --git a/apps/checker/checker/export_test.go b/apps/checker/checker/export_test.go new file mode 100644 index 00000000..46c0c9e6 --- /dev/null +++ b/apps/checker/checker/export_test.go @@ -0,0 +1,16 @@ +package checker + +import ( + "net" + "testing" +) + +// SetResolverForTest swaps the package resolver so tests can point lookups at a +// stub, and restores it when the test ends. +func SetResolverForTest(t *testing.T, r *net.Resolver) { + t.Helper() + + previous := resolver + resolver = r + t.Cleanup(func() { resolver = previous }) +} diff --git a/apps/checker/cmd/private/main.go b/apps/checker/cmd/private/main.go index 1c6f5f75..9caf4422 100644 --- a/apps/checker/cmd/private/main.go +++ b/apps/checker/cmd/private/main.go @@ -20,9 +20,16 @@ import ( const ( configRefreshInterval = 10 * time.Minute + platformTimeout = 30 * time.Second ) func main() { + apiKey := getEnv("OPENSTATUS_KEY", "") + if apiKey == "" { + fmt.Fprintln(os.Stderr, "OPENSTATUS_KEY is required: the probe cannot authenticate against openstatus") + os.Exit(1) + } + ctx, cancel := context.WithCancel(context.Background()) defer cancel() @@ -37,8 +44,6 @@ func main() { s := tasks.New() defer s.Stop() - apiKey := getEnv("OPENSTATUS_KEY", "") - monitorManager := scheduler.MonitorManager{ Client: getClient(apiKey), JobRunner: job.NewJobRunner(), @@ -69,8 +74,12 @@ func getEnv(key, fallback string) string { func getClient(apiKey string) v1.PrivateLocationServiceClient { ingestUrl := getEnv("OPENSTATUS_INGEST_URL", "https://openstatus-private-location.fly.dev") + // Tasks run with RunSingleInstance, so an untimed request that hangs would + // wedge that monitor: it never checks again until the probe restarts. + httpClient := &http.Client{Timeout: platformTimeout} + client := v1.NewPrivateLocationServiceClient( - http.DefaultClient, + httpClient, ingestUrl, connect.WithHTTPGet(), connect.WithInterceptors(NewAuthInterceptor(apiKey)), diff --git a/apps/checker/handlers/dns.go b/apps/checker/handlers/dns.go index 11a6dbd7..e53f2c00 100644 --- a/apps/checker/handlers/dns.go +++ b/apps/checker/handlers/dns.go @@ -169,7 +169,7 @@ func (h Handler) DNSHandler(c *gin.Context) { result, err := backoff.Retry(ctx, op, backoff.WithBackOff(backoff.NewExponentialBackOff()), backoff.WithMaxTries(uint(retry))) data.Latency = latency if result != nil { - data.Records = FormatDNSResult(result) + data.Records = checker.FormatDNSRecords(result) } if len(req.RawAssertions) > 0 { @@ -356,7 +356,7 @@ func (h Handler) DNSHandlerRegion(c *gin.Context) { return } - data.Records = FormatDNSResult(result) + data.Records = checker.FormatDNSRecords(result) if req.RequestId != 0 { if tbEvent, err := data.tinybirdEvent(); err != nil { log.Ctx(ctx).Error().Err(err).Msg("failed to marshal dns records") @@ -369,40 +369,6 @@ func (h Handler) DNSHandlerRegion(c *gin.Context) { } -func FormatDNSResult(result *checker.DnsResponse) map[string][]string { - r := make(map[string][]string) - a := make([]string, 0) - aaaa := make([]string, 0) - mx := make([]string, 0) - ns := make([]string, 0) - txt := make([]string, 0) - - for _, v := range result.A { - a = append(a, v) - } - r["A"] = a - - for _, v := range result.AAAA { - aaaa = append(aaaa, v) - } - r["AAAA"] = aaaa - - r["CNAME"] = []string{result.CNAME} - for _, v := range result.MX { - mx = append(mx, v) - } - r["MX"] = mx - for _, v := range result.NS { - ns = append(ns, v) - } - r["NS"] = ns - for _, v := range result.TXT { - txt = append(txt, v) - } - r["TXT"] = txt - return r -} - func EvaluateDNSAssertions(rawAssertions []json.RawMessage, response *checker.DnsResponse) (bool, error) { for _, a := range rawAssertions { var assert assertions.RecordTarget diff --git a/apps/checker/handlers/dns_test.go b/apps/checker/handlers/dns_test.go index d478804f..f921d5f7 100644 --- a/apps/checker/handlers/dns_test.go +++ b/apps/checker/handlers/dns_test.go @@ -10,7 +10,7 @@ import ( "github.com/openstatushq/openstatus/apps/checker/handlers" ) -// Mock DNSResult struct to match the expected input for FormatDNSResult. +// Mock DNSResult struct to match the expected input for FormatDNSRecords. // If the real struct is in another package, import it accordingly. type DNSResult struct { A []string @@ -21,7 +21,7 @@ type DNSResult struct { TXT []string } -func TestFormatDNSResult(t *testing.T) { +func TestFormatDNSRecords(t *testing.T) { tests := []struct { name string input DNSResult @@ -88,7 +88,7 @@ func TestFormatDNSResult(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - got := handlers.FormatDNSResult(&checker.DnsResponse{ + got := checker.FormatDNSRecords(&checker.DnsResponse{ A: tt.input.A, AAAA: tt.input.AAAA, CNAME: tt.input.CNAME, @@ -97,7 +97,7 @@ func TestFormatDNSResult(t *testing.T) { TXT: tt.input.TXT, }) if !reflect.DeepEqual(got, tt.expected) { - t.Errorf("FormatDNSResult() = %v, want %v", got, tt.expected) + t.Errorf("FormatDNSRecords() = %v, want %v", got, tt.expected) } }) } diff --git a/apps/checker/pkg/job/dns_job.go b/apps/checker/pkg/job/dns_job.go index 8cac0fff..4d2325fe 100644 --- a/apps/checker/pkg/job/dns_job.go +++ b/apps/checker/pkg/job/dns_job.go @@ -2,13 +2,193 @@ package job import ( "context" + "errors" + "fmt" + "time" + "github.com/cenkalti/backoff/v5" + "github.com/google/uuid" + "github.com/openstatushq/openstatus/apps/checker/checker" + "github.com/openstatushq/openstatus/apps/checker/pkg/assertions" v1 "github.com/openstatushq/openstatus/apps/checker/proto/private_location/v1" + "github.com/openstatushq/openstatus/apps/checker/request" ) -type DNSPrivateRegionData struct {} +type DNSPrivateRegionData struct { + ID string `json:"id"` + URI string `json:"uri"` + RequestStatus string `json:"requestStatus,omitempty"` + Message string `json:"message,omitempty"` + Records map[string][]string `json:"records"` + Latency int64 `json:"latency"` + CronTimestamp int64 `json:"cronTimestamp"` + Timestamp int64 `json:"timestamp"` + Error uint8 `json:"error"` +} + +func ProtoRecordAssertionToComparator(assertion v1.RecordComparator) (request.RecordComparator, error) { + switch assertion { + case v1.RecordComparator_RECORD_COMPARATOR_EQUAL: + return request.RecordEquals, nil + case v1.RecordComparator_RECORD_COMPARATOR_NOT_EQUAL: + return request.RecordNotEquals, nil + case v1.RecordComparator_RECORD_COMPARATOR_CONTAINS: + return request.RecordContains, nil + case v1.RecordComparator_RECORD_COMPARATOR_NOT_CONTAINS: + return request.RecordNotContains, nil + } + return "", fmt.Errorf("unknown comparator type: %v", assertion) +} + +func evaluateRecordAssertions(recordAssertions []*v1.RecordAssertion, res *checker.DnsResponse) (bool, error) { + for _, recordAssertion := range recordAssertions { + comparator, err := ProtoRecordAssertionToComparator(recordAssertion.GetComparator()) + if err != nil { + return false, fmt.Errorf("error while parsing record assertion comparator: %w", err) + } + + assert := assertions.RecordTarget{ + Comparator: comparator, + Target: recordAssertion.GetTarget(), + Key: request.Record(recordAssertion.GetRecord()), + } + + var isSuccessful bool + switch assert.Key { + case request.RecordA: + isSuccessful = assert.RecordEvaluate(res.A) + case request.RecordAAAA: + isSuccessful = assert.RecordEvaluate(res.AAAA) + case request.RecordCNAME: + isSuccessful = assert.RecordEvaluate([]string{res.CNAME}) + case request.RecordMX: + isSuccessful = assert.RecordEvaluate(res.MX) + case request.RecordNS: + isSuccessful = assert.RecordEvaluate(res.NS) + case request.RecordTXT: + isSuccessful = assert.RecordEvaluate(res.TXT) + default: + return false, fmt.Errorf("unknown record type in assertion: %s", assert.Key) + } + + if !isSuccessful { + return false, nil + } + } + + return true, nil +} + +// Cancellation cause for the retry budget, so an expiry we imposed can be told +// apart from the caller cancelling us (shutdown), which must not be reported as +// an outage. +var errRetryBudgetExpired = errors.New("DNS retry budget expired") + +func (jobRunner) DNSJob(ctx context.Context, monitor *v1.DNSMonitor) (*DNSPrivateRegionData, error) { + retry := monitor.Retry + if retry == 0 { + retry = 3 + } + + var degradedAfter int64 + if monitor.DegradedAt != nil { + degradedAfter = *monitor.DegradedAt + } + + // One budget for the whole retry loop: checker.Dns honours the deadline, so + // each attempt is bounded by whatever is left of it. + if monitor.Timeout > 0 { + var cancel context.CancelFunc + ctx, cancel = context.WithTimeoutCause(ctx, time.Duration(monitor.Timeout)*time.Millisecond, errRetryBudgetExpired) + defer cancel() + } + + var called int + var lastFailure *DNSPrivateRegionData + + op := func() (*DNSPrivateRegionData, error) { + called++ + start := time.Now().UTC().UnixMilli() + res, lookupErr := checker.Dns(ctx, monitor.Uri) + latency := time.Now().UTC().UnixMilli() - start + + data, err := newDNSData(monitor.Uri, start, latency) + if err != nil { + return nil, err + } + + if lookupErr != nil { + // Build the reportable failure on every attempt, not just the last: + // a dropped result never reaches the platform, so a resolver outage + // would go unalerted. + data.RequestStatus = "error" + data.Error = 1 + data.Message = lookupErr.Error() + lastFailure = data + + if called < int(retry) { + return nil, fmt.Errorf("DNS lookup failed for %s: %w", monitor.Uri, lookupErr) + } + return data, nil + } + + data.Records = checker.FormatDNSRecords(res) + + isSuccessful, assertErr := evaluateRecordAssertions(monitor.RecordAssertions, res) + if assertErr != nil { + // Returning nil here would stop the monitor reporting entirely and + // leave it looking healthy. Retrying can't help — a malformed + // assertion stays malformed — so report it and stop. + data.RequestStatus = "error" + data.Error = 1 + data.Message = fmt.Sprintf("invalid DNS assertion for %s: %s", monitor.Uri, assertErr) + return data, nil + } + + switch { + case !isSuccessful: + data.RequestStatus = "error" + data.Error = 1 + data.Message = fmt.Sprintf("DNS assertions failed for %s", monitor.Uri) + lastFailure = data + + if called < int(retry) { + return nil, errors.New(data.Message) + } + case degradedAfter > 0 && latency > degradedAfter: + data.RequestStatus = "degraded" + default: + data.RequestStatus = "success" + } + + return data, nil + } + + data, err := backoff.Retry(ctx, op, + backoff.WithMaxTries(uint(retry)), + backoff.WithBackOff(backoff.NewExponentialBackOff()), + ) + // backoff.Retry discards the operation result when the context expires, so + // without this the pending failure is lost and no outage is ingested. + if err != nil && lastFailure != nil && errors.Is(err, errRetryBudgetExpired) { + return lastFailure, nil + } + + return data, err +} -func (jobRunner) DNSJob(ctx context.Context, monitor *v1.DNSMonitor) ( *DNSPrivateRegionData, error) { +func newDNSData(uri string, start, latency int64) (*DNSPrivateRegionData, error) { + id, err := uuid.NewV7() + if err != nil { + return nil, fmt.Errorf("error while generating uuid: %w", err) + } - return nil,nil + return &DNSPrivateRegionData{ + ID: id.String(), + URI: uri, + Latency: latency, + Timestamp: start, + CronTimestamp: start, + Records: map[string][]string{}, + }, nil } diff --git a/apps/checker/pkg/job/dns_job_internal_test.go b/apps/checker/pkg/job/dns_job_internal_test.go new file mode 100644 index 00000000..9311402d --- /dev/null +++ b/apps/checker/pkg/job/dns_job_internal_test.go @@ -0,0 +1,97 @@ +package job + +import ( + "testing" + + "github.com/openstatushq/openstatus/apps/checker/checker" + v1 "github.com/openstatushq/openstatus/apps/checker/proto/private_location/v1" +) + +func TestEvaluateRecordAssertions(t *testing.T) { + response := &checker.DnsResponse{ + A: []string{"1.2.3.4", "5.6.7.8"}, + AAAA: []string{"::1"}, + CNAME: "openstatus.dev.", + MX: []string{"mx1.openstatus.dev:10"}, + NS: []string{"ns1.openstatus.dev"}, + TXT: []string{"v=spf1"}, + } + + tests := []struct { + name string + assertions []*v1.RecordAssertion + want bool + wantErr bool + }{ + { + name: "no assertions", + assertions: nil, + want: true, + }, + { + name: "A record matches", + assertions: []*v1.RecordAssertion{ + {Record: "A", Comparator: v1.RecordComparator_RECORD_COMPARATOR_EQUAL, Target: "1.2.3.4"}, + }, + want: true, + }, + { + name: "A record does not match", + assertions: []*v1.RecordAssertion{ + {Record: "A", Comparator: v1.RecordComparator_RECORD_COMPARATOR_EQUAL, Target: "9.9.9.9"}, + }, + want: false, + }, + { + name: "CNAME contains", + assertions: []*v1.RecordAssertion{ + {Record: "CNAME", Comparator: v1.RecordComparator_RECORD_COMPARATOR_CONTAINS, Target: "openstatus.dev"}, + }, + want: true, + }, + { + name: "TXT not contains", + assertions: []*v1.RecordAssertion{ + {Record: "TXT", Comparator: v1.RecordComparator_RECORD_COMPARATOR_NOT_CONTAINS, Target: "dkim"}, + }, + want: true, + }, + { + name: "every assertion must hold", + assertions: []*v1.RecordAssertion{ + {Record: "A", Comparator: v1.RecordComparator_RECORD_COMPARATOR_EQUAL, Target: "1.2.3.4"}, + {Record: "NS", Comparator: v1.RecordComparator_RECORD_COMPARATOR_EQUAL, Target: "ns2.openstatus.dev"}, + }, + want: false, + }, + { + name: "unknown record type", + assertions: []*v1.RecordAssertion{ + {Record: "FOO", Comparator: v1.RecordComparator_RECORD_COMPARATOR_EQUAL, Target: "bar"}, + }, + wantErr: true, + }, + { + name: "unspecified comparator", + assertions: []*v1.RecordAssertion{ + {Record: "A", Comparator: v1.RecordComparator_RECORD_COMPARATOR_UNSPECIFIED, Target: "1.2.3.4"}, + }, + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := evaluateRecordAssertions(tt.assertions, response) + if (err != nil) != tt.wantErr { + t.Fatalf("error = %v, wantErr %v", err, tt.wantErr) + } + if err != nil { + return + } + if got != tt.want { + t.Errorf("evaluateRecordAssertions() = %v, want %v", got, tt.want) + } + }) + } +} diff --git a/apps/checker/pkg/job/dns_job_test.go b/apps/checker/pkg/job/dns_job_test.go new file mode 100644 index 00000000..7cf161f0 --- /dev/null +++ b/apps/checker/pkg/job/dns_job_test.go @@ -0,0 +1,203 @@ +package job_test + +import ( + "context" + "errors" + "testing" + + "github.com/openstatushq/openstatus/apps/checker/pkg/job" + v1 "github.com/openstatushq/openstatus/apps/checker/proto/private_location/v1" +) + +func TestDNSJob_Success(t *testing.T) { + monitor := &v1.DNSMonitor{ + Uri: "openstatus.dev", + Timeout: 5000, + Retry: 1, + } + + data, err := job.NewJobRunner().DNSJob(context.Background(), monitor) + if err != nil { + t.Fatalf("expected no error, got %v", err) + } + if data.RequestStatus != "success" { + t.Errorf("expected RequestStatus 'success', got '%s'", data.RequestStatus) + } + if data.Error != 0 { + t.Errorf("expected Error 0, got %d", data.Error) + } + if len(data.Records["A"]) == 0 { + t.Errorf("expected at least one A record, got %v", data.Records) + } + if data.Timestamp <= 0 { + t.Errorf("expected a positive timestamp, got %d", data.Timestamp) + } +} + +// An assertion the probe can't interpret — typically a comparator or record +// type added by a newer server than this checker build — must still produce a +// datapoint. Dropping it leaves the monitor blank while looking healthy. +func TestDNSJob_UnsupportedAssertionIsReported(t *testing.T) { + tests := []struct { + name string + assertion *v1.RecordAssertion + }{ + { + name: "unknown comparator", + assertion: &v1.RecordAssertion{ + Record: "A", + Comparator: v1.RecordComparator(9999), + Target: "1.2.3.4", + }, + }, + { + name: "unspecified comparator", + assertion: &v1.RecordAssertion{ + Record: "A", + Comparator: v1.RecordComparator_RECORD_COMPARATOR_UNSPECIFIED, + Target: "1.2.3.4", + }, + }, + { + name: "unknown record type", + assertion: &v1.RecordAssertion{ + Record: "SOA", + Comparator: v1.RecordComparator_RECORD_COMPARATOR_EQUAL, + Target: "ns1.openstatus.dev", + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + monitor := &v1.DNSMonitor{ + Uri: "openstatus.dev", + Timeout: 5000, + Retry: 1, + RecordAssertions: []*v1.RecordAssertion{tt.assertion}, + } + + data, err := job.NewJobRunner().DNSJob(context.Background(), monitor) + if err != nil { + t.Fatalf("expected no error, got %v", err) + } + if data == nil { + t.Fatal("expected a reportable datapoint, got nil") + } + if data.RequestStatus != "error" { + t.Errorf("expected RequestStatus 'error', got '%s'", data.RequestStatus) + } + if data.Error != 1 { + t.Errorf("expected Error 1, got %d", data.Error) + } + if data.Message == "" { + t.Errorf("expected a message explaining the misconfiguration") + } + }) + } +} + +// A resolver failure has to come back as a reportable result, not an error: +// returning an error drops the datapoint and the outage goes unalerted. +func TestDNSJob_LookupFailureIsReported(t *testing.T) { + monitor := &v1.DNSMonitor{ + Uri: "openstatus-does-not-resolve.invalid", + Timeout: 5000, + Retry: 1, + } + + data, err := job.NewJobRunner().DNSJob(context.Background(), monitor) + if err != nil { + t.Fatalf("expected no error, got %v", err) + } + if data.RequestStatus != "error" { + t.Errorf("expected RequestStatus 'error', got '%s'", data.RequestStatus) + } + if data.Error != 1 { + t.Errorf("expected Error 1, got %d", data.Error) + } + if data.Message == "" { + t.Errorf("expected a failure message to forward to the platform") + } +} + +// The retry budget running out mid-loop must still report the pending failure: +// backoff.Retry drops the operation result on context expiry, which would +// otherwise leave a resolver outage silently un-ingested. +func TestDNSJob_LookupFailureIsReportedWhenRetryBudgetExpires(t *testing.T) { + monitor := &v1.DNSMonitor{ + Uri: "openstatus-does-not-resolve.invalid", + // Shorter than the backoff between attempts, so the deadline fires + // while retries are still pending. + Timeout: 1, + Retry: 3, + } + + data, err := job.NewJobRunner().DNSJob(context.Background(), monitor) + if err != nil { + t.Fatalf("expected no error, got %v", err) + } + if data == nil { + t.Fatal("expected a reportable datapoint, got nil") + } + if data.RequestStatus != "error" { + t.Errorf("expected RequestStatus 'error', got '%s'", data.RequestStatus) + } + if data.Error != 1 { + t.Errorf("expected Error 1, got %d", data.Error) + } + if data.Message == "" { + t.Errorf("expected a failure message to forward to the platform") + } +} + +// A cancelled parent is us shutting down, not an outage — it must not be +// reported as one. +func TestDNSJob_CallerCancellationIsNotReportedAsOutage(t *testing.T) { + monitor := &v1.DNSMonitor{ + Uri: "openstatus-does-not-resolve.invalid", + Timeout: 5000, + Retry: 3, + } + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + data, err := job.NewJobRunner().DNSJob(ctx, monitor) + if err == nil { + t.Fatalf("expected an error, got data %v", data) + } + if !errors.Is(err, context.Canceled) { + t.Errorf("expected a context.Canceled error, got %v", err) + } +} + +func TestDNSJob_FailedAssertionIsReported(t *testing.T) { + monitor := &v1.DNSMonitor{ + Uri: "openstatus.dev", + Timeout: 5000, + Retry: 1, + RecordAssertions: []*v1.RecordAssertion{ + { + Record: "A", + Comparator: v1.RecordComparator_RECORD_COMPARATOR_EQUAL, + Target: "203.0.113.1", + }, + }, + } + + data, err := job.NewJobRunner().DNSJob(context.Background(), monitor) + if err != nil { + t.Fatalf("expected no error, got %v", err) + } + if data.RequestStatus != "error" { + t.Errorf("expected RequestStatus 'error', got '%s'", data.RequestStatus) + } + if data.Error != 1 { + t.Errorf("expected Error 1, got %d", data.Error) + } + // The records still have to reach Tinybird so the failure can be debugged. + if len(data.Records["A"]) == 0 { + t.Errorf("expected the resolved records to be reported, got %v", data.Records) + } +} diff --git a/apps/checker/pkg/job/http_job.go b/apps/checker/pkg/job/http_job.go index 1f05b54c..c0161e33 100644 --- a/apps/checker/pkg/job/http_job.go +++ b/apps/checker/pkg/job/http_job.go @@ -54,6 +54,19 @@ func ProtoStringAssertionToComparator(assertion v1.StringComparator) (request.St return "", fmt.Errorf("unknown comparator type: %v", assertion) } +// httpFailureMessage explains a failed check in the alert body: checker.Http +// only fills Error for transport failures such as timeouts. +func httpFailureMessage(res checker.Response, statusOK bool) string { + switch { + case res.Error != "": + return res.Error + case !statusOK: + return fmt.Sprintf("Request failed with status code %d", res.Status) + default: + return "Assertions failed" + } +} + func (jr jobRunner) HTTPJob(ctx context.Context, monitor *v1.HTTPMonitor, region string) (*HttpPrivateRegionData, error) { retry := monitor.Retry @@ -219,6 +232,7 @@ func (jr jobRunner) HTTPJob(ctx context.Context, monitor *v1.HTTPMonitor, region } } else { data.Error = 1 + data.Message = httpFailureMessage(res, status.IsSuccessful()) // Mark the recorded response as errored so OTel emits the error counter // for non-2xx / failed assertions, matching the public checker. lastRes.Error = "Error" diff --git a/apps/checker/pkg/job/http_job_test.go b/apps/checker/pkg/job/http_job_test.go index f0512529..36d91a3d 100644 --- a/apps/checker/pkg/job/http_job_test.go +++ b/apps/checker/pkg/job/http_job_test.go @@ -198,3 +198,60 @@ func TestHTTPJob_HeaderAssertions(t *testing.T) { assert.Equal(t, uint8(0), data.Error) }) } + +// A failed check has to carry a message: it ends up as the alert body, which +// was empty for every private location HTTP monitor. +func TestHTTPJob_FailureMessage(t *testing.T) { + t.Run("reports the status code", func(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + })) + defer srv.Close() + + monitor := &v1.HTTPMonitor{Url: srv.URL, Method: "GET", Timeout: 10000, Retry: 1} + + data, err := job.NewJobRunner().HTTPJob(context.Background(), monitor, "test-region") + if err != nil { + t.Fatalf("expected no error, got %v", err) + } + assert.Equal(t, uint8(1), data.Error) + assert.Equal(t, "Request failed with status code 500", data.Message) + }) + + t.Run("reports a failed assertion on a 2xx", func(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + })) + defer srv.Close() + + monitor := &v1.HTTPMonitor{ + Url: srv.URL, Method: "GET", Timeout: 10000, Retry: 1, + HeaderAssertions: []*v1.HeaderAssertion{ + {Key: "X-Missing", Comparator: v1.StringComparator_STRING_COMPARATOR_EQUAL, Target: "expected"}, + }, + } + + data, err := job.NewJobRunner().HTTPJob(context.Background(), monitor, "test-region") + if err != nil { + t.Fatalf("expected no error, got %v", err) + } + assert.Equal(t, uint8(1), data.Error) + assert.Equal(t, "Assertions failed", data.Message) + }) + + t.Run("keeps a successful check message empty", func(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + })) + defer srv.Close() + + monitor := &v1.HTTPMonitor{Url: srv.URL, Method: "GET", Timeout: 10000, Retry: 1} + + data, err := job.NewJobRunner().HTTPJob(context.Background(), monitor, "test-region") + if err != nil { + t.Fatalf("expected no error, got %v", err) + } + assert.Equal(t, uint8(0), data.Error) + assert.Empty(t, data.Message) + }) +} diff --git a/apps/checker/pkg/scheduler/scheduler.go b/apps/checker/pkg/scheduler/scheduler.go index c9c52a95..566276fe 100644 --- a/apps/checker/pkg/scheduler/scheduler.go +++ b/apps/checker/pkg/scheduler/scheduler.go @@ -204,30 +204,30 @@ func (mm *MonitorManager) UpdateMonitors(ctx context.Context) { monitor := m c := context.Background() log.Printf("Starting DNS job for monitor %s (%s)", monitor.Id, monitor.Uri) - _, err := mm.JobRunner.DNSJob(c, monitor) + data, err := mm.JobRunner.DNSJob(c, monitor) if err != nil { log.Printf("DNS monitor check failed for %s (%s): %v", monitor.Id, monitor.Uri, err) return err } resp, ingestErr := mm.Client.IngestDNS(c, &connect.Request[v1.IngestDNSRequest]{ Msg: &v1.IngestDNSRequest{ - MonitorId: monitor.Id, - - // Id: data.ID, - // Uri: monitor.Uri, - // Message: data.Message, - // Latency: data.Latency, - // RequestStatus: data.RequestStatus, - // Error: int64(data.Error), - // CronTimestamp: data.CronTimestamp, - // Timestamp: data.Timestamp, + MonitorId: monitor.Id, + Id: data.ID, + Uri: monitor.Uri, + Message: data.Message, + Latency: data.Latency, + RequestStatus: data.RequestStatus, + Error: int64(data.Error), + CronTimestamp: data.CronTimestamp, + Timestamp: data.Timestamp, + Records: toProtoRecords(data.Records), }, }) if ingestErr != nil { log.Printf("Failed to ingest DNS result for %s (%s): %v", monitor.Id, monitor.Uri, ingestErr) return ingestErr } - log.Printf("DNS monitor check for %s (%s) ingested, ingest response: %v", monitor.Id, monitor.Uri, resp) + log.Printf("DNS monitor check for %s (%s) ingested with status %q, ingest response: %v", monitor.Id, monitor.Uri, data.RequestStatus, resp) return nil }, @@ -251,6 +251,19 @@ func (mm *MonitorManager) UpdateMonitors(ctx context.Context) { } +func toProtoRecords(records map[string][]string) map[string]*v1.Records { + if len(records) == 0 { + return nil + } + + protoRecords := make(map[string]*v1.Records, len(records)) + for recordType, values := range records { + protoRecords[recordType] = &v1.Records{Record: values} + } + + return protoRecords +} + func intervalToSecond(interval string) int { switch interval { case Interval30s: diff --git a/apps/checker/pkg/scheduler/scheduler_test.go b/apps/checker/pkg/scheduler/scheduler_test.go index 0d86d315..45514456 100644 --- a/apps/checker/pkg/scheduler/scheduler_test.go +++ b/apps/checker/pkg/scheduler/scheduler_test.go @@ -62,9 +62,16 @@ func (m *mockJobRunner) TCPRegion() string { } func (m *mockJobRunner) DNSJob(ctx context.Context, monitor *v1.DNSMonitor) (*job.DNSPrivateRegionData, error) { - - m.TCPJobCalled.Store(true) - return &job.DNSPrivateRegionData{}, nil + m.DNSJobCalled.Store(true) + return &job.DNSPrivateRegionData{ + ID: "dns-result-1", + URI: monitor.Uri, + RequestStatus: "success", + Latency: 42, + Timestamp: 1700000000000, + CronTimestamp: 1700000000000, + Records: map[string][]string{"A": {"192.168.1.1"}}, + }, nil } // mockClient implements v1.PrivateLocationServiceClient for testing @@ -238,3 +245,68 @@ func TestMonitorManager_ReschedulesOnConfigChange(t *testing.T) { t.Errorf("expected the added header to reach the job, got %v", got.Headers) } } + +// TestMonitorManager_IngestsDNSResult guards against sending an ingest request +// that carries only the monitor id: the check result was dropped on the floor +// and the server rejected every DNS request for a zero timestamp. +func TestMonitorManager_IngestsDNSResult(t *testing.T) { + ctx := t.Context() + + // Long periodicity: the task only runs when the test invokes it. + dnsMonitor := &v1.DNSMonitor{Id: "dns1", Uri: "openstatus.dev", Periodicity: "1h"} + + var ingested *v1.IngestDNSRequest + client := &mockClient{ + MonitorsFunc: func(ctx context.Context, req *connect.Request[v1.MonitorsRequest]) (*connect.Response[v1.MonitorsResponse], error) { + return connect.NewResponse(&v1.MonitorsResponse{ + DnsMonitors: []*v1.DNSMonitor{dnsMonitor}, + Region: "frankfurt-dc1", + }), nil + }, + IngestDNSFunc: func(ctx context.Context, req *connect.Request[v1.IngestDNSRequest]) (*connect.Response[v1.IngestDNSResponse], error) { + ingested = req.Msg + return connect.NewResponse(&v1.IngestDNSResponse{}), nil + }, + } + jobRunner := &mockJobRunner{} + + s := tasks.New() + defer s.Stop() + + mm := &scheduler.MonitorManager{Client: client, JobRunner: jobRunner, Scheduler: s} + + mm.UpdateMonitors(ctx) + runScheduledTask(t, mm.Scheduler, "dns1") + + if !jobRunner.DNSJobCalled.Load() { + t.Fatalf("expected DNSJob to be called") + } + if ingested == nil { + t.Fatalf("expected IngestDNS to be called") + } + if ingested.MonitorId != "dns1" { + t.Errorf("expected monitor id %q, got %q", "dns1", ingested.MonitorId) + } + if ingested.Id != "dns-result-1" { + t.Errorf("expected the check result id to be forwarded, got %q", ingested.Id) + } + if ingested.Uri != "openstatus.dev" { + t.Errorf("expected uri %q, got %q", "openstatus.dev", ingested.Uri) + } + if ingested.RequestStatus != "success" { + t.Errorf("expected request status %q, got %q", "success", ingested.RequestStatus) + } + if ingested.Latency != 42 { + t.Errorf("expected latency 42, got %d", ingested.Latency) + } + // The ingest server rejects a non-positive timestamp outright. + if ingested.Timestamp <= 0 { + t.Errorf("expected a positive timestamp, got %d", ingested.Timestamp) + } + if ingested.CronTimestamp <= 0 { + t.Errorf("expected a positive cron timestamp, got %d", ingested.CronTimestamp) + } + if got := ingested.Records["A"].GetRecord(); len(got) != 1 || got[0] != "192.168.1.1" { + t.Errorf("expected the A records to be forwarded, got %v", got) + } +} diff --git a/apps/private-location/internal/server/ingest_dns.go b/apps/private-location/internal/server/ingest_dns.go index 9021b525..d4a2c356 100644 --- a/apps/private-location/internal/server/ingest_dns.go +++ b/apps/private-location/internal/server/ingest_dns.go @@ -83,9 +83,18 @@ func (h *privateLocationHandler) IngestDNS(ctx context.Context, req *connect.Req URI: req.Msg.Uri, RequestStatus: req.Msg.RequestStatus, Records: string(recordsJSON), + ErrorMessage: req.Msg.Message, } h.sendEventAndUpdateLastSeen(ctx, data, tinybird.DatasourceDNS, ic.Region.ID) + h.forwardStatusUpdate(ctx, ic, statusUpdateInput{ + RequestStatus: data.RequestStatus, + Message: data.ErrorMessage, + Latency: data.Latency, + CronTimestamp: data.CronTimestamp, + ErrorFlag: data.Error, + }) + return connect.NewResponse(&private_locationv1.IngestDNSResponse{}), nil } diff --git a/apps/private-location/internal/server/ingest_dns_test.go b/apps/private-location/internal/server/ingest_dns_test.go index a1700ee8..adf3656d 100644 --- a/apps/private-location/internal/server/ingest_dns_test.go +++ b/apps/private-location/internal/server/ingest_dns_test.go @@ -6,10 +6,12 @@ import ( "io" "net/http" "testing" + "time" "connectrpc.com/connect" "github.com/openstatushq/openstatus/apps/private-location/internal/server" "github.com/openstatushq/openstatus/apps/private-location/internal/tinybird" + "github.com/openstatushq/openstatus/apps/private-location/internal/workflows" private_locationv1 "github.com/openstatushq/openstatus/apps/private-location/proto/private_location/v1" "github.com/stretchr/testify/require" ) @@ -240,6 +242,50 @@ func TestIngestDNS_RecordsKeyedByType(t *testing.T) { require.Equal(t, []string{"::1"}, records["AAAA"]) } +type recordingWorkflows struct { + called chan workflows.Payload +} + +func (c recordingWorkflows) Report(ctx context.Context, payload workflows.Payload) error { + c.called <- payload + return nil +} + +// TestIngestDNS_ForwardsStatusUpdate guards against DNS results reaching +// Tinybird without ever reaching the alerting pipeline. +func TestIngestDNS_ForwardsStatusUpdate(t *testing.T) { + h := server.NewPrivateLocationServer(testDB(), getTBClient(context.Background())) + client := recordingWorkflows{called: make(chan workflows.Payload, 1)} + h.WorkflowsClient = client + + req := connect.NewRequest(&private_locationv1.IngestDNSRequest{ + Id: "dns-result-4", + MonitorId: "5", + Timestamp: 1234567890, + Latency: 12, + CronTimestamp: 1234567800, + Uri: "openstatus.dev", + RequestStatus: "error", + Error: 1, + Message: "DNS assertions failed for openstatus.dev", + }) + req.Header().Set("openstatus-token", "my-secret-key") + + _, err := h.IngestDNS(context.Background(), req) + require.NoError(t, err) + + select { + case payload := <-client.called: + require.Equal(t, "5", payload.MonitorID) + require.Equal(t, "error", payload.Status) + require.Equal(t, "DNS assertions failed for openstatus.dev", payload.Message) + require.Equal(t, int64(1234567800), payload.CronTimestamp) + require.Equal(t, int64(12), payload.Latency) + case <-time.After(2 * time.Second): + t.Fatal("expected the DNS result to be forwarded to the workflows service") + } +} + func TestIngestDNS_WithError(t *testing.T) { h := server.NewPrivateLocationServer(testDB(), getTBClient(context.Background())) diff --git a/apps/private-location/internal/server/ingest_http.go b/apps/private-location/internal/server/ingest_http.go index a9da4f37..945c413e 100644 --- a/apps/private-location/internal/server/ingest_http.go +++ b/apps/private-location/internal/server/ingest_http.go @@ -76,6 +76,8 @@ func (h *privateLocationHandler) IngestHTTP(ctx context.Context, req *connect.Re Trigger: "cron", RequestStatus: req.Msg.RequestStatus, Assertions: ic.Monitor.Assertions.String, + Message: req.Msg.Message, + Error: uint8(req.Msg.Error), } h.sendEventAndUpdateLastSeen(ctx, data, tinybird.DatasourceHTTP, ic.Region.ID) diff --git a/apps/private-location/internal/server/ingest_http_test.go b/apps/private-location/internal/server/ingest_http_test.go index 0fdb55f4..00b07fd4 100644 --- a/apps/private-location/internal/server/ingest_http_test.go +++ b/apps/private-location/internal/server/ingest_http_test.go @@ -2,17 +2,22 @@ package server_test import ( "context" + "encoding/json" + "io" "log" "net/http" "os" "testing" + "time" "connectrpc.com/connect" "github.com/jmoiron/sqlx" _ "github.com/mattn/go-sqlite3" "github.com/openstatushq/openstatus/apps/private-location/internal/server" "github.com/openstatushq/openstatus/apps/private-location/internal/tinybird" + "github.com/openstatushq/openstatus/apps/private-location/internal/workflows" private_locationv1 "github.com/openstatushq/openstatus/apps/private-location/proto/private_location/v1" + "github.com/stretchr/testify/require" ) func testDB() *sqlx.DB { @@ -223,6 +228,57 @@ func TestIngestHTTP_WithFullData(t *testing.T) { } } +// TestIngestHTTP_ForwardsErrorAndMessage guards against the probe's error flag +// and failure message being dropped on the way to Tinybird and alerting: the +// `error` column stayed 0 for every private check and alerts had empty bodies. +func TestIngestHTTP_ForwardsErrorAndMessage(t *testing.T) { + var capturedBody []byte + interceptor := &interceptorHTTPClient{ + f: func(req *http.Request) (*http.Response, error) { + if req.Body != nil { + capturedBody, _ = io.ReadAll(req.Body) + } + return &http.Response{StatusCode: http.StatusAccepted}, nil + }, + } + h := server.NewPrivateLocationServer(testDB(), tinybird.NewClient(interceptor.GetHTTPClient(), "apiKey")) + workflowsClient := recordingWorkflows{called: make(chan workflows.Payload, 1)} + h.WorkflowsClient = workflowsClient + + const message = "Request failed with status code 500" + req := connect.NewRequest(&private_locationv1.IngestHTTPRequest{ + Id: "request-err", + MonitorId: "5", + Timestamp: 1234567890, + CronTimestamp: 1234567800, + Url: "https://example.com/api", + RequestStatus: "error", + StatusCode: 500, + Error: 1, + Message: message, + }) + req.Header().Set("openstatus-token", "my-secret-key") + + _, err := h.IngestHTTP(context.Background(), req) + require.NoError(t, err) + + var event struct { + Error uint8 `json:"error"` + Message string `json:"message"` + } + require.NoError(t, json.Unmarshal(capturedBody, &event)) + require.Equal(t, uint8(1), event.Error) + require.Equal(t, message, event.Message) + + select { + case payload := <-workflowsClient.called: + require.Equal(t, "error", payload.Status) + require.Equal(t, message, payload.Message) + case <-time.After(2 * time.Second): + t.Fatal("expected the failed check to be forwarded to the workflows service") + } +} + func TestIngestHTTP_WithError(t *testing.T) { h := server.NewPrivateLocationServer(testDB(), getTBClient(context.Background())) diff --git a/apps/private-location/internal/server/ingest_tcp.go b/apps/private-location/internal/server/ingest_tcp.go index 3816ef5a..821a4f88 100644 --- a/apps/private-location/internal/server/ingest_tcp.go +++ b/apps/private-location/internal/server/ingest_tcp.go @@ -66,6 +66,7 @@ func (h *privateLocationHandler) IngestTCP(ctx context.Context, req *connect.Req Trigger: "cron", URI: req.Msg.Uri, RequestStatus: req.Msg.RequestStatus, + ErrorMessage: req.Msg.Message, } h.sendEventAndUpdateLastSeen(ctx, data, tinybird.DatasourceTCP, ic.Region.ID) diff --git a/apps/private-location/internal/server/ingest_tcp_test.go b/apps/private-location/internal/server/ingest_tcp_test.go index 88452674..2285b16f 100644 --- a/apps/private-location/internal/server/ingest_tcp_test.go +++ b/apps/private-location/internal/server/ingest_tcp_test.go @@ -2,16 +2,68 @@ package server_test import ( "context" + "encoding/json" + "io" "net/http" "testing" + "time" "connectrpc.com/connect" "github.com/openstatushq/openstatus/apps/private-location/internal/server" "github.com/openstatushq/openstatus/apps/private-location/internal/tinybird" + "github.com/openstatushq/openstatus/apps/private-location/internal/workflows" private_locationv1 "github.com/openstatushq/openstatus/apps/private-location/proto/private_location/v1" + "github.com/stretchr/testify/require" ) +// TestIngestTCP_ForwardsErrorMessage guards against the probe's failure message +// being dropped: it never reached Tinybird or the alert body. +func TestIngestTCP_ForwardsErrorMessage(t *testing.T) { + var capturedBody []byte + interceptor := &interceptorHTTPClient{ + f: func(req *http.Request) (*http.Response, error) { + if req.Body != nil { + capturedBody, _ = io.ReadAll(req.Body) + } + return &http.Response{StatusCode: http.StatusAccepted}, nil + }, + } + h := server.NewPrivateLocationServer(testDB(), tinybird.NewClient(interceptor.GetHTTPClient(), "apiKey")) + workflowsClient := recordingWorkflows{called: make(chan workflows.Payload, 1)} + h.WorkflowsClient = workflowsClient + + const message = "dial tcp 10.0.0.1:5432: connect: connection refused" + req := connect.NewRequest(&private_locationv1.IngestTCPRequest{ + Id: "tcp-err", + MonitorId: "6", + Timestamp: 1234567890, + CronTimestamp: 1234567800, + Uri: "10.0.0.1:5432", + RequestStatus: "error", + Error: 1, + Message: message, + }) + req.Header().Set("openstatus-token", "my-secret-key") + + _, err := h.IngestTCP(context.Background(), req) + require.NoError(t, err) + + var event struct { + ErrorMessage string `json:"errorMessage"` + } + require.NoError(t, json.Unmarshal(capturedBody, &event)) + require.Equal(t, message, event.ErrorMessage) + + select { + case payload := <-workflowsClient.called: + require.Equal(t, "error", payload.Status) + require.Equal(t, message, payload.Message) + case <-time.After(2 * time.Second): + t.Fatal("expected the failed check to be forwarded to the workflows service") + } +} + func TestIngestTCP_Unauthenticated(t *testing.T) { h := server.NewPrivateLocationServer(testDB(), tinybird.NewClient(http.DefaultClient, "")) diff --git a/apps/private-location/internal/server/monitors.go b/apps/private-location/internal/server/monitors.go index b80badaf..e227e02c 100644 --- a/apps/private-location/internal/server/monitors.go +++ b/apps/private-location/internal/server/monitors.go @@ -181,7 +181,12 @@ func (h *privateLocationHandler) Monitors(ctx context.Context, req *connect.Requ } var location database.PrivateLocation - if err := h.db.Get(&location, "SELECT id, name FROM private_location WHERE token = ?", token); err != nil && !errors.Is(err, sql.ErrNoRows) { + if err := h.db.Get(&location, "SELECT id, name FROM private_location WHERE token = ?", token); err != nil { + // An unknown token used to fall through to an empty monitor list, so a + // probe configured with a typo looked healthy while checking nothing. + if errors.Is(err, sql.ErrNoRows) { + return nil, connect.NewError(connect.CodeUnauthenticated, ErrPrivateLocationNotFound) + } return nil, connect.NewError(connect.CodeInternal, err) } diff --git a/apps/private-location/internal/server/monitors_test.go b/apps/private-location/internal/server/monitors_test.go index 258d52e5..7f764768 100644 --- a/apps/private-location/internal/server/monitors_test.go +++ b/apps/private-location/internal/server/monitors_test.go @@ -152,6 +152,9 @@ func TestMonitors_Unauthenticated(t *testing.T) { } } +// An unknown token has to be rejected rather than answered with an empty +// monitor list: a probe with a mistyped token would otherwise report healthy +// while silently checking nothing. func TestMonitors_InvalidToken(t *testing.T) { h := server.NewPrivateLocationServer(testDB(), getTBClient(context.Background())) @@ -159,20 +162,14 @@ func TestMonitors_InvalidToken(t *testing.T) { req.Header().Set("openstatus-token", "invalid-token") resp, err := h.Monitors(context.Background(), req) - if err != nil { - t.Fatalf("expected no error for invalid token (just empty results), got %v", err) - } - if resp == nil { - t.Fatalf("expected non-nil response") - } - if len(resp.Msg.HttpMonitors) != 0 { - t.Errorf("expected 0 HTTP monitors for invalid token, got %d", len(resp.Msg.HttpMonitors)) + if err == nil { + t.Fatalf("expected an error for an unknown token, got nil") } - if len(resp.Msg.TcpMonitors) != 0 { - t.Errorf("expected 0 TCP monitors for invalid token, got %d", len(resp.Msg.TcpMonitors)) + if connect.CodeOf(err) != connect.CodeUnauthenticated { + t.Errorf("expected unauthenticated code, got %v", connect.CodeOf(err)) } - if len(resp.Msg.DnsMonitors) != 0 { - t.Errorf("expected 0 DNS monitors for invalid token, got %d", len(resp.Msg.DnsMonitors)) + if resp != nil { + t.Errorf("expected nil response, got %v", resp) } }