<commit_before>// Copyright 2012, 2013 Canonical Ltd.
// Licensed under the AGPLv3, see LICENCE file for details.

package provisioner

import (
	"fmt"
	"sync"

	"launchpad.net/golxc"
	"launchpad.net/juju-core/environs"
	"launchpad.net/juju-core/environs/config"
	"launchpad.net/juju-core/state"
	"launchpad.net/juju-core/state/watcher"
	"launchpad.net/juju-core/worker"
	"launchpad.net/loggo"
	"launchpad.net/tomb"
)

type ProvisionerType string

var (
	logger = loggo.GetLogger("juju.provisioner")

	// ENVIRON provisioners create machines from the environment
	ENVIRON ProvisionerType = "environ"
	// LXC provisioners create lxc containers on their parent machine
	LXC ProvisionerType = "lxc"
)

// Provisioner represents a running provisioning worker.
type Provisioner struct {
	pt        ProvisionerType
	st        *state.State
	machineId string // Which machine runs the provisioner.
	machine   *state.Machine
	environ   environs.Environ
	tomb      tomb.Tomb

	configObserver
}

type configObserver struct {
	sync.Mutex
	observer chan<- *config.Config
}

// nofity notifies the observer of a configuration change.
func (o *configObserver) notify(cfg *config.Config) {
	o.Lock()
	if o.observer != nil {
		o.observer <- cfg
	}
	o.Unlock()
}

// NewProvisioner returns a new Provisioner. When new machines
// are added to the state, it allocates instances from the environment
// and allocates them to the new machines.
func NewProvisioner(pt ProvisionerType, st *state.State, machineId string) *Provisioner {
	p := &Provisioner{
		pt:        pt,
		st:        st,
		machineId: machineId,
	}
	go func() {
		defer p.tomb.Done()
		p.tomb.Kill(p.loop())
	}()
	return p
}

func (p *Provisioner) loop() error {
	environWatcher := p.st.WatchEnvironConfig()
	defer watcher.Stop(environWatcher, &p.tomb)

	var err error
	p.environ, err = worker.WaitForEnviron(environWatcher, p.tomb.Dying())
	if err != nil {
		return err
	}

	auth, err := NewSimpleAuthenticator(p.environ)
	if err != nil {
		return err
	}

	// Start a new worker for the environment provider.

	// Start responding to changes in machines, and to any further updates
	// to the environment config.
	machineBroker, err := p.getBroker()
	if err != nil {
		return err
	}
	machineWatcher, err := p.getWatcher()
	if err != nil {
		return err
	}
	environmentProvisioner := NewProvisionerTask(
		p.machineId,
		p.st,
		machineWatcher,
		machineBroker,
		auth)
	defer watcher.Stop(environmentProvisioner, &p.tomb)

	for {
		select {
		case <-p.tomb.Dying():
			return tomb.ErrDying
		case <-environmentProvisioner.Dying():
			err := environmentProvisioner.Err()
			logger.Errorf("environment provisioner died: %v", err)
			return err
		case cfg, ok := <-environWatcher.Changes():
			if !ok {
				return watcher.MustErr(environWatcher)
			}
			if err := p.setConfig(cfg); err != nil {
				logger.Errorf("loaded invalid environment configuration: %v", err)
			}
		}
	}
	panic("not reached")
}

func (p *Provisioner) getMachine() (*state.Machine, error) {
	if p.machine == nil {
		var err error
		if p.machine, err = p.st.Machine(p.machineId); err != nil {
			logger.Errorf("machine %s is not in state", p.machineId)
			return nil, err
		}
	}
	return p.machine, nil
}

