add gosuv remove support

.fsw.yml

@@ -0,0 +1,16 @@
+desc: Auto generated by fswatch [gosuv]
+triggers:
+- name: ""
+  pattens:
+  - '**/*.go'
+  - '**/*.c'
+  - '**/*.py'
+  env:
+    DEBUG: "1"
+  cmd: sh ./build.sh
+  shell: true
+  delay: 100ms
+  signal: KILL
+watch_paths:
+- .
+watch_depth: 5

.gopack.yml

@@ -13,4 +13,4 @@ excludes:
   - \.git
 script:
   - go get -v
-  - go build
+  - sh build.sh

build.sh

@@ -0,0 +1,20 @@
+#!/bin/bash -
+#
+
+
+# get current tag
+VERSION=$(git name-rev --tags --name-only $(git rev-parse HEAD))
+
+# use the latest tag
+if test "X$VERSION" = "Xundefined"
+then
+    VERSION="$(git describe --abbrev=0 --tags)"
+    if test -z "$VERSION"
+    then
+        VERSION="0.0.1"
+    fi
+    VERSION="${VERSION}.dev"
+fi
+
+SHA=$(git rev-parse HEAD)
+exec go build -ldflags "-X main.GOSUV_VERSION=$VERSION"

gosuv.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"bufio"
 	"fmt"
 	"io"
 	"io/ioutil"
