S3 log store: single cluster (csimplestring#16)

* set up s3 local test docker

* s3 single driver log store

* add examples

* update readme

* update comments

Author: csimplestring

Dockerfile.aws

@@ -0,0 +1,9 @@
+FROM localstack/localstack:latest
+
+# Copy init_azurite.py script
+COPY ./init_gcs.py init_gcs.py
+
+# Copy local blobs to azurite
+COPY ./tests /tmp/localstack/tests-data
+
+COPY ./init_s3.sh /etc/localstack/init/ready.d/init_s3.sh

README.md

@@ -20,7 +20,8 @@ What is it not?
 - [x] Local file system 
 - [x] Azure Blob Storage 
 - [x] Google Cloud Storage 
-- [ ] AWS S3
+- [X] AWS S3 (single driver)
+- [ ] AWS S3 (multi drivers)
 
 ## Usage Example
 

docker-compose.yaml

@@ -17,5 +17,19 @@ services:
       dockerfile: Dockerfile.gcs
     ports:
       - 4443:4443 
-    #command: bash -c "python init_gcs.py & gcp-storage-emulator start --host=localhost --port=9023 --in-memory --default-bucket=golden"
     command: sh -c "python init_gcs.py & /bin/fake-gcs-server -scheme http -public-host localhost:4443"
+
+  aws:
+    build:
+      context: ./
+      dockerfile: Dockerfile.aws
+    ports:
+      - "4566:4566" 
+    environment:
+      - SERVICES=s3
+      - DEBUG=0 
+      - START_WEB=0 
+      - LAMBDA_REMOTE_DOCKER=0
+      - DEFAULT_REGION=us-east-1
+    volumes:
+      - '/var/run/docker.sock:/var/run/docker.sock'

examples/gcs_example_test.go

@@ -0,0 +1,41 @@
+package examples
+
+import (
+	"log"
+	"os"
+	"testing"
+
+	delta "github.com/csimplestring/delta-go"
+)
+
+func TestGCSLogStoreExample(t *testing.T) {
+
+	// for local emulator only
+	os.Setenv("STORAGE_EMULATOR_HOST", "localhost:4443")
+
+	path := "gs://golden/snapshot-data0/"
+	config := delta.Config{
+		StoreType: "gs",
+	}
+
+	table, err := delta.ForTable(path, config, &delta.SystemClock{})
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	s, err := table.Snapshot()
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	version := s.Version()
+	log.Println(version)
+
+	files, err := s.AllFiles()
+	if err != nil {
+		log.Fatal(err)
+	}
+	for _, f := range files {
+		log.Println(f.Path)
+	}
+}

examples/s3_example_test.go

@@ -0,0 +1,46 @@
+package examples
+
+import (
+	"log"
+	"os"
+	"testing"
+
+	delta "github.com/csimplestring/delta-go"
+)
+
+func TestS3LogStoreExample(t *testing.T) {
+
+	os.Setenv("AWS_DEFAULT_REGION", "us-east-1")
+	os.Setenv("AWS_DISABLE_SSL", "true")
+	os.Setenv("AWS_S3_FORCE_PATH_STYLE", "true")
+	os.Setenv("AWS_ACCESS_KEY_ID", "foo")
+	os.Setenv("AWS_SECRET_ACCESS_KEY", "bar")
+	// for localstack test only
+	os.Setenv("AWS_ENDPOINT_URL", "http://localhost:4566")
+
+	path := "s3://golden/snapshot-data0/"
+	config := delta.Config{
+		StoreType: "s3",
+	}
+
+	table, err := delta.ForTable(path, config, &delta.SystemClock{})
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	s, err := table.Snapshot()
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	version := s.Version()
+	log.Println(version)
+
+	files, err := s.AllFiles()
+	if err != nil {
+		log.Fatal(err)
+	}
+	for _, f := range files {
+		log.Println(f.Path)
+	}
+}

go.mod

@@ -33,6 +33,26 @@ require (
 	github.com/Azure/go-autorest/autorest/to v0.4.0 // indirect
 	github.com/AzureAD/microsoft-authentication-library-for-go v0.8.1 // indirect
 	github.com/apache/thrift v0.16.0 // indirect
+	github.com/aws/aws-sdk-go v1.44.200 // indirect
+	github.com/aws/aws-sdk-go-v2 v1.17.4 // indirect
+	github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.4.10 // indirect
+	github.com/aws/aws-sdk-go-v2/config v1.18.12 // indirect
+	github.com/aws/aws-sdk-go-v2/credentials v1.13.12 // indirect
+	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.12.22 // indirect
+	github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.11.51 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/configsources v1.1.28 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.4.22 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/ini v1.3.29 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/v4a v1.0.19 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.9.11 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.1.23 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.9.22 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.13.22 // indirect
+	github.com/aws/aws-sdk-go-v2/service/s3 v1.30.2 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sso v1.12.1 // indirect
+	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.14.1 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sts v1.18.3 // indirect
+	github.com/aws/smithy-go v1.13.5 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/golang-jwt/jwt/v4 v4.4.3 // indirect
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
@@ -42,6 +62,7 @@ require (
 	github.com/google/wire v0.5.0 // indirect
 	github.com/googleapis/enterprise-certificate-proxy v0.2.3 // indirect
 	github.com/googleapis/gax-go/v2 v2.7.1 // indirect
+	github.com/jmespath/go-jmespath v0.4.0 // indirect
 	github.com/kylelemons/godebug v1.1.0 // indirect
 	github.com/pkg/browser v0.0.0-20210911075715-681adbf594b8 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect

go.sum

@@ -1318,6 +1318,7 @@ github.com/jmespath/go-jmespath v0.0.0-20160202185014-0b12d6b521d8/go.mod h1:Nht
 github.com/jmespath/go-jmespath v0.0.0-20160803190731-bd40a432e4c7/go.mod h1:Nht3zPeWKUH0NzdCt2Blrr5ys8VGpn0CEB0cQHVjt7k=
 github.com/jmespath/go-jmespath v0.4.0 h1:BEgLn5cpjn8UN1mAw4NjwDrS35OdebyEtFe+9YPoQUg=
 github.com/jmespath/go-jmespath v0.4.0/go.mod h1:T8mJZnbsbmF+m6zOOFylbeCJqk5+pHWvzYPziyZiYoo=
+github.com/jmespath/go-jmespath/internal/testify v1.5.1 h1:shLQSRRSCCPj3f2gpwzGwWFoC7ycTf1rcQZHOlsJ6N8=
 github.com/jmespath/go-jmespath/internal/testify v1.5.1/go.mod h1:L3OGu8Wl2/fWfCI6z80xFu9LTZmf1ZRjMHUOPmWr69U=
 github.com/joefitzgerald/rainbow-reporter v0.1.0/go.mod h1:481CNgqmVHQZzdIbN52CupLJyoVwB10FQ/IQlF1pdL8=
 github.com/joho/godotenv v1.3.0/go.mod h1:7hK45KPybAkOC6peb+G5yklZfMxEjkZhHbwpqxOKXbg=

init_s3.sh

@@ -0,0 +1,9 @@
+#!/usr/bin/env bash
+
+set -euo pipefail
+
+aws configure set aws_access_key_id "AKIAI44QH8DHBEXAMPLE" --profile user2 && aws configure set aws_secret_access_key "je7MtGbClwBF/2Zp9Utk/h3yCo8nvbEXAMPLEKEY" --profile user2 && aws configure set region "us-east-1" --profile user2 && aws configure set output "text" --profile user2
+
+aws --endpoint-url=http://localhost:4566 s3 mb s3://golden 
+
+aws --endpoint-url="http://localhost:4566" s3 sync /tmp/localstack/tests-data/golden s3://golden

internal/util/path/path.go

@@ -2,6 +2,7 @@ package path
 
 import (
 	"net/url"
+	"os"
 	"strings"
 
 	"github.com/rotisserie/eris"
@@ -17,7 +18,7 @@ func ConvertToBlobURL(urlstr string) (string, error) {
 		return "", err
 	}
 
-	if len(u.Query()) > 0 {
+	if len(u.Query()) > 0 && u.Scheme != "s3" {
 		return "", eris.New("path url cannot have query parameters!")
 	}
 
@@ -35,6 +36,30 @@ func ConvertToBlobURL(urlstr string) (string, error) {
 			v.Set("prefix", prefix)
 		}
 
+		q, err := url.QueryUnescape(v.Encode())
+		if err != nil {
+			return "", eris.Wrap(err, "")
+		}
+		u.RawQuery = q
+		return u.String(), nil
+	} else if u.Scheme == "s3" {
+		bucket, prefix, found := strings.Cut(u.Path, "/")
+		u.Path = bucket
+		// set prefix
+		if found {
+			v.Set("prefix", prefix)
+		}
+		// set the endpoint url if localstack is used
+		if endpoint, ok := os.LookupEnv("AWS_ENDPOINT_URL"); ok {
+			v.Set("endpoint", endpoint)
+		}
+		if disableSSL, ok := os.LookupEnv("AWS_DISABLE_SSL"); ok {
+			v.Set("disableSSL", disableSSL)
+		}
+		if s3ForcePathStyle, ok := os.LookupEnv("AWS_S3_FORCE_PATH_STYLE"); ok {
+			v.Set("s3ForcePathStyle", s3ForcePathStyle)
+		}
+
 		q, err := url.QueryUnescape(v.Encode())
 		if err != nil {
 			return "", eris.Wrap(err, "")

internal/util/path/resolver.go

@@ -34,6 +34,8 @@ func Relative(base string, path string) (string, error) {
 		return azureBlobRelative(base, path)
 	} else if p.Scheme == "gs" {
 		return gcsRelative(base, path)
+	} else if p.Scheme == "s3" {
+		return s3Relative(base, path)
 	}
 	return "", eris.New(fmt.Sprintf("unsupported scheme %s", p.Scheme))
 }
@@ -78,6 +80,10 @@ func gcsRelative(base string, path string) (string, error) {
 	return path, nil
 }
 
+func s3Relative(base string, path string) (string, error) {
+	return path, nil
+}
+
 func Canonicalize(path string, schema string) (string, error) {
 
 	if schema == "file" {
@@ -86,6 +92,8 @@ func Canonicalize(path string, schema string) (string, error) {
 		return azblobCanonicalize(path)
 	} else if schema == "gs" {
 		return gcsCanonicalize(path)
+	} else if schema == "s3" {
+		return s3Canonicalize(path)
 	}
 
 	return "", eris.Wrap(errno.UnsupportedFileSystem("unsupported schema to canonicalize"), schema)
@@ -112,3 +120,7 @@ func azblobCanonicalize(p string) (string, error) {
 func gcsCanonicalize(p string) (string, error) {
 	return p, nil
 }
+
+func s3Canonicalize(p string) (string, error) {
+	return p, nil
+}