// checkCache checks the cache for metric types.
// returns:
//  - array of metrics that need to be collected
//  - array of metrics that were returned from the cache
func checkCache(mts []core.Metric) ([]plugin.PluginMetricType, []core.Metric) {
	var fromCache []core.Metric
	var metricsToCollect []plugin.PluginMetricType
	for _, mt := range mts {
		if m := metricCache.get(core.JoinNamespace(mt.Namespace()), mt.Version()); m != nil {
			switch metric := m.(type) {
			case core.Metric:
				fromCache = append(fromCache, metric)
			case []core.Metric:
				for _, met := range metric {
					fromCache = append(fromCache, met)
				}
			default:
				log.WithFields(log.Fields{
					"_module": "client",
					"_block":  "checkCache",
				}).Error("unsupported type found in the cache")
			}
		} else {
			mt := plugin.PluginMetricType{
				Namespace_:          mt.Namespace(),
				LastAdvertisedTime_: mt.LastAdvertisedTime(),
				Version_:            mt.Version(),
				Tags_:               mt.Tags(),
				Labels_:             mt.Labels(),
				Config_:             mt.Config(),
			}
			metricsToCollect = append(metricsToCollect, mt)
		}
	}
	return metricsToCollect, fromCache
}

func respondWithMetrics(host string, mets []core.CatalogedMetric, w http.ResponseWriter) {
	b := rbody.NewMetricsReturned()

	for _, met := range mets {
		rt := met.Policy().RulesAsTable()
		policies := make([]rbody.PolicyTable, 0, len(rt))
		for _, r := range rt {
			policies = append(policies, rbody.PolicyTable{
				Name:     r.Name,
				Type:     r.Type,
				Default:  r.Default,
				Required: r.Required,
				Minimum:  r.Minimum,
				Maximum:  r.Maximum,
			})
		}
		b = append(b, rbody.Metric{
			Namespace:               core.JoinNamespace(met.Namespace()),
			Version:                 met.Version(),
			LastAdvertisedTimestamp: met.LastAdvertisedTime().Unix(),
			Policy:                  policies,
			Href:                    catalogedMetricURI(host, met),
		})
	}
	sort.Sort(b)
	respond(200, b, w)
}

func (s *Server) getMetricsFromTree(w http.ResponseWriter, r *http.Request, params httprouter.Params) {
	ns := parseNamespace(params.ByName("namespace"))

	var (
		ver int
		err error
	)
	q := r.URL.Query()
	v := q.Get("ver")
	if v == "" {
		ver = -1
	} else {
		ver, err = strconv.Atoi(v)
		if err != nil {
			respond(400, rbody.FromError(err), w)
			return
		}
	}

	if ns[len(ns)-1] == "*" {
		mets, err := s.mm.FetchMetrics(ns[:len(ns)-1], ver)
		if err != nil {
			respond(404, rbody.FromError(err), w)
			return
		}
		respondWithMetrics(r.Host, mets, w)
		return
	}

	mt, err := s.mm.GetMetric(ns, ver)
	if err != nil {
		respond(404, rbody.FromError(err), w)
		return
	}

	b := &rbody.MetricReturned{}
	mb := &rbody.Metric{
		Namespace:               core.JoinNamespace(mt.Namespace()),
		Version:                 mt.Version(),
		LastAdvertisedTimestamp: mt.LastAdvertisedTime().Unix(),
		Href: catalogedMetricURI(r.Host, mt),
	}
	rt := mt.Policy().RulesAsTable()
	policies := make([]rbody.PolicyTable, 0, len(rt))
	for _, r := range rt {
		policies = append(policies, rbody.PolicyTable{
			Name:     r.Name,
			Type:     r.Type,
			Default:  r.Default,
			Required: r.Required,
			Minimum:  r.Minimum,
			Maximum:  r.Maximum,
		})
	}
	mb.Policy = policies
	b.Metric = mb
	respond(200, b, w)
}

