houston/client/operation.go

package client

import (
	"archive/zip"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"os"
	"os/exec"
	"path"
	"path/filepath"
	"regexp"
	"runtime/debug"
	"strings"
	"syscall"
	"time"

	"repositories.action2quare.com/ayo/gocommon/logger"
	"repositories.action2quare.com/ayo/houston/shared"
	"repositories.action2quare.com/ayo/houston/shared/protos"
)

func lastExecutionArgs(verpath string) []string {
	argf, err := os.Open(path.Join(verpath, "@args"))
	if os.IsNotExist(err) {
		argf, err = os.Open(path.Clean(path.Join(verpath, "..", "@args")))
		if os.IsNotExist(err) {
			return nil
		}
	}
	defer argf.Close()

	var out []string
	dec := json.NewDecoder(argf)
	dec.Decode(&out)
	return out
}

var errUploadZipLogFailed = errors.New("not ok")

func (hc *houstonClient) uploadZipLogFile(zipFile string, name string, version string) error {
	zf, err := os.Open(zipFile)
	if err != nil {
		return err
	}

	if zf == nil {
		return errUploadZipLogFailed
	}

	defer zf.Close()

	req, err := http.NewRequest("POST", hc.httpAddr+"/upload", zf)
	if err != nil {
		logger.Println(err)
	}
	req.Header.Set("Houston-Service-Name", name)
	req.Header.Set("Houston-Service-Version", version)
	req.Header.Set("Houston-Service-Filename", path.Base(zipFile))
	req.Header.Set("Content-Type", "application/zip")
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		return errUploadZipLogFailed
	}

	return nil
}

func zipLogFiles(req *shared.UploadRequest, start, except string) (string, []string, error) {
	root := path.Join(req.Name, req.Version)
	matches, err := filepath.Glob(path.Join(root, req.Filter))
	if err != nil {
		return "", nil, err
	}

	if len(matches) == 0 {
		return "", nil, nil
	}

	for i, file := range matches {
		file = filepath.ToSlash(file)
		matches[i] = file
	}

	root = path.Join(root, path.Dir(req.Filter))
	zipFileName := path.Join(os.TempDir(), path.Base(matches[0])) + ".zip"
	f, err := os.OpenFile(zipFileName, os.O_RDWR|os.O_CREATE|os.O_EXCL, 0600)

	if err != nil {
		return "", nil, err
	}
	defer f.Close()

	w := zip.NewWriter(f)
	defer w.Close()

	oldestFile := ""
	for _, file := range matches {
		if file == root {
			continue
		}
		if len(except) > 0 && file >= except {
			break
		}
		if len(start) > 0 && file < start {
			continue
		}
		if len(oldestFile) == 0 {
			oldestFile = path.Base(file)
		}

		relative := file[len(root)+1:]
		fw, err := w.Create(relative)
		if err != nil {
			logger.Println(err)
			return "", nil, err
		}

		src, err := os.Open(file)
		if err != nil {
			logger.Println(err)
			return "", nil, err
		}
		defer src.Close()

		if _, err = io.Copy(fw, src); err != nil {
			logger.Println(err)
			return "", nil, err
		}
	}

	return f.Name(), matches, nil
	// defer func() {
	// 	tempname := f.Name()
	// 	f.Close()

	// 	resp, _ := http.Post(req.Url, "application/zip", f)
	// 	if resp != nil && resp.Body != nil {
	// 		resp.Body.Close()
	// 	}
	// 	os.Remove(tempname)
	// 	if del, err := strconv.ParseBool(req.DeleteAfterUploaded); del && err == nil {
	// 		for _, file := range matches {
	// 			if strings.HasSuffix(file, except) {
	// 				continue
	// 			}
	// 			os.Remove(file)
	// 		}
	// 	}
	// }()

	// Create a new zip archive.

	//}(f)

	//return nil
}

func prepareProcessLaunch(req *shared.StartProcessRequest) *procmeta {
	re := regexp.MustCompile(`[^\s"']+|"([^"]*)"|'([^']*)`)
	args := re.FindAllString(req.Args, -1)

	if len(args) == 0 {
		return nil
	}
	verpath := path.Join(req.Name, req.Version)
	fi, err := os.Stat(verpath)

	if err == nil && fi.IsDir() {
		args[0] = "./" + path.Clean(strings.TrimPrefix(args[0], "/"))
		os.Chmod(path.Join(verpath, args[0]), 0777)

		cmd := exec.Command(args[0], args[1:]...)
		cmd.Dir = verpath
		stdin, _ := cmd.StdinPipe()

		return &procmeta{
			cmd:           cmd,
			name:          req.Name,
			version:       req.Version,
			state:         protos.ProcessState_Stopped,
			stdin:         stdin,
			logUploadChan: make(chan *shared.UploadRequest),
			buffers:       bufferStack{cursor: 0},
		}
	}
	return nil
}

