func modifyVcapServicesEnvNameEnv(env []kapi.EnvVar, bsName string, vsp *VcapServiceParameters, bsiName string) (bool, []kapi.EnvVar) {
	_, json_env := env_get(env, VcapServicesEnvName)

	vs := VcapServices{}
	if len(strings.TrimSpace(json_env)) > 0 {
		err := json.Unmarshal([]byte(json_env), &vs)
		if err != nil {
			glog.Warningln("unmarshalVcapServices error: ", err.Error())
		}
	}

	if vsp != nil {
		vs = addVcapServiceParameters(vs, bsName, vsp)
	}
	if bsiName != "" {
		vs = removeVcapServiceParameters(vs, bsName, bsiName)
	}

	if len(vs) == 0 {
		return env_unset(env, VcapServicesEnvName)
	}
	json_data, err := json.Marshal(vs)
	if err != nil {
		glog.Warningln("marshalVcapServices error: ", err.Error())
		return false, env
	}

	json_env = string(json_data)

	glog.Info("new ", VcapServicesEnvName, " = ", json_env)

	return env_set(env, VcapServicesEnvName, json_env)
}

func fetchNetDevs() []machinedata.NetDev {
	classNetDir := "/sys/class/net"
	classNet, err := os.Open(classNetDir)
	if err != nil {
		glog.Warningln(err)
		return []machinedata.NetDev{}
	}

	finfos, err := classNet.Readdir(1000)
	if err != nil {
		glog.Warningln(err)
		return []machinedata.NetDev{}
	}

	rv := []machinedata.NetDev{}

	for _, fi := range finfos {
		fullPath := path.Join(classNetDir, fi.Name())
		link, err := os.Readlink(fullPath)
		if err != nil {
			continue
		}
		if !strings.Contains(link, "virtual") {
			macAddr, _ := readContents(path.Join(fullPath, "address"))
			rv = append(rv, machinedata.NetDev{
				Name:       fi.Name(),
				MacAddress: macAddr,
			})
		}
	}

	return rv
}

func (this *tallySubscriberImpl) Queue(queue string, inbound <-chan interface{}) {
	go func() {
		var queueLength int = 0
		var err error
		queueLength, _ = redis.Int(this.queue.Do("LLEN", queue))
		for {
			select {
			case message := <-inbound:
				if queueLength >= this.settings.MaxQueueLength {
					// do a read of the length in the hope that at some point the queue starts to drain
					queueLength, err = redis.Int(this.queue.Do("LLEN", queue))
					if err != nil {
						glog.Warningln("error-llen", queue, err, this.settings)
					}
					if queueLength >= this.settings.MaxQueueLength {
						glog.Warningln("queue-length-exceeds-limit", this.settings.MaxQueueLength, queueLength)
						// drop the message
						continue
					}
				}
				queueLength, err = redis.Int(this.queue.Do("LPUSH", queue, message))
				if err != nil {
					glog.Warningln("error-lpush", queue, err, this.settings)
				}
			case stop := <-this.stop:
				if stop {
					return
				}
			}
		}
	}()
}

func fetchIPMIAddress() (ip net.IP) {
	ipmitool, err := Asset("embedded/ipmitool")
	if err != nil {
		glog.Warningln(err)
		return
	}

	tempFile, err := ioutil.TempFile("", "ipmitool")
	if err != nil {
		glog.Warningln(err)
		return
	}

	tempFile.Write(ipmitool)
	tempFile.Chmod(0700)
	tempFile.Close()

	ipmitoolCmd := exec.Command(tempFile.Name(), "lan", "print")
	output, err := ipmitoolCmd.CombinedOutput()
	if err != nil {
		glog.Warningln(err)
		return
	}
	lines := strings.Split(string(output), "\n")
	for _, line := range lines {
		if strings.HasPrefix(line, "IP Address    ") {
			fields := strings.Fields(line)
			return net.ParseIP(fields[len(fields)-1])
		}
	}

	return
}

