implement generic service informer

cmd/climc/main.go

@@ -1,3 +1,17 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package main
 
 import (

cmd/climc/promputils/complete_zsh.go

@@ -1,3 +1,17 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package promputils
 
 import (

cmd/climc/shell/k8s/event.go

@@ -1,3 +1,17 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package k8s
 
 import (

cmd/climc/shell/watch.go

@@ -0,0 +1,94 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package shell
+
+import (
+	"context"
+
+	"yunion.io/x/jsonutils"
+	"yunion.io/x/log"
+	"yunion.io/x/pkg/errors"
+	"yunion.io/x/pkg/util/sets"
+
+	"yunion.io/x/onecloud/pkg/mcclient"
+	"yunion.io/x/onecloud/pkg/mcclient/informer"
+	"yunion.io/x/onecloud/pkg/mcclient/modulebase"
+)
+
+type eventHandler struct {
+	man informer.IResourceManager
+}
+
+func (e eventHandler) keyword() string {
+	return e.man.GetKeyword()
+}
+
+func (e eventHandler) OnAdd(obj *jsonutils.JSONDict) {
+	log.Infof("%s [CREATED]: \n%s", e.keyword(), obj.String())
+}
+
+func (e eventHandler) OnUpdate(oldObj, newObj *jsonutils.JSONDict) {
+	log.Infof("%s [UPDATED]: \n[NEW]: %s\n[OLD]: %s", e.keyword(), newObj.String(), oldObj.String())
+}
+
+func (e eventHandler) OnDelete(obj *jsonutils.JSONDict) {
+	log.Infof("%s [DELETED]: \n%s", e.keyword(), obj.String())
+}
+
+func init() {
+	type WatchOptions struct {
+		Resource []string `help:"Resource manager plural keyword, e.g.'servers, disks, guestdisks'" short-token:"s"`
+		All      bool     `help:"Watch all resources"`
+	}
+
+	R(&WatchOptions{}, "watch", "Watch resources", func(s *mcclient.ClientSession, opts *WatchOptions) error {
+		watchMan, err := informer.NewWatchManagerBySession(s, nil)
+		if err != nil {
+			return err
+		}
+		resources := opts.Resource
+		if opts.All {
+			resSets := sets.NewString()
+			mods, _ := modulebase.GetRegisterdModules()
+			for _, ress := range mods {
+				resSets.Insert(ress...)
+			}
+			resources = resSets.List()
+		}
+		if len(resources) == 0 {
+			return errors.Errorf("no watch resources specified")
+		}
+		for _, res := range resources {
+			var resMan informer.IResourceManager
+			if modMan, _ := modulebase.GetModule(s, res); modMan != nil {
+				resMan = modMan
+			}
+			if resMan == nil {
+				if jointModMan, _ := modulebase.GetJointModule(s, res); jointModMan != nil {
+					resMan = jointModMan
+				}
+			}
+			if resMan == nil {
+				//return errors.Errorf("Not found %q resource manager", res)
+				log.Warningf("Not found %q resource manager", res)
+				continue
+			}
+			if err := watchMan.For(resMan).AddEventHandler(context.Background(), eventHandler{resMan}); err != nil {
+				return errors.Wrapf(err, "watch resource %s", res)
+			}
+		}
+		select {}
+	})
+}

pkg/apis/compute/vpcs_ovn.go

@@ -1,3 +1,17 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package compute
 
 import (

pkg/apis/identity/endpoint.go

@@ -38,5 +38,5 @@ type CertificateDetails struct {
 }
 
 const (
-	ENDPOINT_ETCD_INTERNAL = "etcd-internal"
+	SERVICE_TYPE_ETCD = "etcd"
 )

pkg/cloudcommon/app/auth.go

@@ -19,7 +19,6 @@ import (
 	"time"
 
 	"yunion.io/x/log"
-	"yunion.io/x/pkg/errors"
 	"yunion.io/x/pkg/utils"
 
 	"yunion.io/x/onecloud/pkg/apis/identity"
@@ -27,9 +26,7 @@ import (
 	"yunion.io/x/onecloud/pkg/cloudcommon/notifyclient"
 	common_options "yunion.io/x/onecloud/pkg/cloudcommon/options"
 	"yunion.io/x/onecloud/pkg/cloudcommon/policy"
-	"yunion.io/x/onecloud/pkg/httperrors"
 	"yunion.io/x/onecloud/pkg/mcclient/auth"
-	"yunion.io/x/onecloud/pkg/mcclient/modules"
 )
 
 func InitAuth(options *common_options.CommonOptions, authComplete auth.AuthCompletedCallback) {
@@ -63,7 +60,7 @@ func InitAuth(options *common_options.CommonOptions, authComplete auth.AuthCompl
 
 	if options.SessionEndpointType != "" {
 		if !utils.IsInStringArray(options.SessionEndpointType,
-			[]string{auth.PublicEndpointType, auth.InternalEndpointType}) {
+			[]string{identity.EndpointInterfacePublic, identity.EndpointInterfaceInternal}) {
 			log.Fatalf("Invalid session endpoint type %s", options.SessionEndpointType)
 		}
 		auth.SetEndpointType(options.SessionEndpointType)
@@ -97,16 +94,5 @@ func InitBaseAuth(options *common_options.BaseOptions) {
 
 func FetchEtcdServiceInfo() (*identity.EndpointDetails, error) {
 	s := auth.GetAdminSession(context.Background(), "", "")
-	ret, err := modules.EndpointsV3.GetByName(s, identity.ENDPOINT_ETCD_INTERNAL, nil)
-	if err != nil && errors.Cause(err) == httperrors.ErrNotFound {
-		return nil, nil
-	} else if err != nil {
-		return nil, err
-	}
-	endpoint := new(identity.EndpointDetails)
-	err = ret.Unmarshal(endpoint)
-	if err != nil {
-		return nil, errors.Wrap(err, "unmarshal endpoint")
-	}
-	return endpoint, nil
+	return s.GetCommonEtcdEndpoint()
 }

pkg/cloudcommon/database.go

@@ -27,6 +27,8 @@ import (
 	"yunion.io/x/onecloud/pkg/appsrv"
 	"yunion.io/x/onecloud/pkg/cloudcommon/consts"
 	"yunion.io/x/onecloud/pkg/cloudcommon/db/lockman"
+	"yunion.io/x/onecloud/pkg/cloudcommon/etcd"
+	"yunion.io/x/onecloud/pkg/cloudcommon/informer"
 	common_options "yunion.io/x/onecloud/pkg/cloudcommon/options"
 )
 
@@ -77,6 +79,26 @@ func InitDB(options *common_options.DBOptions) {
 		lockman.Init(lm)
 	}
 	// lm := lockman.NewNoopLockManager()
+
+	if len(options.EtcdEndpoints) != 0 {
+		log.Infof("using etcd as resource informer backend")
+		tlsCfg, err := options.GetEtcdTLSConfig()
+		if err != nil {
+			log.Fatalf("get etcd informer backend tls config err: %v", err)
+		}
+		informerBackend, err := informer.NewEtcdBackend(&etcd.SEtcdOptions{
+			EtcdEndpoint:              options.EtcdEndpoints,
+			EtcdTimeoutSeconds:        5,
+			EtcdRequestTimeoutSeconds: 2,
+			EtcdLeaseExpireSeconds:    5,
+			EtcdEnabldSsl:             options.EtcdUseTLS,
+			TLSConfig:                 tlsCfg,
+		}, nil)
+		if err != nil {
+			log.Fatalf("new etcd informer backend error: %v", err)
+		}
+		informer.Init(informerBackend)
+	}
 }
 
 func CloseDB() {

pkg/cloudcommon/db/certificateresource.go

@@ -1,3 +1,17 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package db
 
 import (

pkg/cloudcommon/db/interface.go

@@ -40,7 +40,7 @@ type IModelManager interface {
 	GetIModelManager() IModelManager
 
 	// Table() *sqlchemy.STable
-	TableSpec() *sqlchemy.STableSpec
+	TableSpec() ITableSpec
 
 	// Keyword() string
 	KeywordPlural() string

pkg/cloudcommon/db/modelbase.go

@@ -44,7 +44,7 @@ type SModelBase struct {
 type SModelBaseManager struct {
 	object.SObject
 
-	tableSpec     *sqlchemy.STableSpec
+	tableSpec     ITableSpec
 	keyword       string
 	keywordPlural string
 	alias         string
@@ -52,7 +52,7 @@ type SModelBaseManager struct {
 }
 
 func NewModelBaseManager(model interface{}, tableName string, keyword string, keywordPlural string) SModelBaseManager {
-	ts := sqlchemy.NewTableSpecFromStruct(model, tableName)
+	ts := newTableSpec(model, tableName)
 	modelMan := SModelBaseManager{tableSpec: ts, keyword: keyword, keywordPlural: keywordPlural}
 	return modelMan
 }
@@ -78,7 +78,7 @@ func (manager *SModelBaseManager) SetAlias(alias string, aliasPlural string) {
 	manager.aliasPlural = aliasPlural
 }
 
-func (manager *SModelBaseManager) TableSpec() *sqlchemy.STableSpec {
+func (manager *SModelBaseManager) TableSpec() ITableSpec {
 	return manager.tableSpec
 }
 

pkg/cloudcommon/db/sharablebase_test.go

@@ -1,3 +1,17 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package db
 
 import (

pkg/cloudcommon/db/tablespec.go

@@ -0,0 +1,154 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package db
+
+import (
+	"context"
+	"reflect"
+
+	"yunion.io/x/jsonutils"
+	"yunion.io/x/log"
+	"yunion.io/x/pkg/errors"
+	"yunion.io/x/sqlchemy"
+
+	"yunion.io/x/onecloud/pkg/cloudcommon/informer"
+	"yunion.io/x/onecloud/pkg/util/nopanic"
+)
+
+type ITableSpec interface {
+	Name() string
+	DataType() reflect.Type
+	Insert(dt interface{}) error
+	InsertOrUpdate(dt interface{}) error
+	Update(dt interface{}, doUpdate func() error) (sqlchemy.UpdateDiffs, error)
+	Instance() *sqlchemy.STable
+	ColumnSpec(name string) sqlchemy.IColumnSpec
+	PrimaryColumns() []sqlchemy.IColumnSpec
+	Columns() []sqlchemy.IColumnSpec
+	Fetch(dt interface{}) error
+	FetchAll(dest interface{}) error
+	SyncSQL() []string
+	DropForeignKeySQL() []string
+	AddIndex(unique bool, cols ...string) bool
+}
+
+type sTableSpec struct {
+	*sqlchemy.STableSpec
+}
+
+func newTableSpec(model interface{}, tableName string) ITableSpec {
+	return &sTableSpec{
+		STableSpec: sqlchemy.NewTableSpecFromStruct(model, tableName),
+	}
+}
+
+func (ts *sTableSpec) newInformerModel(dt interface{}) (*informer.ModelObject, error) {
+	obj, ok := dt.(IModel)
+	if !ok {
+		return nil, errors.Errorf("informer model is not IModel")
+	}
+	if obj.GetVirtualObject() == nil {
+		return nil, errors.Errorf("object %v virtual object is nil", obj)
+	}
+	if obj.GetModelManager() == nil {
+		return nil, errors.Errorf("object %v model manager is nil", obj)
+	}
+	jointObj, isJoint := obj.(IJointModel)
+	if isJoint {
+		mObj := jointObj.Master()
+		sObj := jointObj.Slave()
+		return informer.NewJointModel(jointObj, jointObj.KeywordPlural(), mObj.GetId(), sObj.GetId()), nil
+	}
+	return informer.NewModel(obj, obj.KeywordPlural(), obj.GetId()), nil
+}
+
+func (ts *sTableSpec) isMarkDeleted(dt interface{}) (bool, error) {
+	if vObj, ok := dt.(IVirtualModel); ok {
+		if vObj.GetPendingDeleted() {
+			return true, nil
+		}
+	}
+	obj, ok := dt.(IModel)
+	if !ok {
+		return false, errors.Errorf("informer model is not IModel")
+	}
+	return obj.GetDeleted(), nil
+}
+
+func (ts *sTableSpec) Insert(dt interface{}) error {
+	if err := ts.STableSpec.Insert(dt); err != nil {
+		return err
+	}
+	ts.inform(dt, informer.Create)
+	return nil
+}
+
+func (ts *sTableSpec) InsertOrUpdate(dt interface{}) error {
+	if err := ts.STableSpec.InsertOrUpdate(dt); err != nil {
+		return err
+	}
+	ts.inform(dt, informer.Create)
+	return nil
+}
+
+func (ts *sTableSpec) Update(dt interface{}, doUpdate func() error) (sqlchemy.UpdateDiffs, error) {
+	oldObj := jsonutils.Marshal(dt)
+	diffs, err := ts.STableSpec.Update(dt, doUpdate)
+	if err != nil {
+		return nil, err
+	}
+	if diffs == nil {
+		// no data to update
+		return nil, nil
+	}
+	isDeleted, err := ts.isMarkDeleted(dt)
+	if err != nil {
+		return nil, errors.Wrap(err, "check is mark deleted")
+	}
+	if isDeleted {
+		ts.inform(dt, informer.Delete)
+	} else {
+		ts.informUpdate(dt, oldObj.(*jsonutils.JSONDict))
+	}
+	return diffs, nil
+}
+
+func (ts *sTableSpec) inform(dt interface{}, f func(ctx context.Context, obj *informer.ModelObject) error) {
+	nf := func() {
+		obj, err := ts.newInformerModel(dt)
+		if err != nil {
+			log.Warningf("newInformerModel error: %v", err)
+			return
+		}
+		if err := f(context.Background(), obj); err != nil {
+			log.Errorf("call informer func error: %v", err)
+		}
+	}
+	nopanic.Run(nf)
+}
+
+func (ts *sTableSpec) informUpdate(dt interface{}, oldObj *jsonutils.JSONDict) {
+	nf := func() {
+		obj, err := ts.newInformerModel(dt)
+		if err != nil {
+			log.Warningf("newInformerModel error: %v", err)
+			return
+		}
+		if err := informer.Update(context.Background(), obj, oldObj); err != nil {
+			log.Errorf("call informer update func error: %v", err)
+		}
+	}
+	nopanic.Run(nf)
+}

pkg/cloudcommon/etcd/etcd.go

@@ -17,7 +17,6 @@ package etcd
 import (
 	"context"
 	"crypto/tls"
-	"errors"
 	"fmt"
 	"time"
 
@@ -25,12 +24,13 @@ import (
 	"google.golang.org/grpc"
 
 	"yunion.io/x/log"
+	"yunion.io/x/pkg/errors"
 
 	"yunion.io/x/onecloud/pkg/util/seclib2"
 )
 
 var (
-	ErrNoSuchKey = errors.New("No such key")
+	ErrNoSuchKey = errors.Error("No such key")
 )
 
 type SEtcdClient struct {
@@ -178,7 +178,7 @@ func (cli *SEtcdClient) startSession() error {
 
 func (cli *SEtcdClient) RestartSession() error {
 	if cli.leaseLiving {
-		return errors.New("session is living, can't restart")
+		return errors.Error("session is living, can't restart")
 	}
 	return cli.startSession()
 }
@@ -199,6 +199,34 @@ func (cli *SEtcdClient) PutSession(ctx context.Context, key string, val string)
 	return cli.put(ctx, key, val, true)
 }
 
+func (cli *SEtcdClient) grantLease(ctx context.Context, ttlSeconds int64) (*clientv3.LeaseGrantResponse, error) {
+	nctx, cancel := context.WithTimeout(ctx, cli.requestTimeout)
+	defer cancel()
+	resp, err := cli.client.Grant(nctx, ttlSeconds)
+	if err != nil {
+		return nil, errors.Wrap(err, "grant lease")
+	}
+	return resp, err
+}
+
+func (cli *SEtcdClient) PutWithLease(ctx context.Context, key string, val string, ttlSeconds int64) error {
+	resp, err := cli.grantLease(ctx, ttlSeconds)
+	if err != nil {
+		return errors.Wrap(err, "put with grant lease")
+	}
+
+	nctx, cancel := context.WithTimeout(ctx, cli.requestTimeout)
+	defer cancel()
+
+	key = cli.getKey(key)
+	leaseId := resp.ID
+	opts := []clientv3.OpOption{
+		clientv3.WithLease(leaseId),
+	}
+	_, err = cli.client.Put(nctx, key, val, opts...)
+	return err
+}
+
 func (cli *SEtcdClient) put(ctx context.Context, key string, val string, session bool) error {
 	nctx, cancel := context.WithTimeout(ctx, cli.requestTimeout)
 	defer cancel()
@@ -269,10 +297,10 @@ func (w *SEtcdWatcher) Cancel() {
 	w.cancel()
 }
 
-func (cli *SEtcdClient) Watch(ctx context.Context, prefix string, onCreate TEtcdCreateEventFunc, onModify TEtcdModifyEventFunc) {
+func (cli *SEtcdClient) Watch(ctx context.Context, prefix string, onCreate TEtcdCreateEventFunc, onModify TEtcdModifyEventFunc) error {
 	_, ok := cli.watchers[prefix]
 	if ok {
-		return
+		return errors.Errorf("watch prefix %s already registered", prefix)
 	}
 
 	watcher := clientv3.NewWatcher(cli.client)
@@ -298,6 +326,7 @@ func (cli *SEtcdClient) Watch(ctx context.Context, prefix string, onCreate TEtcd
 		}
 		log.Infof("stop watching %s", prefix)
 	}()
+	return nil
 }
 
 func (cli *SEtcdClient) Unwatch(prefix string) {

pkg/cloudcommon/informer/doc.go

@@ -0,0 +1,15 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package informer // import "yunion.io/x/onecloud/pkg/cloudcommon/informer"

pkg/cloudcommon/informer/etcd.go

@@ -0,0 +1,184 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package informer
+
+import (
+	"context"
+	"fmt"
+	"path/filepath"
+
+	"yunion.io/x/jsonutils"
+	"yunion.io/x/log"
+	"yunion.io/x/pkg/errors"
+
+	"yunion.io/x/onecloud/pkg/cloudcommon/etcd"
+)
+
+const (
+	EtcdInformerPrefix = "/onecloud/informer"
+
+	EventTypeCreate = "CREATE"
+	EventTypeUpdate = "UPDATE"
+	EventTypeDelete = "DELETE"
+)
+
+type TEventType string
+
+type modelObject struct {
+	EventType TEventType          `json:"event_type"`
+	Object    *jsonutils.JSONDict `json:"object"`
+	OldObject *jsonutils.JSONDict `json:"old_object"`
+}
+
+func (obj modelObject) ToKey() string {
+	return jsonutils.Marshal(obj).String()
+}
+
+func newModelObjectFromValue(val []byte) (*modelObject, error) {
+	jObj, err := jsonutils.Parse(val)
+	if err != nil {
+		return nil, errors.Wrapf(err, "parse key %s", val)
+	}
+	ret := new(modelObject)
+	if err := jObj.Unmarshal(ret); err != nil {
+		return nil, errors.Wrap(err, "unmarshal to model object")
+	}
+	return ret, nil
+}
+
+type EtcdBackend struct {
+	client   *etcd.SEtcdClient
+	leaseTTL int64
+}
+
+func NewEtcdBackend(opt *etcd.SEtcdOptions, onKeepaliveFailure func()) (*EtcdBackend, error) {
+	opt.EtcdNamspace = EtcdInformerPrefix
+	be := new(EtcdBackend)
+	be.leaseTTL = int64(opt.EtcdLeaseExpireSeconds)
+	if onKeepaliveFailure == nil {
+		onKeepaliveFailure = be.onKeepaliveFailure
+	}
+	cli, err := etcd.NewEtcdClient(opt, onKeepaliveFailure)
+	if err != nil {
+		return nil, errors.Wrap(err, "new etcd client")
+	}
+	be.client = cli
+	return be, nil
+}
+
+func (b *EtcdBackend) getObjectKey(obj *ModelObject) string {
+	if obj.IsJoint {
+		return fmt.Sprintf("/%s/%s/%s", obj.KeywordPlural, obj.MasterId, obj.SlaveId)
+	}
+	return fmt.Sprintf("/%s/%s", obj.KeywordPlural, obj.Id)
+}
+
+func (b *EtcdBackend) getValue(eventType TEventType, obj *ModelObject) string {
+	modelObject := modelObject{
+		EventType: eventType,
+		Object:    obj.Object,
+	}
+	return modelObject.ToKey()
+}
+
+func (b *EtcdBackend) getUpdateValue(obj *ModelObject, oldObj *jsonutils.JSONDict) string {
+	modelObject := modelObject{
+		EventType: EventTypeUpdate,
+		Object:    obj.Object,
+		OldObject: oldObj,
+	}
+	return modelObject.ToKey()
+}
+
+func (b *EtcdBackend) GetType() string {
+	return "etcd"
+}
+
+func (b *EtcdBackend) Create(ctx context.Context, obj *ModelObject) error {
+	key := b.getObjectKey(obj)
+	val := b.getValue(EventTypeCreate, obj)
+	return b.put(ctx, key, val)
+}
+
+func (b *EtcdBackend) Update(ctx context.Context, obj *ModelObject, oldObj *jsonutils.JSONDict) error {
+	key := b.getObjectKey(obj)
+	val := b.getUpdateValue(obj, oldObj)
+	return b.put(ctx, key, val)
+}
+
+func (b *EtcdBackend) Delete(ctx context.Context, obj *ModelObject) error {
+	key := b.getObjectKey(obj)
+	val := b.getValue(EventTypeDelete, obj)
+	err := b.put(ctx, key, val)
+	return err
+}
+
+func (b *EtcdBackend) put(ctx context.Context, key, val string) error {
+	return b.client.PutWithLease(ctx, key, val, b.leaseTTL)
+}
+
+func (b *EtcdBackend) onKeepaliveFailure() {
+	if err := b.client.RestartSession(); err != nil {
+		log.Errorf("restart etcd session error: %v", err)
+	}
+}
+
+func (b *EtcdBackend) getWatchKey(key string) string {
+	return filepath.Join("/", key)
+}
+
+func (b *EtcdBackend) Watch(ctx context.Context, key string, handler ResourceEventHandler) error {
+	return b.client.Watch(ctx, b.getWatchKey(key), b.onCreate(handler), b.onModify(handler))
+}
+
+func (b *EtcdBackend) Unwatch(key string) {
+	b.client.Unwatch(b.getWatchKey(key))
+}
+
+func (b *EtcdBackend) onCreate(handler ResourceEventHandler) etcd.TEtcdCreateEventFunc {
+	return func(key, value []byte) {
+		b.processEvent(handler, string(key), value)
+	}
+}
+
+func (b *EtcdBackend) onModify(handler ResourceEventHandler) etcd.TEtcdModifyEventFunc {
+	return func(key, _, value []byte) {
+		// not care about oldvalue, so ignore it
+		b.processEvent(handler, string(key), value)
+	}
+}
+
+func (b *EtcdBackend) processEvent(handler ResourceEventHandler, key string, value []byte) {
+	if len(value) == 0 {
+		// object already deleted by lease out of ttl
+		return
+	}
+	mObj, err := newModelObjectFromValue(value)
+	if err != nil {
+		log.Errorf("new %s model objecd from value error: %v", key, err)
+		return
+	}
+	eType := mObj.EventType
+	switch eType {
+	case EventTypeCreate:
+		handler.OnAdd(mObj.Object)
+	case EventTypeUpdate:
+		handler.OnUpdate(mObj.OldObject, mObj.Object)
+	case EventTypeDelete:
+		handler.OnDelete(mObj.Object)
+	default:
+		log.Errorf("Invalid event type: %s, mObj: %#v", eType, mObj)
+	}
+}

pkg/cloudcommon/informer/informer.go

@@ -0,0 +1,114 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package informer
+
+import (
+	"context"
+
+	"yunion.io/x/jsonutils"
+	"yunion.io/x/log"
+	"yunion.io/x/pkg/errors"
+)
+
+const (
+	ErrBackendNotInit = errors.Error("InformerBackend not init")
+)
+
+var (
+	defaultBackend IInformerBackend
+)
+
+type IInformerBackend interface {
+	GetType() string
+	Create(ctx context.Context, obj *ModelObject) error
+	Update(ctx context.Context, obj *ModelObject, oldObj *jsonutils.JSONDict) error
+	Delete(ctx context.Context, obj *ModelObject) error
+}
+
+type IWatcher interface {
+	Watch(ctx context.Context, key string, handler ResourceEventHandler) error
+	Unwatch(key string)
+}
+
+func Init(be IInformerBackend) {
+	if defaultBackend != nil {
+		log.Fatalf("informer backend %q already init", be.GetType())
+	}
+	defaultBackend = be
+}
+
+func Set(be IInformerBackend) {
+	defaultBackend = be
+}
+
+func GetDefaultBackend() IInformerBackend {
+	if defaultBackend == nil {
+		log.Warningf("default informer backend is not init")
+	}
+	return defaultBackend
+}
+
+func IsInit() bool {
+	return defaultBackend != nil
+}
+
+type ModelObject struct {
+	Object        *jsonutils.JSONDict
+	KeywordPlural string
+	Id            string
+	IsJoint       bool
+	MasterId      string
+	SlaveId       string
+}
+
+func NewModel(obj interface{}, keywordPlural, id string) *ModelObject {
+	return &ModelObject{
+		Object:        jsonutils.Marshal(obj).(*jsonutils.JSONDict),
+		KeywordPlural: keywordPlural,
+		Id:            id,
+	}
+}
+
+func NewJointModel(obj interface{}, keywordPlural, masterId, slaveId string) *ModelObject {
+	model := NewModel(obj, keywordPlural, "")
+	model.IsJoint = true
+	model.MasterId = masterId
+	model.SlaveId = slaveId
+	return model
+}
+
+func Create(ctx context.Context, obj *ModelObject) error {
+	return run(func(be IInformerBackend) error {
+		return be.Create(ctx, obj)
+	})
+}
+
+func Update(ctx context.Context, obj *ModelObject, oldObj *jsonutils.JSONDict) error {
+	return run(func(be IInformerBackend) error {
+		return be.Update(ctx, obj, oldObj)
+	})
+}
+
+func Delete(ctx context.Context, obj *ModelObject) error {
+	return run(func(be IInformerBackend) error {
+		return be.Delete(ctx, obj)
+	})
+}
+
+type ResourceEventHandler interface {
+	OnAdd(obj *jsonutils.JSONDict)
+	OnUpdate(oldObj, newObj *jsonutils.JSONDict)
+	OnDelete(obj *jsonutils.JSONDict)
+}

pkg/cloudcommon/informer/worker.go

@@ -0,0 +1,46 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package informer
+
+import (
+	"yunion.io/x/log"
+
+	"yunion.io/x/onecloud/pkg/appsrv"
+	"yunion.io/x/onecloud/pkg/util/nopanic"
+)
+
+var (
+	informerWorkerMan *appsrv.SWorkerManager
+)
+
+func init() {
+	informerWorkerMan = appsrv.NewWorkerManager("InformerWorkerManager", 1024, 10240, false)
+}
+
+func run(f func(be IInformerBackend) error) error {
+	be := GetDefaultBackend()
+	if be == nil {
+		return ErrBackendNotInit
+	}
+	wf := func() {
+		nopanic.Run(func() {
+			if err := f(be); err != nil {
+				log.Errorf("run informer error: %v", err)
+			}
+		})
+	}
+	informerWorkerMan.Run(wf, nil, nil)
+	return nil
+}

pkg/compute/models/host_health.go

@@ -1,3 +1,17 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package models
 
 import (

pkg/compute/models/vpcs_mapped_address.go

@@ -1,3 +1,17 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package models
 
 import (

pkg/devtool/models/actions.go

@@ -22,6 +22,7 @@ import (
 	"yunion.io/x/log"
 
 	apis "yunion.io/x/onecloud/pkg/apis/ansible"
+	apiidentity "yunion.io/x/onecloud/pkg/apis/identity"
 	"yunion.io/x/onecloud/pkg/mcclient"
 	"yunion.io/x/onecloud/pkg/mcclient/auth"
 	"yunion.io/x/onecloud/pkg/mcclient/modules"
@@ -59,7 +60,7 @@ func getServerAttrs(ID string, s *mcclient.ClientSession) (map[string]string, er
 func getInfluxdbURL() (string, error) {
 
 	s := auth.GetAdminSessionWithPublic(nil, "", "")
-	url, err := s.GetServiceURL("influxdb", auth.PublicEndpointType)
+	url, err := s.GetServiceURL("influxdb", apiidentity.EndpointInterfacePublic)
 
 	if err != nil {
 		log.Errorf("get influxdb Endpoint error %s", err)

pkg/hostman/host_health/doc.go

@@ -1 +1,15 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package host_health // import "yunion.io/x/onecloud/pkg/hostman/host_health"

pkg/hostman/host_health/etcd.go

@@ -1,3 +1,17 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package host_health
 
 import (

pkg/hostman/host_health/health_manager.go

@@ -1,3 +1,17 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package host_health
 
 import (

pkg/hostman/hosthandler/doc.go

@@ -1 +1,15 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package hosthandler // import "yunion.io/x/onecloud/pkg/hostman/hosthandler"

pkg/hostman/hosthandler/handler.go

@@ -1,3 +1,17 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package hosthandler
 
 import (

pkg/hostman/hostinfo/hostinfo.go

@@ -35,6 +35,7 @@ import (
 	"yunion.io/x/pkg/utils"
 
 	api "yunion.io/x/onecloud/pkg/apis/compute"
+	identityapi "yunion.io/x/onecloud/pkg/apis/identity"
 	"yunion.io/x/onecloud/pkg/hostman/guestfs/fsdriver"
 	"yunion.io/x/onecloud/pkg/hostman/host_health"
 	deployapi "yunion.io/x/onecloud/pkg/hostman/hostdeployer/apis"
@@ -45,7 +46,6 @@ import (
 	"yunion.io/x/onecloud/pkg/hostman/storageman"
 	"yunion.io/x/onecloud/pkg/hostman/system_service"
 	"yunion.io/x/onecloud/pkg/mcclient"
-	"yunion.io/x/onecloud/pkg/mcclient/auth"
 	"yunion.io/x/onecloud/pkg/mcclient/modules"
 	"yunion.io/x/onecloud/pkg/util/cgrouputils"
 	"yunion.io/x/onecloud/pkg/util/fileutils2"
@@ -1500,7 +1500,7 @@ func (h *SHostInfo) OnCatalogChanged(catalog mcclient.KeystoneServiceCatalogV3)
 	// TODO: dynamic probe endpoint type
 	defaultEndpointType := options.HostOptions.SessionEndpointType
 	if len(defaultEndpointType) == 0 {
-		defaultEndpointType = auth.PublicEndpointType
+		defaultEndpointType = identityapi.EndpointInterfacePublic
 	}
 	if options.HostOptions.ManageNtpConfiguration {
 		ntpd := system_service.GetService("ntpd")

pkg/keystone/models/endpoints.go

@@ -16,11 +16,13 @@ package models
 
 import (
 	"context"
+	"crypto/tls"
 	"database/sql"
 	"fmt"
 	"strings"
 
 	"yunion.io/x/jsonutils"
+	"yunion.io/x/log"
 	"yunion.io/x/pkg/errors"
 	"yunion.io/x/pkg/tristate"
 	"yunion.io/x/sqlchemy"
@@ -28,9 +30,13 @@ import (
 	"yunion.io/x/onecloud/pkg/apis"
 	api "yunion.io/x/onecloud/pkg/apis/identity"
 	"yunion.io/x/onecloud/pkg/cloudcommon/db"
+	"yunion.io/x/onecloud/pkg/cloudcommon/etcd"
+	"yunion.io/x/onecloud/pkg/cloudcommon/informer"
 	"yunion.io/x/onecloud/pkg/httperrors"
+	"yunion.io/x/onecloud/pkg/keystone/options"
 	"yunion.io/x/onecloud/pkg/mcclient"
 	"yunion.io/x/onecloud/pkg/util/logclient"
+	"yunion.io/x/onecloud/pkg/util/seclib2"
 	"yunion.io/x/onecloud/pkg/util/stringutils2"
 )
 
@@ -38,6 +44,8 @@ type SEndpointManager struct {
 	db.SStandaloneResourceBaseManager
 	SServiceResourceBaseManager
 	SRegionResourceBaseManager
+
+	informerBackends map[string]informer.IInformerBackend
 }
 
 var EndpointManager *SEndpointManager
@@ -108,6 +116,119 @@ func (manager *SEndpointManager) InitializeData() error {
 			})
 		}
 	}
+	if err := manager.SetInformerBackend(); err != nil {
+		log.Errorf("init informer backend error: %v", err)
+	}
+	return nil
+}
+
+func (manager *SEndpointManager) SetInformerBackend() error {
+	informerEp, err := manager.fetchInformerEndpoint()
+	if err != nil {
+		return errors.Wrap(err, "fetch informer endpoint")
+	}
+	return manager.SetInformerBackendByEndpoint(informerEp)
+}
+
+func (manager *SEndpointManager) getSessionEndpointType() string {
+	epType := api.EndpointInterfaceInternal
+	if options.Options.SessionEndpointType != "" {
+		epType = options.Options.SessionEndpointType
+	}
+	return epType
+}
+
+func (manager *SEndpointManager) IsEtcdInformerBackend(ep *SEndpoint) bool {
+	if ep == nil {
+		return false
+	}
+	svc := ep.getService()
+	if svc == nil {
+		return false
+	}
+	if svc.GetName() != api.SERVICE_TYPE_ETCD {
+		return false
+	}
+	epType := manager.getSessionEndpointType()
+	if ep.Interface != epType {
+		return false
+	}
+	return true
+}
+
+func (manager *SEndpointManager) SetInformerBackendByEndpoint(ep *SEndpoint) error {
+	if !manager.IsEtcdInformerBackend(ep) {
+		return nil
+	}
+	return manager.SetEtcdInformerBackend(ep)
+}
+
+func (manager *SEndpointManager) fetchInformerEndpoint() (*SEndpoint, error) {
+	epType := manager.getSessionEndpointType()
+
+	endpoints := manager.Query().SubQuery()
+	services := ServiceManager.Query().SubQuery()
+	regions := RegionManager.Query().SubQuery()
+	q := endpoints.Query()
+	q = q.Join(regions, sqlchemy.Equals(endpoints.Field("region_id"), regions.Field("id")))
+	q = q.Join(services, sqlchemy.Equals(endpoints.Field("service_id"), services.Field("id")))
+	q = q.Filter(sqlchemy.AND(
+		sqlchemy.Equals(endpoints.Field("interface"), epType),
+		sqlchemy.IsTrue(endpoints.Field("enabled"))))
+	q = q.Filter(sqlchemy.AND(
+		sqlchemy.IsTrue(services.Field("enabled")),
+		sqlchemy.Equals(services.Field("type"), api.SERVICE_TYPE_ETCD)))
+
+	informerEp := new(SEndpoint)
+	if err := q.First(informerEp); err != nil {
+		return nil, err
+	}
+	informerEp.SetModelManager(manager, informerEp)
+
+	return informerEp, nil
+}
+
+func (manager *SEndpointManager) newEtcdInformerBackend(ep *SEndpoint) (informer.IInformerBackend, error) {
+	useTLS := false
+	var (
+		tlsCfg *tls.Config
+	)
+	if ep.ServiceCertificateId != "" {
+		useTLS = true
+		cert, err := ep.getServiceCertificate()
+		if err != nil {
+			return nil, errors.Wrap(err, "get service certificate")
+		}
+		caData := []byte(cert.CaCertificate)
+		certData := []byte(cert.Certificate)
+		keyData := []byte(cert.PrivateKey)
+		cfg, err := seclib2.InitTLSConfigByData(caData, certData, keyData)
+		if err != nil {
+			return nil, errors.Wrap(err, "build TLS config")
+		}
+		// always set insecure skip verify
+		cfg.InsecureSkipVerify = true
+		tlsCfg = cfg
+	}
+	opt := &etcd.SEtcdOptions{
+		EtcdEndpoint:              []string{ep.Url},
+		EtcdTimeoutSeconds:        5,
+		EtcdRequestTimeoutSeconds: 10,
+		EtcdLeaseExpireSeconds:    5,
+	}
+	if useTLS {
+		opt.TLSConfig = tlsCfg
+		opt.EtcdEnabldSsl = true
+	}
+	return informer.NewEtcdBackend(opt, nil)
+}
+
+func (manager *SEndpointManager) SetEtcdInformerBackend(ep *SEndpoint) error {
+	be, err := manager.newEtcdInformerBackend(ep)
+	if err != nil {
+		return errors.Wrap(err, "new etcd informer backend")
+	}
+	informer.Set(be)
 	return nil
 }
 
@@ -255,9 +376,8 @@ func (endpoint *SEndpoint) GetExtraDetails(
 
 func (endpoint *SEndpoint) getMoreDetails(details api.EndpointDetails) (api.EndpointDetails, error) {
 	if len(endpoint.ServiceCertificateId) > 0 {
-		icert, _ := ServiceCertificateManager.FetchById(endpoint.ServiceCertificateId)
-		if icert != nil {
-			cert := icert.(*SServiceCertificate)
+		cert, _ := endpoint.getServiceCertificate()
+		if cert != nil {
 			certOutput := cert.ToOutput()
 			details.CertificateDetails = *certOutput
 		}
@@ -265,6 +385,18 @@ func (endpoint *SEndpoint) getMoreDetails(details api.EndpointDetails) (api.Endp
 	return details, nil
 }
 
+func (endpoint *SEndpoint) getServiceCertificate() (*SServiceCertificate, error) {
+	certId := endpoint.ServiceCertificateId
+	if certId == "" {
+		return nil, nil
+	}
+	icert, err := ServiceCertificateManager.FetchById(certId)
+	if err != nil {
+		return nil, err
+	}
+	return icert.(*SServiceCertificate), nil
+}
+
 func (manager *SEndpointManager) FetchCustomizeColumns(
 	ctx context.Context,
 	userCred mcclient.TokenCredential,
@@ -449,22 +581,34 @@ func (manager *SEndpointManager) QueryDistinctExtraField(q *sqlchemy.SQuery, fie
 	return q, httperrors.ErrNotFound
 }
 
+func (endpoint *SEndpoint) trySetInformerBackend() {
+	if err := EndpointManager.SetInformerBackendByEndpoint(endpoint); err != nil {
+		log.Errorf("Set informer by endpoint %s error: %v", endpoint.GetName(), err)
+	}
+}
+
 func (endpoint *SEndpoint) PostCreate(ctx context.Context, userCred mcclient.TokenCredential, ownerId mcclient.IIdentityProvider, query jsonutils.JSONObject, data jsonutils.JSONObject) {
 	endpoint.SStandaloneResourceBase.PostCreate(ctx, userCred, ownerId, query, data)
 	logclient.AddActionLogWithContext(ctx, endpoint, logclient.ACT_CREATE, data, userCred, true)
 	refreshDefaultClientServiceCatalog()
+	endpoint.trySetInformerBackend()
 }
 
 func (endpoint *SEndpoint) PostUpdate(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, data jsonutils.JSONObject) {
 	endpoint.SStandaloneResourceBase.PostUpdate(ctx, userCred, query, data)
 	logclient.AddActionLogWithContext(ctx, endpoint, logclient.ACT_UPDATE, data, userCred, true)
 	refreshDefaultClientServiceCatalog()
+	endpoint.trySetInformerBackend()
 }
 
 func (endpoint *SEndpoint) PostDelete(ctx context.Context, userCred mcclient.TokenCredential) {
 	endpoint.SStandaloneResourceBase.PostDelete(ctx, userCred)
 	logclient.AddActionLogWithContext(ctx, endpoint, logclient.ACT_DELETE, nil, userCred, true)
 	refreshDefaultClientServiceCatalog()
+	if EndpointManager.IsEtcdInformerBackend(endpoint) {
+		// remove informer backend
+		informer.Set(nil)
+	}
 }
 
 func (endpoint *SEndpoint) ValidateUpdateData(

pkg/keystone/models/servicecertificates.go

@@ -1,3 +1,17 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package models
 
 import (

pkg/keystone/options/options.go

@@ -49,6 +49,8 @@ type SKeystoneOptions struct {
 	DefaultProjectQuota int `default:"100" help:"default quota for project per domain, default is 500"`
 	DefaultRoleQuota    int `default:"100" help:"default quota for role per domain, default is 500"`
 	DefaultPolicyQuota  int `default:"100" help:"default quota for policy per domain, default is 500"`
+
+	SessionEndpointType string `help:"Client session end point type"`
 }
 
 var (

pkg/mcclient/auth/auth.go

@@ -24,6 +24,7 @@ import (
 	"yunion.io/x/log"
 	"yunion.io/x/pkg/util/cache"
 
+	"yunion.io/x/onecloud/pkg/apis/identity"
 	"yunion.io/x/onecloud/pkg/mcclient"
 )
 
@@ -35,11 +36,6 @@ var (
 	globalEndpointType string
 )
 
-const (
-	PublicEndpointType   string = "public"
-	InternalEndpointType string = "internal"
-)
-
 type AuthInfo struct {
 	AuthUrl string
 	// Domain not need when v2 auth
@@ -271,7 +267,7 @@ func GetServiceURL(service, region, zone, endpointType string) (string, error) {
 }
 
 func GetPublicServiceURL(service, region, zone string) (string, error) {
-	return manager.GetServiceURL(service, region, zone, PublicEndpointType)
+	return manager.GetServiceURL(service, region, zone, identity.EndpointInterfacePublic)
 }
 
 func GetServiceURLs(service, region, zone, endpointType string) ([]string, error) {
@@ -369,11 +365,11 @@ func GetSession(ctx context.Context, token mcclient.TokenCredential, region stri
 }
 
 func GetSessionWithInternal(ctx context.Context, token mcclient.TokenCredential, region string, apiVersion string) *mcclient.ClientSession {
-	return getSessionByType(ctx, token, region, apiVersion, InternalEndpointType)
+	return getSessionByType(ctx, token, region, apiVersion, identity.EndpointInterfaceInternal)
 }
 
 func GetSessionWithPublic(ctx context.Context, token mcclient.TokenCredential, region string, apiVersion string) *mcclient.ClientSession {
-	return getSessionByType(ctx, token, region, apiVersion, PublicEndpointType)
+	return getSessionByType(ctx, token, region, apiVersion, identity.EndpointInterfacePublic)
 }
 
 func getSessionByType(ctx context.Context, token mcclient.TokenCredential, region string, apiVersion string, epType string) *mcclient.ClientSession {

pkg/mcclient/informer/doc.go

@@ -0,0 +1,15 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package informer // import "yunion.io/x/onecloud/pkg/mcclient/informer"

pkg/mcclient/informer/watcher.go

@@ -0,0 +1,183 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package informer
+
+import (
+	"context"
+
+	"yunion.io/x/jsonutils"
+	"yunion.io/x/pkg/errors"
+
+	"yunion.io/x/onecloud/pkg/cloudcommon/etcd"
+	"yunion.io/x/onecloud/pkg/cloudcommon/informer"
+	"yunion.io/x/onecloud/pkg/mcclient"
+)
+
+type SWatchManager struct {
+	client       *mcclient.Client
+	watchBackend informer.IWatcher
+}
+
+func NewWatchManagerBySession(session *mcclient.ClientSession, onKeepaliveFailure func()) (*SWatchManager, error) {
+	return NewWatchManager(session.GetClient(), session.GetToken(), session.GetRegion(), session.GetEndpointType(), onKeepaliveFailure)
+}
+
+func NewWatchManager(client *mcclient.Client, token mcclient.TokenCredential, region, interfaceType string, onKeepaliveFailure func()) (*SWatchManager, error) {
+	endpoint, err := client.GetCommonEtcdEndpoint(token, region, interfaceType)
+	if err != nil {
+		return nil, errors.Wrap(err, "get common etcd endpoint")
+	}
+	tlsCfg, err := client.GetCommonEtcdTLSConfig(endpoint)
+	if err != nil {
+		return nil, errors.Wrap(err, "get common etcd tls config")
+	}
+	opt := &etcd.SEtcdOptions{
+		EtcdEndpoint:              []string{endpoint.Url},
+		EtcdTimeoutSeconds:        5,
+		EtcdRequestTimeoutSeconds: 10,
+		EtcdLeaseExpireSeconds:    5,
+	}
+	if tlsCfg != nil {
+		tlsCfg.InsecureSkipVerify = true
+		opt.EtcdEnabldSsl = true
+		opt.TLSConfig = tlsCfg
+	}
+	be, err := informer.NewEtcdBackend(opt, onKeepaliveFailure)
+	if err != nil {
+		return nil, errors.Wrap(err, "new etcd informer backend")
+	}
+	man := &SWatchManager{
+		client:       client,
+		watchBackend: be,
+	}
+	return man, nil
+}
+
+type IResourceManager interface {
+	KeyString() string
+	GetKeyword() string
+}
+
+type EventHandler interface {
+	OnAdd(obj *jsonutils.JSONDict)
+	OnUpdate(oldObj, newObj *jsonutils.JSONDict)
+	OnDelete(obj *jsonutils.JSONDict)
+}
+
+type IWatcher interface {
+	AddEventHandler(ctx context.Context, handler EventHandler) error
+}
+
+type sWatcher struct {
+	manager         *SWatchManager
+	resourceManager IResourceManager
+	ctx             context.Context
+	eventHandler    EventHandler
+}
+
+func (man *SWatchManager) For(resourceManager IResourceManager) IWatcher {
+	return &sWatcher{
+		manager:         man,
+		resourceManager: resourceManager,
+	}
+}
+
+func (w *sWatcher) AddEventHandler(ctx context.Context, handler EventHandler) error {
+	w.ctx = ctx
+	w.eventHandler = w.wrapEventHandler(handler)
+	return w.manager.watch(w.ctx, w.resourceManager, w.eventHandler)
+}
+
+func (man *SWatchManager) watch(ctx context.Context, resourceManager IResourceManager, handler informer.ResourceEventHandler) error {
+	return man.watchBackend.Watch(ctx, resourceManager.KeyString(), handler)
+}
+
+func (w *sWatcher) wrapEventHandler(handler EventHandler) informer.ResourceEventHandler {
+	return &wrapEventHandler{handler}
+}
+
+type wrapEventHandler struct {
+	handler EventHandler
+}
+
+func (h *wrapEventHandler) OnAdd(obj *jsonutils.JSONDict) {
+	h.handler.OnAdd(obj)
+}
+
+func (h *wrapEventHandler) OnUpdate(oldObj, newObj *jsonutils.JSONDict) {
+	h.handler.OnUpdate(oldObj, newObj)
+}
+
+func (h *wrapEventHandler) OnDelete(obj *jsonutils.JSONDict) {
+	h.handler.OnDelete(obj)
+}
+
+type EventHandlerFuncs struct {
+	AddFunc    func(obj *jsonutils.JSONDict)
+	UpdateFunc func(oldObj, newObj *jsonutils.JSONDict)
+	DeleteFunc func(obj *jsonutils.JSONDict)
+}
+
+func (r EventHandlerFuncs) OnAdd(obj *jsonutils.JSONDict) {
+	if r.AddFunc != nil {
+		r.AddFunc(obj)
+	}
+}
+
+func (r EventHandlerFuncs) OnUpdate(oldObj, newObj *jsonutils.JSONDict) {
+	if r.UpdateFunc != nil {
+		r.UpdateFunc(oldObj, newObj)
+	}
+}
+
+func (r EventHandlerFuncs) OnDelete(obj *jsonutils.JSONDict) {
+	if r.DeleteFunc != nil {
+		r.DeleteFunc(obj)
+	}
+}
+
+type FilteringEventHandler struct {
+	FilterFunc func(obj *jsonutils.JSONDict) bool
+	Handler    EventHandler
+}
+
+func (r FilteringEventHandler) OnAdd(obj *jsonutils.JSONDict) {
+	if !r.FilterFunc(obj) {
+		return
+	}
+	r.Handler.OnAdd(obj)
+}
+
+func (r FilteringEventHandler) OnUpdate(oldObj, newObj *jsonutils.JSONDict) {
+	newer := r.FilterFunc(newObj)
+	older := r.FilterFunc(oldObj)
+	switch {
+	case newer && older:
+		r.Handler.OnUpdate(oldObj, newObj)
+	case newer && !older:
+		r.Handler.OnAdd(newObj)
+	case !newer && older:
+		r.Handler.OnDelete(oldObj)
+	default:
+		// do nothing
+	}
+}
+
+func (r FilteringEventHandler) OnDelete(obj *jsonutils.JSONDict) {
+	if !r.FilterFunc(obj) {
+		return
+	}
+	r.Handler.OnDelete(obj)
+}

pkg/mcclient/mcclient.go

@@ -25,9 +25,11 @@ import (
 
 	"yunion.io/x/jsonutils"
 	"yunion.io/x/log"
+	"yunion.io/x/pkg/errors"
 	"yunion.io/x/pkg/gotypes"
 
 	api "yunion.io/x/onecloud/pkg/apis/identity"
+	"yunion.io/x/onecloud/pkg/httperrors"
 	"yunion.io/x/onecloud/pkg/util/httputils"
 	"yunion.io/x/onecloud/pkg/util/seclib2"
 )
@@ -334,6 +336,50 @@ func (this *Client) SetProject(tenantId, tenantName, tenantDomain string, token
 	}
 }
 
+func (this *Client) GetCommonEtcdEndpoint(token TokenCredential, region, interfaceType string) (*api.EndpointDetails, error) {
+	if this.AuthVersion() != "v3" {
+		return nil, errors.Errorf("current version %s not support get internal etcd endpoint", this.AuthVersion())
+	}
+
+	params := jsonutils.NewDict()
+	params.Add(jsonutils.NewString(interfaceType), "interface")
+	params.Add(jsonutils.JSONTrue, "enabled")
+	params.Add(jsonutils.NewString(api.SERVICE_TYPE_ETCD), "service")
+	params.Add(jsonutils.JSONTrue, "details")
+	params.Add(jsonutils.NewString(region), "region")
+
+	epUrl := "/endpoints?" + params.QueryString()
+	_, rbody, err := this.jsonRequest(context.Background(), this.authUrl, token.GetTokenString(), httputils.GET, epUrl, nil, nil)
+	if err != nil {
+		return nil, errors.Wrap(err, "get internal etcd endpoint")
+	}
+	rets, err := rbody.GetArray("endpoints")
+	if err != nil {
+		return nil, errors.Wrap(err, "get endpoints response")
+	}
+	if len(rets) == 0 {
+		return nil, errors.Wrapf(httperrors.ErrNotFound, "not found service %s %s endpoint", api.SERVICE_TYPE_ETCD, interfaceType)
+	}
+	if len(rets) > 1 {
+		return nil, errors.Errorf("fond %d duplicate serivce %s %s endpoint", len(rets), api.SERVICE_TYPE_ETCD, interfaceType)
+	}
+	endpoint := new(api.EndpointDetails)
+	if err := rets[0].Unmarshal(endpoint); err != nil {
+		return nil, errors.Wrap(err, "unmarshal endpoint")
+	}
+	return endpoint, nil
+}
+
+func (this *Client) GetCommonEtcdTLSConfig(endpoint *api.EndpointDetails) (*tls.Config, error) {
+	if endpoint.CertId == "" {
+		return nil, nil
+	}
+	caData := []byte(endpoint.CaCertificate)
+	certData := []byte(endpoint.Certificate)
+	keyData := []byte(endpoint.PrivateKey)
+	return seclib2.InitTLSConfigByData(caData, certData, keyData)
+}
+
 func (this *Client) NewSession(ctx context.Context, region, zone, endpointType string, token TokenCredential, apiVersion string) *ClientSession {
 	cata := token.GetServiceCatalog()
 	if this.serviceCatalog == nil {

pkg/mcclient/modulebase/resource.go

@@ -47,27 +47,27 @@ type ResourceManager struct {
 	idFieldName   string
 }
 
-func (this *ResourceManager) GetKeyword() string {
+func (this ResourceManager) GetKeyword() string {
 	return this.Keyword
 }
 
-func (this *ResourceManager) KeyString() string {
+func (this ResourceManager) KeyString() string {
 	return this.KeywordPlural
 }
 
-func (this *ResourceManager) Version() string {
+func (this ResourceManager) Version() string {
 	return this.version
 }
 
-func (this *ResourceManager) ServiceType() string {
+func (this ResourceManager) ServiceType() string {
 	return this.serviceType
 }
 
-func (this *ResourceManager) EndpointType() string {
+func (this ResourceManager) EndpointType() string {
 	return this.endpointType
 }
 
-func (this *ResourceManager) URLPath() string {
+func (this ResourceManager) URLPath() string {
 	return strings.Replace(this.KeywordPlural, ":", "/", -1)
 }
 

pkg/mcclient/modules/mod_service_certficates.go

@@ -1,3 +1,17 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package modules
 
 import "yunion.io/x/onecloud/pkg/mcclient/modulebase"

pkg/mcclient/options/monitor/influxdbshema.go

@@ -1,3 +1,17 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package monitor
 
 import (

pkg/mcclient/options/servicecertificates.go

@@ -1,3 +1,17 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package options
 
 import (

pkg/mcclient/session.go

@@ -386,3 +386,14 @@ func (this *ClientSession) ToJson() jsonutils.JSONObject {
 func (cs *ClientSession) GetToken() TokenCredential {
 	return cs.token
 }
+
+func (cs *ClientSession) GetContext() context.Context {
+	if cs.ctx == nil {
+		return context.Background()
+	}
+	return cs.ctx
+}
+
+func (cs *ClientSession) GetCommonEtcdEndpoint() (*api.EndpointDetails, error) {
+	return cs.GetClient().GetCommonEtcdEndpoint(cs.GetToken(), cs.region, cs.endpointType)
+}

pkg/monitor/alerting/conditions/suggestrulereducer.go

@@ -1,3 +1,17 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package conditions
 
 import (

pkg/monitor/models/datasource.go

@@ -29,6 +29,7 @@ import (
 	"yunion.io/x/pkg/tristate"
 	"yunion.io/x/pkg/util/wait"
 
+	identityapi "yunion.io/x/onecloud/pkg/apis/identity"
 	"yunion.io/x/onecloud/pkg/apis/monitor"
 	"yunion.io/x/onecloud/pkg/cloudcommon/db"
 	"yunion.io/x/onecloud/pkg/httperrors"
@@ -98,7 +99,7 @@ func (man *SDataSourceManager) initDefaultDataSource(ctx context.Context) error
 			log.Errorf("get empty public session for region %s", region)
 			return
 		}
-		url, err := s.GetServiceURL("influxdb", auth.PublicEndpointType)
+		url, err := s.GetServiceURL("influxdb", identityapi.EndpointInterfacePublic)
 		if err != nil {
 			log.Errorf("get influxdb public url: %v", err)
 			return

pkg/monitor/suggestsysdrivers/common.go

@@ -1,3 +1,17 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package suggestsysdrivers
 
 import (

pkg/monitor/suggestsysdrivers/diskunused.go

@@ -1,3 +1,17 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package suggestsysdrivers
 
 import (

pkg/monitor/suggestsysdrivers/lbunused.go

@@ -1,3 +1,17 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package suggestsysdrivers
 
 import (

pkg/monitor/suggestsysdrivers/scaledown.go

@@ -1,3 +1,17 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package suggestsysdrivers
 
 import (

pkg/util/seclib2/tls.go

@@ -22,6 +22,7 @@ import (
 	"io/ioutil"
 
 	"yunion.io/x/log"
+	"yunion.io/x/pkg/errors"
 )
 
 var CERT_SEP = []byte("-END CERTIFICATE-")
@@ -152,3 +153,31 @@ func InitTLSConfig(certFile, keyFile string) (*tls.Config, error) {
 	tlsConfig.BuildNameToCertificate()
 	return tlsConfig, nil
 }
+
+func InitTLSConfigByData(caCertBlock, certPEMBlock, keyPEMBlock []byte) (*tls.Config, error) {
+	cert, err := tls.X509KeyPair(certPEMBlock, keyPEMBlock)
+	if err != nil {
+		return nil, err
+	}
+
+	caCertPool := x509.NewCertPool()
+	for {
+		var block *pem.Block
+		block, caCertBlock = pem.Decode(caCertBlock)
+		if block == nil {
+			break
+		}
+		caCert, err := x509.ParseCertificate(block.Bytes)
+		if err != nil {
+			return nil, errors.Wrap(err, "parse caCert data")
+		}
+		caCertPool.AddCert(caCert)
+	}
+
+	tlsConfig := &tls.Config{
+		Certificates: []tls.Certificate{cert},
+		RootCAs:      caCertPool,
+	}
+	tlsConfig.BuildNameToCertificate()
+	return tlsConfig, nil
+}