Skip to content

Commit 6bd2c30

Browse files
AlCutterAlCutter
committed
Enable mirror hammer to use an existing log as a source.
1 parent bddd697 commit 6bd2c30

3 files changed

Lines changed: 194 additions & 79 deletions

File tree

‎internal/mirror/hammer/README.md‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,16 @@ For real load-testing applications, especially headless runs as part of a CI pip
1515

1616
## Usage
1717

18+
There are two modes in which the hammer can be run, in both modes the hammer continuously tracks a source log and replicates its data to one or more target mirrors.
19+
The difference between the two modes is where the data for the source log originates:
20+
21+
1. **Standalone log**: The hammer creates and continuously populates its own local POSIX log.
22+
2. **Replication**: The hammer reads from an existing log.
23+
24+
In both cases the hammer uses the data from the source log as the data to send to the mirror server.
25+
26+
### Mode 1: Standalone Hammer Log
27+
1828
First, store the hammer log test private key:
1929

2030
```shell
@@ -53,3 +63,19 @@ go run ./internal/mirror/hammer \
5363
--leaf_write_goal=2500 \
5464
--show_ui=false
5565
```
66+
67+
### Mode 2: Replication
68+
69+
Determine the source log's public key and URL, and ensure that the target mirror is configured to accept data from this log.
70+
71+
Write the source log's `vkey` to a file, e.g. `/tmp/source-log.pub`.
72+
73+
Then start the hammer in replication mode, e.g.:
74+
75+
```shell
76+
go run ./internal/mirror/hammer \
77+
--mirror_url=http://localhost:8080 \
78+
--source_log_url=https://example.com/source-log \
79+
--source_log_pubkey_path=/tmp/source-log.pub \
80+
--show_ui=false
81+
```

‎internal/mirror/hammer/hammer.go‎

Lines changed: 118 additions & 43 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,8 @@ import (
3737
"github.com/transparency-dev/tessera/storage/posix"
3838
"golang.org/x/net/http2"
3939

40+
sdbNote "golang.org/x/mod/sumdb/note"
41+
4042
"log/slog"
4143
)
4244

