Merge pull request #119 from aeroyu/master

etcd basic auth support

Author: zhangxu19830126

cmd/api/api_server.go

@@ -24,6 +24,8 @@ var (
 	addr           = flag.String("addr", "127.0.0.1:9092", "Addr: client grpc entrypoint")
 	addrHTTP       = flag.String("addr-http", "127.0.0.1:9093", "Addr: client http restful entrypoint")
 	addrStore      = flag.String("addr-store", "etcd://127.0.0.1:2379", "Addr: store address")
+	addrStoreUser  = flag.String("addr-store-user", "", "addr Store UserName")
+	addrStorePwd   = flag.String("addr-store-pwd", "", "addr Store Password")
 	namespace      = flag.String("namespace", "dev", "The namespace to isolation the environment.")
 	discovery      = flag.Bool("discovery", false, "Publish apiserver service via discovery.")
 	servicePrefix  = flag.String("service-prefix", "/services", "The prefix for service name.")
@@ -46,13 +48,15 @@ func main() {
 
 	log.Infof("addr: %s", *addr)
 	log.Infof("addr-store: %s", *addrStore)
+	log.Infof("addr-store-user: %s", *addrStoreUser)
+	log.Infof("addr-store-pwd: %s", *addrStorePwd)
 	log.Infof("namespace: %s", *namespace)
 	log.Infof("discovery: %v", *discovery)
 	log.Infof("service-prefix: %s", *servicePrefix)
 	log.Infof("publish-lease: %d", *publishLease)
 	log.Infof("publish-timeout: %d", *publishTimeout)
 
-	db, err := store.GetStoreFrom(*addrStore, fmt.Sprintf("/%s", *namespace))
+	db, err := store.GetStoreFrom(*addrStore, fmt.Sprintf("/%s", *namespace), *addrStoreUser, *addrStorePwd)
 	if err != nil {
 		log.Fatalf("init store failed for %s, errors:\n%+v",
 			*addrStore,

cmd/proxy/proxy.go

@@ -34,6 +34,8 @@ var (
 	addr                          = flag.String("addr", "127.0.0.1:80", "Addr: http request entrypoint")
 	addrRPC                       = flag.String("addr-rpc", "127.0.0.1:9091", "Addr: manager request entrypoint")
 	addrStore                     = flag.String("addr-store", "etcd://127.0.0.1:2379", "Addr: store of meta data, support etcd")
+	addrStoreUser                 = flag.String("addr-store-user", "", "addr Store UserName")
+	addrStorePwd                  = flag.String("addr-store-pwd", "", "addr Store Password")
 	addrPPROF                     = flag.String("addr-pprof", "", "Addr: pprof addr")
 	namespace                     = flag.String("namespace", "dev", "The namespace to isolation the environment.")
 	limitCpus                     = flag.Int("limit-cpus", 0, "Limit: schedule threads count")
@@ -139,6 +141,8 @@ func getCfg() *proxy.Cfg {
 	cfg.AddrRPC = *addrRPC
 	cfg.AddrPPROF = *addrPPROF
 	cfg.AddrStore = *addrStore
+	cfg.AddrStoreUserName = *addrStoreUser
+	cfg.AddrStorePwd = *addrStorePwd
 	cfg.TTLProxy = *ttlProxy
 	cfg.Namespace = fmt.Sprintf("/%s", *namespace)
 	cfg.Option.LimitBytesBody = *limitBytesBodyMB * 1024 * 1024

pkg/proxy/cfg.go

@@ -31,13 +31,15 @@ type Option struct {
 
 // Cfg proxy config
 type Cfg struct {
-	Addr      string
-	AddrRPC   string
-	AddrStore string
-	AddrPPROF string
-	Namespace string
-	TTLProxy  int64
-	Filers    []*FilterSpec
+	Addr              string
+	AddrRPC           string
+	AddrStore         string
+	AddrStoreUserName string
+	AddrStorePwd      string
+	AddrPPROF         string
+	Namespace         string
+	TTLProxy          int64
+	Filers            []*FilterSpec
 
 	Option *Option
 	Metric *util.MetricCfg

pkg/proxy/proxy.go

@@ -221,7 +221,7 @@ func (p *Proxy) init() {
 }
 
 func (p *Proxy) initDispatcher() error {
-	s, err := store.GetStoreFrom(p.cfg.AddrStore, p.cfg.Namespace)
+	s, err := store.GetStoreFrom(p.cfg.AddrStore, p.cfg.Namespace, p.cfg.AddrStoreUserName, p.cfg.AddrStorePwd)
 
 	if err != nil {
 		return err

pkg/store/store.go

@@ -19,7 +19,7 @@ var (
 )
 
 var (
-	supportSchema = make(map[string]func(string, string) (Store, error))
+	supportSchema = make(map[string]func(string, string, BasicAuth) (Store, error))
 )
 
 // EvtType event type
@@ -28,6 +28,11 @@ type EvtType int
 // EvtSrc event src
 type EvtSrc int
 
+type BasicAuth struct {
+	userName string
+	password string
+}
+
 const (
 	// EventTypeNew event type new
 	EventTypeNew = EvtType(0)
@@ -65,7 +70,7 @@ func init() {
 }
 
 // GetStoreFrom returns a store implemention, if not support returns error
-func GetStoreFrom(registryAddr, prefix string) (Store, error) {
+func GetStoreFrom(registryAddr, prefix string, userName string, password string) (Store, error) {
 	u, err := url.Parse(registryAddr)
 	if err != nil {
 		panic(fmt.Sprintf("parse registry addr failed, errors:%+v", err))
@@ -74,21 +79,21 @@ func GetStoreFrom(registryAddr, prefix string) (Store, error) {
 	schema := strings.ToLower(u.Scheme)
 	fn, ok := supportSchema[schema]
 	if ok {
-		return fn(u.Host, prefix)
+		return fn(u.Host, prefix, BasicAuth{userName: userName, password: password})
 	}
 
 	return nil, fmt.Errorf("not support: %s", registryAddr)
 }
 
-func getEtcdStoreFrom(addr, prefix string) (Store, error) {
+func getEtcdStoreFrom(addr, prefix string, basicAuth BasicAuth) (Store, error) {
 	var addrs []string
 	values := strings.Split(addr, ",")
 
 	for _, value := range values {
 		addrs = append(addrs, fmt.Sprintf("http://%s", value))
 	}
 
-	return NewEtcdStore(addrs, prefix)
+	return NewEtcdStore(addrs, prefix, basicAuth)
 }
 
 // Store store interface

pkg/store/store_etcd.go

@@ -61,7 +61,7 @@ type EtcdStore struct {
 }
 
 // NewEtcdStore create a etcd store
-func NewEtcdStore(etcdAddrs []string, prefix string) (Store, error) {
+func NewEtcdStore(etcdAddrs []string, prefix string, basicAuth BasicAuth) (Store, error) {
 	store := &EtcdStore{
 		prefix:             prefix,
 		clustersDir:        fmt.Sprintf("%s/clusters", prefix),
@@ -76,10 +76,18 @@ func NewEtcdStore(etcdAddrs []string, prefix string) (Store, error) {
 		end:                100,
 	}
 
-	cli, err := clientv3.New(clientv3.Config{
+	config := &clientv3.Config{
 		Endpoints:   etcdAddrs,
 		DialTimeout: DefaultTimeout,
-	})
+	}
+	if basicAuth.userName != "" {
+		config.Username = basicAuth.userName
+	}
+	if basicAuth.password != "" {
+		config.Password = basicAuth.password
+	}
+
+	cli, err := clientv3.New(*config)
 
 	if err != nil {
 		return nil, err