diff --git a/.claude/settings.json b/.claude/settings.json index 05ae344e6..0f7c50942 100644 --- a/.claude/settings.json +++ b/.claude/settings.json @@ -1,16 +1,4 @@ { - // "hooks": { - // "Stop": [ - // { - // "hooks": [ - // { - // "type": "command", - // "command": "make test" - // } - // ] - // } - // ] - // }, "permissions": { "allow": [ "Bash(make build)", @@ -27,10 +15,10 @@ "Bash(ls*)", "Bash(ls -la*)", "Bash(ls -l*)", + "Bash(ll*)", "Bash(ls -a*)", "Bash(ls -lh*)", "Bash(ls -la */)", - "Bash(ll*)", "Bash(cat *)", "Bash(cat -n *)", "Bash(less *)", @@ -53,6 +41,13 @@ "Bash(find . -name *)", "Bash(find . -type *)", "Bash(find . -path *)", + "Bash(diff *)", + "Bash(tree *)", + "Bash(pwd*)", + "Bash(cd *)", + "Bash(cd * && *)", + "Bash(cd ../viperblock && *)", + "Bash(cd ../predastore && *)", "Bash(pnpm fix)", "Bash(pnpm check)", diff --git a/TODO.md b/TODO.md index 908d477dd..60fbce8a3 100644 --- a/TODO.md +++ b/TODO.md @@ -28,7 +28,9 @@ - placement group instances, allow spreading instance group between nodes. - capabilities flag - more instance types - per cpu gen -- multi part uploads on frontend +- [DONE] multi part uploads on frontend +- delete volumes on termination +- fix describe instances ## EC2 create-volume diff --git a/hive/config/config.go b/hive/config/config.go index 0922c3128..35cd67bcc 100644 --- a/hive/config/config.go +++ b/hive/config/config.go @@ -119,6 +119,16 @@ type EBSUnMountResponse struct { Error string } +type EBSSyncRequest struct { + Volume string +} + +type EBSSyncResponse struct { + Volume string + Synced bool + Error string +} + type EBSDeleteRequest struct { Volume string } diff --git a/hive/daemon/daemon.go b/hive/daemon/daemon.go index 33143f776..cce7a5a48 100644 --- a/hive/daemon/daemon.go +++ b/hive/daemon/daemon.go @@ -627,6 +627,15 @@ func (d *Daemon) Start() error { return fmt.Errorf("failed to subscribe to NATS ec2.DescribeVolumes: %w", err) } + slog.Info("Subscribing to subject pattern", "subject", "ec2.ModifyVolume") + + // Subscribe to EC2 ModifyVolume with queue group + d.natsSubscriptions["ec2.ModifyVolume"], err = d.natsConn.QueueSubscribe("ec2.ModifyVolume", "hive-workers", d.handleEC2ModifyVolume) + + if err != nil { + return fmt.Errorf("failed to subscribe to NATS ec2.ModifyVolume: %w", err) + } + log.Printf("Subscribing to subject pattern: %s", "ec2.DescribeInstances") // Subscribe to EC2 DescribeInstances - no queue group for multi-node fan-out @@ -1601,6 +1610,13 @@ func (d *Daemon) stopInstance(instances map[string]*vm.VM, deleteVolume bool) er } else { slog.Info("Unmounted Viperblock volume", "id", instance.ID, "data", string(msg.Data)) } + + // Update volume state to "available" for boot volumes + if ebsRequest.Boot { + if err := d.volumeService.UpdateVolumeState(ebsRequest.Name, "available", ""); err != nil { + slog.Error("Failed to update volume state to available", "volumeId", ebsRequest.Name, "err", err) + } + } } // If flagged for termination (delete Volume) @@ -1835,6 +1851,17 @@ func (d *Daemon) LaunchInstance(instance *vm.VM) (err error) { d.Instances.VMS[instance.ID] = instance d.Instances.Mu.Unlock() + // Step 10: Mark boot volumes as "in-use" now that instance is confirmed running + instance.EBSRequests.Mu.Lock() + for _, ebsReq := range instance.EBSRequests.Requests { + if ebsReq.Boot { + if err := d.volumeService.UpdateVolumeState(ebsReq.Name, "in-use", instance.ID); err != nil { + slog.Error("Failed to update volume state to in-use", "volumeId", ebsReq.Name, "err", err) + } + } + } + instance.EBSRequests.Mu.Unlock() + err = d.WriteState() if err != nil { @@ -2458,6 +2485,59 @@ func (d *Daemon) handleEC2DescribeVolumes(msg *nats.Msg) { slog.Info("handleEC2DescribeVolumes completed", "count", len(output.Volumes)) } +// handleEC2ModifyVolume processes incoming EC2 ModifyVolume requests +func (d *Daemon) handleEC2ModifyVolume(msg *nats.Msg) { + slog.Debug("Received message", "subject", msg.Subject) + slog.Debug("Message data", "data", string(msg.Data)) + + modifyVolumeInput := &ec2.ModifyVolumeInput{} + var errResp []byte + + errResp = utils.UnmarshalJsonPayload(modifyVolumeInput, msg.Data) + + if errResp != nil { + msg.Respond(errResp) + slog.Error("Request does not match ModifyVolumeInput") + return + } + + slog.Info("Processing ModifyVolume request", "volumeId", modifyVolumeInput.VolumeId) + + output, err := d.volumeService.ModifyVolume(modifyVolumeInput) + + if err != nil { + slog.Error("handleEC2ModifyVolume service.ModifyVolume failed", "err", err) + errResp = utils.GenerateErrorPayload(err.Error()) + msg.Respond(errResp) + return + } + + jsonResponse, err := json.Marshal(output) + if err != nil { + slog.Error("handleEC2ModifyVolume failed to marshal output", "err", err) + errResp = utils.GenerateErrorPayload(awserrors.ErrorServerInternal) + msg.Respond(errResp) + return + } + msg.Respond(jsonResponse) + + // Notify viperblockd to reload state after volume modification (e.g. resize) + if modifyVolumeInput.VolumeId != nil { + syncData, err := json.Marshal(config.EBSSyncRequest{Volume: *modifyVolumeInput.VolumeId}) + if err != nil { + slog.Error("failed to marshal ebs.sync request", "volumeId", *modifyVolumeInput.VolumeId, "err", err) + } else { + _, syncErr := d.natsConn.Request("ebs.sync", syncData, 5*time.Second) + if syncErr != nil { + slog.Warn("ebs.sync notification failed (volume may not be mounted)", + "volumeId", *modifyVolumeInput.VolumeId, "err", syncErr) + } + } + } + + slog.Info("handleEC2ModifyVolume completed", "volumeId", modifyVolumeInput.VolumeId) +} + // 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 2b61191e5..6d9dfa38b 100644 --- a/hive/gateway/ec2.go +++ b/hive/gateway/ec2.go @@ -327,6 +327,28 @@ func (gw *GatewayConfig) EC2_Request(ctx *fiber.Ctx) error { return errors.New("failed to marshal response to XML") } + case "ModifyVolume": + var input = &ec2.ModifyVolumeInput{} + err = awsec2query.QueryParamsToStruct(queryArgs, input) + + if err != nil { + return err + } + + output, err := gateway_ec2_volume.ModifyVolume(input, gw.NATSConn) + + if err != nil { + return err + } + + // Convert to XML + payload := utils.GenerateXMLPayload("ModifyVolumeResponse", 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/ModifyVolume.go b/hive/gateway/ec2/volume/ModifyVolume.go new file mode 100644 index 000000000..518d3f5aa --- /dev/null +++ b/hive/gateway/ec2/volume/ModifyVolume.go @@ -0,0 +1,47 @@ +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" +) + +// ValidateModifyVolumeInput validates the input parameters +func ValidateModifyVolumeInput(input *ec2.ModifyVolumeInput) error { + if input == nil { + return errors.New("InvalidParameterValue") + } + + if input.VolumeId == nil || !strings.HasPrefix(*input.VolumeId, "vol-") { + return errors.New("InvalidVolumeID.Malformed") + } + + if input.Size != nil && *input.Size <= 0 { + return errors.New("InvalidParameterValue") + } + + return nil +} + +// ModifyVolume handles the ModifyVolume API call +func ModifyVolume(input *ec2.ModifyVolumeInput, natsConn *nats.Conn) (ec2.ModifyVolumeOutput, error) { + var output ec2.ModifyVolumeOutput + + err := ValidateModifyVolumeInput(input) + if err != nil { + return output, err + } + + volumeService := handlers_ec2_volume.NewNATSVolumeService(natsConn) + result, err := volumeService.ModifyVolume(input) + + if err != nil { + return output, err + } + + output = *result + return output, nil +} diff --git a/hive/gateway/ec2/volume/ModifyVolume_test.go b/hive/gateway/ec2/volume/ModifyVolume_test.go new file mode 100644 index 000000000..2485b59b4 --- /dev/null +++ b/hive/gateway/ec2/volume/ModifyVolume_test.go @@ -0,0 +1,131 @@ +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 TestValidateModifyVolumeInput(t *testing.T) { + tests := []struct { + name string + input *ec2.ModifyVolumeInput + wantErr bool + errMsg string + }{ + { + name: "NilInput", + input: nil, + wantErr: true, + errMsg: "InvalidParameterValue", + }, + { + name: "EmptyInput_NoVolumeId", + input: &ec2.ModifyVolumeInput{}, + wantErr: true, + errMsg: "InvalidVolumeID.Malformed", + }, + { + name: "ValidVolumeId_WithSize", + input: &ec2.ModifyVolumeInput{ + VolumeId: aws.String("vol-0123456789abcdef0"), + Size: aws.Int64(64), + }, + wantErr: false, + }, + { + name: "ValidVolumeId_NoSize", + input: &ec2.ModifyVolumeInput{ + VolumeId: aws.String("vol-abc123"), + }, + wantErr: false, + }, + { + name: "InvalidVolumeId_NoPrefix", + input: &ec2.ModifyVolumeInput{ + VolumeId: aws.String("invalid-id"), + Size: aws.Int64(64), + }, + wantErr: true, + errMsg: "InvalidVolumeID.Malformed", + }, + { + name: "InvalidVolumeId_WrongPrefix", + input: &ec2.ModifyVolumeInput{ + VolumeId: aws.String("ami-0123456789abcdef0"), + Size: aws.Int64(64), + }, + wantErr: true, + errMsg: "InvalidVolumeID.Malformed", + }, + { + name: "InvalidVolumeId_EmptyString", + input: &ec2.ModifyVolumeInput{ + VolumeId: aws.String(""), + Size: aws.Int64(64), + }, + wantErr: true, + errMsg: "InvalidVolumeID.Malformed", + }, + { + name: "InvalidVolumeId_Nil", + input: &ec2.ModifyVolumeInput{ + VolumeId: nil, + Size: aws.Int64(64), + }, + wantErr: true, + errMsg: "InvalidVolumeID.Malformed", + }, + { + name: "InvalidSize_Zero", + input: &ec2.ModifyVolumeInput{ + VolumeId: aws.String("vol-abc123"), + Size: aws.Int64(0), + }, + wantErr: true, + errMsg: "InvalidParameterValue", + }, + { + name: "InvalidSize_Negative", + input: &ec2.ModifyVolumeInput{ + VolumeId: aws.String("vol-abc123"), + Size: aws.Int64(-10), + }, + wantErr: true, + errMsg: "InvalidParameterValue", + }, + { + name: "ValidWithVolumeType", + input: &ec2.ModifyVolumeInput{ + VolumeId: aws.String("vol-abc123"), + Size: aws.Int64(100), + VolumeType: aws.String("gp3"), + }, + wantErr: false, + }, + { + name: "ValidWithIops", + input: &ec2.ModifyVolumeInput{ + VolumeId: aws.String("vol-abc123"), + Size: aws.Int64(100), + Iops: aws.Int64(3000), + }, + wantErr: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := ValidateModifyVolumeInput(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 index fb6f62463..210ba3b82 100644 --- a/hive/handlers/ec2/volume/service.go +++ b/hive/handlers/ec2/volume/service.go @@ -5,4 +5,5 @@ 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) + ModifyVolume(input *ec2.ModifyVolumeInput) (*ec2.ModifyVolumeOutput, error) } diff --git a/hive/handlers/ec2/volume/service_impl.go b/hive/handlers/ec2/volume/service_impl.go index c5b4e2003..76323ad0e 100644 --- a/hive/handlers/ec2/volume/service_impl.go +++ b/hive/handlers/ec2/volume/service_impl.go @@ -1,14 +1,17 @@ package handlers_ec2_volume import ( + "bytes" "crypto/tls" "encoding/json" "errors" + "fmt" "io" "log/slog" "net/http" "strings" "sync" + "time" "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/aws/awserr" @@ -111,7 +114,7 @@ func (s *VolumeServiceImpl) DescribeVolumes(input *ec2.DescribeVolumesInput) (*e vol, err := s.getVolumeByID(volumeID) if err != nil { - slog.Debug("Failed to get volume", "volumeId", volumeID, "err", err) + slog.Error("Failed to get volume", "volumeId", volumeID, "err", err) continue } @@ -160,40 +163,15 @@ func (s *VolumeServiceImpl) fetchVolumesByIDs(volumeIDs []*string) []*ec2.Volume // 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), - }) + cfg, err := s.getVolumeConfig(volumeID) 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 + volMeta := cfg.VolumeMetadata if volMeta.VolumeID == "" { - slog.Debug("Volume ID is empty in config", "key", configKey) + slog.Debug("Volume ID is empty in config", "key", volumeID+"/config.json") return nil, errors.New("volume ID is empty") } @@ -253,3 +231,202 @@ func (s *VolumeServiceImpl) getVolumeByID(volumeID string) (*ec2.Volume, error) return volume, nil } + +// volumeConfigWrapper matches the JSON structure stored in S3 config.json files +type volumeConfigWrapper struct { + VolumeConfig viperblock.VolumeConfig `json:"VolumeConfig"` +} + +// getVolumeConfig reads the raw VolumeConfig from S3 for a given volume ID +func (s *VolumeServiceImpl) getVolumeConfig(volumeID string) (*viperblock.VolumeConfig, 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") { + return nil, errors.New(awserrors.ErrorInvalidVolumeNotFound) + } + return nil, fmt.Errorf("failed to get config: %w", err) + } + defer getResult.Body.Close() + + body, err := io.ReadAll(getResult.Body) + if err != nil { + return nil, fmt.Errorf("failed to read config body: %w", err) + } + + var wrapper volumeConfigWrapper + if err := json.Unmarshal(body, &wrapper); err != nil { + return nil, fmt.Errorf("failed to unmarshal config: %w", err) + } + + return &wrapper.VolumeConfig, nil +} + +// putVolumeConfig writes a VolumeConfig back to S3 as config.json. +// It performs a read-modify-write to preserve full VBState if viperblock +// has already written state (BlockSize, SeqNum, WALNum, etc.) to config.json. +func (s *VolumeServiceImpl) putVolumeConfig(volumeID string, cfg *viperblock.VolumeConfig) error { + configKey := volumeID + "/config.json" + + data, err := s.mergeVolumeConfig(configKey, cfg) + if err != nil { + return err + } + + _, err = s.s3Client.PutObject(&s3.PutObjectInput{ + Bucket: aws.String(s.config.Predastore.Bucket), + Key: aws.String(configKey), + Body: bytes.NewReader(data), + }) + if err != nil { + return fmt.Errorf("failed to write config to S3: %w", err) + } + + return nil +} + +// mergeVolumeConfig reads existing config.json from S3 and merges the new +// VolumeConfig into it, preserving full VBState when present. If no existing +// VBState is found, it returns a plain volumeConfigWrapper. +func (s *VolumeServiceImpl) mergeVolumeConfig(configKey string, cfg *viperblock.VolumeConfig) ([]byte, error) { + 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") { + // No existing config.json -- write wrapper for new volume + return json.Marshal(volumeConfigWrapper{VolumeConfig: *cfg}) + } + return nil, fmt.Errorf("failed to read existing config for merge: %w", err) + } + defer getResult.Body.Close() + + body, err := io.ReadAll(getResult.Body) + if err != nil { + return nil, fmt.Errorf("failed to read existing config: %w", err) + } + + var state viperblock.VBState + if json.Unmarshal(body, &state) != nil || state.BlockSize == 0 { + // Not a full VBState (new volume or wrapper-only) -- write wrapper + return json.Marshal(volumeConfigWrapper{VolumeConfig: *cfg}) + } + + // Full VBState exists -- update VolumeConfig and reconcile VolumeSize + state.VolumeConfig = *cfg + configSizeBytes := uint64(cfg.VolumeMetadata.SizeGiB) * 1024 * 1024 * 1024 + if configSizeBytes > 0 && configSizeBytes > state.VolumeSize { + state.VolumeSize = configSizeBytes + } + + slog.Info("putVolumeConfig: preserved VBState", "volumeId", strings.TrimSuffix(configKey, "/config.json"), + "blockSize", state.BlockSize, "seqNum", state.SeqNum) + + return json.Marshal(state) +} + +// UpdateVolumeState updates the State and AttachedInstance fields for a volume in S3. +func (s *VolumeServiceImpl) UpdateVolumeState(volumeID, state, attachedInstance string) error { + cfg, err := s.getVolumeConfig(volumeID) + if err != nil { + return fmt.Errorf("failed to get volume config for state update: %w", err) + } + + cfg.VolumeMetadata.State = state + cfg.VolumeMetadata.AttachedInstance = attachedInstance + + if err := s.putVolumeConfig(volumeID, cfg); err != nil { + return fmt.Errorf("failed to write volume config for state update: %w", err) + } + + slog.Info("Updated volume state", "volumeId", volumeID, "state", state, "attachedInstance", attachedInstance) + return nil +} + +// ModifyVolume modifies an EBS volume (grow-only, requires stopped instance) +func (s *VolumeServiceImpl) ModifyVolume(input *ec2.ModifyVolumeInput) (*ec2.ModifyVolumeOutput, error) { + if input.VolumeId == nil || *input.VolumeId == "" { + return nil, errors.New(awserrors.ErrorInvalidVolumeIDMalformed) + } + + volumeID := *input.VolumeId + slog.Info("ModifyVolume request", "volumeId", volumeID) + + cfg, err := s.getVolumeConfig(volumeID) + if err != nil { + slog.Error("ModifyVolume failed to get volume config", "volumeId", volumeID, "err", err) + return nil, err + } + + volMeta := &cfg.VolumeMetadata + + // Record original values before modification + originalSize := utils.SafeUint64ToInt64(volMeta.SizeGiB) + originalType := volMeta.VolumeType + if originalType == "" { + originalType = "gp3" + } + originalIOPS := int64(volMeta.IOPS) + + // Validate: grow only (new size must be greater than current) + if input.Size != nil && *input.Size <= originalSize { + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + + // Validate: if volume is attached, instance must not be in-use (must be stopped) + if volMeta.AttachedInstance != "" && volMeta.State == "in-use" { + return nil, errors.New(awserrors.ErrorIncorrectState) + } + + // Apply modifications + if input.Size != nil { + volMeta.SizeGiB = utils.SafeInt64ToUint64(*input.Size) + } + if input.VolumeType != nil { + volMeta.VolumeType = *input.VolumeType + } + if input.Iops != nil { + volMeta.IOPS = int(*input.Iops) + } + + // Persist updated config + if err := s.putVolumeConfig(volumeID, cfg); err != nil { + slog.Error("ModifyVolume failed to write config", "volumeId", volumeID, "err", err) + return nil, errors.New(awserrors.ErrorServerInternal) + } + + // Build target values (after modification) + targetSize := utils.SafeUint64ToInt64(volMeta.SizeGiB) + targetType := volMeta.VolumeType + if targetType == "" { + targetType = "gp3" + } + targetIOPS := int64(volMeta.IOPS) + + now := time.Now() + modification := &ec2.VolumeModification{ + VolumeId: aws.String(volumeID), + ModificationState: aws.String("completed"), + Progress: aws.Int64(100), + OriginalSize: aws.Int64(originalSize), + OriginalVolumeType: aws.String(originalType), + OriginalIops: aws.Int64(originalIOPS), + TargetSize: aws.Int64(targetSize), + TargetVolumeType: aws.String(targetType), + TargetIops: aws.Int64(targetIOPS), + StartTime: aws.Time(now), + EndTime: aws.Time(now), + } + + slog.Info("ModifyVolume completed", "volumeId", volumeID, + "originalSize", originalSize, "targetSize", targetSize) + + return &ec2.ModifyVolumeOutput{ + VolumeModification: modification, + }, nil +} diff --git a/hive/handlers/ec2/volume/service_nats.go b/hive/handlers/ec2/volume/service_nats.go index 65ebf4d95..74801a788 100644 --- a/hive/handlers/ec2/volume/service_nats.go +++ b/hive/handlers/ec2/volume/service_nats.go @@ -2,6 +2,7 @@ package handlers_ec2_volume import ( "encoding/json" + "errors" "fmt" "time" @@ -37,7 +38,7 @@ func (s *NATSVolumeService) DescribeVolumes(input *ec2.DescribeVolumesInput) (*e // Validate error response responseError, err := utils.ValidateErrorPayload(msg.Data) if err != nil { - return nil, fmt.Errorf("daemon returned error: %s", *responseError.Code) + return nil, errors.New(*responseError.Code) } // Unmarshal successful response @@ -49,3 +50,29 @@ func (s *NATSVolumeService) DescribeVolumes(input *ec2.DescribeVolumesInput) (*e return &output, nil } + +// ModifyVolume sends a ModifyVolume request via NATS and waits for response +func (s *NATSVolumeService) ModifyVolume(input *ec2.ModifyVolumeInput) (*ec2.ModifyVolumeOutput, error) { + jsonData, err := json.Marshal(input) + if err != nil { + return nil, fmt.Errorf("failed to marshal input: %w", err) + } + + msg, err := s.natsConn.Request("ec2.ModifyVolume", jsonData, 30*time.Second) + if err != nil { + return nil, fmt.Errorf("NATS request failed: %w", err) + } + + responseError, err := utils.ValidateErrorPayload(msg.Data) + if err != nil { + return nil, errors.New(*responseError.Code) + } + + var output ec2.ModifyVolumeOutput + err = json.Unmarshal(msg.Data, &output) + if err != nil { + return nil, fmt.Errorf("failed to unmarshal response: %w", err) + } + + return &output, nil +} diff --git a/hive/services/viperblockd/viperblockd.go b/hive/services/viperblockd/viperblockd.go index 3e5c5b155..3d5c2e7bd 100644 --- a/hive/services/viperblockd/viperblockd.go +++ b/hive/services/viperblockd/viperblockd.go @@ -29,6 +29,7 @@ type MountedVolume struct { Socket string // Unix socket path (when using socket transport) NBDURI string // Full NBD URI (nbd:unix:/path.sock or nbd://host:port) PID int + VB *viperblock.VB // Reference to viperblock instance for state sync/flush } type Config struct { @@ -137,37 +138,47 @@ func launchService(cfg *Config) (err error) { return } - // Find the volume in the mounted volumes + // Find the volume and extract references while holding the lock, + // then release before calling VB.Close() (which does heavy S3 I/O). var ebsResponse config.EBSUnMountResponse - var match bool + var matched MountedVolume + var matchIdx int = -1 cfg.mu.Lock() - for _, volume := range cfg.MountedVolumes { - - // TODO: Confirm KVM/QEMU is not using the volume first. + for i, volume := range cfg.MountedVolumes { if volume.Name == ebsRequest.Name { - ebsResponse = config.EBSUnMountResponse{ - Volume: volume.Name, - Mounted: false, + matched = volume + matchIdx = i + // Remove from slice while we hold the lock + cfg.MountedVolumes = append(cfg.MountedVolumes[:i], cfg.MountedVolumes[i+1:]...) + break + } + } + cfg.mu.Unlock() + + if matchIdx >= 0 { + ebsResponse = config.EBSUnMountResponse{ + Volume: matched.Name, + Mounted: false, + } + + // Clean up the VB instance's background goroutine. + // This VB is state-only (LoadState/sync) — actual I/O is in the nbdkit plugin process. + if matched.VB != nil { + matched.VB.StopWALSyncer() + } + + utils.KillProcess(matched.PID) + + // Remove the socket file if using socket transport + if matched.Socket != "" { + slog.Info("Removing socket file", "socket", matched.Socket) + if err := os.Remove(matched.Socket); err != nil && !os.IsNotExist(err) { + slog.Error("Failed to delete nbd socket", "err", err, "socket", matched.Socket) } - - utils.KillProcess(volume.PID) - match = true - - // Remove the socket file if using socket transport - if volume.Socket != "" { - slog.Info("Removing socket file", "socket", volume.Socket) - err := os.Remove(volume.Socket) - if err != nil && !os.IsNotExist(err) { - slog.Error("Failed to delete nbd socket", "err", err, "socket", volume.Socket) - } - } - } } - cfg.mu.Unlock() - - if !match { + if matchIdx < 0 { ebsResponse = config.EBSUnMountResponse{ Volume: ebsRequest.Name, Error: fmt.Sprintf("Volume %s not found", ebsRequest.Name), @@ -185,6 +196,52 @@ func launchService(cfg *Config) (err error) { nc.Publish("ebs.unmount.response", response) }) + nc.QueueSubscribe("ebs.sync", "hive-workers", func(msg *nats.Msg) { + slog.Info("Received ebs.sync message", "data", string(msg.Data)) + + var syncRequest config.EBSSyncRequest + if err := json.Unmarshal(msg.Data, &syncRequest); err != nil { + slog.Error("Failed to unmarshal ebs.sync message", "err", err) + errResp, _ := json.Marshal(config.EBSSyncResponse{Error: fmt.Sprintf("bad request: %v", err)}) + msg.Respond(errResp) + return + } + + syncResponse := config.EBSSyncResponse{Volume: syncRequest.Volume} + + // Find the mounted volume and reload its state from the backend + cfg.mu.Lock() + var foundVB *viperblock.VB + for _, volume := range cfg.MountedVolumes { + if volume.Name == syncRequest.Volume && volume.VB != nil { + foundVB = volume.VB + break + } + } + cfg.mu.Unlock() + + if foundVB == nil { + syncResponse.Error = fmt.Sprintf("volume %s not mounted or has no VB instance", syncRequest.Volume) + slog.Warn("ebs.sync: volume not found", "volume", syncRequest.Volume) + } else if err := foundVB.LoadState(); err != nil { + syncResponse.Error = fmt.Sprintf("failed to reload state: %v", err) + slog.Error("ebs.sync: LoadState failed", "volume", syncRequest.Volume, "err", err) + } else { + syncResponse.Synced = true + slog.Info("ebs.sync: state reloaded", "volume", syncRequest.Volume, + "volumeSize", foundVB.GetVolumeSize()) + } + + response, err := json.Marshal(syncResponse) + if err != nil { + slog.Error("Failed to marshal ebs.sync response", "err", err) + msg.Respond([]byte(`{"Error":"internal marshal failure"}`)) + return + } + + msg.Respond(response) + }) + nc.QueueSubscribe("ebs.mount", "hive-workers", func(msg *nats.Msg) { slog.Info("Received message:", "data", string(msg.Data)) @@ -425,6 +482,7 @@ func launchService(cfg *Config) (err error) { Socket: nbdSocket, NBDURI: nbdURI, PID: pid, + VB: vb, }) cfg.mu.Unlock() @@ -452,13 +510,21 @@ func launchService(cfg *Config) (err error) { nc.Close() - // Close all nbdkit processes + // Snapshot mounted volumes and clear the list while holding the lock, + // then flush/kill outside the lock (VB.Close does heavy I/O). cfg.mu.Lock() - for _, volume := range cfg.MountedVolumes { + volumes := make([]MountedVolume, len(cfg.MountedVolumes)) + copy(volumes, cfg.MountedVolumes) + cfg.MountedVolumes = nil + cfg.mu.Unlock() + + for _, volume := range volumes { + if volume.VB != nil { + volume.VB.StopWALSyncer() + } slog.Info("Killing nbdkit process", "pid", volume.PID) utils.KillProcess(volume.PID) } - cfg.mu.Unlock() return nil