func (zkClient *ZookeeperClient) getOffsetForPartition(consumerGroup string, topic string, partition string) {
	offsetStr, zkNodeStat, err := zkClient.conn.Get(zkClient.app.Config.Kafka[zkClient.cluster].ZookeeperPath + "/consumers/" + consumerGroup + "/offsets/" + topic + "/" + partition)
	if err != nil {
		log.Warnf("Failed to read partition %s:%v for group %s in cluster %s: %s", topic, partition, consumerGroup, zkClient.cluster, err)
		return
	}

	partitionNum, err := strconv.ParseInt(partition, 10, 32)
	if err != nil {
		log.Errorf("Partition (%s) for topic %s for group %s in cluster %s is not an integer", partition, topic, consumerGroup, zkClient.cluster)
		return
	}

	offset, err := strconv.ParseInt(string(offsetStr), 10, 64)
	if err != nil {
		log.Errorf("Offset value (%s) for partition %s:%v for group %s in cluster %s is not an integer", string(offsetStr), topic, partition, consumerGroup, zkClient.cluster)
		return
	}

	partitionOffset := &PartitionOffset{
		Cluster:   zkClient.cluster,
		Topic:     topic,
		Partition: int32(partitionNum),
		Group:     consumerGroup,
		Timestamp: zkNodeStat.Mtime,
		Offset:    offset,
	}
	timeoutSendOffset(zkClient.app.Storage.offsetChannel, partitionOffset, 1)
}

func (imageManager *dockerImageManager) deleteImage(imageID string, imageState *image.ImageState) {
	if imageID == "" {
		seelog.Errorf("Image ID to be deleted is null")
		return
	}
	seelog.Infof("Removing Image: %s", imageID)
	err := imageManager.client.RemoveImage(imageID, removeImageTimeout)
	if err != nil {
		if err.Error() == imageNotFoundForDeletionError {
			seelog.Errorf("Image already removed from the instance: %v", err)
		} else {
			seelog.Errorf("Error removing Image %v - %v", imageID, err)
			delete(imageManager.imageStatesConsideredForDeletion, imageState.Image.ImageID)
			return
		}
	}
	seelog.Infof("Image removed: %v", imageID)
	imageState.RemoveImageName(imageID)
	if len(imageState.Image.Names) == 0 {
		delete(imageManager.imageStatesConsideredForDeletion, imageState.Image.ImageID)
		imageManager.removeImageState(imageState)
		imageManager.state.RemoveImageState(imageState)
		imageManager.saver.Save()
	}
}

/**
 * When allocation changed, modify the state of slot and update slotmaps.
 */
func HandleAllocationChange(oldAllocations, newAllocations *SlotAllocation, slotinfoMaps *SlotInfoMaps, zkHelper *utils.ZkHelper) {
	isChanged := false
	newSlotInfoMap := make(map[string]*SlotInfo) //create a new slotinfo maps
	for i := 0; i < oldAllocations.SlotCount; i++ {
		oldNodeId := oldAllocations.Allocations[strconv.Itoa(i)]
		newNodeId := newAllocations.Allocations[strconv.Itoa(i)]
		newSlotInfoMap[strconv.Itoa(i)] = slotinfoMaps.GetSlotInfoMap()[strconv.Itoa(i)].Clone()
		if oldNodeId != newNodeId {
			isChanged = true
			log.Infof("The slot %d's node changed to %d from %d.", i, newNodeId, oldNodeId)
			newSlotInfoMap[strconv.Itoa(i)].MigrateState = MigStateMigrating
			newSlotInfoMap[strconv.Itoa(i)].NodeId = strconv.Itoa(newNodeId)
			newSlotInfoMap[strconv.Itoa(i)].SrcNodeId = strconv.Itoa(oldNodeId)
			newSlotInfoMap[strconv.Itoa(i)].TargetNodeId = strconv.Itoa(newNodeId)
			//update the new slotinfo to zk
			jsonStr, err := utils.ToJson(newSlotInfoMap[strconv.Itoa(i)])
			if err != nil {
				log.Errorf("Can not convert to json string from obj [%s]", newSlotInfoMap[strconv.Itoa(i)])
			} else {
				_, err = zkHelper.CoverCreate("/yundis/ids/"+strconv.Itoa(i), []byte(jsonStr), 0, zk.WorldACL(zk.PermAll))
				if err != nil {
					log.Errorf("Change the value of /yundis/ids/%d fail, err:%s.", i, err)
				}
				//zkHelper.Set("/yundis/ids/"+strconv.Itoa(i), []byte(jsonStr), 1)
			}
		}
	}
	if isChanged {
		log.Info("Update the slotinfoMaps.")
		//slotinfoMaps.SetSlotInfoMap(infoMap)
		slotinfoMaps.SetSlotInfoMap(newSlotInfoMap)
	}
}

