From 25dd435bc65bc15efcb480758cae17774c5c8fcd Mon Sep 17 00:00:00 2001 From: Julian Sommer <115516918+juliansommer@users.noreply.github.com> Date: Mon, 9 Feb 2026 09:52:12 +1000 Subject: [PATCH] fix: use object store instead of s3 client --- hive/daemon/daemon.go | 4 +- hive/daemon/daemon_test.go | 3 +- hive/handlers/ec2/image/service_impl.go | 2 +- hive/handlers/ec2/instance/service_impl.go | 35 ++-- hive/handlers/ec2/key/service_impl.go | 2 +- hive/handlers/ec2/volume/service_impl.go | 2 +- hive/handlers/ec2/volume/service_impl_test.go | 2 +- .../ec2 => }/objectstore/objectstore.go | 0 .../ec2 => }/objectstore/objectstore_test.go | 0 hive/s3client/s3client.go | 166 ------------------ 10 files changed, 27 insertions(+), 189 deletions(-) rename hive/{handlers/ec2 => }/objectstore/objectstore.go (100%) rename hive/{handlers/ec2 => }/objectstore/objectstore_test.go (100%) delete mode 100644 hive/s3client/s3client.go diff --git a/hive/daemon/daemon.go b/hive/daemon/daemon.go index 9256e3336..e464a0b9d 100644 --- a/hive/daemon/daemon.go +++ b/hive/daemon/daemon.go @@ -29,6 +29,7 @@ import ( handlers_ec2_image "github.com/mulgadc/hive/hive/handlers/ec2/image" handlers_ec2_instance "github.com/mulgadc/hive/hive/handlers/ec2/instance" handlers_ec2_key "github.com/mulgadc/hive/hive/handlers/ec2/key" + "github.com/mulgadc/hive/hive/objectstore" handlers_ec2_volume "github.com/mulgadc/hive/hive/handlers/ec2/volume" "github.com/mulgadc/hive/hive/qmp" "github.com/mulgadc/hive/hive/utils" @@ -474,7 +475,8 @@ func (d *Daemon) Start() error { } // Create services before loading/launching instances, since LaunchInstance depends on them - d.instanceService = handlers_ec2_instance.NewInstanceServiceImpl(d.config, d.resourceMgr.instanceTypes, d.natsConn, &d.Instances) + store := objectstore.NewS3ObjectStoreFromConfig(d.config.Predastore.Host, d.config.Predastore.Region, d.config.AccessKey, d.config.SecretKey) + d.instanceService = handlers_ec2_instance.NewInstanceServiceImpl(d.config, d.resourceMgr.instanceTypes, d.natsConn, &d.Instances, store) d.keyService = handlers_ec2_key.NewKeyServiceImpl(d.config) d.imageService = handlers_ec2_image.NewImageServiceImpl(d.config) d.volumeService = handlers_ec2_volume.NewVolumeServiceImpl(d.config, d.natsConn) diff --git a/hive/daemon/daemon_test.go b/hive/daemon/daemon_test.go index 109ab9f06..c24efffc5 100644 --- a/hive/daemon/daemon_test.go +++ b/hive/daemon/daemon_test.go @@ -15,6 +15,7 @@ import ( "github.com/aws/aws-sdk-go/service/ec2" "github.com/mulgadc/hive/hive/config" handlers_ec2_instance "github.com/mulgadc/hive/hive/handlers/ec2/instance" + "github.com/mulgadc/hive/hive/objectstore" handlers_ec2_volume "github.com/mulgadc/hive/hive/handlers/ec2/volume" "github.com/mulgadc/hive/hive/qmp" "github.com/mulgadc/hive/hive/vm" @@ -101,7 +102,7 @@ func createTestDaemon(t *testing.T, natsURL string) *Daemon { daemon.natsConn = nc // Initialize services (needed for handler tests) - daemon.instanceService = handlers_ec2_instance.NewInstanceServiceImpl(cfg, daemon.resourceMgr.instanceTypes, nc, &daemon.Instances) + daemon.instanceService = handlers_ec2_instance.NewInstanceServiceImpl(cfg, daemon.resourceMgr.instanceTypes, nc, &daemon.Instances, objectstore.NewMemoryObjectStore()) daemon.volumeService = handlers_ec2_volume.NewVolumeServiceImpl(cfg, nc) t.Cleanup(func() { diff --git a/hive/handlers/ec2/image/service_impl.go b/hive/handlers/ec2/image/service_impl.go index 82ff84808..5882391f6 100644 --- a/hive/handlers/ec2/image/service_impl.go +++ b/hive/handlers/ec2/image/service_impl.go @@ -12,7 +12,7 @@ import ( "github.com/aws/aws-sdk-go/service/s3" "github.com/mulgadc/hive/hive/awserrors" "github.com/mulgadc/hive/hive/config" - "github.com/mulgadc/hive/hive/handlers/ec2/objectstore" + "github.com/mulgadc/hive/hive/objectstore" "github.com/mulgadc/hive/hive/utils" "github.com/mulgadc/viperblock/viperblock" ) diff --git a/hive/handlers/ec2/instance/service_impl.go b/hive/handlers/ec2/instance/service_impl.go index ae242bafa..e5190e84a 100644 --- a/hive/handlers/ec2/instance/service_impl.go +++ b/hive/handlers/ec2/instance/service_impl.go @@ -5,6 +5,7 @@ import ( "crypto/rand" "errors" "fmt" + "io" "log/slog" "math/big" "os" @@ -12,11 +13,13 @@ import ( "text/template" "time" + "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" + awss3 "github.com/aws/aws-sdk-go/service/s3" "github.com/kdomanski/iso9660" "github.com/mulgadc/hive/hive/awserrors" "github.com/mulgadc/hive/hive/config" - "github.com/mulgadc/hive/hive/s3client" + "github.com/mulgadc/hive/hive/objectstore" "github.com/mulgadc/hive/hive/utils" "github.com/mulgadc/hive/hive/vm" "github.com/mulgadc/viperblock/types" @@ -90,15 +93,17 @@ type InstanceServiceImpl struct { instanceTypes map[string]*ec2.InstanceTypeInfo natsConn *nats.Conn instances *vm.Instances + objectStore objectstore.ObjectStore } // NewInstanceServiceImpl creates a new instance service implementation for daemon use -func NewInstanceServiceImpl(cfg *config.Config, instanceTypes map[string]*ec2.InstanceTypeInfo, nc *nats.Conn, instances *vm.Instances) *InstanceServiceImpl { +func NewInstanceServiceImpl(cfg *config.Config, instanceTypes map[string]*ec2.InstanceTypeInfo, nc *nats.Conn, instances *vm.Instances, store objectstore.ObjectStore) *InstanceServiceImpl { return &InstanceServiceImpl{ config: cfg, instanceTypes: instanceTypes, natsConn: nc, instances: instances, + objectStore: store, } } @@ -611,32 +616,28 @@ func (s *InstanceServiceImpl) createCloudInitISO(input *ec2.RunInstancesInput, i hostname := generateHostname(instanceId) // Retrieve SSH pubkey from S3 - s3c := s3client.New(s3client.S3Config{ - AccessKey: s.config.AccessKey, - SecretKey: s.config.SecretKey, - Host: s.config.Predastore.Host, - Bucket: s.config.Predastore.Bucket, - Region: s.config.Predastore.Region, - }) - - err = s3c.Init() - if err != nil { - slog.Error("failed to initialize S3 client", "err", err) - return errors.New(awserrors.ErrorServerInternal) - } - keyName := "" if input.KeyName != nil { keyName = *input.KeyName } // TODO: Mock for account ID, replace with real account ID retrieval - sshKey, err := s3c.Read(fmt.Sprintf("/keys/123456789/%s", keyName)) + keyPath := fmt.Sprintf("/keys/123456789/%s", keyName) + result, err := s.objectStore.GetObject(&awss3.GetObjectInput{ + Bucket: aws.String(s.config.Predastore.Bucket), + Key: aws.String(keyPath), + }) if err != nil { slog.Error("failed to read SSH key", "err", err) return errors.New(awserrors.ErrorServerInternal) } + sshKey, err := io.ReadAll(result.Body) + if err != nil { + slog.Error("failed to read SSH key body", "err", err) + return errors.New(awserrors.ErrorServerInternal) + } + userData := CloudInitData{ Username: "ec2-user", SSHKey: string(sshKey), diff --git a/hive/handlers/ec2/key/service_impl.go b/hive/handlers/ec2/key/service_impl.go index e83d2c54a..050169212 100644 --- a/hive/handlers/ec2/key/service_impl.go +++ b/hive/handlers/ec2/key/service_impl.go @@ -23,7 +23,7 @@ import ( "github.com/aws/aws-sdk-go/service/s3" "github.com/mulgadc/hive/hive/awserrors" "github.com/mulgadc/hive/hive/config" - "github.com/mulgadc/hive/hive/handlers/ec2/objectstore" + "github.com/mulgadc/hive/hive/objectstore" "github.com/mulgadc/hive/hive/utils" ) diff --git a/hive/handlers/ec2/volume/service_impl.go b/hive/handlers/ec2/volume/service_impl.go index 42600c687..9872d5f24 100644 --- a/hive/handlers/ec2/volume/service_impl.go +++ b/hive/handlers/ec2/volume/service_impl.go @@ -18,7 +18,7 @@ import ( "github.com/aws/aws-sdk-go/service/s3" "github.com/mulgadc/hive/hive/awserrors" "github.com/mulgadc/hive/hive/config" - "github.com/mulgadc/hive/hive/handlers/ec2/objectstore" + "github.com/mulgadc/hive/hive/objectstore" "github.com/mulgadc/hive/hive/utils" "github.com/mulgadc/viperblock/viperblock" s3backend "github.com/mulgadc/viperblock/viperblock/backends/s3" diff --git a/hive/handlers/ec2/volume/service_impl_test.go b/hive/handlers/ec2/volume/service_impl_test.go index 1e2f84cbb..b2f6f2487 100644 --- a/hive/handlers/ec2/volume/service_impl_test.go +++ b/hive/handlers/ec2/volume/service_impl_test.go @@ -7,7 +7,7 @@ import ( "github.com/aws/aws-sdk-go/service/ec2" "github.com/mulgadc/hive/hive/awserrors" "github.com/mulgadc/hive/hive/config" - "github.com/mulgadc/hive/hive/handlers/ec2/objectstore" + "github.com/mulgadc/hive/hive/objectstore" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) diff --git a/hive/handlers/ec2/objectstore/objectstore.go b/hive/objectstore/objectstore.go similarity index 100% rename from hive/handlers/ec2/objectstore/objectstore.go rename to hive/objectstore/objectstore.go diff --git a/hive/handlers/ec2/objectstore/objectstore_test.go b/hive/objectstore/objectstore_test.go similarity index 100% rename from hive/handlers/ec2/objectstore/objectstore_test.go rename to hive/objectstore/objectstore_test.go diff --git a/hive/s3client/s3client.go b/hive/s3client/s3client.go deleted file mode 100644 index d97a2e34b..000000000 --- a/hive/s3client/s3client.go +++ /dev/null @@ -1,166 +0,0 @@ -// Copyright 2025 Mulga Defense Corporation (MDC). All rights reserved. -// Use of this source code is governed by an Apache 2.0 license -// that can be found in the LICENSE file. - -package s3client - -import ( - "bytes" - "crypto/tls" - "fmt" - "io" - "log/slog" - "net/http" - "time" - - "github.com/aws/aws-sdk-go/aws" - "github.com/aws/aws-sdk-go/aws/credentials" - "github.com/aws/aws-sdk-go/aws/session" - "github.com/aws/aws-sdk-go/service/s3" - "golang.org/x/net/http2" -) - -// 2. Define config structs -type S3Config struct { - VolumeName string - - Region string - Bucket string - AccessKey string - SecretKey string - - Host string - - S3Client *s3.S3 -} - -type S3Backend struct { - config S3Config -} - -type Backend struct { - S3Backend - Config S3Config -} - -func New(config any) (backend *Backend) { - return &Backend{S3Backend: S3Backend{config: config.(S3Config)}} -} - -func (backend *Backend) Init() error { - - slog.Info("Initializing S3 backend", "config", backend.config) - - // Create HTTP client with HTTP/2 support for connection multiplexing. - // HTTP/2 allows multiple requests over a single TCP connection. - tr := &http.Transport{ - TLSClientConfig: &tls.Config{ - InsecureSkipVerify: true, - ClientSessionCache: tls.NewLRUClientSessionCache(256), - NextProtos: []string{"h2", "http/1.1"}, - }, - MaxIdleConns: 100, - MaxIdleConnsPerHost: 100, - IdleConnTimeout: 120 * time.Second, - ForceAttemptHTTP2: true, - } - - // CRITICAL: Configure HTTP/2 support with custom TLS config - if err := http2.ConfigureTransport(tr); err != nil { - slog.Warn("Failed to configure HTTP/2", "error", err) - } - - client := &http.Client{ - Transport: tr, - Timeout: 120 * time.Second, - } - - // Use the AWS SDK to initialize the S3 backend - sess, err := session.NewSession(&aws.Config{ - // Specify the endpoint - Endpoint: aws.String(backend.config.Host), - S3ForcePathStyle: aws.Bool(true), - Region: aws.String(backend.config.Region), - HTTPClient: client, - Credentials: credentials.NewStaticCredentials(backend.config.AccessKey, backend.config.SecretKey, ""), - }) - - if err != nil { - slog.Error("Error creating session", "error", err) - return err - } - - backend.config.S3Client = s3.New(sess) - - _, err = backend.config.S3Client.ListObjectsV2(&s3.ListObjectsV2Input{ - Bucket: aws.String(backend.config.Bucket), - }) - - if err != nil { - slog.Error("Error listing objects", "error", err) - return err - } - - return nil -} - -func (backend *Backend) Read(object string) (data []byte, err error) { - - slog.Info("[S3 READ] Reading object", "object", object) - - if backend.config.S3Client == nil { - return nil, fmt.Errorf("S3 client not initialized") - } - - // Fetch the object from S3 with a byte range - requestObject := &s3.GetObjectInput{ - Bucket: aws.String(backend.config.Bucket), - Key: aws.String(object), - } - - // TODO: Add ctx support and retry from S3 timeout/500/etc - textResult, err := backend.config.S3Client.GetObject(requestObject) - - if err != nil { - return nil, err - } - - res, err := io.ReadAll(textResult.Body) - - if err != nil { - return nil, err - } - - // Copy the res to the data for the specified length - //copy(data, res) - - return res, nil -} - -func (backend *Backend) Write(filename string, data *[]byte) (err error) { - - if backend.config.S3Client == nil { - return fmt.Errorf("S3 client not initialized") - } - - // Open the specified file - //filename := fmt.Sprintf("%s/chunk.%08d.bin", backend.config.VolumeName, objectId) - - // Create a new S3 object - object := &s3.PutObjectInput{ - Bucket: aws.String(backend.config.Bucket), - Key: aws.String(filename), - Body: bytes.NewReader(*data), - } - - _, err = backend.config.S3Client.PutObject(object) - - //slog.Info("Write object", "output", output) - - if err != nil { - slog.Error("Error writing object", "error", err) - return err - } - - return nil -}