概要

fabric2のThreadingGroupクラスを利用して複数のホストに対して並列にコマンドを実行したい時にsudo権限が必要なコマンドが失敗する問題を解決します。

解決の方針

fabric2のConnectionクラスではpassword引数を与えるとsudoパスワードを自動入力してくれる機能があるsudo関数がありますが、ThreadingGroupクラスにはありません。
並列実行時にrunの中でsudoコマンドを実行するとsudoパスワードを聞いてくるプロンプトが発生するのですが、複数サーバに対して並列に実行しているとsudoパスワードの入力がうまくいかず、コマンドは失敗してしまいます。
つまり複数サーバに対して並列にsudoコマンドを実行するにはThreadingGroupにsudo関数が必要となります。
しかし本家のコードコメントを見ると将来的に載りそうなものの詳細は検討中で未実装のようです。
かと言ってすぐに使いたい場合は待つこともできないのでThreadingGroupを継承したMyThreadingGroupというクラスを作って対応します。

sudoを実装したMyThreadingGroup

ソース: https://github.com/muumu/fabric2-parallel-sample/commit/2650fc999cccc1f973ac35fb6a1c5af89ec99a78
ThreadingGroupを継承した上でThreadingGroupのrunの実装にsudoパスワードを保存して引数として渡す機能を付け加えたものを用意します。
sudoパスワードはMyThreadingGroupインスタンス毎に最初にsudo関数が呼ばれたタイミングで入力用のプロンプトが表示されてsudoパスワードをプロパティに保存するようにしています。

sudoを実装したMyThreadingGroup

#-*- coding:utf-8 -*-
try:
    from invoke.vendor.six.moves.queue import Queue
except ImportError:
    from six.moves.queue import Queue

from invoke.util import ExceptionHandlingThread

from fabric.exceptions import GroupException
from fabric import Connection, ThreadingGroup, GroupResult
from getpass import getpass

def thread_worker_sudo(cxn, queue, args, kwargs):
    result = cxn.sudo(*args, **kwargs)
    queue.put((cxn, result))

class MyThreadingGroup(ThreadingGroup):
    def sudo(self, *args, **kwargs):
        if not hasattr(self, 'password'):
            self.password = getpass('Input sudo password: ')
        kwargs['password'] = self.password
        results = GroupResult()
        queue = Queue()
        threads = []
        for cxn in self:
            my_kwargs = dict(cxn=cxn, queue=queue, args=args, kwargs=kwargs)
            thread = ExceptionHandlingThread(
                target=thread_worker_sudo, kwargs=my_kwargs
            )
            threads.append(thread)
        for thread in threads:
            thread.start()
        for thread in threads:
            thread.join()
        while not queue.empty():
            cxn, result = queue.get(block=False)
            results[cxn] = result
        excepted = False
        for thread in threads:
            wrapper = thread.exception()
            if wrapper is not None:
                cxn = wrapper.kwargs["kwargs"]["cxn"]
                results[cxn] = wrapper.value
                excepted = True
        if excepted:
            raise GroupException(results)
return results

そしてgroupをセットする時にこのMyThreadingGroupを使うように変更します。

set_groupの修正

def set_group(name):
    global group
    hosts = get_hosts(name)
    group = MyThreadingGroup(*hosts)
    group.hosts = hosts
    group.component = get_component(name)()
    group.environment = get_environment(name)
return group

あとはこのgroupインスタンスを使えばsudoを呼び出せます。
たとえばsudoタスクを作ればタスクとして直接呼び出せます。

sudoタスクの作成

@task
def sudo(c, command=None, user='root', warn=False, print_result=True):
    c = get_group()
    r = c.sudo(command, user=user, warn=warn)
    if print_result:
        for connection, result in r.items():
            print(connection.host + ': ' + result.stdout.strip())
return r

これでsudoタスクを実行できるようになります。

実行例

$ fab group web.prod.2 sudo --command='mkdir /var/mytemp' --warn
...
Input sudo password:
...

※groupタスクを呼び出して実行対象サーバーグループをセットし次のsudoタスクでそのサーバーグループを対象に実行する仕組みの実装は前回の記事『fabric2でホストグループを定義して並列実行できるようにする』を参照してください。
このgroupインスタンスが渡ってくる関数からもsudoを使えるようになります。

groupインスタンスを渡す

@task
def build(c):
    c = get_group()
return c.component.build(c)

渡されたgroupインスタンスからsudoを呼び出す

class Web:
    def build(self, c):
return c.sudo('echo "Building web server..."')

まとめ

ThreadingGroupを継承したMyThreadingGroupでsudo関数を実装することにより、複数のサーバに対して並列にsudoコマンドを実行できるようになりました。
最初にMyThreadingGroupのsudo関数を呼び出したタイミングでsudoパスワードを聞くプロンプトが表示され、そこで入力したsudoパスワードがfab実行プロセス内に保存されて以降は自動入力してくれるようになります。
fabric2本家でThreadingGroupにsudo関数が追加されるまでの間はこの独自の実装で凌ぐことにします。

fabric2のThreadingGroupでsudoを実行できるようにする

概要

解決の方針

sudoを実装したMyThreadingGroup

まとめ