// updateCache updates the cache with the given array of metrics.
func (l *lru) UpdateCache(mts []core.Metric) {
	results := []core.Metric{}
	dc := map[string][]core.Metric{}
	for _, mt := range mts {
		if mt.Labels() == nil {
			// cache the individual metric
			l.metricCache.put(core.JoinNamespace(mt.Namespace()), mt.Version(), mt)
			l.logger.Debugf("putting %v:%v in the cache", mt.Namespace(), mt.Version())
		} else {
			// collect the dynamic query results so we can cache
			ns := make([]string, len(mt.Namespace()))
			copy(ns, mt.Namespace())
			for _, label := range mt.Labels() {
				ns[label.Index] = "*"
			}
			if _, ok := dc[core.JoinNamespace(ns)]; !ok {
				dc[core.JoinNamespace(ns)] = []core.Metric{}
			}
			dc[core.JoinNamespace(ns)] = append(dc[core.JoinNamespace(ns)], mt)
			l.metricCache.put(core.JoinNamespace(ns), mt.Version(), dc[core.JoinNamespace(ns)])
			l.logger.Debugf("putting %v:%v in the cache", ns, mt.Version())
		}
		results = append(results, mt)
	}
}

func (p *pluginControl) validateMetricTypeSubscription(mt core.RequestedMetric, cd *cdata.ConfigDataNode) (core.Metric, []serror.SnapError) {
	var serrs []serror.SnapError
	controlLogger.WithFields(log.Fields{
		"_block":    "validate-metric-subscription",
		"namespace": mt.Namespace(),
		"version":   mt.Version(),
	}).Info("subscription called on metric")

	m, err := p.metricCatalog.Get(mt.Namespace(), mt.Version())
	if err != nil {
		serrs = append(serrs, serror.New(err, map[string]interface{}{
			"name":    core.JoinNamespace(mt.Namespace()),
			"version": mt.Version(),
		}))
		return nil, serrs
	}

	// No metric found return error.
	if m == nil {
		serrs = append(serrs, serror.New(fmt.Errorf("no metric found cannot subscribe: (%s) version(%d)", mt.Namespace(), mt.Version())))
		return nil, serrs
	}

	m.config = cd

	typ, serr := core.ToPluginType(m.Plugin.TypeName())
	if serr != nil {
		return nil, []serror.SnapError{serror.New(err)}
	}

	// merge global plugin config
	if m.config != nil {
		m.config.Merge(p.Config.Plugins.getPluginConfigDataNode(typ, m.Plugin.Name(), m.Plugin.Version()))
	} else {
		m.config = p.Config.Plugins.getPluginConfigDataNode(typ, m.Plugin.Name(), m.Plugin.Version())
	}

	// When a metric is added to the MetricCatalog, the policy of rules defined by the plugin is added to the metric's policy.
	// If no rules are defined for a metric, we set the metric's policy to an empty ConfigPolicyNode.
	// Checking m.policy for nil will not work, we need to check if rules are nil.
	if m.policy.HasRules() {
		if m.Config() == nil {
			serrs = append(serrs, serror.New(fmt.Errorf("Policy defined for metric, (%s) version (%d), but no config defined in manifest", mt.Namespace(), mt.Version())))
			return nil, serrs
		}
		ncdTable, errs := m.policy.Process(m.Config().Table())
		if errs != nil && errs.HasErrors() {
			for _, e := range errs.Errors() {
				serrs = append(serrs, serror.New(e))
			}
			return nil, serrs
		}
		m.config = cdata.FromTable(*ncdTable)
	}

	return m, serrs
}

func (mtt *mttNode) find(ns []string) (*mttNode, serror.SnapError) {
	node, index := mtt.walk(ns)
	if index != len(ns) {
		se := serror.New(errorMetricNotFound(ns))
		se.SetFields(map[string]interface{}{
			"name": core.JoinNamespace(ns),
		})
		return nil, se
	}
	return node, nil
}

func (t *TaskWatchHandler) CatchCollection(m []core.Metric) {
	sm := make([]rbody.StreamedMetric, len(m))
	for i, _ := range m {
		sm[i] = rbody.StreamedMetric{
			Namespace: core.JoinNamespace(m[i].Namespace()),
			Data:      m[i].Data(),
			Source:    m[i].Source(),
			Timestamp: m[i].Timestamp(),
		}
	}
	t.mChan <- rbody.StreamedTaskEvent{
		EventType: rbody.TaskWatchMetricEvent,
		Message:   "",
		Event:     sm,
	}
}

// Get works like fetch, but only returns the MT at the given node
// and does not gather the node's children.
func (mtt *mttNode) Get(ns []string) ([]*metricType, serror.SnapError) {
	node, err := mtt.find(ns)
	if err != nil {
		return nil, err
	}
	if node.mts == nil {
		se := serror.New(errorMetricNotFound(ns))
		se.SetFields(map[string]interface{}{
			"name": core.JoinNamespace(ns),
		})
		return nil, se
	}
	var mts []*metricType
	for _, mt := range node.mts {
		mts = append(mts, mt)
	}
	return mts, nil
}

