func (t *tribe) decodeTags(buf []byte) map[string]string {
	tags := make(map[string]string)
	r := bytes.NewReader(buf)
	dec := codec.NewDecoder(r, &codec.MsgpackHandle{})
	if err := dec.Decode(&tags); err != nil {
		t.logger.WithFields(log.Fields{
			"_block": "decode-tags",
			"error":  err,
		}).Error("Failed to decode tags")
	}
	return tags
}
decodeTags is used to decode a tag map
package tribe
import (
	"bytes"
	"errors"
	"fmt"
	"math"
	"net"
	"strconv"
	"sync"
	"time"
	"github.com/intelsdi-x/gomit"
	"github.com/intelsdi-x/snap/core"
	"github.com/intelsdi-x/snap/core/control_event"
	"github.com/intelsdi-x/snap/core/scheduler_event"
	"github.com/intelsdi-x/snap/core/serror"
	"github.com/intelsdi-x/snap/core/tribe_event"
	"github.com/intelsdi-x/snap/mgmt/tribe/agreement"
	"github.com/intelsdi-x/snap/mgmt/tribe/worker"
	"github.com/pborman/uuid"
	log "github.com/sirupsen/logrus"
	"github.com/hashicorp/go-msgpack/codec"
	"github.com/hashicorp/memberlist"
)
const (
	HandlerRegistrationName = "tribe"
)
var (
	errAgreementDoesNotExist          = errors.New("Agreement does not exist")
	errAgreementAlreadyExists         = errors.New("Agreement already exists")
	errUnknownMember                  = errors.New("Unknown member")
	errAlreadyMemberOfPluginAgreement = errors.New("Already a member of a plugin agreement")
	errNotAMember                     = errors.New("Not a member of agreement")
	errTaskAlreadyExists              = errors.New("Task already exists")
	errTaskDoesNotExist               = errors.New("Task does not exist")
	errCreateMemberlist               = errors.New("Failed to start tribe")
	errMemberlistJoin                 = errors.New("Failed to join tribe")
	errPluginCatalogNotSet            = errors.New("Plugin Catalog not set")
	errTaskManagerNotSet              = errors.New("Task Manager not set")
)
var logger = log.WithFields(log.Fields{
	"_module": "tribe",
})
type tribe struct {
	clock              LClock
	agreements         map[string]*agreement.Agreement
	mutex              sync.RWMutex
	msgBuffer          []msg
	intentBuffer       []msg
	broadcasts         *memberlist.TransmitLimitedQueue
	memberlist         *memberlist.Memberlist
	logger             *log.Entry
	taskStartStopCache *cache
	taskStateResponses map[string]*taskStateQueryResponse
	members            map[string]*agreement.Member
	tags               map[string]string
	EventManager       *gomit.EventController
	config             *Config
	pluginCatalog   worker.ManagesPlugins
	taskManager     worker.ManagesTasks
	pluginWorkQueue chan worker.PluginRequest
	taskWorkQueue   chan worker.TaskRequest
	workerQuitChan  chan struct{}
	workerWaitGroup *sync.WaitGroup
}
func New(cfg *Config) (*tribe, error) {
	cfg.MemberlistConfig.Name = cfg.Name
	cfg.MemberlistConfig.BindAddr = cfg.BindAddr
	cfg.MemberlistConfig.BindPort = cfg.BindPort
	logger := logger.WithFields(log.Fields{
		"port": cfg.MemberlistConfig.BindPort,
		"addr": cfg.MemberlistConfig.BindAddr,
		"name": cfg.MemberlistConfig.Name,
	})
	tribe := &tribe{
		agreements:         map[string]*agreement.Agreement{},
		members:            map[string]*agreement.Member{},
		taskStateResponses: map[string]*taskStateQueryResponse{},
		taskStartStopCache: newCache(),
		msgBuffer:          make([]msg, 512),
		intentBuffer:       []msg{},
		logger:             logger.WithField("_name", cfg.MemberlistConfig.Name),
		tags: map[string]string{
			agreement.RestPort:               strconv.Itoa(cfg.RestAPIPort),
			agreement.RestProtocol:           cfg.RestAPIProto,
			agreement.RestInsecureSkipVerify: cfg.RestAPIInsecureSkipVerify,
		},
		pluginWorkQueue: make(chan worker.PluginRequest, 999),
		taskWorkQueue:   make(chan worker.TaskRequest, 999),
		workerQuitChan:  make(chan struct{}),
		workerWaitGroup: &sync.WaitGroup{},
		config:          cfg,
		EventManager:    gomit.NewEventController(),
	}
	tribe.broadcasts = &memberlist.TransmitLimitedQueue{
		NumNodes: func() int {
			return len(tribe.memberlist.Members())
		},
		RetransmitMult: memberlist.DefaultLANConfig().RetransmitMult,
	}
	cfg.MemberlistConfig.Delegate = &delegate{tribe: tribe}
	cfg.MemberlistConfig.Events = &memberDelegate{tribe: tribe}
	ml, err := memberlist.Create(cfg.MemberlistConfig)
	if err != nil {
		logger.Error(err)
		return nil, err
	}
	tribe.memberlist = ml
	if cfg.Seed != "" {
		_, err := ml.Join([]string{cfg.Seed})
		if err != nil {
			logger.WithFields(log.Fields{
				"seed": cfg.Seed,
			}).Error(errMemberlistJoin)
			return nil, errMemberlistJoin
		}
		logger.WithFields(log.Fields{
			"seed": cfg.Seed,
		}).Infoln("tribe started")
		return tribe, nil
	}
	logger.WithFields(log.Fields{
		"seed": "none",
	}).Infoln("tribe started")
	return tribe, nil
}
type cache struct {
	sync.RWMutex
	table map[string]time.Time
}
func newCache() *cache {
	return &cache{
		table: make(map[string]time.Time),
	}
}
func (c *cache) get(m msg) (time.Time, bool) {
	c.RLock()
	defer c.RUnlock()
	key := fmt.Sprintf("%v:%v", m.GetType(), m.ID())
	v, ok := c.table[key]
	return v, ok
}
func (c *cache) put(m msg, duration time.Duration) bool {
	c.Lock()
	defer c.Unlock()
	key := fmt.Sprintf("%v:%v", m.GetType(), m.ID())
	if ti, ok := c.table[key]; ok {
		logger.WithFields(log.Fields{
			"task-id":      m.ID(),
			"time-started": ti,
			"_block":       "task-cache-put",
			"key":          key,
			"cache-size":   len(c.table),
			"ltime":        m.Time(),
		}).Debugln("task cache entry exists")
		return false
	}
	c.table[key] = time.Now()
	time.AfterFunc(duration, func() {
		c.Lock()
		delete(c.table, key)
		c.Unlock()
	})
	return true
}
func (t *tribe) SetPluginCatalog(p worker.ManagesPlugins) {
	t.pluginCatalog = p
}
func (t *tribe) SetTaskManager(m worker.ManagesTasks) {
	t.taskManager = m
}
func (t *tribe) Name() string {
	return "tribe"
}
func (t *tribe) Start() error {
	if t.pluginCatalog == nil {
		return errPluginCatalogNotSet
	}
	if t.taskManager == nil {
		return errTaskManagerNotSet
	}
	worker.DispatchWorkers(
		4,
		t.pluginWorkQueue,
		t.taskWorkQueue,
		t.workerQuitChan,
		t.workerWaitGroup,
		t.pluginCatalog,
		t.taskManager,
		t)
	return nil
}
func (t *tribe) Stop() {
	logger := t.logger.WithFields(log.Fields{
		"_block": "stop",
	})
	err := t.memberlist.Leave(1 * time.Second)
	if err != nil {
		logger.Error(err)
	}
	err = t.memberlist.Shutdown()
	if err != nil {
		logger.Error(err)
	}
	close(t.workerQuitChan)
	t.workerWaitGroup.Wait()
}
func (t *tribe) GetTaskAgreementMembers() ([]worker.Member, error) {
	m, ok := t.members[t.memberlist.LocalNode().Name]
	if !ok || m.TaskAgreements == nil {
		return nil, errNotAMember
	}
	mm := map[*agreement.Member]struct{}{}
	for name := range m.TaskAgreements {
		for _, mem := range t.agreements[name].Members {
			mm[mem] = struct{}{}
		}
	}
	members := make([]worker.Member, 0, len(mm))
	for k := range mm {
		members = append(members, k)
	}
	return members, nil
}
func (t *tribe) GetPluginAgreementMembers() ([]worker.Member, error) {
	m, ok := t.members[t.memberlist.LocalNode().Name]
	if !ok || m.PluginAgreement == nil {
		return nil, errNotAMember
	}
	members := make([]worker.Member, 0, len(t.agreements[m.PluginAgreement.Name].Members))
	for _, v := range t.agreements[m.PluginAgreement.Name].Members {
		members = append(members, v)
	}
	return members, nil
}
func (t *tribe) encodeTags(tags map[string]string) []byte {
	var buf bytes.Buffer
	enc := codec.NewEncoder(&buf, &codec.MsgpackHandle{})
	if err := enc.Encode(tags); err != nil {
		panic(fmt.Sprintf("Failed to encode tags: %v", err))
	}
	return buf.Bytes()
}