bel/pg
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

accept P

Browse Source
This commit is contained in:
Bel LaPointe
2026-03-09 13:24:54 -06:00
parent 82400ca0b2
commit 515feed98b
1 changed files with 33 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
cmd/pg-pulse/main.go
View File

@@ -24,16 +24,28 @@ func run(ctx context.Context) error {
fs := flag.NewFlagSet(os.Args[0], flag.ContinueOnError) fs := flag.NewFlagSet(os.Args[0], flag.ContinueOnError)
c := fs.String("c", "postgresql://pulsegres:pulsegres@localhost:15432", "conn string") c := fs.String("c", "postgresql://pulsegres:pulsegres@localhost:15432", "conn string")
d := fs.Duration("d", time.Second, "interval") d := fs.Duration("d", time.Second, "interval")
p := fs.Int("p", 1, "concurrent goroutines")
if err := fs.Parse(os.Args[1:]); err != nil { if err := fs.Parse(os.Args[1:]); err != nil {
panic(err) panic(err)
} }
for i := 0; i < *p-1; i++ {
go with.PSQL(ctx, *c, func(pg *sql.DB) error {
with.GoEvery(ctx, *d, func() {
if _, err := pg.ExecContext(ctx, `SELECT 1`); err != nil {
log.Println("!", err)
}
})
return ctx.Err()
})
}
return with.PSQL(ctx, *c, func(pg *sql.DB) error { return with.PSQL(ctx, *c, func(pg *sql.DB) error {
log.Println("staging...") log.Println("staging...")
if _, err := pg.ExecContext(ctx, ` if _, err := pg.ExecContext(ctx, `
DROP TABLE IF EXISTS "pg-pulse"; DROP TABLE IF EXISTS "pg-pulse";
CREATE TABLE IF NOT EXISTS "pg-pulse" (k TEXT); CREATE TABLE IF NOT EXISTS "pg-pulse" (k TEXT);
`); err != nil { `); err != nil {
return err return err
} }
@@ -78,8 +90,8 @@ func run(ctx context.Context) error {
with.GoEvery(ctx, *d, func() { with.GoEvery(ctx, *d, func() {
if _, err := pg.ExecContext(ctx, fmt.Sprintf(` if _, err := pg.ExecContext(ctx, fmt.Sprintf(`
INSERT INTO "pg-pulse" (k) SELECT substr(md5(random()::text), 1, 25) INSERT INTO "pg-pulse" (k) SELECT substr(md5(random()::text), 1, 25)
`)); err != nil { `)); err != nil {
pushOK(false) pushOK(false)
log.Println("\n! failed nonzero insert:", err) log.Println("\n! failed nonzero insert:", err)
} else { } else {
@@ -89,22 +101,22 @@ func run(ctx context.Context) error {
ha, ha1 := false, false ha, ha1 := false, false
row := pg.QueryRowContext(ctx, ` row := pg.QueryRowContext(ctx, `
SELECT SELECT
( (
SELECT COUNT(*) > 0 SELECT COUNT(*) > 0
FROM pg_replication_slots FROM pg_replication_slots
WHERE slot_name LIKE 'dpg_%_a_ha__' WHERE slot_name LIKE 'dpg_%_a_ha__'
) AS is_ha, ) AS is_ha,
( (
SELECT COUNT(*) > 0 SELECT COUNT(*) > 0
FROM pg_replication_slots FROM pg_replication_slots
WHERE slot_name LIKE 'dpg_%_a_ha_0' WHERE slot_name LIKE 'dpg_%_a_ha_0'
) AS is_ha1, ) AS is_ha1,
( (
SELECT $1 - COUNT(*) SELECT $1 - COUNT(*)
FROM "pg-pulse" FROM "pg-pulse"
) AS lost_writes ) AS lost_writes
`, n) `, n)
if err := row.Err(); err != nil { if err := row.Err(); err != nil {
log.Println("\n! failed getting ha-stuff:", err) log.Println("\n! failed getting ha-stuff:", err)
} else if err := row.Scan(&ha, &ha1, &lostWrites); err != nil { } else if err := row.Scan(&ha, &ha1, &lostWrites); err != nil {