func (pc *KafkaPartitionConsumer) collectorFunc(messages *[]*MessageAndMetadata) func(topic string, partition int32, offset int64, key []byte, value []byte) error {
	return func(topic string, partition int32, offset int64, key []byte, value []byte) error {
		decodedKey, err := pc.config.KeyDecoder.Decode(key)
		if err != nil {
			log.Warning(err.Error())
			return err
		}
		decodedValue, err := pc.config.ValueDecoder.Decode(value)
		if err != nil {
			log.Warning(err.Error())
			return err
		}

		*messages = append(*messages, &MessageAndMetadata{
			Key:          key,
			Value:        value,
			Topic:        topic,
			Partition:    partition,
			Offset:       offset,
			DecodedKey:   decodedKey,
			DecodedValue: decodedValue,
		})
		return nil
	}
}

// 处理 tls 请求
func (s *TlsServer) HandlerTls(conn net.Conn) {
	c, err := vhost.TLS(conn)

	if err != nil || c.Host() != s.httpsDomain {
		// 不匹配，直接转发
		defer c.Close()
		c.Free()

		remoteConn, err := net.Dial("tcp", s.forwardAddr)
		if err != nil {
			glog.Warning(fmt.Printf("[ERR] dial(\"tcp\",%v):%v", s.forwardAddr, err))
			return
		}
		defer remoteConn.Close()

		go io.Copy(c, remoteConn)
		io.Copy(remoteConn, c)
	} else {
		c.Free()
		tlsConn := tls.Server(c, s.tlsConfig)

		err := tlsConn.Handshake()
		if err != nil {
			glog.Warning(err)
			return
		}

		s.http.HandlerHttp(tlsConn)
	}
}

func validateVolumeMounts(mounts []VolumeMount, volumes util.StringSet) errorList {
	allErrs := errorList{}

	for i := range mounts {
		mnt := &mounts[i] // so we can set default values
		if len(mnt.Name) == 0 {
			allErrs.Append(makeInvalidError("VolumeMount.Name", mnt.Name))
		} else if !volumes.Has(mnt.Name) {
			allErrs.Append(makeNotFoundError("VolumeMount.Name", mnt.Name))
		}
		if len(mnt.MountPath) == 0 {
			// Backwards compat.
			if len(mnt.Path) == 0 {
				allErrs.Append(makeInvalidError("VolumeMount.MountPath", mnt.MountPath))
			} else {
				glog.Warning("DEPRECATED: VolumeMount.Path has been replaced by VolumeMount.MountPath")
				mnt.MountPath = mnt.Path
				mnt.Path = ""
			}
		}
		if len(mnt.MountType) != 0 {
			glog.Warning("DEPRECATED: VolumeMount.MountType will be removed. The Volume struct will handle types")
		}
	}
	return allErrs
}

func (storage *Storage) ReadMessage(file *os.File) *Message {
	//校验消息起始位置的magic
	var magic int32
	err := binary.Read(file, binary.BigEndian, &magic)
	if err != nil {
		log.Info("read file err:", err)
		return nil
	}

	if magic != MAGIC {
		log.Warning("magic err:", magic)
		return nil
	}
	msg := ReceiveMessage(file)
	if msg == nil {
		return msg
	}

	err = binary.Read(file, binary.BigEndian, &magic)
	if err != nil {
		log.Info("read file err:", err)
		return nil
	}

	if magic != MAGIC {
		log.Warning("magic err:", magic)
		return nil
	}
	return msg
}

// InstanceID returns the cloud provider ID of the specified instance.
func (i *Instances) InstanceID(name string) (string, error) {
	// Create context
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// Create vSphere client
	c, err := vsphereLogin(i.cfg, ctx)
	if err != nil {
		return "", err
	}
	defer c.Logout(ctx)

	vm, err := getVirtualMachineByName(i.cfg, ctx, c, name)

	var mvm mo.VirtualMachine
	err = getVirtualMachineManagedObjectReference(ctx, c, vm, "summary", &mvm)
	if err != nil {
		return "", err
	}

	if mvm.Summary.Runtime.PowerState == ActivePowerState {
		return "/" + vm.InventoryPath, nil
	}

	if mvm.Summary.Config.Template == false {
		glog.Warning("VM %s, is not in %s state", name, ActivePowerState)
	} else {
		glog.Warning("VM %s, is a template", name)
	}

	return "", cloudprovider.InstanceNotFound
}

