sneak
918533d897
Some checks failed
check / check (push) Has been cancelled
A delivery could reach a bad end without the engine recording why, and a retrying delivery could fail to reach an end at all. An unknown target type marked the delivery failed and wrote no DeliveryResult, so the event log showed "failed" with no attempts and the only account of why was one line in the server log. It now records a result naming the type before failing the delivery. A deleted target left its retrying deliveries stranded. Both recovery and the sweep began with a scoped loadTarget, which cannot see a soft deleted row, so both logged and returned: the delivery stayed retrying for the life of the database while the sweep repeated the same error every minute. Both now terminalise it with a recorded reason. Deleting a target also did not stop deliveries to it. A scheduled retry is a time.AfterFunc holding the target's configuration from when the chain began, and nothing on that path read the target row, so the timer kept firing and kept sending to the destination the operator had removed for the rest of the backoff chain; terminalising in recovery and the sweep alone would only have caught it after a restart. processRetryTask now confirms the target still exists before it attempts, and abandons the chain when it does not. Only a target confirmed gone stops anything. A lookup that fails for any other reason is the main database being unreadable, which is transient, and every path leaves the delivery exactly as it was rather than failing it. The reason text comes from one Unscoped lookup confined to these terminal paths, because a soft deleted row is what distinguishes a target the operator deleted from an id that never named one. The engine's normal target loading stays scoped, or deleting a target would stop nothing. Terminal writes reached from recovery keep going through the existing retainIdle ownership gate; the retry path writes directly, as a target's own Deliver does, because the worker already holds that delivery. Nothing was added to either sweep dispatch arm. Retry fixtures that drove processRetryTask for a target id with no row in the main database now create one. That state is not reachable in service: the handler reads the target to build the task.
565 lines
14 KiB
Go
565 lines
14 KiB
Go
package delivery_test
|
|
|
|
import (
|
|
"context"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"testing"
|
|
|
|
"github.com/google/uuid"
|
|
"github.com/prometheus/client_golang/prometheus"
|
|
dto "github.com/prometheus/client_model/go"
|
|
"github.com/stretchr/testify/assert"
|
|
"github.com/stretchr/testify/require"
|
|
"sneak.berlin/go/webhooker/internal/database"
|
|
"sneak.berlin/go/webhooker/internal/delivery"
|
|
"sneak.berlin/go/webhooker/internal/metrics"
|
|
)
|
|
|
|
// Metric names as exposed on /metrics.
|
|
const (
|
|
mAttempts = "webhooker_delivery_attempts_total"
|
|
mSucceeded = "webhooker_deliveries_succeeded_total"
|
|
mFailed = "webhooker_deliveries_failed_total"
|
|
mRetries = "webhooker_delivery_retries_total"
|
|
mDuration = "webhooker_delivery_duration_seconds"
|
|
mPending = "webhooker_deliveries_pending"
|
|
mRetrying = "webhooker_deliveries_retrying"
|
|
mBreakers = "webhooker_circuit_breakers_open"
|
|
)
|
|
|
|
const (
|
|
mTypeHTTP = "http"
|
|
mTypeLog = "log"
|
|
mTypeUnknown = "unknown"
|
|
)
|
|
|
|
// mIsolate gives the setup's engine a metric set registered on a
|
|
// private registry. The process-wide collectors are moved by every
|
|
// other delivery test running in parallel, so exact assertions are
|
|
// only possible against a registry this test owns.
|
|
func mIsolate(
|
|
t *testing.T, s iSetup,
|
|
) *prometheus.Registry {
|
|
t.Helper()
|
|
|
|
reg := prometheus.NewRegistry()
|
|
s.Engine.ExportSetMetrics(metrics.New(reg))
|
|
|
|
return reg
|
|
}
|
|
|
|
// mFind returns the series of the named metric carrying the given
|
|
// target_type label.
|
|
func mFind(
|
|
t *testing.T,
|
|
reg *prometheus.Registry,
|
|
name, targetType string,
|
|
) *dto.Metric {
|
|
t.Helper()
|
|
|
|
families, err := reg.Gather()
|
|
require.NoError(t, err)
|
|
|
|
for _, fam := range families {
|
|
if fam.GetName() != name {
|
|
continue
|
|
}
|
|
|
|
for _, m := range fam.GetMetric() {
|
|
for _, label := range m.GetLabel() {
|
|
if label.GetName() == "target_type" &&
|
|
label.GetValue() == targetType {
|
|
return m
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
t.Fatalf(
|
|
"metric %s{target_type=%q} not found",
|
|
name, targetType,
|
|
)
|
|
|
|
return nil
|
|
}
|
|
|
|
func mCounter(
|
|
t *testing.T,
|
|
reg *prometheus.Registry,
|
|
name, targetType string,
|
|
) float64 {
|
|
t.Helper()
|
|
|
|
return mFind(t, reg, name, targetType).
|
|
GetCounter().GetValue()
|
|
}
|
|
|
|
func mGauge(
|
|
t *testing.T,
|
|
reg *prometheus.Registry,
|
|
name, targetType string,
|
|
) float64 {
|
|
t.Helper()
|
|
|
|
return mFind(t, reg, name, targetType).
|
|
GetGauge().GetValue()
|
|
}
|
|
|
|
// mHTTPDurations returns how many samples the delivery duration
|
|
// histogram holds for the http target type, which is the type every
|
|
// test here times.
|
|
func mHTTPDurations(
|
|
t *testing.T, reg *prometheus.Registry,
|
|
) uint64 {
|
|
t.Helper()
|
|
|
|
return mFind(t, reg, mDuration, mTypeHTTP).
|
|
GetHistogram().GetSampleCount()
|
|
}
|
|
|
|
// TestDeliveryMetrics_SuccessAndRetryExhaustion drives one delivery
|
|
// that succeeds and one that fails every attempt until its retries
|
|
// are exhausted, and asserts every delivery counter across both.
|
|
func TestDeliveryMetrics_SuccessAndRetryExhaustion(
|
|
t *testing.T,
|
|
) {
|
|
t.Parallel()
|
|
|
|
s := newISetup(t)
|
|
reg := mIsolate(t, s)
|
|
|
|
mDeliverOK(t, s)
|
|
|
|
assert.InDelta(t, 1.0,
|
|
mCounter(t, reg, mAttempts, mTypeHTTP), 0)
|
|
assert.InDelta(t, 1.0,
|
|
mCounter(t, reg, mSucceeded, mTypeHTTP), 0)
|
|
assert.InDelta(t, 0.0,
|
|
mCounter(t, reg, mFailed, mTypeHTTP), 0)
|
|
assert.InDelta(t, 0.0,
|
|
mCounter(t, reg, mRetries, mTypeHTTP), 0)
|
|
assert.Equal(t, uint64(1),
|
|
mHTTPDurations(t, reg))
|
|
|
|
mExhaustRetries(t, s)
|
|
|
|
// Two further attempts: the first is retried, the second is
|
|
// the last one allowed and fails the delivery terminally.
|
|
assert.InDelta(t, 3.0,
|
|
mCounter(t, reg, mAttempts, mTypeHTTP), 0)
|
|
assert.InDelta(t, 1.0,
|
|
mCounter(t, reg, mSucceeded, mTypeHTTP), 0)
|
|
assert.InDelta(t, 1.0,
|
|
mCounter(t, reg, mRetries, mTypeHTTP), 0)
|
|
assert.InDelta(t, 1.0,
|
|
mCounter(t, reg, mFailed, mTypeHTTP), 0)
|
|
assert.Equal(t, uint64(3),
|
|
mHTTPDurations(t, reg))
|
|
|
|
// Two consecutive failures are below the trip threshold.
|
|
assert.InDelta(t, 0.0,
|
|
mGauge(t, reg, mBreakers, mTypeHTTP), 0)
|
|
|
|
// The label is the target type and nothing finer: two http
|
|
// targets shared one series, and no other type's moved.
|
|
assert.InDelta(t, 0.0,
|
|
mCounter(t, reg, mAttempts, mTypeLog), 0)
|
|
assert.InDelta(t, 0.0,
|
|
mCounter(t, reg, mFailed, mTypeLog), 0)
|
|
}
|
|
|
|
// mDeliverOK delivers one event to a target that answers 200.
|
|
func mDeliverOK(t *testing.T, s iSetup) {
|
|
t.Helper()
|
|
|
|
ts := httptest.NewServer(http.HandlerFunc(
|
|
func(w http.ResponseWriter, _ *http.Request) {
|
|
w.WriteHeader(http.StatusOK)
|
|
},
|
|
))
|
|
defer ts.Close()
|
|
|
|
event := iSeedEvent(
|
|
t, s.WebhookDB, s.WebhookID, `{"ok":true}`,
|
|
)
|
|
targetID := uuid.New().String()
|
|
|
|
d := iSeedDelivery(
|
|
t, s.WebhookDB, event.ID, targetID,
|
|
database.DeliveryStatusPending,
|
|
)
|
|
|
|
body := event.Body
|
|
task := iTask(
|
|
d, event, s.WebhookID, targetID,
|
|
"metrics-ok", iHTTPConfig(ts.URL), 3, 1, &body,
|
|
)
|
|
|
|
s.Engine.ExportProcessNewTask(context.TODO(), &task)
|
|
|
|
iAssertStatus(t, s.WebhookDB, d.ID,
|
|
database.DeliveryStatusDelivered,
|
|
)
|
|
}
|
|
|
|
// mExhaustRetries delivers to a target that answers 500 with a
|
|
// two-attempt budget, driving both attempts so the delivery ends
|
|
// terminally failed.
|
|
func mExhaustRetries(t *testing.T, s iSetup) {
|
|
t.Helper()
|
|
|
|
ts := httptest.NewServer(http.HandlerFunc(
|
|
func(w http.ResponseWriter, _ *http.Request) {
|
|
w.WriteHeader(http.StatusInternalServerError)
|
|
},
|
|
))
|
|
defer ts.Close()
|
|
|
|
event := iSeedEvent(
|
|
t, s.WebhookDB, s.WebhookID, `{"ok":false}`,
|
|
)
|
|
targetID := uuid.New().String()
|
|
|
|
d := iSeedDelivery(
|
|
t, s.WebhookDB, event.ID, targetID,
|
|
database.DeliveryStatusPending,
|
|
)
|
|
|
|
body := event.Body
|
|
cfg := iHTTPConfig(ts.URL)
|
|
|
|
// The retry below is only run if its target still exists; see
|
|
// https://git.eeqj.de/sneak/webhooker/issues/107.
|
|
iCreateTarget(
|
|
t, s.MainDB, targetID, s.WebhookID, "metrics-fail",
|
|
database.TargetTypeHTTP, cfg, 2,
|
|
)
|
|
|
|
first := iTask(
|
|
d, event, s.WebhookID, targetID,
|
|
"metrics-fail", cfg, 2, 1, &body,
|
|
)
|
|
|
|
s.Engine.ExportProcessNewTask(context.TODO(), &first)
|
|
|
|
iAssertStatus(t, s.WebhookDB, d.ID,
|
|
database.DeliveryStatusRetrying,
|
|
)
|
|
|
|
// The engine's own scheduler would re-enqueue this after the
|
|
// backoff; driving the second attempt directly keeps the test
|
|
// deterministic and off the wall clock.
|
|
second := iTask(
|
|
d, event, s.WebhookID, targetID,
|
|
"metrics-fail", cfg, 2, 2, &body,
|
|
)
|
|
|
|
s.Engine.ExportProcessRetryTask(
|
|
context.TODO(), &second,
|
|
)
|
|
|
|
iAssertStatus(t, s.WebhookDB, d.ID,
|
|
database.DeliveryStatusFailed,
|
|
)
|
|
}
|
|
|
|
// TestDeliveryMetrics_CircuitBreakerGauge proves the open-breaker
|
|
// gauge follows a breaker that trips.
|
|
func TestDeliveryMetrics_CircuitBreakerGauge(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
s := newISetup(t)
|
|
reg := mIsolate(t, s)
|
|
|
|
ts := httptest.NewServer(http.HandlerFunc(
|
|
func(w http.ResponseWriter, _ *http.Request) {
|
|
w.WriteHeader(http.StatusInternalServerError)
|
|
},
|
|
))
|
|
defer ts.Close()
|
|
|
|
event := iSeedEvent(
|
|
t, s.WebhookDB, s.WebhookID, `{"trip":true}`,
|
|
)
|
|
targetID := uuid.New().String()
|
|
|
|
d := iSeedDelivery(
|
|
t, s.WebhookDB, event.ID, targetID,
|
|
database.DeliveryStatusPending,
|
|
)
|
|
|
|
body := event.Body
|
|
cfg := iHTTPConfig(ts.URL)
|
|
|
|
// A retry budget above the failure threshold, so the breaker
|
|
// rather than the budget is what stops the delivery.
|
|
maxRetries := delivery.ExportDefaultFailureThreshold + 5
|
|
|
|
// The retries below are only run if their target still exists;
|
|
// see https://git.eeqj.de/sneak/webhooker/issues/107.
|
|
iCreateTarget(
|
|
t, s.MainDB, targetID, s.WebhookID, "metrics-trip",
|
|
database.TargetTypeHTTP, cfg, maxRetries,
|
|
)
|
|
|
|
first := iTask(
|
|
d, event, s.WebhookID, targetID,
|
|
"metrics-trip", cfg, maxRetries, 1, &body,
|
|
)
|
|
|
|
s.Engine.ExportProcessNewTask(context.TODO(), &first)
|
|
|
|
assert.InDelta(t, 0.0,
|
|
mGauge(t, reg, mBreakers, mTypeHTTP), 0)
|
|
|
|
for attempt := 2; attempt <= delivery.
|
|
ExportDefaultFailureThreshold; attempt++ {
|
|
task := iTask(
|
|
d, event, s.WebhookID, targetID,
|
|
"metrics-trip", cfg, maxRetries, attempt, &body,
|
|
)
|
|
|
|
s.Engine.ExportProcessRetryTask(
|
|
context.TODO(), &task,
|
|
)
|
|
}
|
|
|
|
assert.InDelta(t, 1.0,
|
|
mGauge(t, reg, mBreakers, mTypeHTTP), 0)
|
|
}
|
|
|
|
// TestDeliveryMetrics_BreakerBlockedIsNotAnAttempt proves a delivery
|
|
// an open circuit breaker refuses is neither counted as an attempt
|
|
// nor observed in the duration histogram.
|
|
//
|
|
// It sends nothing and records no result row, so counting it would
|
|
// climb the attempts counter with no traffic behind it and pull the
|
|
// duration quantiles down with near-zero samples for as long as the
|
|
// breaker stayed open — the metric moving the wrong way during the
|
|
// outage it exists to reveal.
|
|
func TestDeliveryMetrics_BreakerBlockedIsNotAnAttempt(
|
|
t *testing.T,
|
|
) {
|
|
t.Parallel()
|
|
|
|
s := newISetup(t)
|
|
reg := mIsolate(t, s)
|
|
|
|
ts := httptest.NewServer(http.HandlerFunc(
|
|
func(w http.ResponseWriter, _ *http.Request) {
|
|
w.WriteHeader(http.StatusInternalServerError)
|
|
},
|
|
))
|
|
defer ts.Close()
|
|
|
|
event := iSeedEvent(
|
|
t, s.WebhookDB, s.WebhookID, `{"blocked":true}`,
|
|
)
|
|
targetID := uuid.New().String()
|
|
|
|
d := iSeedDelivery(
|
|
t, s.WebhookDB, event.ID, targetID,
|
|
database.DeliveryStatusPending,
|
|
)
|
|
|
|
body := event.Body
|
|
cfg := iHTTPConfig(ts.URL)
|
|
maxRetries := delivery.ExportDefaultFailureThreshold + 5
|
|
|
|
iCreateTarget(
|
|
t, s.MainDB, targetID, s.WebhookID, "metrics-blocked",
|
|
database.TargetTypeHTTP, cfg, maxRetries,
|
|
)
|
|
|
|
first := iTask(
|
|
d, event, s.WebhookID, targetID,
|
|
"metrics-blocked", cfg, maxRetries, 1, &body,
|
|
)
|
|
|
|
s.Engine.ExportProcessNewTask(context.TODO(), &first)
|
|
|
|
for attempt := 2; attempt <= delivery.
|
|
ExportDefaultFailureThreshold; attempt++ {
|
|
task := iTask(
|
|
d, event, s.WebhookID, targetID,
|
|
"metrics-blocked", cfg, maxRetries, attempt, &body,
|
|
)
|
|
|
|
s.Engine.ExportProcessRetryTask(context.TODO(), &task)
|
|
}
|
|
|
|
require.InDelta(t, 1.0,
|
|
mGauge(t, reg, mBreakers, mTypeHTTP), 0,
|
|
"breaker should be open before the blocked attempt")
|
|
|
|
threshold := float64(
|
|
delivery.ExportDefaultFailureThreshold,
|
|
)
|
|
|
|
assert.InDelta(t, threshold,
|
|
mCounter(t, reg, mAttempts, mTypeHTTP), 0)
|
|
assert.Equal(t, uint64(threshold),
|
|
mHTTPDurations(t, reg))
|
|
|
|
retriesBefore := mCounter(t, reg, mRetries, mTypeHTTP)
|
|
|
|
blocked := iTask(
|
|
d, event, s.WebhookID, targetID,
|
|
"metrics-blocked", cfg, maxRetries,
|
|
delivery.ExportDefaultFailureThreshold+1, &body,
|
|
)
|
|
|
|
s.Engine.ExportProcessRetryTask(context.TODO(), &blocked)
|
|
|
|
// The breaker refused it: rescheduled, so the retry counter
|
|
// moved, but nothing was attempted or timed.
|
|
assert.InDelta(t, retriesBefore+1,
|
|
mCounter(t, reg, mRetries, mTypeHTTP), 0)
|
|
assert.InDelta(t, threshold,
|
|
mCounter(t, reg, mAttempts, mTypeHTTP), 0)
|
|
assert.Equal(t, uint64(threshold),
|
|
mHTTPDurations(t, reg))
|
|
}
|
|
|
|
// TestDeliveryMetrics_OrphanedRetryFailureLabelled proves the
|
|
// terminal failure of a delivery whose target no longer retries is
|
|
// counted against the target's real type, not against unknown. The
|
|
// type is threaded in as an argument because populating d.Target on
|
|
// that path would write the target row into the per-webhook database
|
|
// (https://git.eeqj.de/sneak/webhooker/issues/206).
|
|
func TestDeliveryMetrics_OrphanedRetryFailureLabelled(
|
|
t *testing.T,
|
|
) {
|
|
t.Parallel()
|
|
|
|
s := newISetup(t)
|
|
reg := mIsolate(t, s)
|
|
|
|
iCreateWebhook(
|
|
t, s.MainDB, s.WebhookID, "orphaned-label",
|
|
)
|
|
|
|
deliveryID := iSeedRetryingWithType(
|
|
t, s, database.TargetTypeLog,
|
|
)
|
|
|
|
s.Engine.ExportSweepWebhookRetries(
|
|
context.Background(), s.WebhookID,
|
|
)
|
|
|
|
iAssertStatus(t, s.WebhookDB, deliveryID,
|
|
database.DeliveryStatusFailed,
|
|
)
|
|
|
|
assert.InDelta(t, 1.0,
|
|
mCounter(t, reg, mFailed, mTypeLog), 0)
|
|
}
|
|
|
|
// TestDeliveryMetrics_QueueDepthGauges proves the sampler publishes
|
|
// the queued deliveries it finds in the per-webhook databases, and
|
|
// that a drained queue reads zero rather than keeping its last
|
|
// value.
|
|
func TestDeliveryMetrics_QueueDepthGauges(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
s := newISetup(t)
|
|
reg := mIsolate(t, s)
|
|
|
|
iCreateWebhook(
|
|
t, s.MainDB, s.WebhookID, "queue-depth",
|
|
)
|
|
|
|
targetID := uuid.New().String()
|
|
|
|
iCreateTarget(t, s.MainDB, targetID, s.WebhookID,
|
|
"queue-depth-target", database.TargetTypeHTTP,
|
|
iHTTPConfig("https://example.com/hook"), 3,
|
|
)
|
|
|
|
event := iSeedEvent(
|
|
t, s.WebhookDB, s.WebhookID, `{"queued":true}`,
|
|
)
|
|
|
|
pending := iSeedDelivery(
|
|
t, s.WebhookDB, event.ID, targetID,
|
|
database.DeliveryStatusPending,
|
|
)
|
|
|
|
iSeedDelivery(
|
|
t, s.WebhookDB, event.ID, targetID,
|
|
database.DeliveryStatusPending,
|
|
)
|
|
|
|
retrying := iSeedDelivery(
|
|
t, s.WebhookDB, event.ID, targetID,
|
|
database.DeliveryStatusRetrying,
|
|
)
|
|
|
|
s.Engine.ExportSampleQueueDepths(context.Background())
|
|
|
|
assert.InDelta(t, 2.0,
|
|
mGauge(t, reg, mPending, mTypeHTTP), 0)
|
|
assert.InDelta(t, 1.0,
|
|
mGauge(t, reg, mRetrying, mTypeHTTP), 0)
|
|
assert.InDelta(t, 0.0,
|
|
mGauge(t, reg, mPending, mTypeLog), 0)
|
|
|
|
require.NoError(t, s.WebhookDB.
|
|
Model(&database.Delivery{}).
|
|
Where("id IN ?", []string{pending.ID, retrying.ID}).
|
|
Update(
|
|
"status", database.DeliveryStatusDelivered,
|
|
).Error)
|
|
|
|
s.Engine.ExportSampleQueueDepths(context.Background())
|
|
|
|
assert.InDelta(t, 1.0,
|
|
mGauge(t, reg, mPending, mTypeHTTP), 0)
|
|
assert.InDelta(t, 0.0,
|
|
mGauge(t, reg, mRetrying, mTypeHTTP), 0)
|
|
}
|
|
|
|
// TestDeliveryMetrics_QueueDepthDeletedTarget proves a backlog queued
|
|
// against a target that has since been deleted stays visible, in the
|
|
// unknown series, instead of being dropped. That backlog is the one
|
|
// nobody is watching, so losing it would defeat the queue-depth
|
|
// alerting this metric exists for.
|
|
func TestDeliveryMetrics_QueueDepthDeletedTarget(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
s := newISetup(t)
|
|
reg := mIsolate(t, s)
|
|
|
|
iCreateWebhook(
|
|
t, s.MainDB, s.WebhookID, "deleted-target",
|
|
)
|
|
|
|
// No target row is created: this is a delivery whose target was
|
|
// deleted out from under it.
|
|
targetID := uuid.New().String()
|
|
|
|
event := iSeedEvent(
|
|
t, s.WebhookDB, s.WebhookID, `{"orphan":true}`,
|
|
)
|
|
|
|
iSeedDelivery(
|
|
t, s.WebhookDB, event.ID, targetID,
|
|
database.DeliveryStatusPending,
|
|
)
|
|
|
|
iSeedDelivery(
|
|
t, s.WebhookDB, event.ID, targetID,
|
|
database.DeliveryStatusRetrying,
|
|
)
|
|
|
|
s.Engine.ExportSampleQueueDepths(context.Background())
|
|
|
|
assert.InDelta(t, 1.0,
|
|
mGauge(t, reg, mPending, mTypeUnknown), 0)
|
|
assert.InDelta(t, 1.0,
|
|
mGauge(t, reg, mRetrying, mTypeUnknown), 0)
|
|
assert.InDelta(t, 0.0,
|
|
mGauge(t, reg, mPending, mTypeHTTP), 0)
|
|
}
|