gsadmin/spinifex
1
0
Fork 0
mirror of https://github.com/mulgadc/spinifex.git synced 2026-10-21 03:36:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: nats topic

Browse Source
This commit is contained in:
Julian Sommer
2026-02-12 08:30:23 +10:00
parent 32e2f7bf78
commit c7d832a823
4 changed files with 15 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
DESIGN.md
+1 -1
View File
@@ -227,7 +227,7 @@ type ResourceManager struct {
|-------|-------------|---------|
| `ec2.RunInstances` | `hive-workers` | Launch new instances |
| `ec2.DescribeInstances` | None (fan-out) | Query all nodes for instances |
| `ec2.cmd.{instanceID}` | `hive-events` | Per-instance commands (start/stop/terminate, attach/detach volume) |
| `ec2.cmd.{instanceID}` | None (owner only) | Per-instance commands (start/stop/terminate, attach/detach volume) |
| `ec2.CreateKeyPair` | `hive-workers` | Generate SSH keypair |
| `ec2.DescribeKeyPairs` | `hive-workers` | List SSH keypairs |
| `ec2.DescribeImages` | `hive-workers` | List AMIs |
hive/daemon/daemon.go
+3 -3
View File
@@ -624,7 +624,7 @@ func (d *Daemon) restoreInstances() {
if instance.Attributes.StopInstance || instance.Status == vm.StateStopped {
slog.Info("Instance is stopped, subscribing to NATS for start commands", "instance", instance.ID)
d.mu.Lock()
sub, err := d.natsConn.QueueSubscribe(fmt.Sprintf("ec2.cmd.%s", instance.ID), "hive-events", d.handleEC2Events)
sub, err := d.natsConn.Subscribe(fmt.Sprintf("ec2.cmd.%s", instance.ID), d.handleEC2Events)
if err != nil {
d.mu.Unlock()
slog.Error("Failed to subscribe stopped instance to NATS", "instance", instance.ID, "err", err)
@@ -704,7 +704,7 @@ func (d *Daemon) reconnectInstance(instance *vm.VM) error {
}
d.mu.Lock()
sub, err := d.natsConn.QueueSubscribe(fmt.Sprintf("ec2.cmd.%s", instance.ID), "hive-events", d.handleEC2Events)
sub, err := d.natsConn.Subscribe(fmt.Sprintf("ec2.cmd.%s", instance.ID), d.handleEC2Events)
if err != nil {
d.mu.Unlock()
// Clean up the QMP connection we just opened
@@ -1340,7 +1340,7 @@ func (d *Daemon) LaunchInstance(instance *vm.VM) (err error) {
d.mu.Lock()
defer d.mu.Unlock()
d.natsSubscriptions[instance.ID], err = d.natsConn.QueueSubscribe(fmt.Sprintf("ec2.cmd.%s", instance.ID), "hive-events", d.handleEC2Events)
d.natsSubscriptions[instance.ID], err = d.natsConn.Subscribe(fmt.Sprintf("ec2.cmd.%s", instance.ID), d.handleEC2Events)
if err != nil {
slog.Error("failed to subscribe to NATS", "err", err)
hive/daemon/daemon_handlers_test.go
+4 -6
View File
@@ -323,9 +323,8 @@ func TestHandleEC2Events_StopInstance(t *testing.T) {
QMPClient: &qmp.QMPClient{},
}
sub, err := daemon.natsConn.QueueSubscribe(
sub, err := daemon.natsConn.Subscribe(
fmt.Sprintf("ec2.cmd.%s", instanceID),
"hive-events",
daemon.handleEC2Events,
)
require.NoError(t, err)
@@ -369,9 +368,8 @@ func TestHandleEC2Events_TerminateInstance(t *testing.T) {
QMPClient: &qmp.QMPClient{},
}
sub, err := daemon.natsConn.QueueSubscribe(
sub, err := daemon.natsConn.Subscribe(
fmt.Sprintf("ec2.cmd.%s", instanceID),
"hive-events",
daemon.handleEC2Events,
)
require.NoError(t, err)
@@ -404,7 +402,7 @@ func TestHandleEC2Events_InstanceNotFound(t *testing.T) {
daemon := createTestDaemon(t, natsURL)
sub, err := daemon.natsConn.QueueSubscribe("ec2.cmd.i-nonexistent", "hive-events", daemon.handleEC2Events)
sub, err := daemon.natsConn.Subscribe("ec2.cmd.i-nonexistent", daemon.handleEC2Events)
require.NoError(t, err)
defer sub.Unsubscribe()
@@ -428,7 +426,7 @@ func TestHandleEC2Events_MalformedJSON(t *testing.T) {
daemon := createTestDaemon(t, natsURL)
sub, err := daemon.natsConn.QueueSubscribe("ec2.cmd.test", "hive-events", daemon.handleEC2Events)
sub, err := daemon.natsConn.Subscribe("ec2.cmd.test", daemon.handleEC2Events)
require.NoError(t, err)
defer sub.Unsubscribe()
hive/daemon/daemon_test.go
+7 -14
View File
@@ -1938,9 +1938,8 @@ func TestHandleEC2Events_AttachVolume(t *testing.T) {
daemon.Instances.VMS[instanceID] = instance
// Subscribe the handler to the instance's per-instance topic
sub, err := daemon.natsConn.QueueSubscribe(
sub, err := daemon.natsConn.Subscribe(
fmt.Sprintf("ec2.cmd.%s", instanceID),
"hive-events",
daemon.handleEC2Events,
)
require.NoError(t, err)
@@ -2084,9 +2083,8 @@ func TestHandleEC2Events_DetachVolume(t *testing.T) {
daemon.Instances.VMS[instanceID] = instance
// Subscribe the handler to the instance's per-instance topic
sub, err := daemon.natsConn.QueueSubscribe(
sub, err := daemon.natsConn.Subscribe(
fmt.Sprintf("ec2.cmd.%s", instanceID),
"hive-events",
daemon.handleEC2Events,
)
require.NoError(t, err)
@@ -2445,9 +2443,8 @@ func TestDetachVolume_SuccessPath(t *testing.T) {
require.NoError(t, err)
defer ebsSub.Unsubscribe()
sub, err := daemon.natsConn.QueueSubscribe(
sub, err := daemon.natsConn.Subscribe(
fmt.Sprintf("ec2.cmd.%s", instanceID),
"hive-events",
daemon.handleEC2Events,
)
require.NoError(t, err)
@@ -2584,9 +2581,8 @@ func TestDetachVolume_ForceFlag(t *testing.T) {
require.NoError(t, err)
defer ebsSub.Unsubscribe()
sub, err := daemon.natsConn.QueueSubscribe(
sub, err := daemon.natsConn.Subscribe(
fmt.Sprintf("ec2.cmd.%s", instanceID),
"hive-events",
daemon.handleEC2Events,
)
require.NoError(t, err)
@@ -2689,9 +2685,8 @@ func TestDetachVolume_BlockdevDelFailure(t *testing.T) {
}
daemon.Instances.VMS[instanceID] = instance
sub, err := daemon.natsConn.QueueSubscribe(
sub, err := daemon.natsConn.Subscribe(
fmt.Sprintf("ec2.cmd.%s", instanceID),
"hive-events",
daemon.handleEC2Events,
)
require.NoError(t, err)
@@ -2789,9 +2784,8 @@ func TestDetachVolume_SuccessWithDeviceMatch(t *testing.T) {
require.NoError(t, err)
defer ebsSub.Unsubscribe()
sub, err := daemon.natsConn.QueueSubscribe(
sub, err := daemon.natsConn.Subscribe(
fmt.Sprintf("ec2.cmd.%s", instanceID),
"hive-events",
daemon.handleEC2Events,
)
require.NoError(t, err)
@@ -2866,9 +2860,8 @@ func TestAttachVolume_ReplacesStaleEBSRequest(t *testing.T) {
require.NoError(t, err)
defer ebsSub.Unsubscribe()
sub, err := daemon.natsConn.QueueSubscribe(
sub, err := daemon.natsConn.Subscribe(
fmt.Sprintf("ec2.cmd.%s", instanceID),
"hive-events",
daemon.handleEC2Events,
)
require.NoError(t, err)