Merge pull request #67 from Badangel/cube-builder

support agent download by http or ftp

cube/cube-transfer/conf/transfer.conf

 [default]
 dict_name: test_dict
 mode: base_only
-storage_place: LOCAL
+download_mode: http
+wget_port:8009
 buildtool_local: /home/work/Serving/build/output/bin/cube-builder
 donefile_address: http://127.0.0.1/home/work/dangyifei/donefile
 output_address: /home/work/dangyifei/test-transfer/test_data/output

cube/cube-transfer/src/cube-transfer.go

@@ -95,13 +95,27 @@ Log options:
 	}
 	logex.Notice(">>> Mode:", transfer.Dict.DictMode)
 
-	transfer.Dict.StoragePlace = configMgr.Read("default", "storage_place")
-	if transfer.Dict.StoragePlace == "" || transfer.Dict.StoragePlace != "LOCAL" {
-		fmt.Fprintln(os.Stderr, "ERROR: nead [default] StoragePlace in config_file! only support Local")
+	transfer.Dict.DownloadMode = configMgr.Read("default", "download_mode")
+	if transfer.Dict.DownloadMode != "http" || transfer.Dict.DownloadMode != "ftp" {
+		fmt.Fprintln(os.Stderr, "ERROR: nead [default] download_mode in config_file! only support ftp or http")
 		fmt.Fprintln(os.Stderr, usage)
 		os.Exit(1)
 	}
-	logex.Notice(">>> StoragePlace:", transfer.Dict.StoragePlace)
+	logex.Notice(">>> DownloadMode:", transfer.Dict.DownloadMode)
+
+    transfer.Dict.WgetPort = configMgr.Read("default", "wget_port")
+    if transfer.Dict.WgetPort == "" {
+        fmt.Fprintln(os.Stderr, "ERROR: nead [default] wget_port in config_file!")
+        fmt.Fprintln(os.Stderr, usage)
+        os.Exit(1)
+    }
+    var wget_port int
+    wget_port, err = strconv.Atoi(transfer.Dict.WgetPort)
+    if err != nil {
+        logex.Fatal("wget_port form is not right need int")
+        os.Exit(1)
+    }
+    logex.Notice(">>> WgetPort:", wget_port)
 
 	transfer.BuildToolLocal = configMgr.Read("default", "buildtool_local")
 	if transfer.BuildToolLocal == "" {

cube/cube-transfer/src/transfer/deployer.go

@@ -72,7 +72,7 @@ func CmdInstsDownload() {
 				json_params.Mode = Dict.WaitVersionInfo.Mode
 				json_params.ShardSeq = inst.Shard
 				json_params.Port = strconv.Itoa(inst.Port)
-				json_params.Source = dict.GetFileHead(Dict.StoragePlace, TransferAddr) + Dict.WaitVersionInfo.Output + "/" + json_params.DictName + "_part" + strconv.Itoa(inst.Shard) + ".tar"
+				json_params.Source = dict.GetFileHead(Dict.DownloadMode, TransferAddr, Dict.WgetPort) + Dict.WaitVersionInfo.Output + "/" + json_params.DictName + "_part" + strconv.Itoa(inst.Shard) + ".tar"
 				var address = fmt.Sprintf("http://%v:%v/agent/cmd", inst.AgentIp, inst.AgentPort)
 				logex.Noticef("[download cmd]%v:%v", address, json_params)
 				go nonBlockSendJsonReq("POST2", address, 120, &json_params, &keyAndRespSlice[i], chs[i])
@@ -121,7 +121,7 @@ func CmdInstsReload() {
 				json_params.Mode = Dict.WaitVersionInfo.Mode
 				json_params.ShardSeq = inst.Shard
 				json_params.Port = strconv.Itoa(inst.Port)
-				json_params.Source = dict.GetFileHead(Dict.StoragePlace, TransferAddr) + Dict.WaitVersionInfo.Output + "/" + json_params.DictName + "_part" + strconv.Itoa(inst.Shard) + ".tar"
+				json_params.Source = dict.GetFileHead(Dict.DownloadMode, TransferAddr, Dict.WgetPort) + Dict.WaitVersionInfo.Output + "/" + json_params.DictName + "_part" + strconv.Itoa(inst.Shard) + ".tar"
 
 				var address = fmt.Sprintf("http://%v:%v/agent/cmd", inst.AgentIp, inst.AgentPort)
 				logex.Noticef("[reload cmd]%v:%v", address, json_params)
@@ -170,7 +170,7 @@ func CmdInstsEnable() {
 				json_params.Mode = Dict.WaitVersionInfo.Mode
 				json_params.ShardSeq = inst.Shard
 				json_params.Port = strconv.Itoa(inst.Port)
-				json_params.Source = dict.GetFileHead(Dict.StoragePlace, TransferAddr) + Dict.WaitVersionInfo.Output + "/" + json_params.DictName + "_part" + strconv.Itoa(inst.Shard) + ".tar"
+				json_params.Source = dict.GetFileHead(Dict.DownloadMode, TransferAddr, Dict.WgetPort) + Dict.WaitVersionInfo.Output + "/" + json_params.DictName + "_part" + strconv.Itoa(inst.Shard) + ".tar"
 
 				var address = fmt.Sprintf("http://%v:%v/agent/cmd", inst.AgentIp, inst.AgentPort)
 				logex.Noticef("[enable cmd]%v:%v", address, json_params)

cube/cube-transfer/src/transfer/dict/dict_info.go

@@ -24,7 +24,8 @@ type DictInfo struct {
 	DonefileAddress    string            `json:"donefile_addr"`
 	OutputAddress      string            `json:"output_addr"`
 	TmpAddress         string            `json:"tmp_addr"`
-	StoragePlace       string            `json:"storage_place"`
+	DownloadMode       string            `json:"download_mode"`
+	WgetPort           string            `json:"wget_port"`
 	DownloadSuccInsts  int               `json:"download_inst"`
 	ReloadSuccInsts    int               `json:"reload_insts"`
 	EnableSuccInsts    int               `json:"enable_insts"`

cube/cube-transfer/src/transfer/dict/dict_shard_info.go

@@ -36,7 +36,7 @@ type DictShardInfo struct {
 	IsActive       bool   `json:"is_active,omitempty"`
 }
 
-func GetDictShardScaler(shard int, dictVersionInfo DictVersionInfo, storagePlace string, transferaddr string)(info DictShardInfo){
+func GetDictShardScaler(shard int, dictVersionInfo DictVersionInfo, downloadMode string, transferAddr string, wgetPort string)(info DictShardInfo){
 	info.Name = dictVersionInfo.DictName
 	info.Version = strconv.Itoa(dictVersionInfo.Version)
 	info.Depend = strconv.Itoa(dictVersionInfo.Depend)
@@ -44,16 +44,17 @@ func GetDictShardScaler(shard int, dictVersionInfo DictVersionInfo, storagePlace
 	info.Key = strconv.Itoa(dictVersionInfo.Key)
 	info.Mode = dictVersionInfo.Mode
 	info.Shard = shard
-	info.Source = GetFileHead(storagePlace, transferaddr) + dictVersionInfo.Output+ "/" + info.Version + "/" + info.Name + "_part" + strconv.Itoa(shard) + ".tar"
+	info.Source = GetFileHead(downloadMode, transferAddr, wgetPort) + dictVersionInfo.Output+ "/" + info.Version + "/" + info.Name + "_part" + strconv.Itoa(shard) + ".tar"
 	return
 }
 
 
-func GetFileHead(storagePlace string, transferaddr string) string {
-	if storagePlace == "LOCAL"{
-		return "ftp://" + transferaddr
+func GetFileHead(downloadMode string, transferAddr string, wgetPort string) string {
+	if downloadMode == "http" {
+		return HTTP_HEADER + transferAddr + ":" + wgetPort
+	} else if downloadMode == "ftp" {
+		return FTP_HEADER + transferAddr + ":" + wgetPort
 	} else {
 		return ""
 	}
-
 }
\ No newline at end of file

cube/cube-transfer/src/transfer/http_get.go

@@ -66,11 +66,11 @@ func GetDictScaler(subpath string, m map[string]string) (string, string, int, er
 	}
 
 	for _, version := range Dict.CurrentVersionInfo {
-		info := dict.GetDictShardScaler(shard, version, Dict.StoragePlace, TransferAddr)
+		info := dict.GetDictShardScaler(shard, version,  Dict.DownloadMode, TransferAddr, Dict.WgetPort)
 		infos = append(infos, info)
 	}
 	if Dict.WaitVersionInfo.Status > dict.Dict_Status_Deploying {
-		info := dict.GetDictShardScaler(shard, Dict.WaitVersionInfo, Dict.StoragePlace, TransferAddr)
+		info := dict.GetDictShardScaler(shard, Dict.WaitVersionInfo,  Dict.DownloadMode, TransferAddr, Dict.WgetPort)
 		infos = append(infos, info)
 	}