func (p *Provisioner) getWatcher() (Watcher, error) {
	switch p.pt {
	case ENVIRON:
		return p.st.WatchEnvironMachines(), nil
	case LXC:
		machine, err := p.getMachine()
		if err != nil {
			return nil, err
		}
		return machine.WatchContainers(state.LXC), nil
	}
	return nil, fmt.Errorf("unknown provisioner type")<commit_msg>orrible ack since jujud doesn't restart individual tasks yet.<commit_after>// Copyright 2012, 2013 Canonical Ltd.
// Licensed under the AGPLv3, see LICENCE file for details.

package provisioner

import (
	"fmt"
	"sync"
	"time"

	"launchpad.net/golxc"
	"launchpad.net/juju-core/environs"
	"launchpad.net/juju-core/environs/config"
	"launchpad.net/juju-core/state"
	"launchpad.net/juju-core/state/watcher"
	"launchpad.net/juju-core/worker"
	"launchpad.net/loggo"
	"launchpad.net/tomb"
)

type ProvisionerType string

var (
	logger = loggo.GetLogger("juju.provisioner")

	// ENVIRON provisioners create machines from the environment
	ENVIRON ProvisionerType = "environ"
	// LXC provisioners create lxc containers on their parent machine
	LXC ProvisionerType = "lxc"
)

// Provisioner represents a running provisioning worker.
type Provisioner struct {
	pt        ProvisionerType
	st        *state.State
	machineId string // Which machine runs the provisioner.
	machine   *state.Machine
	environ   environs.Environ
	tomb      tomb.Tomb

	configObserver
}

type configObserver struct {
	sync.Mutex
	observer chan<- *config.Config
}

// nofity notifies the observer of a configuration change.
func (o *configObserver) notify(cfg *config.Config) {
	o.Lock()
	if o.observer != nil {
		o.observer <- cfg
	}
	o.Unlock()
}

// NewProvisioner returns a new Provisioner. When new machines
// are added to the state, it allocates instances from the environment
// and allocates them to the new machines.
func NewProvisioner(pt ProvisionerType, st *state.State, machineId string) *Provisioner {
	p := &Provisioner{
		pt:        pt,
		st:        st,
		machineId: machineId,
	}
	go func() {
		defer p.tomb.Done()
		p.tomb.Kill(p.loop())
	}()
	return p
}

func (p *Provisioner) loop() error {
	environWatcher := p.st.WatchEnvironConfig()
	defer watcher.Stop(environWatcher, &p.tomb)

	var err error
	p.environ, err = worker.WaitForEnviron(environWatcher, p.tomb.Dying())
	if err != nil {
		return err
	}

	auth, err := NewSimpleAuthenticator(p.environ)
	if err != nil {
		return err
	}

	// Start a new worker for the environment provider.

	// Start responding to changes in machines, and to any further updates
	// to the environment config.
	machineBroker, err := p.getBroker()
	if err != nil {
		return err
	}
	machineWatcher, err := p.getWatcher()
	if err != nil {
		return err
	}
	environmentProvisioner := NewProvisionerTask(
		p.machineId,
		p.st,
		machineWatcher,
		machineBroker,
		auth)
	defer watcher.Stop(environmentProvisioner, &p.tomb)

	for {
		select {
		case <-p.tomb.Dying():
			return tomb.ErrDying
		case <-environmentProvisioner.Dying():
			err := environmentProvisioner.Err()
			logger.Errorf("environment provisioner died: %v", err)
			return err
		case cfg, ok := <-environWatcher.Changes():
			if !ok {
				return watcher.MustErr(environWatcher)
			}
			if err := p.setConfig(cfg); err != nil {
				logger.Errorf("loaded invalid environment configuration: %v", err)
			}
		}
	}
	panic("not reached")
}

func (p *Provisioner) getMachine() (*state.Machine, error) {
	if p.machine == nil {
		var err error
		// Give the machiner time to have actually started up and signed in to set tools.
		time.Sleep(1 * time.Second)
		if p.machine, err = p.st.Machine(p.machineId); err != nil {
			logger.Errorf("machine %s is not in state", p.machineId)
			return nil, err
		}
	}
	return p.machine, nil
}

func (p *Provisioner) getWatcher() (Watcher, error) {
	switch p.pt {
	case ENVIRON:
		return p.st.WatchEnvironMachines(), nil
	case LXC:
		machine, err := p.getMachine()
		if err != nil {
			return nil, err
		}
		return machine.WatchContainers(state.LXC), nil
	}
	return nil, fmt.Errorf("unknown provisioner type")
{'fraction_non_alphanumeric': 0.11515151515151516, 'fraction_numerical': 0.0026170798898071624, 'mean_word_length': 3.6189567430025447, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2250237', 'n_tokens_mistral': 2698, 'n_tokens_neox': 2503, 'n_words': 703}