sqlreplay, api: add job parameters to the output of show traffic jobs (#749)

Author: djshow832

pkg/server/api/traffic.go

@@ -40,7 +40,7 @@ func (h *Server) TrafficCapture(c *gin.Context) {
 		}
 		cfg.Duration = duration
 	}
-	cfg.EncryptMethod = c.PostForm("encrypt-method")
+	cfg.EncryptionMethod = c.PostForm("encrypt-method")
 
 	compress := true
 	if compressStr := c.PostForm("compress"); compressStr != "" {

pkg/server/api/traffic_test.go

@@ -74,7 +74,7 @@ func TestTraffic(t *testing.T) {
 		require.NoError(t, err)
 		require.Equal(t, "capture started", string(all))
 		require.Equal(t, "capture", mgr.curJob)
-		require.Equal(t, capture.CaptureConfig{Duration: time.Hour, Output: "/tmp", EncryptMethod: "aes256-ctr", Compress: false,
+		require.Equal(t, capture.CaptureConfig{Duration: time.Hour, Output: "/tmp", EncryptionMethod: "aes256-ctr", Compress: false,
 			StartTime: mgr.captureCfg.StartTime}, mgr.captureCfg)
 	})
 	// job is running error

pkg/sqlreplay/capture/capture.go

@@ -53,9 +53,9 @@ type Capture interface {
 }
 
 type CaptureConfig struct {
-	Output        string
-	EncryptMethod string
-	KeyFile       string
+	Output           string
+	EncryptionMethod string
+	KeyFile          string
 	// It's specified when executing with the statement `TRAFFIC CAPTURE` so that all TiProxy instances
 	// use the same start time and the time acts as the job ID.
 	StartTime          time.Time
@@ -245,7 +245,7 @@ func (c *capture) flushBuffer(bufCh <-chan *bytes.Buffer) {
 		var err error
 		cmdLogger, err = store.NewWriter(c.lg.Named("writer"), c.storage, store.WriterCfg{
 			Dir:           c.cfg.Output,
-			EncryptMethod: c.cfg.EncryptMethod,
+			EncryptMethod: c.cfg.EncryptionMethod,
 			KeyFile:       c.cfg.KeyFile,
 			Compress:      c.cfg.Compress,
 		})
@@ -370,7 +370,7 @@ func (c *capture) putCommand(command *cmd.Command) bool {
 }
 
 func (c *capture) writeMeta(storage storage.ExternalStorage, duration time.Duration, cmds, filteredCmds uint64) {
-	meta := store.NewMeta(duration, cmds, filteredCmds, c.cfg.EncryptMethod)
+	meta := store.NewMeta(duration, cmds, filteredCmds, c.cfg.EncryptionMethod)
 	if err := meta.Write(storage); err != nil {
 		c.lg.Error("failed to write meta", zap.Error(err))
 	}

pkg/sqlreplay/manager/job.go

@@ -38,17 +38,12 @@ type job struct {
 }
 
 type job4Marshal struct {
-	Type      string  `json:"type"`
-	Status    string  `json:"status"`
-	StartTime string  `json:"start_time"`
-	EndTime   string  `json:"end_time,omitempty"`
-	Duration  string  `json:"duration,omitempty"`
-	Output    string  `json:"output,omitempty"`
-	Input     string  `json:"input,omitempty"`
-	Username  string  `json:"username,omitempty"`
-	Speed     float64 `json:"speed,omitempty"`
-	Progress  string  `json:"progress"`
-	Err       string  `json:"error,omitempty"`
+	Type      string `json:"type"`
+	Status    string `json:"status"`
+	StartTime string `json:"start_time"`
+	EndTime   string `json:"end_time,omitempty"`
+	Progress  string `json:"progress"`
+	Err       string `json:"error,omitempty"`
 }
 
 func (job *job) IsRunning() bool {
@@ -90,16 +85,29 @@ type captureJob struct {
 	cfg capture.CaptureConfig
 }
 
+type captureJob4Marshal struct {
+	job4Marshal
+	Output           string `json:"output,omitempty"`
+	Duration         string `json:"duration,omitempty"`
+	Compress         bool   `json:"compress,omitempty"`
+	EncryptionMethod string `json:"encryption-method,omitempty"`
+}
+
 func (job *captureJob) Type() JobType {
 	return Capture
 }
 
 func (job *captureJob) MarshalJSON() ([]byte, error) {
 	job4Marshal := job.getJob4Marshal()
 	job4Marshal.Type = "capture"
-	job4Marshal.Output = ast.RedactURL(job.cfg.Output)
-	job4Marshal.Duration = job.cfg.Duration.String()
-	return json.Marshal(job4Marshal)
+	c := captureJob4Marshal{
+		job4Marshal:      *job4Marshal,
+		Output:           ast.RedactURL(job.cfg.Output),
+		Duration:         job.cfg.Duration.String(),
+		Compress:         job.cfg.Compress,
+		EncryptionMethod: job.cfg.EncryptionMethod,
+	}
+	return json.Marshal(c)
 }
 
 func (job *captureJob) String() string {
@@ -117,20 +125,29 @@ type replayJob struct {
 	cfg replay.ReplayConfig
 }
 
+type replayJob4Marshal struct {
+	job4Marshal
+	Input    string  `json:"input,omitempty"`
+	Username string  `json:"username,omitempty"`
+	Speed    float64 `json:"speed,omitempty"`
+	ReadOnly bool    `json:"readonly,omitempty"`
+}
+
 func (job *replayJob) Type() JobType {
 	return Replay
 }
 
 func (job *replayJob) MarshalJSON() ([]byte, error) {
 	job4Marshal := job.getJob4Marshal()
 	job4Marshal.Type = "replay"
-	job4Marshal.Input = ast.RedactURL(job.cfg.Input)
-	job4Marshal.Username = job.cfg.Username
-	job4Marshal.Speed = job.cfg.Speed
-	if job4Marshal.Speed == 0 {
-		job4Marshal.Speed = 1
+	r := replayJob4Marshal{
+		job4Marshal: *job4Marshal,
+		Input:       ast.RedactURL(job.cfg.Input),
+		Username:    job.cfg.Username,
+		Speed:       job.cfg.Speed,
+		ReadOnly:    job.cfg.ReadOnly,
 	}
-	return json.Marshal(job4Marshal)
+	return json.Marshal(r)
 }
 
 func (job *replayJob) String() string {

pkg/sqlreplay/manager/job_test.go

@@ -127,7 +127,7 @@ func TestMarshalJob(t *testing.T) {
 					Duration: 2 * time.Hour,
 				},
 			},
-			marshal: `{"type":"capture","status":"canceled","start_time":"2020-01-01T00:00:00Z","end_time":"2020-01-01T02:01:01Z","duration":"2h0m0s","output":"/tmp/traffic","progress":"50%","error":"mock error"}`,
+			marshal: `{"type":"capture","status":"canceled","start_time":"2020-01-01T00:00:00Z","end_time":"2020-01-01T02:01:01Z","progress":"50%","error":"mock error","output":"/tmp/traffic","duration":"2h0m0s"}`,
 		},
 		{
 			job: &captureJob{
@@ -138,11 +138,13 @@ func TestMarshalJob(t *testing.T) {
 					done:      true,
 				},
 				cfg: capture.CaptureConfig{
-					Output:   "s3://bucket/prefix?access-key=abcdefghi&secret-access-key=123&force-path-style=true",
-					Duration: 2 * time.Hour,
+					Output:           "s3://bucket/prefix?access-key=abcdefghi&secret-access-key=123&force-path-style=true",
+					Duration:         2 * time.Hour,
+					Compress:         true,
+					EncryptionMethod: "aes256-ctr",
 				},
 			},
-			marshal: `{"type":"capture","status":"done","start_time":"2020-01-01T00:00:00Z","end_time":"2020-01-01T02:01:01Z","duration":"2h0m0s","output":"s3://bucket/prefix?access-key=xxxxxx\u0026force-path-style=true\u0026secret-access-key=xxxxxx","progress":"50%"}`,
+			marshal: `{"type":"capture","status":"done","start_time":"2020-01-01T00:00:00Z","end_time":"2020-01-01T02:01:01Z","progress":"50%","output":"s3://bucket/prefix?access-key=xxxxxx\u0026force-path-style=true\u0026secret-access-key=xxxxxx","duration":"2h0m0s","compress":true,"encryption-method":"aes256-ctr"}`,
 		},
 		{
 			job: &replayJob{
@@ -155,7 +157,7 @@ func TestMarshalJob(t *testing.T) {
 					Username: "root",
 				},
 			},
-			marshal: `{"type":"replay","status":"running","start_time":"2020-01-01T00:00:00Z","input":"/tmp/traffic","username":"root","speed":1,"progress":"0%"}`,
+			marshal: `{"type":"replay","status":"running","start_time":"2020-01-01T00:00:00Z","progress":"0%","input":"/tmp/traffic","username":"root"}`,
 		},
 		{
 			job: &replayJob{
@@ -168,10 +170,12 @@ func TestMarshalJob(t *testing.T) {
 				cfg: replay.ReplayConfig{
 					Input:    "/tmp/traffic",
 					Username: "root",
+					Password: "123456",
 					Speed:    0.5,
+					ReadOnly: true,
 				},
 			},
-			marshal: `{"type":"replay","status":"done","start_time":"2020-01-01T00:00:00Z","end_time":"2020-01-01T02:01:01Z","input":"/tmp/traffic","username":"root","speed":0.5,"progress":"100%"}`,
+			marshal: `{"type":"replay","status":"done","start_time":"2020-01-01T00:00:00Z","end_time":"2020-01-01T02:01:01Z","progress":"100%","input":"/tmp/traffic","username":"root","speed":0.5,"readonly":true}`,
 		},
 		{
 			job: &replayJob{
@@ -184,7 +188,7 @@ func TestMarshalJob(t *testing.T) {
 					Username: "root",
 				},
 			},
-			marshal: `{"type":"replay","status":"running","start_time":"2020-01-01T00:00:00Z","input":"s3://bucket/prefix?access-key=xxxxxx\u0026force-path-style=true\u0026secret-access-key=xxxxxx","username":"root","speed":1,"progress":"0%"}`,
+			marshal: `{"type":"replay","status":"running","start_time":"2020-01-01T00:00:00Z","progress":"0%","input":"s3://bucket/prefix?access-key=xxxxxx\u0026force-path-style=true\u0026secret-access-key=xxxxxx","username":"root"}`,
 		},
 	}
 

pkg/sqlreplay/manager/manager.go

@@ -95,17 +95,17 @@ func (jm *jobManager) StartCapture(cfg capture.CaptureConfig) error {
 	if running != nil {
 		return errors.Errorf("a job is running: %s", running.String())
 	}
-	if err := jm.capture.Start(cfg); err != nil {
-		jm.lg.Warn("start capture failed", zap.Error(err))
-		return errors.Wrapf(err, "start capture failed")
-	}
 	newJob := &captureJob{
 		job: job{
 			// cfg.StartTime may act as the job ID in a TiProxy cluster.
 			startTime: cfg.StartTime,
 		},
 		cfg: cfg,
 	}
+	if err := jm.capture.Start(cfg); err != nil {
+		jm.lg.Warn("start capture failed", zap.String("job", newJob.String()), zap.Error(err))
+		return errors.Wrapf(err, "start capture failed")
+	}
 	jm.lg.Info("start capture", zap.String("job", newJob.String()))
 	jm.addToHistory(newJob)
 	return nil
@@ -116,6 +116,13 @@ func (jm *jobManager) StartReplay(cfg replay.ReplayConfig) error {
 	if running != nil {
 		return errors.Errorf("a job is running: %s", running.String())
 	}
+	newJob := &replayJob{
+		job: job{
+			// cfg.StartTime may act as the job ID in a TiProxy cluster.
+			startTime: cfg.StartTime,
+		},
+		cfg: cfg,
+	}
 	// TODO: support update configs online
 	err := jm.replay.Start(cfg, jm.certManager.SQLTLS(), jm.hsHandler, &backend.BCConfig{
 		ProxyProtocol:      jm.cfg.Proxy.ProxyProtocol != "",
@@ -125,16 +132,9 @@ func (jm *jobManager) StartReplay(cfg replay.ReplayConfig) error {
 		ConnBufferSize:     jm.cfg.Proxy.ConnBufferSize,
 	})
 	if err != nil {
-		jm.lg.Warn("start replay failed", zap.Error(err))
+		jm.lg.Warn("start replay failed", zap.String("job", newJob.String()), zap.Error(err))
 		return errors.Wrapf(err, "start replay failed")
 	}
-	newJob := &replayJob{
-		job: job{
-			// cfg.StartTime may act as the job ID in a TiProxy cluster.
-			startTime: cfg.StartTime,
-		},
-		cfg: cfg,
-	}
 	jm.lg.Info("start replay", zap.String("job", newJob.String()))
 	jm.addToHistory(newJob)
 	return nil

pkg/sqlreplay/manager/manager_test.go

@@ -110,29 +110,28 @@ func TestMarshalJobHistory(t *testing.T) {
     "status": "canceled",
     "start_time": "2020-01-01T00:00:00Z",
     "end_time": "2020-01-01T02:01:01Z",
-    "duration": "2h0m0s",
-    "output": "/tmp/traffic",
     "progress": "50%",
-    "error": "mock error"
+    "error": "mock error",
+    "output": "/tmp/traffic",
+    "duration": "2h0m0s"
   },
   {
     "type": "replay",
     "status": "running",
     "start_time": "2020-01-01T00:00:00Z",
+    "progress": "0%",
     "input": "/tmp/traffic",
-    "username": "root",
-    "speed": 1,
-    "progress": "0%"
+    "username": "root"
   },
   {
     "type": "replay",
     "status": "done",
     "start_time": "2020-01-01T00:00:00Z",
     "end_time": "2020-01-01T02:01:01Z",
+    "progress": "100%",
     "input": "/tmp/traffic",
     "username": "root",
-    "speed": 0.5,
-    "progress": "100%"
+    "speed": 0.5
   }
 ]`, mgr.Jobs())
 }