(t *testing.T)
| 19 | ) |
| 20 | |
| 21 | func TestPGPubsub_Metrics(t *testing.T) { |
| 22 | t.Parallel() |
| 23 | |
| 24 | logger := testutil.Logger(t) |
| 25 | connectionURL, err := dbtestutil.Open(t) |
| 26 | require.NoError(t, err) |
| 27 | db, err := sql.Open("postgres", connectionURL) |
| 28 | require.NoError(t, err) |
| 29 | defer db.Close() |
| 30 | registry := prometheus.NewRegistry() |
| 31 | ctx := testutil.Context(t, testutil.WaitLong) |
| 32 | |
| 33 | uut, err := pubsub.New(ctx, logger, db, connectionURL) |
| 34 | require.NoError(t, err) |
| 35 | defer uut.Close() |
| 36 | |
| 37 | err = registry.Register(uut) |
| 38 | require.NoError(t, err) |
| 39 | |
| 40 | // each Gather measures pubsub latency by publishing a message & subscribing to it |
| 41 | var gatherCount float64 |
| 42 | |
| 43 | metrics, err := registry.Gather() |
| 44 | gatherCount++ |
| 45 | require.NoError(t, err) |
| 46 | require.True(t, testutil.PromGaugeHasValue(t, metrics, 0, "coder_pubsub_current_events")) |
| 47 | require.True(t, testutil.PromGaugeHasValue(t, metrics, 0, "coder_pubsub_current_subscribers")) |
| 48 | |
| 49 | event := "test" |
| 50 | data := "testing" |
| 51 | messageChannel := make(chan []byte) |
| 52 | unsub0, err := uut.Subscribe(event, func(_ context.Context, message []byte) { |
| 53 | messageChannel <- message |
| 54 | }) |
| 55 | require.NoError(t, err) |
| 56 | defer unsub0() |
| 57 | go func() { |
| 58 | err := uut.Publish(event, []byte(data)) |
| 59 | assert.NoError(t, err) |
| 60 | }() |
| 61 | _ = testutil.TryReceive(ctx, t, messageChannel) |
| 62 | |
| 63 | require.Eventually(t, func() bool { |
| 64 | latencyBytes := gatherCount * pubsub.LatencyMessageLength |
| 65 | metrics, err = registry.Gather() |
| 66 | gatherCount++ |
| 67 | assert.NoError(t, err) |
| 68 | return testutil.PromGaugeHasValue(t, metrics, 1, "coder_pubsub_current_events") && |
| 69 | testutil.PromGaugeHasValue(t, metrics, 1, "coder_pubsub_current_subscribers") && |
| 70 | testutil.PromGaugeHasValue(t, metrics, 1, "coder_pubsub_connected") && |
| 71 | testutil.PromCounterHasValue(t, metrics, gatherCount, "coder_pubsub_publishes_total", "true") && |
| 72 | testutil.PromCounterHasValue(t, metrics, gatherCount, "coder_pubsub_subscribes_total", "true") && |
| 73 | testutil.PromCounterHasValue(t, metrics, gatherCount, "coder_pubsub_messages_total", "normal") && |
| 74 | testutil.PromCounterHasValue(t, metrics, float64(len(data))+latencyBytes, "coder_pubsub_received_bytes_total") && |
| 75 | testutil.PromCounterHasValue(t, metrics, float64(len(data))+latencyBytes, "coder_pubsub_published_bytes_total") && |
| 76 | testutil.PromGaugeAssertion(t, metrics, func(in float64) bool { return in > 0 }, "coder_pubsub_send_latency_seconds") && |
| 77 | testutil.PromGaugeAssertion(t, metrics, func(in float64) bool { return in > 0 }, "coder_pubsub_receive_latency_seconds") && |
| 78 | testutil.PromCounterHasValue(t, metrics, gatherCount, "coder_pubsub_latency_measures_total") && |
nothing calls this directly
no test coverage detected