Files
mgmt/exec.go
James Shubin 6f3ac4bf2a Rework the converged detection and provide a clean interface
The old converged detection was hacked in code, instead of something
with a nice interface. This cleans it up, splits it into a separate
file, and removes a race condition that happened with the old code.

We also take the time to get rid of the ugly Set* methods and replace
them all with a single AssociateData method. This might be unnecessary
if we can pass in the Converger method at Resource construction.

Lastly, and most interesting, we suspend the individual timeout callers
when they've already converged, thus reducing unnecessary traffic, and
avoiding fast (eg: < 5 second) timers triggering more than once if they
stay converged!

A quick note on theory for any future readers... What happens if we have
--converged-timeout=0 ? Well, for this and any other positive value,
it's important to realize that deciding if something is converged is
actually a race between if the converged timer will fire and if some
random new event will get triggered. This is because there is nothing
that can actually predict if or when a new event will happen (eg the
user modifying a file). As a result, a race is always inherent, and
actually not a negative or "incorrect" algorithm.

A future improvement could be to add a global lock to each resource, and
to lock all resources when computing if we are converged or not. In
practice, this hasn't been necessary. The worst case scenario would be
(in theory, because this hasn't been tested) if an event happens
*during* the converged calculation, and starts running, the exit command
then runs, and the event finishes, but it doesn't get a chance to notify
some service to restart. A lock could probably fix this theoretical
case.
2016-03-29 20:27:38 -04:00

426 lines
12 KiB
Go

// Mgmt
// Copyright (C) 2013-2016+ James Shubin and the project contributors
// Written by James Shubin <james@shubin.ca> and the project contributors
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package main
import (
"bufio"
"bytes"
"encoding/gob"
"errors"
"log"
"os/exec"
"strings"
)
func init() {
gob.Register(&ExecRes{})
}
type ExecRes struct {
BaseRes `yaml:",inline"`
State string `yaml:"state"` // state: exists/present?, absent, (undefined?)
Cmd string `yaml:"cmd"` // the command to run
Shell string `yaml:"shell"` // the (optional) shell to use to run the cmd
Timeout int `yaml:"timeout"` // the cmd timeout in seconds
WatchCmd string `yaml:"watchcmd"` // the watch command to run
WatchShell string `yaml:"watchshell"` // the (optional) shell to use to run the watch cmd
IfCmd string `yaml:"ifcmd"` // the if command to run
IfShell string `yaml:"ifshell"` // the (optional) shell to use to run the if cmd
PollInt int `yaml:"pollint"` // the poll interval for the ifcmd
}
func NewExecRes(name, cmd, shell string, timeout int, watchcmd, watchshell, ifcmd, ifshell string, pollint int, state string) *ExecRes {
obj := &ExecRes{
BaseRes: BaseRes{
Name: name,
},
Cmd: cmd,
Shell: shell,
Timeout: timeout,
WatchCmd: watchcmd,
WatchShell: watchshell,
IfCmd: ifcmd,
IfShell: ifshell,
PollInt: pollint,
State: state,
}
obj.Init()
return obj
}
func (obj *ExecRes) Init() {
obj.BaseRes.kind = "Exec"
obj.BaseRes.Init() // call base init, b/c we're overriding
}
// validate if the params passed in are valid data
// FIXME: where should this get called ?
func (obj *ExecRes) Validate() bool {
if obj.Cmd == "" { // this is the only thing that is really required
return false
}
// if we have a watch command, then we don't poll with the if command!
if obj.WatchCmd != "" && obj.PollInt > 0 {
return false
}
return true
}
// wraps the scanner output in a channel
func (obj *ExecRes) BufioChanScanner(scanner *bufio.Scanner) (chan string, chan error) {
ch, errch := make(chan string), make(chan error)
go func() {
for scanner.Scan() {
ch <- scanner.Text() // blocks here ?
if e := scanner.Err(); e != nil {
errch <- e // send any misc errors we encounter
//break // TODO ?
}
}
close(ch)
errch <- scanner.Err() // eof or some err
close(errch)
}()
return ch, errch
}
// Exec watcher
func (obj *ExecRes) Watch(processChan chan Event) {
if obj.IsWatching() {
return
}
obj.SetWatching(true)
defer obj.SetWatching(false)
cuuid := obj.converger.Register()
defer cuuid.Unregister()
var send = false // send event?
var exit = false
bufioch, errch := make(chan string), make(chan error)
if obj.WatchCmd != "" {
var cmdName string
var cmdArgs []string
if obj.WatchShell == "" {
// call without a shell
// FIXME: are there still whitespace splitting issues?
split := strings.Fields(obj.WatchCmd)
cmdName = split[0]
//d, _ := os.Getwd() // TODO: how does this ever error ?
//cmdName = path.Join(d, cmdName)
cmdArgs = split[1:len(split)]
} else {
cmdName = obj.Shell // usually bash, or sh
cmdArgs = []string{"-c", obj.WatchCmd}
}
cmd := exec.Command(cmdName, cmdArgs...)
//cmd.Dir = "" // look for program in pwd ?
cmdReader, err := cmd.StdoutPipe()
if err != nil {
log.Printf("%v[%v]: Error creating StdoutPipe for Cmd: %v", obj.Kind(), obj.GetName(), err)
log.Fatal(err) // XXX: how should we handle errors?
}
scanner := bufio.NewScanner(cmdReader)
defer cmd.Wait() // XXX: is this necessary?
defer func() {
// FIXME: without wrapping this in this func it panic's
// when running examples/graph8d.yaml
cmd.Process.Kill() // TODO: is this necessary?
}()
if err := cmd.Start(); err != nil {
log.Printf("%v[%v]: Error starting Cmd: %v", obj.Kind(), obj.GetName(), err)
log.Fatal(err) // XXX: how should we handle errors?
}
bufioch, errch = obj.BufioChanScanner(scanner)
}
for {
obj.SetState(resStateWatching) // reset
select {
case text := <-bufioch:
cuuid.SetConverged(false)
// each time we get a line of output, we loop!
log.Printf("%v[%v]: Watch output: %s", obj.Kind(), obj.GetName(), text)
if text != "" {
send = true
}
case err := <-errch:
cuuid.SetConverged(false) // XXX ?
if err == nil { // EOF
// FIXME: add an "if watch command ends/crashes"
// restart or generate error option
log.Printf("%v[%v]: Reached EOF", obj.Kind(), obj.GetName())
return
}
log.Printf("%v[%v]: Error reading input?: %v", obj.Kind(), obj.GetName(), err)
log.Fatal(err)
// XXX: how should we handle errors?
case event := <-obj.events:
cuuid.SetConverged(false)
if exit, send = obj.ReadEvent(&event); exit {
return // exit
}
case _ = <-cuuid.ConvergedTimer():
cuuid.SetConverged(true) // converged!
continue
}
// do all our event sending all together to avoid duplicate msgs
if send {
send = false
// it is okay to invalidate the clean state on poke too
obj.isStateOK = false // something made state dirty
resp := NewResp()
processChan <- Event{eventNil, resp, "", true} // trigger process
resp.ACKWait() // wait for the ACK()
}
}
}
// TODO: expand the IfCmd to be a list of commands
func (obj *ExecRes) CheckApply(apply bool) (stateok bool, err error) {
log.Printf("%v[%v]: CheckApply(%t)", obj.Kind(), obj.GetName(), apply)
// if there is a watch command, but no if command, run based on state
if obj.WatchCmd != "" && obj.IfCmd == "" {
if obj.isStateOK {
return true, nil
}
// if there is no watcher, but there is an onlyif check, run it to see
} else if obj.IfCmd != "" { // && obj.WatchCmd == ""
// there is a watcher, but there is also an if command
//} else if obj.IfCmd != "" && obj.WatchCmd != "" {
if obj.PollInt > 0 { // && obj.WatchCmd == ""
// XXX have the Watch() command output onlyif poll events...
// XXX we can optimize by saving those results for returning here
// return XXX
}
var cmdName string
var cmdArgs []string
if obj.IfShell == "" {
// call without a shell
// FIXME: are there still whitespace splitting issues?
split := strings.Fields(obj.IfCmd)
cmdName = split[0]
//d, _ := os.Getwd() // TODO: how does this ever error ?
//cmdName = path.Join(d, cmdName)
cmdArgs = split[1:len(split)]
} else {
cmdName = obj.IfShell // usually bash, or sh
cmdArgs = []string{"-c", obj.IfCmd}
}
err = exec.Command(cmdName, cmdArgs...).Run()
if err != nil {
// TODO: check exit value
return true, nil // don't run
}
// if there is no watcher and no onlyif check, assume we should run
} else { // if obj.WatchCmd == "" && obj.IfCmd == "" {
// just run if state is dirty
if obj.isStateOK {
return true, nil
}
}
// state is not okay, no work done, exit, but without error
if !apply {
return false, nil
}
// apply portion
log.Printf("%v[%v]: Apply", obj.Kind(), obj.GetName())
var cmdName string
var cmdArgs []string
if obj.Shell == "" {
// call without a shell
// FIXME: are there still whitespace splitting issues?
// TODO: we could make the split character user selectable...!
split := strings.Fields(obj.Cmd)
cmdName = split[0]
//d, _ := os.Getwd() // TODO: how does this ever error ?
//cmdName = path.Join(d, cmdName)
cmdArgs = split[1:len(split)]
} else {
cmdName = obj.Shell // usually bash, or sh
cmdArgs = []string{"-c", obj.Cmd}
}
cmd := exec.Command(cmdName, cmdArgs...)
//cmd.Dir = "" // look for program in pwd ?
var out bytes.Buffer
cmd.Stdout = &out
if err = cmd.Start(); err != nil {
log.Printf("%v[%v]: Error starting Cmd: %v", obj.Kind(), obj.GetName(), err)
return false, err
}
timeout := obj.Timeout
if timeout == 0 { // zero timeout means no timer, so disable it
timeout = -1
}
done := make(chan error)
go func() { done <- cmd.Wait() }()
select {
case err = <-done:
if err != nil {
log.Printf("%v[%v]: Error waiting for Cmd: %v", obj.Kind(), obj.GetName(), err)
return false, err
}
case <-TimeAfterOrBlock(timeout):
log.Printf("%v[%v]: Timeout waiting for Cmd", obj.Kind(), obj.GetName())
//cmd.Process.Kill() // TODO: is this necessary?
return false, errors.New("Timeout waiting for Cmd!")
}
// TODO: if we printed the stdout while the command is running, this
// would be nice, but it would require terminal log output that doesn't
// interleave all the parallel parts which would mix it all up...
if s := out.String(); s == "" {
log.Printf("Exec[%v]: Command output is empty!", obj.Name)
} else {
log.Printf("Exec[%v]: Command output is:", obj.Name)
log.Printf(out.String())
}
// XXX: return based on exit value!!
// the state tracking is for exec resources that can't "detect" their
// state, and assume it's invalid when the Watch() function triggers.
// if we apply state successfully, we should reset it here so that we
// know that we have applied since the state was set not ok by event!
obj.isStateOK = true // reset
return false, nil // success
}
type ExecUUID struct {
BaseUUID
Cmd string
IfCmd string
// TODO: add more elements here
}
// if and only if they are equivalent, return true
// if they are not equivalent, return false
func (obj *ExecUUID) IFF(uuid ResUUID) bool {
res, ok := uuid.(*ExecUUID)
if !ok {
return false
}
if obj.Cmd != res.Cmd {
return false
}
// TODO: add more checks here
//if obj.Shell != res.Shell {
// return false
//}
//if obj.Timeout != res.Timeout {
// return false
//}
//if obj.WatchCmd != res.WatchCmd {
// return false
//}
//if obj.WatchShell != res.WatchShell {
// return false
//}
if obj.IfCmd != res.IfCmd {
return false
}
//if obj.PollInt != res.PollInt {
// return false
//}
//if obj.State != res.State {
// return false
//}
return true
}
func (obj *ExecRes) AutoEdges() AutoEdge {
// TODO: parse as many exec params to look for auto edges, for example
// the path of the binary in the Cmd variable might be from in a pkg
return nil
}
// include all params to make a unique identification of this object
func (obj *ExecRes) GetUUIDs() []ResUUID {
x := &ExecUUID{
BaseUUID: BaseUUID{name: obj.GetName(), kind: obj.Kind()},
Cmd: obj.Cmd,
IfCmd: obj.IfCmd,
// TODO: add more params here
}
return []ResUUID{x}
}
func (obj *ExecRes) GroupCmp(r Res) bool {
_, ok := r.(*SvcRes)
if !ok {
return false
}
return false // not possible atm
}
func (obj *ExecRes) Compare(res Res) bool {
switch res.(type) {
case *ExecRes:
res := res.(*ExecRes)
if obj.Name != res.Name {
return false
}
if obj.Cmd != res.Cmd {
return false
}
if obj.Shell != res.Shell {
return false
}
if obj.Timeout != res.Timeout {
return false
}
if obj.WatchCmd != res.WatchCmd {
return false
}
if obj.WatchShell != res.WatchShell {
return false
}
if obj.IfCmd != res.IfCmd {
return false
}
if obj.PollInt != res.PollInt {
return false
}
if obj.State != res.State {
return false
}
default:
return false
}
return true
}