实现log文件内容传输

go.mod

@@ -9,7 +9,7 @@ require (
 	github.com/containerd/cgroups/v3 v3.0.1
 	github.com/dustin/go-humanize v1.0.1
 	github.com/gorilla/websocket v1.5.0
-	github.com/jkstack/anet v0.0.0-20230206080421-e4bd2e304c2e
+	github.com/jkstack/anet v0.0.0-20230207071751-5a225913e4b2
 	github.com/jkstack/jkframe v1.3.0
 	github.com/kardianos/service v1.2.1
 	github.com/opencontainers/runtime-spec v1.0.2
@@ -22,6 +22,7 @@ require (
 	github.com/docker/go-units v0.5.0 // indirect
 	github.com/go-ole/go-ole v1.2.6 // indirect
 	github.com/godbus/dbus/v5 v5.1.0 // indirect
+	github.com/klauspost/compress v1.15.11 // indirect
 	github.com/lufia/plan9stats v0.0.0-20230110061619-bbe2e5e100de // indirect
 	github.com/power-devops/perfstat v0.0.0-20221212215047-62379fc7944b // indirect
 	github.com/sirupsen/logrus v1.9.0 // indirect

go.sum

@@ -26,10 +26,12 @@ github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38=
 github.com/google/go-cmp v0.5.9/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
 github.com/gorilla/websocket v1.5.0 h1:PPwGk2jz7EePpoHN/+ClbZu8SPxiqlu12wZP/3sWmnc=
 github.com/gorilla/websocket v1.5.0/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
-github.com/jkstack/anet v0.0.0-20230206080421-e4bd2e304c2e h1:wMQuhQkz1v2xmhTi1TPCYqbws3dLL5otVQY59Zqd490=
-github.com/jkstack/anet v0.0.0-20230206080421-e4bd2e304c2e/go.mod h1:L9J4lHwDxTsCPX5MNaVa/0UG36JIiDWIJxlp3jmCQxQ=
+github.com/jkstack/anet v0.0.0-20230207071751-5a225913e4b2 h1:3x4xZ9Ur9bu21wWJaAsdx0zq3BUS4o5pj1hp2lL0zo4=
+github.com/jkstack/anet v0.0.0-20230207071751-5a225913e4b2/go.mod h1:L9J4lHwDxTsCPX5MNaVa/0UG36JIiDWIJxlp3jmCQxQ=
 github.com/jkstack/jkframe v1.3.0 h1:BV9PTUMJ1fbqYoRjdDQFk2kp3akgFBczc30rxbMr7Qg=
 github.com/jkstack/jkframe v1.3.0/go.mod h1:9f/mU26CFA4eBx/Zp8EZtuZxTS+4zokiO7yhuMl2zTA=
+github.com/klauspost/compress v1.15.11 h1:Lcadnb3RKGin4FYM/orgq0qde+nc15E5Cbqg4B9Sx9c=
+github.com/klauspost/compress v1.15.11/go.mod h1:QPwzmACJjUTFsnSHH934V6woptycfrDDJnH7hvFVbGM=
 github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0/go.mod h1:zJYVVT2jmtg6P3p1VtQj7WsuWi/y4VnjVBn7F8KPB3I=

system_handler.go

@@ -1,18 +1,27 @@
 package agent
 
 import (
+	"archive/zip"
+	"crypto/md5"
+	"io"
 	"os"
 	"path/filepath"
 
 	"github.com/jkstack/anet"
+	"github.com/jkstack/jkframe/compress"
 	"github.com/jkstack/jkframe/logging"
 )
 
+const blockSize = 32_000
+
 func (app *app) handleSystemPacket(msg *anet.Msg) bool {
 	switch msg.Type {
 	case anet.TypeLogLsReq:
 		app.handleLogLs(msg.TaskID)
 		return false
+	case anet.TypeLogDownloadReq:
+		go app.downloadLogs(msg.TaskID, msg.LogDownload.Files)
+		return false
 	}
 	return true
 }
@@ -39,3 +48,124 @@ func (app *app) handleLogLs(id string) {
 	}
 	app.chWrite <- &msg
 }
+
+func (app *app) downloadLogs(id string, files []string) {
+	tmpDir, err := zipFiles(app.a.Configure().Log.Dir, files)
+	if len(tmpDir) != 0 {
+		defer os.Remove(tmpDir)
+	}
+	if err != nil {
+		app.downloadLogFailed(id, err.Error())
+		return
+	}
+	app.downloadLogOK(id, tmpDir)
+}
+
+func zipFiles(dir string, files []string) (string, error) {
+	f, err := os.CreateTemp(os.TempDir(), "log")
+	if err != nil {
+		return "", err
+	}
+	defer f.Close()
+	zw := zip.NewWriter(f)
+	defer zw.Close()
+	for _, file := range files {
+		src, err := os.Open(filepath.Join(dir, file))
+		if err != nil {
+			return "", err
+		}
+		defer src.Close()
+		fi, err := src.Stat()
+		if err != nil {
+			return "", err
+		}
+		hdr, err := zip.FileInfoHeader(fi)
+		if err != nil {
+			return "", err
+		}
+		dst, err := zw.CreateHeader(hdr)
+		if err != nil {
+			return "", err
+		}
+		_, err = io.Copy(dst, src)
+		if err != nil {
+			return "", err
+		}
+	}
+	return f.Name(), nil
+}
+
+func (app *app) downloadLogFailed(id, msg string) {
+	var m anet.Msg
+	m.Type = anet.TypeLogDownloadInfo
+	m.TaskID = id
+	m.LogDownloadInfo = &anet.LogDownloadInfo{
+		OK:     false,
+		ErrMsg: msg,
+	}
+	app.chWrite <- &m
+}
+
+func (app *app) downloadLogOK(id, dir string) {
+	err := app.responseLogInfo(id, dir)
+	if err != nil {
+		logging.Error("response log info: %v", err)
+		return
+	}
+	app.responseLogData(id, dir)
+}
+
+func (app *app) responseLogInfo(id, dir string) error {
+	f, err := os.Open(dir)
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+	fi, err := f.Stat()
+	if err != nil {
+		return err
+	}
+	enc := md5.New()
+	_, err = io.Copy(enc, f)
+	if err != nil {
+		return err
+	}
+	var sum [md5.Size]byte
+	copy(sum[:], enc.Sum(nil))
+	var msg anet.Msg
+	msg.Type = anet.TypeLogDownloadInfo
+	msg.TaskID = id
+	msg.LogDownloadInfo = &anet.LogDownloadInfo{
+		OK:        true,
+		Size:      uint64(fi.Size()),
+		BlockSize: blockSize,
+		MD5:       sum,
+	}
+	app.chWrite <- &msg
+	return nil
+}
+
+func (app *app) responseLogData(id, dir string) {
+	f, err := os.Open(dir)
+	if err != nil {
+		logging.Error("open file: %v", err)
+		return
+	}
+	buf := make([]byte, blockSize)
+	var msg anet.Msg
+	msg.Type = anet.TypeLogDownloadData
+	msg.TaskID = id
+	var offset uint64
+	for {
+		n, err := f.Read(buf)
+		if err != nil {
+			logging.Error("read file: %v", err)
+			return
+		}
+		msg.LogDownloadData = &anet.LogDownloadData{
+			Offset: offset,
+			Data:   compress.Compress(buf[:n]),
+		}
+		app.chWrite <- &msg
+	}
+}