all job status

Author: wyxxxcat

pkg/service/http_service.go

@@ -10,6 +10,7 @@ import (
 	"strconv"
 	"strings"
 
+	"github.com/olekukonko/tablewriter"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"github.com/selectdb/ccr_syncer/pkg/ccr"
 	"github.com/selectdb/ccr_syncer/pkg/ccr/base"
@@ -106,6 +107,51 @@ func (s *HttpService) versionHandler(w http.ResponseWriter, r *http.Request) {
 	writeJson(w, result)
 }
 
+func writeTable(w http.ResponseWriter, data interface{}) {
+	if data == nil || (reflect.ValueOf(data).Kind() == reflect.Ptr && reflect.ValueOf(data).IsNil()) {
+		return
+	}
+
+	jsonData, err := json.Marshal(data)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+
+	var jsonMap map[string]interface{}
+	if err := json.Unmarshal(jsonData, &jsonMap); err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+
+	table := tablewriter.NewWriter(w)
+	table.SetHeader([]string{"Key", "Value"})
+	table.SetBorder(true)
+	table.SetCenterSeparator("+")
+	table.SetColumnSeparator("|")
+	table.SetRowSeparator("-")
+	table.SetAlignment(tablewriter.ALIGN_LEFT)
+	table.SetAutoWrapText(false)
+	table.SetAutoFormatHeaders(true)
+	table.SetHeaderAlignment(tablewriter.ALIGN_LEFT)
+	table.SetAlignment(tablewriter.ALIGN_LEFT)
+	table.SetHeaderLine(true)
+	table.SetRowLine(true)
+
+	for key, value := range jsonMap {
+		switch v := value.(type) {
+		case []interface{}:
+			for i, item := range v {
+				table.Append([]string{fmt.Sprintf("%s[%d]", key, i), fmt.Sprintf("%v", item)})
+			}
+		default:
+			table.Append([]string{key, fmt.Sprintf("%v", v)})
+		}
+	}
+
+	table.Render()
+}
+
 // createCcr creates a new CCR job and adds it to the job manager.
 // It takes a CreateCcrRequest as input and returns an error if there was a problem creating the job or adding it to the job manager.
 func createCcr(request *CreateCcrRequest, db storage.DB, jobManager *ccr.JobManager) error {
@@ -512,6 +558,49 @@ func (s *HttpService) listJobsHandler(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
+// ListJobsStatus service
+func (s *HttpService) listJobsStatusHandler(w http.ResponseWriter, r *http.Request) {
+	log.Infof("list all jobs status")
+
+	type result struct {
+		*defaultResult
+		Jobs      []string         `json:"jobs,omitempty"`
+		JobStatus []*ccr.JobStatus `json:"status,omitempty"`
+	}
+
+	var allJobStatus *result
+	defer func() { writeTable(w, allJobStatus) }()
+
+	// use GetAllData to get all jobs
+	if ans, err := s.db.GetAllData(); err != nil {
+		log.Warnf("when list jobs, get all data failed: %+v", err)
+
+		allJobStatus = &result{
+			defaultResult: newErrorResult(err.Error()),
+		}
+	} else {
+		var jobData []string
+		jobData = ans["jobs"]
+		allJobs := make([]string, 0)
+		alljobStatus := make([]*ccr.JobStatus, 0)
+		for _, eachJob := range jobData {
+			jobName := strings.Trim(strings.Split(eachJob, ",")[0], " ")
+			jobStatus, err := s.jobManager.GetJobStatus(jobName)
+			allJobs = append(allJobs, jobName)
+			if err != nil {
+				log.Warnf("when list all jobs, get job status failed: %+v", err)
+			}
+			alljobStatus = append(alljobStatus, jobStatus)
+		}
+
+		allJobStatus = &result{
+			defaultResult: newSuccessResult(),
+			Jobs:          allJobs,
+			JobStatus:     alljobStatus,
+		}
+	}
+}
+
 // get job progress
 func (s *HttpService) jobProgressHandler(w http.ResponseWriter, r *http.Request) {
 	log.Infof("get job progress")
@@ -688,6 +777,7 @@ func (s *HttpService) RegisterHandlers() {
 	s.mux.HandleFunc("/resume", s.resumeHandler)
 	s.mux.HandleFunc("/delete", s.deleteHandler)
 	s.mux.HandleFunc("/job_status", s.statusHandler)
+	s.mux.HandleFunc("/all_job_status", s.listJobsStatusHandler)
 	s.mux.HandleFunc("/desync", s.desyncHandler)
 	s.mux.HandleFunc("/update_job", s.updateJobHandler)
 	s.mux.HandleFunc("/list_jobs", s.listJobsHandler)