Replace `actionc` with a mutex
As much as I love finding places to put channels, in this case it is entirely
unnecessary. The only use of actionc can be replaced with a mutex, and any
future methods can just compete for the lock instead of competing for a channel
write.
@@ -3,6 +3,7 @@ package kubernetes
import (
"bytes"
"fmt"
+ "sync"
k8syaml "github.com/ghodss/yaml"
"github.com/go-kit/kit/log"
@@ -92,10 +93,11 @@ type Applier interface {
type Cluster struct {
client extendedClient
applier Applier
- actionc chan func()
version string // string response for the version command.
logger log.Logger
sshKeyRing ssh.KeyRing
+
+ mu sync.Mutex
}
// NewCluster returns a usable cluster. Host should be of the form
@@ -114,27 +116,13 @@ func NewCluster(clientset k8sclient.Interface,
clientset.BatchV2alpha1(),
},
applier: applier,
- actionc: make(chan func()),
logger: logger,
sshKeyRing: sshKeyRing,
}
- go c.loop()
return c, nil
}
-// Stop terminates the goroutine that serializes and executes requests against
-// the cluster. A stopped cluster cannot be restarted.
-func (c *Cluster) Stop() {
- close(c.actionc)
-}
-
-func (c *Cluster) loop() {
- for f := range c.actionc {
- f()
- }
-}
-
// --- cluster.Cluster
// SomeControllers returns the controllers named, missing out any that don't
@@ -202,17 +190,17 @@ func (c *Cluster) AllControllers(namespace string) (res []cluster.Controller, er
// Sync performs the given actions on resources. Operations are
// asynchronous, but serialised.
func (c *Cluster) Sync(spec cluster.SyncDef) error {
- errc := make(chan error)
+ c.mu.Lock()
+ defer c.mu.Unlock()
logger := log.With(c.logger, "method", "Sync")
- c.actionc <- func() {
+
errs := cluster.SyncError{}
for _, action := range spec.Actions {
if len(action.Delete) > 0 {
obj, err := definitionObj(action.Delete)
if err == nil {
c.applier.stageDelete(action.ResourceID, obj)
- }
- if err != nil {
+ } else {
errs[action.ResourceID] = err
continue
}
@@ -221,16 +209,13 @@ func (c *Cluster) Sync(spec cluster.SyncDef) error {
obj, err := definitionObj(action.Apply)
if err == nil {
c.applier.stageApply(action.ResourceID, obj)
- }
- if err != nil {
+ } else {
errs[action.ResourceID] = err
continue
}
}
}
- errc <- c.applier.execute(logger, errs)
- }
- return <-errc
+ return c.applier.execute(logger, errs)
}
func (c *Cluster) Ping() error {
