slack-19.0: support consul topo stale reads

Signed-off-by: Tim Vaillancourt <[email protected]>

go/vt/topo/consultopo/file.go

@@ -87,7 +87,9 @@ func (s *Server) Update(ctx context.Context, filePath string, contents []byte, v
 func (s *Server) Get(ctx context.Context, filePath string) ([]byte, topo.Version, error) {
 	nodePath := path.Join(s.root, filePath)
 
-	pair, _, err := s.kv.Get(nodePath, nil)
+	pair, _, err := s.kv.Get(nodePath, &api.QueryOptions{
+		AllowStale: consulAllowStaleReads,
+	})
 	if err != nil {
 		return nil, nil, err
 	}
@@ -102,7 +104,9 @@ func (s *Server) Get(ctx context.Context, filePath string) ([]byte, topo.Version
 func (s *Server) List(ctx context.Context, filePathPrefix string) ([]topo.KVInfo, error) {
 	nodePathPrefix := path.Join(s.root, filePathPrefix)
 
-	pairs, _, err := s.kv.List(nodePathPrefix, nil)
+	pairs, _, err := s.kv.List(nodePathPrefix, &api.QueryOptions{
+		AllowStale: consulAllowStaleReads,
+	})
 	if err != nil {
 		return []topo.KVInfo{}, err
 	}

go/vt/topo/consultopo/server.go

@@ -45,6 +45,7 @@ var (
 	consulMaxConnsPerHost   int = 250 // do not use client default of 0/unlimited
 	consulMaxIdleConns      int
 	consulIdleConnTimeout   time.Duration
+	consulAllowStaleReads   bool
 )
 
 func init() {
@@ -64,6 +65,7 @@ func registerServerFlags(fs *pflag.FlagSet) {
 	fs.IntVar(&consulMaxConnsPerHost, "topo_consul_max_conns_per_host", consulMaxConnsPerHost, "Maximum number of consul connections per host.")
 	fs.IntVar(&consulMaxIdleConns, "topo_consul_max_idle_conns", defaultConsulPooledTransport.MaxIdleConns, "Maximum number of idle consul connections.")
 	fs.DurationVar(&consulIdleConnTimeout, "topo_consul_idle_conn_timeout", defaultConsulPooledTransport.IdleConnTimeout, "Maximum amount of time to pool idle connections.")
+	fs.BoolVar(&consulAllowStaleReads, "topo_consul_allow_stale_reads", consulAllowStaleReads, "Allow stale reads from consul servers")
 }
 
 // ClientAuthCred credential to use for consul clusters