From 36f820c859a108fda04b289010cb713b49f8dc1a Mon Sep 17 00:00:00 2001 From: Julian Sommer <115516918+juliansommer@users.noreply.github.com> Date: Fri, 13 Mar 2026 16:01:00 +1000 Subject: [PATCH] fix: error handling (#76) --- cmd/hive/cmd/root.go | 8 ++- cmd/hive/cmd/service.go | 65 +++++++++++++++++++----- hive/daemon/daemon.go | 49 +++++++++++++----- hive/daemon/jetstream.go | 15 ++++-- hive/daemon/shutdown.go | 8 ++- hive/gateway/ec2.go | 6 ++- hive/gateway/iam.go | 6 ++- hive/qmp/qmp.go | 16 +++++- hive/services/awsgw/awsgw.go | 8 ++- hive/services/viperblockd/viperblockd.go | 38 +++++++++++--- 10 files changed, 173 insertions(+), 46 deletions(-) diff --git a/cmd/hive/cmd/root.go b/cmd/hive/cmd/root.go index a1ccf3181..ffc0b9e06 100644 --- a/cmd/hive/cmd/root.go +++ b/cmd/hive/cmd/root.go @@ -116,7 +116,11 @@ func initConfig() { appConfig, err = config.LoadConfig(cfgFile) if err != nil { - fmt.Fprintf(os.Stderr, "Warning: %v\n", err) - fmt.Fprintln(os.Stderr, "Continuing with environment variables and defaults...") + // If a config file was explicitly provided, treat load failure as fatal + if cfgFile != "" { + fmt.Fprintf(os.Stderr, "Error: failed to load config %s: %v\n", cfgFile, err) + os.Exit(1) + } + // No config specified — continue with env/defaults (e.g., hive --help) } } diff --git a/cmd/hive/cmd/service.go b/cmd/hive/cmd/service.go index 965c02e6d..514c510c4 100644 --- a/cmd/hive/cmd/service.go +++ b/cmd/hive/cmd/service.go @@ -148,7 +148,10 @@ var predastoreStartCmd = &cobra.Command{ return } - service.Start() + if _, err := service.Start(); err != nil { + fmt.Println("Error starting predastore service:", err) + os.Exit(1) + } fmt.Println("Predastore service started", service) }, @@ -167,7 +170,10 @@ var predastoreStopCmd = &cobra.Command{ return } - service.Stop() + if err = service.Stop(); err != nil { + fmt.Println("Error stopping predastore service:", err) + os.Exit(1) + } fmt.Println("Predastore service stopped") @@ -319,7 +325,10 @@ var viperblockStopCmd = &cobra.Command{ return } - service.Stop() + if err = service.Stop(); err != nil { + fmt.Println("Error stopping viperblock service:", err) + os.Exit(1) + } fmt.Println("Viperblock service stopped") @@ -363,7 +372,10 @@ var natsStartCmd = &cobra.Command{ return } - service.Start() + if _, err = service.Start(); err != nil { + fmt.Println("Error starting nats service:", err) + os.Exit(1) + } fmt.Println("NATS service started") }, } @@ -381,7 +393,10 @@ var natsStopCmd = &cobra.Command{ return } - service.Stop() + if err = service.Stop(); err != nil { + fmt.Println("Error stopping nats service:", err) + os.Exit(1) + } fmt.Println("Nats service stopped") }, @@ -448,7 +463,10 @@ var hiveStartCmd = &cobra.Command{ hiveSvc.SetConfigPath(cfgFile) } - svc.Start() + if _, err = svc.Start(); err != nil { + fmt.Println("Error starting hive service:", err) + os.Exit(1) + } fmt.Println("HIVE service started") }, } @@ -466,7 +484,10 @@ var hiveStopCmd = &cobra.Command{ return } - service.Stop() + if err = service.Stop(); err != nil { + fmt.Println("Error stopping hive service:", err) + os.Exit(1) + } fmt.Println("Hive service stopped") }, @@ -542,7 +563,10 @@ var awsgwStartCmd = &cobra.Command{ return } - service.Start() + if _, err = service.Start(); err != nil { + fmt.Println("Error starting awsgw service:", err) + os.Exit(1) + } fmt.Println("AWSGW service started") }, } @@ -560,7 +584,10 @@ var awsgwStopCmd = &cobra.Command{ return } - service.Stop() + if err = service.Stop(); err != nil { + fmt.Println("Error stopping awsgw service:", err) + os.Exit(1) + } fmt.Println("AWSGW service stopped") }, @@ -597,7 +624,10 @@ var hiveUIStartCmd = &cobra.Command{ return } - svc.Start() + if _, err = svc.Start(); err != nil { + fmt.Println("Error starting hive-ui service:", err) + os.Exit(1) + } fmt.Println("hive-ui service started") }, } @@ -615,7 +645,10 @@ var hiveUIStopCmd = &cobra.Command{ return } - svc.Stop() + if err = svc.Stop(); err != nil { + fmt.Println("Error stopping hive-ui service:", err) + os.Exit(1) + } fmt.Println("hive-ui service stopped") }, } @@ -674,7 +707,10 @@ var vpcdStartCmd = &cobra.Command{ return } - svc.Start() + if _, err = svc.Start(); err != nil { + fmt.Println("Error starting vpcd service:", err) + os.Exit(1) + } fmt.Println("vpcd service started") }, } @@ -691,7 +727,10 @@ var vpcdStopCmd = &cobra.Command{ return } - svc.Stop() + if err = svc.Stop(); err != nil { + fmt.Println("Error stopping vpcd service:", err) + os.Exit(1) + } fmt.Println("vpcd service stopped") }, } diff --git a/hive/daemon/daemon.go b/hive/daemon/daemon.go index bfa342ee8..597dd2f66 100644 --- a/hive/daemon/daemon.go +++ b/hive/daemon/daemon.go @@ -15,6 +15,7 @@ import ( "os/signal" "path/filepath" "runtime" + "runtime/debug" "strconv" "strings" "sync" @@ -22,6 +23,8 @@ import ( "syscall" "time" + "net/http" + "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" "github.com/go-chi/chi/v5" @@ -46,7 +49,6 @@ import ( "github.com/mulgadc/viperblock/viperblock" "github.com/nats-io/nats.go" "github.com/pelletier/go-toml/v2" - "net/http" ) type BlockDeviceMapping struct { @@ -910,6 +912,11 @@ func (d *Daemon) restoreInstances() { go func(inst *vm.VM) { defer wg.Done() defer func() { <-sem }() // release + defer func() { + if r := recover(); r != nil { + slog.Error("Panic during instance recovery", "instanceId", inst.ID, "panic", r, "stack", string(debug.Stack())) + } + }() slog.Info("Launching instance (recovery)", "instance", inst.ID) if err := d.LaunchInstance(inst); err != nil { slog.Error("Failed to launch instance", "instanceId", inst.ID, "err", err) @@ -1087,7 +1094,9 @@ func (d *Daemon) ClusterManager() error { } w.Header().Set("Content-Type", "application/json") - _ = json.NewEncoder(w).Encode(response) + if err := json.NewEncoder(w).Encode(response); err != nil { + slog.Error("Failed to encode health response", "error", err) + } }) // Join endpoint - accepts new nodes joining the cluster @@ -1096,10 +1105,12 @@ func (d *Daemon) ClusterManager() error { if err := json.NewDecoder(r.Body).Decode(&req); err != nil { w.Header().Set("Content-Type", "application/json") w.WriteHeader(400) - _ = json.NewEncoder(w).Encode(types.NodeJoinResponse{ + if err := json.NewEncoder(w).Encode(types.NodeJoinResponse{ Success: false, Message: "invalid request body", - }) + }); err != nil { + slog.Error("Failed to encode join error response", "error", err) + } return } @@ -1109,10 +1120,12 @@ func (d *Daemon) ClusterManager() error { if req.Node == "" || req.Region == "" || req.AZ == "" { w.Header().Set("Content-Type", "application/json") w.WriteHeader(400) - _ = json.NewEncoder(w).Encode(types.NodeJoinResponse{ + if err := json.NewEncoder(w).Encode(types.NodeJoinResponse{ Success: false, Message: "node, region, and az are required", - }) + }); err != nil { + slog.Error("Failed to encode join validation response", "error", err) + } return } @@ -1120,10 +1133,12 @@ func (d *Daemon) ClusterManager() error { if _, exists := d.clusterConfig.Nodes[req.Node]; exists { w.Header().Set("Content-Type", "application/json") w.WriteHeader(409) - _ = json.NewEncoder(w).Encode(types.NodeJoinResponse{ + if err := json.NewEncoder(w).Encode(types.NodeJoinResponse{ Success: false, Message: fmt.Sprintf("node %s already exists in cluster", req.Node), - }) + }); err != nil { + slog.Error("Failed to encode join conflict response", "error", err) + } return } @@ -1153,10 +1168,12 @@ func (d *Daemon) ClusterManager() error { slog.Error("Failed to save cluster config", "error", err) w.Header().Set("Content-Type", "application/json") w.WriteHeader(500) - _ = json.NewEncoder(w).Encode(types.NodeJoinResponse{ + if err := json.NewEncoder(w).Encode(types.NodeJoinResponse{ Success: false, Message: "failed to save cluster config", - }) + }); err != nil { + slog.Error("Failed to encode join save-error response", "error", err) + } return } @@ -1209,7 +1226,7 @@ func (d *Daemon) ClusterManager() error { } w.Header().Set("Content-Type", "application/json") - _ = json.NewEncoder(w).Encode(types.NodeJoinResponse{ + if err := json.NewEncoder(w).Encode(types.NodeJoinResponse{ Success: true, Message: fmt.Sprintf("node %s successfully joined cluster", req.Node), SharedData: sharedData, @@ -1218,7 +1235,9 @@ func (d *Daemon) ClusterManager() error { CACert: caCert, CAKey: caKey, PredastoreConfig: predastoreConfig, - }) + }); err != nil { + slog.Error("Failed to encode join success response", "error", err) + } }) // Get cluster config endpoint @@ -1226,10 +1245,12 @@ func (d *Daemon) ClusterManager() error { configHash, _ := d.computeConfigHash() w.Header().Set("Content-Type", "application/json") - _ = json.NewEncoder(w).Encode(map[string]any{ + if err := json.NewEncoder(w).Encode(map[string]any{ "config": d.clusterConfig, "config_hash": configHash, - }) + }); err != nil { + slog.Error("Failed to encode config response", "error", err) + } }) // Start HTTP server in goroutine diff --git a/hive/daemon/jetstream.go b/hive/daemon/jetstream.go index ff2c6c4a1..bd778110b 100644 --- a/hive/daemon/jetstream.go +++ b/hive/daemon/jetstream.go @@ -3,6 +3,7 @@ package daemon import ( "encoding/json" "errors" + "fmt" "log/slog" "strings" "sync" @@ -301,11 +302,14 @@ func (m *JetStreamManager) WriteShutdownMarker(nodeID string) error { if m.clusterKV == nil { return errors.New("cluster state KV not initialized") } - data, _ := json.Marshal(map[string]any{ + data, err := json.Marshal(map[string]any{ "node": nodeID, "timestamp": time.Now().UTC().Format(time.RFC3339), }) - _, err := m.clusterKV.Put("shutdown."+nodeID, data) + if err != nil { + return fmt.Errorf("marshal shutdown marker: %w", err) + } + _, err = m.clusterKV.Put("shutdown."+nodeID, data) return err } @@ -341,14 +345,17 @@ func (m *JetStreamManager) WriteServiceManifest(nodeID string, services []string if m.clusterKV == nil { return errors.New("cluster state KV not initialized") } - data, _ := json.Marshal(map[string]any{ + data, err := json.Marshal(map[string]any{ "node": nodeID, "services": services, "nats_host": natsHost, "predastore_host": predastoreHost, "timestamp": time.Now().UTC().Format(time.RFC3339), }) - _, err := m.clusterKV.Put("node."+nodeID+".services", data) + if err != nil { + return fmt.Errorf("marshal service manifest: %w", err) + } + _, err = m.clusterKV.Put("node."+nodeID+".services", data) return err } diff --git a/hive/daemon/shutdown.go b/hive/daemon/shutdown.go index 8322d7f16..d890630c2 100644 --- a/hive/daemon/shutdown.go +++ b/hive/daemon/shutdown.go @@ -53,6 +53,7 @@ func (d *Daemon) handleShutdownGate(msg *nats.Msg) { var req ShutdownRequest if err := json.Unmarshal(msg.Data, &req); err != nil { slog.Error("handleShutdownGate: failed to unmarshal request", "error", err) + d.respondShutdownACK(msg, ShutdownACK{Node: d.node, Phase: "gate", Error: err.Error()}) return } @@ -105,6 +106,7 @@ func (d *Daemon) handleShutdownDrain(msg *nats.Msg) { var req ShutdownRequest if err := json.Unmarshal(msg.Data, &req); err != nil { slog.Error("handleShutdownDrain: failed to unmarshal request", "error", err) + d.respondShutdownACK(msg, ShutdownACK{Node: d.node, Phase: "drain", Error: err.Error()}) return } @@ -162,6 +164,7 @@ func (d *Daemon) handleShutdownStorage(msg *nats.Msg) { var req ShutdownRequest if err := json.Unmarshal(msg.Data, &req); err != nil { slog.Error("handleShutdownStorage: failed to unmarshal request", "error", err) + d.respondShutdownACK(msg, ShutdownACK{Node: d.node, Phase: "storage", Error: err.Error()}) return } @@ -196,6 +199,7 @@ func (d *Daemon) handleShutdownPersist(msg *nats.Msg) { var req ShutdownRequest if err := json.Unmarshal(msg.Data, &req); err != nil { slog.Error("handleShutdownPersist: failed to unmarshal request", "error", err) + d.respondShutdownACK(msg, ShutdownACK{Node: d.node, Phase: "persist", Error: err.Error()}) return } @@ -226,6 +230,7 @@ func (d *Daemon) handleShutdownInfra(msg *nats.Msg) { var req ShutdownRequest if err := json.Unmarshal(msg.Data, &req); err != nil { slog.Error("handleShutdownInfra: failed to unmarshal request", "error", err) + d.respondShutdownACK(msg, ShutdownACK{Node: d.node, Phase: "infra", Error: err.Error()}) return } @@ -290,9 +295,10 @@ func (d *Daemon) publishShutdownProgress(phase string, total, remaining int) { } data, err := json.Marshal(progress) if err != nil { + slog.Error("Failed to marshal shutdown progress", "error", err) return } if err := d.natsConn.Publish("hive.cluster.shutdown.progress", data); err != nil { - slog.Debug("Failed to publish shutdown progress", "error", err) + slog.Warn("Failed to publish shutdown progress", "error", err) } } diff --git a/hive/gateway/ec2.go b/hive/gateway/ec2.go index 65626ea9c..e942c7d3c 100644 --- a/hive/gateway/ec2.go +++ b/hive/gateway/ec2.go @@ -234,7 +234,11 @@ var ec2LocalActions = map[string]bool{ } func (gw *GatewayConfig) EC2_Request(w http.ResponseWriter, r *http.Request) error { - body, _ := io.ReadAll(r.Body) + body, err := io.ReadAll(r.Body) + if err != nil { + slog.Error("Failed to read EC2 request body", "error", err) + return errors.New(awserrors.ErrorInternalError) + } queryArgs := ParseAWSQueryArgs(string(body)) action := queryArgs["Action"] diff --git a/hive/gateway/iam.go b/hive/gateway/iam.go index 7f8e5c5fe..e18803510 100644 --- a/hive/gateway/iam.go +++ b/hive/gateway/iam.go @@ -94,7 +94,11 @@ var iamActions = map[string]IAMHandler{ } func (gw *GatewayConfig) IAM_Request(w http.ResponseWriter, r *http.Request) error { - body, _ := io.ReadAll(r.Body) + body, err := io.ReadAll(r.Body) + if err != nil { + slog.Error("Failed to read IAM request body", "error", err) + return errors.New(awserrors.ErrorInternalError) + } queryArgs := ParseAWSQueryArgs(string(body)) action := queryArgs["Action"] diff --git a/hive/qmp/qmp.go b/hive/qmp/qmp.go index 80739b461..89e9f0d70 100644 --- a/hive/qmp/qmp.go +++ b/hive/qmp/qmp.go @@ -2,8 +2,10 @@ package qmp import ( "encoding/json" + "fmt" "net" "sync" + "time" ) type unmarshalTarget any @@ -117,7 +119,11 @@ func NewQMPClient(path string) (*QMPClient, error) { return nil, err } - //conn.SetReadDeadline(time.Now().Add(30 * time.Second)) + // Set deadline for greeting — a hung QEMU would block forever without this + if err := conn.SetReadDeadline(time.Now().Add(30 * time.Second)); err != nil { + _ = conn.Close() + return nil, fmt.Errorf("set read deadline: %w", err) + } client := &QMPClient{ Conn: conn, @@ -129,7 +135,13 @@ func NewQMPClient(path string) (*QMPClient, error) { var greeting QMPGreeting if err := client.Decoder.Decode(&greeting); err != nil { _ = conn.Close() - return nil, err + return nil, fmt.Errorf("waiting for QMP greeting: %w", err) + } + + // Clear deadline for normal operations + if err := conn.SetReadDeadline(time.Time{}); err != nil { + _ = conn.Close() + return nil, fmt.Errorf("clear read deadline: %w", err) } // enable capabilities diff --git a/hive/services/awsgw/awsgw.go b/hive/services/awsgw/awsgw.go index 6b376a0e3..8494cb04f 100644 --- a/hive/services/awsgw/awsgw.go +++ b/hive/services/awsgw/awsgw.go @@ -96,7 +96,9 @@ func launchService(config *config.ClusterConfig) error { // First boot: consume bootstrap.json → seed IAM users into NATS KV → delete file bootstrapPath := filepath.Join(nodeConfig.BaseDir, "config", "bootstrap.json") - if data, err := handlers_iam.LoadBootstrapData(bootstrapPath); err == nil { + data, err := handlers_iam.LoadBootstrapData(bootstrapPath) + switch { + case err == nil: slog.Info("Bootstrap file found, seeding IAM users") if err := iamService.SeedBootstrap(data); err != nil { return fmt.Errorf("seed bootstrap from bootstrap.json: %w", err) @@ -105,6 +107,10 @@ func launchService(config *config.ClusterConfig) error { slog.Warn("Failed to delete bootstrap file", "path", bootstrapPath, "err", err) } slog.Info("Bootstrap complete, bootstrap.json deleted") + case os.IsNotExist(err): + // No bootstrap file — normal after first boot + default: + return fmt.Errorf("load bootstrap from %s: %w", bootstrapPath, err) } // Create gateway with NATS connection diff --git a/hive/services/viperblockd/viperblockd.go b/hive/services/viperblockd/viperblockd.go index f99ba2f6d..aee848c85 100644 --- a/hive/services/viperblockd/viperblockd.go +++ b/hive/services/viperblockd/viperblockd.go @@ -240,7 +240,7 @@ func launchService(cfg *Config) (err error) { slog.Error("Failed to respond to ebs.delete request", "err", err) } }); err != nil { - slog.Error("Failed to subscribe to ebs.delete", "err", err) + return fmt.Errorf("failed to subscribe to ebs.delete: %w", err) } // Subscribe to node-specific unmount topic if NodeName is set, otherwise fall back to generic queue group @@ -261,7 +261,11 @@ func launchService(cfg *Config) (err error) { var ebsRequest types.EBSRequest err := json.Unmarshal(msg.Data, &ebsRequest) if err != nil { - slog.Error("Failed to unmarshal message", "err", err) + slog.Error("Failed to unmarshal ebs.unmount message", "err", err) + errResp, _ := json.Marshal(types.EBSUnMountResponse{Error: fmt.Sprintf("bad request: %v", err)}) + if err := msg.Respond(errResp); err != nil { + slog.Error("Failed to respond to ebs.unmount request", "err", err) + } return } @@ -335,7 +339,7 @@ func launchService(cfg *Config) (err error) { slog.Error("Failed to publish ebs.unmount.response", "err", err) } }); err != nil { - slog.Error("Failed to subscribe to unmount topic", "topic", unmountTopic, "err", err) + return fmt.Errorf("failed to subscribe to %s: %w", unmountTopic, err) } if _, err := nc.QueueSubscribe("ebs.sync", "hive-workers", func(msg *nats.Msg) { @@ -389,7 +393,7 @@ func launchService(cfg *Config) (err error) { slog.Error("Failed to respond to ebs.sync request", "err", err) } }); err != nil { - slog.Error("Failed to subscribe to ebs.sync", "err", err) + return fmt.Errorf("failed to subscribe to ebs.sync: %w", err) } // Note: ebs.snapshot is handled per-volume via ebs.snapshot.{volumeID} topics, @@ -414,7 +418,11 @@ func launchService(cfg *Config) (err error) { var ebsRequest types.EBSRequest err := json.Unmarshal(msg.Data, &ebsRequest) if err != nil { - slog.Error("Failed to unmarshal message", "err", err) + slog.Error("Failed to unmarshal ebs.mount message", "err", err) + errResp, _ := json.Marshal(types.EBSMountResponse{Error: fmt.Sprintf("bad request: %v", err)}) + if err := msg.Respond(errResp); err != nil { + slog.Error("Failed to respond to ebs.mount request", "err", err) + } return } @@ -546,6 +554,14 @@ func launchService(cfg *Config) (err error) { nbdPort, err = strconv.Atoi(parts[len(parts)-1]) if err != nil { slog.Error("Failed to convert port to int", "err", err) + ebsResponse.Error = fmt.Sprintf("failed to parse port: %v", err) + response, _ := json.Marshal(ebsResponse) + if err := msg.Respond(response); err != nil { + slog.Error("Failed to respond to ebs.mount request", "err", err) + } + if err := nc.Publish("ebs.mount.response", response); err != nil { + slog.Error("Failed to publish ebs.mount.response", "err", err) + } return } @@ -573,7 +589,15 @@ func launchService(cfg *Config) (err error) { // Generate PID file for nbdkit process nbdPidFile, err := utils.GeneratePidFile(fmt.Sprintf("nbdkit-vol-%s", ebsRequest.Name)) if err != nil { - slog.Error("Failed to generate nbdkit pid file:", "err", err) + slog.Error("Failed to generate nbdkit pid file", "err", err) + ebsResponse.Error = fmt.Sprintf("failed to generate pid file: %v", err) + response, _ := json.Marshal(ebsResponse) + if err := msg.Respond(response); err != nil { + slog.Error("Failed to respond to ebs.mount request", "err", err) + } + if err := nc.Publish("ebs.mount.response", response); err != nil { + slog.Error("Failed to publish ebs.mount.response", "err", err) + } return } @@ -709,7 +733,7 @@ func launchService(cfg *Config) (err error) { slog.Debug("Sent ebs.mount response", "response", string(response)) }); err != nil { - slog.Error("Failed to subscribe to mount topic", "topic", mountTopic, "err", err) + return fmt.Errorf("failed to subscribe to %s: %w", mountTopic, err) } // Create a channel to receive shutdown signals