func (pc *KafkaPartitionConsumer) initOffset() bool {
	log.Infof("Initializing offset for topic %s, partition %d", pc.topic, pc.partition)
	for {
		offset, err := pc.client.GetOffset(pc.config.Group, pc.topic, pc.partition)
		if err != nil {
			if err == siesta.ErrUnknownTopicOrPartition {
				return pc.resetOffset()
			}
			log.Warning("Cannot get offset for group %s, topic %s, partition %d: %s\n", pc.config.Group, pc.topic, pc.partition, err)
			select {
			case <-pc.stop:
				{
					log.Warning("PartitionConsumer told to stop trying to get offset, returning")
					return false
				}
			default:
			}
		} else {
			validOffset := offset + 1
			log.Infof("Initialized offset to %d", validOffset)
			atomic.StoreInt64(&pc.offset, validOffset)
			atomic.StoreInt64(&pc.highwaterMarkOffset, validOffset)
			return true
		}
		time.Sleep(pc.config.InitOffsetBackoff)
	}
}

func (client *Client) HandleGroupIMMessage(msg *IMMessage, seq int) {
	if client.uid == 0 {
		log.Warning("client has't been authenticated")
		return
	}

	msg.timestamp = int32(time.Now().Unix())
	m := &Message{cmd: MSG_GROUP_IM, body: msg}

	group := group_manager.FindGroup(msg.receiver)
	if group == nil {
		log.Warning("can't find group:", msg.receiver)
		return
	}
	members := group.Members()
	for member := range members {
		//群消息不再发送给自己
		if member == client.uid {
			continue
		}
		msgid, err := SaveMessage(client.appid, member, m)
		if err != nil {
			return
		}

		emsg := &EMessage{msgid: msgid, msg: m}
		SendEMessage(client.appid, member, emsg)
	}

	client.wt <- &Message{cmd: MSG_ACK, body: &MessageACK{int32(seq)}}
	atomic.AddInt64(&server_summary.in_message_count, 1)
	log.Infof("group message sender:%d group id:%d", msg.sender, msg.receiver)
}

func (storage *Storage) SaveSyncMessage(emsg *EMessage) error {
	storage.mutex.Lock()
	defer storage.mutex.Unlock()

	filesize, err := storage.file.Seek(0, os.SEEK_END)
	if err != nil {
		log.Fatalln(err)
	}
	if emsg.msgid != filesize {
		log.Warningf("file size:%d, msgid:%d is't equal", filesize, emsg.msgid)
		if emsg.msgid < filesize {
			log.Warning("skip msg:", emsg.msgid)
		} else {
			log.Warning("write padding:", emsg.msgid-filesize)
			padding := make([]byte, emsg.msgid-filesize)
			_, err = storage.file.Write(padding)
			if err != nil {
				log.Fatal("file write:", err)
			}
		}
	}

	storage.WriteMessage(storage.file, emsg.msg)
	storage.ExecMessage(emsg.msg, emsg.msgid)
	log.Info("save sync message:", emsg.msgid)
	return nil
}

func Warning(ctx context.Context, args ...interface{}) {
	if ctx == nil || !hasTraceKey(ctx) {
		glog.Warning(args)
		return
	}
	glog.Warning(prependParam(args, ctx)...)
}