/**
 * CNAME -> A
 */
func dnss(host, dnsServer string) *net.IP {
	addrs, err := Lookup("CNAME", host, dnsServer)
	if err != nil {
		seelog.Errorf("dns cname fail with the host[%s]. error: [%s]", host, err.Error())
		return nil
	}

	for {
		if len(addrs.Answer) == 0 {
			break
		}
		host = addrs.Answer[0].(*dns.CNAME).Target
		addrs, err = Lookup("CNAME", host, dnsServer)
		if err != nil {
			seelog.Errorf("dns cname fail with the host[%s]. error: [%s]", host, err.Error())
			return nil
		}
	}
	addrs, err = Lookup("A", host, dnsServer)
	if err != nil {
		seelog.Errorf("dns a fail with the host[%s]. error: [%s]", host, err.Error())
		return nil
	}
	for _, a := range addrs.Answer {
		if a.(*dns.A).A != nil {
			return &a.(*dns.A).A
		}
	}
	return nil
}

func New(endpoint []string, path string) *opt {
	cfg := client.Config{
		Endpoints:               endpoint,
		Transport:               client.DefaultTransport,
		HeaderTimeoutPerRequest: time.Second,
	}

	etcdClient, err := client.New(cfg)
	if err != nil {
		log.Errorf("new etcd client error: ", err)
		log.Flush()
		panic(0)
	}
	api := client.NewKeysAPI(etcdClient)
	resp, err := api.Get(context.Background(), "/swarm/docker/swarm/leader", nil)
	if err != nil {
		log.Errorf("get swarm leader error: %v", err)
		log.Flush()
		panic(0)
	}
	return &opt{
		Client:   etcdClient,
		Endpoint: endpoint,
		Path:     path,
		Api:      api,
		Leader:   fmt.Sprintf("http://%s", resp.Node.Value),
	}
}

func QueryTokenByUidClientID(uid int, clientID string) (token PGToken, err error) {
	err = db.Ping()
	if err != nil {
		seelog.Errorf("uid %v clientID %v ping mysql %v", uid, clientID, err)
		return
	}

	qrySql := "select oauth_token,client_id,expires,scope,uid from pg_token where uid=? and client_id=?"
	rs, err := db.Query(qrySql, uid, clientID)
	if err != nil {
		seelog.Errorf("uid %v clientID %v query sql %v", uid, clientID, err)
		return
	}

	if rs.Next() {
		if err = rs.Err(); err != nil {
			seelog.Errorf("uid %v clientID %v read rows %v", uid, clientID, err)
			return
		}
		rs.Scan(&token.OauthToken, &token.ClientId, &token.Expires, &token.Scope, &token.Uid)
	} else {
		err = pgpub.ErrNotExist
	}

	return
}

// publishError publishes an event when a handler returns an error
func publishError(req *Request, e errors.Error) {
	if !PublishErrors {
		return
	}

	stacktrace := ""
	if e.MultiStack() != nil {
		stacktrace = e.MultiStack().String()
	}

	msg := map[string]interface{}{
		"created":     time.Now(),
		"service":     Name,
		"version":     Version,
		"azName":      az,
		"hostname":    hostname,
		"instanceId":  InstanceID,
		"error":       e.Error(),
		"type":        e.Type(),
		"code":        e.Code(),
		"description": e.Description(),
		"httpCode":    e.HttpCode(),
		"context":     e.Context(),
		"traceId":     req.TraceID(),
		"stacktrace":  stacktrace,
	}

	payload, err := json.Marshal(msg)
	if err != nil {
		log.Errorf("[Server] Failed to JSON encode error event: %v", err)
	}
	if err = nsq.Publish(errorTopic, payload); err != nil {
		log.Errorf("[Server] Failed to publish error event: %v", err)
	}
}

