crawler_manager_go/crawler_manager.go

package crawler_manager

import (
	"loafle.com/overflow/crawler_go/config"
	"encoding/json"
	"io/ioutil"
	"google.golang.org/grpc"
	"strconv"
	"os/exec"
	"time"
	"os"
	"log"


	"path/filepath"
	"errors"
)

const (
	address = "localhost:"
	portArgsName = "-Port="
	defaultPort = 50000

	rootFolder = "/home/cm/"
	ConfigFolder = rootFolder + "/config/" // +container
	BinaryFolder = rootFolder + "/container/"
	PidFolder = rootFolder + "/pids/"
	runFile = "ttnc"
)

var g_CrawlerMananger *CrawlerManager = nil

type CrawlerManager struct {
	currentPort int
	portMap map[string]string
	pidMap map[string]string
}

func init() {

	if g_CrawlerMananger == nil {
		g_CrawlerMananger = &CrawlerManager{portMap:make(map[string]string),pidMap:make(map[string]string), currentPort:defaultPort}
	}

	g_CrawlerMananger.init()
}

func GetClient(container string)  (*grpc.ClientConn, error) {
	crm := GetInstance()
	b := crm.checkContainer(&container)

	if b == false {
		err := crm.runAndInitContainer(&container)
		if err != nil {
			return nil, err
		}
	}
	gport := crm.portMap[container]

	return grpc.Dial(address+gport, grpc.WithInsecure())
}

func GetInstance() *CrawlerManager {
	return g_CrawlerMananger;
}



func readConfig(path string ) *config.Config {

	bytes, err := ioutil.ReadFile(path)

	if err != nil {
		return nil
	}

	cc := config.Config{}

	json.Unmarshal(bytes, &cc)

	return &cc
}



func (c *CrawlerManager)init() {

	c.checkPid()

	cs := c.isStartContainer()

	for _, cc := range cs {

		err := c.runAndInitContainer(&cc)

		if err != nil {
			log.Println(err)
			continue
		}

	}

}

func (c *CrawlerManager)isStartContainer() []string {

	files, _ := ioutil.ReadDir(ConfigFolder)

	var cs []string

	for _,file := range files {

		if file.IsDir() {
			b := existConfigFile(ConfigFolder, file.Name())
			if b {
				cs = append(cs, file.Name())
			}
		}

	}
	return cs
}

func (c *CrawlerManager)checkContainer(container *string) bool {
	return callStatus(container)
}


func (c *CrawlerManager)checkPid() {

	files, err :=  ioutil.ReadDir(PidFolder)

	if err != nil {
		log.Println(err)
		return
	}

	for _, file := range files {

		if file.IsDir() {
			continue
		}

		str := file.Name()

		c.stopProcess(&str)
		c.removeProcessFile(&str)

	}

}


func (c *CrawlerManager)runAndInitContainer(container *string) error {

	err := c.runContainer(container)

	if err != nil {
		return err
	}

	dirs := getConfigPaths(container)

	if dirs == nil {
		return errors.New("not found config")
	}
	b := callInit(container, dirs)
	if b == false {
		return errors.New("call init failed")
	}


	return nil
}

func (c *CrawlerManager)runContainer(container *string) error {

	b := c.checkContainer(container)
	if b {
		return nil
	}

	cmdStr := getRunCommand(container)

	for {
		pArg := portArgsName + strconv.Itoa(c.currentPort)
		cmd := exec.Command(cmdStr, pArg)

		err := cmd.Start()
		if err != nil {
			//run error break;
			log.Println(err)
			return err
		}

		time.Sleep(time.Duration( time.Second * 2))

		b := callStatusAddress(address + strconv.Itoa(c.currentPort))

		if b == false {
			c.currentPort++
			continue;
		}
		log.Println(*container + " run success port:" , c.currentPort , "pid:", cmd.Process.Pid )

		writePid(cmd.Process.Pid)

		c.portMap[*container] = strconv.Itoa(c.currentPort)
		c.pidMap[*container] = strconv.Itoa(cmd.Process.Pid)

		c.currentPort++

		break;
	}

	return nil
}


func (c *CrawlerManager)stopContainerAll() {

	for k, _ := range c.pidMap {
		c.stopContainer(&k)
	}

}

func (c *CrawlerManager)stopContainer(container *string) {

	pid := c.pidMap[*container]

	if len(pid) <= 0 {
		return
	}

	c.stopProcess(&pid)
	c.removeProcessFile(&pid)

	delete(c.pidMap, *container)
	delete(c.portMap, *container)

}

func (c *CrawlerManager)stopProcess(pid *string) {

	pidi, err := strconv.Atoi(*pid)

	if err != nil {
		log.Println(err)
	}

	p, err := os.FindProcess(pidi)
	if err != nil {
		log.Println(err)
	}
	p.Kill()


}

func (c *CrawlerManager)removeProcessFile(pid *string) {
	err := os.Remove(PidFolder + "/" +*pid)
	if err != nil {
		log.Println(err)
	}
}


func (c *CrawlerManager)activeCrawler(container *string) []string {

	var dirs []string
	existConfigFileDir(ConfigFolder,*container,&dirs)

	if len(dirs) <= 0 {
		return nil
	}

	var r []string
	for _, path := range dirs {
		r = append(r, filepath.Base(path))
	}

	return r

}