func (self *Router) handleMsgServerClient(msc *link.Session) {
	msc.ReadLoop(func(msg link.InBuffer) {
		glog.Info("msg_server", msc.Conn().RemoteAddr().String(), " say: ", string(msg.Get()))
		var c protocol.CmdInternal
		pp := NewProtoProc(self)
		err := json.Unmarshal(msg.Get(), &c)
		if err != nil {
			glog.Error("error:", err)
		}
		switch c.GetCmdName() {
		case protocol.SEND_MESSAGE_P2P_CMD:
			err := pp.procSendMsgP2P(c, msc)
			if err != nil {
				glog.Warning(err.Error())
			}
		case protocol.CREATE_TOPIC_CMD:
			err := pp.procCreateTopic(c, msc)
			if err != nil {
				glog.Warning(err.Error())
			}
		case protocol.JOIN_TOPIC_CMD:
			err := pp.procJoinTopic(c, msc)
			if err != nil {
				glog.Warning(err.Error())
			}
		case protocol.SEND_MESSAGE_TOPIC_CMD:
			err := pp.procSendMsgTopic(c, msc)
			if err != nil {
				glog.Warning(err.Error())
			}

		}
	})
}

func (pk *Peekachu) applyMetricFilters(
	client *Client,
	tableName string,
	row RowMap,
) RowMap {
	for _, filterName := range Filters.FilterNames() {
		if _, ok := pk.config.Influxdb.MetricFilters[filterName]; ok {
			filterer, err := Filters.GetFilter(filterName, client, pk)

			if err != nil {
				msg := "Error retrieving %s filter: %s\n"
				glog.Errorf(msg, filterName, err)
				glog.Warning("Filter will not be applied!")
				break
			}
			filteredRow, err := filterer.Filter(tableName, row)

			if err != nil {
				glog.Errorf("Error applying filter: %s\n", err)
				glog.Warning("Filter will not be applyed!")
				break
			} else {
				row = filteredRow
			}

			if row == nil {
				return nil
			}

		}
	}
	return row
}

func (client *Client) HandleGroupIMMessage(msg *IMMessage, seq int) {
	if client.uid == 0 {
		log.Warning("client has't been authenticated")
		return
	}

	msg.timestamp = int32(time.Now().Unix())
	m := &Message{cmd: MSG_GROUP_IM, version: DEFAULT_VERSION, body: msg}

	group := group_manager.FindGroup(msg.receiver)
	if group == nil {
		log.Warning("can't find group:", msg.receiver)
		return
	}
	if group.super {
		_, err := SaveGroupMessage(client.appid, msg.receiver, client.device_ID, m)
		if err != nil {
			return
		}
	} else {
		members := group.Members()
		for member := range members {
			_, err := SaveMessage(client.appid, member, client.device_ID, m)
			if err != nil {
				continue
			}
		}
	}

	client.wt <- &Message{cmd: MSG_ACK, body: &MessageACK{int32(seq)}}
	atomic.AddInt64(&server_summary.in_message_count, 1)
	log.Infof("group message sender:%d group id:%d", msg.sender, msg.receiver)
}

func connect(addrs []string, tries int, hint int) (net.Conn, int, error) {
	var conn net.Conn
	var err error

	// first, try on to connect to the most likely leader
	glog.Info("Trying to connect to ", addrs[hint])
	conn, err = net.Dial("tcp", addrs[hint])
	// if successful
	if err == nil {
		glog.Infof("Connect established to %s", addrs[hint])
		return conn, hint, err
	}
	//if unsuccessful
	glog.Warning(err)

	// if fails, try everyone else
	for i := range addrs {
		for t := tries; t > 0; t-- {
			glog.Info("Trying to connect to ", addrs[i])
			conn, err = net.Dial("tcp", addrs[i])

			// if successful
			if err == nil {
				glog.Infof("Connect established to %s", addrs[i])
				return conn, i, err
			}

			//if unsuccessful
			glog.Warning(err)
		}
	}

	return conn, hint + 1, err
}

func AllMigrateTasks() ([]*MigrateMeta, error) {
	appname := meta.appName
	zconn := meta.zconn
	taskspath := "/r3/app/" + appname + "/migrate"
	exists, _, err := zconn.Exists(taskspath)
	if err != nil {
		return nil, err
	}
	if exists {
		tasks, _, err := zconn.Children(taskspath)
		if err != nil {
			return nil, err
		}
		var alltasks []*MigrateMeta
		for _, t := range tasks {
			glog.Info("Get task from zk: ", t)
			task, _, err := zconn.Get(taskspath + "/" + t)
			if err != nil {
				glog.Warning("Get task failed ", err)
				return nil, err
			}
			var migrateMeta MigrateMeta
			err = json.Unmarshal(task, &migrateMeta)
			if err != nil {
				glog.Warning("Cluster", "Unmarshal failed: ", err)
				return nil, err
			}
			alltasks = append(alltasks, &migrateMeta)
		}
		return alltasks, nil
	} else {
		return nil, nil
	}
}