// handleSingleMessage processes a single refresh credentials message.
func (refreshHandler *refreshCredentialsHandler) handleSingleMessage(message *ecsacs.IAMRoleCredentialsMessage) error {
	// Validate fields in the message
	err := validateIAMRoleCredentialsMessage(message)
	if err != nil {
		seelog.Errorf("Error validating credentials message: %v", err)
		return err
	}
	taskArn := aws.StringValue(message.TaskArn)
	messageId := aws.StringValue(message.MessageId)
	task, ok := refreshHandler.taskEngine.GetTaskByArn(taskArn)
	if !ok {
		seelog.Errorf("Task not found in the engine for the arn in credentials message, arn: %s, messageId: %s", taskArn, messageId)
		return fmt.Errorf("Task not found in the engine for the arn in credentials message, arn: %s", taskArn)
	}
	taskCredentials := credentials.TaskIAMRoleCredentials{
		ARN:                taskArn,
		IAMRoleCredentials: credentials.IAMRoleCredentialsFromACS(message.RoleCredentials),
	}
	err = refreshHandler.credentialsManager.SetTaskCredentials(taskCredentials)
	if err != nil {
		seelog.Errorf("Error updating credentials, err: %v messageId: %s", err, messageId)
		return fmt.Errorf("Error updating credentials %v", err)
	}
	task.SetCredentialsId(aws.StringValue(message.RoleCredentials.CredentialsId))

	go func() {
		response := &ecsacs.IAMRoleCredentialsAckRequest{
			Expiration:    message.RoleCredentials.Expiration,
			MessageId:     message.MessageId,
			CredentialsId: message.RoleCredentials.CredentialsId,
		}
		refreshHandler.ackRequest <- response
	}()
	return nil
}

// ServeHttp serves IAM Role Credentials for Tasks being managed by the agent.
func ServeHttp(credentialsManager credentials.Manager, containerInstanceArn string, cfg *config.Config) {
	// Create and initialize the audit log
	// TODO Use seelog's programmatic configuration instead of xml.
	logger, err := log.LoggerFromConfigAsString(audit.AuditLoggerConfig(cfg))
	if err != nil {
		log.Errorf("Error initializing the audit log: %v", err)
		// If the logger cannot be initialized, use the provided dummy seelog.LoggerInterface, seelog.Disabled.
		logger = log.Disabled
	}

	auditLogger := audit.NewAuditLog(containerInstanceArn, cfg, logger)

	server := setupServer(credentialsManager, auditLogger)

	for {
		utils.RetryWithBackoff(utils.NewSimpleBackoff(time.Second, time.Minute, 0.2, 2), func() error {
			// TODO, make this cancellable and use the passed in context;
			err := server.ListenAndServe()
			if err != nil {
				log.Errorf("Error running http api: %v", err)
			}
			return err
		})
	}
}

func StartTcpServer(host, port string, service ConnectorService) {

	if service == nil {
		connService = &MqService{}
	} else {
		connService = service
	}
	connService.InitService()

	addr, err := net.ResolveTCPAddr("tcp", ":9000")
	if err != nil {
		logger.Errorf("Can not resolve tcp address for server. host: %s. port: %s. address string: ")
		panic(err)
	}

	listener, err := net.ListenTCP("tcp", addr)
	if err != nil {
		logger.Criticalf("Can not start tcp server on address: %s:%s. Error: %s", addr.IP, addr.Port, err)
		panic(err)
	}
	defer listener.Close()

	for {
		conn, err := listener.AcceptTCP()
		if err != nil {
			logger.Errorf("Create tcp connection error. Err: %s", err)
			continue
		}

		handleConnection(conn)
	}
}

func (this *Orchestrator) handleAction(action Action) {
	var err error = nil
	ocSideOnly := false
	ocSide, ocSideOk := action.(OrchestratorSideAction)
	action.SetTriggeredTime(time.Now())
	log.Debugf("action %s is executable on the orchestrator side: %t", action, ocSideOk)
	if ocSideOk {
		ocSideOnly = ocSide.OrchestratorSideOnly()
		log.Debugf("action %s is executable on only the orchestrator side: %t", action, ocSideOnly)
		err = ocSide.ExecuteOnOrchestrator()
		if err != nil {
			log.Errorf("ignoring an error occurred while ExecuteOnOrchestrator: %s", err)
		}
	}

	if !ocSideOnly {
		// pass to the inspector handler.
		entity := GetTransitionEntity(action.EntityID())
		if entity == nil {
			err = fmt.Errorf("could find entity %s for %s", action.EntityID(), action)
			log.Errorf("ignoring an error: %s", err)
		} else {
			log.Debugf("Main[%s]->Handler: sending an action %s", entity.ID, action)
			entity.ActionFromMain <- action
			log.Debugf("Main[%s]->Handler: sent an action %s", entity.ID, action)
		}
	}

	// make sequence for tracing
	if this.collectTrace {
		this.actionSequence = append(this.actionSequence, action)
	}
}