func (this *rapperVideo) updateTask() {
	for {
		// 取一个任务
		oneVideoTask := <-this.toUpdate

		// 回调
		para := map[string]string{"type": confJson["tasktype"].(string), "tid": oneVideoTask.Tid, "rid": oneVideoTask.Rid}
		if oneVideoTask.err != nil {
			para["msg"] = base64.StdEncoding.EncodeToString([]byte(oneVideoTask.err.Error()))
		}
		if oneVideoTask.nfid != "" {
			para["nfid"] = oneVideoTask.nfid
		}
		if oneVideoTask.nimg != "" {
			para["img"] = oneVideoTask.nimg
		}

		olderr := ""
		if oneVideoTask.err != nil {
			olderr = oneVideoTask.err.Error()
		}
		glog.Warningln("callbackTask: ", this.no, oneVideoTask.toString(), para)
		body, err := getRequest(oneVideoTask.Callback, &para)
		if err != nil {
			oneVideoTask.err = fmt.Errorf("%s\ncallbackERR: %s", olderr, err.Error())
			glog.Errorln("updateTask callbackERR:", oneVideoTask.toString())
		} else if string(body) != "true" {
			oneVideoTask.err = fmt.Errorf("%s\ncallbackERR: %s", olderr, body)
			glog.Errorln("updateTask callbackERR:", oneVideoTask.toString(), body)
		}

		// 更新任务状态
		delete(para, "rid")
		delete(para, "nfid")
		delete(para, "img")
		para["stat"] = "1"
		para["name"] = confJson["rappername"].(string)
		if oneVideoTask.err != nil {
			para["stat"] = "-1"
			para["msg"] = base64.StdEncoding.EncodeToString([]byte(oneVideoTask.err.Error()))
		}

		glog.Infoln("updateTask: ", this.no, oneVideoTask.toString(), para)
		_, err = getRequest(confJson["taskServ"].(string)+"/uptask", &para)
		if err == nil {
			glog.Warningln("updateTaskOK: ", this.no, oneVideoTask.toString())
		} else {
			glog.Errorln("updateTaskERR:", this.no, oneVideoTask.toString(), para, err.Error())
		}

		glog.Flush()

		// 删除临时文件
		fn := confJson["tmpdir"].(string) + oneVideoTask.Tid
		os.Remove(fn)
		os.Remove(fn + ".mp4")
		os.Remove(fn + ".jpg")
	}
}

func (p *AppInfo) GetNewAccessToken() (*AccessTokenType, error) {
	p.mu.Lock()
	defer p.mu.Unlock()

	glog.V(2).Infoln("GetNewAccessToken start")
	defer glog.V(2).Infoln("GetNewAccessToken finish")

	glog.V(2).Infoln("server_url:", OAUTH_SERVER_URL)

	vs := make(url.Values)
	vs.Add("grant_type", p.GrantType)
	vs.Add("client_id", p.ApiKey)
	vs.Add("client_secret", p.SecretKey)
	vs.Add("scope", strings.Join(p.Scope, ","))

	qs := vs.Encode()
	glog.V(2).Infoln("get access_token querystring:", qs)

	body := bytes.NewBufferString(qs)
	req, err := http.NewRequest("POST", OAUTH_SERVER_URL, body)
	if err != nil {
		glog.Warningln("build request failed:", err)
		return nil, err
	}

	if glog.V(2) {
		dump, _ := httputil.DumpRequest(req, true)
		glog.Infoln("request_dump,", string(dump))
	}

	resp, err := p.HttpClient.Do(req)
	if err != nil {
		glog.Warningln("send request failed:", err)
		return nil, err
	}
	glog.V(2).Infoln("resp status:", resp.Status)
	defer resp.Body.Close()
	dataBs, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		glog.Warningln("read response failed:", err)
		return nil, err
	}
	glog.V(2).Info("response:", string(dataBs))
	var token *AccessTokenType
	err = json.Unmarshal(dataBs, &token)
	if err != nil {
		return nil, fmt.Errorf("%s,resp:%s", err.Error(), string(dataBs))
	}
	token.TokenGetTime = time.Now()
	p.AccessToken = token
	err = p.Save2File()
	return token, err
}

