d7b4060a28
default port 50000
331 lines
5.7 KiB
Go
331 lines
5.7 KiB
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"
|
|
|
|
|
|
|
|
"errors"
|
|
"loafle.com/overflow/agent_api/config_manager"
|
|
)
|
|
|
|
const (
|
|
address = "localhost:"
|
|
portArgsName = "-Port="
|
|
defaultPort = 50000
|
|
|
|
//rootFolder = "/home/cm2/"
|
|
//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
|
|
ConfigMgr config_manager.ConfigManager
|
|
}
|
|
|
|
|
|
func init() {
|
|
|
|
GetInstance();
|
|
|
|
listenEvent()
|
|
|
|
//g_CrawlerMananger.init()
|
|
}
|
|
|
|
func SettingPath() {
|
|
|
|
//test
|
|
//GetInstance().ConfigMgr.GetGlobalConfig().Paths.RootFolder = "/home/cm2/"
|
|
|
|
GetInstance().ConfigMgr.GetGlobalConfig().Paths.BinaryFolder = GetInstance().ConfigMgr.GetGlobalConfig().Paths.RootFolder + GetInstance().ConfigMgr.GetGlobalConfig().Paths.BinaryFolder
|
|
GetInstance().ConfigMgr.GetGlobalConfig().Paths.ConfigFolder = GetInstance().ConfigMgr.GetGlobalConfig().Paths.RootFolder + GetInstance().ConfigMgr.GetGlobalConfig().Paths.ConfigFolder
|
|
GetInstance().ConfigMgr.GetGlobalConfig().Paths.PidFolder = GetInstance().ConfigMgr.GetGlobalConfig().Paths.RootFolder + GetInstance().ConfigMgr.GetGlobalConfig().Paths.PidFolder
|
|
|
|
}
|
|
|
|
|
|
func GetInstance() *CrawlerManager {
|
|
|
|
if g_CrawlerMananger == nil {
|
|
g_CrawlerMananger = &CrawlerManager{portMap:make(map[string]string),pidMap:make(map[string]string), currentPort:defaultPort}
|
|
}
|
|
|
|
return g_CrawlerMananger;
|
|
}
|
|
|
|
|
|
func (c *CrawlerManager)GetClient(container string) (*grpc.ClientConn, error) {
|
|
|
|
b := c.checkContainer(&container)
|
|
|
|
if b == false {
|
|
err := c.runAndInitContainerOne(&container)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
gport := c.portMap[container]
|
|
|
|
return grpc.Dial(address+gport, grpc.WithInsecure())
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
func (c *CrawlerManager)init() error {
|
|
|
|
c.checkPid()
|
|
|
|
cmap := c.ConfigMgr.GetSensors()
|
|
scm := sortContainer(cmap)
|
|
|
|
|
|
for ctn := range scm {
|
|
err := c.runAndInitContainer(&ctn, scm[ctn])
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
|
|
func (c *CrawlerManager)checkContainer(container *string) bool {
|
|
return callStatus(container)
|
|
}
|
|
|
|
func (c *CrawlerManager)checkAndRunContainer(container *string) bool {
|
|
|
|
b := callStatus(container)
|
|
|
|
if b == false {
|
|
err := c.runAndInitContainerOne(container)
|
|
if err != nil {
|
|
return false
|
|
}
|
|
}
|
|
|
|
return true
|
|
}
|
|
|
|
|
|
func (c *CrawlerManager)checkPid() {
|
|
|
|
files, err := ioutil.ReadDir(c.ConfigMgr.GetGlobalConfig().Paths.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)runAndInitContainerOne(container *string) error {
|
|
|
|
err := c.runContainer(container)
|
|
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
cmap := c.ConfigMgr.GetSensors()
|
|
scm := sortContainer(cmap)
|
|
|
|
b := callInitConfig(container, scm[*container])
|
|
|
|
if b == false {
|
|
return errors.New("call init failed")
|
|
}
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
func (c *CrawlerManager)runAndInitContainer(container *string,cl []*config_manager.Config) error {
|
|
|
|
err := c.runContainer(container)
|
|
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
b := callInitConfig(container, cl)
|
|
|
|
if b == false {
|
|
return errors.New("call init failed")
|
|
}
|
|
|
|
|
|
return nil
|
|
}
|
|
|
|
func sortContainer(cm map[string]*config_manager.Config) map[string][]*config_manager.Config {
|
|
|
|
m := make(map[string][]*config_manager.Config)
|
|
|
|
var cn string;
|
|
for key := range cm {
|
|
cn = cm[key].Crawler.Container
|
|
m[cn] = append(m[cn], cm[key])
|
|
}
|
|
|
|
return m
|
|
}
|
|
|
|
|
|
func (c *CrawlerManager)runContainer(container *string) error {
|
|
|
|
b := c.checkContainer(container)
|
|
if b {
|
|
return nil
|
|
}
|
|
|
|
cmdStr := getRunCommand(container)
|
|
|
|
for {
|
|
pArg := 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 {
|
|
log.Println("false " + strconv.Itoa(c.currentPort))
|
|
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++
|
|
|
|
log.Println(*container + "started")
|
|
|
|
break;
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (c *CrawlerManager)addSensor(id string) error {
|
|
|
|
conf := c.ConfigMgr.GetSensorById(id)
|
|
|
|
b := c.checkAndRunContainer(&conf.Crawler.Name)
|
|
|
|
if b == false {
|
|
return errors.New("run container error")
|
|
}
|
|
|
|
b = callAdd(&conf.Crawler.Container, conf)
|
|
|
|
if b == false {
|
|
return errors.New("Call Add Fail")
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (c *CrawlerManager)removeSensor(id string) {
|
|
|
|
conf := c.ConfigMgr.GetSensorById(id)
|
|
|
|
callRemove(&conf.Crawler.Name, conf)
|
|
|
|
//remove and stop
|
|
|
|
}
|
|
|
|
|
|
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(c.ConfigMgr.GetGlobalConfig().Paths.PidFolder + "/" +*pid)
|
|
if err != nil {
|
|
log.Println(err)
|
|
}
|
|
}
|