Merge remote-tracking branch 'couchbase/unstable' into HEAD

http: //ci2i-unstable.northscale.in/gsi-17.06.2020-05.30.pass.html
Change-Id: Ide39724c434d5f3cb46495cf2db1200999a83f71

secondary/cmd/cbindexperf/executor.go

@@ -206,7 +206,7 @@ func RunCommands(cluster string, cfg *Config, statsW io.Writer) (*Result, error)
 	}
 
 	time.Sleep(time.Second * time.Duration(cfg.ClientBootTime))
-	indexes, _, _, err := client.Refresh()
+	indexes, _, _, _, err := client.Refresh()
 	if err != nil {
 		return nil, err
 	}

secondary/common/index.go

@@ -652,6 +652,8 @@ func IsValidIndexType(t string) bool {
 func IsEquivalentIndex(d1, d2 *IndexDefn) bool {
 
 	if d1.Bucket != d2.Bucket ||
+		d1.Scope != d2.Scope ||
+		d1.Collection != d2.Collection ||
 		d1.IsPrimary != d2.IsPrimary ||
 		d1.ExprType != d2.ExprType ||
 		d1.PartitionScheme != d2.PartitionScheme ||

secondary/indexer/ddl_service_manager.go

@@ -113,7 +113,7 @@ func NewDDLServiceMgr(indexerId common.IndexerId, supvCmdch MsgChannel, supvMsgc
 		donech:          nil,
 		killch:          make(chan bool),
 		allowDDL:        true,
-		commandListener: mc.NewCommandListener(donech, true, false, false, false, false),
+		commandListener: mc.NewCommandListener(donech, true, false, false, false, false, false),
 		listenerDonech:  donech,
 	}
 
@@ -125,6 +125,7 @@ func NewDDLServiceMgr(indexerId common.IndexerId, supvCmdch MsgChannel, supvMsgc
 	mux.HandleFunc("/listDeleteTokens", mgr.handleListDeleteTokens)
 	mux.HandleFunc("/listDropInstanceTokens", mgr.handleListDropInstanceTokens)
 	mux.HandleFunc("/listScheduleCreateTokens", mgr.handleListScheduleCreateTokens)
+	mux.HandleFunc("/listStopScheduleCreateTokens", mgr.handleListStopScheduleCreateTokens)
 
 	go mgr.run()
 
@@ -862,7 +863,7 @@ func (m *DDLServiceMgr) processCreateCommand() {
 		case _, ok := <-m.listenerDonech:
 			if !ok {
 				m.listenerDonech = make(chan bool)
-				m.commandListener = mc.NewCommandListener(m.listenerDonech, true, false, false, false, false)
+				m.commandListener = mc.NewCommandListener(m.listenerDonech, true, false, false, false, false, false)
 				m.commandListener.ListenTokens()
 			}
 
@@ -1335,7 +1336,46 @@ func (m *DDLServiceMgr) handleListScheduleCreateTokens(w http.ResponseWriter, r
 
 		buf, err := mc.MarshallScheduleCreateTokenList(list)
 		if err != nil {
-			logging.Errorf("DDLServiceMgr::handleListDropInstanceTokens Error %v", err)
+			logging.Errorf("DDLServiceMgr::handleListScheduleCreateTokens Error %v", err)
+			w.WriteHeader(http.StatusInternalServerError)
+			w.Write([]byte(err.Error() + "\n"))
+			return
+		}
+
+		w.WriteHeader(http.StatusOK)
+		w.Write(buf)
+	}
+}
+
+func (m *DDLServiceMgr) handleListStopScheduleCreateTokens(w http.ResponseWriter, r *http.Request) {
+
+	if !m.validateAuth(w, r) {
+		logging.Errorf("DDLServiceMgr::handleListStopScheduleCreateTokens Validation Failure for Request %v", r)
+		return
+	}
+
+	if r.Method == "GET" {
+
+		logging.Infof("DDLServiceMgr::handleListStopScheduleCreateTokens Processing Request %v", r)
+
+		scheduleTokens, err := mc.ListAllStopScheduleCreateTokens()
+		if err != nil {
+			logging.Errorf("DDLServiceMgr::handleListStopScheduleCreateTokens Error %v", err)
+			w.WriteHeader(http.StatusInternalServerError)
+			w.Write([]byte(err.Error() + "\n"))
+			return
+		}
+
+		list := &mc.StopScheduleCreateTokenList{}
+		list.Tokens = make([]mc.StopScheduleCreateToken, 0, len(scheduleTokens))
+
+		for _, token := range scheduleTokens {
+			list.Tokens = append(list.Tokens, *token)
+		}
+
+		buf, err := mc.MarshallStopScheduleCreateTokenList(list)
+		if err != nil {
+			logging.Errorf("DDLServiceMgr::handleListStopScheduleCreateTokens Error %v", err)
 			w.WriteHeader(http.StatusInternalServerError)
 			w.Write([]byte(err.Error() + "\n"))
 			return

secondary/indexer/indexer.go

@@ -5396,19 +5396,15 @@ func (idx *indexer) startKeyspaceIdStream(streamId common.StreamId, keyspaceId s
 	numVb := idx.config["numVbuckets"].Int()
 	enableAsync := idx.config["enableAsyncOpenStream"].Bool()
 
-	var clusterVer uint64
+	clusterVer := idx.clusterInfoClient.ClusterVersion()
 	if !inRepair {
-		clusterVer = idx.clusterInfoClient.ClusterVersion()
 		async = enableAsync && clusterVer >= common.INDEXER_65_VERSION
 	}
 
 	var cid string
 	var ok bool
 	if cid, ok = idx.streamKeyspaceIdCollectionId[streamId][keyspaceId]; !ok {
 		//if the cid has not been set e.g. in warmup, set it from the first index
-		if clusterVer == 0 {
-			clusterVer = clusterVersion(clustAddr)
-		}
 		//get cid of any index and determine if it needs to be used
 		cid = indexList[0].Defn.CollectionId
 		cid = idx.makeCollectionIdForStreamRequest(streamId, keyspaceId, cid, clusterVer)

secondary/indexer/restapi.go

@@ -76,7 +76,7 @@ func (api *testServer) validateAuth(w http.ResponseWriter, r *http.Request) (cba
 
 func (api *testServer) authorize(w http.ResponseWriter, creds cbauth.Creds) bool {
 
-	indexes, _, _, err := api.client.Refresh()
+	indexes, _, _, _, err := api.client.Refresh()
 	if err != nil {
 		log.Errorf("Fail to authorize.  Reason: unable to fetch index metadata.  %v", err)
 		http.Error(w, jsonstr("Authroziation check fails", err), http.StatusBadRequest)
@@ -456,7 +456,7 @@ func (api *testServer) doBuildOne(w http.ResponseWriter, request *http.Request)
 
 //GET    /internal/index/{id}
 func (api *testServer) doGetAll(w http.ResponseWriter, request *http.Request) {
-	indexes, _, _, err := api.client.Refresh()
+	indexes, _, _, _, err := api.client.Refresh()
 	if err != nil {
 		msg := `cannot refresh metadata: %v`
 		http.Error(w, jsonstr(msg, err), http.StatusBadRequest)
@@ -1290,7 +1290,7 @@ func (api *testServer) getIndex(path string) (*mclient.IndexMetadata, string) {
 		return nil, jsonstr(`invalid index id, ParseUint failed %v`, err)
 	}
 
-	indexes, _, _, err := api.client.Refresh()
+	indexes, _, _, _, err := api.client.Refresh()
 	if err != nil {
 		return nil, jsonstr(`cannot refresh metadata: %v`, err)
 	}