overflow_probes/service/ContainerService.go

package service

import (
	"fmt"
	"io/ioutil"
	"os"
	"reflect"
	"strconv"
	"time"

	cda "git.loafle.net/commons_go/di/annotation"
	cdr "git.loafle.net/commons_go/di/registry"
	"git.loafle.net/commons_go/logging"
	crc "git.loafle.net/commons_go/rpc/client"
	crr "git.loafle.net/commons_go/rpc/registry"
	"git.loafle.net/overflow/overflow_probes/client/container"
)

func init() {
	cdr.RegisterType(reflect.TypeOf((*ContainerService)(nil)))
}

type ContainerService struct {
	cda.TypeAnnotation `annotation:"@overFlow:Service()"`

	ProbeRPCInvoker crr.RPCInvoker `annotation:"@Resource()"`

	clients map[string]*containerState
}

type containerState struct {
	pid    int
	port   int
	client crc.Client
}

func (cs *ContainerService) Call(name string, result interface{}, method string, params ...interface{}) error {
	c, err := cs.GetClient(name)
	if nil != err {
		return err
	}
	return c.Call(result, method, params...)
}

func (cs *ContainerService) Send(name string, method string, params ...interface{}) error {
	c, err := cs.GetClient(name)
	if nil != err {
		return err
	}
	return c.Send(method, params...)
}

func (cs *ContainerService) GetClient(name string) (crc.Client, error) {
	cState, ok := cs.clients[name]
	if !ok {
		if err := cs.runProcess(name); nil != err {
			return nil, err
		}
		cState, _ = cs.clients[name]
	}

	return cState.client, nil
}

func (cs *ContainerService) Connect(name string) error {
	return nil
}

func (cs *ContainerService) runProcess(name string) error {
	ok := cs.checkProcess(name)

	if ok {
		return nil
	}

	cmd, pidPath := container.GetContainerCommand(name)
	removePidFile(pidPath)

	if err := cmd.Start(); nil != err {
		logging.Logger().Errorf("Probe: To run container(%s) failed err %v", name, err)
		return err
	}

	port, err := watchPidFileCreate(pidPath, time.Duration(time.Second*2))
	if nil != err {
		return err
	}
	// port := 60000

	client := container.New(port, cs.ProbeRPCInvoker)
	if err := client.Connect(); nil != err {
		cmd.Process.Kill()
		return err
	}

	logging.Logger().Debugf("Container[%s] pid: %d", name, cmd.Process.Pid)

	cState := &containerState{
		pid:    cmd.Process.Pid,
		port:   port,
		client: client,
	}

	if nil == cs.clients {
		cs.clients = make(map[string]*containerState, 0)
	}
	cs.clients[name] = cState

	return nil
}

func (cs *ContainerService) checkProcess(name string) bool {
	cState, ok := cs.clients[name]
	if !ok || nil == cState || nil == cState.client {
		return false
	}

	_, err := os.FindProcess(cState.pid)
	if nil != err {
		return false
	}

	return true
}

func (cs *ContainerService) killProcess(name string) error {
	cState, ok := cs.clients[name]
	if !ok || nil == cState || nil == cState.client {
		return fmt.Errorf("Probe: Container[%s] is not exist", name)
	}

	p, err := os.FindProcess(cState.pid)
	if nil != err {
		return err
	}

	if err = p.Signal(os.Interrupt); nil != err {
		return err
	}
	// if err = p.Kill(); nil != err {
	// 	return err
	// }

	return nil
}

func removePidFile(pidFilePath string) {
	if _, err := os.Stat(pidFilePath); err == nil {
		if err := os.Remove(pidFilePath); nil != err {
			logging.Logger().Errorf("Probe: Removing pid file has been failed [%v]", err)
		}
	}
}

func watchPidFileCreate(pidFilePath string, waitTime time.Duration) (int, error) {
	startTime := time.Now()

	for {
		if _, err := os.Stat(pidFilePath); err == nil {
			buf, err := ioutil.ReadFile(pidFilePath)
			if nil != err {
				return 0, err
			}
			portNumber, err := strconv.ParseInt(string(buf), 10, 32)
			if nil != err {
				return 0, err
			}
			return int(portNumber), nil
		}

		if time.Since(startTime) > waitTime {
			return 0, fmt.Errorf("Probe: pid file not exist")
		}

		time.Sleep(time.Duration(time.Millisecond * 100))
	}
}