Something went wrong. Try again.
[READ-ONLY] Mirror of https://github.com/openstatusHQ/openstatus. 🫖 Status page with uptime monitoring & API monitoring as code 🫖 openstatus.dev
bun drizzle-orm monitoring monitoring-as-code nextjs observability on-call open-source shadcn-ui status-page statuspage synthetic-monitoring tinybird turso uptime uptime-checker uptime-monitor
Something went wrong. Try again.
8.6 kB · 278 lines
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279package queue_test
import ( "context" "encoding/json" "errors" "sync" "sync/atomic" "testing" "time"
"cloud.google.com/go/pubsub/v2" pb "cloud.google.com/go/pubsub/v2/apiv1/pubsubpb" "cloud.google.com/go/pubsub/v2/pstest" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "google.golang.org/api/option" "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure"
"github.com/openstatushq/openstatus/apps/checker/checker" "github.com/openstatushq/openstatus/apps/checker/handlers" "github.com/openstatushq/openstatus/apps/checker/pkg/queue" "github.com/openstatushq/openstatus/apps/checker/request")
type fakeRunner struct { run func(ctx context.Context, monitorID string) error}
func (f fakeRunner) RunHTTPCheck(ctx context.Context, req request.HttpCheckerRequest) (checker.Response, error) { return checker.Response{}, f.run(ctx, req.MonitorID)}
func (f fakeRunner) RunTCPCheck(ctx context.Context, req request.TCPCheckerRequest) (checker.TCPResponse, error) { return checker.TCPResponse{}, f.run(ctx, req.MonitorID)}
func (f fakeRunner) RunDNSCheck(ctx context.Context, req request.DNSCheckerRequest) (handlers.DNSResponse, error) { return handlers.DNSResponse{}, f.run(ctx, req.MonitorID)}
func (f fakeRunner) RunICMPCheck(ctx context.Context, req request.ICMPCheckerRequest) (checker.ICMPResponse, error) { return checker.ICMPResponse{}, f.run(ctx, req.MonitorID)}
func (f fakeRunner) RunGRPCCheck(ctx context.Context, req request.GRPCCheckerRequest) (checker.GRPCResponse, error) { return checker.GRPCResponse{}, f.run(ctx, req.MonitorID)}
const ( project = "projects/test" subscription = "checker-local")
type harness struct { srv *pstest.Server client *pubsub.Client events chan queue.Event}
func newHarness(t *testing.T) *harness { t.Helper() ctx := context.Background() srv := pstest.NewServer() t.Cleanup(func() { srv.Close() })
_, err := srv.GServer.CreateTopic(ctx, &pb.Topic{Name: project + "/topics/" + subscription}) require.NoError(t, err) _, err = srv.GServer.CreateSubscription(ctx, &pb.Subscription{ Name: project + "/subscriptions/" + subscription, Topic: project + "/topics/" + subscription, AckDeadlineSeconds: 10, }) require.NoError(t, err)
conn, err := grpc.NewClient(srv.Addr, grpc.WithTransportCredentials(insecure.NewCredentials())) require.NoError(t, err) t.Cleanup(func() { conn.Close() })
client, err := pubsub.NewClient(ctx, "test", option.WithGRPCConn(conn)) require.NoError(t, err) t.Cleanup(func() { client.Close() })
return &harness{srv: srv, client: client, events: make(chan queue.Event, 16)}}
func (h *harness) publish(t *testing.T, jobType string, monitorID string, cronTimestamp time.Time) string { t.Helper() data, err := json.Marshal(map[string]any{ "monitorId": monitorID, "workspaceId": "1", "cronTimestamp": cronTimestamp.UnixMilli(), }) require.NoError(t, err)
return h.srv.Publish(project+"/topics/"+subscription, data, map[string]string{ "jobType": jobType, "monitorId": monitorID, })}
func (h *harness) start(t *testing.T, runner queue.Runner, loop queue.Loop, drain time.Duration) (context.CancelFunc, <-chan error) { t.Helper() loop.Subscription = subscription c := &queue.Consumer{ Client: h.client, Runner: runner, Loops: []queue.Loop{loop}, Drain: drain, OnEvent: func(_ context.Context, e queue.Event) { h.events <- e }, } ctx, cancel := context.WithCancel(context.Background()) done := make(chan error, 1) go func() { done <- c.Run(ctx) }() t.Cleanup(cancel)
return cancel, done}
func (h *harness) next(t *testing.T) queue.Event { t.Helper() select { case e := <-h.events: return e case <-time.After(10 * time.Second): t.Fatal("no event") return queue.Event{} }}
func TestConsumer_AcksRecordedCheck(t *testing.T) { h := newHarness(t) id := h.publish(t, "http", "1", time.Now()) h.start(t, fakeRunner{run: func(context.Context, string) error { return nil }}, queue.Loop{MaxOutstanding: 10}, time.Second)
e := h.next(t)
assert.Equal(t, queue.OutcomeAck, e.Outcome) assert.Equal(t, "http", e.JobType) assert.Equal(t, "1", e.MonitorID) assert.Eventually(t, func() bool { return h.srv.Message(id).Acks == 1 }, 5*time.Second, 20*time.Millisecond)}
func TestConsumer_NacksRecordFailure(t *testing.T) { h := newHarness(t) id := h.publish(t, "tcp", "1", time.Now()) var calls atomic.Int32 h.start(t, fakeRunner{run: func(context.Context, string) error { if calls.Add(1) == 1 { return errors.New("tinybird down") } return nil }}, queue.Loop{MaxOutstanding: 10}, time.Second)
assert.Equal(t, queue.OutcomeNack, h.next(t).Outcome) assert.Equal(t, queue.OutcomeAck, h.next(t).Outcome) assert.GreaterOrEqual(t, h.srv.Message(id).Deliveries, 2)}
func TestConsumer_AcksPoisonWithoutRetrying(t *testing.T) { h := newHarness(t) unknown := h.publish(t, "smtp", "1", time.Now()) invalid := h.publish(t, "dns", "2", time.Now()) var calls atomic.Int32 h.start(t, fakeRunner{run: func(context.Context, string) error { calls.Add(1) return handlers.InvalidRequestError{Reason: "invalid monitor id"} }}, queue.Loop{MaxOutstanding: 10}, time.Second)
assert.Equal(t, queue.OutcomePoison, h.next(t).Outcome) assert.Equal(t, queue.OutcomePoison, h.next(t).Outcome) assert.Eventually(t, func() bool { return h.srv.Message(unknown).Acks == 1 && h.srv.Message(invalid).Acks == 1 }, 5*time.Second, 20*time.Millisecond) assert.Equal(t, int32(1), calls.Load())}
func TestConsumer_WaitsForCronTimestamp(t *testing.T) { h := newHarness(t) at := time.Now().Add(500 * time.Millisecond) h.publish(t, "icmp", "1", at) var ranAt atomic.Int64 h.start(t, fakeRunner{run: func(context.Context, string) error { ranAt.Store(time.Now().UnixMilli()) return nil }}, queue.Loop{MaxOutstanding: 10}, time.Second)
e := h.next(t)
assert.Equal(t, queue.OutcomeAck, e.Outcome) assert.Positive(t, e.Wait) assert.GreaterOrEqual(t, ranAt.Load(), at.UnixMilli())}
func TestConsumer_SleepingMessageHoldsNoProbeSlot(t *testing.T) { h := newHarness(t) h.publish(t, "http", "later", time.Now().Add(2*time.Second)) h.publish(t, "http", "now", time.Now()) var mu sync.Mutex var order []string h.start(t, fakeRunner{run: func(_ context.Context, monitorID string) error { mu.Lock() order = append(order, monitorID) mu.Unlock() return nil }}, queue.Loop{MaxOutstanding: 10, MaxConcurrent: 1}, time.Second)
first := h.next(t)
assert.Equal(t, "now", first.MonitorID) assert.Equal(t, "later", h.next(t).MonitorID) mu.Lock() defer mu.Unlock() assert.Equal(t, []string{"now", "later"}, order)}
func TestConsumer_NacksCheckStillRunningAfterDrain(t *testing.T) { h := newHarness(t) h.publish(t, "grpc", "1", time.Now()) started := make(chan struct{}) var stopped atomic.Bool cancel, done := h.start(t, fakeRunner{run: func(ctx context.Context, _ string) error { close(started) <-ctx.Done() stopped.Store(true) return ctx.Err() }}, queue.Loop{MaxOutstanding: 10}, 100*time.Millisecond)
<-started cancel()
e := h.next(t) assert.Equal(t, queue.OutcomeDrain, e.Outcome) assert.True(t, stopped.Load(), "check still running after its message was nacked") select { case err := <-done: assert.NoError(t, err) case <-time.After(10 * time.Second): t.Fatal("consumer did not stop") }}
func TestConsumer_AcksMissingCronTimestampAsPoison(t *testing.T) { h := newHarness(t) id := h.srv.Publish(project+"/topics/"+subscription, []byte(`{"monitorId":"1"}`), map[string]string{ "jobType": "http", "monitorId": "1", }) var calls atomic.Int32 h.start(t, fakeRunner{run: func(context.Context, string) error { calls.Add(1) return nil }}, queue.Loop{MaxOutstanding: 10}, time.Second)
assert.Equal(t, queue.OutcomePoison, h.next(t).Outcome) assert.Eventually(t, func() bool { return h.srv.Message(id).Acks == 1 }, 5*time.Second, 20*time.Millisecond) assert.Equal(t, int32(0), calls.Load())}
func TestConsumer_RejectsNonPositiveMaxOutstanding(t *testing.T) { h := newHarness(t) _, done := h.start(t, fakeRunner{run: func(context.Context, string) error { return nil }}, queue.Loop{}, time.Second)
select { case err := <-done: assert.ErrorContains(t, err, "must be positive") case <-time.After(5 * time.Second): t.Fatal("consumer did not fail") }}
func TestSubscription(t *testing.T) { assert.Equal(t, "checker-ams", queue.Subscription("ams", false)) assert.Equal(t, "checker-koyeb_fra-30s", queue.Subscription("koyeb_fra", true))}