func destroyStorage(env environs.Environ) error {
	logger.Infof("destroying storage")
	environConfig := env.Config()
	storageProviderTypes, ok := registry.EnvironStorageProviders(environConfig.Type())
	if !ok {
		return nil
	}
	for _, storageProviderType := range storageProviderTypes {
		storageProvider, err := registry.StorageProvider(storageProviderType)
		if err != nil {
			return errors.Trace(err)
		}
		if !storageProvider.Dynamic() {
			continue
		}
		if storageProvider.Scope() != storage.ScopeEnviron {
			continue
		}
		if err := destroyVolumes(environConfig, storageProviderType, storageProvider); err != nil {
			return errors.Trace(err)
		}
		// TODO(axw) destroy env-level filesystems when we have them.
	}
	return nil
}

// machineVolumeParams retrieves VolumeParams for the volumes that should be
// provisioned with, and attached to, the machine. The client should ignore
// parameters that it does not know how to handle.
func (p *ProvisionerAPI) machineVolumeParams(m *state.Machine) ([]params.VolumeParams, error) {
	volumeAttachments, err := m.VolumeAttachments()
	if err != nil {
		return nil, err
	}
	if len(volumeAttachments) == 0 {
		return nil, nil
	}
	envConfig, err := p.st.EnvironConfig()
	if err != nil {
		return nil, err
	}
	poolManager := poolmanager.New(state.NewStateSettings(p.st))
	allVolumeParams := make([]params.VolumeParams, 0, len(volumeAttachments))
	for _, volumeAttachment := range volumeAttachments {
		volumeTag := volumeAttachment.Volume()
		volume, err := p.st.Volume(volumeTag)
		if err != nil {
			return nil, errors.Annotatef(err, "getting volume %q", volumeTag.Id())
		}
		storageInstance, err := storagecommon.MaybeAssignedStorageInstance(
			volume.StorageInstance, p.st.StorageInstance,
		)
		if err != nil {
			return nil, errors.Annotatef(err, "getting volume %q storage instance", volumeTag.Id())
		}
		volumeParams, err := storagecommon.VolumeParams(volume, storageInstance, envConfig, poolManager)
		if err != nil {
			return nil, errors.Annotatef(err, "getting volume %q parameters", volumeTag.Id())
		}
		provider, err := registry.StorageProvider(storage.ProviderType(volumeParams.Provider))
		if err != nil {
			return nil, errors.Annotate(err, "getting storage provider")
		}
		if provider.Dynamic() {
			// Leave dynamic storage to the storage provisioner.
			continue
		}
		volumeAttachmentParams, ok := volumeAttachment.Params()
		if !ok {
			// Attachment is already provisioned; this is an insane
			// state, so we should not proceed with the volume.
			return nil, errors.Errorf(
				"volume %s already attached to machine %s",
				volumeTag.Id(), m.Id(),
			)
		}
		// Not provisioned yet, so ask the cloud provisioner do it.
		volumeParams.Attachment = &params.VolumeAttachmentParams{
			volumeTag.String(),
			m.Tag().String(),
			"", // we're creating the volume, so it has no volume ID.
			"", // we're creating the machine, so it has no instance ID.
			volumeParams.Provider,
			volumeAttachmentParams.ReadOnly,
		}
		allVolumeParams = append(allVolumeParams, volumeParams)
	}
	return allVolumeParams, nil
}

// Create is defined on PoolManager interface.
func (pm *poolManager) Create(name string, providerType storage.ProviderType, attrs map[string]interface{}) (*storage.Config, error) {
	if name == "" {
		return nil, MissingNameError
	}
	if providerType == "" {
		return nil, MissingTypeError
	}

	cfg, err := storage.NewConfig(name, providerType, attrs)
	if err != nil {
		return nil, errors.Trace(err)
	}
	p, err := registry.StorageProvider(providerType)
	if err != nil {
		return nil, errors.Trace(err)
	}
	if err := provider.ValidateConfig(p, cfg); err != nil {
		return nil, errors.Annotate(err, "validating storage provider config")
	}

	poolAttrs := cfg.Attrs()
	poolAttrs[Name] = name
	poolAttrs[Type] = string(providerType)
	if err := pm.settings.CreateSettings(globalKey(name), poolAttrs); err != nil {
		return nil, errors.Annotatef(err, "creating pool %q", name)
	}
	return cfg, nil
}

func (s *providerRegistrySuite) TestRegisterProvider(c *gc.C) {
	p1 := &mockProvider{}
	ptype := storage.ProviderType("foo")
	registry.RegisterProvider(ptype, p1)
	p, err := registry.StorageProvider(ptype)
	c.Assert(err, jc.ErrorIsNil)
	c.Assert(p, gc.Equals, p1)
}

func (s *providerRegistrySuite) TestUnregisterProvider(c *gc.C) {
	ptype := storage.ProviderType("foo")

	// No-op, since there's nothing registered yet.
	registry.RegisterProvider(ptype, nil)

	// Register and then unregister, ensure that the provider cannot
	// be accessed.
	registry.RegisterProvider(ptype, &mockProvider{})
	registry.RegisterProvider(ptype, nil)
	_, err := registry.StorageProvider(storage.ProviderType("foo"))
	c.Assert(err, gc.ErrorMatches, `storage provider "foo" not found`)
}