// Interpolate alert information into summary/description templates.
func interpolateMessage(msg string, labels clientmodel.LabelSet, value clientmodel.SampleValue) string {
	t := template.New("message")

	// Inject some convenience variables that are easier to remember for users
	// who are not used to Go's templating system.
	defs :=
		"{{$labels := .Labels}}" +
			"{{$value := .Value}}"

	if _, err := t.Parse(defs + msg); err != nil {
		glog.Warning("Error parsing template: ", err)
		return msg
	}

	l := map[string]string{}
	for k, v := range labels {
		l[string(k)] = string(v)
	}

	tmplData := struct {
		Labels map[string]string
		Value  clientmodel.SampleValue
	}{
		Labels: l,
		Value:  value,
	}

	var buf bytes.Buffer
	if err := t.Execute(&buf, &tmplData); err != nil {
		glog.Warning("Error executing template: ", err)
		return msg
	}
	return buf.String()
}

func (storage *Storage) SaveSyncMessage(emsg *EMessage) error {
	storage.mutex.Lock()
	defer storage.mutex.Unlock()

	filesize, err := storage.file.Seek(0, os.SEEK_END)
	if err != nil {
		log.Fatalln(err)
	}
	if emsg.msgid != filesize {
		log.Warningf("file size:%d, msgid:%d is't equal", filesize, emsg.msgid)
		if emsg.msgid < filesize {
			log.Warning("skip msg:", emsg.msgid)
		} else {
			log.Warning("write padding:", emsg.msgid-filesize)
			padding := make([]byte, emsg.msgid-filesize)
			_, err = storage.file.Write(padding)
			if err != nil {
				log.Fatal("file write:", err)
			}
		}
	}

	storage.WriteMessage(storage.file, emsg.msg)

	if emsg.msg.cmd == MSG_OFFLINE {
		off := emsg.msg.body.(*OfflineMessage)
		storage.AddOffline(off.msgid, off.appid, off.receiver)
		storage.SetLastMessageID(off.appid, off.receiver, emsg.msgid)
	} else if emsg.msg.cmd == MSG_ACK_IN {
		off := emsg.msg.body.(*OfflineMessage)
		storage.RemoveOffline(off.msgid, off.appid, off.receiver)
	}
	log.Info("save sync message:", emsg.msgid)
	return nil
}

// Update handles PUT
func (ctl *WatcherController) Update(c *models.Context) {
	m := models.WatcherType{}
	err := c.Fill(&m)

	if err != nil {
		glog.Warning(err)
		c.RespondWithErrorMessage(
			fmt.Sprintf("The post data is invalid: %v", err.Error()),
			http.StatusBadRequest,
		)
		return
	}

	itemType := strings.ToLower(m.ItemType)
	if itemType != "" {
		if _, exists := h.ItemTypes[itemType]; !exists {
			glog.Warning(err)
			c.RespondWithErrorMessage(
				fmt.Sprintf("Watcher could not be saved: Item type not found"),
				http.StatusBadRequest,
			)
			return
		}

		m.ItemTypeID = h.ItemTypes[itemType]
	}

	var status int
	// watcher must exist to be updated
	// Also the returned watcher ID belongs to the authed person by definition
	// - no need to check later
	m.ID, _, _, _, status, err = models.GetWatcherAndIgnoreStatus(
		m.ItemTypeID,
		m.ItemID,
		c.Auth.ProfileID,
	)
	if err != nil {
		glog.Error(err)
		c.RespondWithErrorDetail(err, status)
		return
	}

	// To update we only need id, SendEmail and SendSMS
	status, err = m.Update()
	if err != nil {
		glog.Error(err)
		c.RespondWithErrorMessage(
			fmt.Sprintf("Could not update watcher: %v", err.Error()),
			http.StatusBadRequest,
		)
		return
	}

	// Respond
	c.RespondWithOK()
}

