added read only policies

Author: htemelski-redis

command.go

@@ -4292,6 +4292,9 @@ func (cmd *CommandsInfoCmd) readReply(rd *proto.Reader) error {
 			}
 
 			rawTips := make(map[string]string, tipsLen)
+			if cmdInfo.ReadOnly {
+				rawTips[routing.ReadOnlyCMD] = ""
+			}
 			for f := 0; f < tipsLen; f++ {
 				tip, err := rd.ReadString()
 				if err != nil {

command_policy_resolver.go

@@ -21,10 +21,16 @@ var defaultPolicies = map[module]map[commandName]*routing.CommandPolicy{
 		"search": {
 			Request:  routing.ReqDefault,
 			Response: routing.RespDefaultKeyless,
+			Tips: map[string]string{
+				routing.ReadOnlyCMD: "",
+			},
 		},
 		"aggregate": {
 			Request:  routing.ReqDefault,
 			Response: routing.RespDefaultKeyless,
+			Tips: map[string]string{
+				routing.ReadOnlyCMD: "",
+			},
 		},
 		"dictadd": {
 			Request:  routing.ReqDefault,
@@ -33,6 +39,9 @@ var defaultPolicies = map[module]map[commandName]*routing.CommandPolicy{
 		"dictdump": {
 			Request:  routing.ReqDefault,
 			Response: routing.RespDefaultKeyless,
+			Tips: map[string]string{
+				routing.ReadOnlyCMD: "",
+			},
 		},
 		"dictdel": {
 			Request:  routing.ReqDefault,
@@ -41,10 +50,16 @@ var defaultPolicies = map[module]map[commandName]*routing.CommandPolicy{
 		"suglen": {
 			Request:  routing.ReqDefault,
 			Response: routing.RespDefaultHashSlot,
+			Tips: map[string]string{
+				routing.ReadOnlyCMD: "",
+			},
 		},
 		"cursor": {
 			Request:  routing.ReqSpecial,
 			Response: routing.RespDefaultKeyless,
+			Tips: map[string]string{
+				routing.ReadOnlyCMD: "",
+			},
 		},
 		"sugadd": {
 			Request:  routing.ReqDefault,
@@ -53,6 +68,9 @@ var defaultPolicies = map[module]map[commandName]*routing.CommandPolicy{
 		"sugget": {
 			Request:  routing.ReqDefault,
 			Response: routing.RespDefaultHashSlot,
+			Tips: map[string]string{
+				routing.ReadOnlyCMD: "",
+			},
 		},
 		"sugdel": {
 			Request:  routing.ReqDefault,
@@ -61,14 +79,23 @@ var defaultPolicies = map[module]map[commandName]*routing.CommandPolicy{
 		"spellcheck": {
 			Request:  routing.ReqDefault,
 			Response: routing.RespDefaultKeyless,
+			Tips: map[string]string{
+				routing.ReadOnlyCMD: "",
+			},
 		},
 		"explain": {
 			Request:  routing.ReqDefault,
 			Response: routing.RespDefaultKeyless,
+			Tips: map[string]string{
+				routing.ReadOnlyCMD: "",
+			},
 		},
 		"explaincli": {
 			Request:  routing.ReqDefault,
 			Response: routing.RespDefaultKeyless,
+			Tips: map[string]string{
+				routing.ReadOnlyCMD: "",
+			},
 		},
 		"aliasadd": {
 			Request:  routing.ReqDefault,
@@ -85,14 +112,23 @@ var defaultPolicies = map[module]map[commandName]*routing.CommandPolicy{
 		"info": {
 			Request:  routing.ReqDefault,
 			Response: routing.RespDefaultKeyless,
+			Tips: map[string]string{
+				routing.ReadOnlyCMD: "",
+			},
 		},
 		"tagvals": {
 			Request:  routing.ReqDefault,
 			Response: routing.RespDefaultKeyless,
+			Tips: map[string]string{
+				routing.ReadOnlyCMD: "",
+			},
 		},
 		"syndump": {
 			Request:  routing.ReqDefault,
 			Response: routing.RespDefaultKeyless,
+			Tips: map[string]string{
+				routing.ReadOnlyCMD: "",
+			},
 		},
 		"synupdate": {
 			Request:  routing.ReqDefault,
@@ -101,6 +137,9 @@ var defaultPolicies = map[module]map[commandName]*routing.CommandPolicy{
 		"profile": {
 			Request:  routing.ReqDefault,
 			Response: routing.RespDefaultKeyless,
+			Tips: map[string]string{
+				routing.ReadOnlyCMD: "",
+			},
 		},
 		"alter": {
 			Request:  routing.ReqDefault,

internal/routing/policy.go

@@ -19,6 +19,10 @@ const (
 	ReqSpecial
 )
 
+const (
+	ReadOnlyCMD string = "readonly"
+)
+
 func (p RequestPolicy) String() string {
 	switch p {
 	case ReqDefault:
@@ -133,3 +137,9 @@ type CommandPolicy struct {
 func (p *CommandPolicy) CanBeUsedInPipeline() bool {
 	return p.Request != ReqAllNodes && p.Request != ReqAllShards && p.Request != ReqMultiShard
 }
+
+func (p *CommandPolicy) IsReadOnly() bool {
+	_, readOnly := p.Tips[ReadOnlyCMD]
+	fmt.Println(readOnly)
+	return readOnly
+}

osscluster.go

@@ -944,9 +944,7 @@ func NewClusterClient(opt *ClusterOptions) *ClusterClient {
 
 	c.state = newClusterStateHolder(c.loadState)
 
-	dynamicResolver := c.NewDynamicResolver()
-	dynamicResolver.SetFallbackResolver(NewDefaultCommandPolicyResolver())
-	c.SetCommandInfoResolver(dynamicResolver)
+	c.SetCommandInfoResolver(NewDefaultCommandPolicyResolver())
 
 	c.cmdable = c.Process
 	c.initHooks(hooks{
@@ -1895,6 +1893,7 @@ func (c *ClusterClient) cmdInfo(ctx context.Context, name string) *CommandInfo {
 	if info == nil {
 		internal.Logger.Printf(cmdInfoCtx, "info for cmd=%s not found", name)
 	}
+
 	return info
 }
 

osscluster_router.go

@@ -27,7 +27,7 @@ func (c *ClusterClient) routeAndRun(ctx context.Context, cmd Cmder, node *cluste
 	}
 
 	if policy == nil {
-		return c.executeDefault(ctx, cmd, node)
+		return c.executeDefault(ctx, cmd, policy, node)
 	}
 	switch policy.Request {
 	case routing.ReqAllNodes:
@@ -39,16 +39,25 @@ func (c *ClusterClient) routeAndRun(ctx context.Context, cmd Cmder, node *cluste
 	case routing.ReqSpecial:
 		return c.executeSpecialCommand(ctx, cmd, policy, node)
 	default:
-		return c.executeDefault(ctx, cmd, node)
+		return c.executeDefault(ctx, cmd, policy, node)
 	}
 }
 
 // executeDefault handles standard command routing based on keys
-func (c *ClusterClient) executeDefault(ctx context.Context, cmd Cmder, node *clusterNode) error {
+func (c *ClusterClient) executeDefault(ctx context.Context, cmd Cmder, policy *routing.CommandPolicy, node *clusterNode) error {
 	if c.hasKeys(cmd) {
 		// execute on key based shard
 		return node.Client.Process(ctx, cmd)
 	}
+	if policy != nil {
+
+		fmt.Println(policy.Tips)
+		if c.readOnlyEnabled() && policy.IsReadOnly() {
+			fmt.Println("will execute on arbitrary node")
+			return c.executeOnArbitraryNode(ctx, cmd)
+		}
+	}
+
 	return c.executeOnArbitraryShard(ctx, cmd)
 }
 
@@ -61,6 +70,15 @@ func (c *ClusterClient) executeOnArbitraryShard(ctx context.Context, cmd Cmder)
 	return node.Client.Process(ctx, cmd)
 }
 
+// executeOnArbitraryNode routes command to an arbitrary node
+func (c *ClusterClient) executeOnArbitraryNode(ctx context.Context, cmd Cmder) error {
+	node := c.pickArbitraryNode(ctx)
+	if node == nil {
+		return errClusterNoNodes
+	}
+	return node.Client.Process(ctx, cmd)
+}
+
 // executeOnAllNodes executes command on all nodes (masters and replicas)
 func (c *ClusterClient) executeOnAllNodes(ctx context.Context, cmd Cmder, policy *routing.CommandPolicy) error {
 	state, err := c.state.Get(ctx)
@@ -252,7 +270,7 @@ func (c *ClusterClient) executeSpecialCommand(ctx context.Context, cmd Cmder, po
 	case "ft.cursor":
 		return c.executeCursorCommand(ctx, cmd)
 	default:
-		return c.executeDefault(ctx, cmd, node)
+		return c.executeDefault(ctx, cmd, policy, node)
 	}
 }
 
@@ -479,12 +497,29 @@ func (c *ClusterClient) pickArbitraryShard(ctx context.Context) *clusterNode {
 	return state.Masters[idx]
 }
 
+// pickArbitraryNode selects a master or slave shard using the configured ShardPicker
+func (c *ClusterClient) pickArbitraryNode(ctx context.Context) *clusterNode {
+	state, err := c.state.Get(ctx)
+	if err != nil || len(state.Masters) == 0 {
+		return nil
+	}
+
+	allNodes := append(state.Masters, state.Slaves...)
+
+	idx := c.opt.ShardPicker.Next(len(allNodes))
+	return allNodes[idx]
+}
+
 // hasKeys checks if a command operates on keys
 func (c *ClusterClient) hasKeys(cmd Cmder) bool {
 	firstKeyPos := cmdFirstKeyPos(cmd)
 	return firstKeyPos > 0
 }
 
+func (c *ClusterClient) readOnlyEnabled() bool {
+	return c.opt.ReadOnly
+}
+
 // setCommandValue sets the aggregated value on a command using the enum-based approach
 func (c *ClusterClient) setCommandValue(cmd Cmder, value interface{}) error {
 	// If value is nil, it might mean ExtractCommandValue couldn't extract the value