func HandlePush(agent *Agent, pkt *packet.Pkt) {
	dataMsg := packet.PktDataMessage{}
	err := packet.Unpack(pkt, &dataMsg)
	if err != nil {
		log.Errorf("Error unpack push msg: %s", err.Error())
	}
	log.Infof("Received push message: MsgId: %d, Type: %d,  Appid: %s, Msg: %s\n",
		dataMsg.MsgId, dataMsg.MsgType, dataMsg.AppId, dataMsg.Msg)

	dataAck := packet.PktDataACK{
		MsgId: dataMsg.MsgId,
		AppId: dataMsg.AppId,
		RegId: agent.RegIds[dataMsg.AppId],
	}

	pktAck, err := packet.Pack(packet.PKT_ACK, 0, dataAck)
	if err != nil {
		log.Errorf("Pack error: %s", err.Error())
		return
	}

	agent.SendPkt(pktAck)

	if agent.OnReceiveMsg != nil {
		agent.OnReceiveMsg(dataMsg.AppId, dataMsg.MsgId, dataMsg.MsgType, dataMsg.Msg)
	}
}

func (migrator *Migrator) migrateAttachments(c *desk.Case, requester *zendesk.Customer) ([]zendesk.Comment, error) {
	comments := make([]zendesk.Comment, 0)
	for _, attachment := range c.Attachments {
		err := migrator.deskClient.DownloadFile(&attachment, fmt.Sprintf("%s/attachments/", migrator.path))
		if err != nil {
			log.Errorf("Can't download attachment: %v. Ignoring.", err)
			continue
		}
		token, err := migrator.migrateAttachment(&attachment, fmt.Sprintf("%s/attachments/", migrator.path))
		if err != nil {
			log.Errorf("Can't migrate attachment: %v. Ignoring", err)
			continue
		}

		if token != "" {
			comments = append(comments, zendesk.Comment{
				AuthorId: requester.Id,
				Value:    attachment.FileName,
				Uploads:  []string{token},
			})
		}
	}

	return comments, nil
}

/**
 * watch the slot list change.
 */
func (self *SlotInfoMaps) WatchSlotInfoMap() {
	_, _, ch, err := self.zk.GetZkConn().GetW("/yundis/ids")
	if err != nil {
		log.Errorf("Can not watch path /yundis/ids, err:%s", err)
	}

	go func() {
		for {
			event := <-ch
			log.Infof("Slotinfo list changed event, %+v", event)
			data, _, ch1, err1 := self.zk.GetZkConn().GetW("/yundis/ids")
			if err1 == nil {
				ch = ch1
				//handle the node list change event
				log.Infof("Slotinfo list changed : %s", data)
				infoMap := self.GetSlotInfoMapFromZk()
				//change the slotinfo state.
				self.SetSlotInfoMap(infoMap) //refresh nodeinfo map by new zk data.
				log.Info("Refresh slotinfo map by new zk data.")
			} else {
				log.Errorf("Can not watching the children of /yundis/ids, err:%s", err1)
				break
			}
			time.Sleep(time.Second)
		}
	}()
}

/**
 * watch the node list change.
 */
func (self *NodeInfoMaps) WatchNodeInfoMap() {
	_, _, ch, err := self.zk.GetZkConn().ChildrenW("/yundis/nodes")
	if err != nil {
		log.Errorf("Can not watch path /yundis/nodes, err:%s", err)
	}
	go func() {
		for {
			event := <-ch
			log.Infof("node list change, %+v", event)
			children, _, ch1, err1 := self.zk.GetZkConn().ChildrenW("/yundis/nodes")
			if err1 == nil {
				ch = ch1
				//handle the node list change event
				log.Infof("node list changed : %s", children)
				infoMap := self.getNodeInfoMapFromZk()
				//change the slotinfo state.
				log.Info("The node list changed, begin to change the affected slot's info.")
				self.SetNodeInfoMap(infoMap) //refresh nodeinfo map by new zk data.
				self.ModifySlotState(infoMap)
				log.Info("Refresh nodeinfo map by new zk data.")
			} else {
				log.Errorf("Can not watching the children of /yundis/nodes, err:%s", err1)
				break
			}
			time.Sleep(time.Second)
		}
	}()
}

