diff --git a/cluster.go b/cluster.go index b02bd50d2..c2062383a 100644 --- a/cluster.go +++ b/cluster.go @@ -721,7 +721,7 @@ func (c *Cluster) fragSources(to *Cluster, idx *Index) (map[string][]*internal.R // Determine if a node is being added or removed. action, diffNodeID, err := c.diff(to) if err != nil { - return nil, err + return nil, errors.Wrap(err, "diffing") } // Initialize the map with all the nodes in `to`. @@ -905,7 +905,7 @@ func (c *Cluster) Open() error { // Load topology file if it exists. if err := c.loadTopology(); err != nil { - return fmt.Errorf("load topology: %v", err) + return errors.Wrap(err, "loading topology") } c.ID = c.Topology.ClusterID @@ -1006,7 +1006,7 @@ func (c *Cluster) handleNodeAction(nodeAction nodeAction) error { if err := c.setStateAndBroadcast(ClusterStateNormal); err != nil { c.Logger.Printf("setStateAndBroadcast error: err=%s", err) } - return err + return errors.Wrap(err, "setting state") } // j.Run() runs in a goroutine because in the case where the @@ -1023,14 +1023,14 @@ func (c *Cluster) handleNodeAction(nodeAction nodeAction) error { // Make sure j.Run() didn't return an error. if eg.Wait() != nil { - return err + return errors.Wrap(err, "running job") } c.Logger.Printf("received jobResult: %s", jobResult) switch jobResult { case ResizeJobStateDone: if err := c.CompleteCurrentJob(ResizeJobStateDone); err != nil { - return err + return errors.Wrap(err, "completing finished job") } // Add/remove uri to/from the cluster. if j.action == ResizeJobActionRemove { @@ -1040,7 +1040,7 @@ func (c *Cluster) handleNodeAction(nodeAction nodeAction) error { } case ResizeJobStateAborted: if err := c.CompleteCurrentJob(ResizeJobStateAborted); err != nil { - return err + return errors.Wrap(err, "completing aborted job") } } return nil @@ -1055,7 +1055,7 @@ func (c *Cluster) setStateAndBroadcast(state string) error { func (c *Cluster) sendTo(node *Node, msg proto.Message) error { if err := c.Broadcaster.SendTo(node, msg); err != nil { - return err + return errors.Wrap(err, "sending") } return nil } @@ -1124,7 +1124,7 @@ func (c *Cluster) generateResizeJob(nodeAction nodeAction) (*ResizeJob, error) { j, err := c.generateResizeJobByAction(nodeAction) if err != nil { - return nil, err + return nil, errors.Wrap(err, "generating job") } c.Logger.Printf("generated ResizeJob: %d", j.ID) @@ -1171,7 +1171,7 @@ func (c *Cluster) generateResizeJobByAction(nodeAction nodeAction) (*ResizeJob, for _, idx := range c.Holder.Indexes() { fragSources, err := c.fragSources(toCluster, idx) if err != nil { - return nil, err + return nil, errors.Wrap(err, "getting sources") } for id, sources := range fragSources { @@ -1223,7 +1223,7 @@ func (c *Cluster) FollowResizeInstruction(instr *internal.ResizeInstruction) err // Make sure the cluster status on this node agrees with the Coordinator // before attempting a resize. if err := c.MergeClusterStatus(instr.ClusterStatus); err != nil { - return err + return errors.Wrap(err, "merging cluster status") } c.Logger.Printf("MergeClusterStatus done, start goroutine") @@ -1248,7 +1248,7 @@ func (c *Cluster) FollowResizeInstruction(instr *internal.ResizeInstruction) err // Sync the schema received in the resize instruction. c.Logger.Printf("Holder ApplySchema") if err := c.Holder.ApplySchema(instr.Schema); err != nil { - return err + return errors.Wrap(err, "applying schema") } // Create a client for calling remote nodes. @@ -1269,13 +1269,13 @@ func (c *Cluster) FollowResizeInstruction(instr *internal.ResizeInstruction) err // Create view. v, err := f.CreateViewIfNotExists(src.View) if err != nil { - return err + return errors.Wrap(err, "creating view") } // Create the local fragment. frag, err := v.CreateFragmentIfNotExists(src.Slice) if err != nil { - return err + return errors.Wrap(err, "creating fragment") } // Stream slice from remote node. @@ -1291,7 +1291,7 @@ func (c *Cluster) FollowResizeInstruction(instr *internal.ResizeInstruction) err if err == ErrFragmentNotFound { return nil } - return err + return errors.Wrap(err, "retrieving slice") } else if rd == nil { return fmt.Errorf("slice %v doesn't exist on host: %s", src.Slice, src.Node.URI) } @@ -1304,7 +1304,7 @@ func (c *Cluster) FollowResizeInstruction(instr *internal.ResizeInstruction) err } return nil }(); err != nil { - return err + return errors.Wrap(err, "copying remote slice") } } return nil @@ -1438,7 +1438,7 @@ func (j *ResizeJob) Run() error { err := j.distributeResizeInstructions() if err != nil { j.result <- ResizeJobStateAborted - return err + return errors.Wrap(err, "distributing instructions") } return nil } @@ -1475,7 +1475,7 @@ func (j *ResizeJob) distributeResizeInstructions() error { } j.Logger.Printf("send resize instructions: %v", instr) if err := j.Broadcaster.SendTo(node, instr); err != nil { - return err + return errors.Wrap(err, "sending instruction") } } return nil @@ -1581,16 +1581,16 @@ func (c *Cluster) loadTopology() error { c.Topology = NewTopology() return nil } else if err != nil { - return err + return errors.Wrap(err, "reading file") } var pb internal.Topology if err := proto.Unmarshal(buf, &pb); err != nil { - return err + return errors.Wrap(err, "unmarshalling") } top, err := decodeTopology(&pb) if err != nil { - return err + return errors.Wrap(err, "decoding") } c.Topology = top @@ -1601,13 +1601,13 @@ func (c *Cluster) loadTopology() error { func (c *Cluster) saveTopology() error { if err := os.MkdirAll(c.Path, 0777); err != nil { - return err + return errors.Wrap(err, "creating directory") } if buf, err := proto.Marshal(encodeTopology(c.Topology)); err != nil { - return err + return errors.Wrap(err, "marshalling") } else if err := ioutil.WriteFile(filepath.Join(c.Path, ".topology"), buf, 0666); err != nil { - return err + return errors.Wrap(err, "writing file") } return nil } @@ -1704,7 +1704,7 @@ func (c *Cluster) nodeJoin(node *Node) error { } if err := c.AddNode(node); err != nil { - return err + return errors.Wrap(err, "adding node for agreement") } // Only change to normal if there is no existing data. Otherwise, @@ -1742,7 +1742,7 @@ func (c *Cluster) nodeJoin(node *Node) error { // If the holder does not yet contain data, go ahead and add the node. if ok, err := c.Holder.HasData(); !ok && err == nil { if err := c.AddNode(node); err != nil { - return err + return errors.Wrap(err, "adding node") } return c.setStateAndBroadcast(ClusterStateNormal) } else if err != nil { @@ -1752,7 +1752,7 @@ func (c *Cluster) nodeJoin(node *Node) error { // If the cluster has data, we need to change to RESIZING and // kick off the resizing process. if err := c.setStateAndBroadcast(ClusterStateResizing); err != nil { - return err + return errors.Wrap(err, "broadcasting state") } c.joiningLeavingNodes <- nodeAction{node, ResizeJobActionAdd} @@ -1784,7 +1784,7 @@ func (c *Cluster) NodeLeave(node *Node) error { _, err := c.generateResizeJobByAction(nodeAction{c.nodeByID(node.ID), ResizeJobActionRemove}) if err != nil { - return err + return errors.Wrap(err, "generating job") } return c.nodeLeave(node) @@ -1802,7 +1802,7 @@ func (c *Cluster) nodeLeave(node *Node) error { // If the holder does not yet contain data, go ahead and remove the node. if ok, err := c.Holder.HasData(); !ok && err == nil { if err := c.RemoveNode(n); err != nil { - return err + return errors.Wrap(err, "removing node") } return c.setStateAndBroadcast(ClusterStateNormal) } else if err != nil { @@ -1812,7 +1812,7 @@ func (c *Cluster) nodeLeave(node *Node) error { // If the cluster has data then change state to RESIZING and // kick off the resizing process. if err := c.setStateAndBroadcast(ClusterStateResizing); err != nil { - return err + return errors.Wrap(err, "broadcasting state") } c.joiningLeavingNodes <- nodeAction{n, ResizeJobActionRemove} @@ -1836,7 +1836,7 @@ func (c *Cluster) MergeClusterStatus(cs *internal.ClusterStatus) error { // Add all nodes from the coordinator. for _, node := range officialNodes { if err := c.AddNode(node); err != nil { - return err + return errors.Wrap(err, "adding node") } } @@ -1857,7 +1857,7 @@ func (c *Cluster) MergeClusterStatus(cs *internal.ClusterStatus) error { for _, nodeID := range nodeIDsToRemove { if err := c.RemoveNode(c.nodeByID(nodeID)); err != nil { - return err + return errors.Wrap(err, "removing node") } } diff --git a/cluster_internal_test.go b/cluster_internal_test.go index da9989760..6b1f6e72b 100644 --- a/cluster_internal_test.go +++ b/cluster_internal_test.go @@ -17,6 +17,7 @@ package pilosa import ( "io/ioutil" "reflect" + "strings" "testing" "github.com/pilosa/pilosa/internal" @@ -243,8 +244,8 @@ func TestFragSources(t *testing.T) { actual, err := (test.from).fragSources(test.to, test.idx) if test.err != "" { - if err.Error() != test.err { - t.Fatalf("expected error: %s", test.err) + if !strings.Contains(err.Error(), test.err) { + t.Fatalf("expected error: %s, got: %s", test.err, err.Error()) } } else { if err != nil {