func (s *Server) getPluginConfigItem(w http.ResponseWriter, r *http.Request, p httprouter.Params) {
	var err error
	styp := p.ByName("type")
	if styp == "" {
		cdn := s.mc.GetPluginConfigDataNodeAll()
		item := &rbody.PluginConfigItem{ConfigDataNode: cdn}
		respond(200, item, w)
		return
	}

	typ, err := getPluginType(styp)
	if err != nil {
		respond(400, rbody.FromError(err), w)
		return
	}

	name := p.ByName("name")
	sver := p.ByName("version")
	var iver int
	if sver != "" {
		if iver, err = strconv.Atoi(sver); err != nil {
			respond(400, rbody.FromError(err), w)
			return
		}
	} else {
		iver = -2
	}

	cdn := s.mc.GetPluginConfigDataNode(typ, name, iver)
	item := &rbody.PluginConfigItem{ConfigDataNode: cdn}
	respond(200, item, 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)
}

func (s *Server) removeTask(w http.ResponseWriter, r *http.Request, p httprouter.Params) {
	id := p.ByName("id")
	err := s.mt.RemoveTask(id)
	if err != nil {
		if strings.Contains(err.Error(), ErrTaskNotFound.Error()) {
			respond(404, rbody.FromError(err), w)
			return
		}
		respond(500, rbody.FromError(err), w)
		return
	}
	respond(200, &rbody.ScheduledTaskRemoved{ID: id}, w)
}

//enableTask changes the task state from Disabled to Stopped
func (s *Server) enableTask(w http.ResponseWriter, r *http.Request, p httprouter.Params) {
	id := p.ByName("id")
	tsk, err := s.mt.EnableTask(id)
	if err != nil {
		if strings.Contains(err.Error(), ErrTaskNotFound.Error()) {
			respond(404, rbody.FromError(err), w)
			return
		}
		respond(500, rbody.FromError(err), w)
		return
	}
	task := &rbody.ScheduledTaskEnabled{}
	task.AddScheduledTask = *rbody.AddSchedulerTaskFromTask(tsk)
	respond(200, task, w)
}

func (s *Server) getMetrics(w http.ResponseWriter, r *http.Request, _ httprouter.Params) {
	mets, err := s.mm.MetricCatalog()
	if err != nil {
		respond(500, rbody.FromError(err), w)
		return
	}
	respondWithMetrics(r.Host, mets, w)
}

func (s *Server) addAgreement(w http.ResponseWriter, r *http.Request, p httprouter.Params) {
	tribeLogger = tribeLogger.WithField("_block", "addAgreement")
	b, err := ioutil.ReadAll(r.Body)
	if err != nil {
		tribeLogger.Error(err)
		respond(500, rbody.FromError(err), w)
		return
	}

	a := struct{ Name string }{}
	err = json.Unmarshal(b, &a)
	if err != nil {
		fields := map[string]interface{}{
			"error": err,
			"hint":  `The body of the request should be of the form '{"name": "agreement_name"}'`,
		}
		se := serror.New(ErrInvalidJSON, fields)
		tribeLogger.WithFields(fields).Error(ErrInvalidJSON)
		respond(400, rbody.FromSnapError(se), w)
		return
	}

	if a.Name == "" {
		fields := map[string]interface{}{
			"hint": `The body of the request should be of the form '{"name": "agreement_name"}'`,
		}
		se := serror.New(ErrInvalidJSON, fields)
		tribeLogger.WithFields(fields).Error(ErrInvalidJSON)
		respond(400, rbody.FromSnapError(se), w)
		return
	}

	err = s.tr.AddAgreement(a.Name)
	if err != nil {
		tribeLogger.WithField("agreement-name", a.Name).Error(err)
		respond(400, rbody.FromError(err), w)
		return
	}

	res := &rbody.TribeAddAgreement{}
	res.Agreements = s.tr.GetAgreements()

	respond(200, res, w)
}

