From 07d31ce5e54803d98f2f370b24cd35395d0f6c29 Mon Sep 17 00:00:00 2001 From: Dongsu Park Date: Fri, 19 Aug 2016 15:30:08 +0200 Subject: [PATCH 1/4] fleetctl: fix up minor style issues To be as idiomatic as possible, do so: * make tryWaitForSystemdActiveState() return error instead of int * pass getBlockAttempts(cCmd) to tryWaitForSystemdActiveState() to handle such cases of --no-block or --block-attempts. * make assertSystemdActiveState(), waitForSystemdActiveState() return just error, not with 'found', as it's unnecessary. * remove unnecessary checking against conn == nil. * compare strings with '!=' instead of strings.Compare() != 0 --- fleetctl/fleetctl.go | 49 +++++++++++++++++++++++--------------------- fleetctl/start.go | 6 +++--- 2 files changed, 29 insertions(+), 26 deletions(-) diff --git a/fleetctl/fleetctl.go b/fleetctl/fleetctl.go index 97fc150b6..470bc9b7b 100644 --- a/fleetctl/fleetctl.go +++ b/fleetctl/fleetctl.go @@ -1067,22 +1067,25 @@ func assertUnitState(name string, js job.JobState, out io.Writer) (ret bool) { // tryWaitForSystemdActiveState tries to wait for systemd units to reach an // active state, making use of the systemd's DBUS interface. First it takes one // or more units as input, and ensures that every unit in the []units must be -// in the active state. If yes, return 0. Otherwise return 1. +// in the active state. If yes, return nil. Otherwise return error. // // NOTE: To avoid unnecessary conflicts with unit tests in start_test.go, we -// should return 0 for the case of sd_dbus.New() returning error. In that case, +// should return nil for the case of sd_dbus.New() returning error. In that case, // it just means that the unit test environment based on Travis is not capable // of setting up a DBUS connection of systemd, which would actually work fine // for production systems or functional tests. -func tryWaitForSystemdActiveState(units []string) (ret int) { +func tryWaitForSystemdActiveState(units []string, maxAttempts int) (err error) { + if maxAttempts <= -1 { + for _, name := range units { + stdout("Triggered unit %s start", name) + } + return nil + } + conn, err := sd_dbus.New() if err != nil { stderr("Failed to get systemd-dbus conn: %v", err) - return 0 - } - if conn == nil { - stderr("Got a nil connection") - return 0 + return nil } defer conn.Close() @@ -1090,31 +1093,31 @@ func tryWaitForSystemdActiveState(units []string) (ret int) { // If any unit in []units cannot be found, then return an error. // That should be the only case of returning 1. for _, uName := range units { - found, err := waitForSystemdActiveState(conn, uName) - if !found || err != nil { + err := waitForSystemdActiveState(conn, uName) + if err != nil { stderr("cannot find an active unit: %v", err) - return 1 + return err } } - return 0 + return nil } // waitForSystemdActiveState tries to assert that the given unit becomes // active, retrying periodically until the unit gets active. It will retry -// up to maxAttempts. -func waitForSystemdActiveState(conn *sd_dbus.Conn, unitName string) (found bool, err error) { - return func() (found bool, errWait error) { +// up to 15 seconds. +func waitForSystemdActiveState(conn *sd_dbus.Conn, unitName string) (err error) { + return func() error { timeout := 15 * time.Second alarm := time.After(timeout) ticker := time.Tick(250 * time.Millisecond) for { select { case <-alarm: - return false, fmt.Errorf("Failed to reach expected state within %v.", timeout) + return fmt.Errorf("Failed to reach expected state within %v.", timeout) case <-ticker: - if found, errA := assertSystemdActiveState(conn, unitName); found && errA == nil { - return found, nil + if errA := assertSystemdActiveState(conn, unitName); errA == nil { + return nil } } } @@ -1123,22 +1126,22 @@ func waitForSystemdActiveState(conn *sd_dbus.Conn, unitName string) (found bool, // assertSystemdActiveState determines if a given systemd unit is actually // in the active state, making use of systemd's DBUS connection. -func assertSystemdActiveState(conn *sd_dbus.Conn, unitName string) (found bool, err error) { +func assertSystemdActiveState(conn *sd_dbus.Conn, unitName string) (err error) { listUnits, err := conn.ListUnits() if err != nil { - return false, fmt.Errorf("Failed to list systemd unit: %v", err) + return fmt.Errorf("Failed to list systemd unit: %v", err) } for _, u := range listUnits { - if strings.Compare(u.Name, unitName) != 0 { + if u.Name != unitName { continue } if u.ActiveState != "active" || u.LoadState != "loaded" { - return false, fmt.Errorf("Failed to find an active unit %s", unitName) + return fmt.Errorf("Failed to find an active unit %s", unitName) } } - return true, nil + return nil } func machineState(machID string) (*machine.MachineState, error) { diff --git a/fleetctl/start.go b/fleetctl/start.go index c7f2982cb..6b8ac6edc 100644 --- a/fleetctl/start.go +++ b/fleetctl/start.go @@ -87,9 +87,9 @@ func runStartUnit(cCmd *cobra.Command, args []string) (exit int) { return exitVal } - exitVal = tryWaitForSystemdActiveState(starting) - if exitVal != 0 { - return exitVal + if err := tryWaitForSystemdActiveState(starting, getBlockAttempts(cCmd)); err != nil { + stderr("Error waiting for systemd unit states, err: %v", err) + return 1 } return 0 From 57b583f75d10d481ba5df3f2935e6840a9fcf8b4 Mon Sep 17 00:00:00 2001 From: Dongsu Park Date: Fri, 19 Aug 2016 16:05:00 +0200 Subject: [PATCH 2/4] fleetctl: make tryWaitForUnitStates return error instead of int Now that tryWaitForSystemdActiveState() returns error instead of int, tryWaitForUnitStates() should also return error instead of int, to be as idiomatic as possible. --- fleetctl/fleetctl.go | 10 +++++----- fleetctl/load.go | 6 +++--- fleetctl/start.go | 7 +++---- fleetctl/stop.go | 12 ++++++------ fleetctl/unload.go | 12 ++++++------ 5 files changed, 23 insertions(+), 24 deletions(-) diff --git a/fleetctl/fleetctl.go b/fleetctl/fleetctl.go index 470bc9b7b..8544c432f 100644 --- a/fleetctl/fleetctl.go +++ b/fleetctl/fleetctl.go @@ -960,23 +960,23 @@ func getBlockAttempts(cCmd *cobra.Command) int { // wait, it will assume that all units reached their desired state. // If maxAttempts is zero tryWaitForUnitStates will retry forever, and // if it is greater than zero, it will retry up to the indicated value. -// It returns 0 on success or 1 on errors. -func tryWaitForUnitStates(units []string, state string, js job.JobState, maxAttempts int, out io.Writer) (ret int) { +// It returns nil on success or error on failure. +func tryWaitForUnitStates(units []string, state string, js job.JobState, maxAttempts int, out io.Writer) error { // We do not wait just assume we reached the desired state if maxAttempts <= -1 { for _, name := range units { stdout("Triggered unit %s %s", name, state) } - return + return nil } errchan := waitForUnitStates(units, js, maxAttempts, out) for err := range errchan { stderr("Error waiting for units: %v", err) - ret = 1 + return err } - return + return nil } // waitForUnitStates polls each of the indicated units until each of their diff --git a/fleetctl/load.go b/fleetctl/load.go index bf3265888..8689911a9 100644 --- a/fleetctl/load.go +++ b/fleetctl/load.go @@ -73,9 +73,9 @@ func runLoadUnit(cCmd *cobra.Command, args []string) (exit int) { } } - exitVal := tryWaitForUnitStates(loading, "load", job.JobStateLoaded, getBlockAttempts(cCmd), os.Stdout) - if exitVal != 0 { - stderr("Error waiting for unit states, exit status: %d", exitVal) + err = tryWaitForUnitStates(loading, "load", job.JobStateLoaded, getBlockAttempts(cCmd), os.Stdout) + if err != nil { + stderr("Error waiting for unit states, exit status: %v", err) return 1 } diff --git a/fleetctl/start.go b/fleetctl/start.go index 6b8ac6edc..d005da458 100644 --- a/fleetctl/start.go +++ b/fleetctl/start.go @@ -81,10 +81,9 @@ func runStartUnit(cCmd *cobra.Command, args []string) (exit int) { } } - exitVal := tryWaitForUnitStates(starting, "start", job.JobStateLaunched, getBlockAttempts(cCmd), os.Stdout) - if exitVal != 0 { - stderr("Error waiting for unit states, exit status: %d", exitVal) - return exitVal + if err := tryWaitForUnitStates(starting, "start", job.JobStateLaunched, getBlockAttempts(cCmd), os.Stdout); err != nil { + stderr("Error waiting for unit states, exit status: %v", err) + return 1 } if err := tryWaitForSystemdActiveState(starting, getBlockAttempts(cCmd)); err != nil { diff --git a/fleetctl/stop.go b/fleetctl/stop.go index 8a6dd60a2..517ae9e3d 100644 --- a/fleetctl/stop.go +++ b/fleetctl/stop.go @@ -92,12 +92,12 @@ func runStopUnit(cCmd *cobra.Command, args []string) (exit int) { } } - exit = tryWaitForUnitStates(stopping, "stop", job.JobStateLoaded, getBlockAttempts(cCmd), os.Stdout) - if exit == 0 { - stderr("Successfully stopped units %v.", stopping) - } else { - stderr("Failed to stop units %v. exit == %d.", stopping, exit) + err = tryWaitForUnitStates(stopping, "stop", job.JobStateLoaded, getBlockAttempts(cCmd), os.Stdout) + if err != nil { + stderr("Failed to stop units %v. err: %v", stopping, err) + return 1 } - return + stderr("Successfully stopped units %v.", stopping) + return 0 } diff --git a/fleetctl/unload.go b/fleetctl/unload.go index c5bb5050f..be2ffdf1a 100644 --- a/fleetctl/unload.go +++ b/fleetctl/unload.go @@ -71,12 +71,12 @@ func runUnloadUnit(cCmd *cobra.Command, args []string) (exit int) { } } - exit = tryWaitForUnitStates(wait, "unload", job.JobStateInactive, getBlockAttempts(cCmd), os.Stdout) - if exit == 0 { - stderr("Successfully unloaded units %v.", wait) - } else { - stderr("Failed to unload units %v. exit == %d.", wait, exit) + err = tryWaitForUnitStates(wait, "unload", job.JobStateInactive, getBlockAttempts(cCmd), os.Stdout) + if err != nil { + stderr("Failed to unload units %v. err: %v", wait, err) + return 1 } - return + stderr("Successfully unloaded units %v.", wait) + return 0 } From cc2837f8bbcdb68bb336101ffc491f1ae8094b0a Mon Sep 17 00:00:00 2001 From: Dongsu Park Date: Tue, 6 Sep 2016 15:43:34 +0200 Subject: [PATCH 3/4] fleetctl: check systemd active state via cAPI.UnitStates() It's not appropriate to check systemd active states with DBUS connection, as that will return correct states only on the same machine. So we should introduce a check using cAPI.UnitStates(), to check states on other machines. --- fleetctl/fleetctl.go | 73 +++++++++++++++++++++++--------------------- 1 file changed, 39 insertions(+), 34 deletions(-) diff --git a/fleetctl/fleetctl.go b/fleetctl/fleetctl.go index 8544c432f..9329d96d5 100644 --- a/fleetctl/fleetctl.go +++ b/fleetctl/fleetctl.go @@ -34,7 +34,6 @@ import ( "github.com/spf13/pflag" etcd "github.com/coreos/etcd/client" - sd_dbus "github.com/coreos/go-systemd/dbus" "github.com/coreos/fleet/api" "github.com/coreos/fleet/client" @@ -177,6 +176,12 @@ type Command struct { } +type suState struct { + LoadState string + ActiveState string + SubState string +} + func getFlags(flagset *flag.FlagSet) (flags []*flag.Flag) { flags = make([]*flag.Flag, 0) flagset.VisitAll(func(f *flag.Flag) { @@ -1065,15 +1070,9 @@ func assertUnitState(name string, js job.JobState, out io.Writer) (ret bool) { } // tryWaitForSystemdActiveState tries to wait for systemd units to reach an -// active state, making use of the systemd's DBUS interface. First it takes one -// or more units as input, and ensures that every unit in the []units must be -// in the active state. If yes, return nil. Otherwise return error. -// -// NOTE: To avoid unnecessary conflicts with unit tests in start_test.go, we -// should return nil for the case of sd_dbus.New() returning error. In that case, -// it just means that the unit test environment based on Travis is not capable -// of setting up a DBUS connection of systemd, which would actually work fine -// for production systems or functional tests. +// active state, making use of cAPI. It takes one or more units as input, and +// ensures that every unit in the []units must be in the active state. +// If yes, return nil. Otherwise return error. func tryWaitForSystemdActiveState(units []string, maxAttempts int) (err error) { if maxAttempts <= -1 { for _, name := range units { @@ -1082,18 +1081,8 @@ func tryWaitForSystemdActiveState(units []string, maxAttempts int) (err error) { return nil } - conn, err := sd_dbus.New() - if err != nil { - stderr("Failed to get systemd-dbus conn: %v", err) - return nil - } - defer conn.Close() - - // Get systemd state and check the state is active & loaded. - // If any unit in []units cannot be found, then return an error. - // That should be the only case of returning 1. for _, uName := range units { - err := waitForSystemdActiveState(conn, uName) + err := waitForSystemdActiveState(uName) if err != nil { stderr("cannot find an active unit: %v", err) return err @@ -1106,7 +1095,7 @@ func tryWaitForSystemdActiveState(units []string, maxAttempts int) (err error) { // waitForSystemdActiveState tries to assert that the given unit becomes // active, retrying periodically until the unit gets active. It will retry // up to 15 seconds. -func waitForSystemdActiveState(conn *sd_dbus.Conn, unitName string) (err error) { +func waitForSystemdActiveState(unitName string) (err error) { return func() error { timeout := 15 * time.Second alarm := time.After(timeout) @@ -1116,7 +1105,7 @@ func waitForSystemdActiveState(conn *sd_dbus.Conn, unitName string) (err error) case <-alarm: return fmt.Errorf("Failed to reach expected state within %v.", timeout) case <-ticker: - if errA := assertSystemdActiveState(conn, unitName); errA == nil { + if errA := assertSystemdActiveState(unitName); errA == nil { return nil } } @@ -1125,25 +1114,41 @@ func waitForSystemdActiveState(conn *sd_dbus.Conn, unitName string) (err error) } // assertSystemdActiveState determines if a given systemd unit is actually -// in the active state, making use of systemd's DBUS connection. -func assertSystemdActiveState(conn *sd_dbus.Conn, unitName string) (err error) { - listUnits, err := conn.ListUnits() +// in the active state, making use of cAPI +func assertSystemdActiveState(unitName string) (err error) { + uState, err := getSingleUnitState(unitName) if err != nil { - return fmt.Errorf("Failed to list systemd unit: %v", err) + return err } - for _, u := range listUnits { - if u.Name != unitName { - continue - } - if u.ActiveState != "active" || u.LoadState != "loaded" { - return fmt.Errorf("Failed to find an active unit %s", unitName) - } + // Get systemd state and check the state is active & loaded. + if uState.ActiveState != "active" || uState.LoadState != "loaded" { + return fmt.Errorf("Failed to find an active unit %s", unitName) } return nil } +// getSingleUnitState returns a single uState of type suState, which consists +// of necessary systemd states, only for a given unit name. +func getSingleUnitState(unitName string) (suState, error) { + apiStates, err := cAPI.UnitStates() + if err != nil { + return suState{}, fmt.Errorf("Error retrieving list of units: %v", err) + } + + for _, us := range apiStates { + if us.Name == unitName { + return suState{ + us.SystemdLoadState, + us.SystemdActiveState, + us.SystemdSubState, + }, nil + } + } + return suState{}, fmt.Errorf("unit %s not found", unitName) +} + func machineState(machID string) (*machine.MachineState, error) { machines, err := cAPI.Machines() if err != nil { From 988fd84899ff4c87351f5e26e34752867c1b4d18 Mon Sep 17 00:00:00 2001 From: Dongsu Park Date: Mon, 19 Sep 2016 14:19:53 +0200 Subject: [PATCH 4/4] fleetctl: call cAPI.UnitStates() only once before checking unit states As it's basically not efficient to call cAPI.UnitStates() for each unit, let's call it only once outside goroutines for processing units, i.e. calling in waitForSystemdActiveState(), not in getSingleUnitState(). To do that, we need to share the result apiStates across all units. Also make use of goroutines in waitForSystemdActiveState(), instead of sequential iteration over each unit. Note that we still need to keep calling UnitStates() when assertSystemdActiveState() failed, because in the next attempt the old apiState is not necessarily going to be valid. Ideally in that case, we should be able to fetch the state only for a single unit. However, we cannot do that for now, because cAPI.UnitState() is not available. In the future we would need to implement cAPI.UnitState() and all dependendent parts all over the tree in fleet, e.g. schema, etcdRegistry, rpcRegistry, etc. to replace UnitStates() in this place with the new method UnitState(). In practice, calling UnitStates() here is not as badly inefficient as it looks, because it will be anyway rarely called only when the assertion failed. --- fleetctl/fleetctl.go | 110 +++++++++++++++++++++++++++++++------------ 1 file changed, 79 insertions(+), 31 deletions(-) diff --git a/fleetctl/fleetctl.go b/fleetctl/fleetctl.go index 9329d96d5..c588736ef 100644 --- a/fleetctl/fleetctl.go +++ b/fleetctl/fleetctl.go @@ -20,6 +20,7 @@ import ( "fmt" "io" "io/ioutil" + "math" "net" "net/http" "net/url" @@ -1081,42 +1082,94 @@ func tryWaitForSystemdActiveState(units []string, maxAttempts int) (err error) { return nil } - for _, uName := range units { - err := waitForSystemdActiveState(uName) - if err != nil { - stderr("cannot find an active unit: %v", err) - return err - } + errchan := waitForSystemdActiveState(units, maxAttempts) + for err := range errchan { + stderr("Error waiting for units: %v", err) + return err } return nil } // waitForSystemdActiveState tries to assert that the given unit becomes -// active, retrying periodically until the unit gets active. It will retry -// up to 15 seconds. -func waitForSystemdActiveState(unitName string) (err error) { - return func() error { - timeout := 15 * time.Second - alarm := time.After(timeout) - ticker := time.Tick(250 * time.Millisecond) - for { - select { - case <-alarm: - return fmt.Errorf("Failed to reach expected state within %v.", timeout) - case <-ticker: - if errA := assertSystemdActiveState(unitName); errA == nil { - return nil - } - } - } +// active, making use of multiple goroutines that check unit states. +func waitForSystemdActiveState(units []string, maxAttempts int) (errch chan error) { + apiStates, err := cAPI.UnitStates() + if err != nil { + errch <- fmt.Errorf("Error retrieving list of units: %v", err) + return + } + + errchan := make(chan error) + var wg sync.WaitGroup + for _, name := range units { + wg.Add(1) + go checkSystemdActiveState(apiStates, name, maxAttempts, &wg, errchan) + } + + go func() { + wg.Wait() + close(errchan) }() + + return errchan +} + +func checkSystemdActiveState(apiStates []*schema.UnitState, name string, maxAttempts int, wg *sync.WaitGroup, errchan chan error) { + defer wg.Done() + + // "isInf == true" means "blocking forever until it succeeded". + // In that case, maxAttempts is set to an arbitrary large integer number. + var isInf bool + if maxAttempts < 1 { + isInf = true + maxAttempts = math.MaxInt32 + } + + for attempt := 0; attempt < maxAttempts; attempt++ { + if err := assertFetchSystemdActiveState(apiStates, name); err == nil { + return + } else { + errchan <- err + } + + if !isInf { + errchan <- fmt.Errorf("timed out waiting for unit %s to report active state", name) + } + } +} + +func assertFetchSystemdActiveState(apiStates []*schema.UnitState, name string) error { + if err := assertSystemdActiveState(apiStates, name); err == nil { + return nil + } + + // If the assertion failed, we again need to get unit states via cAPI, + // to retry the assertion repeatedly. + // + // NOTE: Ideally we should be able to fetch the state only for a single + // unit. However, we cannot do that for now, because cAPI.UnitState() + // is not available. In the future we need to implement cAPI.UnitState() + // and all dependendent parts all over the tree in fleet, (schema, + // etcdRegistry, rpcRegistry, etc) to replace UnitStates() in this place + // with the new method UnitState(). In practice, calling UnitStates() here + // is not as badly inefficient as it looks, because it will be anyway + // rarely called only when the assertion failed. - dpark 20160907 + + time.Sleep(defaultSleepTime) + + var errU error + apiStates, errU = cAPI.UnitStates() + if errU != nil { + return fmt.Errorf("Error retrieving list of units: %v", errU) + } + return nil } // assertSystemdActiveState determines if a given systemd unit is actually // in the active state, making use of cAPI -func assertSystemdActiveState(unitName string) (err error) { - uState, err := getSingleUnitState(unitName) +func assertSystemdActiveState(apiStates []*schema.UnitState, unitName string) error { + uState, err := getSingleUnitState(apiStates, unitName) if err != nil { return err } @@ -1131,12 +1184,7 @@ func assertSystemdActiveState(unitName string) (err error) { // getSingleUnitState returns a single uState of type suState, which consists // of necessary systemd states, only for a given unit name. -func getSingleUnitState(unitName string) (suState, error) { - apiStates, err := cAPI.UnitStates() - if err != nil { - return suState{}, fmt.Errorf("Error retrieving list of units: %v", err) - } - +func getSingleUnitState(apiStates []*schema.UnitState, unitName string) (suState, error) { for _, us := range apiStates { if us.Name == unitName { return suState{