Add a statistics pane to the webhook page (closes #368)
check / check (push) Failing after 1m15s
check / check (push) Failing after 1m15s
Each webhook's event database keeps one row of running totals: events, deliveries and failures, and how many of each retention removed. Storing an event, creating a delivery, a delivery becoming failed and the retention sweep each update it in the transaction that writes or deletes the rows it counts. Deliveries get a finished_at column, the last column of the status index, so the last-10-minutes and last-24-hours figures are index-range counts. The pane is its own template, included at the top of the page. The schema changes in place with nothing back-filled, so an existing database must be recreated. Model: opus-5-5
This commit is contained in:
@@ -1070,7 +1070,7 @@ unconditionally against whatever files it finds:
|
||||
- the main database on connect — `Setting`, `User`, `APIKey`, `Webhook`,
|
||||
`Entrypoint`, `Target`
|
||||
- each event database when it is lazily opened — `Event`, `Delivery`,
|
||||
`DeliveryResult`
|
||||
`DeliveryResult`, `Totals`
|
||||
- each archive database on every open and reopen
|
||||
|
||||
There is no schema version table, no migration ledger, and no down
|
||||
@@ -1384,7 +1384,7 @@ The codebase uses consistent naming throughout (rename completed in
|
||||
|
||||
### Data Model
|
||||
|
||||
webhooker's data model has nine entities organized into two tiers: the
|
||||
webhooker's data model has ten entities organized into two tiers: the
|
||||
**application tier** (user and webhook configuration) and the **event
|
||||
tier** (event ingestion, delivery, and logging).
|
||||
|
||||
@@ -1413,6 +1413,10 @@ tier** (event ingestion, delivery, and logging).
|
||||
│ ┌──────────┐ ┌──────────┐ ┌─────────────────┐ │
|
||||
│ │ Event │──1:N──│ Delivery │──1:N──│ DeliveryResult │ │
|
||||
│ └──────────┘ └──────────┘ └─────────────────┘ │
|
||||
│ │
|
||||
│ ┌──────────┐ │
|
||||
│ │ Totals │ (one row of running counts) │
|
||||
│ └──────────┘ │
|
||||
└─────────────────────────────────────────────────────────────┘
|
||||
```
|
||||
|
||||
@@ -1665,6 +1669,7 @@ status across potentially multiple attempts.
|
||||
| `event_id` | UUID | Foreign key → Event |
|
||||
| `target_id`| UUID | Foreign key → Target |
|
||||
| `status` | DeliveryStatus | One of: `pending`, `delivered`, `failed`, `retrying` |
|
||||
| `finished_at` | timestamp | When the delivery became `delivered` or `failed` (nullable; empty while `pending` or `retrying`) |
|
||||
|
||||
**Relations:** Belongs to Event. Belongs to Target. Has many
|
||||
DeliveryResults.
|
||||
@@ -1732,6 +1737,29 @@ retries) is individually logged for full observability.
|
||||
|
||||
**Relations:** Belongs to Delivery.
|
||||
|
||||
#### Totals
|
||||
|
||||
The one row of running counts in each event database, read by the
|
||||
statistics pane at the top of the webhook page.
|
||||
|
||||
| Field | Type | Description |
|
||||
| -------------------- | ------- | ----------- |
|
||||
| `events` | integer | Events ever stored, resubmitted copies included |
|
||||
| `deliveries` | integer | Deliveries ever created, replays included |
|
||||
| `failures` | integer | Deliveries that ever became `failed` |
|
||||
| `events_removed` | integer | Events retention has deleted |
|
||||
| `deliveries_removed` | integer | Deliveries retention has deleted |
|
||||
| `failures_removed` | integer | Failed deliveries retention has deleted |
|
||||
|
||||
Each count changes in the transaction that writes or deletes the rows it
|
||||
counts. The pane shows each of the first three as a lifetime figure, and
|
||||
less what retention removed as the figure within retention, so neither
|
||||
needs the rows themselves. Its last-10-minutes and last-24-hours figures
|
||||
are counted from the `events` and `deliveries` indexes over just that
|
||||
window. Its failure percentage for a window is the deliveries that became
|
||||
`failed` in it out of all that became `delivered` or `failed` in it, and
|
||||
a dash when none did.
|
||||
|
||||
#### Event-tier indexes
|
||||
|
||||
These indexes on the per-webhook event databases are declared in the model
|
||||
@@ -1739,10 +1767,10 @@ tags, so `AutoMigrate` creates them on a fresh and on an existing database:
|
||||
|
||||
| Table | Columns | Serves |
|
||||
| ------------------ | --------------------------- | ------ |
|
||||
| `deliveries` | `status`, `deleted_at` | Startup recovery, the retry and pending sweeps every 60 seconds and the queue-depth sampler every 30 seconds, which select deliveries by status |
|
||||
| `deliveries` | `status`, `deleted_at`, `finished_at` | Startup recovery, the retry and pending sweeps every 60 seconds and the queue-depth sampler every 30 seconds, which select deliveries by status, and the webhook page's statistics, which count deliveries by status and when they finished |
|
||||
| `deliveries` | `event_id`, `deleted_at` | The event log, which loads each event's deliveries, and retention, which selects and deletes the deliveries of expired events |
|
||||
| `delivery_results` | `delivery_id`, `deleted_at` | The event log, which loads the attempts of a page's deliveries, and retention, which deletes the attempts of expired events |
|
||||
| `events` | `deleted_at`, `created_at` | Retention, which selects expired events by age |
|
||||
| `events` | `deleted_at`, `created_at` | Retention, which selects expired events by age, and the webhook page's statistics, which count recent events and find the newest |
|
||||
| `events` | `created_at` | Retention's delete of the expired events themselves |
|
||||
|
||||
GORM's soft delete adds `deleted_at IS NULL` to these queries; retention's
|
||||
@@ -1756,9 +1784,10 @@ and SQLite narrows by a `<` only on the last column it uses.
|
||||
|
||||
#### Common Fields
|
||||
|
||||
Every entity except `Setting` includes these fields from `BaseModel`.
|
||||
`Setting` is a bare key-value row with no `id`, no timestamps and no
|
||||
soft delete:
|
||||
Every entity except `Setting` and `Totals` includes these fields from
|
||||
`BaseModel`. `Setting` is a bare key-value row with no `id`, no
|
||||
timestamps and no soft delete, and `Totals` is a single row of counts
|
||||
with only a numeric `id`:
|
||||
|
||||
| Field | Type | Description |
|
||||
| ------------ | --------- | ----------- |
|
||||
@@ -1800,6 +1829,7 @@ encryption key is generated and stored, and an `admin` user is created.
|
||||
- **Events** — captured incoming webhook payloads
|
||||
- **Deliveries** — event-to-target pairings and their status
|
||||
- **DeliveryResults** — individual delivery attempt logs
|
||||
- **Totals** — running counts of the above, kept through retention
|
||||
|
||||
Per-webhook databases are created automatically when a webhook is
|
||||
created (and lazily on first access for webhooks that predate this
|
||||
|
||||
@@ -123,7 +123,7 @@ func TestEventTierQueriesUseTheirIndexes(t *testing.T) {
|
||||
Order("attempt_num ASC").Find(&results),
|
||||
"idx_delivery_results_delivery_id (delivery_id=? AND deleted_at=?)")
|
||||
|
||||
// Retention's three deletes (reapExpired), whose subqueries are built
|
||||
// Retention's deletes (deleteExpired), whose subqueries are built
|
||||
// afresh for each statement as it builds them.
|
||||
expiredEventIDs := func() *gorm.DB {
|
||||
return dry.Model(&database.Event{}).Select("id").
|
||||
@@ -135,6 +135,14 @@ func TestEventTierQueriesUseTheirIndexes(t *testing.T) {
|
||||
Select("id").Where("event_id IN (?)", expiredEventIDs()),
|
||||
).Delete(&database.DeliveryResult{}),
|
||||
"idx_delivery_results_delivery_id (delivery_id=?)", byEvent, byAge)
|
||||
var failed struct{ Count int64 }
|
||||
|
||||
assertPlanUses(t, db, dry.Unscoped().Model(&database.Delivery{}).
|
||||
Select("count(CASE WHEN status = ? THEN 1 END) AS count",
|
||||
database.DeliveryStatusFailed).
|
||||
Where("event_id IN (?)", expiredEventIDs()).
|
||||
Take(&failed),
|
||||
"idx_deliveries_event_id (event_id=?)", byAge)
|
||||
assertPlanUses(t, db, dry.Unscoped().Where(
|
||||
"event_id IN (?)", expiredEventIDs(),
|
||||
).Delete(&database.Delivery{}),
|
||||
@@ -142,6 +150,36 @@ func TestEventTierQueriesUseTheirIndexes(t *testing.T) {
|
||||
assertPlanUses(t, db, dry.Unscoped().Where(
|
||||
"created_at < ?", cutoff,
|
||||
).Delete(&database.Event{}), "idx_events_created_at (created_at<?)")
|
||||
|
||||
// The webhook page's statistics (readEventStats in the handlers):
|
||||
// deliveries in progress, deliveries finished and events received
|
||||
// since a time, and the newest event, which must come straight
|
||||
// off an index rather than from sorting every event.
|
||||
var (
|
||||
count int64
|
||||
newest []time.Time
|
||||
)
|
||||
|
||||
assertPlanUses(t, db, dry.Model(&database.Delivery{}).
|
||||
Where("status IN ?", []database.DeliveryStatus{
|
||||
database.DeliveryStatusPending,
|
||||
database.DeliveryStatusRetrying,
|
||||
}).Count(&count), byStatus)
|
||||
assertPlanUses(t, db, dry.Model(&database.Delivery{}).
|
||||
Where("status = ? AND finished_at >= ?",
|
||||
database.DeliveryStatusFailed, cutoff).Count(&count),
|
||||
"idx_deliveries_status "+
|
||||
"(status=? AND deleted_at=? AND finished_at>?)")
|
||||
assertPlanUses(t, db, dry.Model(&database.Event{}).
|
||||
Where("created_at >= ?", cutoff).Count(&count),
|
||||
"idx_events_deleted_at_created_at "+
|
||||
"(deleted_at=? AND created_at>?)")
|
||||
|
||||
newestEvent := dry.Model(&database.Event{}).
|
||||
Order("created_at DESC").Limit(1).Pluck("created_at", &newest)
|
||||
assertPlanUses(t, db, newestEvent,
|
||||
"idx_events_deleted_at_created_at (deleted_at=?)")
|
||||
assert.NotContains(t, queryPlan(t, db, newestEvent), "TEMP B-TREE")
|
||||
}
|
||||
|
||||
// assertPlanUses asserts that SQLite's plan for a statement GORM built
|
||||
@@ -152,6 +190,18 @@ func assertPlanUses(
|
||||
) {
|
||||
t.Helper()
|
||||
|
||||
plan := queryPlan(t, db, built)
|
||||
|
||||
for _, index := range indexes {
|
||||
assert.Contains(t, plan, index, built.Statement.SQL.String())
|
||||
}
|
||||
}
|
||||
|
||||
// queryPlan returns SQLite's plan for a statement GORM built in a dry
|
||||
// run, run with the same SQL and arguments GORM would send.
|
||||
func queryPlan(t *testing.T, db, built *gorm.DB) string {
|
||||
t.Helper()
|
||||
|
||||
var plan []struct{ Detail string }
|
||||
|
||||
require.NoError(t, db.Raw(
|
||||
@@ -159,8 +209,5 @@ func assertPlanUses(
|
||||
built.Statement.Vars...,
|
||||
).Scan(&plan).Error)
|
||||
|
||||
for _, index := range indexes {
|
||||
assert.Contains(t, fmt.Sprint(plan), index,
|
||||
built.Statement.SQL.String())
|
||||
}
|
||||
return fmt.Sprint(plan)
|
||||
}
|
||||
|
||||
@@ -1,6 +1,10 @@
|
||||
package database
|
||||
|
||||
import "gorm.io/gorm"
|
||||
import (
|
||||
"time"
|
||||
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
// DeliveryStatus represents the status of a delivery
|
||||
type DeliveryStatus string
|
||||
@@ -45,6 +49,12 @@ type Delivery struct {
|
||||
// gives.
|
||||
DeletedAt gorm.DeletedAt `gorm:"index:idx_deliveries_event_id,priority:2;index:idx_deliveries_status,priority:2" json:"deletedAt,omitzero"`
|
||||
|
||||
// FinishedAt is when the delivery became delivered or failed, and
|
||||
// nil while it is pending or retrying. It ends the status index,
|
||||
// so the webhook page counts the deliveries that finished in a
|
||||
// recent window by reading that window from the index.
|
||||
FinishedAt *time.Time `gorm:"index:idx_deliveries_status,priority:3" json:"finishedAt,omitempty"`
|
||||
|
||||
// Relations
|
||||
Event Event `json:"event,omitzero"`
|
||||
Target Target `json:"target,omitzero"`
|
||||
|
||||
@@ -0,0 +1,73 @@
|
||||
package database
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
// Totals is the single row of running totals in a webhook's event
|
||||
// database. It is what keeps the webhook page's lifetime figures right
|
||||
// after retention has removed the rows they count, and what lets the
|
||||
// page show them without counting every row.
|
||||
//
|
||||
// Storing an event, creating a delivery and failing a delivery each
|
||||
// add one, and retention adds what it deletes to the Removed columns.
|
||||
// Every addition goes through AddTotals, in the transaction that
|
||||
// writes or deletes the rows it counts.
|
||||
type Totals struct {
|
||||
ID int64 `gorm:"primaryKey"`
|
||||
|
||||
Events int64 `gorm:"not null"`
|
||||
Deliveries int64 `gorm:"not null"`
|
||||
Failures int64 `gorm:"not null"`
|
||||
|
||||
EventsRemoved int64 `gorm:"not null"`
|
||||
DeliveriesRemoved int64 `gorm:"not null"`
|
||||
FailuresRemoved int64 `gorm:"not null"`
|
||||
}
|
||||
|
||||
// TableName names the table AddTotals updates.
|
||||
func (Totals) TableName() string {
|
||||
return "totals"
|
||||
}
|
||||
|
||||
// EventsWithinRetention is how many of the webhook's events are still
|
||||
// stored.
|
||||
func (t Totals) EventsWithinRetention() int64 {
|
||||
return t.Events - t.EventsRemoved
|
||||
}
|
||||
|
||||
// DeliveriesWithinRetention is how many of the webhook's deliveries
|
||||
// are still stored.
|
||||
func (t Totals) DeliveriesWithinRetention() int64 {
|
||||
return t.Deliveries - t.DeliveriesRemoved
|
||||
}
|
||||
|
||||
// FailuresWithinRetention is how many of the webhook's failed
|
||||
// deliveries are still stored.
|
||||
func (t Totals) FailuresWithinRetention() int64 {
|
||||
return t.Failures - t.FailuresRemoved
|
||||
}
|
||||
|
||||
// AddTotals adds each count in add to the webhook's running totals.
|
||||
// Call it on the transaction that writes or deletes the rows it
|
||||
// counts, so the totals change exactly when those rows do.
|
||||
func AddTotals(tx *gorm.DB, add Totals) error {
|
||||
err := tx.Exec(
|
||||
`UPDATE totals SET
|
||||
events = events + ?,
|
||||
deliveries = deliveries + ?,
|
||||
failures = failures + ?,
|
||||
events_removed = events_removed + ?,
|
||||
deliveries_removed = deliveries_removed + ?,
|
||||
failures_removed = failures_removed + ?`,
|
||||
add.Events, add.Deliveries, add.Failures,
|
||||
add.EventsRemoved, add.DeliveriesRemoved, add.FailuresRemoved,
|
||||
).Error
|
||||
if err != nil {
|
||||
return fmt.Errorf("adding to running totals: %w", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
@@ -2,7 +2,7 @@ package database
|
||||
|
||||
// Migrate runs database migrations for the main application database.
|
||||
// Only configuration-tier models are stored in the main database.
|
||||
// Event-tier models (Event, Delivery, DeliveryResult) live in
|
||||
// Event-tier models (Event, Delivery, DeliveryResult, Totals) live in
|
||||
// per-webhook dedicated databases managed by WebhookDBManager.
|
||||
func (d *Database) Migrate() error {
|
||||
return d.db.AutoMigrate(
|
||||
|
||||
@@ -267,55 +267,101 @@ func retentionCutoff(
|
||||
|
||||
// reapExpired hard-deletes, in foreign-key-safe order, the delivery
|
||||
// results, deliveries, and events associated with events older than
|
||||
// cutoff. Deletes are unscoped so rows are physically removed rather
|
||||
// than soft-deleted, reclaiming disk. It returns the number of events
|
||||
// deleted.
|
||||
// cutoff, and adds what it deleted to the running totals, all in one
|
||||
// transaction. Deletes are unscoped so rows are physically removed
|
||||
// rather than soft-deleted, reclaiming disk. It returns the number of
|
||||
// events deleted.
|
||||
func reapExpired(db *gorm.DB, cutoff time.Time) (int64, error) {
|
||||
var removed Totals
|
||||
|
||||
err := db.Transaction(func(tx *gorm.DB) error {
|
||||
var err error
|
||||
|
||||
removed, err = deleteExpired(tx, cutoff)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return AddTotals(tx, removed)
|
||||
})
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
return removed.EventsRemoved, nil
|
||||
}
|
||||
|
||||
// deleteExpired runs reapExpired's deletes and returns how many
|
||||
// events, deliveries and failed deliveries they removed.
|
||||
func deleteExpired(tx *gorm.DB, cutoff time.Time) (Totals, error) {
|
||||
var removed Totals
|
||||
|
||||
// Fresh subqueries are built per statement to avoid reusing a
|
||||
// mutated builder across executions.
|
||||
expiredEventIDs := func() *gorm.DB {
|
||||
return db.Model(&Event{}).
|
||||
return tx.Model(&Event{}).
|
||||
Select("id").
|
||||
Where("created_at < ?", cutoff)
|
||||
}
|
||||
expiredDeliveryIDs := func() *gorm.DB {
|
||||
return db.Model(&Delivery{}).
|
||||
return tx.Model(&Delivery{}).
|
||||
Select("id").
|
||||
Where("event_id IN (?)", expiredEventIDs())
|
||||
}
|
||||
|
||||
// 1. Delivery results whose delivery belongs to an expired event.
|
||||
res := db.Unscoped().
|
||||
res := tx.Unscoped().
|
||||
Where("delivery_id IN (?)", expiredDeliveryIDs()).
|
||||
Delete(&DeliveryResult{})
|
||||
if res.Error != nil {
|
||||
return 0, fmt.Errorf(
|
||||
return removed, fmt.Errorf(
|
||||
"deleting expired delivery results: %w",
|
||||
res.Error,
|
||||
)
|
||||
}
|
||||
|
||||
// 2. Deliveries belonging to an expired event.
|
||||
del := db.Unscoped().
|
||||
// 2. Deliveries belonging to an expired event, after counting the
|
||||
// failed ones among them. The status is tested in the select list
|
||||
// rather than the WHERE clause: there, SQLite would read every
|
||||
// failed delivery the webhook has through the status index,
|
||||
// instead of only the expired ones through the event_id index.
|
||||
var failed struct{ Count int64 }
|
||||
|
||||
err := tx.Unscoped().Model(&Delivery{}).
|
||||
Select("count(CASE WHEN status = ? THEN 1 END) AS count",
|
||||
DeliveryStatusFailed).
|
||||
Where("event_id IN (?)", expiredEventIDs()).
|
||||
Take(&failed).Error
|
||||
if err != nil {
|
||||
return removed, fmt.Errorf(
|
||||
"counting expired failed deliveries: %w", err,
|
||||
)
|
||||
}
|
||||
|
||||
del := tx.Unscoped().
|
||||
Where("event_id IN (?)", expiredEventIDs()).
|
||||
Delete(&Delivery{})
|
||||
if del.Error != nil {
|
||||
return 0, fmt.Errorf(
|
||||
return removed, fmt.Errorf(
|
||||
"deleting expired deliveries: %w",
|
||||
del.Error,
|
||||
)
|
||||
}
|
||||
|
||||
// 3. The expired events themselves.
|
||||
ev := db.Unscoped().
|
||||
ev := tx.Unscoped().
|
||||
Where("created_at < ?", cutoff).
|
||||
Delete(&Event{})
|
||||
if ev.Error != nil {
|
||||
return 0, fmt.Errorf(
|
||||
return removed, fmt.Errorf(
|
||||
"deleting expired events: %w",
|
||||
ev.Error,
|
||||
)
|
||||
}
|
||||
|
||||
return ev.RowsAffected, nil
|
||||
removed.EventsRemoved = ev.RowsAffected
|
||||
removed.DeliveriesRemoved = del.RowsAffected
|
||||
removed.FailuresRemoved = failed.Count
|
||||
|
||||
return removed, nil
|
||||
}
|
||||
|
||||
@@ -0,0 +1,126 @@
|
||||
package database_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"gorm.io/gorm"
|
||||
"sneak.berlin/go/webhooker/internal/database"
|
||||
)
|
||||
|
||||
// readTotals reads a webhook database's row of running totals,
|
||||
// asserting that it has exactly one.
|
||||
func readTotals(t *testing.T, db *gorm.DB) database.Totals {
|
||||
t.Helper()
|
||||
|
||||
var rows []database.Totals
|
||||
|
||||
require.NoError(t, db.Find(&rows).Error)
|
||||
require.Len(t, rows, 1)
|
||||
|
||||
return rows[0]
|
||||
}
|
||||
|
||||
// TestWebhookDBManager_TotalsRowSurvivesReopen verifies that a new
|
||||
// event database starts with one row of zero totals, and that opening
|
||||
// it again keeps that row and what was added to it.
|
||||
func TestWebhookDBManager_TotalsRowSurvivesReopen(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
mgr, lc := setupTestWebhookDBManager(t)
|
||||
ctx := context.Background()
|
||||
require.NoError(t, lc.Start(ctx))
|
||||
|
||||
defer func() { require.NoError(t, lc.Stop(ctx)) }()
|
||||
|
||||
webhookID := uuid.New().String()
|
||||
|
||||
db, err := mgr.GetDB(webhookID)
|
||||
require.NoError(t, err)
|
||||
|
||||
fresh := readTotals(t, db)
|
||||
assert.Equal(t, database.Totals{ID: fresh.ID}, fresh)
|
||||
|
||||
require.NoError(t, database.AddTotals(db, database.Totals{
|
||||
Events: 2, Deliveries: 3, Failures: 1,
|
||||
}))
|
||||
|
||||
// Drop the cached connection so the next open reopens the file,
|
||||
// as a restart would.
|
||||
require.NoError(t, mgr.CloseAll())
|
||||
|
||||
db, err = mgr.GetDB(webhookID)
|
||||
require.NoError(t, err)
|
||||
|
||||
assert.Equal(t, database.Totals{
|
||||
ID: fresh.ID, Events: 2, Deliveries: 3, Failures: 1,
|
||||
}, readTotals(t, db))
|
||||
}
|
||||
|
||||
// TestRetentionReaper_AddsWhatItRemovesToTotals verifies that a sweep
|
||||
// leaves the lifetime totals alone and adds the events, deliveries and
|
||||
// failed deliveries it deletes to the removed totals, so the totals
|
||||
// within retention match the rows still stored.
|
||||
func TestRetentionReaper_AddsWhatItRemovesToTotals(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
env := setupRetentionTest(t)
|
||||
|
||||
webhookID := createWebhook(t, env.mainDB.DB(), 30)
|
||||
|
||||
db, err := env.mgr.GetDB(webhookID)
|
||||
require.NoError(t, err)
|
||||
|
||||
now := time.Now()
|
||||
expired := now.Add(-40 * 24 * time.Hour)
|
||||
|
||||
seedEventChain(t, db, webhookID, expired)
|
||||
expiredFailure := seedEventChain(t, db, webhookID, expired)
|
||||
recentFailure := seedEventChain(
|
||||
t, db, webhookID, now.Add(-24*time.Hour),
|
||||
)
|
||||
|
||||
for _, id := range []string{
|
||||
expiredFailure.deliveryID, recentFailure.deliveryID,
|
||||
} {
|
||||
require.NoError(t, db.Model(&database.Delivery{}).
|
||||
Where("id = ?", id).
|
||||
Update("status", database.DeliveryStatusFailed).Error)
|
||||
}
|
||||
|
||||
// The totals storing those rows would have left.
|
||||
require.NoError(t, database.AddTotals(db, database.Totals{
|
||||
Events: 3, Deliveries: 3, Failures: 2,
|
||||
}))
|
||||
|
||||
env.reaper.ExportSweep(context.Background())
|
||||
|
||||
totals := readTotals(t, db)
|
||||
assert.Equal(t, database.Totals{
|
||||
ID: totals.ID,
|
||||
Events: 3, Deliveries: 3, Failures: 2,
|
||||
EventsRemoved: 2, DeliveriesRemoved: 2, FailuresRemoved: 1,
|
||||
}, totals)
|
||||
|
||||
var events, deliveries, failures int64
|
||||
|
||||
require.NoError(t, db.Model(&database.Event{}).Count(&events).Error)
|
||||
require.NoError(t, db.Model(&database.Delivery{}).
|
||||
Count(&deliveries).Error)
|
||||
require.NoError(t, db.Model(&database.Delivery{}).
|
||||
Where("status = ?", database.DeliveryStatusFailed).
|
||||
Count(&failures).Error)
|
||||
|
||||
assert.Equal(t, events, totals.EventsWithinRetention())
|
||||
assert.Equal(t, deliveries, totals.DeliveriesWithinRetention())
|
||||
assert.Equal(t, failures, totals.FailuresWithinRetention())
|
||||
|
||||
// A sweep with nothing left to remove changes nothing.
|
||||
env.reaper.ExportSweep(context.Background())
|
||||
|
||||
assert.Equal(t, totals, readTotals(t, db))
|
||||
}
|
||||
@@ -35,7 +35,8 @@ var errInvalidCachedDBType = errors.New(
|
||||
|
||||
// WebhookDBManager manages per-webhook SQLite database files
|
||||
// for event storage. Each webhook gets its own dedicated
|
||||
// database containing Events, Deliveries, and DeliveryResults.
|
||||
// database containing Events, Deliveries, DeliveryResults and the
|
||||
// running Totals of them.
|
||||
// Database connections are opened lazily and cached.
|
||||
type WebhookDBManager struct {
|
||||
dataDir string
|
||||
@@ -294,7 +295,7 @@ func (m *WebhookDBManager) openDB(
|
||||
|
||||
// Run migrations for event-tier models only
|
||||
err = db.AutoMigrate(
|
||||
&Event{}, &Delivery{}, &DeliveryResult{},
|
||||
&Event{}, &Delivery{}, &DeliveryResult{}, &Totals{},
|
||||
)
|
||||
if err != nil {
|
||||
_ = sqlDB.Close()
|
||||
@@ -305,6 +306,17 @@ func (m *WebhookDBManager) openDB(
|
||||
)
|
||||
}
|
||||
|
||||
// A new database gets its row of running totals, all zero.
|
||||
err = db.FirstOrCreate(&Totals{}).Error
|
||||
if err != nil {
|
||||
_ = sqlDB.Close()
|
||||
|
||||
return nil, fmt.Errorf(
|
||||
"creating running totals for webhook database %s: %w",
|
||||
webhookID, err,
|
||||
)
|
||||
}
|
||||
|
||||
m.log.Info(
|
||||
"opened per-webhook database",
|
||||
"webhook_id", webhookID,
|
||||
|
||||
@@ -0,0 +1,100 @@
|
||||
package delivery_test
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"gorm.io/gorm"
|
||||
"sneak.berlin/go/webhooker/internal/database"
|
||||
)
|
||||
|
||||
// failureTotal reads the running failure total of a webhook database.
|
||||
func failureTotal(t *testing.T, db *gorm.DB) int64 {
|
||||
t.Helper()
|
||||
|
||||
var totals database.Totals
|
||||
|
||||
require.NoError(t, db.Take(&totals).Error)
|
||||
|
||||
return totals.Failures
|
||||
}
|
||||
|
||||
// TestUpdateDeliveryStatus_FinishTimeAndFailureTotal pins what a status
|
||||
// write records for the webhook page's statistics: the time a delivery
|
||||
// finished, set only when it becomes delivered or failed, and one more
|
||||
// on the failure total when it fails.
|
||||
func TestUpdateDeliveryStatus_FinishTimeAndFailureTotal(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
tests := []struct {
|
||||
status database.DeliveryStatus
|
||||
finished bool
|
||||
failures int64
|
||||
}{
|
||||
{database.DeliveryStatusRetrying, false, 0},
|
||||
{database.DeliveryStatusDelivered, true, 0},
|
||||
{database.DeliveryStatusFailed, true, 1},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(string(tt.status), func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
db := testWebhookDB(t)
|
||||
e := testEngine(t, 1)
|
||||
event := seedEvent(t, db, `{}`)
|
||||
d := seedDelivery(
|
||||
t, db, event.ID, uuid.New().String(),
|
||||
database.DeliveryStatusPending,
|
||||
)
|
||||
|
||||
before := time.Now()
|
||||
|
||||
require.NoError(t, e.ExportUpdateDeliveryStatus(
|
||||
db, &d, tt.status,
|
||||
))
|
||||
|
||||
var stored database.Delivery
|
||||
|
||||
require.NoError(t, db.First(&stored, "id = ?", d.ID).Error)
|
||||
assert.Equal(t, tt.status, stored.Status)
|
||||
|
||||
if tt.finished {
|
||||
require.NotNil(t, stored.FinishedAt)
|
||||
assert.False(t, stored.FinishedAt.Before(before))
|
||||
} else {
|
||||
assert.Nil(t, stored.FinishedAt)
|
||||
}
|
||||
|
||||
assert.Equal(t, tt.failures, failureTotal(t, db))
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestUpdateDeliveryStatus_DeletedDeliveryIsNotCounted covers a
|
||||
// delivery retention deleted while the engine still held it. Failing
|
||||
// it afterwards writes no row, so it adds no failure either: retention
|
||||
// has already counted what it removed.
|
||||
func TestUpdateDeliveryStatus_DeletedDeliveryIsNotCounted(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
db := testWebhookDB(t)
|
||||
e := testEngine(t, 1)
|
||||
event := seedEvent(t, db, `{}`)
|
||||
d := seedDelivery(
|
||||
t, db, event.ID, uuid.New().String(),
|
||||
database.DeliveryStatusRetrying,
|
||||
)
|
||||
|
||||
require.NoError(t, db.Unscoped().
|
||||
Delete(&database.Delivery{}, "id = ?", d.ID).Error)
|
||||
|
||||
require.NoError(t, e.ExportUpdateDeliveryStatus(
|
||||
db, &d, database.DeliveryStatusFailed,
|
||||
))
|
||||
|
||||
assert.Zero(t, failureTotal(t, db))
|
||||
}
|
||||
@@ -1554,8 +1554,9 @@ func (e *Engine) updateDeliveryStatus(
|
||||
targetType database.TargetType,
|
||||
status database.DeliveryStatus,
|
||||
) error {
|
||||
err := webhookDB.Model(d).
|
||||
Update("status", status).Error
|
||||
err := webhookDB.Transaction(func(tx *gorm.DB) error {
|
||||
return writeDeliveryStatus(tx, d, status)
|
||||
})
|
||||
if err != nil {
|
||||
return fmt.Errorf(
|
||||
"updating delivery %s to status %s: %w",
|
||||
@@ -1574,6 +1575,33 @@ func (e *Engine) updateDeliveryStatus(
|
||||
return nil
|
||||
}
|
||||
|
||||
// writeDeliveryStatus writes a delivery's new status. A delivery that
|
||||
// becomes delivered or failed also gets the time it finished, and a
|
||||
// failed one is added to the webhook's running failure total. The
|
||||
// failure is counted only if the row was still there to update:
|
||||
// retention may have deleted it while the engine was working on it.
|
||||
func writeDeliveryStatus(
|
||||
tx *gorm.DB,
|
||||
d *database.Delivery,
|
||||
status database.DeliveryStatus,
|
||||
) error {
|
||||
columns := map[string]any{"status": status}
|
||||
if status.Terminal() {
|
||||
columns["finished_at"] = time.Now()
|
||||
}
|
||||
|
||||
res := tx.Model(d).Updates(columns)
|
||||
if res.Error != nil {
|
||||
return res.Error
|
||||
}
|
||||
|
||||
if status == database.DeliveryStatusFailed && res.RowsAffected > 0 {
|
||||
return database.AddTotals(tx, database.Totals{Failures: 1})
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// settleStatus moves a delivery to its outcome status and reports a
|
||||
// failed write through bookkeepingFailed, which leaves the row
|
||||
// recoverable. It exists so the target call sites read as one
|
||||
|
||||
@@ -57,7 +57,9 @@ func testWebhookDB(t *testing.T) *gorm.DB {
|
||||
&database.Event{},
|
||||
&database.Delivery{},
|
||||
&database.DeliveryResult{},
|
||||
&database.Totals{},
|
||||
))
|
||||
require.NoError(t, db.Create(&database.Totals{}).Error)
|
||||
|
||||
return db
|
||||
}
|
||||
|
||||
@@ -150,6 +150,16 @@ func (e *Engine) ExportDeliverSlack(
|
||||
)
|
||||
}
|
||||
|
||||
// ExportUpdateDeliveryStatus exposes updateDeliveryStatus. It passes no
|
||||
// target type, so no metric moves.
|
||||
func (e *Engine) ExportUpdateDeliveryStatus(
|
||||
webhookDB *gorm.DB,
|
||||
d *database.Delivery,
|
||||
status database.DeliveryStatus,
|
||||
) error {
|
||||
return e.updateDeliveryStatus(webhookDB, d, "", status)
|
||||
}
|
||||
|
||||
// ExportProcessNewTask exposes processNewTask.
|
||||
func (e *Engine) ExportProcessNewTask(
|
||||
ctx context.Context, task *Task,
|
||||
|
||||
@@ -299,7 +299,8 @@ func countInFlightDeliveries(
|
||||
return count, err
|
||||
}
|
||||
|
||||
// createReplayDelivery writes the new pending delivery row and returns
|
||||
// createReplayDelivery writes the new pending delivery row, adds it to
|
||||
// the webhook's running totals in the same transaction, and returns
|
||||
// the task that carries it to the delivery engine.
|
||||
//
|
||||
// The row is written with associations omitted, and neither Event nor
|
||||
@@ -319,7 +320,14 @@ func createReplayDelivery(
|
||||
Status: database.DeliveryStatusPending,
|
||||
}
|
||||
|
||||
err := webhookDB.Omit(clause.Associations).Create(dlv).Error
|
||||
err := webhookDB.Transaction(func(tx *gorm.DB) error {
|
||||
err := tx.Omit(clause.Associations).Create(dlv).Error
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return database.AddTotals(tx, database.Totals{Deliveries: 1})
|
||||
})
|
||||
if err != nil {
|
||||
return delivery.Task{}, err
|
||||
}
|
||||
|
||||
@@ -69,6 +69,21 @@ func (s *Handlers) LoadEventLogViewsForTest(
|
||||
return views
|
||||
}
|
||||
|
||||
// WebhookStatsForTest returns the figures the statistics pane on a
|
||||
// webhook's page shows, from the webhook's entrypoints and targets
|
||||
// loaded as that page loads them.
|
||||
func (s *Handlers) WebhookStatsForTest(webhookID string) *WebhookStats {
|
||||
var entrypoints []database.Entrypoint
|
||||
|
||||
s.db.DB().Where("webhook_id = ?", webhookID).Find(&entrypoints)
|
||||
|
||||
var targets []database.Target
|
||||
|
||||
s.db.DB().Where("webhook_id = ?", webhookID).Find(&targets)
|
||||
|
||||
return s.loadWebhookStats(webhookID, entrypoints, targets)
|
||||
}
|
||||
|
||||
// AddTemplateForTest registers a template under a page name so that
|
||||
// the handlers_test package can drive the render path with a
|
||||
// template of its own.
|
||||
|
||||
@@ -91,18 +91,22 @@ type Handlers struct {
|
||||
|
||||
// parsePageTemplate parses a page-specific template set from the
|
||||
// embedded FS. Each page template is combined with the shared
|
||||
// base, htmlheader, and navbar templates. The page file must be
|
||||
// listed first so that its root action ({{template "base" .}})
|
||||
// becomes the template set's entry point.
|
||||
func parsePageTemplate(pageFile string) *template.Template {
|
||||
// base, htmlheader, and navbar templates, and with any further files
|
||||
// the page includes. The page file must be listed first so that its
|
||||
// root action ({{template "base" .}}) becomes the template set's entry
|
||||
// point.
|
||||
func parsePageTemplate(
|
||||
pageFile string, included ...string,
|
||||
) *template.Template {
|
||||
files := append([]string{
|
||||
pageFile,
|
||||
"base.html",
|
||||
"htmlheader.html",
|
||||
"navbar.html",
|
||||
}, included...)
|
||||
|
||||
return template.Must(
|
||||
template.ParseFS(
|
||||
templates.Templates,
|
||||
pageFile,
|
||||
"base.html",
|
||||
"htmlheader.html",
|
||||
"navbar.html",
|
||||
),
|
||||
template.ParseFS(templates.Templates, files...),
|
||||
)
|
||||
}
|
||||
|
||||
@@ -131,7 +135,7 @@ func New(
|
||||
"profile.html": parsePageTemplate("profile.html"),
|
||||
"sources_list.html": parsePageTemplate("sources_list.html"),
|
||||
"sources_new.html": parsePageTemplate("sources_new.html"),
|
||||
"source_detail.html": parsePageTemplate("source_detail.html"),
|
||||
"source_detail.html": parsePageTemplate("source_detail.html", "webhook_stats.html"),
|
||||
"source_edit.html": parsePageTemplate("source_edit.html"),
|
||||
"source_logs.html": parsePageTemplate("source_logs.html"),
|
||||
"target_edit.html": parsePageTemplate("target_edit.html"),
|
||||
|
||||
@@ -450,6 +450,7 @@ func (h *Handlers) renderSourceDetail(
|
||||
"Targets": delivery.NewTargetViews(targets),
|
||||
"Events": events,
|
||||
"BaseURL": baseURL,
|
||||
"Stats": h.loadWebhookStats(webhook.ID, entrypoints, targets),
|
||||
}
|
||||
|
||||
h.renderTemplate(w, r, "source_detail.html", data)
|
||||
|
||||
@@ -252,11 +252,12 @@ func requestEventSource(
|
||||
}
|
||||
}
|
||||
|
||||
// createAndFanOut writes the event and one pending delivery per target
|
||||
// in a single transaction, then hands the tasks to the delivery
|
||||
// engine. It is the only path by which an event and its deliveries are
|
||||
// created, so a resubmitted event is retried, SSRF-guarded and
|
||||
// circuit-broken exactly as a received one is.
|
||||
// createAndFanOut writes the event and one pending delivery per target,
|
||||
// and adds them to the webhook's running totals, in a single
|
||||
// transaction, then hands the tasks to the delivery engine. It is the
|
||||
// only path by which an event and its deliveries are created, so a
|
||||
// resubmitted event is retried, SSRF-guarded and circuit-broken
|
||||
// exactly as a received one is.
|
||||
//
|
||||
// The tasks are returned as well as queued, so a caller can report how
|
||||
// many targets the event went to.
|
||||
@@ -296,6 +297,16 @@ func (h *Handlers) createAndFanOut(
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
err = database.AddTotals(tx, database.Totals{
|
||||
Events: 1,
|
||||
Deliveries: int64(len(tasks)),
|
||||
})
|
||||
if err != nil {
|
||||
tx.Rollback()
|
||||
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
err = tx.Commit().Error
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf(
|
||||
|
||||
@@ -0,0 +1,211 @@
|
||||
package handlers
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"gorm.io/gorm"
|
||||
"sneak.berlin/go/webhooker/internal/database"
|
||||
)
|
||||
|
||||
// The spans of the two recent windows the statistics pane reports on.
|
||||
const (
|
||||
tenMinutes = 10 * time.Minute
|
||||
twentyFourHours = 24 * time.Hour
|
||||
)
|
||||
|
||||
// percent turns a fraction into a percentage.
|
||||
const percent = 100
|
||||
|
||||
// WebhookStats holds the figures in the statistics pane at the top of
|
||||
// the webhook page.
|
||||
type WebhookStats struct {
|
||||
Entrypoints int
|
||||
ActiveEntrypoints int
|
||||
Targets int
|
||||
ActiveTargets int
|
||||
|
||||
// Totals holds the lifetime counts of events, deliveries and
|
||||
// failures, and how many of each retention has removed.
|
||||
Totals database.Totals
|
||||
|
||||
// InProgress counts the deliveries still pending or retrying.
|
||||
InProgress int64
|
||||
|
||||
// LastEventAt is when the newest stored event arrived, or nil when
|
||||
// none is stored.
|
||||
LastEventAt *time.Time
|
||||
|
||||
Last10Minutes RecentWindow
|
||||
Last24Hours RecentWindow
|
||||
}
|
||||
|
||||
// RecentWindow holds what happened in one recent window: the events
|
||||
// received in it, and the deliveries that became delivered or failed in
|
||||
// it.
|
||||
type RecentWindow struct {
|
||||
Events int64
|
||||
Delivered int64
|
||||
Failed int64
|
||||
}
|
||||
|
||||
// FailurePercent is the share of the deliveries finished in the window
|
||||
// that failed, or a dash when none finished. Deliveries still pending
|
||||
// or retrying are not counted either way.
|
||||
func (w RecentWindow) FailurePercent() string {
|
||||
finished := w.Delivered + w.Failed
|
||||
if finished == 0 {
|
||||
return "—"
|
||||
}
|
||||
|
||||
return fmt.Sprintf(
|
||||
"%.1f%%", percent*float64(w.Failed)/float64(finished),
|
||||
)
|
||||
}
|
||||
|
||||
// loadWebhookStats gathers the figures for the statistics pane from the
|
||||
// webhook's entrypoints and targets, as the page has already loaded
|
||||
// them, and from its event database. It returns nil, and logs why, when
|
||||
// the event database cannot be read.
|
||||
func (h *Handlers) loadWebhookStats(
|
||||
webhookID string,
|
||||
entrypoints []database.Entrypoint,
|
||||
targets []database.Target,
|
||||
) *WebhookStats {
|
||||
stats := &WebhookStats{
|
||||
Entrypoints: len(entrypoints),
|
||||
Targets: len(targets),
|
||||
}
|
||||
|
||||
for i := range entrypoints {
|
||||
if entrypoints[i].Active {
|
||||
stats.ActiveEntrypoints++
|
||||
}
|
||||
}
|
||||
|
||||
for i := range targets {
|
||||
if targets[i].Active {
|
||||
stats.ActiveTargets++
|
||||
}
|
||||
}
|
||||
|
||||
// Opening an event database that does not exist would create it,
|
||||
// and it would hold nothing to count.
|
||||
if !h.dbMgr.DBExists(webhookID) {
|
||||
return stats
|
||||
}
|
||||
|
||||
webhookDB, err := h.dbMgr.GetDB(webhookID)
|
||||
if err == nil {
|
||||
err = readEventStats(webhookDB, time.Now(), stats)
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
h.log.Error(
|
||||
"failed to read webhook statistics",
|
||||
"webhook_id", webhookID,
|
||||
"error", err,
|
||||
)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
return stats
|
||||
}
|
||||
|
||||
// readEventStats fills in the figures that come from the webhook's
|
||||
// event database. None of them reads every stored row: the totals are
|
||||
// one row, and every other figure is read from an index, over only the
|
||||
// rows it counts.
|
||||
func readEventStats(
|
||||
db *gorm.DB, now time.Time, stats *WebhookStats,
|
||||
) error {
|
||||
err := db.Take(&stats.Totals).Error
|
||||
if err != nil {
|
||||
return fmt.Errorf("reading running totals: %w", err)
|
||||
}
|
||||
|
||||
err = db.Model(&database.Delivery{}).
|
||||
Where("status IN ?", []database.DeliveryStatus{
|
||||
database.DeliveryStatusPending,
|
||||
database.DeliveryStatusRetrying,
|
||||
}).
|
||||
Count(&stats.InProgress).Error
|
||||
if err != nil {
|
||||
return fmt.Errorf("counting deliveries in progress: %w", err)
|
||||
}
|
||||
|
||||
var newest []time.Time
|
||||
|
||||
err = db.Model(&database.Event{}).
|
||||
Order("created_at DESC").
|
||||
Limit(1).
|
||||
Pluck("created_at", &newest).Error
|
||||
if err != nil {
|
||||
return fmt.Errorf("reading newest event time: %w", err)
|
||||
}
|
||||
|
||||
if len(newest) > 0 {
|
||||
stats.LastEventAt = &newest[0]
|
||||
}
|
||||
|
||||
stats.Last10Minutes, err = readRecentWindow(
|
||||
db, now.Add(-tenMinutes),
|
||||
)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
stats.Last24Hours, err = readRecentWindow(
|
||||
db, now.Add(-twentyFourHours),
|
||||
)
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
// readRecentWindow counts the events received, and the deliveries that
|
||||
// became delivered or failed, since the given time.
|
||||
func readRecentWindow(
|
||||
db *gorm.DB, since time.Time,
|
||||
) (RecentWindow, error) {
|
||||
var w RecentWindow
|
||||
|
||||
err := db.Model(&database.Event{}).
|
||||
Where("created_at >= ?", since).
|
||||
Count(&w.Events).Error
|
||||
if err != nil {
|
||||
return w, fmt.Errorf("counting recent events: %w", err)
|
||||
}
|
||||
|
||||
w.Delivered, err = countFinishedSince(
|
||||
db, database.DeliveryStatusDelivered, since,
|
||||
)
|
||||
if err != nil {
|
||||
return w, err
|
||||
}
|
||||
|
||||
w.Failed, err = countFinishedSince(
|
||||
db, database.DeliveryStatusFailed, since,
|
||||
)
|
||||
|
||||
return w, err
|
||||
}
|
||||
|
||||
// countFinishedSince counts the deliveries that reached the given
|
||||
// final status since the given time.
|
||||
func countFinishedSince(
|
||||
db *gorm.DB, status database.DeliveryStatus, since time.Time,
|
||||
) (int64, error) {
|
||||
var n int64
|
||||
|
||||
err := db.Model(&database.Delivery{}).
|
||||
Where("status = ? AND finished_at >= ?", status, since).
|
||||
Count(&n).Error
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf(
|
||||
"counting deliveries %s recently: %w", status, err,
|
||||
)
|
||||
}
|
||||
|
||||
return n, nil
|
||||
}
|
||||
@@ -0,0 +1,331 @@
|
||||
package handlers_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/go-chi/chi"
|
||||
"github.com/google/uuid"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"go.uber.org/fx/fxtest"
|
||||
"gorm.io/gorm"
|
||||
"gorm.io/gorm/clause"
|
||||
"sneak.berlin/go/webhooker/internal/config"
|
||||
"sneak.berlin/go/webhooker/internal/database"
|
||||
"sneak.berlin/go/webhooker/internal/handlers"
|
||||
"sneak.berlin/go/webhooker/internal/logger"
|
||||
"sneak.berlin/go/webhooker/internal/session"
|
||||
)
|
||||
|
||||
// statsReceive posts one event to an entrypoint through the real
|
||||
// receiver.
|
||||
func statsReceive(t *testing.T, h *handlers.Handlers, path string) {
|
||||
t.Helper()
|
||||
|
||||
req := httptest.NewRequestWithContext(
|
||||
context.Background(), http.MethodPost, "/",
|
||||
strings.NewReader(`{"stats":true}`),
|
||||
)
|
||||
|
||||
rctx := chi.NewRouteContext()
|
||||
rctx.URLParams.Add("uuid", path)
|
||||
|
||||
req = req.WithContext(context.WithValue(
|
||||
req.Context(), chi.RouteCtxKey, rctx,
|
||||
))
|
||||
|
||||
w := httptest.NewRecorder()
|
||||
h.HandleWebhook().ServeHTTP(w, req)
|
||||
|
||||
require.Equal(t, http.StatusOK, w.Code)
|
||||
}
|
||||
|
||||
// statsEntrypoint adds an entrypoint to a webhook and returns its path.
|
||||
func statsEntrypoint(
|
||||
t *testing.T, db *database.Database, webhookID string, active bool,
|
||||
) string {
|
||||
t.Helper()
|
||||
|
||||
ep := &database.Entrypoint{
|
||||
WebhookID: webhookID,
|
||||
Path: uuid.New().String(),
|
||||
}
|
||||
|
||||
require.NoError(t, db.DB().Omit(clause.Associations).Create(ep).Error)
|
||||
require.NoError(t, db.DB().Model(ep).Update("active", active).Error)
|
||||
|
||||
return ep.Path
|
||||
}
|
||||
|
||||
// statsDelivery returns the id of an event's delivery to a target.
|
||||
func statsDelivery(
|
||||
t *testing.T, webhookDB *gorm.DB, eventID, targetID string,
|
||||
) string {
|
||||
t.Helper()
|
||||
|
||||
var d database.Delivery
|
||||
|
||||
require.NoError(t, webhookDB.Where(
|
||||
"event_id = ? AND target_id = ?", eventID, targetID,
|
||||
).First(&d).Error)
|
||||
|
||||
return d.ID
|
||||
}
|
||||
|
||||
// statsFinish settles a delivery as the delivery engine does: its
|
||||
// final status and the time it finished, and for a failure one more on
|
||||
// the webhook's failure total, in one transaction.
|
||||
func statsFinish(
|
||||
t *testing.T,
|
||||
webhookDB *gorm.DB,
|
||||
deliveryID string,
|
||||
status database.DeliveryStatus,
|
||||
at time.Time,
|
||||
) {
|
||||
t.Helper()
|
||||
|
||||
require.NoError(t, webhookDB.Transaction(func(tx *gorm.DB) error {
|
||||
err := tx.Model(&database.Delivery{}).
|
||||
Where("id = ?", deliveryID).
|
||||
Updates(map[string]any{"status": status, "finished_at": at}).
|
||||
Error
|
||||
if err != nil || status != database.DeliveryStatusFailed {
|
||||
return err
|
||||
}
|
||||
|
||||
return database.AddTotals(tx, database.Totals{Failures: 1})
|
||||
}))
|
||||
}
|
||||
|
||||
// statsAge moves an event's arrival back to the given time.
|
||||
func statsAge(
|
||||
t *testing.T, webhookDB *gorm.DB, eventID string, at time.Time,
|
||||
) {
|
||||
t.Helper()
|
||||
|
||||
require.NoError(t, webhookDB.Model(&database.Event{}).
|
||||
Where("id = ?", eventID).
|
||||
Update("created_at", at).Error)
|
||||
}
|
||||
|
||||
// statsPrune runs the real retention reaper until it has removed one
|
||||
// event from the webhook's database, then stops it.
|
||||
func statsPrune(
|
||||
t *testing.T,
|
||||
db *database.Database,
|
||||
dbMgr *database.WebhookDBManager,
|
||||
log *logger.Logger,
|
||||
webhookDB *gorm.DB,
|
||||
) {
|
||||
t.Helper()
|
||||
|
||||
lc := fxtest.NewLifecycle(t)
|
||||
database.NewRetentionReaper(lc, database.RetentionReaperParams{
|
||||
Config: &config.Config{
|
||||
RetentionSweepInterval: 10 * time.Millisecond,
|
||||
},
|
||||
Database: db,
|
||||
DBManager: dbMgr,
|
||||
Logger: log,
|
||||
})
|
||||
|
||||
lc.RequireStart()
|
||||
|
||||
require.Eventually(t, func() bool {
|
||||
var totals database.Totals
|
||||
|
||||
err := webhookDB.Take(&totals).Error
|
||||
|
||||
return err == nil && totals.EventsRemoved == 1
|
||||
}, 10*time.Second, 10*time.Millisecond)
|
||||
|
||||
lc.RequireStop()
|
||||
}
|
||||
|
||||
// assertStatsTotals checks the lifetime events, deliveries and
|
||||
// failures, and those within retention.
|
||||
func assertStatsTotals(
|
||||
t *testing.T, totals database.Totals, lifetime, within [3]int64,
|
||||
) {
|
||||
t.Helper()
|
||||
|
||||
assert.Equal(t, lifetime, [3]int64{
|
||||
totals.Events, totals.Deliveries, totals.Failures,
|
||||
}, "lifetime events, deliveries, failures")
|
||||
assert.Equal(t, within, [3]int64{
|
||||
totals.EventsWithinRetention(),
|
||||
totals.DeliveriesWithinRetention(),
|
||||
totals.FailuresWithinRetention(),
|
||||
}, "events, deliveries, failures within retention")
|
||||
}
|
||||
|
||||
// TestWebhookStats_EveryFigureAcrossRetentionPrune builds a webhook's
|
||||
// history through the receiver and a replay, settles its deliveries as
|
||||
// the delivery engine would, and checks every figure the statistics
|
||||
// pane shows, before and after the real retention reaper removes the
|
||||
// oldest event.
|
||||
func TestWebhookStats_EveryFigureAcrossRetentionPrune(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
var (
|
||||
h *handlers.Handlers
|
||||
sess *session.Session
|
||||
db *database.Database
|
||||
dbMgr *database.WebhookDBManager
|
||||
log *logger.Logger
|
||||
)
|
||||
|
||||
app := newTestApp(t, &h, &sess, &db, &dbMgr, &log)
|
||||
app.RequireStart()
|
||||
|
||||
t.Cleanup(app.RequireStop)
|
||||
|
||||
wh := &database.Webhook{
|
||||
UserID: deleteTestUserID, Name: "stats", RetentionDays: 1,
|
||||
}
|
||||
require.NoError(t, db.DB().Omit(clause.Associations).Create(wh).Error)
|
||||
|
||||
path := statsEntrypoint(t, db, wh.ID, true)
|
||||
statsEntrypoint(t, db, wh.ID, false)
|
||||
|
||||
first := seedConfiguredTarget(
|
||||
t, db, wh.ID, database.TargetTypeHTTP,
|
||||
`{"url":"`+replayTargetURL+`"}`,
|
||||
)
|
||||
second := seedTarget(t, db, wh.ID, database.TargetTypeLog)
|
||||
inactive := seedTarget(t, db, wh.ID, database.TargetTypeLog)
|
||||
require.NoError(t, db.DB().Model(inactive).
|
||||
Update("active", false).Error)
|
||||
|
||||
// Three events, each delivered to the two active targets.
|
||||
for range 3 {
|
||||
statsReceive(t, h, path)
|
||||
}
|
||||
|
||||
webhookDB, err := dbMgr.GetDB(wh.ID)
|
||||
require.NoError(t, err)
|
||||
|
||||
events := listEvents(t, webhookDB)
|
||||
require.Len(t, events, 3)
|
||||
|
||||
oldest, middle, newest := events[0], events[1], events[2]
|
||||
now := time.Now()
|
||||
|
||||
// The oldest is past the webhook's one day of retention.
|
||||
statsAge(t, webhookDB, oldest.ID, now.Add(-50*time.Hour))
|
||||
statsAge(t, webhookDB, middle.ID, now.Add(-6*time.Hour))
|
||||
|
||||
oldestFailure := statsDelivery(t, webhookDB, oldest.ID, first.ID)
|
||||
statsFinish(t, webhookDB, oldestFailure,
|
||||
database.DeliveryStatusFailed, now.Add(-49*time.Hour))
|
||||
statsFinish(t, webhookDB,
|
||||
statsDelivery(t, webhookDB, oldest.ID, second.ID),
|
||||
database.DeliveryStatusDelivered, now.Add(-49*time.Hour))
|
||||
statsFinish(t, webhookDB,
|
||||
statsDelivery(t, webhookDB, middle.ID, first.ID),
|
||||
database.DeliveryStatusFailed, now.Add(-5*time.Hour))
|
||||
statsFinish(t, webhookDB,
|
||||
statsDelivery(t, webhookDB, middle.ID, second.ID),
|
||||
database.DeliveryStatusFailed, now.Add(-time.Minute))
|
||||
statsFinish(t, webhookDB,
|
||||
statsDelivery(t, webhookDB, newest.ID, first.ID),
|
||||
database.DeliveryStatusDelivered, now.Add(-2*time.Minute))
|
||||
|
||||
// A replay adds a delivery, still pending, to the oldest event.
|
||||
require.Equal(t, http.StatusSeeOther,
|
||||
postReplay(t, h, sess, wh.ID, oldestFailure).Code)
|
||||
|
||||
stats := h.WebhookStatsForTest(wh.ID)
|
||||
require.NotNil(t, stats)
|
||||
|
||||
assert.Equal(t, 2, stats.Entrypoints)
|
||||
assert.Equal(t, 1, stats.ActiveEntrypoints)
|
||||
assert.Equal(t, 3, stats.Targets)
|
||||
assert.Equal(t, 2, stats.ActiveTargets)
|
||||
assertStatsTotals(t, stats.Totals, [3]int64{3, 7, 3}, [3]int64{3, 7, 3})
|
||||
assert.Equal(t, int64(2), stats.InProgress)
|
||||
require.NotNil(t, stats.LastEventAt)
|
||||
assert.True(t, newest.CreatedAt.Equal(*stats.LastEventAt))
|
||||
assert.Equal(t, handlers.RecentWindow{
|
||||
Events: 1, Delivered: 1, Failed: 1,
|
||||
}, stats.Last10Minutes)
|
||||
assert.Equal(t, handlers.RecentWindow{
|
||||
Events: 2, Delivered: 1, Failed: 2,
|
||||
}, stats.Last24Hours)
|
||||
assert.Equal(t, "50.0%", stats.Last10Minutes.FailurePercent())
|
||||
assert.Equal(t, "66.7%", stats.Last24Hours.FailurePercent())
|
||||
|
||||
// Retention removes the oldest event with its three deliveries,
|
||||
// one of them failed and one the pending replay.
|
||||
statsPrune(t, db, dbMgr, log, webhookDB)
|
||||
|
||||
after := h.WebhookStatsForTest(wh.ID)
|
||||
require.NotNil(t, after)
|
||||
|
||||
assertStatsTotals(t, after.Totals, [3]int64{3, 7, 3}, [3]int64{2, 4, 2})
|
||||
assert.Equal(t, int64(1), after.InProgress)
|
||||
assert.Equal(t, stats.LastEventAt, after.LastEventAt)
|
||||
assert.Equal(t, stats.Last10Minutes, after.Last10Minutes)
|
||||
assert.Equal(t, stats.Last24Hours, after.Last24Hours)
|
||||
|
||||
body := renderSourceDetailPage(t, h, sess, wh.ID)
|
||||
assert.Contains(t, body, "Statistics")
|
||||
assert.Contains(t, body, "Within retention (1 day)")
|
||||
assert.Contains(t, body, "50.0%")
|
||||
assert.Contains(t, body, "66.7%")
|
||||
}
|
||||
|
||||
// TestWebhookStats_WebhookWithNoEvents covers a webhook whose event
|
||||
// database has never been opened: every count is zero, the
|
||||
// percentages are a dash, and showing the page does not create the
|
||||
// database.
|
||||
func TestWebhookStats_WebhookWithNoEvents(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
var (
|
||||
h *handlers.Handlers
|
||||
sess *session.Session
|
||||
db *database.Database
|
||||
dbMgr *database.WebhookDBManager
|
||||
)
|
||||
|
||||
app := newTestApp(t, &h, &sess, &db, &dbMgr)
|
||||
app.RequireStart()
|
||||
|
||||
t.Cleanup(app.RequireStop)
|
||||
|
||||
wh := seedWebhook(t, db)
|
||||
|
||||
assert.Equal(t, &handlers.WebhookStats{}, h.WebhookStatsForTest(wh.ID))
|
||||
assert.Equal(t, "—", handlers.RecentWindow{}.FailurePercent())
|
||||
|
||||
body := renderSourceDetailPage(t, h, sess, wh.ID)
|
||||
assert.Contains(t, body, "Statistics")
|
||||
assert.False(t, dbMgr.DBExists(wh.ID))
|
||||
}
|
||||
|
||||
// TestRecentWindow_FailurePercent pins the percentage: failed
|
||||
// deliveries out of all that finished in the window.
|
||||
func TestRecentWindow_FailurePercent(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
tests := []struct {
|
||||
window handlers.RecentWindow
|
||||
want string
|
||||
}{
|
||||
{handlers.RecentWindow{}, "—"},
|
||||
{handlers.RecentWindow{Events: 4}, "—"},
|
||||
{handlers.RecentWindow{Delivered: 3, Failed: 1}, "25.0%"},
|
||||
{handlers.RecentWindow{Failed: 2}, "100.0%"},
|
||||
{handlers.RecentWindow{Delivered: 2}, "0.0%"},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
assert.Equal(t, tt.want, tt.window.FailurePercent(), tt.window)
|
||||
}
|
||||
}
|
||||
@@ -24,6 +24,8 @@
|
||||
</div>
|
||||
</div>
|
||||
|
||||
{{template "webhook_stats" .}}
|
||||
|
||||
<div class="grid grid-cols-1 lg:grid-cols-2 gap-6">
|
||||
<!-- Entrypoints -->
|
||||
<div class="card">
|
||||
|
||||
@@ -0,0 +1,83 @@
|
||||
{{define "webhook_stats"}}
|
||||
<!-- Statistics pane at the top of the webhook page. -->
|
||||
<div class="card mb-6">
|
||||
<div class="p-4 border-b border-gray-200">
|
||||
<h2 class="text-lg font-medium text-gray-900">Statistics</h2>
|
||||
</div>
|
||||
{{with .Stats}}
|
||||
<div class="p-4 flex flex-wrap gap-6 text-sm border-b border-gray-200">
|
||||
<div>
|
||||
<span class="text-gray-500">Entrypoints</span>
|
||||
<span class="font-medium text-gray-900">{{.Entrypoints}}</span>
|
||||
<span class="text-gray-500">({{.ActiveEntrypoints}} active)</span>
|
||||
</div>
|
||||
<div>
|
||||
<span class="text-gray-500">Targets</span>
|
||||
<span class="font-medium text-gray-900">{{.Targets}}</span>
|
||||
<span class="text-gray-500">({{.ActiveTargets}} active)</span>
|
||||
</div>
|
||||
<div>
|
||||
<span class="text-gray-500">Deliveries in progress</span>
|
||||
<span class="font-medium text-gray-900">{{.InProgress}}</span>
|
||||
</div>
|
||||
<div>
|
||||
<span class="text-gray-500">Last event</span>
|
||||
<span class="font-medium text-gray-900">{{with .LastEventAt}}{{.Format "2006-01-02 15:04:05 UTC"}}{{else}}none{{end}}</span>
|
||||
</div>
|
||||
</div>
|
||||
<div class="p-4 grid grid-cols-1 lg:grid-cols-2 gap-6 text-sm">
|
||||
<div>
|
||||
<div class="flex py-2 border-b border-gray-200 text-xs text-gray-500 uppercase tracking-wide">
|
||||
<span class="flex-1"></span>
|
||||
<span class="w-32 text-center">Lifetime</span>
|
||||
<span class="w-32 text-center">Within retention ({{$.Webhook.RetentionLabel}})</span>
|
||||
</div>
|
||||
<div class="divide-y divide-gray-100">
|
||||
<div class="flex py-2">
|
||||
<span class="flex-1 text-gray-600">Events</span>
|
||||
<span class="w-32 text-center text-gray-900">{{.Totals.Events}}</span>
|
||||
<span class="w-32 text-center text-gray-900">{{.Totals.EventsWithinRetention}}</span>
|
||||
</div>
|
||||
<div class="flex py-2">
|
||||
<span class="flex-1 text-gray-600">Deliveries</span>
|
||||
<span class="w-32 text-center text-gray-900">{{.Totals.Deliveries}}</span>
|
||||
<span class="w-32 text-center text-gray-900">{{.Totals.DeliveriesWithinRetention}}</span>
|
||||
</div>
|
||||
<div class="flex py-2">
|
||||
<span class="flex-1 text-gray-600">Failures</span>
|
||||
<span class="w-32 text-center text-gray-900">{{.Totals.Failures}}</span>
|
||||
<span class="w-32 text-center text-gray-900">{{.Totals.FailuresWithinRetention}}</span>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
<div>
|
||||
<div class="flex py-2 border-b border-gray-200 text-xs text-gray-500 uppercase tracking-wide">
|
||||
<span class="flex-1"></span>
|
||||
<span class="w-32 text-center">Last 10 minutes</span>
|
||||
<span class="w-32 text-center">Last 24 hours</span>
|
||||
</div>
|
||||
<div class="divide-y divide-gray-100">
|
||||
<div class="flex py-2">
|
||||
<span class="flex-1 text-gray-600">Events</span>
|
||||
<span class="w-32 text-center text-gray-900">{{.Last10Minutes.Events}}</span>
|
||||
<span class="w-32 text-center text-gray-900">{{.Last24Hours.Events}}</span>
|
||||
</div>
|
||||
<div class="flex py-2">
|
||||
<span class="flex-1 text-gray-600">Failures</span>
|
||||
<span class="w-32 text-center text-gray-900">{{.Last10Minutes.Failed}}</span>
|
||||
<span class="w-32 text-center text-gray-900">{{.Last24Hours.Failed}}</span>
|
||||
</div>
|
||||
<div class="flex py-2">
|
||||
<span class="flex-1 text-gray-600">Failure percentage</span>
|
||||
<span class="w-32 text-center text-gray-900">{{.Last10Minutes.FailurePercent}}</span>
|
||||
<span class="w-32 text-center text-gray-900">{{.Last24Hours.FailurePercent}}</span>
|
||||
</div>
|
||||
</div>
|
||||
<p class="mt-2 text-xs text-gray-500">Failure percentage is the failed deliveries out of all deliveries that finished in the window. Deliveries still pending or retrying are not counted.</p>
|
||||
</div>
|
||||
</div>
|
||||
{{else}}
|
||||
<div class="p-4 text-sm text-gray-500">The statistics could not be read.</div>
|
||||
{{end}}
|
||||
</div>
|
||||
{{end}}
|
||||
Reference in New Issue
Block a user