raft: async-discover

Signed-off-by: Shlomi Noach <[email protected]>

Author: shlomi-noach

go/http/api.go

@@ -208,6 +208,7 @@ func (this *HttpAPI) Instance(params martini.Params, r render.Render, req *http.
 // AsyncDiscover issues an asynchronous read on an instance. This is
 // useful for bulk loads of a new set of instances and will not block
 // if the instance is slow to respond or not reachable.
+// It will also not block the raft queue in the event ocmmunication to discover instance hangs.
 func (this *HttpAPI) AsyncDiscover(params martini.Params, r render.Render, req *http.Request, user auth.User) {
 	if !isAuthorizedForAction(req, user) {
 		Respond(r, &APIResponse{Code: ERROR, Message: "Unauthorized"})
@@ -218,7 +219,12 @@ func (this *HttpAPI) AsyncDiscover(params martini.Params, r render.Render, req *
 		Respond(r, &APIResponse{Code: ERROR, Message: err.Error()})
 		return
 	}
-	go this.Discover(params, r, req, user)
+
+	if orcraft.IsRaftEnabled() {
+		orcraft.PublishCommand("async-discover", instanceKey)
+	} else {
+		go logic.DiscoverInstance(instanceKey)
+	}
 
 	Respond(r, &APIResponse{Code: OK, Message: fmt.Sprintf("Asynchronous discovery initiated for Instance: %+v", instanceKey)})
 }

go/logic/command_applier.go

@@ -21,7 +21,7 @@ import (
 
 	"github.com/openark/orchestrator/go/inst"
 	"github.com/openark/orchestrator/go/kv"
-	"github.com/openark/orchestrator/go/raft"
+	orcraft "github.com/openark/orchestrator/go/raft"
 
 	"github.com/openark/golib/log"
 )
@@ -45,6 +45,8 @@ func (applier *CommandApplier) ApplyCommand(op string, value []byte) interface{}
 		return applier.registerNode(value)
 	case "discover":
 		return applier.discover(value)
+	case "async-discover":
+		return applier.asyncDiscover(value)
 	case "injected-pseudo-gtid":
 		return applier.injectedPseudoGTID(value)
 	case "forget":
@@ -109,6 +111,14 @@ func (applier *CommandApplier) discover(value []byte) interface{} {
 	return nil
 }
 
+// asyncDiscover discover-s in a goroutine so that the discovery cannot block the raft path
+// (e.g. a situation where communication with discovered instannce hangs could block the entire
+// raft queue; this function is therefore safer)
+func (applier *CommandApplier) asyncDiscover(value []byte) interface{} {
+	go applier.discover(value)
+	return nil
+}
+
 func (applier *CommandApplier) injectedPseudoGTID(value []byte) interface{} {
 	var clusterName string
 	if err := json.Unmarshal(value, &clusterName); err != nil {