Refactor validation logic and add verbose logging support

cmd/root.go

@@ -19,6 +19,7 @@ import (
 	"fmt"
 	"log/slog"
 	"os"
+	"time"
 
 	"github.com/spf13/cobra"
 
@@ -41,7 +42,7 @@ and integration with CockroachDB backup/restore workflows.
 It verifies that the storage provider is correctly configured, 
 runs synthetic workloads, and produces network performance statistics.`,
 	PersistentPreRunE: func(_ *cobra.Command, _ []string) error {
-		if envConfig.DatabaseURL == "" {
+		if envConfig.DatabaseURL == "" && !envConfig.Guess {
 			return errors.New("database URL cannot be blank")
 		}
 		if envConfig.URI != "" {
@@ -74,7 +75,13 @@ func Execute() {
 	f.StringVar(&envConfig.Path, "path", envConfig.Path, "destination path (e.g. bucket/folder)")
 	f.StringVar(&envConfig.Endpoint, "endpoint", envConfig.Path, "http endpoint")
 	f.StringVar(&envConfig.URI, "uri", envConfig.URI, "S3 URI")
+	f.BoolVar(&envConfig.Guess, "guess", false, `perform a short test to guess suggested parameters:
+it only require access to the bucket; 
+it does not try to run a full backup/restore cycle 
+in the CockroachDB cluster.`)
 	f.CountVarP(&verbosity, "verbosity", "v", "increase logging verbosity to debug")
+	f.IntVar(&envConfig.Workers, "workers", 5, "number of concurrent workers")
+	f.DurationVar(&envConfig.WorkloadDuration, "workload-duration", 5*time.Second, "duration of the workload")
 	err := rootCmd.Execute()
 
 	if err != nil {

cmd/s3/s3.go

@@ -34,6 +34,12 @@ func command(env *env.Env) *cobra.Command {
 			if err != nil {
 				return err
 			}
+			if env.Guess {
+				format.Report(cmd.OutOrStdout(), &validate.Report{
+					SuggestedParams: store.Params(),
+				})
+				return nil
+			}
 			validator, err := validate.New(ctx, env, store)
 			if err != nil {
 				return err

internal/env/env.go

@@ -14,16 +14,21 @@
 
 package env
 
+import "time"
+
 // LookupEnv is a function that retrieves the value of an environment variable.
 type LookupEnv func(key string) (string, bool)
 
 // Env holds the environment configuration.
 type Env struct {
-	DatabaseURL string    // the database connection URL
-	Endpoint    string    // the S3 endpoint
-	Path        string    // the S3 bucket path
-	LookupEnv   LookupEnv // allows injection of environment variable lookup for testing
-	Testing     bool      // enables testing mode
-	URI         string    // the S3 object URI (if not provided,will be constructed from Endpoint and Path)
-	Verbose     bool      // enables verbose logging
+	DatabaseURL      string        // the database connection URL
+	Endpoint         string        // the S3 endpoint
+	Guess            bool          // Guess the URL parameters, no validation.
+	LookupEnv        LookupEnv     // allows injection of environment variable lookup for testing
+	Path             string        // the S3 bucket path
+	Testing          bool          // enables testing mode
+	URI              string        // the S3 object URI (if not provided,will be constructed from Endpoint and Path)
+	Verbose          bool          // enables verbose logging
+	Workers          int           // number of concurrent workers
+	WorkloadDuration time.Duration // duration to run the workload
 }

internal/validate/backup.go

@@ -0,0 +1,119 @@
+// Copyright 2025 Cockroach Labs, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package validate
+
+import (
+	"log/slog"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"github.com/cockroachdb/errors"
+	"github.com/cockroachdb/field-eng-powertools/stopper"
+	"github.com/cockroachlabs-field/blobcheck/internal/db"
+)
+
+// checkBackups verifies that there is exactly one full and one incremental backup.
+func (v *Validator) checkBackups(ctx *stopper.Context, extConn *db.ExternalConn) error {
+	conn, err := v.acquireConn(ctx)
+	if err != nil {
+		return err
+	}
+	defer conn.Release()
+
+	backups, err := extConn.ListTableBackups(ctx, conn)
+	if err != nil {
+		return errors.Wrap(err, "failed to list table backups")
+	}
+	if len(backups) != expectedBackupCollections {
+		return errors.Newf("expected exactly %d backup collection, got %d", expectedBackupCollections, len(backups))
+	}
+
+	v.latest = backups[0]
+	info, err := extConn.BackupInfo(ctx, conn, backups[0], v.sourceTable)
+	if err != nil {
+		return errors.Wrap(err, "failed to get backup info")
+	}
+	if len(info) != expectedBackupCount {
+		return errors.Newf("expected exactly %d backups (1 full, 1 incremental), got %d backups", expectedBackupCount, len(info))
+	}
+
+	fullCount := 0
+	for _, i := range info {
+		if i.Full {
+			fullCount++
+		}
+	}
+	if fullCount != expectedFullBackupCount {
+		return errors.Newf("expected exactly %d full backup, got %d", expectedFullBackupCount, fullCount)
+	}
+	return nil
+}
+
+// performRestore restores the backup to a separate database.
+func (v *Validator) performRestore(
+	ctx *stopper.Context, conn *pgxpool.Conn, extConn *db.ExternalConn,
+) error {
+	slog.Info("restoring backup")
+	if err := v.restoredTable.Restore(ctx, conn, extConn, &v.sourceTable); err != nil {
+		return errors.Wrap(err, "failed to restore backup")
+	}
+	return nil
+}
+
+// runFullBackup runs a full backup in a separate database connection.
+func (v *Validator) runFullBackup(ctx *stopper.Context, extConn *db.ExternalConn) error {
+	conn, err := v.acquireConn(ctx)
+	if err != nil {
+		return err
+	}
+	defer conn.Release()
+
+	slog.Info("starting full backup")
+	if err := v.sourceTable.Backup(ctx, conn, extConn, false); err != nil {
+		return errors.Wrap(err, "failed to create full backup")
+	}
+	return nil
+}
+
+// runIncrementalBackup runs an incremental backup.
+func (v *Validator) runIncrementalBackup(
+	ctx *stopper.Context, conn *pgxpool.Conn, extConn *db.ExternalConn,
+) error {
+	slog.Info("starting incremental backup")
+	if err := v.sourceTable.Backup(ctx, conn, extConn, true); err != nil {
+		return errors.Wrap(err, "failed to create incremental backup")
+	}
+	return nil
+}
+
+// verifyIntegrity checks that the restored data matches the original.
+func (v *Validator) verifyIntegrity(ctx *stopper.Context, conn *pgxpool.Conn) error {
+	slog.Info("checking integrity")
+	original, err := v.sourceTable.Fingerprint(ctx, conn)
+	if err != nil {
+		return errors.Wrap(err, "failed to get original table fingerprint")
+	}
+
+	restore, err := v.restoredTable.Fingerprint(ctx, conn)
+	if err != nil {
+		return errors.Wrap(err, "failed to get restored table fingerprint")
+	}
+
+	if original != restore {
+		return errors.Errorf("integrity check failed: got %s, expected %s while comparing restored data with original",
+			restore, original)
+	}
+	return nil
+}

internal/validate/db.go

@@ -0,0 +1,80 @@
+// Copyright 2025 Cockroach Labs, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package validate
+
+import (
+	"log/slog"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"github.com/cockroachdb/errors"
+	"github.com/cockroachdb/field-eng-powertools/stopper"
+	"github.com/cockroachlabs-field/blobcheck/internal/db"
+)
+
+// acquireConn acquires a database connection from the pool.
+func (v *Validator) acquireConn(ctx *stopper.Context) (*pgxpool.Conn, error) {
+	conn, err := v.pool.Acquire(ctx)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to acquire database connection")
+	}
+	return conn, nil
+}
+
+// createSourceTable creates the source database and table.
+func createSourceTable(ctx *stopper.Context, conn *pgxpool.Conn) (db.KvTable, error) {
+	source := db.Database{Name: "_blobcheck"}
+	if err := source.Create(ctx, conn); err != nil {
+		return db.KvTable{}, errors.Wrap(err, "failed to create source database")
+	}
+
+	// TODO (silvano): presplit table to have ranges in all nodes
+	sourceTable := db.KvTable{
+		Database: source,
+		Schema:   db.Public,
+		Name:     "mytable",
+	}
+	if err := sourceTable.Create(ctx, conn); err != nil {
+		return db.KvTable{}, errors.Wrap(err, "failed to create source table")
+	}
+	return sourceTable, nil
+}
+
+// createRestoredTable creates the restored database and table.
+func createRestoredTable(ctx *stopper.Context, conn *pgxpool.Conn) (db.KvTable, error) {
+	dest := db.Database{Name: "_blobcheck_restored"}
+	if err := dest.Create(ctx, conn); err != nil {
+		return db.KvTable{}, errors.Wrap(err, "failed to create restored database")
+	}
+
+	restoredTable := db.KvTable{
+		Database: dest,
+		Schema:   db.Public,
+		Name:     "mytable",
+	}
+	return restoredTable, nil
+}
+
+// captureInitialStats captures initial database statistics.
+func captureInitialStats(
+	ctx *stopper.Context, extConn *db.ExternalConn, conn *pgxpool.Conn,
+) ([]*db.Stats, error) {
+	slog.Info("capturing initial statistics")
+	stats, err := extConn.Stats(ctx, conn)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to capture initial statistics")
+	}
+	return stats, nil
+}

internal/validate/minio_integration_test.go

@@ -82,11 +82,13 @@ func TestMinio(t *testing.T) {
 
 	bucketName := fmt.Sprintf("bucket-%d", time.Now().UnixMilli())
 	var env = &env.Env{
-		DatabaseURL: "postgresql://root@localhost:26257?sslmode=disable",
-		Endpoint:    endpoint,
-		LookupEnv:   lookup,
-		Path:        bucketName,
-		Testing:     true,
+		DatabaseURL:      "postgresql://root@localhost:26257?sslmode=disable",
+		Endpoint:         endpoint,
+		LookupEnv:        lookup,
+		Path:             bucketName,
+		Testing:          true,
+		Workers:          5,
+		WorkloadDuration: 5 * time.Second,
 	}
 	r.NoError(createMinioBucket(ctx, vars, env, bucketName))
 	blobStorage, err := blob.S3FromEnv(ctx, env)