micro/runtime/service.go

package runtime

import (
	"io"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/micro/go-micro/v2/logger"
	"github.com/micro/go-micro/v2/runtime/local/build"
	"github.com/micro/go-micro/v2/runtime/local/process"
	proc "github.com/micro/go-micro/v2/runtime/local/process/os"
)

type service struct {
	sync.RWMutex

	running bool
	closed  chan bool
	err     error
	updated time.Time

	retries    int
	maxRetries int

	// output for logs
	output io.Writer

	// service to manage
	*Service
	// process creator
	Process *proc.Process
	// Exec
	Exec *process.Executable
	// process pid
	PID *process.PID
}

func newService(s *Service, c CreateOptions) *service {
	var exec string
	var args []string

	// set command
	exec = strings.Join(c.Command, " ")
	args = c.Args

	return &service{
		Service: s,
		Process: new(proc.Process),
		Exec: &process.Executable{
			Package: &build.Package{
				Name: s.Name,
				Path: exec,
			},
			Env:  c.Env,
			Args: args,
			Dir:  s.Source,
		},
		closed:     make(chan bool),
		output:     c.Output,
		updated:    time.Now(),
		maxRetries: c.Retries,
	}
}

func (s *service) streamOutput() {
	go io.Copy(s.output, s.PID.Output)
	go io.Copy(s.output, s.PID.Error)
}

func (s *service) shouldStart() bool {
	if s.running {
		return false
	}
	return s.retries <= s.maxRetries
}

func (s *service) ShouldStart() bool {
	s.RLock()
	defer s.RUnlock()
	return s.shouldStart()
}

func (s *service) Running() bool {
	s.RLock()
	defer s.RUnlock()
	return s.running
}

// Start stars the service
func (s *service) Start() error {
	s.Lock()
	defer s.Unlock()

	if !s.shouldStart() {
		return nil
	}

	// reset
	s.err = nil
	s.closed = make(chan bool)

	if s.Metadata == nil {
		s.Metadata = make(map[string]string)
	}

	s.Metadata["status"] = "starting"
	// delete any existing error
	delete(s.Metadata, "error")

	// TODO: pull source & build binary
	if logger.V(logger.DebugLevel, logger.DefaultLogger) {
		logger.Debugf("Runtime service %s forking new process", s.Service.Name)
	}
	p, err := s.Process.Fork(s.Exec)
	if err != nil {
		s.Metadata["status"] = "error"
		s.Metadata["error"] = err.Error()
		return err
	}

	// set the pid
	s.PID = p
	// set to running
	s.running = true
	// set status
	s.Metadata["status"] = "running"
	// set started
	s.Metadata["started"] = time.Now().Format(time.RFC3339)

	if s.output != nil {
		s.streamOutput()
	}

	// wait and watch
	go s.Wait()

	return nil
}

// Stop stops the service
func (s *service) Stop() error {
	s.Lock()
	defer s.Unlock()

	select {
	case <-s.closed:
		return nil
	default:
		close(s.closed)
		s.running = false
		if s.PID == nil {
			return nil
		}

		// set status
		s.Metadata["status"] = "stopping"

		// kill the process
		err := s.Process.Kill(s.PID)
		// wait for it to exit
		s.Process.Wait(s.PID)

		// set status
		s.Metadata["status"] = "stopped"

		// return the kill error
		return err
	}
}

// Error returns the last error service has returned
func (s *service) Error() error {
	s.RLock()
	defer s.RUnlock()
	return s.err
}

// Wait waits for the service to finish running
func (s *service) Wait() {
	// wait for process to exit
	err := s.Process.Wait(s.PID)

	s.Lock()
	defer s.Unlock()

	// save the error
	if err != nil {
		s.retries++
		s.Metadata["status"] = "error"
		s.Metadata["error"] = err.Error()
		s.Metadata["retries"] = strconv.Itoa(s.retries)

		s.err = err
	} else {
		s.Metadata["status"] = "done"
	}

	// no longer running
	s.running = false
}