Merge branch 'master' into kd-live

client/client.go

@@ -15,6 +15,7 @@ import (
 	"reflect"
 	"sort"
 	"strconv"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"syscall"
@@ -116,6 +117,8 @@ type houstonClient struct {
 	timestamp     string
 	wg            sync.WaitGroup
 	config        clientConfig
+	version       string
+	standalone    bool
 }
 
 func unmarshal[T any](val *T, src map[string]string) {
@@ -129,6 +132,9 @@ func unmarshal[T any](val *T, src map[string]string) {
 		if argval.Elem().Field(i).CanInt() {
 			num, _ := strconv.ParseInt(arg, 10, 0)
 			argval.Elem().Field(i).SetInt(num)
+		} else if argval.Elem().Field(i).Kind() == reflect.Array || argval.Elem().Field(i).Kind() == reflect.Slice {
+			conv := strings.Split(arg, "\n")
+			argval.Elem().Field(i).Set(reflect.ValueOf(conv))
 		} else {
 			argval.Elem().Field(i).SetString(arg)
 		}
@@ -158,8 +164,33 @@ func gatherDeployedPrograms(storageRoot, name string) []*protos.VersionAndArgs {
 }
 
 func (hc *houstonClient) makeOperationQueryRequest() *protos.OperationQueryRequest {
-	hn, _ := os.Hostname()
-	procs := make([]*protos.ProcessDescription, 0, len(hc.childProcs))
+	var procs []*protos.ProcessDescription
+	var deploys []*protos.DeployedVersions
+
+	var selfname string
+	var selfargs []string
+	if hc.standalone {
+		selfname = path.Base(os.Args[0])
+		selfargs = os.Args[1:]
+	} else {
+		selfname = "houston"
+		selfargs = []string{}
+	}
+
+	procs = append(procs, &protos.ProcessDescription{
+		Name:    selfname,
+		Args:    selfargs,
+		Version: hc.version,
+		State:   protos.ProcessState_Running,
+		Pid:     int32(os.Getpid()),
+	})
+	deploys = append(deploys, &protos.DeployedVersions{
+		Name: selfname,
+		Versions: []*protos.VersionAndArgs{
+			{Version: hc.version, Args: selfargs},
+		},
+	})
+
 	for _, child := range hc.childProcs {
 		procs = append(procs, &protos.ProcessDescription{
 			Name:    child.name,
@@ -170,7 +201,6 @@ func (hc *houstonClient) makeOperationQueryRequest() *protos.OperationQueryReque
 		})
 	}
 
-	var deploys []*protos.DeployedVersions
 	for name, prog := range hc.deploys {
 		deploys = append(deploys, &protos.DeployedVersions{
 			Name:     name,
@@ -178,6 +208,7 @@ func (hc *houstonClient) makeOperationQueryRequest() *protos.OperationQueryReque
 		})
 	}
 
+	hn, _ := os.Hostname()
 	return &protos.OperationQueryRequest{
 		Hostname: hn,
 		Procs:    procs,
@@ -185,7 +216,7 @@ func (hc *houstonClient) makeOperationQueryRequest() *protos.OperationQueryReque
 	}
 }
 
-func NewClient() (HoustonClient, error) {
+func NewClient(standalone bool) (HoustonClient, error) {
 	clientConfig, err := loadClientConfig()
 	if err != nil {
 		return nil, err
@@ -234,12 +265,19 @@ func NewClient() (HoustonClient, error) {
 		}
 	}
 
+	ver, _ := os.ReadFile("@version")
+	if len(ver) == 0 {
+		ver = []byte("0.0.0")
+	}
+
 	hc := &houstonClient{
 		config:       clientConfig,
 		clientChan:   make(chan *grpc.ClientConn),
 		extraMetrics: unsafe.Pointer(&map[string]float32{}),
 		deploys:      deploys,
 		timestamp:    exefi.ModTime().String(),
+		version:      string(ver),
+		standalone:   standalone,
 	}
 
 	ctx, cancel := context.WithCancel(context.Background())
@@ -415,6 +453,8 @@ func (hc *houstonClient) Start() {
 
 	var client *grpc.ClientConn
 	reconnCount := 0
+	time.Sleep(time.Second)
+
 	for {
 		select {
 		case <-hc.ctx.Done():

client/deploy.go

@@ -264,6 +264,12 @@ func (hc *houstonClient) prepareUpdateSelf(req *shared.DeployRequest) (srcdir st
 		return "", "", err
 	}
 
+	// houston version 파일
+	err = os.WriteFile(path.Join(path.Dir(fname), "@version"), []byte(req.Version), 0644)
+	if err != nil {
+		return "", "", err
+	}
+
 	selfname, _ := os.Executable()
 	srcreplacer := path.Join(path.Dir(fname), "replacer") + path.Ext(selfname)
 	replacer = "./" + filepath.ToSlash("replacer"+path.Ext(selfname))

client/operation.go

@@ -102,11 +102,12 @@ func zipLogFiles(storageRoot string, req *shared.UploadRequest, start, except st
 	defer w.Close()
 
 	oldestFile := ""
-	for _, file := range matches {
+	for i, file := range matches {
 		if file == root {
 			continue
 		}
 		if len(except) > 0 && file >= except {
+			matches = matches[:i]
 			break
 		}
 		if len(start) > 0 && file < start {
@@ -439,7 +440,7 @@ func (hc *houstonClient) restartChildProcess(req *shared.RestartProcessRequest,
 		if proc.cmd.Process.Pid == int(req.Pid) {
 			if len(req.Config) > 0 {
 				// config.json를 먼저 다운로드 시도
-				root := proc.cmd.Path
+				root := proc.cmd.Dir
 				if _, err := download(root, hc.makeDownloadUrl(req.Config), ""); err != nil {
 					return err
 				}
@@ -447,10 +448,7 @@ func (hc *houstonClient) restartChildProcess(req *shared.RestartProcessRequest,
 
 			proc.state = protos.ProcessState_Restart
 			op.Refresh(context.Background(), hc.makeOperationQueryRequest())
-
-			if err := proc.cmd.Process.Signal(syscall.SIGTERM); err != nil {
-				proc.cmd.Process.Signal(os.Kill)
-			}
+			hc.exitChan <- proc.cmd
 
 			break
 		}
@@ -483,11 +481,7 @@ func (hc *houstonClient) uploadFiles(req *shared.UploadRequest) error {
 	// 전체 파일을 대상으로
 	zipFile, srcFiles, err := zipLogFiles(hc.config.StorageRoot, 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 {
+		if err = hc.uploadZipLogFile(zipFile, req.Name, req.Version); err != nil {
 			logger.Println("uploadZipLogFile failed :", err)
 		}
 	} else if err != nil {

houston.sh

@@ -1,4 +1,4 @@
 #!/bin/sh
 
-nohup /home/opdev/houston -client -logfile &
+nohup /home/opdev/houston -client -logfile > /dev/null &
 

main.go

@@ -18,7 +18,7 @@ func main() {
 	}
 
 	if *runAsClient {
-		hc, err := client.NewClient()
+		hc, err := client.NewClient(true)
 		if err != nil {
 			logger.Fatal(err)
 			return

replacer/main.go

@@ -8,15 +8,24 @@ import (
 	"os"
 	"os/exec"
 	"path"
-	"strconv"
 	"time"
 )
 
-func copy(src, dst string) error {
+func copy(src, dst string, stdlog *log.Logger) error {
 	fi, err := os.Stat(src)
 	if err != nil {
 		return err
 	}
+	if fi.IsDir() {
+		entries, _ := os.ReadDir(src)
+		for _, ent := range entries {
+			if err := copy(path.Join(src, ent.Name()), path.Join(dst, ent.Name()), stdlog); err != nil {
+				return err
+			}
+		}
+		return nil
+	}
+
 	inmode := fi.Mode()
 
 	in, err := os.Open(src)
@@ -46,6 +55,7 @@ func copy(src, dst string) error {
 		return err
 	}
 
+	stdlog.Println("file copied :", src, dst)
 	return nil
 }
 
@@ -60,54 +70,34 @@ func main() {
 	// args[3:] : 다시 시작할 때 넘겨줄 arguments(프로세스 이름 포함)
 	stdlog.Println(args)
 
-	pid, err := strconv.Atoi(args[1])
-	if err != nil {
-		stdlog.Fatal(err)
-	}
-
 	for {
 		stdlog.Println("wait for terminating of", args[3])
-		proc, err := os.FindProcess(pid)
-		if err != nil {
-			stdlog.Fatal(err)
-		}
-		state, _ := proc.Wait()
-		if state == nil {
+		cmd := exec.Command("ps", "-p", args[1])
+		if err := cmd.Run(); err != nil {
 			break
 		}
-
 		time.Sleep(time.Second)
 	}
 
-	selfext, _ := os.Executable()
-	selfext = path.Base(selfext)
-	nextArgs := args[4:]
-	entries, _ := os.ReadDir(args[2])
-	for _, ent := range entries {
-		if ent.Name() == selfext {
-			continue
-		}
+	stdlog.Println("target is terminated")
 
-		if ent.IsDir() {
-			if err := os.MkdirAll(ent.Name(), 0775); err != nil {
-				stdlog.Fatal(err)
-			}
-		} else {
-			if ent.Name() == "@args" {
-				var tempArgs []string
-				argfile, _ := os.Open(path.Join(args[2], ent.Name()))
-				dec := json.NewDecoder(argfile)
-				if dec.Decode(&tempArgs) == nil {
-					nextArgs = tempArgs
-				}
-			} else if err := copy(path.Join(args[2], ent.Name()), ent.Name()); err != nil {
-				stdlog.Println("copy failed :", path.Join(args[2], ent.Name()), ent.Name())
-				stdlog.Fatal(err)
-			}
-		}
+	// replacer 제거. 내가 돌고 있으므로 복사는 안된다.
+	// 내가 실행되기 전에 이미 복사가 되서 나는 최신 버전임
+	os.Remove(path.Join(args[2], os.Args[0]))
+	if err := copy(args[2], "", stdlog); err != nil {
+		stdlog.Fatal(err)
 	}
 
-	err = os.RemoveAll(args[2])
+	nextArgs := args[4:]
+	if bt, _ := os.ReadFile("@args"); len(bt) > 0 {
+		var tempArgs []string
+		if json.Unmarshal(bt, &tempArgs) == nil {
+			nextArgs = tempArgs
+		}
+	}
+	os.Remove("@args")
+
+	err := os.RemoveAll(args[2])
 	if err != nil {
 		stdlog.Println("os.RemoveAll failed :", args[2], err)
 	}

server/http_api.go

@@ -34,6 +34,16 @@ func (h *houstonHandler) GetAgents(w http.ResponseWriter, r *http.Request) {
 	enc.Encode(allHosts)
 }
 
+func readTagsFromFile(paths ...string) string {
+	raw, _ := os.ReadFile(path.Join(paths...))
+	if len(raw) > 0 {
+		tag := string(raw)
+		return strings.Trim(tag, "\n")
+	}
+
+	return ""
+}
+
 func (h *houstonHandler) GetDeploySources(w http.ResponseWriter, r *http.Request) {
 	files, err := os.ReadDir(h.deployPath)
 	if err != nil {
@@ -43,19 +53,31 @@ func (h *houstonHandler) GetDeploySources(w http.ResponseWriter, r *http.Request
 	}
 
 	getVersions := func(name string) []string {
-		var out []string
 		vers, _ := os.ReadDir(path.Join(h.deployPath, name))
+		mytags := readTagsFromFile(h.deployPath, name, "@tags")
+		out := []string{
+			mytags,
+		}
 		for _, fd := range vers {
 			if fd.IsDir() {
 				ver := fd.Name()
 				files, _ := os.ReadDir(path.Join(h.deployPath, name, ver))
+				vertags := readTagsFromFile(h.deployPath, name, ver, "@tags")
 				if len(files) > 0 {
-					downloadpath := path.Join(sub_folder_name_deploys, name, ver, files[0].Name())
-					ver = fmt.Sprintf("%s:%s", ver, downloadpath)
+					for _, file := range files {
+						if strings.HasPrefix(file.Name(), "@") {
+							continue
+						}
+
+						downloadpath := path.Join(sub_folder_name_deploys, name, ver, file.Name())
+						ver = fmt.Sprintf("%s:%s", ver+mytags+vertags, downloadpath)
+						break
+					}
 				}
 				out = append(out, ver)
 			}
 		}
+
 		return out
 	}
 
@@ -100,8 +122,12 @@ func (h *houstonHandler) UploadDeploySource(w http.ResponseWriter, r *http.Reque
 
 	if version == "config" {
 		filename = path.Join(h.deployPath, name, version, "config"+ext)
+		tags := readTagsFromFile(h.deployPath, name, version, "@tags")
+		if !strings.Contains(tags, "#hidden") {
+			tags = tags + "#hidden"
+			os.WriteFile(path.Join(h.deployPath, name, version, "@tags"), []byte(tags), 0644)
+		}
 	} else {
-		// deploys 폴더는 파일시스템 서비스이므로 다운로드 가능
 		filename = path.Join(h.deployPath, name, version, name+ext)
 	}
 
@@ -223,6 +249,10 @@ func (h *houstonHandler) Deploy(w http.ResponseWriter, r *http.Request) {
 			continue
 		}
 
+		if strings.HasPrefix(fd.Name(), "@") {
+			continue
+		}
+
 		fi, _ := fd.Info()
 		if fi.ModTime().After(latestTime) {
 			latestFilename = fi.Name()

server/http_handler.go

@@ -15,10 +15,6 @@ import (
 	"repositories.action2quare.com/ayo/gocommon/logger"
 )
 
-const (
-	defaultMaxMemory = 32 << 10 // 32 KB
-)
-
 type HoustonServerWithHandler interface {
 	HoustonServer
 	RegisterHandlers(serveMux *http.ServeMux, prefix string) error
@@ -157,10 +153,8 @@ func (h *houstonHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	var operation string
 	if r.Method == "POST" {
 		operation = r.FormValue("operation")
-		logger.Println("api called :", userinfo["email"], r.Form)
 	} else {
 		operation = r.URL.Query().Get("operation")
-		logger.Println("api called :", userinfo["email"], r.URL.Query())
 	}
 
 	if len(operation) == 0 {

server/operation.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"reflect"
+	"strings"
 	"sync"
 
 	"repositories.action2quare.com/ayo/houston/shared"
@@ -150,6 +151,12 @@ func marshal(argval reflect.Value, output map[string]string) map[string]string {
 			marshal(argval.Field(i), output)
 		} else if argval.Field(i).CanInt() {
 			output[argval.Type().Field(i).Name] = fmt.Sprintf("%d", argval.Field(i).Int())
+		} else if argval.Field(i).Kind() == reflect.Array || argval.Field(i).Kind() == reflect.Slice {
+			var conv []string
+			for j := 0; j < argval.Field(i).Len(); j++ {
+				conv = append(conv, argval.Field(i).Index(j).String())
+			}
+			output[argval.Type().Field(i).Name] = strings.Join(conv, "\n")
 		} else {
 			output[argval.Type().Field(i).Name] = argval.Field(i).String()
 		}
@@ -158,8 +165,6 @@ func marshal(argval reflect.Value, output map[string]string) map[string]string {
 }
 
 func (os *operationServer) Query(svr protos.Operation_QueryServer) error {
-	// 서버는 업데이트가 있는지 확인하고 있으면 stream에 응답을 보낸다.
-	// 업데이트가 없으면 대기
 	desc, err := svr.Recv()
 	if err != nil {
 		return err

server/server.go

@@ -5,7 +5,7 @@ import (
 	"fmt"
 	"net"
 	"os"
-	"time"
+	"sync/atomic"
 
 	"repositories.action2quare.com/ayo/gocommon/logger"
 	"repositories.action2quare.com/ayo/houston/client"
@@ -176,27 +176,34 @@ func (hs *houstonServer) Start() error {
 		return err
 	}
 
+	closeCount := int32(0)
+	var hc client.HoustonClient
 	if loadServerConfig().RunAsClient {
+		hc, err = client.NewClient(false)
+		if err != nil {
+			return err
+		}
+
 		go func() {
-			time.Sleep(time.Second)
-			hc, err := client.NewClient()
-			if err != nil {
-				logger.Fatal(err)
-				return
-			}
 			hc.Start()
+			if atomic.AddInt32(&closeCount, 1) == 1 {
+				hs.Stop()
+			}
 		}()
 	}
 
-	if err := hs.rpcServer.Serve(lis); err != nil {
-		return err
+	err = hs.rpcServer.Serve(lis)
+	if atomic.AddInt32(&closeCount, 1) == 1 {
+		if hc != nil {
+			hc.Shutdown()
+		}
 	}
 
-	return nil
+	return err
 }
 
 func (hs *houstonServer) Stop() {
-	hs.rpcServer.GracefulStop()
+	hs.rpcServer.Stop()
 }
 
 func (hs *houstonServer) Operation() Operation {