func (s *Server) addTask(w http.ResponseWriter, r *http.Request, _ httprouter.Params) {

	tr, err := marshalTask(r.Body)
	if err != nil {
		respond(500, rbody.FromError(err), w)
		return
	}

	sch, err := makeSchedule(tr.Schedule)
	if err != nil {
		respond(500, rbody.FromError(err), w)
		return
	}

	var opts []core.TaskOption
	if tr.Deadline != "" {
		dl, err := time.ParseDuration(tr.Deadline)
		if err != nil {
			respond(500, rbody.FromError(err), w)
			return
		}
		opts = append(opts, core.TaskDeadlineDuration(dl))
	}

	if tr.Name != "" {
		opts = append(opts, core.SetTaskName(tr.Name))
	}
	opts = append(opts, core.OptionStopOnFailure(10))

	task, errs := s.mt.CreateTask(sch, tr.Workflow, tr.Start, opts...)
	if errs != nil && len(errs.Errors()) != 0 {
		var errMsg string
		for _, e := range errs.Errors() {
			errMsg = errMsg + e.Error() + " -- "
		}
		respond(500, rbody.FromError(errors.New(errMsg[:len(errMsg)-4])), w)
		return
	}

	taskB := rbody.AddSchedulerTaskFromTask(task)
	taskB.Href = taskURI(r.Host, task)
	respond(201, taskB, w)
}

func (s *Server) addTask(w http.ResponseWriter, r *http.Request, _ httprouter.Params) {
	task, err := core.CreateTaskFromContent(r.Body, nil, s.mt.CreateTask)
	if err != nil {
		respond(500, rbody.FromError(err), w)
		return
	}
	taskB := rbody.AddSchedulerTaskFromTask(task)
	taskB.Href = taskURI(r.Host, task)
	respond(201, taskB, w)
}

func (s *Server) deletePluginConfigItem(w http.ResponseWriter, r *http.Request, p httprouter.Params) {
	var err error
	var typ core.PluginType
	styp := p.ByName("type")
	if styp != "" {
		typ, err = getPluginType(styp)
		if err != nil {
			respond(400, rbody.FromError(err), w)
			return
		}
	}

	name := p.ByName("name")
	sver := p.ByName("version")
	var iver int
	if sver != "" {
		if iver, err = strconv.Atoi(sver); err != nil {
			respond(400, rbody.FromError(err), w)
			return
		}
	} else {
		iver = -2
	}

	src := []string{}
	errCode, err := core.UnmarshalBody(&src, r.Body)
	if errCode != 0 && err != nil {
		respond(400, rbody.FromError(err), w)
		return
	}

	var res cdata.ConfigDataNode
	if styp == "" {
		res = s.mc.DeletePluginConfigDataNodeFieldAll(src...)
	} else {
		res = s.mc.DeletePluginConfigDataNodeField(typ, name, iver, src...)
	}

	item := &rbody.DeletePluginConfigItem{ConfigDataNode: res}
	respond(200, item, w)
}

func (s *Server) getMetrics(w http.ResponseWriter, r *http.Request, _ httprouter.Params) {
	ver := 0 // 0: get all metrics

	// If we are provided a parameter with the name 'ns' we need to
	// perform a query
	q := r.URL.Query()
	v := q.Get("ver")
	ns_query := q.Get("ns")
	if ns_query != "" {
		ver = 0 // 0: get all versions
		if v != "" {
			var err error
			ver, err = strconv.Atoi(v)
			if err != nil {
				respond(400, rbody.FromError(err), w)
				return
			}
		}
		// strip the leading char and split on the remaining.
		fc := stringutils.GetFirstChar(ns_query)
		ns := strings.Split(strings.TrimLeft(ns_query, fc), fc)
		if ns[len(ns)-1] == "*" {
			ns = ns[:len(ns)-1]
		}

		mets, err := s.mm.FetchMetrics(core.NewNamespace(ns...), ver)
		if err != nil {
			respond(404, rbody.FromError(err), w)
			return
		}
		respondWithMetrics(r.Host, mets, w)
		return
	}

	mets, err := s.mm.MetricCatalog()
	if err != nil {
		respond(500, rbody.FromError(err), w)
		return
	}
	respondWithMetrics(r.Host, mets, w)
}

func (s *Server) getTask(w http.ResponseWriter, r *http.Request, p httprouter.Params) {
	id := p.ByName("id")
	t, err1 := s.mt.GetTask(id)
	if err1 != nil {
		respond(404, rbody.FromError(err1), w)
		return
	}
	task := &rbody.ScheduledTaskReturned{}
	task.AddScheduledTask = *rbody.AddSchedulerTaskFromTask(t)
	task.Href = taskURI(r.Host, t)
	respond(200, task, w)
}