func srvTunnelUDP(conn net.Conn, uconn *net.UDPConn) {
	go func() {
		b := make([]byte, 16*1024)

		for {
			n, addr, err := uconn.ReadFromUDP(b)
			if err != nil {
				if glog.V(LWARNING) {
					glog.Warningln(err)
				}
				return
			}

			udp := gosocks5.NewUDPDatagram(
				gosocks5.NewUDPHeader(uint16(n), 0, ToSocksAddr(addr)), b[:n])
			//log.Println("r", udp.Header)
			if err := udp.Write(conn); err != nil {
				if glog.V(LWARNING) {
					glog.Warningln(err)
				}
				return
			}
		}
	}()

	for {
		udp, err := gosocks5.ReadUDPDatagram(conn)
		if err != nil {
			if glog.V(LWARNING) {
				glog.Warningln(err)
			}
			return
		}
		//log.Println("w", udp.Header)
		addr, err := net.ResolveUDPAddr("udp", udp.Header.Addr.String())
		if err != nil {
			if glog.V(LWARNING) {
				glog.Warningln(err)
			}
			continue // drop silently
		}

		if _, err := uconn.WriteToUDP(udp.Data, addr); err != nil {
			if glog.V(LWARNING) {
				glog.Warningln(err)
			}
			return
		}
	}
}

func checkAuthentication(req *restful.Request, resp *restful.Response, chain *restful.FilterChain) {
	var err error
	var token *http.Cookie
	var id int64
	var valid bool
	token, err = req.Request.Cookie("Token")
	if err != nil {
		if glog.V(2) {
			glog.Warningln("'Token' not set in cookie")
		}
		resp.WriteErrorString(400, "'Token' not set in cookie")
		return
	}
	idStr, err := req.Request.Cookie("Id")
	if err != nil {
		if glog.V(2) {
			glog.Warningln("'Id' not set in cookie")
		}
		resp.WriteErrorString(400, "'Id' not set in cookie")
		return
	}
	id, err = strconv.ParseInt(idStr.Value, 10, 64)
	if err != nil {
		if glog.V(2) {
			glog.Warningln("'Id' not valid int64")
		}
		resp.WriteErrorString(400, "'Id' not valid int64")
		return
	}
	err = q["checkToken"].QueryRow(id, decodeBase64(token.Value)).Scan(&valid)
	if err != nil {
		if glog.V(2) {
			glog.Errorf("Token checking SQL returned error: %v\n", q["checkToken"])
		}
		resp.WriteErrorString(500, "error checking token")
		return
	}
	if !valid {
		if glog.V(1) {
			glog.Infof("Auth BAD for %d\n", id)
		}
		resp.WriteErrorString(401, "Unauthenticated")
		return
	}
	if glog.V(1) {
		glog.Infof("Auth OK for %d\n", id)
	}
	chain.ProcessFilter(req, resp)
}

func (api *Api) BuildRequest() (*http.Request, error) {

	vs := make(url.Values)
	vs.Add("method_name", api.Method)

	bs, err := json.Marshal(api.Data)
	if err != nil {
		return nil, err
	}
	vs.Add("data", string(bs))
	vs.Add("time", fmt.Sprintf("%d", time.Now().Unix()))
	if api.Type != "" {
		vs.Add("type", api.Type)
	}
	qs := vs.Encode()

	glog.V(2).Infoln("goods_api_url:", API_URL)
	glog.V(2).Infoln("callApi queryString:", qs)

	http_req, err := http.NewRequest("POST", API_URL, bytes.NewBufferString(qs))
	if err != nil {
		glog.Warningln("build ApiRequest failed:", err)
		return nil, err
	}

	return http_req, err
}

func (this *Schema) Update(db *sql.DB) (err error) {
	if this.statements == nil {
		this.statements = make(map[StatementKey]*sql.Stmt, 0)
	}

	tx, err := db.Begin()
	if err != nil {
		return
	}

	for _, stmt := range this.AlterTables {
		if _, err := db.Exec(stmt); err != nil {
			return tx.Rollback()
		}
	}
	for _, stmt := range this.UpdateIndexes {
		if _, err := db.Exec(stmt); err != nil {
			glog.Warningln(stmt, "err:", err)
		}
	}

	err = update_schema_version(db, this)
	if err != nil {
		tx.Rollback()
		return
	}
	return tx.Commit()
}

func (p *pusher) Push(entry *protolog.Entry) error {
	if entry.Level == protolog.Level_LEVEL_DEBUG && !p.logDebug {
		return nil
	}
	dataBytes, err := p.marshaller.Marshal(entry)
	if err != nil {
		return err
	}
	data := string(dataBytes)
	switch entry.Level {
	case protolog.Level_LEVEL_DEBUG, protolog.Level_LEVEL_INFO:
		glog.Infoln(data)
	case protolog.Level_LEVEL_WARN:
		glog.Warningln(data)
	case protolog.Level_LEVEL_ERROR:
		glog.Errorln(data)
	case protolog.Level_LEVEL_FATAL:
		// cannot use fatal since this will exit before logging completes,
		// which is particularly important for a multi-pusher
		glog.Errorln(data)
	case protolog.Level_LEVEL_PANIC:
		// cannot use panic since this will panic before logging completes,
		// which is particularly important for a multi-pusher
		glog.Errorln(data)
	}
	return nil
}

