diff --git a/.github/workflows/validate-release.yml b/.github/workflows/validate-release.yml index 018f82f2a..f5080b5a7 100644 --- a/.github/workflows/validate-release.yml +++ b/.github/workflows/validate-release.yml @@ -67,6 +67,7 @@ jobs: env: PROJECT_ID: honk-fake-project RUNTIME_IMAGE: gcr.io/distroless/static:debug-nonroot + GORELEASER_CONFIG: .goreleaser.ci.yml - name: check binaries run: | diff --git a/.goreleaser.ci.yml b/.goreleaser.ci.yml new file mode 100644 index 000000000..9a8b38101 --- /dev/null +++ b/.goreleaser.ci.yml @@ -0,0 +1,96 @@ +project_name: rekor +version: 2 + +env: + - GO111MODULE=on + - CGO_ENABLED=0 + - DOCKER_CLI_EXPERIMENTAL=enabled + - COSIGN_YES=true + +# Prevents parallel builds from stepping on eachothers toes downloading modules +before: + hooks: + - go mod tidy + - /bin/bash -c 'if [ -n "$(git --no-pager diff --exit-code go.mod go.sum)" ]; then exit 1; fi' +# if running a release we will generate the images in this step +# if running in the CI the CI env va is set by github action runner and we dont run the ko steps +# this is needed because we are generating files that goreleaser was not aware to push to GH project release + - /bin/bash -c 'if [ -z "$CI" ]; then make sign-container-release; fi' + +gomod: + proxy: true + +sboms: + - artifacts: binary + +builds: + - id: rekor-server-linux + binary: rekor-server-linux-{{ .Arch }} + no_unique_dist_dir: true + main: ./cmd/rekor-server + goos: + - linux + goarch: + - amd64 + flags: + - -trimpath + mod_timestamp: '{{ .CommitTimestamp }}' + ldflags: + - "{{ .Env.SERVER_LDFLAGS }}" + + - id: rekor-cli + binary: rekor-cli-{{ .Os }}-{{ .Arch }} + no_unique_dist_dir: true + main: ./cmd/rekor-cli + goos: + - linux + goarch: + - amd64 + flags: + - -trimpath + mod_timestamp: '{{ .CommitTimestamp }}' + ldflags: + - "{{ .Env.CLI_LDFLAGS }}" + +signs: + - id: rekor + signature: "${artifact}.sig" + cmd: cosign + args: ["sign-blob", "--output-signature", "${artifact}.sig", "--key", "gcpkms://projects/{{ .Env.PROJECT_ID }}/locations/{{ .Env.KEY_LOCATION }}/keyRings/{{ .Env.KEY_RING }}/cryptoKeys/{{ .Env.KEY_NAME }}/versions/{{ .Env.KEY_VERSION }}", "${artifact}"] + artifacts: binary + # Keyless + - id: rekor-keyless + signature: "${artifact}-keyless.sig" + certificate: "${artifact}-keyless.pem" + cmd: cosign + args: ["sign-blob", "--output-signature", "${artifact}-keyless.sig", "--output-certificate", "${artifact}-keyless.pem", "${artifact}"] + artifacts: binary + - id: checksum-keyless + signature: "${artifact}-keyless.sig" + certificate: "${artifact}-keyless.pem" + cmd: cosign + args: ["sign-blob", "--output-signature", "${artifact}-keyless.sig", "--output-certificate", "${artifact}-keyless.pem", "${artifact}"] + artifacts: checksum + +archives: + - format: binary + name_template: "{{ .Binary }}" + allow_different_binary_count: true + +checksum: + name_template: "{{ .ProjectName }}_checksums.txt" + +snapshot: + name_template: SNAPSHOT-{{ .ShortCommit }} + +release: + prerelease: allow # remove this when we start publishing non-prerelease or set to auto + draft: true # allow for manual edits + github: + owner: sigstore + name: rekor + footer: | + ### Thanks for all contributors! + + extra_files: + - glob: "./rekor*.yaml" diff --git a/cmd/backfill-index/main.go b/cmd/backfill-index/main.go index 6169b2a57..b685e06f1 100644 --- a/cmd/backfill-index/main.go +++ b/cmd/backfill-index/main.go @@ -19,10 +19,18 @@ we need to backfill missing entries into the database for the search API. It can also be used to populate an index storage backend from scratch. + The tool supports automatic checkpointing for both Redis and MySQL backends to allow + resuming interrupted backfills and to maintain progress across scheduled runs. + Use --checkpoint-interval to control how frequently checkpoints are saved + (e.g., --checkpoint-interval 100 saves every 100 entries). On subsequent runs, + the tool automatically continues from the last checkpoint unless --reset-checkpoint is set. + Checkpoints persist indefinitely for scheduled jobs. + To run: go run cmd/backfill-index/main.go --rekor-address
\ - --hostname --port --concurrency \ - --start --end [--dry-run] + --redis-hostname --redis-port --concurrency \ + --start --end \ + [--checkpoint-interval ] [--reset-checkpoint] [--checkpoint-key ] [--dry-run] */ package main @@ -31,7 +39,9 @@ import ( "bytes" "context" "crypto/tls" + "database/sql" "encoding/base64" + "encoding/json" "errors" "flag" "fmt" @@ -79,7 +89,15 @@ const ( PRIMARY KEY(PK), UNIQUE(EntryKey, EntryUUID) )` - maxInsertErrors = 5 + mysqlCheckpointTableStmt = `CREATE TABLE IF NOT EXISTS BackfillCheckpoint ( + CheckpointKey VARCHAR(255) NOT NULL, + LastCompletedIndex INT NOT NULL, + LastUpdated TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY(CheckpointKey) + )` + mysqlCheckpointSaveStmt = `REPLACE INTO BackfillCheckpoint (CheckpointKey, LastCompletedIndex) VALUES (:key, :lastIndex)` + mysqlCheckpointLoadStmt = `SELECT LastCompletedIndex, LastUpdated FROM BackfillCheckpoint WHERE CheckpointKey = ?` + mysqlCheckpointDeleteStmt = `DELETE FROM BackfillCheckpoint WHERE CheckpointKey = ?` ) type provider int @@ -92,6 +110,10 @@ const ( type indexClient interface { idempotentAddToIndex(ctx context.Context, key, value string) error + saveCheckpoint(ctx context.Context, checkpointKey string, state checkpointState) error + loadCheckpoint(ctx context.Context, checkpointKey string) (*checkpointState, error) + deleteCheckpoint(ctx context.Context, checkpointKey string) error + supportsCheckpointing() bool } type redisClient struct { @@ -102,6 +124,15 @@ type mysqlClient struct { client *sqlx.DB } +type checkpointState struct { + LastCompletedIndex int `json:"last_completed_index"` + LastUpdated time.Time `json:"last_updated"` +} + +type checkpointUpdate struct { + index int +} + type headers map[string][]string func (h *headers) String() string { @@ -139,6 +170,9 @@ var ( versionFlag = flag.Bool("version", false, "Print the current version of Backfill MySQL") concurrency = flag.Int("concurrency", 1, "Number of workers to use for backfill") dryRun = flag.Bool("dry-run", false, "Dry run - don't actually insert into MySQL") + checkpointInterval = flag.Int("checkpoint-interval", 100, "Save checkpoint every N entries (0 to disable)") + resetCheckpoint = flag.Bool("reset-checkpoint", false, "Clear checkpoint and start from --start value") + checkpointKey = flag.String("checkpoint-key", "", "Custom Redis key for checkpoint (default: \"default\")") ) func main() { @@ -181,6 +215,9 @@ func main() { if *rekorAddress == "" { log.Fatal("rekor-address must be set") } + if *checkpointInterval < 0 { + log.Fatal("checkpoint-interval must be >= 0") + } log.Printf("running backfill index Version: %s GitCommit: %s BuildDate: %s", versionInfo.GitVersion, versionInfo.GitCommit, versionInfo.BuildDate) @@ -233,6 +270,9 @@ func getIndexClient(backend provider) (indexClient, error) { if _, err = dbClient.Exec(mysqlCreateTableStmt); err != nil { return nil, err } + if _, err = dbClient.Exec(mysqlCheckpointTableStmt); err != nil { + return nil, err + } return &mysqlClient{client: dbClient}, nil default: return nil, fmt.Errorf("could not create client for unexpected provider") @@ -245,36 +285,95 @@ func populate(indexClient indexClient, rekorClient *rekorclient.Rekor) (err erro group, ctx := errgroup.WithContext(ctx) group.SetLimit(*concurrency) - type result struct { - index int - parseErrs []error - insertErrs []error - } - var resultChan = make(chan result) - parseErrs := make([]int, 0) - insertErrs := make([]int, 0) - runningInsertErrs := 0 + var checkpointKeyName string + var checkpointChan chan checkpointUpdate + var checkpointDone chan struct{} + var useCheckpointing = *checkpointInterval > 0 && indexClient.supportsCheckpointing() + originalStartIndex := *startIndex - go func() { - for r := range resultChan { - if len(r.parseErrs) > 0 { - parseErrs = append(parseErrs, r.index) + if useCheckpointing { + checkpointKeyName = *checkpointKey + if checkpointKeyName == "" { + checkpointKeyName = "default" + } + + if *resetCheckpoint { + if err := indexClient.deleteCheckpoint(ctx, checkpointKeyName); err != nil { + log.Printf("Warning: failed to delete checkpoint: %v", err) + } else { + log.Println("Checkpoint reset - starting fresh") } - if len(r.insertErrs) > 0 { - insertErrs = append(insertErrs, r.index) + } else { + checkpoint, err := indexClient.loadCheckpoint(ctx, checkpointKeyName) + switch { + case err != nil: + log.Printf("Warning: failed to load checkpoint: %v", err) + case checkpoint != nil: + log.Printf("Resuming from checkpoint: last completed index %d", checkpoint.LastCompletedIndex) + *startIndex = checkpoint.LastCompletedIndex + 1 + if *startIndex > *endIndex { + log.Printf("Checkpoint at %d is already past end index %d - nothing to do", checkpoint.LastCompletedIndex, *endIndex) + return nil + } + log.Printf("Processing entries from index %d to %d", *startIndex, *endIndex) + default: + log.Printf("No checkpoint found - starting fresh from index %d to %d", *startIndex, *endIndex) } } - }() - defer func() { - close(resultChan) - if len(parseErrs) > 0 { - err = fmt.Errorf("failed to parse %d entries: %v", len(parseErrs), parseErrs) - } - if len(insertErrs) > 0 { - err = fmt.Errorf("failed to insert/remove %d entries: %v", len(insertErrs), insertErrs) - } - }() + checkpointChan = make(chan checkpointUpdate, *concurrency*2) + checkpointDone = make(chan struct{}) + + go func() { + defer close(checkpointDone) + + completedIndices := make(map[int]bool) + highestCompleted := *startIndex - 1 + saveCounter := 0 + + for update := range checkpointChan { + completedIndices[update.index] = true + + // Find highest contiguous completed index + if update.index == highestCompleted+1 { + for i := update.index; completedIndices[i]; i++ { + highestCompleted = i + delete(completedIndices, i) + } + } + + saveCounter++ + + if saveCounter >= *checkpointInterval { + state := checkpointState{ + LastCompletedIndex: highestCompleted, + LastUpdated: time.Now(), + } + saveCtx, saveCancel := context.WithTimeout(context.Background(), 5*time.Second) + if err := indexClient.saveCheckpoint(saveCtx, checkpointKeyName, state); err != nil { + // If save fails we will try to catch it next time + log.Printf("Warning: failed to save checkpoint: %v", err) + } else { + log.Printf("Checkpoint saved: last completed index %d", highestCompleted) + } + saveCancel() + saveCounter = 0 + } + } + + state := checkpointState{ + LastCompletedIndex: highestCompleted, + LastUpdated: time.Now(), + } + saveCtx, saveCancel := context.WithTimeout(context.Background(), 5*time.Second) + if err := indexClient.saveCheckpoint(saveCtx, checkpointKeyName, state); err != nil { + log.Printf("Warning: failed to save final checkpoint: %v", err) + } else { + log.Printf("Final checkpoint saved: last completed index %d", highestCompleted) + } + saveCancel() + }() + } for i := *startIndex; i <= *endIndex; i++ { index := i // capture loop variable for closure @@ -287,69 +386,61 @@ func populate(indexClient indexClient, rekorClient *rekorclient.Rekor) (err erro if errors.Is(err, context.Canceled) { return nil } - return fmt.Errorf("retrieving log uuid by index: %v", err) + return fmt.Errorf("retrieving log uuid by index %d: %w", index, err) } - var parseErrs []error - var insertErrs []error - defer func() { - if len(insertErrs) != 0 || len(parseErrs) != 0 { - fmt.Printf("Errors with log index %d:\n", index) - for _, e := range insertErrs { - fmt.Println(e) - } - for _, e := range parseErrs { - fmt.Println(e) - } - } else { - fmt.Printf("Completed log index %d\n", index) - } - }() + for uuid, entry := range resp.Payload { // uuid is the global UUID - tree ID and entry UUID e, _, _, err := unmarshalEntryImpl(entry.Body.(string)) if err != nil { - parseErrs = append(parseErrs, fmt.Errorf("error unmarshalling entry for %s: %w", uuid, err)) - continue + return fmt.Errorf("error unmarshalling entry at index %d for %s: %w", index, uuid, err) } keys, err := e.IndexKeys() if err != nil { - parseErrs = append(parseErrs, fmt.Errorf("error building index keys for %s: %w", uuid, err)) - continue + return fmt.Errorf("error building index keys at index %d for %s: %w", index, uuid, err) } for _, key := range keys { if err := indexClient.idempotentAddToIndex(ctx, key, uuid); err != nil { if errors.Is(err, context.Canceled) { return nil } - insertErrs = append(insertErrs, fmt.Errorf("error inserting UUID %s with key %s: %w", uuid, key, err)) - runningInsertErrs++ - if runningInsertErrs > maxInsertErrors { - resultChan <- result{ - index: index, - parseErrs: parseErrs, - insertErrs: insertErrs, - } - return fmt.Errorf("too many insertion errors") - } - continue + return fmt.Errorf("error inserting UUID %s with key %s at index %d: %w", uuid, key, index, err) } fmt.Printf("Uploaded entry %s, index %d, key %s\n", uuid, index, key) } } - resultChan <- result{ - index: index, - parseErrs: parseErrs, - insertErrs: insertErrs, + + if useCheckpointing { + select { + case checkpointChan <- checkpointUpdate{index: index}: + case <-ctx.Done(): + return nil + } } + fmt.Printf("Completed log index %d\n", index) return nil }) } err = group.Wait() + + if useCheckpointing { + close(checkpointChan) + <-checkpointDone + } + if err != nil { - return fmt.Errorf("error running backfill: %v", err) + if useCheckpointing { + log.Printf("Backfill failed with error (checkpoint saved, resume from last checkpoint on next run): %v", err) + } + return fmt.Errorf("error running backfill: %w", err) + } + + if useCheckpointing { + log.Printf("Backfill complete: processed %d entries, checkpoint persists for next run", *endIndex-originalStartIndex+1) + } else { + fmt.Println("Backfill complete") } - fmt.Println("Backfill complete") return nil } @@ -386,6 +477,61 @@ func (c *redisClient) idempotentAddToIndex(ctx context.Context, key, value strin return err } +// formatRedisCheckpointKey generates a Redis-specific key format for checkpoint storage +func (c *redisClient) formatRedisCheckpointKey(checkpointKey string) string { + return fmt.Sprintf("backfill/checkpoint/%s", checkpointKey) +} + +func (c *redisClient) saveCheckpoint(ctx context.Context, checkpointKey string, state checkpointState) error { + if *dryRun { + return nil + } + redisKey := c.formatRedisCheckpointKey(checkpointKey) + data, err := json.Marshal(state) + if err != nil { + return fmt.Errorf("marshaling checkpoint state: %w", err) + } + err = c.client.Set(ctx, redisKey, data, 0).Err() + if err != nil { + return fmt.Errorf("saving checkpoint to Redis: %w", err) + } + return nil +} + +func (c *redisClient) loadCheckpoint(ctx context.Context, checkpointKey string) (*checkpointState, error) { + redisKey := c.formatRedisCheckpointKey(checkpointKey) + data, err := c.client.Get(ctx, redisKey).Result() + if err == redis.Nil { + return nil, nil + } + if err != nil { + return nil, fmt.Errorf("loading checkpoint from Redis: %w", err) + } + + var state checkpointState + if err := json.Unmarshal([]byte(data), &state); err != nil { + return nil, fmt.Errorf("unmarshaling checkpoint state: %w", err) + } + return &state, nil +} + +func (c *redisClient) deleteCheckpoint(ctx context.Context, checkpointKey string) error { + if *dryRun { + return nil + } + redisKey := c.formatRedisCheckpointKey(checkpointKey) + err := c.client.Del(ctx, redisKey).Err() + if err != nil { + return fmt.Errorf("deleting checkpoint from Redis: %w", err) + } + log.Printf("Checkpoint deleted: %s", redisKey) + return nil +} + +func (c *redisClient) supportsCheckpointing() bool { + return true +} + func (c *mysqlClient) idempotentAddToIndex(ctx context.Context, key, value string) error { if *dryRun { return nil @@ -393,3 +539,58 @@ func (c *mysqlClient) idempotentAddToIndex(ctx context.Context, key, value strin _, err := c.client.NamedExecContext(ctx, mysqlWriteStmt, map[string]any{"key": key, "uuid": value}) return err } + +func (c *mysqlClient) saveCheckpoint(ctx context.Context, checkpointKey string, state checkpointState) error { + if *dryRun { + return nil + } + _, err := c.client.NamedExecContext(ctx, mysqlCheckpointSaveStmt, map[string]any{ + "key": checkpointKey, + "lastIndex": state.LastCompletedIndex, + }) + if err != nil { + return fmt.Errorf("saving checkpoint to MySQL: %w", err) + } + return nil +} + +func (c *mysqlClient) loadCheckpoint(ctx context.Context, checkpointKey string) (*checkpointState, error) { + var state checkpointState + var lastUpdatedBytes []byte + err := c.client.QueryRowContext(ctx, mysqlCheckpointLoadStmt, checkpointKey).Scan( + &state.LastCompletedIndex, + &lastUpdatedBytes, + ) + if err == sql.ErrNoRows { + return nil, nil + } + if err != nil { + return nil, fmt.Errorf("loading checkpoint from MySQL: %w", err) + } + + if len(lastUpdatedBytes) > 0 { + parsedTime, err := time.Parse("2006-01-02 15:04:05", string(lastUpdatedBytes)) + if err != nil { + return nil, fmt.Errorf("parsing LastUpdated timestamp: %w", err) + } + state.LastUpdated = parsedTime + } + + return &state, nil +} + +func (c *mysqlClient) deleteCheckpoint(ctx context.Context, checkpointKey string) error { + if *dryRun { + return nil + } + _, err := c.client.ExecContext(ctx, mysqlCheckpointDeleteStmt, checkpointKey) + if err != nil { + return fmt.Errorf("deleting checkpoint from MySQL: %w", err) + } + log.Printf("Checkpoint deleted: %s", checkpointKey) + return nil +} + +func (c *mysqlClient) supportsCheckpointing() bool { + return true +} diff --git a/release/release.mk b/release/release.mk index e99dca67b..b95ece474 100644 --- a/release/release.mk +++ b/release/release.mk @@ -2,6 +2,8 @@ # release section ################## +GORELEASER_CONFIG ?= .goreleaser.yml + # used when releasing together with GCP CloudBuild .PHONY: release release: @@ -10,7 +12,7 @@ release: # used when need to validate the goreleaser .PHONY: snapshot snapshot: - CLI_LDFLAGS="$(CLI_LDFLAGS)" SERVER_LDFLAGS="$(SERVER_LDFLAGS)" goreleaser release --skip=sign,publish --snapshot --clean --timeout 120m + CLI_LDFLAGS="$(CLI_LDFLAGS)" SERVER_LDFLAGS="$(SERVER_LDFLAGS)" goreleaser release --skip=sign,publish --snapshot --clean --config $(GORELEASER_CONFIG) --timeout 120m ########################### # sign section diff --git a/tests/backfill-test.sh b/tests/backfill-test.sh index 2076999fd..3c657641f 100755 --- a/tests/backfill-test.sh +++ b/tests/backfill-test.sh @@ -50,7 +50,18 @@ remove_keys() { set +e } +clear_checkpoint() { + local checkpoint_key=${1:-default} + if [ "$INDEX_BACKEND" == "redis" ] ; then + redis_cli DEL "backfill/checkpoint/$checkpoint_key" 2>/dev/null || true + else + mysql_cli -e "DELETE FROM BackfillCheckpoint WHERE CheckpointKey='$checkpoint_key';" 2>/dev/null || true + fi +} + check_all_entries() { + local expected_redis=${1:-21} # 20 index keys + 1 checkpoint key + local expected_mysql=${2:-26} set -e for artifact in "${!expected_artifacts[@]}" ; do local expected_uuids="${expected_artifacts[$artifact]}" @@ -113,10 +124,10 @@ check_all_entries() { local expected_size local actual_size if [ "${INDEX_BACKEND}" == "redis" ] ; then - expected_size=20 + expected_size=$expected_redis actual_size=$(redis_cli DBSIZE) else - expected_size=26 + expected_size=$expected_mysql actual_size=$(mysql_cli -NB -e "SELECT COUNT(*) FROM EntryIndex;") fi if [ $expected_size -ne $actual_size ] ; then @@ -174,6 +185,9 @@ echo echo "##### Scenario 2: backfill last half of entries #####" echo +# Clear checkpoint since we're modifying the index +clear_checkpoint + remove_keys $(seq 6 12) # searching for artifact 0-5 should succeed, but searching for later artifacts should fail @@ -190,6 +204,9 @@ echo echo "##### Scenario 3: backfill sparse entries #####" echo +# Clear checkpoint since we're modifying the index +clear_checkpoint + remove_keys $(seq 2 2 12) # searching for odd artifacts should succeed, but searching for even artifacts should fail unless it was re-signed @@ -212,3 +229,134 @@ run_backfill $end_index check_all_entries echo "Scenario 4: SUCCESS" + +echo +echo "##### Scenario 5: checkpoint and resume #####" +echo + +# Start fresh, make sure we have a clean index +if [ "$INDEX_BACKEND" == "redis" ] ; then + redis_cli FLUSHALL +else + mysql_cli -e "DELETE FROM EntryIndex; DELETE FROM BackfillCheckpoint;" 2>/dev/null || true +fi + +echo "Initial backfill from 0 to $end_index with checkpointing" +set -e +if [ "$INDEX_BACKEND" == "redis" ] ; then + go run cmd/backfill-index/main.go --rekor-address $REKOR_ADDRESS \ + --redis-hostname $REDIS_HOST --redis-port $REDIS_PORT --redis-password $REDIS_PASSWORD \ + --concurrency 5 --start 0 --end $end_index \ + --checkpoint-interval 2 +else + go run cmd/backfill-index/main.go --rekor-address $REKOR_ADDRESS \ + --mysql-dsn "${MYSQL_USER}:${MYSQL_PASSWORD}@tcp(${MYSQL_HOST}:${MYSQL_PORT})/${MYSQL_DB}" \ + --concurrency 5 --start 0 --end $end_index \ + --checkpoint-interval 2 +fi +set +e + +# Capture initial checkpoint value for later +initial_checkpoint=$end_index + +echo "Adding more entries to the log..." +for i in $(seq 13 20) ; do + minisign -GW -p $testdir/mini${i}.pub -s $testdir/mini${i}.key + echo test${i} > $testdir/blob${i} + minisign -S -s $testdir/mini${i}.key -m $testdir/blob${i} + rekor_out=$(rekor-cli --rekor_server $REKOR_ADDRESS upload \ + --artifact $testdir/blob${i} \ + --pki-format=minisign \ + --public-key $testdir/mini${i}.pub \ + --signature $testdir/blob${i}.minisig \ + --format json) + uuid=$(echo $rekor_out | jq -r .Location | cut -d '/' -f 6) + expected_keys["$testdir/mini${i}.pub"]=$uuid + expected_artifacts["$testdir/blob${i}"]=$uuid +done + +set -e +loginfo=$(rekor-cli --rekor_server $REKOR_ADDRESS loginfo --format=json) +let new_end_index=$(echo $loginfo | jq .ActiveTreeSize)-1 +set +e + +echo "New entries added, log now has indices 0-$new_end_index" + +echo "Running backfill from 0 to $new_end_index (should resume from checkpoint)" +if [ "$INDEX_BACKEND" == "redis" ] ; then + go run cmd/backfill-index/main.go --rekor-address $REKOR_ADDRESS \ + --redis-hostname $REDIS_HOST --redis-port $REDIS_PORT --redis-password $REDIS_PASSWORD \ + --concurrency 5 --start 0 --end $new_end_index \ + --checkpoint-interval 2 2>&1 | tee /tmp/backfill-resume.log +else + go run cmd/backfill-index/main.go --rekor-address $REKOR_ADDRESS \ + --mysql-dsn "${MYSQL_USER}:${MYSQL_PASSWORD}@tcp(${MYSQL_HOST}:${MYSQL_PORT})/${MYSQL_DB}" \ + --concurrency 5 --start 0 --end $new_end_index \ + --checkpoint-interval 2 2>&1 | tee /tmp/backfill-resume.log +fi + +if ! grep -q "Resuming from checkpoint: last completed index $initial_checkpoint" /tmp/backfill-resume.log ; then + echo "Scenario 5: FAILED - Did not resume from checkpoint" + cat /tmp/backfill-resume.log + exit 1 +fi +echo "Verified: Backfill resumed from checkpoint index $initial_checkpoint" + +# Verify all entries are indexed and checkpoint is at the end +# After adding 8 new entries (13-20), total is 21 entries: +# Redis: 20 index + 16 new index + 1 checkpoint = 37 keys +# MySQL: 26 + (8 entries × 2 rows) = 42 rows +check_all_entries 37 42 + +if [ "$INDEX_BACKEND" == "redis" ] ; then + checkpoint_data=$(redis_cli GET "backfill/checkpoint/default") + final_checkpoint=$(echo $checkpoint_data | jq -r .last_completed_index) +else + final_checkpoint=$(mysql_cli -NB -e "SELECT LastCompletedIndex FROM BackfillCheckpoint WHERE CheckpointKey='default'") +fi + +if [ "$final_checkpoint" != "$new_end_index" ] ; then + echo "Scenario 5: FAILED - Checkpoint should be at end index $new_end_index, got $final_checkpoint" + exit 1 +fi +echo "Checkpoint at index $final_checkpoint (end of log)" + +echo "Testing --reset-checkpoint flag" +reset_end_index=5 +if [ "$INDEX_BACKEND" == "redis" ] ; then + go run cmd/backfill-index/main.go --rekor-address $REKOR_ADDRESS \ + --redis-hostname $REDIS_HOST --redis-port $REDIS_PORT --redis-password $REDIS_PASSWORD \ + --concurrency 5 --start 0 --end $reset_end_index \ + --checkpoint-interval 2 --reset-checkpoint 2>&1 | tee /tmp/backfill-reset.log +else + go run cmd/backfill-index/main.go --rekor-address $REKOR_ADDRESS \ + --mysql-dsn "${MYSQL_USER}:${MYSQL_PASSWORD}@tcp(${MYSQL_HOST}:${MYSQL_PORT})/${MYSQL_DB}" \ + --concurrency 5 --start 0 --end $reset_end_index \ + --checkpoint-interval 2 --reset-checkpoint 2>&1 | tee /tmp/backfill-reset.log +fi + +if grep -q "Resuming from checkpoint" /tmp/backfill-reset.log ; then + echo "Scenario 5: FAILED - Reset flag did not clear checkpoint" + exit 1 +fi +if ! grep -q "Checkpoint reset - starting fresh" /tmp/backfill-reset.log ; then + echo "Scenario 5: FAILED - Did not see checkpoint reset message" + exit 1 +fi + +if [ "$INDEX_BACKEND" == "redis" ] ; then + checkpoint_data=$(redis_cli GET "backfill/checkpoint/default") + reset_checkpoint=$(echo $checkpoint_data | jq -r .last_completed_index) +else + reset_checkpoint=$(mysql_cli -NB -e "SELECT LastCompletedIndex FROM BackfillCheckpoint WHERE CheckpointKey='default'") +fi + +if [ "$reset_checkpoint" != "$reset_end_index" ] ; then + echo "Scenario 5: FAILED - Checkpoint should be at end index $reset_end_index, got $reset_checkpoint" + exit 1 +fi +echo "Checkpoint reset and at index $reset_checkpoint (end of range)" + +check_all_entries 37 42 + +echo "Scenario 5: SUCCESS" diff --git a/tests/rekor-harness.sh b/tests/rekor-harness.sh index 970f0be62..0ac0c391b 100755 --- a/tests/rekor-harness.sh +++ b/tests/rekor-harness.sh @@ -123,6 +123,10 @@ do echo "Tests passed successfully." echo "=======================================================" done + + # Clean up docker resources between server versions to free disk space + echo "Cleaning up docker resources..." + docker system prune -f --volumes done # Since we add two entries to the log for every test, once all tests are run we should have 2*(($NUM_VERSIONS_TO_TEST+1)^2) entries