'Change the version of flocker installed on all of the nodes to the
version indicated by `package_source`.
:param PackageSource package_source: The :class:`PackageSource` to
install flocker from on all of the nodes.
:param bool destroy_persisted_state: Whether to destroy the control
node\'s state file when upgrading or not.'

@log_method
def install_flocker_version(self, package_source, destroy_persisted_state=False):

    control_node_address = self.control_node.public_address
    all_cluster_nodes = set((list((x.public_address for x in self.nodes)) + [control_node_address]))
    distribution = self.distribution
    def get_flocker_version():
        d = retry_failure(self.reactor, (lambda : timeout(self.reactor, self.client.version(), 10)), ([1] * 10))
        d.addCallback((lambda v: (v.get('flocker', u'').encode('ascii') or None)))
        return d
    d = get_flocker_version()
    d.addErrback(write_failure)
    def reinstall_if_needed(current_version):
        if ((not current_version) or (current_version != package_source.version)):
            return reinstall_flocker_from_package_source(reactor, all_cluster_nodes, control_node_address, package_source, distribution, destroy_persisted_state=destroy_persisted_state)
        return current_version
    d.addCallback(reinstall_if_needed)
    d.addCallback((lambda _: get_flocker_version()))
    def verify_version(current_version):
        if package_source.version:
            if (current_version != package_source.version):
                raise FailureToUpgrade(('Failed    to    set    version    of    flocker    to    %s,    it    is    still    %s.' % (package_source.version, current_version)))
        return current_version
    d.addCallback(verify_version)
    return d
