mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-15 17:11:30 +00:00
feat: describe-volumes
This commit is contained in:
Julian Sommer
@@ -30,6 +30,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"
|
||||
handlers_ec2_volume "github.com/mulgadc/hive/hive/handlers/ec2/volume"
|
||||
"github.com/mulgadc/hive/hive/qmp"
|
||||
"github.com/mulgadc/hive/hive/utils"
|
||||
"github.com/mulgadc/hive/hive/vm"
|
||||
@@ -76,6 +77,7 @@ type Daemon struct {
|
||||
instanceService *handlers_ec2_instance.InstanceServiceImpl
|
||||
keyService *handlers_ec2_key.KeyServiceImpl
|
||||
imageService *handlers_ec2_image.ImageServiceImpl
|
||||
volumeService *handlers_ec2_volume.VolumeServiceImpl
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
shutdownWg sync.WaitGroup
|
||||
@@ -461,6 +463,9 @@ func (d *Daemon) Start() error {
|
||||
// Create image service for handling EC2 AMI operations
|
||||
d.imageService = handlers_ec2_image.NewImageServiceImpl(d.config)
|
||||
|
||||
// Create volume service for handling EC2 EBS volume operations
|
||||
d.volumeService = handlers_ec2_volume.NewVolumeServiceImpl(d.config)
|
||||
|
||||
log.Printf("Subscribing to subject pattern: %s", "ec2.launch")
|
||||
|
||||
// Subscribe to EC2 events with queue group (legacy topic for backward compatibility)
|
||||
@@ -526,6 +531,15 @@ func (d *Daemon) Start() error {
|
||||
return fmt.Errorf("failed to subscribe to NATS ec2.DescribeImages: %w", err)
|
||||
}
|
||||
|
||||
log.Printf("Subscribing to subject pattern: %s", "ec2.DescribeVolumes")
|
||||
|
||||
// Subscribe to EC2 DescribeVolumes with queue group
|
||||
d.natsSubscriptions["ec2.DescribeVolumes"], err = d.natsConn.QueueSubscribe("ec2.DescribeVolumes", "hive-workers", d.handleEC2DescribeVolumes)
|
||||
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to subscribe to NATS ec2.DescribeVolumes: %w", err)
|
||||
}
|
||||
|
||||
log.Printf("Subscribing to subject pattern: %s", "ec2.DescribeInstances")
|
||||
|
||||
// Subscribe to EC2 DescribeInstances - no queue group for multi-node fan-out
|
||||
@@ -2095,6 +2109,48 @@ func (d *Daemon) handleEC2DescribeImages(msg *nats.Msg) {
|
||||
slog.Info("handleEC2DescribeImages completed", "count", len(output.Images))
|
||||
}
|
||||
|
||||
// handleEC2DescribeVolumes processes incoming EC2 DescribeVolumes requests
|
||||
func (d *Daemon) handleEC2DescribeVolumes(msg *nats.Msg) {
|
||||
log.Printf("Received message on subject: %s", msg.Subject)
|
||||
log.Printf("Message data: %s", string(msg.Data))
|
||||
|
||||
// Initialize describeVolumesInput before unmarshaling into it
|
||||
describeVolumesInput := &ec2.DescribeVolumesInput{}
|
||||
var errResp []byte
|
||||
|
||||
errResp = utils.UnmarshalJsonPayload(describeVolumesInput, msg.Data)
|
||||
|
||||
if errResp != nil {
|
||||
msg.Respond(errResp)
|
||||
slog.Error("Request does not match DescribeVolumesInput")
|
||||
return
|
||||
}
|
||||
|
||||
slog.Info("Processing DescribeVolumes request")
|
||||
|
||||
// Delegate to volume service for business logic (S3 listing)
|
||||
output, err := d.volumeService.DescribeVolumes(describeVolumesInput)
|
||||
|
||||
if err != nil {
|
||||
slog.Error("handleEC2DescribeVolumes service.DescribeVolumes failed", "err", err)
|
||||
errResp = utils.GenerateErrorPayload(err.Error())
|
||||
msg.Respond(errResp)
|
||||
return
|
||||
}
|
||||
|
||||
// Respond to NATS with DescribeVolumesOutput
|
||||
jsonResponse, err := json.Marshal(output)
|
||||
if err != nil {
|
||||
slog.Error("handleEC2DescribeVolumes failed to marshal output", "err", err)
|
||||
errResp = utils.GenerateErrorPayload(awserrors.ErrorServerInternal)
|
||||
msg.Respond(errResp)
|
||||
return
|
||||
}
|
||||
msg.Respond(jsonResponse)
|
||||
|
||||
slog.Info("handleEC2DescribeVolumes completed", "count", len(output.Volumes))
|
||||
}
|
||||
|
||||
// handleEC2DescribeInstanceTypes processes incoming EC2 DescribeInstanceTypes requests
|
||||
// This handler responds with instance types that can currently be provisioned on this node
|
||||
// based on available resources (CPU and memory not already allocated to running instances)
|
||||
|
||||
@@ -11,6 +11,7 @@ import (
|
||||
gateway_ec2_instance "github.com/mulgadc/hive/hive/gateway/ec2/instance"
|
||||
gateway_ec2_key "github.com/mulgadc/hive/hive/gateway/ec2/key"
|
||||
gateway_ec2_regions "github.com/mulgadc/hive/hive/gateway/ec2/regions"
|
||||
gateway_ec2_volume "github.com/mulgadc/hive/hive/gateway/ec2/volume"
|
||||
"github.com/mulgadc/hive/hive/utils"
|
||||
)
|
||||
|
||||
@@ -301,6 +302,29 @@ func (gw *GatewayConfig) EC2_Request(ctx *fiber.Ctx) error {
|
||||
if err != nil {
|
||||
return errors.New("failed to marshal response to XML")
|
||||
}
|
||||
|
||||
case "DescribeVolumes":
|
||||
var input = &ec2.DescribeVolumesInput{}
|
||||
err = awsec2query.QueryParamsToStruct(queryArgs, input)
|
||||
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
output, err := gateway_ec2_volume.DescribeVolumes(input, gw.NATSConn)
|
||||
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Convert to XML
|
||||
payload := utils.GenerateXMLPayload("DescribeVolumesResponse", output)
|
||||
xmlOutput, err = utils.MarshalToXML(payload)
|
||||
|
||||
if err != nil {
|
||||
return errors.New("failed to marshal response to XML")
|
||||
}
|
||||
|
||||
default:
|
||||
err = errors.New("InvalidAction")
|
||||
}
|
||||
|
||||
@@ -0,0 +1,51 @@
|
||||
package gateway_ec2_volume
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"strings"
|
||||
|
||||
"github.com/aws/aws-sdk-go/service/ec2"
|
||||
handlers_ec2_volume "github.com/mulgadc/hive/hive/handlers/ec2/volume"
|
||||
"github.com/nats-io/nats.go"
|
||||
)
|
||||
|
||||
// ValidateDescribeVolumesInput validates the input parameters
|
||||
func ValidateDescribeVolumesInput(input *ec2.DescribeVolumesInput) error {
|
||||
if input == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Validate VolumeId format if provided
|
||||
if input.VolumeIds != nil {
|
||||
for _, volumeId := range input.VolumeIds {
|
||||
if volumeId != nil && !strings.HasPrefix(*volumeId, "vol-") {
|
||||
return errors.New("InvalidVolume.Malformed")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// DescribeVolumes handles the DescribeVolumes API call
|
||||
func DescribeVolumes(input *ec2.DescribeVolumesInput, natsConn *nats.Conn) (ec2.DescribeVolumesOutput, error) {
|
||||
var output ec2.DescribeVolumesOutput
|
||||
|
||||
// Validate input
|
||||
err := ValidateDescribeVolumesInput(input)
|
||||
if err != nil {
|
||||
return output, err
|
||||
}
|
||||
|
||||
// Create NATS service and call handler
|
||||
volumeService := handlers_ec2_volume.NewNATSVolumeService(natsConn)
|
||||
result, err := volumeService.DescribeVolumes(input)
|
||||
|
||||
if err != nil {
|
||||
return output, err
|
||||
}
|
||||
|
||||
// Return result
|
||||
output = *result
|
||||
return output, nil
|
||||
}
|
||||
@@ -0,0 +1,102 @@
|
||||
package gateway_ec2_volume
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/aws/aws-sdk-go/aws"
|
||||
"github.com/aws/aws-sdk-go/service/ec2"
|
||||
"github.com/stretchr/testify/assert"
|
||||
)
|
||||
|
||||
func TestValidateDescribeVolumesInput(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
input *ec2.DescribeVolumesInput
|
||||
wantErr bool
|
||||
errMsg string
|
||||
}{
|
||||
{
|
||||
name: "NilInput",
|
||||
input: nil,
|
||||
wantErr: false,
|
||||
},
|
||||
{
|
||||
name: "EmptyInput",
|
||||
input: &ec2.DescribeVolumesInput{},
|
||||
wantErr: false,
|
||||
},
|
||||
{
|
||||
name: "ValidSingleVolumeId",
|
||||
input: &ec2.DescribeVolumesInput{
|
||||
VolumeIds: []*string{aws.String("vol-0123456789abcdef0")},
|
||||
},
|
||||
wantErr: false,
|
||||
},
|
||||
{
|
||||
name: "ValidMultipleVolumeIds",
|
||||
input: &ec2.DescribeVolumesInput{
|
||||
VolumeIds: []*string{
|
||||
aws.String("vol-abc123"),
|
||||
aws.String("vol-def456"),
|
||||
aws.String("vol-ghi789"),
|
||||
},
|
||||
},
|
||||
wantErr: false,
|
||||
},
|
||||
{
|
||||
name: "InvalidVolumeId_NoPrefix",
|
||||
input: &ec2.DescribeVolumesInput{
|
||||
VolumeIds: []*string{aws.String("invalid-id")},
|
||||
},
|
||||
wantErr: true,
|
||||
errMsg: "InvalidVolume.Malformed",
|
||||
},
|
||||
{
|
||||
name: "InvalidVolumeId_WrongPrefix",
|
||||
input: &ec2.DescribeVolumesInput{
|
||||
VolumeIds: []*string{aws.String("ami-0123456789abcdef0")},
|
||||
},
|
||||
wantErr: true,
|
||||
errMsg: "InvalidVolume.Malformed",
|
||||
},
|
||||
{
|
||||
name: "InvalidVolumeId_EmptyString",
|
||||
input: &ec2.DescribeVolumesInput{
|
||||
VolumeIds: []*string{aws.String("")},
|
||||
},
|
||||
wantErr: true,
|
||||
errMsg: "InvalidVolume.Malformed",
|
||||
},
|
||||
{
|
||||
name: "MixedValidAndInvalid",
|
||||
input: &ec2.DescribeVolumesInput{
|
||||
VolumeIds: []*string{
|
||||
aws.String("vol-valid123"),
|
||||
aws.String("invalid-id"),
|
||||
},
|
||||
},
|
||||
wantErr: true,
|
||||
errMsg: "InvalidVolume.Malformed",
|
||||
},
|
||||
{
|
||||
name: "NilVolumeIdInList",
|
||||
input: &ec2.DescribeVolumesInput{
|
||||
VolumeIds: []*string{nil, aws.String("vol-abc123")},
|
||||
},
|
||||
wantErr: false,
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
err := ValidateDescribeVolumesInput(tt.input)
|
||||
|
||||
if tt.wantErr {
|
||||
assert.Error(t, err)
|
||||
assert.Equal(t, tt.errMsg, err.Error())
|
||||
} else {
|
||||
assert.NoError(t, err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,8 @@
|
||||
package handlers_ec2_volume
|
||||
|
||||
import "github.com/aws/aws-sdk-go/service/ec2"
|
||||
|
||||
// VolumeService defines the interface for EBS volume operations
|
||||
type VolumeService interface {
|
||||
DescribeVolumes(input *ec2.DescribeVolumesInput) (*ec2.DescribeVolumesOutput, error)
|
||||
}
|
||||
@@ -0,0 +1,246 @@
|
||||
package handlers_ec2_volume
|
||||
|
||||
import (
|
||||
"crypto/tls"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"io"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
"github.com/aws/aws-sdk-go/aws"
|
||||
"github.com/aws/aws-sdk-go/aws/awserr"
|
||||
"github.com/aws/aws-sdk-go/aws/credentials"
|
||||
"github.com/aws/aws-sdk-go/aws/session"
|
||||
"github.com/aws/aws-sdk-go/service/ec2"
|
||||
"github.com/aws/aws-sdk-go/service/s3"
|
||||
"github.com/mulgadc/hive/hive/awserrors"
|
||||
"github.com/mulgadc/hive/hive/config"
|
||||
"github.com/mulgadc/viperblock/viperblock"
|
||||
)
|
||||
|
||||
// VolumeServiceImpl handles EBS volume operations with S3 storage
|
||||
type VolumeServiceImpl struct {
|
||||
config *config.Config
|
||||
s3Client *s3.S3
|
||||
}
|
||||
|
||||
// NewVolumeServiceImpl creates a new daemon-side volume service
|
||||
func NewVolumeServiceImpl(cfg *config.Config) *VolumeServiceImpl {
|
||||
// Create HTTP client with TLS verification disabled for local S3-compatible endpoints
|
||||
httpClient := &http.Client{
|
||||
Transport: &http.Transport{
|
||||
TLSClientConfig: &tls.Config{
|
||||
InsecureSkipVerify: true,
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
// Create AWS SDK S3 client configured for Predastore endpoint
|
||||
sess := session.Must(session.NewSession(&aws.Config{
|
||||
Endpoint: aws.String(cfg.Predastore.Host),
|
||||
Region: aws.String(cfg.Predastore.Region),
|
||||
Credentials: credentials.NewStaticCredentials(cfg.Predastore.AccessKey, cfg.Predastore.SecretKey, ""),
|
||||
S3ForcePathStyle: aws.Bool(true),
|
||||
HTTPClient: httpClient,
|
||||
}))
|
||||
|
||||
s3Client := s3.New(sess)
|
||||
|
||||
return &VolumeServiceImpl{
|
||||
config: cfg,
|
||||
s3Client: s3Client,
|
||||
}
|
||||
}
|
||||
|
||||
// DescribeVolumes lists EBS volumes by reading config.json files from S3
|
||||
func (s *VolumeServiceImpl) DescribeVolumes(input *ec2.DescribeVolumesInput) (*ec2.DescribeVolumesOutput, error) {
|
||||
if input == nil {
|
||||
input = &ec2.DescribeVolumesInput{}
|
||||
}
|
||||
|
||||
slog.Info("Describing volumes", "volumeIds", input.VolumeIds)
|
||||
|
||||
var volumes []*ec2.Volume
|
||||
|
||||
// Fast path: if specific volume IDs are requested, fetch them directly
|
||||
if len(input.VolumeIds) > 0 {
|
||||
volumes = s.fetchVolumesByIDs(input.VolumeIds)
|
||||
slog.Info("DescribeVolumes completed", "count", len(volumes))
|
||||
return &ec2.DescribeVolumesOutput{Volumes: volumes}, nil
|
||||
}
|
||||
|
||||
// Slow path: list all volumes (no filter provided)
|
||||
result, err := s.s3Client.ListObjects(&s3.ListObjectsInput{
|
||||
Bucket: aws.String(s.config.Predastore.Bucket),
|
||||
Delimiter: aws.String("/"),
|
||||
})
|
||||
|
||||
if err != nil {
|
||||
slog.Error("Failed to list S3 objects", "err", err)
|
||||
return nil, errors.New(awserrors.ErrorServerInternal)
|
||||
}
|
||||
|
||||
// Iterate over CommonPrefixes to find vol-* directories
|
||||
for _, prefix := range result.CommonPrefixes {
|
||||
if prefix.Prefix == nil {
|
||||
continue
|
||||
}
|
||||
|
||||
prefixStr := *prefix.Prefix
|
||||
slog.Debug("Processing S3 prefix", "prefix", prefixStr)
|
||||
|
||||
// Only check prefixes that match pattern: vol-<id>/
|
||||
if !strings.HasPrefix(prefixStr, "vol-") {
|
||||
continue
|
||||
}
|
||||
|
||||
// Extract volume ID from prefix (remove trailing slash)
|
||||
volumeID := strings.TrimSuffix(prefixStr, "/")
|
||||
|
||||
vol, err := s.getVolumeByID(volumeID)
|
||||
if err != nil {
|
||||
slog.Debug("Failed to get volume", "volumeId", volumeID, "err", err)
|
||||
continue
|
||||
}
|
||||
|
||||
volumes = append(volumes, vol)
|
||||
}
|
||||
|
||||
slog.Info("DescribeVolumes completed", "count", len(volumes))
|
||||
|
||||
return &ec2.DescribeVolumesOutput{
|
||||
Volumes: volumes,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// fetchVolumesByIDs fetches multiple volumes in parallel by their IDs
|
||||
func (s *VolumeServiceImpl) fetchVolumesByIDs(volumeIDs []*string) []*ec2.Volume {
|
||||
var (
|
||||
volumes []*ec2.Volume
|
||||
mu sync.Mutex
|
||||
wg sync.WaitGroup
|
||||
)
|
||||
|
||||
for _, volumeID := range volumeIDs {
|
||||
if volumeID == nil {
|
||||
continue
|
||||
}
|
||||
|
||||
wg.Add(1)
|
||||
go func(volID string) {
|
||||
defer wg.Done()
|
||||
|
||||
vol, err := s.getVolumeByID(volID)
|
||||
if err != nil {
|
||||
slog.Debug("Volume not found", "volumeId", volID, "err", err)
|
||||
return
|
||||
}
|
||||
|
||||
mu.Lock()
|
||||
volumes = append(volumes, vol)
|
||||
mu.Unlock()
|
||||
}(*volumeID)
|
||||
}
|
||||
|
||||
wg.Wait()
|
||||
return volumes
|
||||
}
|
||||
|
||||
// getVolumeByID fetches a single volume's config from S3 and builds an EC2 Volume
|
||||
func (s *VolumeServiceImpl) getVolumeByID(volumeID string) (*ec2.Volume, error) {
|
||||
configKey := volumeID + "/config.json"
|
||||
|
||||
getResult, err := s.s3Client.GetObject(&s3.GetObjectInput{
|
||||
Bucket: aws.String(s.config.Predastore.Bucket),
|
||||
Key: aws.String(configKey),
|
||||
})
|
||||
if err != nil {
|
||||
if aerr, ok := err.(awserr.Error); ok && (aerr.Code() == s3.ErrCodeNoSuchKey || aerr.Code() == "NotFound") {
|
||||
slog.Debug("Config file not found", "key", configKey)
|
||||
return nil, errors.New("volume not found")
|
||||
}
|
||||
slog.Debug("Failed to get config file", "key", configKey, "err", err)
|
||||
return nil, err
|
||||
}
|
||||
defer getResult.Body.Close()
|
||||
|
||||
body, err := io.ReadAll(getResult.Body)
|
||||
if err != nil {
|
||||
slog.Debug("Failed to read config body", "key", configKey, "err", err)
|
||||
return nil, err
|
||||
}
|
||||
|
||||
var vbConfig struct {
|
||||
VolumeConfig viperblock.VolumeConfig `json:"VolumeConfig"`
|
||||
}
|
||||
if err := json.Unmarshal(body, &vbConfig); err != nil {
|
||||
slog.Debug("Failed to unmarshal config", "key", configKey, "err", err)
|
||||
return nil, err
|
||||
}
|
||||
|
||||
volMeta := vbConfig.VolumeConfig.VolumeMetadata
|
||||
|
||||
if volMeta.VolumeID == "" {
|
||||
slog.Debug("Volume ID is empty in config", "key", configKey)
|
||||
return nil, errors.New("volume ID is empty")
|
||||
}
|
||||
|
||||
state := volMeta.State
|
||||
if state == "" {
|
||||
state = "available"
|
||||
}
|
||||
volumeType := volMeta.VolumeType
|
||||
if volumeType == "" {
|
||||
volumeType = "gp3"
|
||||
}
|
||||
|
||||
volume := &ec2.Volume{
|
||||
VolumeId: aws.String(volMeta.VolumeID),
|
||||
Size: aws.Int64(int64(volMeta.SizeGiB)),
|
||||
State: aws.String(state),
|
||||
AvailabilityZone: aws.String(volMeta.AvailabilityZone),
|
||||
CreateTime: aws.Time(volMeta.CreatedAt),
|
||||
VolumeType: aws.String(volumeType),
|
||||
Encrypted: aws.Bool(volMeta.IsEncrypted),
|
||||
}
|
||||
|
||||
if volMeta.IOPS > 0 {
|
||||
volume.Iops = aws.Int64(int64(volMeta.IOPS))
|
||||
}
|
||||
|
||||
if volMeta.SnapshotID != "" {
|
||||
volume.SnapshotId = aws.String(volMeta.SnapshotID)
|
||||
}
|
||||
|
||||
if volMeta.AttachedInstance != "" {
|
||||
attachState := "attached"
|
||||
if volMeta.State != "in-use" {
|
||||
attachState = "detached"
|
||||
}
|
||||
volume.Attachments = []*ec2.VolumeAttachment{
|
||||
{
|
||||
VolumeId: aws.String(volMeta.VolumeID),
|
||||
InstanceId: aws.String(volMeta.AttachedInstance),
|
||||
Device: aws.String(volMeta.DeviceName),
|
||||
State: aws.String(attachState),
|
||||
DeleteOnTermination: aws.Bool(true),
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
if len(volMeta.Tags) > 0 {
|
||||
tags := make([]*ec2.Tag, 0, len(volMeta.Tags))
|
||||
for key, value := range volMeta.Tags {
|
||||
tags = append(tags, &ec2.Tag{
|
||||
Key: aws.String(key),
|
||||
Value: aws.String(value),
|
||||
})
|
||||
}
|
||||
volume.Tags = tags
|
||||
}
|
||||
|
||||
return volume, nil
|
||||
}
|
||||
@@ -0,0 +1,51 @@
|
||||
package handlers_ec2_volume
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/aws/aws-sdk-go/service/ec2"
|
||||
"github.com/mulgadc/hive/hive/utils"
|
||||
"github.com/nats-io/nats.go"
|
||||
)
|
||||
|
||||
// NATSVolumeService handles volume operations via NATS messaging
|
||||
type NATSVolumeService struct {
|
||||
natsConn *nats.Conn
|
||||
}
|
||||
|
||||
// NewNATSVolumeService creates a new NATS-based volume service
|
||||
func NewNATSVolumeService(conn *nats.Conn) VolumeService {
|
||||
return &NATSVolumeService{natsConn: conn}
|
||||
}
|
||||
|
||||
// DescribeVolumes sends a request via NATS and waits for response
|
||||
func (s *NATSVolumeService) DescribeVolumes(input *ec2.DescribeVolumesInput) (*ec2.DescribeVolumesOutput, error) {
|
||||
// Marshal input to JSON
|
||||
jsonData, err := json.Marshal(input)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to marshal input: %w", err)
|
||||
}
|
||||
|
||||
// Send NATS request with 30 second timeout
|
||||
msg, err := s.natsConn.Request("ec2.DescribeVolumes", jsonData, 30*time.Second)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("NATS request failed: %w", err)
|
||||
}
|
||||
|
||||
// Validate error response
|
||||
responseError, err := utils.ValidateErrorPayload(msg.Data)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("daemon returned error: %s", *responseError.Code)
|
||||
}
|
||||
|
||||
// Unmarshal successful response
|
||||
var output ec2.DescribeVolumesOutput
|
||||
err = json.Unmarshal(msg.Data, &output)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to unmarshal response: %w", err)
|
||||
}
|
||||
|
||||
return &output, nil
|
||||
}
|
||||
Reference in New Issue
Block a user