func (s *Server) joinAgreement(w http.ResponseWriter, r *http.Request, p httprouter.Params) {
	tribeLogger = tribeLogger.WithField("_block", "joinAgreement")
	name := p.ByName("name")
	if _, ok := s.tr.GetAgreements()[name]; !ok {
		fields := map[string]interface{}{
			"agreement_name": name,
		}
		tribeLogger.WithFields(fields).Error(ErrAgreementDoesNotExist)
		respond(400, rbody.FromSnapError(serror.New(ErrAgreementDoesNotExist, fields)), w)
		return
	}

	b, err := ioutil.ReadAll(r.Body)
	if err != nil {
		tribeLogger.Error(err)
		respond(500, rbody.FromError(err), w)
		return
	}

	m := struct {
		MemberName string `json:"member_name"`
	}{}
	err = json.Unmarshal(b, &m)
	if err != nil {
		fields := map[string]interface{}{
			"error": err,
			"hint":  `The body of the request should be of the form '{"member_name": "some_value"}'`,
		}
		se := serror.New(ErrInvalidJSON, fields)
		tribeLogger.WithFields(fields).Error(ErrInvalidJSON)
		respond(400, rbody.FromSnapError(se), w)
		return
	}

	serr := s.tr.JoinAgreement(name, m.MemberName)
	if serr != nil {
		tribeLogger.Error(serr)
		respond(400, rbody.FromSnapError(serr), w)
		return
	}
	agreement, _ := s.tr.GetAgreement(name)
	respond(200, &rbody.TribeJoinAgreement{Agreement: agreement}, w)

}