@@ -47,6 +49,9 @@ func init() {
4749
var (
4850
mirrorURL multiStringFlag
4951

52+
sourceLogURL = flag.String("source_log_url", "", "URL of the log to mirror, or empty if using a locally created log.")
53+
sourceLogPubKeyPath = flag.String("source_log_pubkey_path", "", "Path to the public key of the log to mirror. Required if --source_log_url is set.")
54+
5055
storageDir = flag.String("storage_dir", "", "Root directory to store log data")
5156
logPrivKey = flag.String("log_private_key", "", "Location of private key file")
5257

@@ -71,6 +76,12 @@ var (
7176
hc *http.Client
7277
)
7378

79+
type logReader interface {
80+
ReadCheckpoint(ctx context.Context) ([]byte, error)
81+
ReadEntryBundle(ctx context.Context, index uint64, p uint8) ([]byte, error)
82+
ReadTile(ctx context.Context, level, index uint64, p uint8) ([]byte, error)
83+
}
84+
7485
func main() {
7586
flag.Parse()
7687
slog.SetDefault(slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.Level(*slogLevel)})))
@@ -102,42 +113,15 @@ func main() {
102113

103114
ctx, cancel := context.WithCancel(context.Background())
104115

105-
// Gather the info needed for reading/writing checkpoints.
106-
s := getSignerOrDie(ctx)
107-
108-
// Create the Tessera POSIX storage, using the directory from the --storage_dir flag.
109-
driver, err := posix.New(ctx, posix.Config{Path: *storageDir})
110-
if err != nil {
111-
slog.ErrorContext(ctx, "Failed to construct storage", slog.Any("error", err))
112-
os.Exit(1)
113-
}
114-
115-
appender, shutdown, logReader, err := tessera.NewAppender(ctx, driver, tessera.NewAppendOptions().
116-
WithCheckpointSigner(s).
117-
WithCheckpointInterval(time.Second).
118-
WithCheckpointRepublishInterval(time.Minute).
119-
WithBatching(tessera.DefaultBatchMaxSize, time.Second))
120-
if err != nil {
121-
slog.ErrorContext(ctx, "Failed to create new appender", slog.Any("error", err))
122-
os.Exit(1)
123-
}
116+
appender, logReader, shutdown, verifier := logFromFlags(ctx)
124117
defer func() {
125118
if err := shutdown(ctx); err != nil {
126119
slog.ErrorContext(ctx, "Failed to shutdown appender", slog.Any("error", err))
127120
}
128121
}()
129122

130-
leafWriter := func(ctx context.Context, newLeaf []byte) (uint64, error) {
131-
idx, err := appender.Add(ctx, tessera.NewEntry(newLeaf))()
132-
if err != nil {
133-
return 0, fmt.Errorf("failed to add leaf: %v", err)
134-
}
135-
return idx.Index, nil
136-
}
137-
138-
var cpRaw []byte
139123
cons := client.UnilateralConsensus(logReader.ReadCheckpoint)
140-
tracker, err := client.NewLogStateTracker(ctx, logReader.ReadTile, cpRaw, s.Verifier(), s.Verifier().Name(), cons)
124+
tracker, err := client.NewLogStateTracker(ctx, logReader.ReadTile, nil, verifier, verifier.Name(), cons)
141125
if err != nil {
142126
slog.ErrorContext(ctx, "Failed to create LogStateTracker", slog.Any("error", err))
143127
os.Exit(1)
@@ -160,7 +144,7 @@ func main() {
160144
mOpts := mirror.NewOptions().
161145
WithMirrorURL(mURL).
162146
WithHTTPClient(hc).
163-
WithLogOrigin(s.Verifier().Name()).
147+
WithLogOrigin(verifier.Name()).
164148
WithTileFetcher(logReader.ReadTile).
165149
WithBundleFetcher(logReader.ReadEntryBundle).
166150
WithMirrorCheckpointFetcher(func(ctx context.Context) ([]byte, error) {
@@ -175,18 +159,13 @@ func main() {
175159
go runMirrorSync(ctx, tracker, mc, mURL)
176160
}
177161

178-
ha := loadtest.NewHammerAnalyser(func() uint64 { return tracker.Latest().Size })
179-
ha.Run(ctx)
180-
181-
gen := newLeafGenerator(tracker.Latest().Size, *leafMinSize, *dupChance)
182-
opts := loadtest.HammerOpts{
183-
MaxReadOpsPerSecond: 0,
184-
MaxWriteOpsPerSecond: *maxWriteOpsPerSecond,
185-
NumReadersRandom: 0,
186-
NumReadersFull: 0,
187-
NumWriters: *numWriters,
162+
var hammer *loadtest.Hammer
163+
var ha *loadtest.HammerAnalyser
164+
if appender != nil {
165+
hammer, ha = newHammer(ctx, appender, logReader, tracker)
166+
ha.Run(ctx)
167+
hammer.Run(ctx)
188168
}
189-
hammer := loadtest.NewHammer(tracker, logReader.ReadEntryBundle, leafWriter, gen, ha.SeqLeafChan, ha.ErrChan, opts)
190169

191170
exitCode := 0
192171
if *leafWriteGoal > 0 {
@@ -226,10 +205,9 @@ func main() {
226205
}
227206
}()
228207
}
229-
hammer.Run(ctx)
230208

231209
if *showUI {
232-
c := loadtest.NewController(hammer, ha)
210+
c := loadtest.NewController(hammer, ha, tracker)
233211
c.Run(ctx)
234212
} else {
235213
<-ctx.Done()
@@ -350,3 +328,100 @@ func runMirrorSync(ctx context.Context, tracker *client.LogStateTracker, mClient
350328
}
351329
}
352330
}
331+
332+
func logFromFlags(ctx context.Context) (*tessera.Appender, logReader, func(context.Context) error, sdbNote.Verifier) {
333+
if *sourceLogURL != "" {
334+
if *sourceLogPubKeyPath == "" {
335+
slog.ErrorContext(ctx, "Source log URL provided but no public key path.")
336+
os.Exit(1)
337+
}
338+
lr, shutdown, v, err := newRemoteLog(ctx, *sourceLogURL, *sourceLogPubKeyPath)
339+
if err != nil {
340+
slog.ErrorContext(ctx, "Failed to create remote log", slog.Any("error", err))
341+
os.Exit(1)
342+
}
343+
return nil, lr, shutdown, v
344+
}
345+
346+
if *storageDir == "" {
347+
slog.ErrorContext(ctx, "No storage directory provided.")
348+
os.Exit(1)
349+
}
350+
app, lr, shutdown, v, err := newLocalLog(ctx, *storageDir, getSignerOrDie(ctx))
351+
if err != nil {
352+
slog.ErrorContext(ctx, "Failed to create local log", slog.Any("error", err))
353+
os.Exit(1)
354+
}
355+
return app, lr, shutdown, v
356+
}
357+
358+
// newLocalLog creates a new local POSIX log which can be used by a hammer to hammer the mirror.
359+
//
360+
// Returns an Appender for adding leaves to the log, a LogReader that reads from the local log,
361+
func newLocalLog(ctx context.Context, storageDir string, s note.Signer) (*tessera.Appender, logReader, func(context.Context) error, sdbNote.Verifier, error) {
362+
// Create the Tessera POSIX storage, using the provided directory.
363+
driver, err := posix.New(ctx, posix.Config{Path: storageDir})
364+
if err != nil {
365+
return nil, nil, nil, nil, fmt.Errorf("failed to construct storage: %w", err)
366+
}
367+
368+
appender, shutdown, logReader, err := tessera.NewAppender(ctx, driver, tessera.NewAppendOptions().
369+
WithCheckpointSigner(s).
370+
WithCheckpointInterval(time.Second).
371+
WithCheckpointRepublishInterval(time.Minute).
372+
WithBatching(tessera.DefaultBatchMaxSize, time.Second))
373+
if err != nil {
374+
return nil, nil, nil, nil, err
375+
}
376+
return appender, logReader, shutdown, s.Verifier(), nil
377+
}
378+
379+
// newRemoteLog creates a logReader that reads from a remote log, and a shutdown func.
380+
func newRemoteLog(_ context.Context, logURL string, pubKeyPath string) (logReader, func(context.Context) error, sdbNote.Verifier, error) {
381+
pubKRaw, err := getKeyFile(pubKeyPath)
382+
if err != nil {
383+
return nil, nil, nil, fmt.Errorf("failed to read public key %q: %v", pubKeyPath, err)
384+
}
385+
386+
pubKey, err := note.NewVerifier(pubKRaw)
387+
if err != nil {
388+
return nil, nil, nil, fmt.Errorf("failed to parse public key %q: %v", pubKeyPath, err)
389+
}
390+
391+
var lr logReader
392+
switch logURL := strings.ToLower(logURL); {
393+
case strings.HasPrefix(logURL, "http://"), strings.HasPrefix(logURL, "https://"):
394+
u, err := url.Parse(logURL)
395+
if err != nil {
396+
return nil, nil, nil, fmt.Errorf("failed to parse log URL %q: %v", logURL, err)
397+
}
398+
lr, err = client.NewHTTPFetcher(u, hc)
399+
default:
400+
lr = client.FileFetcher{Root: logURL}
401+
}
402+
403+
return lr, func(context.Context) error { return nil }, pubKey, nil
404+
}
405+
406+
func newHammer(_ context.Context, appender *tessera.Appender, logReader logReader, tracker *client.LogStateTracker) (*loadtest.Hammer, *loadtest.HammerAnalyser) {
407+
leafWriter := func(ctx context.Context, newLeaf []byte) (uint64, error) {
408+
idx, err := appender.Add(ctx, tessera.NewEntry(newLeaf))()
409+
if err != nil {
410+
return 0, fmt.Errorf("failed to add leaf: %v", err)
411+
}
412+
return idx.Index, nil
413+
}
414+
ha := loadtest.NewHammerAnalyser(func() uint64 { return tracker.Latest().Size })
415+
416+
gen := newLeafGenerator(tracker.Latest().Size, *leafMinSize, *dupChance)
417+
opts := loadtest.HammerOpts{
418+
MaxReadOpsPerSecond: 0,
419+
MaxWriteOpsPerSecond: *maxWriteOpsPerSecond,
420+
NumReadersRandom: 0,
421+
NumReadersFull: 0,
422+
NumWriters: *numWriters,
423+
}
424+
hammer := loadtest.NewHammer(tracker, logReader.ReadEntryBundle, leafWriter, gen, ha.SeqLeafChan, ha.ErrChan, opts)
425+
426+
return hammer, ha
427+
}

‎internal/mirror/hammer/loadtest/tui.go‎

Lines changed: 50 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -25,21 +25,24 @@ import (
2525
movingaverage "github.com/RobinUS2/golang-moving-average"
2626
"github.com/gdamore/tcell/v2"
2727
"github.com/rivo/tview"
28+
"github.com/transparency-dev/tessera/client"
2829
)
2930

3031
type tuiController struct {
3132
hammer *Hammer
3233
analyser *HammerAnalyser
34+
tracker *client.LogStateTracker
3335
app *tview.Application
3436
statusView *tview.TextView
3537
logView *tview.TextView
3638
helpView *tview.TextView
3739
}
3840

39-
func NewController(h *Hammer, a *HammerAnalyser) *tuiController {
41+
func NewController(h *Hammer, a *HammerAnalyser, t *client.LogStateTracker) *tuiController {
4042
c := tuiController{
4143
hammer: h,
4244
analyser: a,
45+
tracker: t,
4346
app: tview.NewApplication(),
4447
}
4548
grid := tview.NewGrid()
@@ -74,29 +77,31 @@ func (c *tuiController) Run(ctx context.Context) {
7477
go c.updateStatsLoop(ctx, 500*time.Millisecond)
7578

7679
c.app.SetInputCapture(func(event *tcell.EventKey) *tcell.EventKey {
77-
switch event.Rune() {
78-
case '+':
79-
slog.InfoContext(ctx, "Increasing the read operations per second")
80-
c.hammer.readThrottle.Increase()
81-
case '-':
82-
slog.InfoContext(ctx, "Decreasing the read operations per second")
83-
c.hammer.readThrottle.Decrease()
84-
case '>':
85-
slog.InfoContext(ctx, "Increasing the write operations per second")
86-
c.hammer.writeThrottle.Increase()
87-
case '<':
88-
slog.InfoContext(ctx, "Decreasing the write operations per second")
89-
c.hammer.writeThrottle.Decrease()
90-
case 'w':
91-
slog.InfoContext(ctx, "Increasing the number of workers")
92-
c.hammer.randomReaders.Grow(ctx)
93-
c.hammer.fullReaders.Grow(ctx)
94-
c.hammer.writers.Grow(ctx)
95-
case 'W':
96-
slog.InfoContext(ctx, "Decreasing the number of workers")
97-
c.hammer.randomReaders.Shrink(ctx)
98-
c.hammer.fullReaders.Shrink(ctx)
99-
c.hammer.writers.Shrink(ctx)
80+
if c.hammer != nil {
81+
switch event.Rune() {
82+
case '+':
83+
slog.InfoContext(ctx, "Increasing the read operations per second")
84+
c.hammer.readThrottle.Increase()
85+
case '-':
86+
slog.InfoContext(ctx, "Decreasing the read operations per second")
87+
c.hammer.readThrottle.Decrease()
88+
case '>':
89+
slog.InfoContext(ctx, "Increasing the write operations per second")
90+
c.hammer.writeThrottle.Increase()
91+
case '<':
92+
slog.InfoContext(ctx, "Decreasing the write operations per second")
93+
c.hammer.writeThrottle.Decrease()
94+
case 'w':
95+
slog.InfoContext(ctx, "Increasing the number of workers")
96+
c.hammer.randomReaders.Grow(ctx)
97+
c.hammer.fullReaders.Grow(ctx)
98+
c.hammer.writers.Grow(ctx)
99+
case 'W':
100+
slog.InfoContext(ctx, "Decreasing the number of workers")
101+
c.hammer.randomReaders.Shrink(ctx)
102+
c.hammer.fullReaders.Shrink(ctx)
103+
c.hammer.writers.Shrink(ctx)
104+
}
100105
}
101106
return event
102107
})
@@ -114,32 +119,41 @@ func (c *tuiController) updateStatsLoop(ctx context.Context, interval time.Durat
114119
}
115120

116121
ticker := time.NewTicker(interval)
117-
lastSize := c.hammer.tracker.Latest().Size
122+
lastSize := c.tracker.Latest().Size
118123
maSlots := int((30 * time.Second) / interval)
119124
growth := movingaverage.New(maSlots)
120125
for {
121126
select {
122127
case <-ctx.Done():
123128
return
124129
case <-ticker.C:
125-
s := c.hammer.tracker.Latest().Size
130+
s := c.tracker.Latest().Size
126131
growth.Add(float64(s - lastSize))
127132
lastSize = s
128133
qps := growth.Avg() * float64(time.Second/interval)
129-
readWorkersLine := fmt.Sprintf("Read (%d workers): %s",
130-
c.hammer.fullReaders.Size()+c.hammer.randomReaders.Size(),
131-
c.hammer.readThrottle.String())
132-
writeWorkersLine := fmt.Sprintf("Write (%d workers): %s",
133-
c.hammer.writers.Size(),
134-
c.hammer.writeThrottle.String())
134+
135+
var numR, numW int
136+
var rThrottle, wThrottle string
137+
if c.hammer != nil {
138+
numR = c.hammer.fullReaders.Size() + c.hammer.randomReaders.Size()
139+
rThrottle = c.hammer.readThrottle.String()
140+
numW = c.hammer.writers.Size()
141+
wThrottle = c.hammer.writeThrottle.String()
142+
}
143+
readWorkersLine := fmt.Sprintf("Read (%d workers): %s", numR, rThrottle)
144+
writeWorkersLine := fmt.Sprintf("Write (%d workers): %s", numW, wThrottle)
135145
treeSizeLine := fmt.Sprintf("TreeSize: %d (Δ %.0fqps over %ds)",
136146
s,
137147
qps,
138148
time.Duration(maSlots*int(interval))/time.Second)
139-
queueLine := fmt.Sprintf("Time-in-queue: %s",
140-
formatMovingAverage(c.analyser.QueueTime))
141-
integrateLine := fmt.Sprintf("Observed-time-to-integrate: %s",
142-
formatMovingAverage(c.analyser.IntegrationTime))
149+
150+
var qTime, iTime string
151+
if c.analyser != nil {
152+
qTime = formatMovingAverage(c.analyser.QueueTime)
153+
iTime = formatMovingAverage(c.analyser.IntegrationTime)
154+
}
155+
queueLine := fmt.Sprintf("Time-in-queue: %s", qTime)
156+
integrateLine := fmt.Sprintf("Observed-time-to-integrate: %s", iTime)
143157
text := strings.Join([]string{readWorkersLine, writeWorkersLine, treeSizeLine, queueLine, integrateLine}, "\n")
144158
c.statusView.SetText(text)
145159
c.app.Draw()

0 commit comments

Comments
 (0)