@@ -8,6 +9,7 @@ import (
 	"os"
 	"os/exec"
 	"path/filepath"
+	"strings"
 	"time"
 
 	"github.com/codegangsta/cli"
@@ -18,7 +20,7 @@ import (
 	"google.golang.org/grpc"
 )
 
-const GOSUV_VERSION = "0.0.3"
+var GOSUV_VERSION = "Unknown"
 
 var (
 	GOSUV_HOME           = os.ExpandEnv("$HOME/.gosuv")
@@ -44,23 +46,21 @@ func connect(ctx *cli.Context) (cc *grpc.ClientConn, err error) {
 	return conn, err
 }
 
-func testConnection(network, address string) error {
-	log.Debugf("test connection")
-	testconn, err := net.DialTimeout(network, address, time.Millisecond*100)
-	if err != nil {
-		log.Debugf("start run server")
+func DialWithRetry(network, address string) (conn *grpc.ClientConn, err error) {
+	conn, err = grpcDial(network, address)
+	if err == nil {
+		return
+	} else {
 		cmd := exec.Command(os.Args[0], "serv")
 		timeout := time.Millisecond * 500
 		er := <-GoTimeoutFunc(timeout, cmd.Run)
 		if er == ErrGoTimeout {
 			fmt.Println("server started")
 		} else {
-			return fmt.Errorf("server stared failed, %v", er)
+			return nil, fmt.Errorf("server stared failed, %v", er)
 		}
-	} else {
-		testconn.Close()
+		return grpcDial(network, address)
 	}
-	return nil
 }
 
 func wrap(f interface{}) func(*cli.Context) {
@@ -69,15 +69,12 @@ func wrap(f interface{}) func(*cli.Context) {
 			log.SetOutputLevel(log.Ldebug)
 		}
 
-		if err := testConnection("unix", GOSUV_SOCK_PATH); err != nil {
-			log.Fatal(err)
-		}
-
-		conn, err := connect(ctx)
+		conn, err := DialWithRetry("unix", GOSUV_SOCK_PATH)
 		if err != nil {
 			log.Fatal(err)
 		}
 		defer conn.Close()
+
 		programClient := pb.NewProgramClient(conn)
 		gosuvClient := pb.NewGoSuvClient(conn)
 
@@ -134,9 +131,9 @@ func ActionAdd(ctx *cli.Context, client pb.GoSuvClient) {
 	fmt.Println(res.Message)
 }
 
-func buildURI(ctx *cli.Context, uri string) string {
-	return fmt.Sprintf("http://%s%s", ctx.GlobalString("addr"), uri)
-}
+// func buildURI(ctx *cli.Context, uri string) string {
+// 	return fmt.Sprintf("http://%s%s", ctx.GlobalString("addr"), uri)
+// }
 
 func ActionStop(ctx *cli.Context) {
 	conn, err := connect(ctx)
@@ -154,15 +151,17 @@ func ActionStop(ctx *cli.Context) {
 	fmt.Println(res.Message)
 }
 
-func ActionRemove(ctx *cli.Context) {
-	conn, err := connect(ctx)
-	if err != nil {
-		log.Fatal(err)
+func ActionRemove(ctx *cli.Context, client pb.ProgramClient) {
+	name := ctx.Args().First()
+	reader := bufio.NewReader(os.Stdin)
+	fmt.Print("Danger operation, Enter name again: ")
+	text, _ := reader.ReadString('\n')
+	text = strings.TrimSpace(text)
+	if text != name {
+		fmt.Println("Canceled.")
+		return
 	}
-	defer conn.Close()
 
-	name := ctx.Args().First()
-	client := pb.NewProgramClient(conn)
 	res, err := client.Remove(context.Background(), &pb.Request{Name: name})
 	if err != nil {
 		Errorf("ERR: %#v\n", err)
@@ -215,10 +214,13 @@ func ActionStart(ctx *cli.Context) {
 
 // grpc.Dial can't set network, so I have to implement this func
 func grpcDial(network, addr string) (*grpc.ClientConn, error) {
-	return grpc.Dial(addr, grpc.WithInsecure(), grpc.WithDialer(
-		func(address string, timeout time.Duration) (conn net.Conn, err error) {
-			return net.DialTimeout(network, address, timeout)
-		}))
+	return grpc.Dial(addr,
+		grpc.WithInsecure(),
+		grpc.WithTimeout(time.Second*2),
+		grpc.WithDialer(
+			func(address string, timeout time.Duration) (conn net.Conn, err error) {
+				return net.DialTimeout(network, address, timeout)
+			}))
 }
 
 func ActionShutdown(ctx *cli.Context) {
@@ -238,12 +240,12 @@ func ActionShutdown(ctx *cli.Context) {
 }
 
 func ActionVersion(ctx *cli.Context, client pb.GoSuvClient) {
-	fmt.Printf("Client: %s\n", GOSUV_VERSION)
+	fmt.Printf("gosuv version:\n  client: %s\n", GOSUV_VERSION)
 	res, err := client.Version(context.Background(), &pb.NopRequest{})
 	if err != nil {
 		log.Fatal(err)
 	}
-	fmt.Printf("Server: %s\n", res.Message)
+	fmt.Printf("  server: %s\n", res.Message)
 }
 
 var app *cli.App
@@ -320,7 +322,7 @@ func initCli() {
 					Value: 10,
 					Usage: "The location is number lines.",
 				},
-				cli.BoolFlag{
+				cli.BoolTFlag{
 					Name:  "follow, f",
 					Usage: "Constantly show log",
 				},

program.go

@@ -6,14 +6,13 @@ import (
 	"io"
 	"io/ioutil"
 	"os"
-	"os/exec"
 	"path/filepath"
 	"sort"
 	"sync"
 	"syscall"
 	"time"
 
-	"github.com/codeskyblue/kproc"
+	"github.com/codeskyblue/kexec"
 	"github.com/qiniu/log"
 )
 
@@ -66,18 +65,18 @@ type ProgramInfo struct {
 	StartSeconds int      `json:"startsecs"`
 }
 
-func (p *ProgramInfo) buildCmd() *exec.Cmd {
-	cmd := exec.Command(p.Command[0], p.Command[1:]...)
+func (p *ProgramInfo) buildCmd() *kexec.KCommand {
+	cmd := kexec.Command(p.Command[0], p.Command[1:]...)
 	cmd.Dir = p.Dir
 	cmd.Env = append(os.Environ(), p.Environ...)
 	return cmd
 }
 
 type Program struct {
-	*kproc.Process `json:"-"`
-	Status         string         `json:"state"`
-	Sig            chan os.Signal `json:"-"`
-	Info           *ProgramInfo   `json:"info"`
+	*kexec.KCommand `json:"-"`
+	Status          string         `json:"state"`
+	Sig             chan os.Signal `json:"-"`
+	Info            *ProgramInfo   `json:"info"`
 
 	retry int
 	stopc chan bool
@@ -194,7 +193,7 @@ func (p *Program) Stop() error {
 }
 
 func (p *Program) Start() error {
-	p.Process = kproc.ProcCommand(p.Info.buildCmd())
+	p.KCommand = p.Info.buildCmd()
 	logFd, err := p.createLog()
 	if err != nil {
 		return err
@@ -217,7 +216,7 @@ func initProgramTable() {
 type ProgramTable struct {
 	table map[string]*Program
 	ch    chan string
-	mu    sync.Mutex
+	mu    sync.RWMutex
 }
 
 var (
@@ -262,6 +261,7 @@ func (pt *ProgramTable) loadConfig() error {
 func (pt *ProgramTable) AddProgram(p *Program) error {
 	pt.mu.Lock()
 	defer pt.mu.Unlock()
+
 	name := p.Info.Name
 	if _, exists := pt.table[name]; exists {
 		return ErrProgramDuplicate
@@ -272,25 +272,22 @@ func (pt *ProgramTable) AddProgram(p *Program) error {
 }
 
 func (pt *ProgramTable) Programs() []*Program {
-	pt.mu.Lock()
-	defer pt.mu.Unlock()
+	pt.mu.RLock()
+	defer pt.mu.RUnlock()
 	ps := make([]*Program, 0, len(pt.table))
 	names := []string{}
 	for name, _ := range pt.table {
 		names = append(names, name)
 	}
-	// log.Println(names)
 	sort.Strings(names)
-	// log.Println(names)
+
 	for _, name := range names {
 		ps = append(ps, pt.table[name])
 	}
-	// for _, p := range pt.table {
-	// ps = append(ps, p)
-	// }
 	return ps
 }
 
+// golang map get will alwyas be safe.
 func (pt *ProgramTable) Get(name string) (*Program, error) {
 	program, exists := pt.table[name]
 	if !exists {
@@ -303,6 +300,20 @@ func (pt *ProgramTable) StopAll() {
 	pt.mu.Lock()
 	defer pt.mu.Unlock()
 	for _, program := range pt.table {
-		program.Stop()
+		program.InputData(EVENT_STOP)
+	}
+}
+
+func (pt *ProgramTable) Remove(name string) error {
+	pt.mu.Lock()
+	defer pt.mu.Unlock()
+
+	program, exists := pt.table[name]
+	if !exists {
+		return ErrProgramNotExists
 	}
+	program.InputData(EVENT_STOP)
+	// program.Stop()
+	delete(pt.table, name)
+	return pt.saveConfig()
 }

serv.go

@@ -0,0 +1,56 @@
+package main
+
+import (
+	"fmt"
+	"net"
+	"os"
+	"os/signal"
+	"syscall"
+
+	pb "github.com/codeskyblue/gosuv/gosuvpb"
+	"github.com/qiniu/log"
+	"google.golang.org/grpc"
+)
+
+func handleSignal(lis net.Listener) {
+	sigc := make(chan os.Signal, 2)
+	signal.Notify(sigc, syscall.SIGINT, syscall.SIGHUP)
+	go func() {
+		for sig := range sigc {
+			log.Println("Receive signal:", sig)
+			if sig == syscall.SIGHUP {
+				return // ignore, when shell session closed, gosuv will receive SIGHUP signal
+			}
+			lis.Close()
+			log.Println("Kill all running gosuv programs")
+			programTable.StopAll()
+			os.Exit(0)
+			return
+		}
+	}()
+}
+
+type PbSuvServer struct {
+	lis net.Listener
+}
+
+func RunGosuvService(addr string) error {
+	initProgramTable()
+
+	lis, err := net.Listen("unix", GOSUV_SOCK_PATH)
+	if err != nil {
+		log.Fatal(err)
+	}
+	handleSignal(lis)
+
+	pbServ := &PbSuvServer{}
+	pbProgram := &PbProgram{}
+
+	grpcServ := grpc.NewServer()
+	pb.RegisterGoSuvServer(grpcServ, pbServ)
+	pb.RegisterProgramServer(grpcServ, pbProgram)
+
+	pbServ.lis = lis
+	grpcServ.Serve(lis)
+	return fmt.Errorf("Address: %s has been used", addr)
+}

serv_cmds.go

@@ -2,19 +2,65 @@ package main
 
 import (
 	"fmt"
-	"net"
+	"log"
 	"os"
 	"os/exec"
-	"os/signal"
-	"syscall"
 	"time"
 
 	pb "github.com/codeskyblue/gosuv/gosuvpb"
-	"github.com/qiniu/log"
 	"golang.org/x/net/context"
-	"google.golang.org/grpc"
 )
 
+func (s *PbSuvServer) Create(ctx context.Context, in *pb.ProgramInfo) (*pb.Response, error) {
+	pinfo := new(ProgramInfo)
+	pinfo.Name = in.Name
+	pinfo.Command = in.Command
+	pinfo.Dir = in.Directory
+	pinfo.Environ = in.Environ
+	pinfo.AutoStart = true
+
+	log.Println(in.Name)
+	program := NewProgram(pinfo)
+	if err := programTable.AddProgram(program); err != nil {
+		return nil, err
+	}
+	program.InputData(EVENT_START)
+	res := new(pb.Response)
+	res.Code = 200
+	res.Message = fmt.Sprintf("%s: created", pinfo.Name)
+	return res, nil
+}
+
+func (s *PbSuvServer) Shutdown(ctx context.Context, in *pb.NopRequest) (*pb.Response, error) {
+	go func() {
+		time.Sleep(50 * time.Millisecond)
+		s.lis.Close()
+		programTable.StopAll()
+		os.Exit(2)
+	}()
+	res := &pb.Response{}
+	res.Message = "gosuv shutdown"
+	return res, nil
+}
+func (s *PbSuvServer) Version(ctx context.Context, in *pb.NopRequest) (res *pb.Response, err error) {
+	res = &pb.Response{
+		Message: GOSUV_VERSION,
+	}
+	return
+}
+
+func (s *PbSuvServer) Status(ctx context.Context, in *pb.NopRequest) (res *pb.StatusResponse, err error) {
+	res = &pb.StatusResponse{}
+	for _, program := range programTable.Programs() {
+		ps := &pb.ProgramStatus{}
+		ps.Name = program.Info.Name
+		ps.Status = program.Status
+		ps.Extra = "..."
+		res.Programs = append(res.Programs, ps)
+	}
+	return
+}
+
 type PbProgram struct{}
 
 func (this *PbProgram) Start(ctx context.Context, in *pb.Request) (res *pb.Response, err error) {
@@ -42,8 +88,9 @@ func (this *PbProgram) Stop(ctx context.Context, in *pb.Request) (res *pb.Respon
 // Remove(ctx context.Context, in *Request, opts ...grpc.CallOption) (*Response, error)
 func (s *PbProgram) Remove(ctx context.Context, in *pb.Request) (res *pb.Response, err error) {
 	res = &pb.Response{}
-	res.Message = "TODO: remove not finished. " + in.Name
-	return res, nil
+	err = programTable.Remove(in.Name)
+	res.Message = fmt.Sprintf("Program %s removed.", in.Name)
+	return
 }
 
 //func (this *PbProgram) Tail(ctx context.Context, in *pb.Request)(stream
@@ -77,95 +124,3 @@ func (c *PbProgram) Tail(in *pb.TailRequest, stream pb.Program_TailServer) (err
 	}
 	return nil
 }
-
-type PbSuvServer struct {
-	lis net.Listener
-}
-
-func (s *PbSuvServer) Create(ctx context.Context, in *pb.ProgramInfo) (*pb.Response, error) {
-	pinfo := new(ProgramInfo)
-	pinfo.Name = in.Name
-	pinfo.Command = in.Command
-	pinfo.Dir = in.Directory
-	pinfo.Environ = in.Environ
-
-	log.Println(in.Name)
-	program := NewProgram(pinfo)
-	if err := programTable.AddProgram(program); err != nil {
-		return nil, err
-	}
-	program.InputData(EVENT_START)
-	res := new(pb.Response)
-	res.Code = 200
-	res.Message = fmt.Sprintf("%s: created", pinfo.Name)
-	return res, nil
-}
-
-func (s *PbSuvServer) Shutdown(ctx context.Context, in *pb.NopRequest) (*pb.Response, error) {
-	go func() {
-		time.Sleep(50 * time.Millisecond)
-		s.lis.Close()
-		programTable.StopAll()
-		os.Exit(2)
-	}()
-	res := &pb.Response{}
-	res.Message = "gosuv shutdown"
-	return res, nil
-}
-
-func (s *PbSuvServer) Version(ctx context.Context, in *pb.NopRequest) (res *pb.Response, err error) {
-	res = &pb.Response{
-		Message: GOSUV_VERSION,
-	}
-	return
-}
-
-func (s *PbSuvServer) Status(ctx context.Context, in *pb.NopRequest) (res *pb.StatusResponse, err error) {
-	res = &pb.StatusResponse{}
-	for _, program := range programTable.Programs() {
-		ps := &pb.ProgramStatus{}
-		ps.Name = program.Info.Name
-		ps.Status = program.Status
-		ps.Extra = "..."
-		res.Programs = append(res.Programs, ps)
-	}
-	return
-}
-
-func handleSignal(lis net.Listener) {
-	sigc := make(chan os.Signal, 2)
-	signal.Notify(sigc, syscall.SIGINT, syscall.SIGHUP)
-	go func() {
-		for sig := range sigc {
-			log.Println("Receive signal:", sig)
-			if sig == syscall.SIGHUP {
-				return // ignore, when shell session closed, gosuv will receive SIGHUP signal
-			}
-			lis.Close()
-			programTable.StopAll()
-			os.Exit(0)
-			return
-		}
-	}()
-}
-
-func RunGosuvService(addr string) error {
-	initProgramTable()
-
-	lis, err := net.Listen("unix", GOSUV_SOCK_PATH)
-	if err != nil {
-		log.Fatal(err)
-	}
-	handleSignal(lis)
-
-	pbServ := &PbSuvServer{}
-	pbProgram := &PbProgram{}
-
-	grpcServ := grpc.NewServer()
-	pb.RegisterGoSuvServer(grpcServ, pbServ)
-	pb.RegisterProgramServer(grpcServ, pbProgram)
-
-	pbServ.lis = lis
-	grpcServ.Serve(lis)
-	return fmt.Errorf("Address: %s has been used", addr)
-}