lazyload: optimiz some svf reseted when master-slave switch

framework/model/metric/accesslog_convertor.go

@@ -17,21 +17,14 @@ type AccessLogConvertor struct {
 }
 
 func NewAccessLogConvertor(config AccessLogConvertorConfig) *AccessLogConvertor {
-
- newCacheResultCopy := make(map[string]map[string]string)
- for meta, value := range config.InitCache {
- tmpValue := make(map[string]string)
- for k, v := range value {
- tmpValue[k] = v
- }
- newCacheResultCopy[meta] = tmpValue
- }
+ result := make(map[string]map[string]string)
+ resultCopy := make(map[string]map[string]string)
 
  return &AccessLogConvertor{
  name: config.Name,
  handler: config.Handler,
- cacheResult: config.InitCache,
- cacheResultCopy: newCacheResultCopy,
+ cacheResult: result,
+ cacheResultCopy: resultCopy,
  }
 }
 

framework/model/metric/accesslog_source.go

@@ -136,3 +136,21 @@ func (s *AccessLogSource) Reset(info string) error {
 
  return nil
 }
+
+func (s *AccessLogSource) Fullfill(cache map[string]map[string]string) error {
+
+ for _, convertor := range s.convertors {
+ convertor.convertorLock.Lock()
+ for meta, value := range cache {
+ convertor.cacheResult[meta] = value
+ tmpValue := make(map[string]string)
+ for k, v := range value {
+ tmpValue[k] = v
+ }
+
+ convertor.cacheResultCopy[meta] = tmpValue
+ }
+ convertor.convertorLock.Unlock()
+ }
+ return nil
+}

framework/model/metric/config.go

@@ -44,7 +44,6 @@ type AccessLogSourceConfig struct {
 }
 
 type AccessLogConvertorConfig struct {
- Name string // handler name
- Handler func(logEntry []*data_accesslog.HTTPAccessLogEntry) (map[string]map[string]string, error)
- InitCache map[string]map[string]string // for cacheResult init
+ Name string // handler name
+ Handler func(logEntry []*data_accesslog.HTTPAccessLogEntry) (map[string]map[string]string, error)
 }

framework/model/metric/mock_source.go

@@ -21,3 +21,7 @@ func (s *MockSource) QueryMetric(queryMap QueryMap) (Metric, error) {
 func (s *MockSource) Reset(info string) error {
  return nil
 }
+
+func (s *MockSource) Fullfill(cache map[string]map[string]string) error {
+ return nil
+}

framework/model/metric/model.go

@@ -18,4 +18,5 @@ type Source interface {
  QueryMetric(queryMap QueryMap) (Metric, error)
  Start() error
  Reset(info string) error
+ Fullfill(map[string]map[string]string) error
 }

framework/model/metric/prometheus_source.go

@@ -67,6 +67,10 @@ func (ps *PrometheusSource) Reset(info string) error {
  return nil
 }
 
+func (ps *PrometheusSource) Fullfill(cache map[string]map[string]string) error {
+ return nil
+}
+
 func defaultConvertor(qv prometheusModel.Value) map[string]string {
  result := make(map[string]string)
 

staging/src/slime.io/slime/modules/lazyload/controllers/producer.go

@@ -127,21 +127,13 @@ func NewProducerConfig(env bootstrap.Environment) (*metric.ProducerConfig, error
  // init log source port
  port := env.Config.Global.Misc["logSourcePort"]
 
- // init initCache
- initCache, err := newInitCache(env)
- if err != nil {
- return nil, err
- }
- log.Debugf("initCache is %+v", initCache)
-
  // init accessLog source config
  accessLogSourceConfig = metric.AccessLogSourceConfig{
  ServePort: port,
  AccessLogConvertorConfigs: []metric.AccessLogConvertorConfig{
  {
- Name: AccessLogConvertorName,
- Handler: nil,
- InitCache: initCache,
+ Name: AccessLogConvertorName,
+ Handler: nil,
  },
  },
  }
@@ -209,7 +201,7 @@ func newPrometheusSourceConfig(env bootstrap.Environment) (metric.PrometheusSour
  }, nil
 }
 
-func newInitCache(env bootstrap.Environment) (map[string]map[string]string, error) {
+func NewCache(env bootstrap.Environment) (map[string]map[string]string, error) {
  result := make(map[string]map[string]string)
 
  svfGvr := schema.GroupVersionResource{

staging/src/slime.io/slime/modules/lazyload/go.mod

@@ -17,6 +17,7 @@ require (
  k8s.io/api v0.22.2
  k8s.io/apimachinery v0.22.2
  k8s.io/client-go v0.22.2
+ k8s.io/kube-openapi v0.0.0-20210421082810-95288971da7e
  sigs.k8s.io/controller-runtime v0.10.3
  sigs.k8s.io/yaml v1.2.0
  slime.io/slime/framework v0.0.0-00010101000000-000000000000

staging/src/slime.io/slime/modules/lazyload/module/module.go

@@ -77,7 +77,17 @@ func (m *Module) Setup(opts module.ModuleOptions) error {
  sfReconciler.StartIpToSvcCache(ctx)
  })
 
+ // build metric source
  source := metric.NewSource(pc)
+
+ cache, err := controllers.NewCache(env)
+ if err != nil {
+ return fmt.Errorf("GetCacheFromServicefence occured err: %s", err)
+ }
+ source.Fullfill(cache)
+ log.Debugf("GetCacheFromServicefence %+v", cache)
+
+ // register svf reset
  handler := &server.Handler{
  HttpPathHandler: env.HttpPathHandler,
  Source: source,
@@ -122,6 +132,17 @@ func (m *Module) Setup(opts module.ModuleOptions) error {
  return fmt.Errorf("unable to create controller,%+v", err)
  }
 
+ le.AddOnStartedLeading(func(ctx context.Context) {
+ log.Infof("retrieve metric from svf status.metric")
+ cache, err := controllers.NewCache(env)
+ if err != nil {
+ log.Warnf("GetCacheFromServicefence occured err in StartedLeading: %s", err)
+ return
+ }
+ source.Fullfill(cache)
+ log.Debugf("GetCacheFromServicefence is %+v", cache)
+ })
+
  le.AddOnStartedLeading(func(ctx context.Context) {
  log.Infof("producers starts")
  metric.NewProducer(pc, source)