添加进度条和配置信息

Author: 张俊杰

.gitignore

@@ -14,6 +14,8 @@
 # Dependency directories (remove the comment below to include it)
 vendor/
 
+config/*.json
+
 #MAC
 .DS_Store
 #IDE

README.md

@@ -2,10 +2,12 @@
  虽然 logstash dts 等工具提供了很好的多数据源同步订阅方案，但是实际项目中并不会把服务器的上机权限或云厂商账号开放给所有成员。
  这个小工具算对这类场景的一种补充,它可以帮助我们从mysql中取出数据并推送到elasticsearch上
 
+![16291791332153](http://pic.phpzjj.com/mweb/2021/08/17/16291791332153.jpg)
+
 # 特点
 
  * 基于 elasticsearch  REST APIs 理论无版本上的兼容问题
- * 配置简单仅需3步即可完成 1. 配置 mapping  2. 编写获取源数据sql语句 3. 对呀字段映射关系
+ * 配置简单仅需3步即可完成 1. 配置 mapping  2. 编写获取源数据sql语句 3. 对应字段映射关系
  * 提供WebGUI配置完成后日常维护无需上机
 
 
@@ -14,6 +16,7 @@
  下载对应平台运行，或源码编译安装
 
 MAC 
+
 ```
 # Linux
 CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build
@@ -108,14 +111,3 @@ filePath.json //单个配置文件
 
 
 ```
-
- 
-# 计划
-
- * 加入更多支持的数据关键字
- * 错误写日志
- * WebGUI 加入 push 时的进度条
- * 在线编辑配置文件
- * docker版
-
- 

config/Config.go

@@ -6,6 +6,8 @@ import (
 )
 
 type Config struct {
+	Title string
+	Port int
 	JobList []JobList
 }
 

config/config.json

@@ -1,4 +1,6 @@
 {
+  "title" : "es推送工具",
+  "port" : 9002,
   "jobList" : [
     {"name": "测试" , "filePath" : "example.json"}
   ]

go.mod

@@ -10,4 +10,8 @@ require github.com/go-sql-driver/mysql v1.5.0
 
 require github.com/olivere/elastic/v7 v7.0.24
 
-require github.com/Jeffail/gabs/v2 v2.6.1
+require (
+	github.com/Jeffail/gabs/v2 v2.6.1
+	github.com/fvbock/endless v0.0.0-20170109170031-447134032cb6
+	github.com/gorilla/websocket v1.4.2
+)

go.sum

@@ -50,6 +50,8 @@ github.com/fortytw2/leaktest v1.3.0 h1:u8491cBMTQ8ft8aeV+adlcytMZylmA5nnwwkRZjI8
 github.com/fortytw2/leaktest v1.3.0/go.mod h1:jDsjWgpAGjm2CA7WthBh/CdZYEPF31XHquHwclZch5g=
 github.com/fsnotify/fsnotify v1.4.7 h1:IXs+QLmnXW2CcXuY+8Mzv/fWEsPGWxqefPtCP5CnV9I=
 github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
+github.com/fvbock/endless v0.0.0-20170109170031-447134032cb6 h1:6VSn3hB5U5GeA6kQw4TwWIWbOhtvR2hmbBJnTOtqTWc=
+github.com/fvbock/endless v0.0.0-20170109170031-447134032cb6/go.mod h1:YxOVT5+yHzKvwhsiSIWmbAYM3Dr9AEEbER2dVayfBkg=
 github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
 github.com/gin-contrib/sse v0.1.0 h1:Y/yl/+YNO8GZSjAhjMsSuLt29uWRFHdHYUb5lYOV9qE=
 github.com/gin-contrib/sse v0.1.0/go.mod h1:RHrZQHXnP2xjPF+u1gW/2HnVO7nvIa9PG3Gm+fLHvGI=
@@ -111,6 +113,7 @@ github.com/googleapis/gax-go/v2 v2.0.4/go.mod h1:0Wqv26UfaUD9n4G6kQubkQ+KchISgw+
 github.com/googleapis/gax-go/v2 v2.0.5/go.mod h1:DWXyrwAJ9X0FpwwEdw+IPEYBICEFu5mhpdKc/us6bOk=
 github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1 h1:EGx4pi6eqNxGaHF6qqu48+N2wcFQ5qg5FXgOdqsJ5d8=
 github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY=
+github.com/gorilla/websocket v1.4.2 h1:+/TMaTYc4QFitKJxsQ7Yye35DkWvkdLcvGKqM+x0Ufc=
 github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/grpc-ecosystem/go-grpc-middleware v1.0.0/go.mod h1:FiyG127CGDf3tlThmgyCl78X/SZQqEOJBCDaAfeWzPs=
 github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0/go.mod h1:8NvIoxWQoOIhqOTXgfV/d3M/q6VIi02HzZEHgUlZvzk=

service/monitor/progress.go

@@ -0,0 +1,17 @@
+package monitor
+
+import "sync"
+
+var ProgressBars  = map[string]ProgressBar{}
+
+type ProgressBar struct {
+	Total int
+	Progress *sync.Map
+}
+
+type ProgressBarJson struct {
+	Name string
+	Total int
+	Progress int
+	Status int  // 101 运行中  200 执行完成
+}

service/push/bulkPush.go

@@ -8,6 +8,7 @@ import (
 	"github.com/olivere/elastic/v7"
 	"main/config"
 	"main/service"
+	"main/service/monitor"
 	"main/service/parse"
 	"strconv"
 	"sync"
@@ -21,7 +22,7 @@ type section struct {
 
 
 func BulkPushRun(esConfig service.EsConfig, name string,
-	conn config.Content, channel int,) error {
+	conn config.Content, channel int, configName string) error {
 
 	// 最小 最大 启动数 计算区间
 	var max,min int
@@ -38,11 +39,31 @@ func BulkPushRun(esConfig service.EsConfig, name string,
 
 	channelData := generate(max, min, channel)
 
+
+	channelWorkNumbers := (max-min)/conn.Writer.Parameter.BatchSize
+	monitor.ProgressBars[configName] = monitor.ProgressBar{Total: channelWorkNumbers, Progress: &sync.Map{}}
+	monitor.ProgressBars[configName].Progress.Store("number", -1)
+
+	//fmt.Printf("max:%s, min:%s, = : %s \n", max, min, channelWorkNumbers)
+
+
+
 	var wg = sync.WaitGroup{}
 
+
+/*	go func() {
+		for true {
+			a,_ := monitor.ProgressBars[name].Progress.Load("number")
+			fmt.Printf("【%s】total:%s, ing:%s \n",name,monitor.ProgressBars[name].Total, a)
+			time.Sleep(time.Second*10)
+		}
+
+	}()*/
+
 	for _,v := range channelData{
 		wg.Add(1)
-		go workProcess(v.Max, v.Min, conn, esConfig, name, &wg)
+		//fmt.Printf("最小:%s, 最大:%s \n",v.Min, v.Max)
+		go workProcess(v.Max, v.Min, conn, esConfig, name, &wg, configName)
 	}
 
 	wg.Wait()