func (p *pluginControl) gatherCollectors(mts []core.Metric) ([]core.Plugin, []serror.SnapError) {
	var (
		plugins []core.Plugin
		serrs   []serror.SnapError
	)

	// here we resolve and retrieve plugins for each metric type.
	// if the incoming metric type version is < 1, we treat that as
	// latest as with plugins.  The following two loops create a set
	// of plugins with proper versions needed to discern the subscription
	// types.
	colPlugins := make(map[string]*loadedPlugin)
	for _, mt := range mts {
		m, err := p.metricCatalog.Get(mt.Namespace(), mt.Version())
		if err != nil {
			serrs = append(serrs, serror.New(err, map[string]interface{}{
				"name":    core.JoinNamespace(mt.Namespace()),
				"version": mt.Version(),
			}))
			continue
		}
		// if the metric subscription is to version -1, we need to carry
		// that forward in the subscription.
		if mt.Version() < 1 {
			// make a copy of the loadedPlugin and overwrite the version.
			npl := *m.Plugin
			npl.Meta.Version = -1
			colPlugins[npl.Key()] = &npl
		} else {
			colPlugins[m.Plugin.Key()] = m.Plugin
		}
	}
	if len(serrs) > 0 {
		return plugins, serrs
	}

	for _, lp := range colPlugins {
		plugins = append(plugins, lp)
	}

	return plugins, nil
}

func (c *cache) checkCache(mts []core.Metric) (metricsToCollect []core.Metric, fromCache []core.Metric) {
	for _, mt := range mts {
		if m := c.get(core.JoinNamespace(mt.Namespace()), mt.Version()); m != nil {
			switch metric := m.(type) {
			case core.Metric:
				fromCache = append(fromCache, metric)
			case []core.Metric:
				for _, met := range metric {
					fromCache = append(fromCache, met)
				}
			default:
				cacheLog.WithFields(log.Fields{
					"_block": "checkCache",
				}).Error("unsupported type found in the cache")
			}
		} else {
			metricsToCollect = append(metricsToCollect, mt)
		}
	}
	return metricsToCollect, fromCache
}

func (mc *metricCatalog) get(ns []string, ver int) (*metricType, serror.SnapError) {
	mts, err := mc.tree.Get(ns)
	if err != nil {
		return nil, err
	}
	if mts == nil {
		return nil, serror.New(errMetricNotFound)
	}
	// a version IS given
	if ver > 0 {
		l, err := getVersion(mts, ver)
		if err != nil {
			se := serror.New(errorMetricNotFound(ns, ver))
			se.SetFields(map[string]interface{}{
				"name":    core.JoinNamespace(ns),
				"version": ver,
			})
			return nil, se
		}
		return l, nil
	}
	// ver is less than or equal to 0 get the latest
	return getLatest(mts), nil
}

// checkCache checks the cache for metric types.
// returns:
//  - array of metrics that need to be collected
//  - array of metrics that were returned from the cache
func (l *lru) CheckCache(mts []core.Metric) ([]core.Metric, []core.Metric) {
	var fromCache []core.Metric
	var metricsToCollect []core.Metric
	for _, mt := range mts {
		if m := l.metricCache.get(core.JoinNamespace(mt.Namespace()), mt.Version()); m != nil {
			switch metric := m.(type) {
			case core.Metric:
				fromCache = append(fromCache, metric)
			case []core.Metric:
				for _, met := range metric {
					fromCache = append(fromCache, met)
				}
			default:
				l.logger.WithFields(log.Fields{
					"_module": "client",
					"_block":  "checkCache",
				}).Error("unsupported type found in the cache")
			}
		} else {
			metricsToCollect = append(metricsToCollect, mt)
		}
	}
	return metricsToCollect, fromCache
}

// updateCache updates the cache with the given array of metrics.
func updateCache(mts []plugin.PluginMetricType) {
	results := []core.Metric{}
	dc := map[string][]core.Metric{}
	for _, mt := range mts {
		if mt.Labels == nil {
			// cache the individual metric
			metricCache.put(core.JoinNamespace(mt.Namespace_), mt.Version(), mt)
		} else {
			// collect the dynamic query results so we can cache
			ns := make([]string, len(mt.Namespace()))
			copy(ns, mt.Namespace())
			for _, label := range mt.Labels_ {
				ns[label.Index] = "*"
			}
			if _, ok := dc[core.JoinNamespace(ns)]; !ok {
				dc[core.JoinNamespace(ns)] = []core.Metric{}
			}
			dc[core.JoinNamespace(ns)] = append(dc[core.JoinNamespace(ns)], mt)
			metricCache.put(core.JoinNamespace(ns), mt.Version(), dc[core.JoinNamespace(ns)])
		}
		results = append(results, mt)
	}
}