func (conn *TCPConn) Write(b []byte) (n int, err error) {
	defer func() {
		if r := recover(); r != nil {
			glog.Warningln(r)
			err = r.(error)
		}
	}()
	iv, err := cipher.NewIV()
	if err != nil {
		glog.V(3).Infoln("fail to create iv", err)
		return 0, err
	}

	enc := cipher.NewEncrypter(conn.cipher, iv)
	encrypted := make([]byte, len(iv)+len(b))
	copy(encrypted, iv)
	enc.Encrypt(encrypted[len(iv):], b)

	if _, err := conn.Conn.Write(encrypted); err != nil {
		glog.V(3).Infoln("fail to write data through connection", err)
		return 0, err
	}

	glog.V(3).Infoln("data writed", encrypted)
	return len(encrypted), nil
}

func HandleSignals(shutdownc <-chan io.Closer) {
	c := make(chan os.Signal, 1)
	signal.Notify(c, syscall.SIGHUP)
	signal.Notify(c, syscall.SIGINT)
	for {
		sig := <-c
		sysSig, ok := sig.(syscall.Signal)
		if !ok {
			glog.Fatal("Not a unix signal")
		}
		switch sysSig {
		case syscall.SIGHUP:
		case syscall.SIGINT:
			glog.Warningln("Got SIGTERM: shutting down")
			donec := make(chan bool)
			go func() {
				cl := <-shutdownc
				if err := cl.Close(); err != nil {
					exitf("Error shutting down: %v", err)
				}
				donec <- true
			}()
			select {
			case <-donec:
				glog.Infoln("Shut down completed.")
				os.Exit(0)
			case <-time.After(5 * time.Second):
				exitf("Timeout shutting down. Exiting uncleanly.")
			}
		default:
			glog.Fatal("Received another signal, should not happen.")
		}
	}
}

func (g *RpcHandler) handleRpcRequest(cmd string, seq int, args []interface{}) (reply []interface{}) {
	glog.Infoln("rpc", cmd, args)
	if cmd == "echo" {
		return []interface{}{seq, "", args}
	}

	defer func() {
		errMsg := ""
		switch v := recover().(type) {
		case nil:
			// no error
		case string:
			errMsg = v
		case error:
			errMsg = v.Error()
		default:
			errMsg = fmt.Sprintf("%T: %v", v, v)
		}
		if errMsg != "" {
			glog.Warningln("rpc-error", cmd, args, errMsg)
			reply = []interface{}{seq, errMsg}
		}
	}()

	// FIXME: hack alert, this is special-cased to deal with the database
	switch cmd {
	case "db-get":
		result := database.Get(args[0].(string))
		return []interface{}{seq, "", result}
	case "db-keys":
		result := database.Keys(args[0].(string))
		return []interface{}{seq, "", result}
	case "db-put":
		var value interface{}
		if len(args) > 1 {
			value = args[1]
		}
		database.Put(args[0].(string), value)
		return nil
	}

	// if there's registered circuit for cmd, set it up and return as a stream
	if _, ok := flow.Registry[cmd]; ok && len(args) == 1 {
		c := flow.NewCircuit()
		c.Add("x", cmd)
		c.AddCircuitry("y", &streamRpcResults{seqNum: seq, replies: g})
		c.Connect("x.Out", "y.In", 0)
		for k, v := range args[0].(map[string]interface{}) {
			c.Feed("x."+k, tryToConvertToTag(v))
		}
		go func() {
			defer flow.DontPanic()
			c.Run()
			g.Out.Send([]interface{}{seq, false}) // end streaming
		}()
		return []interface{}{seq, true} // start streaming
	}

	panic(cmd + "?")
}

