Merge pull request #12 from chai2010/dev

refactoring module struct

Author: chai2010

.travis.yml

@@ -11,5 +11,11 @@ go:
 go_import_path: openpitrix.io/libconfd
 
 script:
-- env GO111MODULE=on go test ./...
-- env GO111MODULE=on go vet ./...
+- env GO111MODULE=on go test openpitrix.io/libconfd/...
+- env GO111MODULE=on go vet openpitrix.io/libconfd/...
+
+- env GO111MODULE=on go test openpitrix.io/libconfd/etcdv3/...
+- env GO111MODULE=on go vet openpitrix.io/libconfd/etcdv3/...
+
+- env GO111MODULE=on go test openpitrix.io/libconfd/metad/...
+- env GO111MODULE=on go vet openpitrix.io/libconfd/metad/...

README.md

@@ -33,4 +33,4 @@ func main() {
 $ go run miniconfd.go -h
 ```
 
-See [miniconfd.go](miniconfd.go)
+See [etcdv3/miniconfd-etcdv3/miniconfd.go](etcdv3/miniconfd-etcdv3/miniconfd.go)

confd-backend.toml

@@ -7,7 +7,7 @@ type = "libconfd-backend-toml"
 
 # backend address
 host = [
-	"./confd/backend-file.toml",
+	"./testdata/confd/backend-file.toml",
 ]
 
 # user name/password

confd.toml

@@ -7,7 +7,7 @@
 #
 # abspath = filepath.Join(ConfigPath, Config.ConfDir)
 #
-confdir = "./confd"
+confdir = "./testdata/confd"
 
 # The backend polling interval in seconds. (10)
 interval = 10

etcdv3/backends/etcdv3.go renamed to etcdv3/backend/etcdv3.go

@@ -2,7 +2,7 @@
 // Use of this source code is governed by a Apache license
 // that can be found in the LICENSE file.
 
-package backends
+package backend_etcdv3
 
 import (
 	"context"

etcdv3/miniconfd-etcdv3/miniconfd.go

@@ -12,7 +12,7 @@ import (
 	"github.com/urfave/cli"
 
 	"openpitrix.io/libconfd"
-	_ "openpitrix.io/libconfd/etcdv3/backends"
+	_ "openpitrix.io/libconfd/etcdv3/backend"
 )
 
 func main() {

metad/backend/metad.go

@@ -0,0 +1,267 @@
+// Copyright 2018 The OpenPitrix Authors. All rights reserved.
+// Use of this source code is governed by a Apache license
+// that can be found in the LICENSE file.
+
+// Copyright 2018 Yunify Inc. All rights reserved.
+// Use of this source code is governed by a Apache license
+// that can be found in the LICENSE file.
+
+// copy from metad/pkg/client/client.go
+
+package backend_metad
+
+import (
+	"container/ring"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io/ioutil"
+	"math/rand"
+	"net"
+	"net/http"
+	"reflect"
+	"strconv"
+	"strings"
+	"sync/atomic"
+	"time"
+
+	"openpitrix.io/libconfd"
+)
+
+var (
+	_      libconfd.BackendClient = (*MetadClient)(nil)
+	logger                        = libconfd.GetLogger()
+)
+
+const MetadBackendType = "libconfd-backend-metad"
+
+func init() {
+	libconfd.RegisterBackendClient(
+		MetadBackendType,
+		func(cfg *libconfd.BackendConfig) (libconfd.BackendClient, error) {
+			return NewMetadClient(cfg.Host)
+		},
+	)
+}
+
+type MetadClient struct {
+	connections *ring.Ring
+	current     *MetadConnection
+}
+
+type MetadConnection struct {
+	url        string
+	httpClient *http.Client
+	waitIndex  uint64
+	errTimes   uint32
+}
+
+func (c *MetadClient) Type() string       { return MetadBackendType }
+func (c *MetadClient) WatchEnabled() bool { return true }
+func (c *MetadClient) Close() error       { return nil }
+
+func NewMetadClient(backendNodes []string) (*MetadClient, error) {
+	connections := ring.New(len(backendNodes))
+	for _, backendNode := range backendNodes {
+		url := "http://" + backendNode
+		connection := &MetadConnection{
+			url: url,
+			httpClient: &http.Client{
+				Transport: &http.Transport{
+					Proxy: http.ProxyFromEnvironment,
+					DialContext: (&net.Dialer{
+						KeepAlive: 1 * time.Second,
+						DualStack: true,
+					}).DialContext,
+				},
+			},
+		}
+		connections.Value = connection
+		connections = connections.Next()
+	}
+
+	client := &MetadClient{
+		connections: connections,
+	}
+
+	err := client.selectConnection()
+
+	return client, err
+}
+
+func (c *MetadClient) selectConnection() error {
+	maxTime := 15 * time.Second
+	i := 1 * time.Second
+	for ; i < maxTime; i *= time.Duration(2) {
+		if conn, err := c.testConnection(); err == nil {
+			//found available conn
+			if c.current != nil {
+				atomic.StoreUint32(&c.current.errTimes, 0)
+			}
+			c.current = conn
+			break
+		}
+		time.Sleep(i)
+	}
+	if i >= maxTime {
+		return fmt.Errorf("fail to connect any backend.")
+	}
+	logger.Info("Using Metad URL: " + c.current.url)
+	return nil
+}
+
+func (c *MetadClient) testConnection() (*MetadConnection, error) {
+	//random start
+	if c.current == nil {
+		rand.Seed(time.Now().Unix())
+		r := rand.Intn(c.connections.Len())
+		c.connections = c.connections.Move(r)
+	}
+	c.connections = c.connections.Next()
+	conn := c.connections.Value.(*MetadConnection)
+	startConn := conn
+	_, err := conn.makeMetaDataRequest("/")
+	for err != nil {
+		logger.Error("connection to [%s], error: [%v]", conn.url, err)
+		c.connections = c.connections.Next()
+		conn = c.connections.Value.(*MetadConnection)
+		if conn == startConn {
+			break
+		}
+		_, err = conn.makeMetaDataRequest("/")
+	}
+	return conn, err
+}
+
+func (c *MetadClient) GetValues(keys []string) (map[string]string, error) {
+	vars := map[string]string{}
+
+	for _, key := range keys {
+		body, err := c.current.makeMetaDataRequest(key)
+		if err != nil {
+			atomic.AddUint32(&c.current.errTimes, 1)
+			return vars, err
+		}
+
+		var jsonResponse interface{}
+		if err = json.Unmarshal(body, &jsonResponse); err != nil {
+			return vars, err
+		}
+
+		if err = treeWalk(key, jsonResponse, vars); err != nil {
+			return vars, err
+		}
+	}
+	return vars, nil
+}
+
+func treeWalk(root string, val interface{}, vars map[string]string) error {
+	switch val.(type) {
+	case map[string]interface{}:
+		for k := range val.(map[string]interface{}) {
+			treeWalk(strings.Join([]string{root, k}, "/"), val.(map[string]interface{})[k], vars)
+		}
+	case []interface{}:
+		for i, item := range val.([]interface{}) {
+			idx := strconv.Itoa(i)
+			if i, isMap := item.(map[string]interface{}); isMap {
+				if name, exists := i["name"]; exists {
+					idx = name.(string)
+				}
+			}
+
+			treeWalk(strings.Join([]string{root, idx}, "/"), item, vars)
+		}
+	case bool:
+		vars[root] = strconv.FormatBool(val.(bool))
+	case string:
+		vars[root] = val.(string)
+	case float64:
+		vars[root] = strconv.FormatFloat(val.(float64), 'f', -1, 64)
+	case nil:
+		vars[root] = "null"
+	default:
+		logger.Error("Unknown type: " + reflect.TypeOf(val).Name())
+	}
+	return nil
+}
+
+func (c *MetadClient) WatchPrefix(prefix string, keys []string, waitIndex uint64, stopChan chan bool) (uint64, error) {
+	if c.current.errTimes >= 3 {
+		c.selectConnection()
+	}
+
+	conn := c.current
+
+	// return something > 0 to trigger a key retrieval from the store
+	if waitIndex == 0 {
+		conn.waitIndex = 1
+		return conn.waitIndex, nil
+	}
+	// when switch to anther server, so set waitIndex 0, and let server response current version.
+	if conn.waitIndex == 0 {
+		waitIndex = 0
+	}
+
+	done := make(chan struct{})
+	defer close(done)
+	ctx, cancel := context.WithCancel(context.Background())
+	req, err := http.NewRequest("GET", fmt.Sprintf("%s%s?wait=true&prev_version=%d", conn.url, prefix, waitIndex), nil)
+	if err != nil {
+		return conn.waitIndex, err
+	}
+
+	req.Header.Set("Accept", "application/json")
+	req = req.WithContext(ctx)
+	go func() {
+		select {
+		case <-stopChan:
+			cancel()
+		case <-done:
+			return
+		}
+	}()
+
+	// just ignore resp, notify confd to reload metadata from metad
+	resp, err := conn.httpClient.Do(req)
+	if resp != nil {
+		defer resp.Body.Close()
+	}
+	if err != nil {
+		logger.Error("failed to connect to metad when watch prefix")
+		atomic.AddUint32(&conn.errTimes, 1)
+		return conn.waitIndex, err
+	}
+	if resp.StatusCode != 200 {
+		return conn.waitIndex, errors.New(fmt.Sprintf("metad response status [%v], requestID: [%s]", resp.StatusCode, resp.Header.Get("X-Metad-RequestID")))
+	}
+	versionStr := resp.Header.Get("X-Metad-OpVersion")
+	if versionStr != "" {
+		v, err := strconv.ParseUint(versionStr, 10, 64)
+		if err != nil {
+			logger.Error("Parse X-Metad-OpVersion %s error:%s", versionStr, err.Error())
+		}
+		conn.waitIndex = v
+	} else {
+		logger.Warning("Metad response miss X-Metad-OpVersion header.")
+		conn.waitIndex = conn.waitIndex + 1
+	}
+	return conn.waitIndex, nil
+}
+
+func (c *MetadConnection) makeMetaDataRequest(path string) ([]byte, error) {
+	req, err := http.NewRequest("GET", strings.Join([]string{c.url, path}, ""), nil)
+	if err != nil {
+		return nil, err
+	}
+	req.Header.Set("Accept", "application/json")
+
+	resp, err := c.httpClient.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	return ioutil.ReadAll(resp.Body)
+}

metad/dummy.txt

@@ -0,0 +1,7 @@
+// Copyright 2018 The OpenPitrix Authors. All rights reserved.
+// Use of this source code is governed by a Apache license
+// that can be found in the LICENSE file.
+
+module openpitrix.io/libconfd/metad
+
+require openpitrix.io/libconfd v0.1.0

metad/go.mod

@@ -0,0 +1,6 @@
+github.com/BurntSushi/toml v0.3.1 h1:WXkYYl6Yr3qBf1K79EBnL4mak0OimBfB0XUf9Vl28OQ=
+github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
+golang.org/x/crypto v0.0.0-20181015023909-0c41d7ab0a0e h1:IzypfodbhbnViNUO/MEh0FzCUooG97cIGfdggUrUSyU=
+golang.org/x/crypto v0.0.0-20181015023909-0c41d7ab0a0e/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
+openpitrix.io/libconfd v0.1.0 h1:r3jK8TNCQ9jmp796xzBzQ8vQH9astI4gCPPXKNR9Z40=
+openpitrix.io/libconfd v0.1.0/go.mod h1:LfoWsMYAvA/Xw4XOldge+cwt5Zux1+y4ch8T80OkONQ=