mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-15 17:11:30 +00:00
fix: use object store instead of s3 client
This commit is contained in:
@@ -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)
|
||||
|
||||
@@ -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() {
|
||||
|
||||
@@ -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"
|
||||
)
|
||||
|
||||
@@ -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),
|
||||
|
||||
@@ -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"
|
||||
)
|
||||
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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"
|
||||
)
|
||||
|
||||
@@ -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
|
||||
}
|
||||
Reference in New Issue
Block a user