func (s *Server) watchTask(w http.ResponseWriter, r *http.Request, p httprouter.Params) {
	s.wg.Add(1)
	defer s.wg.Done()
	logger := log.WithFields(log.Fields{
		"_module": "api",
		"_block":  "watch-task",
		"client":  r.RemoteAddr,
	})

	id := p.ByName("id")

	logger.WithFields(log.Fields{
		"task-id": id,
	}).Debug("request to watch task")
	tw := &TaskWatchHandler{
		alive: true,
		mChan: make(chan rbody.StreamedTaskEvent),
	}
	tc, err1 := s.mt.WatchTask(id, tw)
	if err1 != nil {
		if strings.Contains(err1.Error(), ErrTaskNotFound.Error()) {
			respond(404, rbody.FromError(err1), w)
			return
		}
		respond(500, rbody.FromError(err1), w)
		return
	}

	// Make this Server Sent Events compatible
	w.Header().Set("Content-Type", "text/event-stream")
	w.Header().Set("Cache-Control", "no-cache")
	w.Header().Set("Connection", "keep-alive")
	w.Header().Set("Access-Control-Allow-Origin", "*")

	// get a flusher type
	flusher, ok := w.(http.Flusher)
	if !ok {
		// This only works on ResponseWriters that support streaming
		respond(500, rbody.FromError(ErrStreamingUnsupported), w)
		return
	}
	// send initial stream open event
	so := rbody.StreamedTaskEvent{
		EventType: rbody.TaskWatchStreamOpen,
		Message:   "Stream opened",
	}
	fmt.Fprintf(w, "data: %s\n\n", so.ToJSON())
	flusher.Flush()

	// Get a channel for if the client notifies us it is closing the connection
	n := w.(http.CloseNotifier).CloseNotify()
	t := time.Now()
	for {
		// Write to the ResponseWriter
		select {
		case e := <-tw.mChan:
			logger.WithFields(log.Fields{
				"task-id":            id,
				"task-watcher-event": e.EventType,
			}).Debug("new event")
			switch e.EventType {
			case rbody.TaskWatchMetricEvent, rbody.TaskWatchTaskStarted:
				// The client can decide to stop receiving on the stream on Task Stopped.
				// We write the event to the buffer
				fmt.Fprintf(w, "data: %s\n\n", e.ToJSON())
			case rbody.TaskWatchTaskDisabled, rbody.TaskWatchTaskStopped:
				// A disabled task should end the streaming and close the connection
				fmt.Fprintf(w, "data: %s\n\n", e.ToJSON())
				// Flush since we are sending nothing new
				flusher.Flush()
				// Close out watcher removing it from the scheduler
				tc.Close()
				// exit since this client is no longer listening
				respond(200, &rbody.ScheduledTaskWatchingEnded{}, w)
			}
			// If we are at least above our minimum buffer time we flush to send
			if time.Now().Sub(t).Seconds() > StreamingBufferWindow {
				flusher.Flush()
				t = time.Now()
			}
		case <-n:
			logger.WithFields(log.Fields{
				"task-id": id,
			}).Debug("client disconnecting")
			// Flush since we are sending nothing new
			flusher.Flush()
			// Close out watcher removing it from the scheduler
			tc.Close()
			// exit since this client is no longer listening
			respond(200, &rbody.ScheduledTaskWatchingEnded{}, w)
			return
		case <-s.killChan:
			logger.WithFields(log.Fields{
				"task-id": id,
			}).Debug("snapd exiting; disconnecting client")
			// Flush since we are sending nothing new
			flusher.Flush()
			// Close out watcher removing it from the scheduler
			tc.Close()
			// exit since this client is no longer listening
//.........这里部分代码省略.........

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 (s *Server) loadPlugin(w http.ResponseWriter, r *http.Request, _ httprouter.Params) {
	mediaType, params, err := mime.ParseMediaType(r.Header.Get("Content-Type"))
	if err != nil {
		respond(500, rbody.FromError(err), w)
		return
	}
	if strings.HasPrefix(mediaType, "multipart/") {
		var pluginPath string
		var signature []byte
		var checkSum [sha256.Size]byte
		lp := &rbody.PluginsLoaded{}
		lp.LoadedPlugins = make([]rbody.LoadedPlugin, 0)
		mr := multipart.NewReader(r.Body, params["boundary"])
		var i int
		for {
			var b []byte
			p, err := mr.NextPart()
			if err == io.EOF {
				break
			}
			if err != nil {
				respond(500, rbody.FromError(err), w)
				return
			}
			if r.Header.Get("Plugin-Compression") == "gzip" {
				g, err := gzip.NewReader(p)
				defer g.Close()
				if err != nil {
					respond(500, rbody.FromError(err), w)
					return
				}
				b, err = ioutil.ReadAll(g)
				if err != nil {
					respond(500, rbody.FromError(err), w)
					return
				}
			} else {
				b, err = ioutil.ReadAll(p)
				if err != nil {
					respond(500, rbody.FromError(err), w)
					return
				}
			}

			// A little sanity checking for files being passed into the API server.
			// First file passed in should be the plugin. If the first file is a signature
			// file, an error is returned. The signature file should be the second
			// file passed to the API server. If the second file does not have the ".asc"
			// extension, an error is returned.
			// If we loop around more than twice before receiving io.EOF, then
			// an error is returned.

			switch {
			case i == 0:
				if filepath.Ext(p.FileName()) == ".asc" {
					e := errors.New("Error: first file passed to load plugin api can not be signature file")
					respond(500, rbody.FromError(e), w)
					return
				}
				if pluginPath, err = writeFile(p.FileName(), b); err != nil {
					respond(500, rbody.FromError(err), w)
					return
				}
				checkSum = sha256.Sum256(b)
			case i == 1:
				if filepath.Ext(p.FileName()) == ".asc" {
					signature = b
				} else {
					e := errors.New("Error: second file passed was not a signature file")
					respond(500, rbody.FromError(e), w)
					return
				}
			case i == 2:
				e := errors.New("Error: More than two files passed to the load plugin api")
				respond(500, rbody.FromError(e), w)
				return
			}
			i++
		}
		rp, err := core.NewRequestedPlugin(pluginPath)
		if err != nil {
			respond(500, rbody.FromError(err), w)
			return
		}
		// Sanity check, verify the checkSum on the file sent is the same
		// as after it is written to disk.
		if rp.CheckSum() != checkSum {
			e := errors.New("Error: CheckSum mismatch on requested plugin to load")
			respond(500, rbody.FromError(e), w)
			return
		}
		rp.SetSignature(signature)
		restLogger.Info("Loading plugin: ", rp.Path())
		pl, err := s.mm.Load(rp)
		if err != nil {
			var ec int
			restLogger.Error(err)
			restLogger.Debugf("Removing file (%s)", rp.Path())
			err2 := os.RemoveAll(filepath.Dir(rp.Path()))
			if err2 != nil {
//.........这里部分代码省略.........