자동 재접속

This commit is contained in:
2023-06-14 01:50:40 +09:00
parent 2e4b7811db
commit 96ee2a4627
3 changed files with 43 additions and 38 deletions

View File

@ -103,7 +103,6 @@ type procmeta struct {
} }
type houstonClient struct { type houstonClient struct {
client *grpc.ClientConn
childProcs []*procmeta childProcs []*procmeta
extraMetrics unsafe.Pointer // map[string]float32 extraMetrics unsafe.Pointer // map[string]float32
deploys map[string][]*protos.VersionAndArgs deploys map[string][]*protos.VersionAndArgs
@ -111,9 +110,10 @@ type houstonClient struct {
ctx context.Context ctx context.Context
operationChan chan *protos.OperationQueryResponse operationChan chan *protos.OperationQueryResponse
exitChan chan *exec.Cmd exitChan chan *exec.Cmd
httpAddr string clientChan chan *grpc.ClientConn
timestamp string timestamp string
wg sync.WaitGroup wg sync.WaitGroup
config clientConfig
} }
func unmarshal[T any](val *T, src map[string]string) { func unmarshal[T any](val *T, src map[string]string) {
@ -195,22 +195,6 @@ func NewClient() (HoustonClient, error) {
return nil, errors.New("client.http_server_address is missing") return nil, errors.New("client.http_server_address is missing")
} }
var client *grpc.ClientConn
for {
logger.Println("grpc.DialContext :", clientConfig.GrpcAddress)
dialContext, cancelDial := context.WithTimeout(context.Background(), 15*time.Second)
client, err = grpc.DialContext(dialContext, clientConfig.GrpcAddress, grpc.WithBlock(), grpc.WithTransportCredentials(insecure.NewCredentials()))
if err == nil {
cancelDial()
break
}
cancelDial()
logger.Println("grpc.DialContext failed :", err)
time.Sleep(5 * time.Second)
}
logger.Println("grpc.DialContext succeeded")
exefile, err := os.Executable() exefile, err := os.Executable()
if err != nil { if err != nil {
return nil, err return nil, err
@ -234,10 +218,10 @@ func NewClient() (HoustonClient, error) {
} }
hc := &houstonClient{ hc := &houstonClient{
client: client, config: clientConfig,
clientChan: make(chan *grpc.ClientConn),
extraMetrics: unsafe.Pointer(&map[string]float32{}), extraMetrics: unsafe.Pointer(&map[string]float32{}),
deploys: deploys, deploys: deploys,
httpAddr: clientConfig.HttpAddress,
timestamp: exefi.ModTime().String(), timestamp: exefi.ModTime().String(),
} }
@ -251,7 +235,7 @@ func NewClient() (HoustonClient, error) {
defer hc.wg.Done() defer hc.wg.Done()
// 메인 operator // 메인 operator
op := protos.NewOperationClient(hc.client) var op protos.OperationClient
myname, _ := os.Executable() myname, _ := os.Executable()
myname = path.Base(filepath.ToSlash(myname)) myname = path.Base(filepath.ToSlash(myname))
if len(path.Ext(myname)) > 0 { if len(path.Ext(myname)) > 0 {
@ -266,6 +250,9 @@ func NewClient() (HoustonClient, error) {
case <-ctx.Done(): case <-ctx.Done():
return return
case newClient := <-hc.clientChan:
op = protos.NewOperationClient(newClient)
case exited := <-exitChan: case exited := <-exitChan:
var newprocs []*procmeta var newprocs []*procmeta
for _, proc := range hc.childProcs { for _, proc := range hc.childProcs {
@ -329,21 +316,21 @@ func NewClient() (HoustonClient, error) {
case shared.Start: case shared.Start:
var sr shared.StartProcessRequest var sr shared.StartProcessRequest
unmarshal(&sr, resp.Args) unmarshal(&sr, resp.Args)
if err := hc.startChildProcess(&sr); err != nil { if err := hc.startChildProcess(&sr, op); err != nil {
logger.Println(err) logger.Println(err)
} }
case shared.Stop: case shared.Stop:
var sr shared.StopProcessRequest var sr shared.StopProcessRequest
unmarshal(&sr, resp.Args) unmarshal(&sr, resp.Args)
if err := hc.stopChildProcess(&sr); err != nil { if err := hc.stopChildProcess(&sr, op); err != nil {
logger.Println(err) logger.Println(err)
} }
case shared.Restart: case shared.Restart:
var rr shared.RestartProcessRequest var rr shared.RestartProcessRequest
unmarshal(&rr, resp.Args) unmarshal(&rr, resp.Args)
if err := hc.restartChildProcess(&rr); err != nil { if err := hc.restartChildProcess(&rr, op); err != nil {
logger.Println(err) logger.Println(err)
} }
@ -392,15 +379,37 @@ func (hc *houstonClient) Start() {
hc.shutdownFunc() hc.shutdownFunc()
}() }()
var client *grpc.ClientConn
reconnCount := 0
for { for {
select { select {
case <-hc.ctx.Done(): case <-hc.ctx.Done():
return return
default: default:
err := hc.checkOperation() if client == nil {
if err != nil { if reconnCount == 0 {
logger.Println("hc.checkUpdate failed :", err) logger.Println("grpc.DialContext :", hc.config.GrpcAddress)
}
reconnCount++
dialContext, cancelDial := context.WithTimeout(context.Background(), 15*time.Second)
client, _ = grpc.DialContext(dialContext, hc.config.GrpcAddress, grpc.WithBlock(), grpc.WithTransportCredentials(insecure.NewCredentials()))
cancelDial()
if client != nil {
reconnCount = 0
logger.Println("grpc.DialContext succeeded")
hc.clientChan <- client
}
}
if client != nil {
err := hc.checkOperation(client)
if err != nil {
logger.Println("hc.checkUpdate failed :", err)
client = nil
}
} }
} }
} }
@ -410,7 +419,7 @@ func (hc *houstonClient) Shutdown() {
hc.shutdownFunc() hc.shutdownFunc()
} }
func (hc *houstonClient) checkOperation() error { func (hc *houstonClient) checkOperation(client *grpc.ClientConn) error {
defer func() { defer func() {
r := recover() r := recover()
if r != nil { if r != nil {
@ -418,7 +427,7 @@ func (hc *houstonClient) checkOperation() error {
} }
}() }()
op := protos.NewOperationClient(hc.client) op := protos.NewOperationClient(client)
cl, err := op.Query(hc.ctx, grpc.WaitForReady(true)) cl, err := op.Query(hc.ctx, grpc.WaitForReady(true))
if err != nil { if err != nil {
return err return err

View File

@ -194,7 +194,7 @@ func (hc *houstonClient) prepareDeploy(name string, version string) (destPath st
func (hc *houstonClient) makeDownloadUrl(rel string) string { func (hc *houstonClient) makeDownloadUrl(rel string) string {
out := rel out := rel
if !strings.HasPrefix(out, "http") { if !strings.HasPrefix(out, "http") {
tks := strings.SplitN(hc.httpAddr, "://", 2) tks := strings.SplitN(hc.config.HttpAddress, "://", 2)
out = fmt.Sprintf("%s://%s", tks[0], path.Join(tks[1], rel)) out = fmt.Sprintf("%s://%s", tks[0], path.Join(tks[1], rel))
} }
return out return out

View File

@ -53,7 +53,7 @@ func (hc *houstonClient) uploadZipLogFile(zipFile string, name string, version s
defer zf.Close() defer zf.Close()
req, err := http.NewRequest("POST", hc.httpAddr+"/upload", zf) req, err := http.NewRequest("POST", hc.config.HttpAddress+"/upload", zf)
if err != nil { if err != nil {
logger.Println(err) logger.Println(err)
} }
@ -323,7 +323,7 @@ func (hc *houstonClient) launch(meta *procmeta) error {
var errPrepareprocessLaunchFailed = errors.New("prepareProcessLaunch failed") var errPrepareprocessLaunchFailed = errors.New("prepareProcessLaunch failed")
func (hc *houstonClient) startChildProcess(req *shared.StartProcessRequest) error { func (hc *houstonClient) startChildProcess(req *shared.StartProcessRequest, op protos.OperationClient) error {
logger.Println("startChildProcess :", *req) logger.Println("startChildProcess :", *req)
if req.Version == "latest" { if req.Version == "latest" {
// 최신 버전을 찾음 // 최신 버전을 찾음
@ -363,8 +363,6 @@ func (hc *houstonClient) startChildProcess(req *shared.StartProcessRequest) erro
} }
hc.childProcs = append(hc.childProcs, meta) hc.childProcs = append(hc.childProcs, meta)
op := protos.NewOperationClient(hc.client)
op.Refresh(context.Background(), hc.makeOperationQueryRequest()) op.Refresh(context.Background(), hc.makeOperationQueryRequest())
return nil return nil
@ -372,7 +370,7 @@ func (hc *houstonClient) startChildProcess(req *shared.StartProcessRequest) erro
var errNoRunningProcess = errors.New("no running processed") var errNoRunningProcess = errors.New("no running processed")
func (hc *houstonClient) stopChildProcess(req *shared.StopProcessRequest) error { func (hc *houstonClient) stopChildProcess(req *shared.StopProcessRequest, op protos.OperationClient) error {
if req.Version == "latest" { if req.Version == "latest" {
// 최신 버전을 찾음 // 최신 버전을 찾음
latest, err := shared.FindLastestVersion(req.Name) latest, err := shared.FindLastestVersion(req.Name)
@ -422,7 +420,6 @@ func (hc *houstonClient) stopChildProcess(req *shared.StopProcessRequest) error
} }
} }
op := protos.NewOperationClient(hc.client)
op.Refresh(context.Background(), hc.makeOperationQueryRequest()) op.Refresh(context.Background(), hc.makeOperationQueryRequest())
for _, proc := range killing { for _, proc := range killing {
@ -439,7 +436,7 @@ func (hc *houstonClient) stopChildProcess(req *shared.StopProcessRequest) error
return errNoRunningProcess return errNoRunningProcess
} }
func (hc *houstonClient) restartChildProcess(req *shared.RestartProcessRequest) error { func (hc *houstonClient) restartChildProcess(req *shared.RestartProcessRequest, op protos.OperationClient) error {
if req.Version == "latest" { if req.Version == "latest" {
// 최신 버전을 찾음 // 최신 버전을 찾음
latest, err := shared.FindLastestVersion(req.Name) latest, err := shared.FindLastestVersion(req.Name)
@ -472,7 +469,6 @@ func (hc *houstonClient) restartChildProcess(req *shared.RestartProcessRequest)
proc.state = protos.ProcessState_Stopping proc.state = protos.ProcessState_Stopping
} }
op := protos.NewOperationClient(hc.client)
op.Refresh(context.Background(), hc.makeOperationQueryRequest()) op.Refresh(context.Background(), hc.makeOperationQueryRequest())
for _, proc := range restarts { for _, proc := range restarts {