func (hc *houstonClient) launch(meta *procmeta) error {
	stdout, err := meta.cmd.StdoutPipe()
	if err != nil {
		return err
	}
	stderr, err := meta.cmd.StderrPipe()
	if err != nil {
		return err
	}

	err = os.MkdirAll(path.Join(meta.cmd.Dir, "logs"), 0775)
	if err != nil {
		return err
	}

	relayChan := make(chan struct {
		size int
		buf  []byte
	})

	go func() {
		defer func() {
			r := recover()
			if r != nil {
				logger.Println(r)
				debug.PrintStack()
			}
			close(relayChan)
			hc.exitChan <- meta.cmd
		}()

		now := time.Now().UTC()
		ext := path.Ext(meta.cmd.Args[0])
		nameonly := path.Base(meta.cmd.Args[0])
		if len(ext) > 0 {
			nameonly = nameonly[:len(nameonly)-len(ext)]
		}
		ts := now.Format("2006-01-02T15-04-05")
		stdPrefix := path.Join(meta.cmd.Dir, "logs", fmt.Sprintf("%s_%s", nameonly, ts))
		logfile, _ := os.Create(stdPrefix + "_0.log")
		defer logfile.Close()

		logfileIdx := 0
		for {
			thisFileSize := 0
			switchToNextFile := func() string {
				logfileIdx++
				nextFile := fmt.Sprintf("%s_%d.log", stdPrefix, logfileIdx)
				if nextLogfile, err := os.Create(nextFile); err == nil {
					logfile.Close()
					logfile = nextLogfile
				}
				thisFileSize = 0
				return nextFile
			}

			uploadStartFile := ""
			select {
			case req := <-meta.logUploadChan:
				nextFile := switchToNextFile()
				startFile := uploadStartFile
				uploadStartFile = nextFile
				go func(startFile, nextFile string) {
					zipFile, srcFiles, err := zipLogFiles(req, startFile, nextFile)
					if err == nil && len(zipFile) > 0 && len(srcFiles) > 0 {
						if err = hc.uploadZipLogFile(zipFile, meta.name, meta.version); err == nil {
							for _, oldf := range srcFiles {
								os.Remove(oldf)
							}
						} else {
							logger.Println("uploadZipLogFile failed :", err)
						}
					} else if err != nil {
						logger.Println("zipLogFiles failed :", err)
					}
				}(startFile, nextFile)

			case bt := <-relayChan:
				if bt.buf == nil {
					return
				}
				logfile.Write(bt.buf[:bt.size])
				meta.buffers.push(bt.buf)
				thisFileSize += bt.size
				if thisFileSize > 1024*1024 {
					switchToNextFile()
				}
			}
		}
	}()

	stdReader := func(r io.Reader) {
		defer func() {
			recover()
			stdout.Close()
		}()

		for {
			buff := meta.buffers.pop()
			size, err := r.Read(buff)
			if err != nil {
				relayChan <- struct {
					size int
					buf  []byte
				}{buf: nil}
				break
			}
			if size > 0 {
				relayChan <- struct {
					size int
					buf  []byte
				}{size: size, buf: buff}
			}
		}
	}

	go stdReader(stderr)
	go stdReader(stdout)

	err = meta.cmd.Start()
	if err == nil {
		meta.state = protos.ProcessState_Running
	}
	return err
}

var errPrepareprocessLaunchFailed = errors.New("prepareProcessLaunch failed")

func (hc *houstonClient) startChildProcess(req *shared.StartProcessRequest) error {
	logger.Println("startChildProcess :", *req)
	if req.Version == "latest" {
		// 최신 버전을 찾음
		latest, err := shared.FindLastestVersion(req.Name)
		if err != nil {
			return err
		}

		req.Version = latest
	}

	meta := prepareProcessLaunch(req)
	if meta == nil {
		return errPrepareprocessLaunchFailed
	}
	if err := hc.launch(meta); err != nil {
		return err
	}

	// launch가 성공하면 args 저장. this and parent folder
	vers := hc.deploys[req.Name]
	for _, ver := range vers {
		if ver.Version == req.Version {
			ver.Args = meta.cmd.Args
		}
	}

	if argfile, err := os.Create(path.Join(req.Name, "@args")); err == nil {
		enc := json.NewEncoder(argfile)
		enc.Encode(meta.cmd.Args)
		argfile.Close()
	}
	if argfile, err := os.Create(path.Join(req.Name, req.Version, "@args")); err == nil {
		enc := json.NewEncoder(argfile)
		enc.Encode(meta.cmd.Args)
		argfile.Close()
	}

	hc.childProcs = append(hc.childProcs, meta)

	op := protos.NewOperationClient(hc.client)
	op.Refresh(context.Background(), hc.makeOperationQueryRequest())

	return nil
}

