mosuka · mosuka · Jan 15, 2022 · Jan 15, 2022
diff --git a/.gitignore b/.gitignore
@@ -8,6 +8,8 @@
 bin/phalanx
 dist/
 
+.env
+
 *.pem
 *.csr
 

diff --git a/clients/s3.go b/clients/s3.go
@@ -0,0 +1,112 @@
+package clients
+
+import (
+	"context"
+	"fmt"
+	"net/url"
+	"os"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/config"
+	"github.com/aws/aws-sdk-go-v2/credentials"
+	"github.com/aws/aws-sdk-go-v2/service/s3"
+	"github.com/mosuka/phalanx/errors"
+)
+
+func NewS3ClientWithUri(uri string) (*s3.Client, error) {
+	// Parse URI.
+	u, err := url.Parse(uri)
+	if err != nil {
+		return nil, err
+	}
+
+	if u.Scheme != "s3" {
+		return nil, errors.ErrInvalidUri
+	}
+
+	ctx := context.Background()
+
+	var cfg aws.Config
+
+	profile := os.Getenv("AWS_PROFILE")
+	if str := u.Query().Get("profile"); str != "" {
+		profile = str
+	}
+	if profile != "" {
+		cfg, err = config.LoadDefaultConfig(ctx, config.WithSharedConfigProfile(profile))
+		if err != nil {
+			return nil, err
+		}
+	} else {
+		cfg, err = config.LoadDefaultConfig(ctx)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	useCredentials := false
+	accessKeyId := os.Getenv("AWS_ACCESS_KEY_ID")
+	if str := u.Query().Get("access_key_id"); str != "" {
+		accessKeyId = str
+	}
+	if accessKeyId != "" {
+		useCredentials = true
+	}
+
+	secretAccessKey := os.Getenv("AWS_SECRET_ACCESS_KEY")
+	if str := u.Query().Get("secret_access_key"); str != "" {
+		secretAccessKey = str
+	}
+	if secretAccessKey != "" {
+		useCredentials = true
+	}
+
+	sessionToken := os.Getenv("AWS_SESSION_TOKEN")
+	if str := u.Query().Get("session_token"); str != "" {
+		sessionToken = str
+	}
+	if sessionToken != "" {
+		useCredentials = true
+	}
+
+	if useCredentials {
+		cfg.Credentials = aws.NewCredentialsCache(credentials.NewStaticCredentialsProvider(accessKeyId, secretAccessKey, sessionToken))
+	}
+
+	region := os.Getenv("AWS_DEFAULT_REGION")
+	if str := u.Query().Get("region"); str != "" {
+		region = str
+	}
+	if region != "" {
+		cfg.Region = region
+	}
+
+	endpoint := os.Getenv("AWS_ENDPOINT")
+	if str := u.Query().Get("endpoint"); str != "" {
+		endpoint = str
+	}
+	if endpoint != "" {
+		customResolver := aws.EndpointResolverWithOptionsFunc(func(service, region string, options ...interface{}) (aws.Endpoint, error) {
+			return aws.Endpoint{
+				PartitionID:   "aws",
+				URL:           endpoint,
+				SigningRegion: cfg.Region,
+			}, nil
+		})
+		cfg.EndpointResolverWithOptions = customResolver
+	}
+
+	usePathStyle := false
+	usePathStyleStr := os.Getenv("AWS_USE_PATH_STYLE")
+	if str := u.Query().Get("use_path_style"); str != "" {
+		usePathStyleStr = str
+	}
+	if usePathStyleStr == "true" {
+		fmt.Println("AWS_USE_PATH_STYLE is set to true. Using path style.")
+		usePathStyle = true
+	}
+
+	return s3.NewFromConfig(cfg, func(options *s3.Options) {
+		options.UsePathStyle = usePathStyle
+	}), nil
+}
diff --git a/clients_integration_test/etcd_test.go b/clients_integration_test/etcd_test.go
@@ -0,0 +1,17 @@
+//go:build integration
+
+package clients_integration_test
+
+import (
+	"testing"
+
+	"github.com/mosuka/phalanx/clients"
+)
+
+func TestNewEtcdClientWithUri(t *testing.T) {
+	uri := "etcd://phalanx-test/metastore/test?endpoints=localhost:2379"
+
+	if _, err := clients.NewEtcdClientWithUri(uri); err != nil {
+		t.Fatalf("error %v\n", err)
+	}
+}
diff --git a/clients_integration_test/minio_test.go b/clients_integration_test/minio_test.go
@@ -0,0 +1,17 @@
+//go:build integration
+
+package clients_integration_test
+
+import (
+	"testing"
+
+	"github.com/mosuka/phalanx/clients"
+)
+
+func TestNewMinioClientWithUri(t *testing.T) {
+	uri := "minio://phalanx-test/indexes/test?endpoint=localhost:9000"
+
+	if _, err := clients.NewMinioClientWithUri(uri); err != nil {
+		t.Fatalf("error %v\n", err)
+	}
+}
diff --git a/clients_integration_test/s3_test.go b/clients_integration_test/s3_test.go
@@ -0,0 +1,17 @@
+//go:build integration
+
+package clients_integration_test
+
+import (
+	"testing"
+
+	"github.com/mosuka/phalanx/clients"
+)
+
+func TestNewS3ClientWithUri(t *testing.T) {
+	uri := "s3://phalanx-test/indexes/test?endpoint=http://localhost:4572"
+
+	if _, err := clients.NewS3ClientWithUri(uri); err != nil {
+		t.Fatalf("error %v\n", err)
+	}
+}
diff --git a/cluster/cluster.go b/cluster/cluster.go
@@ -57,8 +57,6 @@ func NewCluster(host string, bindPort int, nodeMetadata NodeMetadata, isSeedNode
 	// }
 	// members.UpdateNode(10 * time.Second)
 
-	fmt.Println("cluster created", memberList.LocalNode().Name)
-
 	return &Cluster{
 		memberList:           memberList,
 		nodeEventDeliegate:   nodeEventDeliegate,

diff --git a/directory/directory.go b/directory/directory.go
@@ -21,21 +21,30 @@ const (
 	SchemeTypeMem
 	SchemeTypeFile
 	SchemeTypeMinio
+	SchemeTypeS3
 )
 
+type uint64Slice []uint64
+
+func (e uint64Slice) Len() int           { return len(e) }
+func (e uint64Slice) Swap(i, j int)      { e[i], e[j] = e[j], e[i] }
+func (e uint64Slice) Less(i, j int) bool { return e[i] < e[j] }
+
 // Enum value maps for SchemeType.
 var (
 	SchemeType_name = map[SchemeType]string{
 		SchemeTypeUnknown: "unknown",
 		SchemeTypeMem:     "mem",
 		SchemeTypeFile:    "file",
 		SchemeTypeMinio:   "minio",
+		SchemeTypeS3:      "s3",
 	}
 	SchemeType_value = map[string]SchemeType{
 		"unknown": SchemeTypeUnknown,
 		"mem":     SchemeTypeMem,
 		"file":    SchemeTypeFile,
 		"minio":   SchemeTypeMinio,
+		"s3":      SchemeTypeS3,
 	}
 )
 
@@ -64,6 +73,8 @@ func NewIndexConfigWithUri(uri string, lockUri string, logger *zap.Logger) (blug
 		return FileSystemIndexConfig(uri, directoryLogger), nil
 	case SchemeType_name[SchemeTypeMinio]:
 		return MinioIndexConfig(uri, lockUri, directoryLogger), nil
+	case SchemeType_name[SchemeTypeS3]:
+		return S3IndexConfig(uri, lockUri, directoryLogger), nil
 	default:
 		err := errors.ErrUnsupportedDirectoryType
 		directoryLogger.Error(err.Error(), zap.String("scheme", u.Scheme))

diff --git a/directory/directory_minio.go b/directory/directory_minio.go
@@ -23,13 +23,6 @@ import (
 	"go.uber.org/zap"
 )
 
-type uint64Slice []uint64
-
-func (e uint64Slice) Len() int           { return len(e) }
-func (e uint64Slice) Swap(i, j int)      { e[i], e[j] = e[j], e[i] }
-func (e uint64Slice) Less(i, j int) bool { return e[i] < e[j] }
-
-// func MinioIndexConfig(uri string, lockManager lock.LockManager, logger *zap.Logger) bluge.Config {
 func MinioIndexConfig(uri string, lockUri string, logger *zap.Logger) bluge.Config {
 	return bluge.DefaultConfigWithDirectory(func() index.Directory {
 		return NewMinioDirectoryWithUri(uri, lockUri, logger)
@@ -47,7 +40,6 @@ type MinioDirectory struct {
 	logger         *zap.Logger
 }
 
-// func NewMinioDirectoryWithUri(uri string, lockManager lock.LockManager, logger *zap.Logger) *MinioDirectory {
 func NewMinioDirectoryWithUri(uri string, lockUri string, logger *zap.Logger) *MinioDirectory {
 	directoryLogger := logger.Named("minio")
 
@@ -120,8 +112,7 @@ func (d *MinioDirectory) Setup(readOnly bool) error {
 			Region: region,
 		}
 
-		err = d.client.MakeBucket(ctx, d.bucket, opts)
-		if err != nil {
+		if err = d.client.MakeBucket(ctx, d.bucket, opts); err != nil {
 			d.logger.Error(err.Error(), zap.String("region", region), zap.String("bucket", d.bucket))
 			return err
 		}
@@ -236,8 +227,7 @@ func (d *MinioDirectory) Remove(kind string, id uint64) error {
 	ctx, cancel := context.WithTimeout(d.ctx, d.requestTimeout)
 	defer cancel()
 
-	err := d.client.RemoveObject(ctx, d.bucket, path, opts)
-	if err != nil {
+	if err := d.client.RemoveObject(ctx, d.bucket, path, opts); err != nil {
 		d.logger.Error(err.Error(), zap.String("bucket", d.bucket), zap.String("path", path), zap.Any("opts", opts))
 		return err
 	}
-Original file line number
+Diff line change
@@ Expand Up / @@ -8,6 +8,8 @@ @@
     bin/phalanx
     dist/
+    .env
     *.pem
     *.csr
@@ Expand Down @@