@@ -59,7 +80,7 @@ func generate(max int, min int, channel int) map[int]section {
 	channelPip := make(map[int]section)
 	extent :=  (max - min) / channel
 
-	for i:=0; i <= channel; i++ {
+	for i:=1; i <= channel; i++ {
 		channelPip[i] = section{min,min + extent}
 		min = min + extent
 	}
@@ -72,6 +93,7 @@ func generate(max int, min int, channel int) map[int]section {
 func workProcess (max int , min int, conn config.Content,
 	esConfig service.EsConfig, name string,
 	wg *sync.WaitGroup,
+	configName string,
 	) (error) {
 
 	client := service.NewEsObj(esConfig)
@@ -102,10 +124,14 @@ func workProcess (max int , min int, conn config.Content,
 		return error
 	}
 
-	for i:=min ; i <= max; i = i + conn.Writer.Parameter.BatchSize {
+	for i:=min; i <= max; i = i + conn.Writer.Parameter.BatchSize {
+
+		a,_ := monitor.ProgressBars[configName].Progress.Load("number")
+		monitor.ProgressBars[configName].Progress.Store("number", a.(int)+1)
 
 		temp := conn.Writer.Parameter.BatchSize + i
 
+
 		if temp > max {
 			temp = max
 		}

webGUI/controllers/IndexController.go

@@ -2,15 +2,22 @@ package controllers
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"github.com/gin-gonic/gin"
+	"github.com/gorilla/websocket"
+	"log"
 	"main/config"
 	"main/service"
 	"main/service/errno"
+	"main/service/monitor"
 	"main/service/push"
+	"net/http"
+	"os"
+	"os/exec"
+	"time"
 )
 
-
 func Index(c *gin.Context)  {
 
 	configFile := config.NewConfig()
@@ -45,7 +52,7 @@ func Index(c *gin.Context)  {
 	}
 
 	c.HTML(200, "index.tmpl", gin.H{
-		"title" : "包图网es推送service",
+		"title" : configFile.Title,
 		"jobList" : job,
 	})
 }
@@ -121,6 +128,7 @@ func Push(c *gin.Context)  {
 			synchronousConfig.Job.Content.Writer.Parameter.Index+ "_a",
 			synchronousConfig.Job.Content,
 			synchronousConfig.Job.Setting.Speed.Channel,
+			name,
 		)
 
 		if db_error != nil {
@@ -167,6 +175,7 @@ func Push(c *gin.Context)  {
 			new_index_suffix,
 			synchronousConfig.Job.Content,
 			synchronousConfig.Job.Setting.Speed.Channel,
+			name,
 			)
 
 		if db_error != nil {
@@ -188,7 +197,8 @@ func Push(c *gin.Context)  {
 
 		client.DeleteIndex(now_index_suffix).Do(ctx)
 
-
+		monitor.ProgressBars[name].Progress.Delete("number")
+		delete(monitor.ProgressBars, name)
 
 		c.JSON(200, gin.H{
 			"code": 200,
@@ -197,3 +207,85 @@ func Push(c *gin.Context)  {
 	}
 
 }
+
+func Progress(c *gin.Context)  {
+
+	var upgrader = websocket.Upgrader{
+		ReadBufferSize:  1024,
+		WriteBufferSize: 1024,
+		CheckOrigin: func(r *http.Request) bool {
+			return true
+		},
+	}
+
+	conn, err := upgrader.Upgrade(c.Writer, c.Request, nil) // 升级协议
+
+	defer conn.Close()
+
+	if err != nil {
+		log.Println(err)
+		return
+	}
+
+	for {
+		messageType, message, err := conn.ReadMessage()
+
+		if err != nil {
+			log.Println(err)
+			return
+		}
+
+		var  progressBarJson  []byte
+		var  IsClose = false
+		for {
+			progress,ok := monitor.ProgressBars[string(message[:])]
+
+			if ok == true {
+				progress,_ := progress.Progress.Load("number")
+
+				progressBarJson,_ = json.Marshal(monitor.ProgressBarJson{
+					Total: monitor.ProgressBars[string(message[:])].Total,
+					Progress: progress.(int),
+					Name: string(message[:]),
+					Status: 101,
+				})
+
+			} else {
+
+				progressBarJson,_ = json.Marshal(monitor.ProgressBarJson{
+					Name: string(message[:]),
+					Status: 200,
+				})
+
+				IsClose = true
+
+			}
+
+			if err := conn.WriteMessage(messageType, progressBarJson); err != nil {
+				log.Println(err)
+				return
+			}
+
+			if IsClose {
+				conn.Close()
+			}
+
+
+			time.Sleep(time.Second*1)
+		}
+
+
+	}
+
+}
+
+func Restart(c *gin.Context)  {
+
+	//todo 重新启动还要找其他的解决方法
+	str,_ :=os.Getwd()
+	fmt.Println(str)
+	s := exec.Command("bash", "-c", "kill `lsof -t -i:9100` && " + str +"/main")
+	output, _ := s.CombinedOutput()
+	fmt.Println(string(output))
+
+}