Merge pull request #13 from yunify/feature/wait_util_etcd_is_ready

block process until etcd connection is established

.gitignore

@@ -1,2 +1,6 @@
 .idea
 bin
+/infra-proxy1.etcd/
+/infra3.etcd/
+/infra1.etcd/
+/infra2.etcd/

Procfile

@@ -0,0 +1,6 @@
+# Use goreman to run `go get github.com/mattn/goreman`
+etcd1: etcd --name infra1 --listen-client-urls http://127.0.0.1:2379 --advertise-client-urls http://127.0.0.1:2379 --listen-peer-urls http://127.0.0.1:12380 --initial-advertise-peer-urls http://127.0.0.1:12380 --initial-cluster-token etcd-cluster-1 --initial-cluster 'infra1=http://127.0.0.1:12380,infra2=http://127.0.0.1:22380,infra3=http://127.0.0.1:32380' --initial-cluster-state new --enable-pprof
+etcd2: etcd --name infra2 --listen-client-urls http://127.0.0.1:22379 --advertise-client-urls http://127.0.0.1:22379 --listen-peer-urls http://127.0.0.1:22380 --initial-advertise-peer-urls http://127.0.0.1:22380 --initial-cluster-token etcd-cluster-1 --initial-cluster 'infra1=http://127.0.0.1:12380,infra2=http://127.0.0.1:22380,infra3=http://127.0.0.1:32380' --initial-cluster-state new --enable-pprof
+etcd3: etcd --name infra3 --listen-client-urls http://127.0.0.1:32379 --advertise-client-urls http://127.0.0.1:32379 --listen-peer-urls http://127.0.0.1:32380 --initial-advertise-peer-urls http://127.0.0.1:32380 --initial-cluster-token etcd-cluster-1 --initial-cluster 'infra1=http://127.0.0.1:12380,infra2=http://127.0.0.1:22380,infra3=http://127.0.0.1:32380' --initial-cluster-state new --enable-pprof
+# in future, use proxy to listen on 2379
+#proxy: etcd --name infra-proxy1 --proxy=on --listen-client-urls http://127.0.0.1:2379 --initial-cluster 'infra1=http://127.0.0.1:12380,infra2=http://127.0.0.1:22380,infra3=http://127.0.0.1:32380' --enable-pprof

backends/etcdv3/client.go

