environ: add SetConfig

environs/ec2/ec2.go

 package ec2
 
 import (
         "fmt"
         "launchpad.net/goamz/ec2"
         "launchpad.net/goamz/s3"
         "launchpad.net/juju/go/environs"
         "launchpad.net/juju/go/log"
         "launchpad.net/juju/go/state"
         "launchpad.net/juju/go/version"
+        "sync"
         "time"
 )
 
 const zkPort = 2181
 
 var zkPortSuffix = fmt.Sprintf(":%d", zkPort)
 
 // A request may fail to due "eventual consistency" semantics, which
 // should resolve fairly quickly.  A request may also fail due to a slow
 // state transition (for instance an instance taking a while to release
@@ skipping 11 matching lines @@
 
 func init() {
         environs.RegisterProvider("ec2", environProvider{})
 }
 
 type environProvider struct{}
 
 var _ environs.EnvironProvider = environProvider{}
 
 type environ struct {
-        name          string
-        config        *providerConfig
-        ec2           *ec2.EC2
-        s3            *s3.S3
-        storage       storage
-        publicStorage *storage // optional.
+        name string
+
+        // configMutex protects the *Unlocked fields below.
+        configMutex           sync.Mutex
+        configUnlocked        *providerConfig
+        ec2Unlocked           *ec2.EC2
+        s3Unlocked            *s3.S3
+        storageUnlocked       *storage
+        publicStorageUnlocked *storage // optional.
 }
 
 var _ environs.Environ = (*environ)(nil)
 
 type instance struct {
         e *environ
         *ec2.Instance
 }
 
 func (inst *instance) String() string {
@@ skipping 32 matching lines @@
                 }
         }
         return "", fmt.Errorf("timed out trying to get DNS address for %v", inst.Id())
 }
 
 func (cfg *providerConfig) Open() (environs.Environ, error) {
         log.Printf("environs/ec2: opening environment %q", cfg.name)
         if Regions[cfg.region].EC2Endpoint == "" {
                 return nil, fmt.Errorf("no ec2 endpoint found for region %q, opening %q", cfg.region, cfg.name)
         }
-        e := &environ{
-                config: cfg,
-                ec2:    ec2.New(cfg.auth, Regions[cfg.region]),
-                s3:     s3.New(cfg.auth, Regions[cfg.region]),
+        e := new(environ)
+        e.SetConfig(cfg)
+        return e, nil
+}
+
+func (e *environ) SetConfig(cfg environs.EnvironConfig) {
+        config := cfg.(*providerConfig)
+        e.configMutex.Lock()
+        defer e.configMutex.Unlock()
+        e.configUnlocked = config
+        e.ec2Unlocked = ec2.New(config.auth, Regions[config.region])
+        e.s3Unlocked = s3.New(config.auth, Regions[config.region])
+
+        // create new storage instances, existing instances continue
+        // to reference their existing configuration.
+        e.storageUnlocked = &storage{
+                bucket: e.s3Unlocked.Bucket(config.bucket),
         }
-        e.storage.bucket = e.s3.Bucket(cfg.bucket)
-        if cfg.publicBucket != "" {
-                e.publicStorage = &storage{bucket: e.s3.Bucket(cfg.publicBucket)}
+        if config.publicBucket != "" {
+                e.publicStorageUnlocked = &storage{
+                        bucket: e.s3Unlocked.Bucket(config.publicBucket),
+                }
+        } else {
+                e.publicStorageUnlocked = nil
         }
-        return e, nil
+}
+
+func (e *environ) config() *providerConfig {
+        e.configMutex.Lock()
+        defer e.configMutex.Unlock()
+        return e.configUnlocked
+}
+
+func (e *environ) ec2() *ec2.EC2 {
+        e.configMutex.Lock()
+        defer e.configMutex.Unlock()
+        return e.ec2Unlocked
+}
+
+func (e *environ) s3() *s3.S3 {
+        e.configMutex.Lock()
+        defer e.configMutex.Unlock()
+        return e.s3Unlocked
+}
+
+func (e *environ) Storage() environs.Storage {
+        e.configMutex.Lock()
+        defer e.configMutex.Unlock()
+        return e.storageUnlocked
+}
+
+func (e *environ) PublicStorage() environs.StorageReader {
+        e.configMutex.Lock()
+        defer e.configMutex.Unlock()
+        if e.publicStorageUnlocked == nil {
+                return environs.EmptyStorage
+        }
+        return e.publicStorageUnlocked
 }
 
 func (e *environ) Bootstrap(uploadTools bool) error {
         log.Printf("environs/ec2: bootstrapping environment %q", e.name)
         _, err := e.loadState()
         if err == nil {
                 return fmt.Errorf("environment is already bootstrapped")
         }
         if _, notFound := err.(*environs.NotFoundError); !notFound {
                 return fmt.Errorf("cannot query old bootstrap state: %v", err)
@@ skipping 58 matching lines @@
                 UseSSH: true,
         }, nil
 }
 
 func (e *environ) StartInstance(machineId int, info *state.Info) (environs.Instance, error) {
         log.Printf("environs/ec2: starting machine %d in %q", machineId, e.name)
         return e.startInstance(machineId, info, false)
 }
 
 func (e *environ) userData(machineId int, info *state.Info, master bool, toolsURL string) ([]byte, error) {
+        config := e.config()
         cfg := &machineConfig{
                 provisioner:        master,
                 zookeeper:          master,
                 stateInfo:          info,
                 instanceIdAccessor: "$(curl http://169.254.169.254/1.0/meta-data/instance-id)",
                 providerType:       "ec2",
                 toolsURL:           toolsURL,
                 machineId:          machineId,
         }
 
-        if e.config.authorizedKeys == "" {
+        if config.authorizedKeys == "" {
                 var err error
-                cfg.authorizedKeys, err = authorizedKeys(e.config.authorizedKeysPath)
+                cfg.authorizedKeys, err = authorizedKeys(config.authorizedKeysPath)
                 if err != nil {
                         return nil, fmt.Errorf("cannot get ssh authorized keys: %v", err)
                 }
         }
         cloudcfg, err := newCloudInit(cfg)
         if err != nil {
                 return nil, err
         }
         return cloudcfg.Render()
 }
@@ skipping 15 matching lines @@
         if err != nil {
                 return nil, fmt.Errorf("cannot make user data: %v", err)
         }
         groups, err := e.setUpGroups(machineId)
         if err != nil {
                 return nil, fmt.Errorf("cannot set up groups: %v", err)
         }
         var instances *ec2.RunInstancesResp
 
         for a := shortAttempt.start(); a.next(); {
-                instances, err = e.ec2.RunInstances(&ec2.RunInstances{
+                instances, err = e.ec2().RunInstances(&ec2.RunInstances{
                         ImageId:        spec.imageId,
                         MinCount:       1,
                         MaxCount:       1,
                         UserData:       userData,
                         InstanceType:   "m1.small",
                         SecurityGroups: groups,
                 })
                 if err == nil || ec2ErrCode(err) != "InvalidGroup.NotFound" {
                         break
                 }
@@ skipping 28 matching lines @@
                         need = append(need, ids[i])
                 }
         }
         if len(need) == 0 {
                 return nil
         }
         filter := ec2.NewFilter()
         filter.Add("instance-state-name", "pending", "running")
         filter.Add("group-name", e.groupName())
         filter.Add("instance-id", need...)
-        resp, err := e.ec2.Instances(nil, filter)
+        resp, err := e.ec2().Instances(nil, filter)
         if err != nil {
                 return err
         }
         n := 0
         // For each requested id, add it to the returned instances
         // if we find it in the response.
         for i, id := range ids {
                 if insts[i] != nil {
                         continue
                 }
@@ skipping 42 matching lines @@
         }
         return insts, nil
 }
 
 func (e *environ) Destroy(insts []environs.Instance) error {
         log.Printf("environs/ec2: destroying environment %q", e.name)
         // Try to find all the instances in the environ's group.
         filter := ec2.NewFilter()
         filter.Add("instance-state-name", "pending", "running")
         filter.Add("group-name", e.groupName())
-        resp, err := e.ec2.Instances(nil, filter)
+        resp, err := e.ec2().Instances(nil, filter)
         if err != nil {
                 return fmt.Errorf("cannot get instances: %v", err)
         }
         var ids []string
         found := make(map[string]bool)
         for _, r := range resp.Reservations {
                 for _, inst := range r.Instances {
                         ids = append(ids, inst.InstanceId)
                         found[inst.InstanceId] = true
                 }
         }
 
         // Then add any instances we've been told about but haven't yet shown
         // up in the instance list.
         for _, inst := range insts {
                 id := inst.(*instance).InstanceId
                 if !found[id] {
                         ids = append(ids, id)
                         found[id] = true
                 }
         }
         err = e.terminateInstances(ids)
         if err != nil {
                 return err
         }
-        err = e.storage.deleteAll()
+        // to properly observe e.storageUnlocked we need to get it's value while
+        // holding e.configMutex. e.Storage() does this for us, then we convert
+        // back to the (*storage) to access the private deleteAll() method.
+        st := e.Storage().(*storage)
+        err = st.deleteAll()
         if err != nil {
                 return err
         }
         return nil
 }
 
 func (e *environ) terminateInstances(ids []string) error {
         if len(ids) == 0 {
                 return nil
         }
         var err error
+        ec2inst := e.ec2()
         for a := shortAttempt.start(); a.next(); {
-                _, err = e.ec2.TerminateInstances(ids)
+                _, err = ec2inst.TerminateInstances(ids)
                 if err == nil || ec2ErrCode(err) != "InvalidInstanceID.NotFound" {
                         return err
                 }
         }
         if len(ids) == 1 {
                 return err
         }
         var firstErr error
         // If we get a NotFound error, it means that no instances have been
         // terminated even if some exist, so try them one by one, ignoring
         // NotFound errors.
         for _, id := range ids {
-                _, err = e.ec2.TerminateInstances([]string{id})
+                _, err = ec2inst.TerminateInstances([]string{id})
                 if ec2ErrCode(err) == "InvalidInstanceID.NotFound" {
                         err = nil
                 }
                 if err != nil && firstErr == nil {
                         firstErr = err
                 }
         }
         return firstErr
 }
 
@@ skipping 33 matching lines @@
         return []ec2.SecurityGroup{jujuGroup, jujuMachineGroup}, nil
 }
 
 // zeroGroup holds the zero security group.
 var zeroGroup ec2.SecurityGroup
 
 // ensureGroup returns the security group with name and perms.
 // If a group with name does not exist, one will be created.
 // If it exists, its permissions are set to perms.
 func (e *environ) ensureGroup(name string, perms []ec2.IPPerm) (g ec2.SecurityGroup, err error) {
-        resp, err := e.ec2.CreateSecurityGroup(name, "juju group")
+        ec2inst := e.ec2()
+        resp, err := ec2inst.CreateSecurityGroup(name, "juju group")
         if err != nil && ec2ErrCode(err) != "InvalidGroup.Duplicate" {
                 return zeroGroup, err
         }
 
         want := newPermSet(perms)
         var have permSet
         if err == nil {
                 g = resp.SecurityGroup
         } else {
-                resp, err := e.ec2.SecurityGroups(ec2.SecurityGroupNames(name), nil)
+                resp, err := ec2inst.SecurityGroups(ec2.SecurityGroupNames(name), nil)
                 if err != nil {
                         return zeroGroup, err
                 }
                 // It's possible that the old group has the wrong
                 // description here, but if it does it's probably due
                 // to something deliberately playing games with juju,
                 // so we ignore it.
                 have = newPermSet(resp.Groups[0].IPPerms)
                 g = resp.Groups[0].SecurityGroup
         }
         revoke := make(permSet)
         for p := range have {
                 if !want[p] {
                         revoke[p] = true
                 }
         }
         if len(revoke) > 0 {
-                _, err := e.ec2.RevokeSecurityGroup(g, revoke.ipPerms())
+                _, err := ec2inst.RevokeSecurityGroup(g, revoke.ipPerms())
                 if err != nil {
                         return zeroGroup, fmt.Errorf("cannot revoke security group: %v", err)
                 }
         }
 
         add := make(permSet)
         for p := range want {
                 if !have[p] {
                         add[p] = true
                 }
         }
         if len(add) > 0 {
-                _, err := e.ec2.AuthorizeSecurityGroup(g, add.ipPerms())
+                _, err := ec2inst.AuthorizeSecurityGroup(g, add.ipPerms())
                 if err != nil {
                         return zeroGroup, fmt.Errorf("cannot authorize securityGroup: %v", err)
                 }
         }
         return g, nil
 }
 
 // permKey represents a permission for a group or an ip address range
 // to access the given range of ports. Only one of groupId or ipAddr
 // should be non-empty.
@@ skipping 54 matching lines @@
 
 // If the err is of type *ec2.Error, ec2ErrCode returns
 // its code, otherwise it returns the empty string.
 func ec2ErrCode(err error) string {
         ec2err, _ := err.(*ec2.Error)
         if ec2err == nil {
                 return ""
         }
         return ec2err.Code
 }