var errNoRunningProcess = errors.New("no running processed")

func (hc *houstonClient) stopChildProcess(req *shared.StopProcessRequest) error {
	if req.Version == "latest" {
		// 최신 버전을 찾음
		latest, err := shared.FindLastestVersion(req.Name)
		if err != nil {
			return err
		}

		req.Version = latest
	}

	var remains []*procmeta
	var killing []*procmeta
	for _, proc := range hc.childProcs {
		if proc.state != protos.ProcessState_Running {
			continue
		}

		if req.Pid != 0 {
			if req.Pid == int32(proc.cmd.Process.Pid) {
				// 해당 pid만 제거
				killing = append(killing, proc)
			} else {
				remains = append(remains, proc)
			}
		} else if proc.name == req.Name {
			if len(req.Version) == 0 {
				// program 다 정지
				killing = append(killing, proc)
			} else if req.Version == proc.version {
				// program의 특정 버전만 정지
				killing = append(killing, proc)
			} else {
				// 해당 사항 없음
				remains = append(remains, proc)
			}
		} else {
			// 해당 사항 없음
			remains = append(remains, proc)
		}
	}

	if len(killing) > 0 {
		for _, proc := range killing {
			if err := proc.cmd.Process.Signal(syscall.SIGTERM); err != nil {
				proc.cmd.Process.Signal(os.Kill)
				proc.state = protos.ProcessState_Stopping
			}
		}

		op := protos.NewOperationClient(hc.client)
		op.Refresh(context.Background(), hc.makeOperationQueryRequest())

		for _, proc := range killing {
			proc.cmd.Wait()
		}

		hc.childProcs = remains

		op.Refresh(context.Background(), hc.makeOperationQueryRequest())

		return nil
	}

	return errNoRunningProcess
}

func (hc *houstonClient) restartChildProcess(req *shared.RestartProcessRequest) error {
	if req.Version == "latest" {
		// 최신 버전을 찾음
		latest, err := shared.FindLastestVersion(req.Name)
		if err != nil {
			return err
		}

		req.Version = latest
	}

	var restarts []*procmeta
	for _, proc := range hc.childProcs {
		if proc.name == req.Name {
			if len(req.Version) == 0 {
				restarts = append(restarts, proc)
			} else if req.Version == proc.version {
				restarts = append(restarts, proc)
			}
		}
	}

	if len(restarts) == 0 {
		return errNoRunningProcess
	}

	for _, proc := range restarts {
		if err := proc.cmd.Process.Signal(syscall.SIGTERM); err != nil {
			proc.cmd.Process.Signal(os.Kill)
		}
		proc.state = protos.ProcessState_Stopping
	}

	op := protos.NewOperationClient(hc.client)
	op.Refresh(context.Background(), hc.makeOperationQueryRequest())

	for _, proc := range restarts {
		proc.cmd.Wait()
		proc.state = protos.ProcessState_Stopped
	}
	op.Refresh(context.Background(), hc.makeOperationQueryRequest())

	for _, proc := range restarts {
		if err := hc.launch(proc); err != nil {
			return err
		}
	}
	op.Refresh(context.Background(), hc.makeOperationQueryRequest())

	return nil
}

func (hc *houstonClient) uploadFiles(req *shared.UploadRequest) error {
	if req.Version == "latest" {
		// 최신 버전을 찾음
		latest, err := shared.FindLastestVersion(req.Name)
		if err != nil {
			return err
		}

		req.Version = latest
	}

	logger.Println("uploadFiles req :", *req)
	for _, child := range hc.childProcs {
		if child.version == req.Version && child.name == req.Name {
			logger.Println("uploadFiles found :", child.version, child.name)
			child.logUploadChan <- req
			return nil
		}
	}

	// 실행 중이 아닌 폴더에서도 대상을 찾는다
	// 전체 파일을 대상으로
	zipFile, srcFiles, err := zipLogFiles(req, "", "")
	if err == nil && len(zipFile) > 0 && len(srcFiles) > 0 {
		if err = hc.uploadZipLogFile(zipFile, req.Name, req.Version); err == nil {
			for _, oldf := range srcFiles {
				os.Remove(oldf)
			}
		} else {
			logger.Println("uploadZipLogFile failed :", err)
		}
	} else if err != nil {
		logger.Println("zipLogFiles failed :", err)
	}

	return nil
}