cmd/jujud: strawman provisioning agent

cmd/jujud/provisioning.go

 package main
 
 import (
-        "fmt"
         "launchpad.net/gnuflag"
         "launchpad.net/juju/go/cmd"
+        "launchpad.net/juju/go/environs"
+        "launchpad.net/juju/go/log"
+        "launchpad.net/juju/go/state"
+        "launchpad.net/tomb"
+
+        // register providers
+        _ "launchpad.net/juju/go/environs/dummy"
+        _ "launchpad.net/juju/go/environs/ec2"
 )
 
 // ProvisioningAgent is a cmd.Command responsible for running a provisioning agent.
 type ProvisioningAgent struct {
         Conf AgentConf
 }
 
 // Info returns usage information for the command.
 func (a *ProvisioningAgent) Info() *cmd.Info {
         return &cmd.Info{"provisioning", "", "run a juju provisioning agent", ""}
 }
 
 // Init initializes the command for running.
 func (a *ProvisioningAgent) Init(f *gnuflag.FlagSet, args []string) error {
         a.Conf.addFlags(f)
         if err := f.Parse(true, args); err != nil {
                 return err
         }
         return a.Conf.checkArgs(f.Args())
 }
 
 // Run runs a provisioning agent.
 func (a *ProvisioningAgent) Run(_ *cmd.Context) error {
-        return fmt.Errorf("MachineAgent.Run not implemented")
+        st, err := state.Open(&a.Conf.StateInfo)

[niemeyer, 2012/05/31 11:46:59]

What happens if the connection is broken? We'll need to reestablish it and
restart from the beginning, since all the assumptions are now wrong. Maybe this
has to be moved into the outer loop or something. Will leave details with you at
this point.

That said, I suggest keeping this for the branch coming right after this one. It
feels like what we have here is a solid step forward, and we can comfortably
iterate over the state reestablishment in a new branch before continuing with
everything else.

[dfc, 2012/05/31 22:40:03]

I have been operating under the assumption (I think it's correct based on what
William and Roger were talking about yesterday) that all the agents will run
under upstart so can assume that if they exit or SEGV they will be restarted.
I agree, i'd like to see that as a future enhancement, and lean on upstart for
the moment. See also comments about state.IsClosed/IsValid below.

[dfc, 2012/06/01 01:58:27]

I looked into this more this morning and it looks like if the underlying
connection to ZK is interrupted then the c client reconnects behind the scenes.
I tested this by running the PA, then shutting down zk, then starting it again a
minute later. None of the watchers exited.

I'm going to keep playing with this, but at this point, if gozk never reports an
error when it cant talk to the zk server, I can't think of a way to detect
connection failure.

Cheers

Dave

[niemeyer, 2012/06/04 23:22:23]

We've covered this on juju-dev@. A TODO would be good here as well.

+        if err != nil {
+                return err
+        }
+        p := NewProvisioner(st)
+        return p.tomb.Wait()
 }
