Michael McGuinness
94 lines
2.1 KiB
Go
94 lines
2.1 KiB
Go
package objectstore
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"log/slog"
|
|
awsc "queryorchestration/internal/serviceconfig/aws"
|
|
"time"
|
|
|
|
"github.com/aws/aws-sdk-go-v2/config"
|
|
"github.com/aws/aws-sdk-go-v2/service/s3"
|
|
)
|
|
|
|
type ObjectStoreConfig struct {
|
|
AWSEndpointUrlS3 string `env:"AWS_ENDPOINT_URL_S3"`
|
|
AWSS3UsePathStyle bool `env:"AWS_S3_USE_PATH_STYLE" envDefault:"false"`
|
|
StoreClient S3Client
|
|
}
|
|
|
|
type ConfigProvider interface {
|
|
awsc.ConfigProvider
|
|
SetStoreClient(context.Context) error
|
|
SetStoreClientAndPingByName(context.Context, string) error
|
|
GetStoreClient() S3Client
|
|
PingStoreByName(context.Context, string) error
|
|
SetS3Endpoint(string)
|
|
GetS3Endpoint() string
|
|
SetS3UsePathStyle(bool)
|
|
}
|
|
|
|
func (c *ObjectStoreConfig) SetStoreClientAndPingByName(ctx context.Context, name string) error {
|
|
err := c.SetStoreClient(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return c.PingStoreByName(ctx, name)
|
|
}
|
|
|
|
func (c *ObjectStoreConfig) SetStoreClient(ctx context.Context) error {
|
|
cfg, err := config.LoadDefaultConfig(ctx)
|
|
if err != nil {
|
|
return fmt.Errorf("unable to load SDK config: %v", err)
|
|
}
|
|
|
|
c.StoreClient = s3.NewFromConfig(cfg, func(o *s3.Options) {
|
|
o.UsePathStyle = c.AWSS3UsePathStyle
|
|
})
|
|
|
|
return nil
|
|
}
|
|
|
|
func (c *ObjectStoreConfig) GetStoreClient() S3Client {
|
|
return c.StoreClient
|
|
}
|
|
|
|
func (c *ObjectStoreConfig) PingStoreByName(ctx context.Context, name string) error {
|
|
timeout := time.After(30 * time.Second)
|
|
tick := time.NewTicker(2 * time.Second)
|
|
defer tick.Stop()
|
|
|
|
input := &s3.HeadBucketInput{
|
|
Bucket: &name,
|
|
}
|
|
|
|
for {
|
|
select {
|
|
case <-timeout:
|
|
return fmt.Errorf("unable to ping bucket by name: %s", name)
|
|
case <-tick.C:
|
|
_, err := c.StoreClient.HeadBucket(ctx, input)
|
|
if err == nil {
|
|
slog.Info("Successful bucket ping", "name", name)
|
|
return nil
|
|
}
|
|
slog.Info("Attempted bucket ping", "name", name, "address", c.GetS3Endpoint())
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
}
|
|
}
|
|
}
|
|
|
|
func (c *ObjectStoreConfig) GetS3Endpoint() string {
|
|
return c.AWSEndpointUrlS3
|
|
}
|
|
|
|
func (c *ObjectStoreConfig) SetS3Endpoint(e string) {
|
|
c.AWSEndpointUrlS3 = e
|
|
}
|
|
|
|
func (c *ObjectStoreConfig) SetS3UsePathStyle(e bool) {
|
|
c.AWSS3UsePathStyle = e
|
|
}
|