// StoragePoolConfig returns the storage provider type and
// configuration for a named storage pool. If there is no
// such pool with the specified name, but it identifies a
// storage provider, then that type will be returned with a
// nil configuration.
func StoragePoolConfig(name string, poolManager poolmanager.PoolManager) (storage.ProviderType, *storage.Config, error) {
	pool, err := poolManager.Get(name)
	if errors.IsNotFound(err) {
		// If not a storage pool, then maybe a provider type.
		providerType := storage.ProviderType(name)
		if _, err1 := registry.StorageProvider(providerType); err1 != nil {
			return "", nil, errors.Trace(err)
		}
		return providerType, &storage.Config{}, nil
	} else if err != nil {
		return "", nil, errors.Annotatef(err, "getting pool %q", name)
	}
	return pool.Provider(), pool, nil
}

func poolStorageProvider(st *State, poolName string) (storage.ProviderType, storage.Provider, error) {
	poolManager := poolmanager.New(NewStateSettings(st))
	pool, err := poolManager.Get(poolName)
	if errors.IsNotFound(err) {
		// If there's no pool called poolName, maybe a provider type
		// has been specified directly.
		providerType := storage.ProviderType(poolName)
		provider, err1 := registry.StorageProvider(providerType)
		if err1 != nil {
			// The name can't be resolved as a storage provider type,
			// so return the original "pool not found" error.
			return "", nil, errors.Trace(err)
		}
		return providerType, provider, nil
	} else if err != nil {
		return "", nil, errors.Trace(err)
	}
	providerType := pool.Provider()
	provider, err := registry.StorageProvider(providerType)
	if err != nil {
		return "", nil, errors.Trace(err)
	}
	return providerType, provider, nil
}

// storageConfig returns the provider type and config attributes for the
// specified poolName. If no such pool exists, we check to see if poolName is
// actually a provider type, in which case config will be empty.
func storageConfig(st *state.State, poolName string) (storage.ProviderType, map[string]interface{}, error) {
	pm := poolmanager.New(state.NewStateSettings(st))
	p, err := pm.Get(poolName)
	// If not a storage pool, then maybe a provider type.
	if errors.IsNotFound(err) {
		providerType := storage.ProviderType(poolName)
		if _, err1 := registry.StorageProvider(providerType); err1 != nil {
			return "", nil, errors.Trace(err)
		}
		return providerType, nil, nil
	}
	if err != nil {
		return "", nil, errors.Trace(err)
	}
	return p.Provider(), p.Attrs(), nil
}

func sourceParams(providerType storage.ProviderType, sourceName, baseStorageDir string) (storage.Provider, *storage.Config, error) {
	provider, err := registry.StorageProvider(providerType)
	if err != nil {
		return nil, nil, errors.Annotate(err, "getting provider")
	}
	attrs := make(map[string]interface{})
	if baseStorageDir != "" {
		storageDir := filepath.Join(baseStorageDir, sourceName)
		attrs[storage.ConfigStorageDir] = storageDir
	}
	sourceConfig, err := storage.NewConfig(sourceName, providerType, attrs)
	if err != nil {
		return nil, nil, errors.Annotate(err, "getting config")
	}
	return provider, sourceConfig, nil
}

func configFromSettings(settings map[string]interface{}) (*storage.Config, error) {
	providerType := storage.ProviderType(settings[Type].(string))
	name := settings[Name].(string)
	// Ensure returned attributes are stripped of name and type,
	// as these are not user-specified attributes.
	delete(settings, Name)
	delete(settings, Type)
	cfg, err := storage.NewConfig(name, providerType, settings)
	if err != nil {
		return nil, errors.Trace(err)
	}
	p, err := registry.StorageProvider(providerType)
	if err != nil {
		return nil, errors.Trace(err)
	}
	if err := provider.ValidateConfig(p, cfg); err != nil {
		return nil, errors.Trace(err)
	}
	return cfg, nil
}

func (*maasProviderSuite) TestMAASProviderRegistered(c *gc.C) {
	p, err := registry.StorageProvider(maas.MaasStorageProviderType)
	c.Assert(err, jc.ErrorIsNil)
	_, ok := p.(storage.Provider)
	c.Assert(ok, jc.IsTrue)
}

func (s *providerRegistrySuite) TestNoSuchProvider(c *gc.C) {
	_, err := registry.StorageProvider(storage.ProviderType("foo"))
	c.Assert(err, gc.ErrorMatches, `storage provider "foo" not found`)
}

func (*providerSuite) TestEBSProviderRegistered(c *gc.C) {
	p, err := registry.StorageProvider(ec2.EBS_ProviderType)
	c.Assert(err, jc.ErrorIsNil)
	_, ok := p.(storage.Provider)
	c.Assert(ok, jc.IsTrue)
}