mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-15 17:11:30 +00:00
Feat/describe instance types
This commit is contained in:
@@ -13,7 +13,7 @@ Big rocks:
|
||||
- Add support using the `hive` CLI tool to provision a new user with AWS access-keys/IAM.
|
||||
- Support multi-tenant operations and isolation
|
||||
- Add support to include capabilities when adding a new hardware node to MulgaOS (e.g EC2 target, S3, EBS, NATs, etc) - Features can be turned on/off depending on hardware scope.
|
||||
- Add simple Web UI console, using the AWS JS SDK, communicating to local AWS gateway.
|
||||
- [DONE] Add simple Web UI console, using the AWS JS SDK, communicating to local AWS gateway.
|
||||
- Implement ShadCNblocks for UI framework
|
||||
- Simple Go webserver, static files, easy build process.
|
||||
|
||||
@@ -27,14 +27,17 @@ Implementation gaps:
|
||||
|
||||
### EC2
|
||||
|
||||
- [DONE] - create-key-pair
|
||||
- [DONE] - delete-key-pair
|
||||
- [DONE] - describe-images
|
||||
- [DONE] - import-key-pair
|
||||
- [DONE] - describe-instances
|
||||
- [DONE] - run-instances (count not implemented)
|
||||
- [DONE] - start-instances
|
||||
- [DONE] - stop-instances
|
||||
- [DONE] - terminate-instances
|
||||
- [DONE] - describe-instance-types
|
||||
- [DONE] - create-key-pair
|
||||
- [DONE] - delete-key-pair
|
||||
- [DONE] - describe-key-pairs
|
||||
- [DONE] - import-key-pair
|
||||
- [DONE] - describe-images
|
||||
|
||||
### To implement
|
||||
|
||||
@@ -135,7 +138,6 @@ TODO:
|
||||
- describe-egress-only-internet-gateways
|
||||
- describe-hosts
|
||||
- describe-internet-gateways
|
||||
- describe-key-pairs
|
||||
- describe-managed-prefix-lists
|
||||
- describe-nat-gateways
|
||||
- describe-network-acls
|
||||
@@ -173,7 +175,6 @@ TODO:
|
||||
- get-spot-placement-scores
|
||||
- get-subnet-cidr-reservations
|
||||
- import-image
|
||||
- import-key-pair
|
||||
- import-snapshot
|
||||
- modify-instance-attribute
|
||||
- modify-instance-capacity-reservation-attributes
|
||||
|
||||
+314
-62
@@ -21,6 +21,7 @@ import (
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"github.com/aws/aws-sdk-go/aws"
|
||||
"github.com/aws/aws-sdk-go/service/ec2"
|
||||
"github.com/gofiber/fiber/v2"
|
||||
"github.com/mulgadc/hive/hive/awserrors"
|
||||
@@ -55,14 +56,6 @@ type EBS struct {
|
||||
VolumeType string
|
||||
}
|
||||
|
||||
// InstanceType represents the resource requirements for an EC2 instance type
|
||||
type InstanceType struct {
|
||||
Name string
|
||||
VCPUs int
|
||||
MemoryGB float64
|
||||
Architecture string // e.g., "x86_64", "arm64"
|
||||
}
|
||||
|
||||
// ResourceManager handles the allocation and tracking of system resources
|
||||
type ResourceManager struct {
|
||||
mu sync.RWMutex
|
||||
@@ -70,7 +63,7 @@ type ResourceManager struct {
|
||||
availableMem float64
|
||||
allocatedVCPU int
|
||||
allocatedMem float64
|
||||
instanceTypes map[string]InstanceType
|
||||
instanceTypes map[string]*ec2.InstanceTypeInfo
|
||||
}
|
||||
|
||||
// Daemon represents the main daemon service
|
||||
@@ -101,6 +94,30 @@ type Daemon struct {
|
||||
mu sync.Mutex
|
||||
}
|
||||
|
||||
// cpuToInstanceFamily maps CPU model patterns to AWS instance family prefixes
|
||||
var cpuToInstanceFamily = map[string]string{
|
||||
"EPYC": "m8a", // AMD EPYC processors
|
||||
"Xeon": "m7i", // Intel Xeon processors
|
||||
"ARM": "m8g", // ARM-based processors
|
||||
"Apple": "m8g", // Apple Silicon (ARM-based)
|
||||
"Graviton": "m8g", // AWS Graviton
|
||||
}
|
||||
|
||||
// getInstanceFamilyFromCPU returns the AWS instance family based on CPU model
|
||||
func getInstanceFamilyFromCPU(cpuModel string) string {
|
||||
cpuUpper := strings.ToUpper(cpuModel)
|
||||
for pattern, family := range cpuToInstanceFamily {
|
||||
if strings.Contains(cpuUpper, strings.ToUpper(pattern)) {
|
||||
return family
|
||||
}
|
||||
}
|
||||
// Default fallback based on architecture
|
||||
if runtime.GOARCH == "arm64" {
|
||||
return "t4g"
|
||||
}
|
||||
return "t3" // fallback for unknown x86_64
|
||||
}
|
||||
|
||||
// getSystemMemory returns the total system memory in GB
|
||||
func getSystemMemory() (float64, error) {
|
||||
switch runtime.GOOS {
|
||||
@@ -144,6 +161,83 @@ func getSystemMemory() (float64, error) {
|
||||
}
|
||||
}
|
||||
|
||||
// getCPUModel returns the CPU model name for the host system
|
||||
func getCPUModel() (string, error) {
|
||||
switch runtime.GOOS {
|
||||
case "darwin":
|
||||
// macOS: use sysctl
|
||||
cmd := exec.Command("sysctl", "-n", "machdep.cpu.brand_string")
|
||||
output, err := cmd.Output()
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("failed to get CPU model on macOS: %w", err)
|
||||
}
|
||||
return strings.TrimSpace(string(output)), nil
|
||||
|
||||
case "linux":
|
||||
// Linux: read from /proc/cpuinfo
|
||||
file, err := os.Open("/proc/cpuinfo")
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("failed to open /proc/cpuinfo: %w", err)
|
||||
}
|
||||
defer file.Close()
|
||||
|
||||
scanner := bufio.NewScanner(file)
|
||||
for scanner.Scan() {
|
||||
line := scanner.Text()
|
||||
if strings.HasPrefix(line, "model name") {
|
||||
parts := strings.SplitN(line, ":", 2)
|
||||
if len(parts) == 2 {
|
||||
return strings.TrimSpace(parts[1]), nil
|
||||
}
|
||||
}
|
||||
}
|
||||
return "", fmt.Errorf("model name not found in /proc/cpuinfo")
|
||||
|
||||
default:
|
||||
return "", fmt.Errorf("unsupported operating system: %s", runtime.GOOS)
|
||||
}
|
||||
}
|
||||
|
||||
// generateInstanceTypes creates the instance type map for the detected CPU family
|
||||
func generateInstanceTypes(family, arch string) map[string]*ec2.InstanceTypeInfo {
|
||||
sizes := []struct {
|
||||
suffix string
|
||||
vcpus int
|
||||
memoryGB float64
|
||||
}{
|
||||
{"nano", 2, 0.5},
|
||||
{"micro", 2, 1.0},
|
||||
{"small", 2, 2.0},
|
||||
{"medium", 2, 4.0},
|
||||
{"large", 2, 8.0},
|
||||
{"xlarge", 4, 16.0},
|
||||
{"2xlarge", 8, 32.0},
|
||||
}
|
||||
|
||||
instanceTypes := make(map[string]*ec2.InstanceTypeInfo)
|
||||
for _, size := range sizes {
|
||||
name := fmt.Sprintf("%s.%s", family, size.suffix)
|
||||
instanceTypes[name] = &ec2.InstanceTypeInfo{
|
||||
InstanceType: aws.String(name),
|
||||
VCpuInfo: &ec2.VCpuInfo{
|
||||
DefaultVCpus: aws.Int64(int64(size.vcpus)),
|
||||
},
|
||||
MemoryInfo: &ec2.MemoryInfo{
|
||||
SizeInMiB: aws.Int64(int64(size.memoryGB * 1024)),
|
||||
},
|
||||
ProcessorInfo: &ec2.ProcessorInfo{
|
||||
SupportedArchitectures: []*string{aws.String(arch)},
|
||||
},
|
||||
CurrentGeneration: aws.Bool(true),
|
||||
BurstablePerformanceSupported: aws.Bool(strings.HasPrefix(name, "t")),
|
||||
Hypervisor: aws.String("kvm"),
|
||||
SupportedVirtualizationTypes: []*string{aws.String("hvm")},
|
||||
SupportedRootDeviceTypes: []*string{aws.String("ebs")},
|
||||
}
|
||||
}
|
||||
return instanceTypes
|
||||
}
|
||||
|
||||
// NewResourceManager creates a new resource manager with system capabilities
|
||||
func NewResourceManager() *ResourceManager {
|
||||
// Get system CPU cores
|
||||
@@ -156,30 +250,27 @@ func NewResourceManager() *ResourceManager {
|
||||
totalMemGB = 8.0 // Default to 8GB if we can't get the actual memory
|
||||
}
|
||||
|
||||
// Define supported instance types
|
||||
// TODO: Determine host capabilities (x86_64 vs arm64) and adjust available instance types accordingly
|
||||
instanceTypes := map[string]InstanceType{
|
||||
// x86_64
|
||||
"t3.nano": {Name: "t3.nano", VCPUs: 2, MemoryGB: 0.5, Architecture: "x86_64"},
|
||||
"t3.micro": {Name: "t3.micro", VCPUs: 2, MemoryGB: 1.0, Architecture: "x86_64"},
|
||||
"t3.small": {Name: "t3.small", VCPUs: 2, MemoryGB: 2.0, Architecture: "x86_64"},
|
||||
"t3.medium": {Name: "t3.medium", VCPUs: 2, MemoryGB: 4.0, Architecture: "x86_64"},
|
||||
"t3.large": {Name: "t3.large", VCPUs: 2, MemoryGB: 8.0, Architecture: "x86_64"},
|
||||
"t3.xlarge": {Name: "t3.xlarge", VCPUs: 4, MemoryGB: 16.0, Architecture: "x86_64"},
|
||||
"t3.2xlarge": {Name: "t3.2xlarge", VCPUs: 8, MemoryGB: 32.0, Architecture: "x86_64"},
|
||||
|
||||
// ARM
|
||||
"t4g.nano": {Name: "t4g.nano", VCPUs: 2, MemoryGB: 0.5, Architecture: "arm64"},
|
||||
"t4g.micro": {Name: "t4g.micro", VCPUs: 2, MemoryGB: 1.0, Architecture: "arm64"},
|
||||
"t4g.small": {Name: "t4g.small", VCPUs: 2, MemoryGB: 2.0, Architecture: "arm64"},
|
||||
"t4g.medium": {Name: "t4g.medium", VCPUs: 2, MemoryGB: 4.0, Architecture: "arm64"},
|
||||
"t4g.large": {Name: "t4g.large", VCPUs: 2, MemoryGB: 8.0, Architecture: "arm64"},
|
||||
"t4g.xlarge": {Name: "t4g.xlarge", VCPUs: 4, MemoryGB: 16.0, Architecture: "arm64"},
|
||||
"t4g.2xlarge": {Name: "t4g.2xlarge", VCPUs: 8, MemoryGB: 32.0, Architecture: "arm64"},
|
||||
// Get CPU model for instance family detection
|
||||
cpuModel, err := getCPUModel()
|
||||
if err != nil {
|
||||
log.Printf("Warning: Failed to get CPU model: %v, using default", err)
|
||||
cpuModel = "Unknown"
|
||||
}
|
||||
|
||||
log.Printf("System resources: %d vCPUs, %.2f GB RAM (detected on %s)",
|
||||
numCPU, totalMemGB, runtime.GOOS)
|
||||
// Determine instance family from CPU model
|
||||
instanceFamily := getInstanceFamilyFromCPU(cpuModel)
|
||||
|
||||
// Determine architecture
|
||||
arch := "x86_64"
|
||||
if runtime.GOARCH == "arm64" {
|
||||
arch = "arm64"
|
||||
}
|
||||
|
||||
// Generate instance types based on detected CPU family
|
||||
instanceTypes := generateInstanceTypes(instanceFamily, arch)
|
||||
|
||||
log.Printf("System resources: %d vCPUs, %.2f GB RAM, CPU: %s, Family: %s (detected on %s)",
|
||||
numCPU, totalMemGB, cpuModel, instanceFamily, runtime.GOOS)
|
||||
|
||||
return &ResourceManager{
|
||||
availableVCPU: numCPU,
|
||||
@@ -188,6 +279,53 @@ func NewResourceManager() *ResourceManager {
|
||||
}
|
||||
}
|
||||
|
||||
// GetInstanceTypeInfos returns all instance types as ec2.InstanceTypeInfo for AWS API compatibility
|
||||
func (rm *ResourceManager) GetInstanceTypeInfos() []*ec2.InstanceTypeInfo {
|
||||
rm.mu.RLock()
|
||||
defer rm.mu.RUnlock()
|
||||
|
||||
var infos []*ec2.InstanceTypeInfo
|
||||
for _, it := range rm.instanceTypes {
|
||||
infos = append(infos, it)
|
||||
}
|
||||
return infos
|
||||
}
|
||||
|
||||
// GetAvailableInstanceTypeInfos returns only instance types that can currently be provisioned
|
||||
// based on available resources (filters out types that exceed remaining CPU or memory)
|
||||
func (rm *ResourceManager) GetAvailableInstanceTypeInfos() []*ec2.InstanceTypeInfo {
|
||||
rm.mu.RLock()
|
||||
defer rm.mu.RUnlock()
|
||||
|
||||
var infos []*ec2.InstanceTypeInfo
|
||||
|
||||
// Calculate remaining resources
|
||||
remainingVCPU := rm.availableVCPU - rm.allocatedVCPU
|
||||
remainingMem := rm.availableMem - rm.allocatedMem
|
||||
|
||||
for _, it := range rm.instanceTypes {
|
||||
// Skip instance types that exceed available resources
|
||||
vCPUs := int64(0)
|
||||
if it.VCpuInfo != nil && it.VCpuInfo.DefaultVCpus != nil {
|
||||
vCPUs = *it.VCpuInfo.DefaultVCpus
|
||||
}
|
||||
memoryGB := float64(0)
|
||||
if it.MemoryInfo != nil && it.MemoryInfo.SizeInMiB != nil {
|
||||
memoryGB = float64(*it.MemoryInfo.SizeInMiB) / 1024.0
|
||||
}
|
||||
if int(vCPUs) > remainingVCPU || memoryGB > remainingMem {
|
||||
continue
|
||||
}
|
||||
|
||||
infos = append(infos, it)
|
||||
}
|
||||
|
||||
slog.Info("GetAvailableInstanceTypeInfos", "total", len(rm.instanceTypes), "available", len(infos),
|
||||
"remainingVCPU", remainingVCPU, "remainingMem", remainingMem)
|
||||
|
||||
return infos
|
||||
}
|
||||
|
||||
// SetConfigPath sets the configuration file path for cluster management
|
||||
func (d *Daemon) SetConfigPath(path string) {
|
||||
d.configPath = path
|
||||
@@ -285,16 +423,7 @@ func (d *Daemon) Start() error {
|
||||
}
|
||||
|
||||
// Create instance service for handling EC2 instance operations
|
||||
instanceTypes := make(map[string]handlers_ec2_instance.InstanceType)
|
||||
for k, v := range d.resourceMgr.instanceTypes {
|
||||
instanceTypes[k] = handlers_ec2_instance.InstanceType{
|
||||
Name: v.Name,
|
||||
VCPUs: v.VCPUs,
|
||||
MemoryGB: v.MemoryGB,
|
||||
Architecture: v.Architecture,
|
||||
}
|
||||
}
|
||||
d.instanceService = handlers_ec2_instance.NewInstanceServiceImpl(d.config, instanceTypes, d.natsConn, &d.Instances)
|
||||
d.instanceService = handlers_ec2_instance.NewInstanceServiceImpl(d.config, d.resourceMgr.instanceTypes, d.natsConn, &d.Instances)
|
||||
|
||||
// Create key service for handling EC2 key pair operations
|
||||
d.keyService = handlers_ec2_key.NewKeyServiceImpl(d.config)
|
||||
@@ -376,6 +505,15 @@ func (d *Daemon) Start() error {
|
||||
return fmt.Errorf("failed to subscribe to NATS ec2.DescribeInstances: %w", err)
|
||||
}
|
||||
|
||||
log.Printf("Subscribing to subject pattern: %s", "ec2.DescribeInstanceTypes")
|
||||
|
||||
// Subscribe to EC2 DescribeInstanceTypes - no queue group for multi-node fan-out
|
||||
d.natsSubscriptions["ec2.DescribeInstanceTypes"], err = d.natsConn.Subscribe("ec2.DescribeInstanceTypes", d.handleEC2DescribeInstanceTypes)
|
||||
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to subscribe to NATS ec2.DescribeInstanceTypes: %w", err)
|
||||
}
|
||||
|
||||
// Subscribe to EC2 start instance events
|
||||
// TODO: The instance state needs to be shared, not pinned to a single node.
|
||||
// TODO: Handle this in a more generic function to group similar commands (start, stop, launch)
|
||||
@@ -731,13 +869,14 @@ func (d *Daemon) SendQMPCommand(q *qmp.QMPClient, cmd qmp.QMPCommand, instanceId
|
||||
|
||||
var updatedStatus string
|
||||
|
||||
if msg["event"] == "STOP" {
|
||||
switch msg["event"] {
|
||||
case "STOP":
|
||||
updatedStatus = "stopped"
|
||||
} else if msg["event"] == "RESUME" {
|
||||
case "RESUME":
|
||||
updatedStatus = "resuming"
|
||||
} else if msg["event"] == "RESET" {
|
||||
case "RESET":
|
||||
updatedStatus = "restarting"
|
||||
} else if msg["event"] == "POWERDOWN" {
|
||||
case "POWERDOWN":
|
||||
updatedStatus = "powering_down"
|
||||
}
|
||||
|
||||
@@ -809,7 +948,9 @@ func (d *Daemon) handleEC2Events(msg *nats.Msg) {
|
||||
slog.Error("handleEC2RunInstances LaunchInstance failed", "err", err)
|
||||
// TODO: Confirm LaunchInstances does this - Free the resource
|
||||
instanceType := d.resourceMgr.instanceTypes[instance.InstanceType]
|
||||
d.resourceMgr.deallocate(instanceType)
|
||||
if instanceType != nil {
|
||||
d.resourceMgr.deallocate(instanceType)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
@@ -863,7 +1004,9 @@ func (d *Daemon) handleEC2Events(msg *nats.Msg) {
|
||||
|
||||
// Free resources
|
||||
instanceType := d.resourceMgr.instanceTypes[instance.InstanceType]
|
||||
d.resourceMgr.deallocate(instanceType)
|
||||
if instanceType != nil {
|
||||
d.resourceMgr.deallocate(instanceType)
|
||||
}
|
||||
|
||||
// Remove instance from state
|
||||
d.Instances.Mu.Lock()
|
||||
@@ -947,7 +1090,11 @@ func (d *Daemon) handleEC2RunInstances(msg *nats.Msg) {
|
||||
}
|
||||
|
||||
// Delegate to service for business logic (volume creation, cloud-init, etc.)
|
||||
slog.Info("Launching EC2 instance", "instance", instanceType)
|
||||
instanceTypeName := ""
|
||||
if instanceType.InstanceType != nil {
|
||||
instanceTypeName = *instanceType.InstanceType
|
||||
}
|
||||
slog.Info("Launching EC2 instance", "instanceType", instanceTypeName)
|
||||
|
||||
instance, reservation, err := d.instanceService.RunInstances(runInstancesInput)
|
||||
|
||||
@@ -1391,7 +1538,23 @@ func (d *Daemon) StartInstance(instance *vm.VM) error {
|
||||
return err
|
||||
}
|
||||
|
||||
var instanceType = d.resourceMgr.instanceTypes[instance.InstanceType]
|
||||
instanceType := d.resourceMgr.instanceTypes[instance.InstanceType]
|
||||
if instanceType == nil {
|
||||
return fmt.Errorf("instance type %s not found", instance.InstanceType)
|
||||
}
|
||||
|
||||
vCPUs := int(0)
|
||||
if instanceType.VCpuInfo != nil && instanceType.VCpuInfo.DefaultVCpus != nil {
|
||||
vCPUs = int(*instanceType.VCpuInfo.DefaultVCpus)
|
||||
}
|
||||
memoryMiB := int64(0)
|
||||
if instanceType.MemoryInfo != nil && instanceType.MemoryInfo.SizeInMiB != nil {
|
||||
memoryMiB = *instanceType.MemoryInfo.SizeInMiB
|
||||
}
|
||||
architecture := "x86_64"
|
||||
if instanceType.ProcessorInfo != nil && len(instanceType.ProcessorInfo.SupportedArchitectures) > 0 && instanceType.ProcessorInfo.SupportedArchitectures[0] != nil {
|
||||
architecture = *instanceType.ProcessorInfo.SupportedArchitectures[0]
|
||||
}
|
||||
|
||||
instance.Config = vm.Config{
|
||||
Name: instance.ID,
|
||||
@@ -1402,9 +1565,9 @@ func (d *Daemon) StartInstance(instance *vm.VM) error {
|
||||
MachineType: "q35",
|
||||
Serial: "pty",
|
||||
CPUType: "host", // If available, if kvm fails, will use cpu max
|
||||
Memory: int(instanceType.MemoryGB) * 1024,
|
||||
CPUCount: instanceType.VCPUs,
|
||||
Architecture: instanceType.Architecture,
|
||||
Memory: int(memoryMiB),
|
||||
CPUCount: vCPUs,
|
||||
Architecture: architecture,
|
||||
}
|
||||
|
||||
// Loop through each volume in volumes
|
||||
@@ -1686,36 +1849,67 @@ func (d *Daemon) MountVolumes(instance *vm.VM) error {
|
||||
}
|
||||
|
||||
// canAllocate checks if there are enough resources available
|
||||
func (rm *ResourceManager) canAllocate(instanceType InstanceType) bool {
|
||||
func (rm *ResourceManager) canAllocate(instanceType *ec2.InstanceTypeInfo) bool {
|
||||
rm.mu.RLock()
|
||||
defer rm.mu.RUnlock()
|
||||
|
||||
return rm.availableVCPU-rm.allocatedVCPU >= instanceType.VCPUs &&
|
||||
rm.availableMem-rm.allocatedMem >= instanceType.MemoryGB
|
||||
vCPUs := int64(0)
|
||||
if instanceType.VCpuInfo != nil && instanceType.VCpuInfo.DefaultVCpus != nil {
|
||||
vCPUs = *instanceType.VCpuInfo.DefaultVCpus
|
||||
}
|
||||
memoryGB := float64(0)
|
||||
if instanceType.MemoryInfo != nil && instanceType.MemoryInfo.SizeInMiB != nil {
|
||||
memoryGB = float64(*instanceType.MemoryInfo.SizeInMiB) / 1024.0
|
||||
}
|
||||
|
||||
return rm.availableVCPU-rm.allocatedVCPU >= int(vCPUs) &&
|
||||
rm.availableMem-rm.allocatedMem >= memoryGB
|
||||
}
|
||||
|
||||
// allocate reserves resources for an instance
|
||||
func (rm *ResourceManager) allocate(instanceType InstanceType) error {
|
||||
func (rm *ResourceManager) allocate(instanceType *ec2.InstanceTypeInfo) error {
|
||||
|
||||
if !rm.canAllocate(instanceType) {
|
||||
return fmt.Errorf("insufficient resources for instance type %s", instanceType.Name)
|
||||
instanceTypeName := ""
|
||||
if instanceType.InstanceType != nil {
|
||||
instanceTypeName = *instanceType.InstanceType
|
||||
}
|
||||
return fmt.Errorf("insufficient resources for instance type %s", instanceTypeName)
|
||||
}
|
||||
|
||||
rm.mu.Lock()
|
||||
defer rm.mu.Unlock()
|
||||
|
||||
rm.allocatedVCPU += instanceType.VCPUs
|
||||
rm.allocatedMem += instanceType.MemoryGB
|
||||
vCPUs := int64(0)
|
||||
if instanceType.VCpuInfo != nil && instanceType.VCpuInfo.DefaultVCpus != nil {
|
||||
vCPUs = *instanceType.VCpuInfo.DefaultVCpus
|
||||
}
|
||||
memoryGB := float64(0)
|
||||
if instanceType.MemoryInfo != nil && instanceType.MemoryInfo.SizeInMiB != nil {
|
||||
memoryGB = float64(*instanceType.MemoryInfo.SizeInMiB) / 1024.0
|
||||
}
|
||||
|
||||
rm.allocatedVCPU += int(vCPUs)
|
||||
rm.allocatedMem += memoryGB
|
||||
return nil
|
||||
}
|
||||
|
||||
// deallocate releases resources for an instance
|
||||
func (rm *ResourceManager) deallocate(instanceType InstanceType) {
|
||||
func (rm *ResourceManager) deallocate(instanceType *ec2.InstanceTypeInfo) {
|
||||
rm.mu.Lock()
|
||||
defer rm.mu.Unlock()
|
||||
|
||||
rm.allocatedVCPU -= instanceType.VCPUs
|
||||
rm.allocatedMem -= instanceType.MemoryGB
|
||||
vCPUs := int64(0)
|
||||
if instanceType.VCpuInfo != nil && instanceType.VCpuInfo.DefaultVCpus != nil {
|
||||
vCPUs = *instanceType.VCpuInfo.DefaultVCpus
|
||||
}
|
||||
memoryGB := float64(0)
|
||||
if instanceType.MemoryInfo != nil && instanceType.MemoryInfo.SizeInMiB != nil {
|
||||
memoryGB = float64(*instanceType.MemoryInfo.SizeInMiB) / 1024.0
|
||||
}
|
||||
|
||||
rm.allocatedVCPU -= int(vCPUs)
|
||||
rm.allocatedMem -= memoryGB
|
||||
}
|
||||
|
||||
// handleEC2DescribeKeyPairs processes incoming EC2 DescribeKeyPairs requests
|
||||
@@ -1847,6 +2041,64 @@ func (d *Daemon) handleEC2DescribeImages(msg *nats.Msg) {
|
||||
slog.Info("handleEC2DescribeImages completed", "count", len(output.Images))
|
||||
}
|
||||
|
||||
// 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)
|
||||
func (d *Daemon) handleEC2DescribeInstanceTypes(msg *nats.Msg) {
|
||||
log.Printf("Received message on subject: %s", msg.Subject)
|
||||
|
||||
// Initialize input
|
||||
describeInput := &ec2.DescribeInstanceTypesInput{}
|
||||
var errResp []byte
|
||||
|
||||
errResp = utils.UnmarshalJsonPayload(describeInput, msg.Data)
|
||||
if errResp != nil {
|
||||
msg.Respond(errResp)
|
||||
slog.Error("Request does not match DescribeInstanceTypesInput")
|
||||
return
|
||||
}
|
||||
|
||||
slog.Info("Processing DescribeInstanceTypes request from this node")
|
||||
|
||||
// Get instance types that can be provisioned with current available resources
|
||||
availableTypes := d.resourceMgr.GetAvailableInstanceTypeInfos()
|
||||
|
||||
// Filter by requested instance types if specified
|
||||
var filteredTypes []*ec2.InstanceTypeInfo
|
||||
if len(describeInput.InstanceTypes) > 0 {
|
||||
requestedTypes := make(map[string]bool)
|
||||
for _, t := range describeInput.InstanceTypes {
|
||||
if t != nil {
|
||||
requestedTypes[*t] = true
|
||||
}
|
||||
}
|
||||
for _, info := range availableTypes {
|
||||
if info.InstanceType != nil && requestedTypes[*info.InstanceType] {
|
||||
filteredTypes = append(filteredTypes, info)
|
||||
}
|
||||
}
|
||||
} else {
|
||||
filteredTypes = availableTypes
|
||||
}
|
||||
|
||||
// Create the response
|
||||
output := &ec2.DescribeInstanceTypesOutput{
|
||||
InstanceTypes: filteredTypes,
|
||||
}
|
||||
|
||||
// Respond to NATS
|
||||
jsonResponse, err := json.Marshal(output)
|
||||
if err != nil {
|
||||
slog.Error("handleEC2DescribeInstanceTypes failed to marshal output", "err", err)
|
||||
errResp = utils.GenerateErrorPayload(awserrors.ErrorServerInternal)
|
||||
msg.Respond(errResp)
|
||||
return
|
||||
}
|
||||
msg.Respond(jsonResponse)
|
||||
|
||||
slog.Info("handleEC2DescribeInstanceTypes completed", "count", len(filteredTypes))
|
||||
}
|
||||
|
||||
// handleEC2DescribeInstances processes incoming EC2 DescribeInstances requests
|
||||
// This handler responds with all instances running on this node
|
||||
func (d *Daemon) handleEC2DescribeInstances(msg *nats.Msg) {
|
||||
|
||||
+394
-28
@@ -3,6 +3,8 @@ package daemon
|
||||
import (
|
||||
"encoding/json"
|
||||
"os"
|
||||
"runtime"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
@@ -93,16 +95,7 @@ func createTestDaemon(t *testing.T, natsURL string) *Daemon {
|
||||
daemon.natsConn = nc
|
||||
|
||||
// Initialize instance service (needed for handleEC2RunInstances)
|
||||
instanceTypes := make(map[string]handlers_ec2_instance.InstanceType)
|
||||
for k, v := range daemon.resourceMgr.instanceTypes {
|
||||
instanceTypes[k] = handlers_ec2_instance.InstanceType{
|
||||
Name: v.Name,
|
||||
VCPUs: v.VCPUs,
|
||||
MemoryGB: v.MemoryGB,
|
||||
Architecture: v.Architecture,
|
||||
}
|
||||
}
|
||||
daemon.instanceService = handlers_ec2_instance.NewInstanceServiceImpl(cfg, instanceTypes, nc, &daemon.Instances)
|
||||
daemon.instanceService = handlers_ec2_instance.NewInstanceServiceImpl(cfg, daemon.resourceMgr.instanceTypes, nc, &daemon.Instances)
|
||||
|
||||
t.Cleanup(func() {
|
||||
if daemon.natsConn != nil {
|
||||
@@ -113,20 +106,20 @@ func createTestDaemon(t *testing.T, natsURL string) *Daemon {
|
||||
return daemon
|
||||
}
|
||||
|
||||
// createValidRunInstancesInput creates a valid RunInstancesInput for testing
|
||||
func createValidRunInstancesInput() *ec2.RunInstancesInput {
|
||||
return &ec2.RunInstancesInput{
|
||||
ImageId: aws.String("ami-0abcdef1234567890"),
|
||||
InstanceType: aws.String("t3.micro"),
|
||||
MinCount: aws.Int64(1),
|
||||
MaxCount: aws.Int64(1),
|
||||
KeyName: aws.String("test-key-pair"),
|
||||
SecurityGroupIds: []*string{
|
||||
aws.String("sg-0123456789abcdef0"),
|
||||
},
|
||||
SubnetId: aws.String("subnet-6e7f829e"),
|
||||
UserData: aws.String("#!/bin/bash\necho 'test'"),
|
||||
// getTestInstanceType returns a valid instance type for testing based on the system's CPU
|
||||
func getTestInstanceType() string {
|
||||
rm := NewResourceManager()
|
||||
// Find any .micro instance type
|
||||
for key := range rm.instanceTypes {
|
||||
if strings.HasSuffix(key, ".micro") {
|
||||
return key
|
||||
}
|
||||
}
|
||||
// Fallback: return first available instance type
|
||||
for key := range rm.instanceTypes {
|
||||
return key
|
||||
}
|
||||
return "t3.micro" // Default fallback
|
||||
}
|
||||
|
||||
// TestHandleEC2RunInstances_MessageParsing tests that the handler correctly parses NATS messages
|
||||
@@ -394,9 +387,20 @@ func TestResourceManager(t *testing.T) {
|
||||
assert.Greater(t, rm.availableVCPU, 0)
|
||||
assert.Greater(t, rm.availableMem, float64(0))
|
||||
|
||||
// Test allocation
|
||||
instanceType := rm.instanceTypes["t3.micro"]
|
||||
require.NotNil(t, instanceType)
|
||||
// Test allocation using the first available instance type (dynamic based on CPU)
|
||||
require.NotEmpty(t, rm.instanceTypes, "Should have at least one instance type")
|
||||
|
||||
// Find any .micro instance type
|
||||
var instanceType *ec2.InstanceTypeInfo
|
||||
var exists bool
|
||||
for key, it := range rm.instanceTypes {
|
||||
if strings.HasSuffix(key, ".micro") {
|
||||
instanceType = it
|
||||
exists = true
|
||||
break
|
||||
}
|
||||
}
|
||||
require.True(t, exists, "Should have at least one .micro instance type")
|
||||
|
||||
// Check if can allocate
|
||||
canAlloc := rm.canAllocate(instanceType)
|
||||
@@ -407,11 +411,373 @@ func TestResourceManager(t *testing.T) {
|
||||
assert.NoError(t, err)
|
||||
|
||||
// Check resources were allocated
|
||||
assert.Equal(t, instanceType.VCPUs, rm.allocatedVCPU)
|
||||
assert.Equal(t, instanceType.MemoryGB, rm.allocatedMem)
|
||||
vCPUs := int64(0)
|
||||
if instanceType.VCpuInfo != nil && instanceType.VCpuInfo.DefaultVCpus != nil {
|
||||
vCPUs = *instanceType.VCpuInfo.DefaultVCpus
|
||||
}
|
||||
memoryGB := float64(0)
|
||||
if instanceType.MemoryInfo != nil && instanceType.MemoryInfo.SizeInMiB != nil {
|
||||
memoryGB = float64(*instanceType.MemoryInfo.SizeInMiB) / 1024.0
|
||||
}
|
||||
assert.Equal(t, int(vCPUs), rm.allocatedVCPU)
|
||||
assert.Equal(t, memoryGB, rm.allocatedMem)
|
||||
|
||||
// Deallocate
|
||||
rm.deallocate(instanceType)
|
||||
assert.Equal(t, 0, rm.allocatedVCPU)
|
||||
assert.Equal(t, float64(0), rm.allocatedMem)
|
||||
}
|
||||
|
||||
// TestGetInstanceTypeInfos tests the GetInstanceTypeInfos method
|
||||
func TestGetInstanceTypeInfos(t *testing.T) {
|
||||
rm := NewResourceManager()
|
||||
|
||||
infos := rm.GetInstanceTypeInfos()
|
||||
|
||||
require.NotEmpty(t, infos, "Should return at least one instance type")
|
||||
assert.Len(t, infos, 7, "Should have 7 instance sizes (nano, micro, small, medium, large, xlarge, 2xlarge)")
|
||||
|
||||
// Verify structure of returned instance type info
|
||||
for _, info := range infos {
|
||||
assert.NotNil(t, info.InstanceType, "InstanceType should not be nil")
|
||||
assert.NotNil(t, info.VCpuInfo, "VCpuInfo should not be nil")
|
||||
assert.NotNil(t, info.VCpuInfo.DefaultVCpus, "DefaultVCpus should not be nil")
|
||||
assert.NotNil(t, info.MemoryInfo, "MemoryInfo should not be nil")
|
||||
assert.NotNil(t, info.MemoryInfo.SizeInMiB, "SizeInMiB should not be nil")
|
||||
assert.NotNil(t, info.ProcessorInfo, "ProcessorInfo should not be nil")
|
||||
assert.NotEmpty(t, info.ProcessorInfo.SupportedArchitectures, "SupportedArchitectures should not be empty")
|
||||
assert.NotNil(t, info.CurrentGeneration, "CurrentGeneration should not be nil")
|
||||
assert.True(t, *info.CurrentGeneration, "CurrentGeneration should be true")
|
||||
|
||||
t.Logf("Instance type: %s, vCPUs: %d, Memory: %d MiB",
|
||||
*info.InstanceType, *info.VCpuInfo.DefaultVCpus, *info.MemoryInfo.SizeInMiB)
|
||||
}
|
||||
}
|
||||
|
||||
// TestCPUDetection tests CPU model detection
|
||||
func TestCPUDetection(t *testing.T) {
|
||||
cpuModel, err := getCPUModel()
|
||||
require.NoError(t, err, "Should be able to detect CPU model")
|
||||
assert.NotEmpty(t, cpuModel, "CPU model should not be empty")
|
||||
|
||||
t.Logf("Detected CPU model: %s", cpuModel)
|
||||
|
||||
// Test instance family mapping
|
||||
testCases := []struct {
|
||||
cpuModel string
|
||||
}{
|
||||
{"AMD EPYC 7551"},
|
||||
{"Intel Xeon E5-2686"},
|
||||
{"Apple M1"},
|
||||
{"AWS Graviton2"},
|
||||
{"Unknown CPU"},
|
||||
}
|
||||
|
||||
for _, tt := range testCases {
|
||||
t.Run(tt.cpuModel, func(t *testing.T) {
|
||||
family := getInstanceFamilyFromCPU(tt.cpuModel)
|
||||
assert.NotEmpty(t, family, "Should return a family for %s", tt.cpuModel)
|
||||
t.Logf("CPU: %s -> Family: %s", tt.cpuModel, family)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestGetAvailableInstanceTypeInfos_ResourceFiltering tests that instance types are filtered by available resources
|
||||
func TestGetAvailableInstanceTypeInfos_ResourceFiltering(t *testing.T) {
|
||||
rm := NewResourceManager()
|
||||
|
||||
// Get initial count of all available types
|
||||
allTypes := rm.GetInstanceTypeInfos()
|
||||
initialAvailable := rm.GetAvailableInstanceTypeInfos()
|
||||
|
||||
t.Logf("System has %d vCPUs, %.2f GB RAM", rm.availableVCPU, rm.availableMem)
|
||||
t.Logf("All instance types: %d, Initially available: %d", len(allTypes), len(initialAvailable))
|
||||
|
||||
// Initially available types should only include those that fit system resources
|
||||
// (on small machines, xlarge/2xlarge may already be filtered out)
|
||||
assert.LessOrEqual(t, len(initialAvailable), len(allTypes),
|
||||
"Available types should be <= total types")
|
||||
assert.Greater(t, len(initialAvailable), 0, "Should have at least one available type")
|
||||
|
||||
// Verify all initially available types fit within system resources
|
||||
for _, info := range initialAvailable {
|
||||
vcpus := int(*info.VCpuInfo.DefaultVCpus)
|
||||
memGB := float64(*info.MemoryInfo.SizeInMiB) / 1024
|
||||
|
||||
assert.LessOrEqual(t, vcpus, rm.availableVCPU,
|
||||
"Instance type %s vCPUs should fit system", *info.InstanceType)
|
||||
assert.LessOrEqual(t, memGB, rm.availableMem,
|
||||
"Instance type %s memory should fit system", *info.InstanceType)
|
||||
}
|
||||
|
||||
// Allocate the smallest instance type (nano) to consume some resources
|
||||
var nanoKey string
|
||||
var nanoType *ec2.InstanceTypeInfo
|
||||
var exists bool
|
||||
for key, it := range rm.instanceTypes {
|
||||
if strings.HasSuffix(key, ".nano") {
|
||||
nanoKey = key
|
||||
nanoType = it
|
||||
exists = true
|
||||
break
|
||||
}
|
||||
}
|
||||
require.True(t, exists, "Should have at least one .nano instance type")
|
||||
|
||||
err := rm.allocate(nanoType)
|
||||
require.NoError(t, err, "Should be able to allocate %s", nanoKey)
|
||||
|
||||
t.Logf("After allocating %s: allocated %d vCPUs, %.2f GB RAM",
|
||||
nanoKey, rm.allocatedVCPU, rm.allocatedMem)
|
||||
|
||||
// Now get available types - should be fewer or equal (depending on system resources)
|
||||
afterAllocation := rm.GetAvailableInstanceTypeInfos()
|
||||
t.Logf("Available after allocation: %d", len(afterAllocation))
|
||||
|
||||
// Verify all returned types fit within REMAINING resources
|
||||
remainingVCPU := rm.availableVCPU - rm.allocatedVCPU
|
||||
remainingMem := rm.availableMem - rm.allocatedMem
|
||||
|
||||
for _, info := range afterAllocation {
|
||||
typeName := *info.InstanceType
|
||||
vcpus := int(*info.VCpuInfo.DefaultVCpus)
|
||||
memGB := float64(*info.MemoryInfo.SizeInMiB) / 1024
|
||||
|
||||
assert.LessOrEqual(t, vcpus, remainingVCPU,
|
||||
"Instance type %s should not exceed remaining vCPUs", typeName)
|
||||
assert.LessOrEqual(t, memGB, remainingMem,
|
||||
"Instance type %s should not exceed remaining memory", typeName)
|
||||
|
||||
t.Logf("Available: %s (vCPUs: %d, Memory: %.2f GB)", typeName, vcpus, memGB)
|
||||
}
|
||||
|
||||
// Deallocate and verify we get the same available types as before
|
||||
rm.deallocate(nanoType)
|
||||
afterDeallocation := rm.GetAvailableInstanceTypeInfos()
|
||||
assert.Equal(t, len(initialAvailable), len(afterDeallocation),
|
||||
"Should have same available types after deallocation")
|
||||
}
|
||||
|
||||
// TestHandleEC2DescribeInstanceTypes tests the DescribeInstanceTypes handler
|
||||
func TestHandleEC2DescribeInstanceTypes(t *testing.T) {
|
||||
ns, natsURL := startTestNATSServer(t)
|
||||
defer ns.Shutdown()
|
||||
|
||||
daemon := createTestDaemon(t, natsURL)
|
||||
|
||||
// Subscribe to DescribeInstanceTypes (no queue group for fan-out)
|
||||
sub, err := daemon.natsConn.Subscribe("ec2.DescribeInstanceTypes", daemon.handleEC2DescribeInstanceTypes)
|
||||
require.NoError(t, err, "Failed to subscribe to ec2.DescribeInstanceTypes")
|
||||
defer sub.Unsubscribe()
|
||||
|
||||
// Test 1: Get all available instance types and verify CPU architecture
|
||||
t.Run("GetAllAvailableInstanceTypes_VerifyArchitecture", func(t *testing.T) {
|
||||
input := &ec2.DescribeInstanceTypesInput{}
|
||||
msgData, err := json.Marshal(input)
|
||||
require.NoError(t, err)
|
||||
|
||||
reply, err := daemon.natsConn.Request("ec2.DescribeInstanceTypes", msgData, 5*time.Second)
|
||||
require.NoError(t, err, "Request should succeed")
|
||||
require.NotNil(t, reply, "Should receive a reply")
|
||||
|
||||
var output ec2.DescribeInstanceTypesOutput
|
||||
err = json.Unmarshal(reply.Data, &output)
|
||||
require.NoError(t, err, "Should unmarshal response")
|
||||
|
||||
require.NotNil(t, output.InstanceTypes, "InstanceTypes should not be nil")
|
||||
assert.Greater(t, len(output.InstanceTypes), 0, "Should return at least one instance type")
|
||||
|
||||
// Verify CPU architecture is correct
|
||||
expectedArch := "x86_64"
|
||||
if runtime.GOARCH == "arm64" {
|
||||
expectedArch = "arm64"
|
||||
}
|
||||
|
||||
for _, info := range output.InstanceTypes {
|
||||
require.NotNil(t, info.ProcessorInfo, "ProcessorInfo should not be nil")
|
||||
require.NotEmpty(t, info.ProcessorInfo.SupportedArchitectures, "SupportedArchitectures should not be empty")
|
||||
assert.Equal(t, expectedArch, *info.ProcessorInfo.SupportedArchitectures[0],
|
||||
"Instance type %s should have correct architecture", *info.InstanceType)
|
||||
}
|
||||
|
||||
t.Logf("Returned %d instance types with architecture %s", len(output.InstanceTypes), expectedArch)
|
||||
})
|
||||
|
||||
// Test 2: Verify instance types match expected list
|
||||
t.Run("VerifyInstanceTypesMatchExpectedList", func(t *testing.T) {
|
||||
input := &ec2.DescribeInstanceTypesInput{}
|
||||
msgData, err := json.Marshal(input)
|
||||
require.NoError(t, err)
|
||||
|
||||
reply, err := daemon.natsConn.Request("ec2.DescribeInstanceTypes", msgData, 5*time.Second)
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, reply)
|
||||
|
||||
var output ec2.DescribeInstanceTypesOutput
|
||||
err = json.Unmarshal(reply.Data, &output)
|
||||
require.NoError(t, err)
|
||||
|
||||
// Get expected instance types from ResourceManager
|
||||
expectedTypes := daemon.resourceMgr.GetAvailableInstanceTypeInfos()
|
||||
require.NotEmpty(t, expectedTypes, "Should have expected instance types")
|
||||
|
||||
// Build map of expected instance type names
|
||||
expectedTypeMap := make(map[string]bool)
|
||||
for _, it := range expectedTypes {
|
||||
if it.InstanceType != nil {
|
||||
expectedTypeMap[*it.InstanceType] = true
|
||||
}
|
||||
}
|
||||
|
||||
// Verify all returned types are in expected list
|
||||
returnedTypeMap := make(map[string]bool)
|
||||
for _, info := range output.InstanceTypes {
|
||||
if info.InstanceType != nil {
|
||||
typeName := *info.InstanceType
|
||||
returnedTypeMap[typeName] = true
|
||||
assert.True(t, expectedTypeMap[typeName],
|
||||
"Returned instance type %s should be in expected list", typeName)
|
||||
}
|
||||
}
|
||||
|
||||
// Verify counts match (all available types should be returned)
|
||||
assert.Equal(t, len(expectedTypes), len(output.InstanceTypes),
|
||||
"Returned instance types count should match available types count")
|
||||
|
||||
t.Logf("Verified %d instance types match expected list", len(output.InstanceTypes))
|
||||
})
|
||||
|
||||
// Test 3: Filter unavailable types after allocating 2 CPUs
|
||||
t.Run("FilterUnavailableTypesAfterAllocation", func(t *testing.T) {
|
||||
// Get initial available types
|
||||
input := &ec2.DescribeInstanceTypesInput{}
|
||||
msgData, err := json.Marshal(input)
|
||||
require.NoError(t, err)
|
||||
|
||||
reply, err := daemon.natsConn.Request("ec2.DescribeInstanceTypes", msgData, 5*time.Second)
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, reply)
|
||||
|
||||
var initialOutput ec2.DescribeInstanceTypesOutput
|
||||
err = json.Unmarshal(reply.Data, &initialOutput)
|
||||
require.NoError(t, err)
|
||||
|
||||
initialCount := len(initialOutput.InstanceTypes)
|
||||
t.Logf("Initial available instance types: %d", initialCount)
|
||||
|
||||
// Find an instance type that uses 2 vCPUs
|
||||
var instanceType2CPU *ec2.InstanceTypeInfo
|
||||
var instanceTypeName string
|
||||
for _, it := range daemon.resourceMgr.instanceTypes {
|
||||
if it.VCpuInfo != nil && it.VCpuInfo.DefaultVCpus != nil && *it.VCpuInfo.DefaultVCpus == 2 {
|
||||
instanceType2CPU = it
|
||||
if it.InstanceType != nil {
|
||||
instanceTypeName = *it.InstanceType
|
||||
}
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
require.NotNil(t, instanceType2CPU, "Should find an instance type with 2 vCPUs")
|
||||
t.Logf("Allocating instance type: %s (2 vCPUs)", instanceTypeName)
|
||||
|
||||
// Allocate the 2 vCPU instance type
|
||||
err = daemon.resourceMgr.allocate(instanceType2CPU)
|
||||
require.NoError(t, err, "Should be able to allocate 2 vCPU instance")
|
||||
|
||||
// Verify allocation
|
||||
assert.Equal(t, 2, daemon.resourceMgr.allocatedVCPU, "Should have allocated 2 vCPUs")
|
||||
t.Logf("Allocated resources: %d vCPUs, %.2f GB RAM",
|
||||
daemon.resourceMgr.allocatedVCPU, daemon.resourceMgr.allocatedMem)
|
||||
|
||||
// Get available types after allocation
|
||||
reply, err = daemon.natsConn.Request("ec2.DescribeInstanceTypes", msgData, 5*time.Second)
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, reply)
|
||||
|
||||
var afterAllocationOutput ec2.DescribeInstanceTypesOutput
|
||||
err = json.Unmarshal(reply.Data, &afterAllocationOutput)
|
||||
require.NoError(t, err)
|
||||
|
||||
afterAllocationCount := len(afterAllocationOutput.InstanceTypes)
|
||||
t.Logf("Available instance types after allocation: %d", afterAllocationCount)
|
||||
|
||||
// Verify fewer types are available
|
||||
assert.LessOrEqual(t, afterAllocationCount, initialCount,
|
||||
"Should have fewer or equal instance types after allocation")
|
||||
|
||||
// Calculate remaining resources
|
||||
remainingVCPU := daemon.resourceMgr.availableVCPU - daemon.resourceMgr.allocatedVCPU
|
||||
remainingMem := daemon.resourceMgr.availableMem - daemon.resourceMgr.allocatedMem
|
||||
|
||||
t.Logf("Remaining resources: %d vCPUs, %.2f GB RAM", remainingVCPU, remainingMem)
|
||||
|
||||
// Verify all returned types fit within remaining resources
|
||||
for _, info := range afterAllocationOutput.InstanceTypes {
|
||||
require.NotNil(t, info.InstanceType, "InstanceType should not be nil")
|
||||
require.NotNil(t, info.VCpuInfo, "VCpuInfo should not be nil")
|
||||
require.NotNil(t, info.VCpuInfo.DefaultVCpus, "DefaultVCpus should not be nil")
|
||||
require.NotNil(t, info.MemoryInfo, "MemoryInfo should not be nil")
|
||||
require.NotNil(t, info.MemoryInfo.SizeInMiB, "SizeInMiB should not be nil")
|
||||
|
||||
typeName := *info.InstanceType
|
||||
vcpus := int(*info.VCpuInfo.DefaultVCpus)
|
||||
memGB := float64(*info.MemoryInfo.SizeInMiB) / 1024.0
|
||||
|
||||
assert.LessOrEqual(t, vcpus, remainingVCPU,
|
||||
"Instance type %s (%d vCPUs) should not exceed remaining vCPUs (%d)",
|
||||
typeName, vcpus, remainingVCPU)
|
||||
assert.LessOrEqual(t, memGB, remainingMem,
|
||||
"Instance type %s (%.2f GB) should not exceed remaining memory (%.2f GB)",
|
||||
typeName, memGB, remainingMem)
|
||||
|
||||
t.Logf("Available: %s (vCPUs: %d, Memory: %.2f GB)", typeName, vcpus, memGB)
|
||||
}
|
||||
|
||||
// Verify that instance types requiring more than remaining resources are NOT returned
|
||||
// Find instance types that should be filtered out
|
||||
for _, it := range daemon.resourceMgr.instanceTypes {
|
||||
if it.InstanceType == nil || it.VCpuInfo == nil || it.VCpuInfo.DefaultVCpus == nil {
|
||||
continue
|
||||
}
|
||||
|
||||
typeName := *it.InstanceType
|
||||
vcpus := int(*it.VCpuInfo.DefaultVCpus)
|
||||
memGB := float64(0)
|
||||
if it.MemoryInfo != nil && it.MemoryInfo.SizeInMiB != nil {
|
||||
memGB = float64(*it.MemoryInfo.SizeInMiB) / 1024.0
|
||||
}
|
||||
|
||||
// If this type exceeds remaining resources, it should NOT be in the response
|
||||
if vcpus > remainingVCPU || memGB > remainingMem {
|
||||
found := false
|
||||
for _, returnedInfo := range afterAllocationOutput.InstanceTypes {
|
||||
if returnedInfo.InstanceType != nil && *returnedInfo.InstanceType == typeName {
|
||||
found = true
|
||||
break
|
||||
}
|
||||
}
|
||||
assert.False(t, found,
|
||||
"Instance type %s (%d vCPUs, %.2f GB) should NOT be returned as it exceeds remaining resources",
|
||||
typeName, vcpus, memGB)
|
||||
}
|
||||
}
|
||||
|
||||
// Cleanup: deallocate
|
||||
daemon.resourceMgr.deallocate(instanceType2CPU)
|
||||
assert.Equal(t, 0, daemon.resourceMgr.allocatedVCPU, "Should have deallocated all vCPUs")
|
||||
})
|
||||
}
|
||||
|
||||
// createValidRunInstancesInput creates a valid RunInstancesInput for testing
|
||||
func createValidRunInstancesInput() *ec2.RunInstancesInput {
|
||||
return &ec2.RunInstancesInput{
|
||||
ImageId: aws.String("ami-0abcdef1234567890"),
|
||||
InstanceType: aws.String(getTestInstanceType()),
|
||||
MinCount: aws.Int64(1),
|
||||
MaxCount: aws.Int64(1),
|
||||
KeyName: aws.String("test-key"),
|
||||
SubnetId: aws.String("subnet-test"),
|
||||
UserData: aws.String(""), // Empty UserData to bypass cloud-init requirements
|
||||
}
|
||||
}
|
||||
|
||||
+22
-119
@@ -137,6 +137,28 @@ func (gw *GatewayConfig) EC2_Request(ctx *fiber.Ctx) error {
|
||||
return errors.New("failed to marshal response to XML")
|
||||
}
|
||||
|
||||
case "DescribeInstanceTypes":
|
||||
var input = &ec2.DescribeInstanceTypesInput{}
|
||||
err = awsec2query.QueryParamsToStruct(queryArgs, input)
|
||||
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
output, err := gateway_ec2_instance.DescribeInstanceTypes(input, gw.NATSConn, gw.ExpectedNodes)
|
||||
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Convert to XML
|
||||
payload := utils.GenerateXMLPayload("DescribeInstanceTypesResponse", output)
|
||||
xmlOutput, err = utils.MarshalToXML(payload)
|
||||
|
||||
if err != nil {
|
||||
return errors.New("failed to marshal response to XML")
|
||||
}
|
||||
|
||||
case "CreateKeyPair":
|
||||
|
||||
var input = &ec2.CreateKeyPairInput{}
|
||||
@@ -257,7 +279,6 @@ func (gw *GatewayConfig) EC2_Request(ctx *fiber.Ctx) error {
|
||||
if err != nil {
|
||||
return errors.New("failed to marshal response to XML")
|
||||
}
|
||||
|
||||
default:
|
||||
err = errors.New("InvalidAction")
|
||||
}
|
||||
@@ -272,121 +293,3 @@ func (gw *GatewayConfig) EC2_Request(ctx *fiber.Ctx) error {
|
||||
return nil
|
||||
|
||||
}
|
||||
|
||||
/*
|
||||
func (gw *GatewayConfig) EC2_DescribeInstances(ctx *fiber.Ctx, args map[string]string) error {
|
||||
|
||||
slog.Info("EC2 DescribeInstances called")
|
||||
|
||||
// Return a dummy response
|
||||
response := `<?xml version="1.0" encoding="UTF-8"?>
|
||||
<DescribeInstancesResponse xmlns="http://ec2.amazonaws.com/doc/2016-11-15/">
|
||||
<requestId>2b7ac2f1-9acd-4d73-b0a1-1b2e5607f9ab</requestId>
|
||||
<reservationSet>
|
||||
<item>
|
||||
<reservationId>r-0f6b3c4d5e6789abc</reservationId>
|
||||
<ownerId>123456789012</ownerId>
|
||||
<groupSet/>
|
||||
<instancesSet>
|
||||
<item>
|
||||
<instanceId>i-0123456789abcdef0</instanceId>
|
||||
<imageId>ami-08d4ac5b634553e16</imageId>
|
||||
<instanceState>
|
||||
<code>16</code>
|
||||
<name>running</name>
|
||||
</instanceState>
|
||||
<privateDnsName>ip-10-0-1-25.ap-southeast-2.compute.internal</privateDnsName>
|
||||
<dnsName>ec2-3-26-14-112.ap-southeast-2.compute.amazonaws.com</dnsName>
|
||||
<reason/>
|
||||
<keyName>my-keypair</keyName>
|
||||
<amiLaunchIndex>0</amiLaunchIndex>
|
||||
<productCodes/>
|
||||
<instanceType>t3.medium</instanceType>
|
||||
<launchTime>2025-02-18T04:12:07.000Z</launchTime>
|
||||
<placement>
|
||||
<availabilityZone>ap-southeast-2a</availabilityZone>
|
||||
<groupName/>
|
||||
<tenancy>default</tenancy>
|
||||
</placement>
|
||||
<monitoring>
|
||||
<state>disabled</state>
|
||||
</monitoring>
|
||||
<privateIpAddress>10.0.1.25</privateIpAddress>
|
||||
<ipAddress>3.26.14.112</ipAddress>
|
||||
<subnetId>subnet-0ab1c2d3e4f567890</subnetId>
|
||||
<vpcId>vpc-0123abcd4567efgh8</vpcId>
|
||||
<sourceDestCheck>true</sourceDestCheck>
|
||||
<groupSet>
|
||||
<item>
|
||||
<groupId>sg-0abc1234def567890</groupId>
|
||||
<groupName>default</groupName>
|
||||
</item>
|
||||
</groupSet>
|
||||
<stateReason/>
|
||||
<architecture>x86_64</architecture>
|
||||
<rootDeviceType>ebs</rootDeviceType>
|
||||
<rootDeviceName>/dev/xvda</rootDeviceName>
|
||||
<blockDeviceMapping>
|
||||
<item>
|
||||
<deviceName>/dev/xvda</deviceName>
|
||||
<ebs>
|
||||
<volumeId>vol-06f5e4d3c2b1a0f98</volumeId>
|
||||
<status>attached</status>
|
||||
<attachTime>2025-02-18T04:12:09.000Z</attachTime>
|
||||
<deleteOnTermination>true</deleteOnTermination>
|
||||
</ebs>
|
||||
</item>
|
||||
</blockDeviceMapping>
|
||||
<virtualizationType>hvm</virtualizationType>
|
||||
<tagSet>
|
||||
<item>
|
||||
<key>Name</key>
|
||||
<value>web-01</value>
|
||||
</item>
|
||||
</tagSet>
|
||||
<hypervisor>xen</hypervisor>
|
||||
<networkInterfaceSet>
|
||||
<item>
|
||||
<networkInterfaceId>eni-0a12b34c56d78ef90</networkInterfaceId>
|
||||
<subnetId>subnet-0ab1c2d3e4f567890</subnetId>
|
||||
<vpcId>vpc-0123abcd4567efgh8</vpcId>
|
||||
<description/>
|
||||
<ownerId>123456789012</ownerId>
|
||||
<status>in-use</status>
|
||||
<macAddress>0a:1b:2c:3d:4e:5f</macAddress>
|
||||
<privateIpAddress>10.0.1.25</privateIpAddress>
|
||||
<privateDnsName>ip-10-0-1-25.ap-southeast-2.compute.internal</privateDnsName>
|
||||
<sourceDestCheck>true</sourceDestCheck>
|
||||
<groupSet>
|
||||
<item>
|
||||
<groupId>sg-0abc1234def567890</groupId>
|
||||
<groupName>default</groupName>
|
||||
</item>
|
||||
</groupSet>
|
||||
<attachment>
|
||||
<attachmentId>eni-attach-0abc123def4567890</attachmentId>
|
||||
<deviceIndex>0</deviceIndex>
|
||||
<status>attached</status>
|
||||
<attachTime>2025-02-18T04:12:07.000Z</attachTime>
|
||||
<deleteOnTermination>true</deleteOnTermination>
|
||||
</attachment>
|
||||
<privateIpAddressesSet>
|
||||
<item>
|
||||
<privateIpAddress>10.0.1.25</privateIpAddress>
|
||||
<privateDnsName>ip-10-0-1-25.ap-southeast-2.compute.internal</privateDnsName>
|
||||
<primary>true</primary>
|
||||
</item>
|
||||
</privateIpAddressesSet>
|
||||
</item>
|
||||
</networkInterfaceSet>
|
||||
</item>
|
||||
</instancesSet>
|
||||
</item>
|
||||
</reservationSet>
|
||||
</DescribeInstancesResponse>
|
||||
`
|
||||
|
||||
return ctx.SendString(response)
|
||||
}
|
||||
|
||||
*/
|
||||
|
||||
@@ -0,0 +1,109 @@
|
||||
package gateway_ec2_instance
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"time"
|
||||
|
||||
"github.com/aws/aws-sdk-go/service/ec2"
|
||||
"github.com/mulgadc/hive/hive/utils"
|
||||
"github.com/nats-io/nats.go"
|
||||
)
|
||||
|
||||
// DescribeInstanceTypes queries all hive nodes for their instance types via NATS
|
||||
func DescribeInstanceTypes(input *ec2.DescribeInstanceTypesInput, natsConn *nats.Conn, expectedNodes int) (*ec2.DescribeInstanceTypesOutput, error) {
|
||||
// Marshal input to JSON
|
||||
jsonData, err := json.Marshal(input)
|
||||
if err != nil {
|
||||
slog.Error("DescribeInstanceTypes: Failed to marshal input", "err", err)
|
||||
return nil, fmt.Errorf("failed to marshal input: %w", err)
|
||||
}
|
||||
|
||||
// Create an inbox for collecting responses from all nodes
|
||||
inbox := nats.NewInbox()
|
||||
sub, err := natsConn.SubscribeSync(inbox)
|
||||
if err != nil {
|
||||
slog.Error("DescribeInstanceTypes: Failed to create inbox subscription", "err", err)
|
||||
return nil, fmt.Errorf("failed to create inbox: %w", err)
|
||||
}
|
||||
defer sub.Unsubscribe()
|
||||
|
||||
// Publish request to all nodes (no queue group, so all daemons receive it)
|
||||
err = natsConn.PublishRequest("ec2.DescribeInstanceTypes", inbox, jsonData)
|
||||
if err != nil {
|
||||
slog.Error("DescribeInstanceTypes: Failed to publish request", "err", err)
|
||||
return nil, fmt.Errorf("failed to publish request: %w", err)
|
||||
}
|
||||
|
||||
// Collect responses from all nodes
|
||||
// Timeout serves as a safety mechanism in case some nodes don't respond
|
||||
timeout := 3 * time.Second
|
||||
deadline := time.Now().Add(timeout)
|
||||
|
||||
var allInstanceTypes []*ec2.InstanceTypeInfo
|
||||
responsesReceived := 0
|
||||
|
||||
if expectedNodes <= 0 {
|
||||
expectedNodes = -1 // Disable early exit
|
||||
slog.Warn("DescribeInstanceTypes: ExpectedNodes not configured, using timeout-only collection")
|
||||
}
|
||||
|
||||
for time.Now().Before(deadline) {
|
||||
// Check if we've received responses from all expected nodes
|
||||
if expectedNodes > 0 && responsesReceived >= expectedNodes {
|
||||
slog.Info("DescribeInstanceTypes: Received responses from all expected nodes", "expected", expectedNodes, "received", responsesReceived)
|
||||
break
|
||||
}
|
||||
|
||||
// Calculate remaining timeout
|
||||
remaining := time.Until(deadline)
|
||||
if remaining <= 0 {
|
||||
break
|
||||
}
|
||||
|
||||
// Wait for next message with remaining timeout
|
||||
msg, err := sub.NextMsg(remaining)
|
||||
if err != nil {
|
||||
if err == nats.ErrTimeout {
|
||||
// Timeout reached, no more messages
|
||||
break
|
||||
}
|
||||
slog.Error("DescribeInstanceTypes: Error receiving message", "err", err)
|
||||
break
|
||||
}
|
||||
|
||||
// Increment response counter (even for errors, as we heard from the node)
|
||||
responsesReceived++
|
||||
|
||||
// Check if response is an error
|
||||
responseError, err := utils.ValidateErrorPayload(msg.Data)
|
||||
if err != nil {
|
||||
// Response is an error payload - log but continue collecting from other nodes
|
||||
slog.Warn("DescribeInstanceTypes: Received error from node", "code", responseError.Code, "responses_received", responsesReceived)
|
||||
continue
|
||||
}
|
||||
|
||||
// Parse the DescribeInstanceTypesOutput from this node
|
||||
var nodeOutput ec2.DescribeInstanceTypesOutput
|
||||
err = json.Unmarshal(msg.Data, &nodeOutput)
|
||||
if err != nil {
|
||||
slog.Error("DescribeInstanceTypes: Failed to unmarshal node response", "err", err)
|
||||
continue
|
||||
}
|
||||
|
||||
// Aggregate instance types from this node
|
||||
if nodeOutput.InstanceTypes != nil {
|
||||
allInstanceTypes = append(allInstanceTypes, nodeOutput.InstanceTypes...)
|
||||
slog.Info("DescribeInstanceTypes: Collected instance types from node", "count", len(nodeOutput.InstanceTypes), "responses_received", responsesReceived)
|
||||
}
|
||||
}
|
||||
|
||||
// Build final aggregated response
|
||||
output := &ec2.DescribeInstanceTypesOutput{
|
||||
InstanceTypes: allInstanceTypes,
|
||||
}
|
||||
|
||||
slog.Info("DescribeInstanceTypes: Aggregated response", "total_instance_types", len(allInstanceTypes))
|
||||
return output, nil
|
||||
}
|
||||
@@ -103,7 +103,7 @@ func (gw *GatewayConfig) SetupRoutes() *fiber.App {
|
||||
|
||||
// Add CORS middleware for browser requests
|
||||
app.Use(cors.New(cors.Config{
|
||||
AllowOrigins: "http://localhost:3000",
|
||||
AllowOrigins: "https://localhost:3000",
|
||||
AllowMethods: "GET,POST,PUT,DELETE,HEAD,OPTIONS",
|
||||
AllowHeaders: "*",
|
||||
AllowCredentials: true,
|
||||
|
||||
@@ -74,24 +74,16 @@ type CloudInitMetaData struct {
|
||||
Hostname string
|
||||
}
|
||||
|
||||
// InstanceType represents the resource requirements for an EC2 instance type
|
||||
type InstanceType struct {
|
||||
Name string
|
||||
VCPUs int
|
||||
MemoryGB float64
|
||||
Architecture string // e.g., "x86_64", "arm64"
|
||||
}
|
||||
|
||||
// InstanceServiceImpl handles daemon-side EC2 instance operations
|
||||
type InstanceServiceImpl struct {
|
||||
config *config.Config
|
||||
instanceTypes map[string]InstanceType
|
||||
instanceTypes map[string]*ec2.InstanceTypeInfo
|
||||
natsConn *nats.Conn
|
||||
instances *vm.Instances
|
||||
}
|
||||
|
||||
// NewInstanceServiceImpl creates a new instance service implementation for daemon use
|
||||
func NewInstanceServiceImpl(cfg *config.Config, instanceTypes map[string]InstanceType, nc *nats.Conn, instances *vm.Instances) *InstanceServiceImpl {
|
||||
func NewInstanceServiceImpl(cfg *config.Config, instanceTypes map[string]*ec2.InstanceTypeInfo, nc *nats.Conn, instances *vm.Instances) *InstanceServiceImpl {
|
||||
return &InstanceServiceImpl{
|
||||
config: cfg,
|
||||
instanceTypes: instanceTypes,
|
||||
|
||||
Reference in New Issue
Block a user