From 896b70e508b67b147fa378f0bf3e244e2e8e03b8 Mon Sep 17 00:00:00 2001 From: Julian Sommer <115516918+juliansommer@users.noreply.github.com> Date: Tue, 20 Jan 2026 13:07:16 +1100 Subject: [PATCH] feat: describe-volumes --- hive/daemon/daemon.go | 56 ++++ hive/gateway/ec2.go | 24 ++ hive/gateway/ec2/volume/DescribeVolumes.go | 51 ++++ .../ec2/volume/DescribeVolumes_test.go | 102 ++++++++ hive/handlers/ec2/volume/service.go | 8 + hive/handlers/ec2/volume/service_impl.go | 246 ++++++++++++++++++ hive/handlers/ec2/volume/service_nats.go | 51 ++++ 7 files changed, 538 insertions(+) create mode 100644 hive/gateway/ec2/volume/DescribeVolumes.go create mode 100644 hive/gateway/ec2/volume/DescribeVolumes_test.go create mode 100644 hive/handlers/ec2/volume/service.go create mode 100644 hive/handlers/ec2/volume/service_impl.go create mode 100644 hive/handlers/ec2/volume/service_nats.go diff --git a/hive/daemon/daemon.go b/hive/daemon/daemon.go index 8ced8692a..61e8284c4 100644 --- a/hive/daemon/daemon.go +++ b/hive/daemon/daemon.go @@ -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) diff --git a/hive/gateway/ec2.go b/hive/gateway/ec2.go index af9d06e7b..7d513829d 100644 --- a/hive/gateway/ec2.go +++ b/hive/gateway/ec2.go @@ -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") } diff --git a/hive/gateway/ec2/volume/DescribeVolumes.go b/hive/gateway/ec2/volume/DescribeVolumes.go new file mode 100644 index 000000000..66fb44aaf --- /dev/null +++ b/hive/gateway/ec2/volume/DescribeVolumes.go @@ -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 +} diff --git a/hive/gateway/ec2/volume/DescribeVolumes_test.go b/hive/gateway/ec2/volume/DescribeVolumes_test.go new file mode 100644 index 000000000..abff51610 --- /dev/null +++ b/hive/gateway/ec2/volume/DescribeVolumes_test.go @@ -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) + } + }) + } +} diff --git a/hive/handlers/ec2/volume/service.go b/hive/handlers/ec2/volume/service.go new file mode 100644 index 000000000..fb6f62463 --- /dev/null +++ b/hive/handlers/ec2/volume/service.go @@ -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) +} diff --git a/hive/handlers/ec2/volume/service_impl.go b/hive/handlers/ec2/volume/service_impl.go new file mode 100644 index 000000000..c41bd7eb4 --- /dev/null +++ b/hive/handlers/ec2/volume/service_impl.go @@ -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-/ + 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 +} diff --git a/hive/handlers/ec2/volume/service_nats.go b/hive/handlers/ec2/volume/service_nats.go new file mode 100644 index 000000000..65ebf4d95 --- /dev/null +++ b/hive/handlers/ec2/volume/service_nats.go @@ -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 +}