func NewOpenStack(config io.Reader) (*OpenStack, error) {
	var cfg Config
	err := gcfg.ReadInto(&cfg, config)
	if err != nil {
		glog.Warning("Failed to parse openstack configure file: %v", err)
		return nil, err
	}

	provider, err := openstack.AuthenticatedClient(cfg.toAuthOptions())
	if err != nil {
		glog.Warning("Failed to auth openstack: %v", err)
		return nil, err
	}

	identity, err := openstack.NewIdentityV2(provider, gophercloud.EndpointOpts{
		Availability: gophercloud.AvailabilityAdmin,
	})
	if err != nil {
		glog.Warning("Failed to find identity endpoint")
		return nil, err
	}

	network, err := openstack.NewNetworkV2(provider, gophercloud.EndpointOpts{
		Region: cfg.Global.Region,
	})
	if err != nil {
		glog.Warning("Failed to find neutron endpoint: %v", err)
		return nil, err
	}

	os := OpenStack{
		identity:   identity,
		network:    network,
		provider:   provider,
		region:     cfg.Global.Region,
		lbOpts:     cfg.LoadBalancer,
		pluginOpts: cfg.Plugin,
		ExtNetID:   cfg.Global.ExtNetID,
	}

	// init plugin
	if cfg.Plugin.PluginName != "" {
		integrationBriage := "br-int"
		if cfg.Plugin.IntegrationBridge != "" {
			integrationBriage = cfg.Plugin.IntegrationBridge
		}

		plugin, _ := plugins.GetNetworkPlugin(cfg.Plugin.PluginName)
		if plugin != nil {
			plugin.Init(integrationBriage)
			os.Plugin = plugin
		}
	}

	return &os, nil
}

func CleanupTempFile(f *os.File) {
	err := f.Close()
	if err != nil {
		glog.Warning("error closing temp file: %v", err)
	}
	err = os.Remove(f.Name())
	if err != nil {
		glog.Warning("error deleting temp file: %v", err)
	}
}

// handleClientStateChange makes clean transitions as the connection with
// avahi-daemon changes.
//export handleClientStateChange
func handleClientStateChange(client *C.AvahiClient, newState C.AvahiClientState, userdata unsafe.Pointer) {
	z := instance
	z.spMutex.Lock()
	defer z.spMutex.Unlock()

	// Name conflict.
	if newState == C.AVAHI_CLIENT_S_COLLISION {
		glog.Warning("Avahi reports a host name collision.")
	}

	// Transition from not connecting to connecting. Warn in logs.
	if newState == C.AVAHI_CLIENT_CONNECTING {
		glog.Warning("Cannot find Avahi daemon. Is it running?")
	}

	// Transition from running to not running. Free all groups.
	if newState != C.AVAHI_CLIENT_S_RUNNING {
		glog.Info("Local printing disabled (Avahi client is not running).")
		for name, r := range z.printers {
			if r.group != nil {
				if errstr := C.removeAvahiGroup(z.threadedPoll, r.group); errstr != nil {
					err := errors.New(C.GoString(errstr))
					glog.Errorf("Failed to remove Avahi group: %s", err)
				}
				r.group = nil
				z.printers[name] = r
			}
		}
	}

	// Transition from not running to running. Recreate all groups.
	if newState == C.AVAHI_CLIENT_S_RUNNING {
		glog.Info("Local printing enabled (Avahi client is running).")
		for name, r := range z.printers {
			txt := prepareTXT(r.ty, r.url, r.id, r.online)
			defer C.avahi_string_list_free(txt)

			if errstr := C.addAvahiGroup(z.threadedPoll, z.client, &r.group, r.name, C.ushort(r.port), txt); errstr != nil {
				err := errors.New(C.GoString(errstr))
				glog.Errorf("Failed to add Avahi group: %s", err)
			}

			z.printers[name] = r
		}
	}

	// Transition from not failure to failure. Recreate thread poll and client.
	if newState == C.AVAHI_CLIENT_FAILURE {
		z.restart <- struct{}{}
	}

	z.state = newState
}