+
+type Provisioner struct {
+        st      *state.State
+        environ environs.Environ
+        tomb    tomb.Tomb
+
+        environment
+        machines
+}
+
+// environment ensures that the watcher for the environ
+// configuration is valid.
+type environment struct {

[niemeyer, 2012/05/31 11:46:59]

Those two types feel unnecessary. It looks like your model would work equally
well if we had environWatcher and machinesWatcher both within the Provisioner
struct in place of the environment and machines fields, plus stopMachinesWatcher
and stopEnvironWatcher methods (the two invalidate ones), with no further
changes.

[dfc, 2012/05/31 22:40:03]

This was an indulgence. I am besotted with the ability to bind a few methods to
a structure and then embed the structure as a way of composing more complex
types. I've redone it, but retained a wrapper methods.

+        st      *state.State
+        watcher *state.ConfigWatcher
+}
+
+// changes returns a channel that will receive the new *ConfigNode when a
+// change is detected.·
+func (e *environment) changes() <-chan *state.ConfigNode {
+        if e.watcher == nil {
+                e.watcher = e.st.WatchEnvironConfig()
+        }
+        return e.watcher.Changes()
+}
+
+// invalidate stops the current watcher.
+func (e *environment) invalidate() {

[niemeyer, 2012/05/31 11:46:59]

This should return the error it finds. Call sites can selectively ignore it, but
methods such as Provisioner.Stop should return the problems it finds, so we can
have a handle on them if we want (testing should definitely check, for
instance).

[dfc, 2012/05/31 22:40:03]

Done.

+        if e.watcher != nil {
+                log.Printf("provisioner: environment watcher exited: %v", e.watcher.Stop())

[niemeyer, 2012/05/31 11:46:59]

That Stop call is the most critical task done by this method. It shouldn't be
disguised at the end of a log call.

[dfc, 2012/05/31 22:40:03]

Done.

+        }
+        e.watcher = nil
+}
+
+// machines ensures that the watcher for machines changes
+// is valid.
+type machines struct {
+        st      *state.State
+        watcher *state.MachinesWatcher
+}
+
+// changes returns a channel that will receive the new *ConfigNode when a
+// change is detected.·
+func (m *machines) changes() <-chan *state.MachinesChange {
+        if m.watcher == nil {
+                m.watcher = m.st.WatchMachines()
+        }
+        return m.watcher.Changes()
+}
+
+// invalidate stops the current watcher.
+func (m *machines) invalidate() {
+        if m.watcher != nil {
+                log.Printf("provisioner: machines watcher exited: %v", m.watcher.Stop())
+        }
+        m.watcher = nil
+}
+
+// NewProvisioner returns a Provisioner.
+func NewProvisioner(st *state.State) *Provisioner {
+        p := &Provisioner{
+                st:          st,
+                environment: environment{st: st},
+                machines:    machines{st: st},
+        }
+        go p.loop()
+        return p
+}
+
+func (p *Provisioner) loop() {
+        defer p.tomb.Done()
+        for {
+                select {
+                case <-p.tomb.Dying():
+                        return
+                case config, ok := <-p.environment.changes():
+                        if !ok {
+                                p.environment.invalidate()
+                                continue
+                        }
+                        var err error
+                        p.environ, err = environs.NewEnviron(config.Map())
+                        if err != nil {
+                                log.Printf("provisioner: unable to create environment from supplied configuration: %v", err)
+                                continue
+                        }
+                        log.Printf("provisioning: valid environment configured")
+                        p.innerLoop()
+                }
+        }
+}
+
+func (p *Provisioner) innerLoop() {

[niemeyer, 2012/05/31 11:46:59]

Nice organization.

[dfc, 2012/05/31 22:40:03]

Thank you. As much as I would like to have only one loop, there is a hard
requirement to have a valid Environ before reading anything off the machines
channel. To remove that would mean that every machine change has to be recorded
in some structure which is serviced by a goroutine. That means more tomb.Tombs
to check, locks around the Envrion, etc. This is most of what would be required
for a concurrent PA.

I do want to make the PA concurrent in the future, but not just yet.

+        for {
+                select {
+                case <-p.tomb.Dying():
+                        return
+                case change, ok := <-p.environment.changes():
+                        if !ok {
+                                p.environment.invalidate()
+                                continue
+                        }
+                        config, err := environs.NewConfig(change.Map())
+                        if err != nil {
+                                log.Printf("provisioning: new configuration received, but was not valid: %v", err)
+                                continue
+                        }
+                        p.environ.SetConfig(config)
+                        log.Printf("provisioning: new configuartion applied")

[niemeyer, 2012/05/31 11:46:59]

s/configuartion/environment configuration/ (note typo)

[dfc, 2012/05/31 22:40:03]

Done.

+                case machines, ok := <-p.machines.changes():
+                        if !ok {
+                                p.machines.invalidate()
+                                continue
+                        }
+                        p.processMachines(machines)
+                }

[niemeyer, 2012/05/31 11:46:59]

I'm wondering if rather than "continue" in the problems above, we should
"break", and here have something that verifies if the state is still sane? 
Otherwise this will be an infinite loop, won't it?

[dfc, 2012/05/31 22:40:03]

At the moment, by my reading, a watcher is always returned from any of the calls
to state.New*Watcher, but if the underlying state is broken, they will
immediately go into Dead state and close their channel.

What I think is needed here is the state/watcher.go methods to be able to return
an error when you ask for a watcher. Or, a method like state.IsClosed() (or
similar) if it existed could be put a the top of the loop, ie.

for !p.st.IsClosed() { select { ... } }

+        }
+}
+
+// Stop stops the Provisioner and returns any error encountered while
+// provisioning.
+func (p *Provisioner) Stop() error {
+        p.tomb.Kill(nil)
+        p.environment.invalidate()

[niemeyer, 2012/05/31 11:46:59]

I suggest logic like this here:

p.tomb.Kill(nil)
err1 := p.stopEnvironWatcher()
err2 := p.stopMachinesWatcher()
err3 := p.tomb.Wait()
for err := []error{err3, err2, err1} {
    if err != nil {
        return err
    }
}
return nil

[dfc, 2012/05/31 22:40:03]

SGTM. I'm never sure how to merge several error values into one.

+        p.machines.invalidate()
+        return p.tomb.Wait()
+}
+
+func (p *Provisioner) processMachines(changes *state.MachinesChange) {}