func runFileChmod(base_path, path string) bool {
	listener := utils.GetListenerFromDir(base_path)
	rel_path := utils.GetRelativePath(listener, path)
	dbItem, err := datastore.GetOne(base_path, rel_path)
	if err != nil {
		log.Errorf("Error occurred trying to get %s from DB\nError: %s", rel_path, err.Error())
		return false
	}
	fsItem, err := utils.GetFileInfo(path)
	if err != nil {
		log.Errorf("Could not find item on filesystem: %s\nError:%s", path, err.Error())
	}
	if dbItem.Perms != fsItem.Perms {
		iPerm, _ := strconv.Atoi(dbItem.Perms)
		mode := int(iPerm)
		if _, err := os.Stat(path); os.IsNotExist(err) {
			log.Infof("File no longer exists returning")
			return true
		} else {
			err := os.Chmod(path, os.FileMode(mode))
			if err != nil {
				log.Errorf("Error occurred changing file modes: %s", err.Error())
				return false
			}
			return true
		}
	} else {
		log.Info("File modes are correct changing nothing")
		return false
	}
	return true
}

func CreateRecursivePath(zooPath string) error {
	// check if parent dir is exist
	log.Debugf("Create zoo path [%s]", zooPath)
	if zooPath == "/" {
		return nil
	} else {
		//create directly
		bExist, _, err := gZKConn.Exists(zooPath)
		if err != nil {
			log.Errorf("Zookeeper execute Exists error=%s", err.Error())
			return errors.New("Zookeeper Exists path error")
		}
		if bExist == false {
			parentDir := path.Dir(zooPath)
			err := CreateRecursivePath(parentDir)
			if err != nil {
				return err
			}
			// Create it selft
			flags := int32(0)
			acl := zk.WorldACL(zk.PermAll)
			_, err = gZKConn.Create(zooPath, []byte("1"), flags, acl)
			if err != nil {
				log.Errorf("Zookeeper create path [%s] error=%s", zooPath, err.Error())
				return errors.New("Zookeeper create path error")
			}
			return nil
		} else {
			return nil
		}
	}

	return nil
}

func queryUser(colName string, val string) (pguser PGUser, err error) {
	err = db.Ping()
	if err != nil {
		seelog.Errorf("queryUser(%v, %v) open db: %v", colName, val, err)
		return
	}

	qrySql := "select " + strings.Join(getStructCols(pguser), ",") + " from pg_user where " + colName + "=?"
	seelog.Debugf("queryUser(%v, %v) sql: %v", colName, val, qrySql)

	rs, err := db.Query(qrySql, val)
	if err != nil {
		seelog.Errorf("queryUser(%v, %v) qry %v %v", colName, val, qrySql, err)
	}

	if rs.Next() {
		if err = rs.Err(); err != nil {
			seelog.Errorf("queryUser(%v, %v) read rows %v", colName, val, err)
			return
		}

		if err = Scan2Struct(rs, &pguser); err != nil {
			seelog.Error(err)
			return
		}
	} else {
		err = pgpub.ErrNotExist
	}

	return
}

func (slack *SlackNotifier) postToSlack(slackMessage *SlackMessage) error {
	data, err := json.Marshal(slackMessage)
	if err != nil {
		log.Errorf("Unable to marshal slack payload:%+v", err)
		return err
	}
	log.Debugf("struct = %+v, json = %s", slackMessage, string(data))

	b := bytes.NewBuffer(data)
	req, err := http.NewRequest("POST", slack.Url, b)
	req.Header.Set("Content-Type", "application/json")

	if res, err := slack.HttpClient.Do(req); err != nil {
		log.Errorf("Unable to send data to slack:%+v", err)
		return err
	} else {
		defer res.Body.Close()
		statusCode := res.StatusCode
		if statusCode != 200 {
			body, _ := ioutil.ReadAll(res.Body)
			log.Errorf("Unable to notify slack:%s", string(body))
			return errors.New("Send to Slack failed")
		} else {
			log.Debug("Slack notification sent")
			return nil
		}
	}
}

func (r *RabbitConnection) Consume(serverName string) (<-chan amqp.Delivery, *RabbitChannel, error) {
	consumerChannel, err := NewRabbitChannel(r.Connection)
	if err != nil {
		log.Errorf("[Rabbit] Failed to create new channel: %s", err.Error())
		return nil, nil, err
	}

	err = consumerChannel.DeclareQueue(serverName)
	if err != nil {
		log.Errorf("[Rabbit] Failed to declare queue %s: %s", serverName, err.Error())
		return nil, nil, err
	}

	deliveries, err := consumerChannel.ConsumeQueue(serverName)
	if err != nil {
		log.Errorf("[Rabbit] Failed to declare queue %s: %s", serverName, err.Error())
		return nil, nil, err
	}

	err = consumerChannel.BindQueue(serverName, Exchange)
	if err != nil {
		log.Errorf("[Rabbit] Failed to bind %s to %s exchange: %s", serverName, Exchange, err.Error())
		return nil, nil, err
	}

	return deliveries, consumerChannel, nil
}