func TestCollectDynamicMetrics(t *testing.T) {
	Convey("given a plugin using the native client", t, func() {
		config := NewConfig()
		config.Plugins.All.AddItem("password", ctypes.ConfigValueStr{Value: "testval"})
		c := New(OptSetConfig(config), CacheExpiration(time.Second*1))
		c.Start()
		So(strategy.GlobalCacheExpiration, ShouldResemble, time.Second*1)
		lpe := newListenToPluginEvent()
		c.eventManager.RegisterHandler("Control.PluginLoaded", lpe)
		_, e := load(c, PluginPath)
		Convey("Loading native client plugin", func() {
			Convey("Should not error", func() {
				So(e, ShouldBeNil)
			})
		})
		if e != nil {
			t.FailNow()
		}
		<-lpe.done
		_, e = load(c, JSONRPCPluginPath)
		Convey("Loading JSONRPC client plugin", func() {
			Convey("Should not error", func() {
				So(e, ShouldBeNil)
			})
		})
		if e != nil {
			t.FailNow()
		}
		<-lpe.done
		cd := cdata.NewNode()
		metrics, err := c.metricCatalog.Fetch([]string{})
		So(err, ShouldBeNil)
		So(len(metrics), ShouldEqual, 6)
		m, err := c.metricCatalog.Get([]string{"intel", "mock", "*", "baz"}, 2)
		So(err, ShouldBeNil)
		So(m, ShouldNotBeNil)
		jsonm, err := c.metricCatalog.Get([]string{"intel", "mock", "*", "baz"}, 1)
		So(err, ShouldBeNil)
		So(jsonm, ShouldNotBeNil)
		metric, errs := c.validateMetricTypeSubscription(m, cd)
		So(errs, ShouldBeNil)
		So(metric, ShouldNotBeNil)
		Convey("collects metrics from plugin using native client", func() {
			lp, err := c.pluginManager.get("collector:mock:2")
			So(err, ShouldBeNil)
			So(lp, ShouldNotBeNil)
			pool, errp := c.pluginRunner.AvailablePlugins().getOrCreatePool("collector:mock:2")
			So(errp, ShouldBeNil)
			So(pool, ShouldNotBeNil)
			ttl, err := pool.CacheTTL()
			So(err, ShouldResemble, ErrPoolEmpty)
			So(ttl, ShouldEqual, 0)
			pool.subscribe("1", unboundSubscriptionType)
			err = c.pluginRunner.runPlugin(lp.Details)
			So(err, ShouldBeNil)
			ttl, err = pool.CacheTTL()
			So(err, ShouldBeNil)
			// The minimum TTL advertised by the plugin is 100ms therefore the TTL for the
			// pool should be the global cache expiration
			So(ttl, ShouldEqual, strategy.GlobalCacheExpiration)
			mts, errs := c.CollectMetrics([]core.Metric{m}, time.Now().Add(time.Second*1))
			hits, err := pool.CacheHits(core.JoinNamespace(m.namespace), 2)
			So(err, ShouldBeNil)
			So(hits, ShouldEqual, 0)
			So(errs, ShouldBeNil)
			So(len(mts), ShouldEqual, 10)
			mts, errs = c.CollectMetrics([]core.Metric{m}, time.Now().Add(time.Second*1))
			hits, err = pool.CacheHits(core.JoinNamespace(m.namespace), 2)
			So(err, ShouldBeNil)
			So(hits, ShouldEqual, 1)
			So(errs, ShouldBeNil)
			So(len(mts), ShouldEqual, 10)
			pool.unsubscribe("1")
			Convey("collects metrics from plugin using httpjson client", func() {
				lp, err := c.pluginManager.get("collector:mock:1")
				So(err, ShouldBeNil)
				So(lp, ShouldNotBeNil)
				pool, errp := c.pluginRunner.AvailablePlugins().getOrCreatePool("collector:mock:1")
				So(errp, ShouldBeNil)
				So(pool, ShouldNotBeNil)
				ttl, err := pool.CacheTTL()
				So(err, ShouldResemble, ErrPoolEmpty)
				So(ttl, ShouldEqual, 0)
				pool.subscribe("1", unboundSubscriptionType)
				err = c.pluginRunner.runPlugin(lp.Details)
				So(err, ShouldBeNil)
				ttl, err = pool.CacheTTL()
				So(err, ShouldBeNil)
				So(ttl, ShouldEqual, 1100*time.Millisecond)
				mts, errs := c.CollectMetrics([]core.Metric{jsonm}, time.Now().Add(time.Second*1))
				hits, err := pool.CacheHits(core.JoinNamespace(jsonm.namespace), jsonm.version)
				So(pool.subscriptionCount(), ShouldEqual, 1)
				So(pool.strategy, ShouldNotBeNil)
				So(len(mts), ShouldBeGreaterThan, 0)
				So(err, ShouldBeNil)
				So(hits, ShouldEqual, 0)
				So(errs, ShouldBeNil)
				So(len(mts), ShouldEqual, 10)
				mts, errs = c.CollectMetrics([]core.Metric{jsonm}, time.Now().Add(time.Second*1))
				hits, err = pool.CacheHits(core.JoinNamespace(m.namespace), 1)
//.........这里部分代码省略.........

func (m *metricType) NamespaceAsString() string {
	return core.JoinNamespace(m.Namespace())
}

func errorMetricNotFound(ns []string, ver ...int) error {
	if len(ver) > 0 {
		return fmt.Errorf("Metric not found: %s (version: %d)", core.JoinNamespace(ns), ver[0])
	}
	return fmt.Errorf("Metric not found: %s", core.JoinNamespace(ns))
}

func (s *Server) getMetricsFromTree(w http.ResponseWriter, r *http.Request, params httprouter.Params) {
	namespace := params.ByName("namespace")

	// we land here if the request contains a trailing slash, because it matches the tree
	// lookup URL: /v1/metrics/*namespace.  If the length of the namespace param is 1, we
	// redirect the request to getMetrics.  This results in GET /v1/metrics and
	// GET /v1/metrics/ behaving the same way.
	if len(namespace) <= 1 {
		s.getMetrics(w, r, params)
		return
	}

	ns := parseNamespace(namespace)

	var (
		ver int
		err error
	)
	q := r.URL.Query()
	v := q.Get("ver")

	if ns[len(ns)-1] == "*" {
		if v == "" {
			ver = -1
		} else {
			ver, err = strconv.Atoi(v)
			if err != nil {
				respond(400, rbody.FromError(err), w)
				return
			}
		}

		mets, err := s.mm.FetchMetrics(ns[:len(ns)-1], ver)
		if err != nil {
			respond(404, rbody.FromError(err), w)
			return
		}
		respondWithMetrics(r.Host, mets, w)
		return
	}

	// If no version was given, get all that fall at this namespace.
	if v == "" {
		mts, err := s.mm.GetMetricVersions(ns)
		if err != nil {
			respond(404, rbody.FromError(err), w)
			return
		}
		respondWithMetrics(r.Host, mts, w)
		return
	}

	// if an explicit version is given, get that single one.
	ver, err = strconv.Atoi(v)
	if err != nil {
		respond(400, rbody.FromError(err), w)
		return
	}
	mt, err := s.mm.GetMetric(ns, ver)
	if err != nil {
		respond(404, rbody.FromError(err), w)
		return
	}

	b := &rbody.MetricReturned{}
	mb := &rbody.Metric{
		Namespace:               core.JoinNamespace(mt.Namespace()),
		Version:                 mt.Version(),
		LastAdvertisedTimestamp: mt.LastAdvertisedTime().Unix(),
		Href: catalogedMetricURI(r.Host, mt),
	}
	rt := mt.Policy().RulesAsTable()
	policies := make([]rbody.PolicyTable, 0, len(rt))
	for _, r := range rt {
		policies = append(policies, rbody.PolicyTable{
			Name:     r.Name,
			Type:     r.Type,
			Default:  r.Default,
			Required: r.Required,
			Minimum:  r.Minimum,
			Maximum:  r.Maximum,
		})
	}
	mb.Policy = policies
	b.Metric = mb
	respond(200, b, w)
}

func catalogedMetricURI(host string, mt core.CatalogedMetric) string {
	return fmt.Sprintf("%s://%s/v1/metrics%s?ver=%d", protocolPrefix, host, core.JoinNamespace(mt.Namespace()), mt.Version())
}