@@ -104,7 +104,9 @@ func (c *Client) Delete(nodePath string, dir bool) error {
 }
 
 func (c *Client) Sync(store store.Store, stopChan chan bool) {
-	go c.internalSync(c.prefix, store, stopChan)
+	startedChan := make(chan bool)
+	go c.internalSync(c.prefix, store, stopChan,startedChan)
+	<-startedChan
 }
 
 func (c *Client) GetMapping(nodePath string, dir bool) (interface{}, error) {
@@ -130,7 +132,9 @@ func (c *Client) DeleteMapping(nodePath string, dir bool) error {
 }
 
 func (c *Client) SyncMapping(mapping store.Store, stopChan chan bool) {
-	go c.internalSync(c.mappingPrefix, mapping, stopChan)
+	startedChan := make(chan bool)
+	go c.internalSync(c.mappingPrefix, mapping, stopChan,startedChan)
+	<-startedChan
 }
 
 func (c *Client) internalGets(prefix, nodePath string) (map[string]string, error) {
@@ -178,7 +182,7 @@ func handleGetResp(prefix string, resp *client.GetResponse, vars map[string]stri
 	return nil
 }
 
-func (c *Client) internalSync(prefix string, store store.Store, stopChan chan bool) {
+func (c *Client) internalSync(prefix string, store store.Store, stopChan chan bool, startedChan chan bool) {
 
 	var rev int64 = 0
 	init := false
@@ -215,6 +219,9 @@ func (c *Client) internalSync(prefix string, store store.Store, stopChan chan bo
 			store.PutBulk("/", val)
 			log.Info("Init store for prefix %s success.", prefix)
 			init = true
+			go func() {
+				startedChan <- true
+			}()
 		}
 		for resp := range watchChan {
 			processSyncChange(prefix, store, &resp)

backends/etcdv3/client_test.go

@@ -20,17 +20,20 @@ func TestClientSyncStop(t *testing.T) {
 	prefix := fmt.Sprintf("/prefix%v", rand.Intn(1000))
 
 	stopChan := make(chan bool)
-
+	log.Info("prefix is %s",prefix)
 	nodes := []string{"http://127.0.0.1:2379"}
 	storeClient, err := NewEtcdClient("default", prefix, nodes, "", "", "", false, "", "")
 	assert.NoError(t, err)
-
 	go func() {
-		time.Sleep(3000 * time.Millisecond)
+		time.Sleep(3 * time.Second)
 		stopChan <- true
 	}()
 
 	metastore := store.New()
 	// expect internalSync not block after stopChan has signal
-	storeClient.internalSync(prefix, metastore, stopChan)
+	startedChan := make(chan bool)
+	storeClient.internalSync(prefix, metastore, stopChan, startedChan)
+	initialized:=<-startedChan
+	log.Info(fmt.Sprint("sync status:",initialized))
+
 }

metad.go

@@ -5,14 +5,6 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"github.com/golang/gddo/httputil"
-	"github.com/gorilla/mux"
-	"github.com/yunify/metad/atomic"
-	"github.com/yunify/metad/backends"
-	"github.com/yunify/metad/log"
-	"github.com/yunify/metad/metadata"
-	"github.com/yunify/metad/util/flatmap"
-	yaml "gopkg.in/yaml.v2"
 	"io/ioutil"
 	"net"
 	"net/http"
@@ -23,6 +15,16 @@ import (
 	"strings"
 	"syscall"
 	"time"
+
+	"github.com/golang/gddo/httputil"
+	"github.com/gorilla/mux"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+	"github.com/yunify/metad/atomic"
+	"github.com/yunify/metad/backends"
+	"github.com/yunify/metad/log"
+	"github.com/yunify/metad/metadata"
+	"github.com/yunify/metad/util/flatmap"
+	yaml "gopkg.in/yaml.v2"
 )
 
 const (
@@ -107,6 +109,13 @@ func (m *Metad) initRouter() {
 
 func (m *Metad) initManageRouter() {
 	m.manageRouter.HandleFunc("/favicon.ico", http.NotFound)
+	m.manageRouter.Handle("/metrics", promhttp.Handler())
+	m.manageRouter.HandleFunc("/health", func(arg1 http.ResponseWriter, arg2 *http.Request) {
+		status := make(map[string]string)
+		status["status"] = "up"
+		result, _ := json.Marshal(status)
+		arg1.Write(result)
+	})
 
 	v1 := m.manageRouter.PathPrefix("/v1").Subrouter()
 

metad_test.go

@@ -3,14 +3,15 @@ package main
 import (
 	"encoding/json"
 	"fmt"
-	"github.com/stretchr/testify/assert"
-	"github.com/yunify/metad/log"
-	"github.com/yunify/metad/util"
 	"net/http/httptest"
 	"strconv"
 	"strings"
 	"testing"
 	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/yunify/metad/log"
+	"github.com/yunify/metad/util"
 )
 
 var (
@@ -62,13 +63,27 @@ func TestMetad(t *testing.T) {
 	data := make(map[string]interface{})
 	json.Unmarshal([]byte(dataJson), &data)
 
-	req := httptest.NewRequest("PUT", "/v1/data/", strings.NewReader(dataJson))
+	req := httptest.NewRequest("GET", "/metrics", strings.NewReader(dataJson))
 	w := httptest.NewRecorder()
 	metad.manageRouter.ServeHTTP(w, req)
 	assert.Equal(t, 200, w.Code)
 
 	time.Sleep(sleepTime)
 
+	req = httptest.NewRequest("GET", "/health", strings.NewReader(dataJson))
+	w = httptest.NewRecorder()
+	metad.manageRouter.ServeHTTP(w, req)
+	assert.Equal(t, 200, w.Code)
+
+	time.Sleep(sleepTime)
+
+	req = httptest.NewRequest("PUT", "/v1/data/", strings.NewReader(dataJson))
+	w = httptest.NewRecorder()
+	metad.manageRouter.ServeHTTP(w, req)
+	assert.Equal(t, 200, w.Code)
+
+	time.Sleep(sleepTime)
+
 	req = httptest.NewRequest("GET", "/v1/data/", nil)
 	req.Header.Set("accept", "application/json")
 	w = httptest.NewRecorder()