func (api *Api) Call(name string, argv interface{}) interface{} {
	if arr, ok := argv.([]interface{}); ok {
		if n := len(arr); n > 0 {
			if m0, ok := arr[0].([]*Metrics); ok {
				for i := 1; i < n; i++ {
					if m, ok := arr[i].([]*Metrics); ok {
						for _, mi := range m {
							m0 = append(m0, mi)
						}
					}
				}
				switch strings.ToLower(name) {
				case "sum", "sumseries":
					return api.sumSeries(m0)
				case "div", "divseries", "divideseries":
					return api.divSeries(m0)
				case "diff", "diffseries":
					return api.diffSeries(m0)
				case "_":
					return m0
				default:
					glog.Warningln("[ ! ]\tFunction not supported: ", name)
				}
			}
		}
	}
	return nil
}

func (selector *clientSelector) OnSelected(method uint8, conn net.Conn) (net.Conn, error) {
	switch method {
	case MethodTLS:
		conn = tls.Client(conn, &tls.Config{InsecureSkipVerify: true})

	case gosocks5.MethodUserPass, MethodTLSAuth:
		if method == MethodTLSAuth {
			conn = tls.Client(conn, &tls.Config{InsecureSkipVerify: true})
		}

		var username, password string
		if selector.arg.User != nil {
			username = selector.arg.User.Username()
			password, _ = selector.arg.User.Password()
		}

		req := gosocks5.NewUserPassRequest(gosocks5.UserPassVer, username, password)
		if err := req.Write(conn); err != nil {
			if glog.V(LWARNING) {
				glog.Warningln("socks5 auth:", err)
			}
			return nil, err
		}
		if glog.V(LDEBUG) {
			glog.Infoln(req)
		}

		res, err := gosocks5.ReadUserPassResponse(conn)
		if err != nil {
			if glog.V(LWARNING) {
				glog.Warningln("socks5 auth:", err)
			}
			return nil, err
		}
		if glog.V(LDEBUG) {
			glog.Infoln(res)
		}

		if res.Status != gosocks5.Succeeded {
			return nil, gosocks5.ErrAuthFailure
		}
	case gosocks5.MethodNoAcceptable:
		return nil, gosocks5.ErrBadMethod
	}

	return conn, nil
}

func (t *suicideTracker) makeJumper(_ jumper) jumper {
	return jumper(func(driver bindings.ExecutorDriver, cancel <-chan struct{}) {
		glog.Warningln("jumping?!")
		if t.jumps != nil {
			atomic.AddUint32(t.jumps, 1)
		}
	})
}

func parseArgs(ss []string) (args []Args) {
	for _, s := range ss {
		if !strings.Contains(s, "://") {
			s = "tcp://" + s
		}
		u, err := url.Parse(s)
		if err != nil {
			if glog.V(LWARNING) {
				glog.Warningln(err)
			}
			continue
		}

		arg := Args{
			Addr: u.Host,
			User: u.User,
		}

		schemes := strings.Split(u.Scheme, "+")
		if len(schemes) == 1 {
			arg.Protocol = schemes[0]
			arg.Transport = schemes[0]
		}
		if len(schemes) == 2 {
			arg.Protocol = schemes[0]
			arg.Transport = schemes[1]
		}

		switch arg.Protocol {
		case "http", "socks", "socks5", "ss":
		default:
			arg.Protocol = "default"
		}
		switch arg.Transport {
		case "ws", "tls", "tcp":
		default:
			arg.Transport = "tcp"
		}

		mp := strings.Split(strings.Trim(u.Path, "/"), ":")
		if len(mp) == 1 {
			arg.EncMeth = mp[0]
		}
		if len(mp) == 2 {
			arg.EncMeth = mp[0]
			arg.EncPass = mp[1]
		}

		if arg.Cert, err = tls.LoadX509KeyPair("cert.pem", "key.pem"); err != nil {
			if glog.V(LFATAL) {
				glog.Fatalln(err)
			}
		}
		args = append(args, arg)
	}

	return
}

// Add a named gadget to the circuit with a unique name.
func (c *Circuit) Add(name, gadget string) {
	constructor := Registry[gadget]
	if constructor == nil {
		glog.Warningln("not found:", gadget)
		return
	}
	c.gnames = append(c.gnames, gadgetDef{name, gadget})
	c.AddCircuitry(name, constructor())
}

func (mgr *pxeManagerT) infoPusher(w http.ResponseWriter, r *http.Request) {
	helper, err := os.Open(path.Join(mgr.staticHTMLPath, "infopusher"))
	if err != nil {
		glog.Warningln(err)
	}
	setContentLength(w, helper)
	defer helper.Close()
	io.Copy(w, helper)
}