From ecae0e38950f17c461152c632c708ffcc21065cf Mon Sep 17 00:00:00 2001 From: Saumya Shah Date: Sun, 20 Sep 2026 10:06:26 +0530 Subject: [PATCH] feat(db): introduce new seaweedfs object store usage for e2e Signed-off-by: Saumya Shah refactor a bit make a generalized constant for s3 secret and access key Signed-off-by: Saumya Shah defer close idle conn in ready() check and add -master.telemetry=false so non-related to e2e stuff doesnt start Signed-off-by: Saumya Shah --- db/db.go | 161 ++++++++++++++++++++++++++++++++++++++++++++++---- db/db_test.go | 28 ++++++++- 2 files changed, 176 insertions(+), 13 deletions(-) diff --git a/db/db.go b/db/db.go index cc140da..0302036 100644 --- a/db/db.go +++ b/db/db.go @@ -6,13 +6,16 @@ package e2edb import ( + "context" "crypto/rand" "crypto/rsa" + "crypto/tls" "crypto/x509" "encoding/pem" "fmt" "math/big" "net" + "net/http" "os" "path/filepath" "strconv" @@ -23,11 +26,13 @@ import ( "github.com/efficientgo/e2e" e2emon "github.com/efficientgo/e2e/monitoring" e2eprof "github.com/efficientgo/e2e/profiling" + "github.com/minio/minio-go/v7" + "github.com/minio/minio-go/v7/pkg/credentials" ) const ( - MinioAccessKey = "Cheescake" - MinioSecretKey = "supersecret" + S3AccessKey = "Cheescake" + S3SecretKey = "supersecret" ) type Option func(*options) @@ -36,6 +41,7 @@ type options struct { image string flagOverride map[string]string minioOptions minioOptions + seaweedOptions seaweedFSOptions azuriteOptions azuriteOptions } @@ -50,6 +56,10 @@ type minioOptions struct { enableTLS bool } +type seaweedFSOptions struct { + enableTLS bool +} + func WithImage(image string) Option { return func(o *options) { o.image = image @@ -74,6 +84,12 @@ func WithMinioTLS() Option { } } +func WithSeaweedFSTLS() Option { + return func(o *options) { + o.seaweedOptions.enableTLS = true + } +} + func WithAzuriteStorageAccounts(acc string) Option { return func(o *options) { o.azuriteOptions.customStorageAccounts = acc @@ -116,8 +132,8 @@ func NewMinio(env e2e.Environment, name, bktName string, opts ...Option) *e2emon userID := strconv.Itoa(os.Getuid()) ports := map[string]int{AccessPortName: 8090} envVars := []string{ - "MINIO_ROOT_USER=" + MinioAccessKey, - "MINIO_ROOT_PASSWORD=" + MinioSecretKey, + "MINIO_ROOT_USER=" + S3AccessKey, + "MINIO_ROOT_PASSWORD=" + S3SecretKey, "MINIO_BROWSER=" + "off", } @@ -141,7 +157,7 @@ func NewMinio(env e2e.Environment, name, bktName string, opts ...Option) *e2emon var readiness e2e.ReadinessProbe if o.minioOptions.enableTLS { - if err := os.MkdirAll(filepath.Join(f.Dir(), "certs", "CAs"), 0750); err != nil { + if err := os.MkdirAll(filepath.Join(f.Dir(), "certs", "CAs"), 0o750); err != nil { return &e2emon.InstrumentedRunnable{Runnable: e2e.NewFailedRunnable(name, errors.Wrap(err, "create certs dir"))} } @@ -226,7 +242,7 @@ func NewAzuriteBlobStorage(env e2e.Environment, name string, opts ...Option) *e2 ) if o.azuriteOptions.enableTLS { - if err := os.MkdirAll(filepath.Join(f.Dir(), "certs", "CAs"), 0750); err != nil { + if err := os.MkdirAll(filepath.Join(f.Dir(), "certs", "CAs"), 0o750); err != nil { return &e2emon.InstrumentedRunnable{Runnable: e2e.NewFailedRunnable(name, errors.Wrap(err, "create certs dir"))} } @@ -296,7 +312,7 @@ func NewAzuriteBlobStorageWriter(env e2e.Environment, name, containerName, tempD // genCerts generates certificates and writes those to the provided paths. func genCerts(certPath, privkeyPath, caPath, serverName string) error { - var caRoot = &x509.Certificate{ + caRoot := &x509.Certificate{ SerialNumber: big.NewInt(2019), NotAfter: time.Now().AddDate(10, 0, 0), IsCA: true, @@ -305,7 +321,7 @@ func genCerts(certPath, privkeyPath, caPath, serverName string) error { BasicConstraintsValid: true, } - var cert = &x509.Certificate{ + cert := &x509.Certificate{ SerialNumber: big.NewInt(1658), DNSNames: []string{serverName}, IPAddresses: []net.IP{net.ParseIP("127.0.0.1"), net.ParseIP("::1")}, @@ -333,7 +349,7 @@ func genCerts(certPath, privkeyPath, caPath, serverName string) error { Type: "CERTIFICATE", Bytes: caBytes, }) - err = os.WriteFile(caPath, caPEM, 0644) + err = os.WriteFile(caPath, caPEM, 0o644) if err != nil { return err } @@ -347,7 +363,7 @@ func genCerts(certPath, privkeyPath, caPath, serverName string) error { Type: "CERTIFICATE", Bytes: certBytes, }) - err = os.WriteFile(certPath, certPEM, 0644) + err = os.WriteFile(certPath, certPEM, 0o644) if err != nil { return err } @@ -356,7 +372,7 @@ func genCerts(certPath, privkeyPath, caPath, serverName string) error { Type: "RSA PRIVATE KEY", Bytes: x509.MarshalPKCS1PrivateKey(certPrivKey), }) - err = os.WriteFile(privkeyPath, certPrivKeyPEM, 0644) + err = os.WriteFile(privkeyPath, certPrivKeyPEM, 0o644) if err != nil { return err } @@ -473,3 +489,126 @@ func NewETCD(env e2e.Environment, name string, opts ...Option) *e2emon.Instrumen }, ), "metrics") } + +func NewSeaweedFS(env e2e.Environment, name, bucket string, opts ...Option) *e2emon.InstrumentedRunnable { + o := options{image: "chrislusf/seaweedfs:4.47"} + for _, opt := range opts { + opt(&o) + } + + const adminPortName = "admin" + ports := map[string]int{ + AccessPortName: 8333, + adminPortName: 23646, + } + f := env.Runnable(name).WithPorts(ports).Future() + dataDir := filepath.Join(f.Dir(), "data") + if err := os.MkdirAll(dataDir, 0o750); err != nil { + return &e2emon.InstrumentedRunnable{Runnable: e2e.NewFailedRunnable(name, errors.Wrap(err, "create SeaweedFS data directory"))} + } + + args := []string{ + "mini", + "-dir=/data", + fmt.Sprintf("-s3.port=%d", ports[AccessPortName]), + "-master.telemetry=false", + "-webdav=false", + "-s3.port.iceberg=0", + "-s3.port.lance=0", + "-s3.autoCreateBucket=false", + } + readiness := e2e.ReadinessProbe(e2e.NewHTTPReadinessProbe(AccessPortName, "/readyz", http.StatusOK, http.StatusOK)) + instrumentedOpts := []e2emon.InstrumentedOption{} + caFile := "" + + if o.seaweedOptions.enableTLS { + certDir := filepath.Join(f.Dir(), "certs") + caDir := filepath.Join(certDir, "CAs") + if err := os.MkdirAll(caDir, 0o750); err != nil { + return &e2emon.InstrumentedRunnable{Runnable: e2e.NewFailedRunnable(name, errors.Wrap(err, "create SeaweedFS certificate directory"))} + } + + certFile := filepath.Join(certDir, "public.crt") + keyFile := filepath.Join(certDir, "private.key") + caFile = filepath.Join(caDir, "ca.crt") + if err := genCerts(certFile, keyFile, caFile, fmt.Sprintf("%s-%s", env.Name(), name)); err != nil { + return &e2emon.InstrumentedRunnable{Runnable: e2e.NewFailedRunnable(name, errors.Wrap(err, "generate SeaweedFS certificates"))} + } + + args = append(args, "-s3.cert.file="+certFile, "-s3.key.file="+keyFile) + readiness = e2e.NewHTTPSReadinessProbe(AccessPortName, "/readyz", http.StatusOK, http.StatusOK) + instrumentedOpts = append(instrumentedOpts, e2emon.WithInstrumentedScheme("https")) + } + + envVars := map[string]string{ + "AWS_ACCESS_KEY_ID": S3AccessKey, + "AWS_SECRET_ACCESS_KEY": S3SecretKey, + } + if bucket != "" { + envVars["S3_BUCKET"] = bucket + readiness = &seaweedFSBucketReadinessProbe{ + readiness: readiness, + bucket: bucket, + enableTLS: o.seaweedOptions.enableTLS, + caFile: caFile, + } + } + + r := f.Init(e2e.StartOptions{ + Image: o.image, + User: strconv.Itoa(os.Getuid()), + Command: e2e.NewCommand(args[0], args[1:]...), + EnvVars: envVars, + Readiness: readiness, + Volumes: []string{dataDir + ":/data:z"}, + }) + return e2emon.AsInstrumented(r, AccessPortName, instrumentedOpts...) +} + +type seaweedFSBucketReadinessProbe struct { + readiness e2e.ReadinessProbe + bucket string + enableTLS bool + caFile string +} + +// This explicitly checks if bucket exists or not (seaweedfs server might report healthy while bucket creation is still pending, which happens in later stage +func (p *seaweedFSBucketReadinessProbe) Ready(runnable e2e.Runnable) error { + if err := p.readiness.Ready(runnable); err != nil { + return err + } + + transport := http.DefaultTransport.(*http.Transport).Clone() + defer transport.CloseIdleConnections() + if p.enableTLS { + caPEM, err := os.ReadFile(p.caFile) + if err != nil { + return errors.Wrap(err, "read SeaweedFS CA certificate") + } + rootCAs := x509.NewCertPool() + if !rootCAs.AppendCertsFromPEM(caPEM) { + return errors.New("parse SeaweedFS CA certificate") + } + transport.TLSClientConfig = &tls.Config{RootCAs: rootCAs} + } + + client, err := minio.New(runnable.Endpoint(AccessPortName), &minio.Options{ + Creds: credentials.NewStaticV4(S3AccessKey, S3SecretKey, ""), + Secure: p.enableTLS, + Transport: transport, + }) + if err != nil { + return errors.Wrap(err, "create SeaweedFS readiness client") + } + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + exists, err := client.BucketExists(ctx, p.bucket) + if err != nil { + return errors.Wrap(err, "check SeaweedFS bucket") + } + if !exists { + return errors.Newf("SeaweedFS bucket %q does not exist", p.bucket) + } + return nil +} diff --git a/db/db_test.go b/db/db_test.go index 37d0639..8b8e8c0 100644 --- a/db/db_test.go +++ b/db/db_test.go @@ -25,8 +25,8 @@ func TestMinio(t *testing.T) { testutil.Ok(t, e2e.StartAndWaitReady(minioContainer)) endpoint := minioContainer.Endpoint("http") - accessKeyID := MinioAccessKey - secretAccessKey := MinioSecretKey + accessKeyID := S3AccessKey + secretAccessKey := S3SecretKey minioClient, err := minio.New(endpoint, &minio.Options{ Creds: credentials.NewStaticV4(accessKeyID, secretAccessKey, ""), Secure: false, @@ -38,3 +38,27 @@ func TestMinio(t *testing.T) { minioClient.MakeBucket(context.Background(), "test-bucket", minio.MakeBucketOptions{}), ) } + +func TestSeaweedFS(t *testing.T) { + for _, tc := range []struct { + name string + secure bool + }{ + {name: "HTTP"}, + {name: "HTTPS", secure: true}, + } { + t.Run(tc.name, func(t *testing.T) { + e, err := e2e.New() + testutil.Ok(t, err) + t.Cleanup(e.Close) + + opts := []Option{} + if tc.secure { + opts = append(opts, WithSeaweedFSTLS()) + } + const bucket = "test-bucket" + seaweedFS := NewSeaweedFS(e, "seaweedfs", bucket, opts...) + testutil.Ok(t, e2e.StartAndWaitReady(seaweedFS)) + }) + } +}