← Back to branch summary

~launchpad-pqm/launchpad/devel

« back to all changes in this revision

Viewing changes to database/replication/generate_migration.py

  • Committer: Launchpad Patch Queue Manager
  • Date: 2011-08-03 11:23:34 UTC
  • mfrom: (13457.6.16 upgrade-stderr)
  • Revision ID: launchpad@pqm.canonical.com-20110803112334-acnupsa7jmzmdeet
[r=stevenk][bug=819751] Fix the implementation of several methods in
 LoggingUIFactory.

Show diffs side-by-side

added added

removed removed

Lines of Context:
database/replication/generate_migration.py
1
 
#!/usr/bin/python -S
2
 
# Copyright 2011 Canonical Ltd.  This software is licensed under the
3
 
# GNU Affero General Public License version 3 (see the file LICENSE).
4
 
 
5
 
"""Generate slonik scripts for Slony 1.2 to 2.0 migration.
6
 
 
7
 
Remove this script after migration is complete.
8
 
"""
9
 
 
10
 
__metaclass__ = type
11
 
__all__ = []
12
 
 
13
 
import _pythonpath
14
 
 
15
 
from optparse import OptionParser
16
 
import os.path
17
 
from textwrap import dedent
18
 
 
19
 
from lp.services import scripts
20
 
from lp.services.database.sqlbase import connect
21
 
import replication.helpers
22
 
from replication.helpers import (
23
 
    LPMAIN_SET_ID,
24
 
    LPMIRROR_SET_ID,
25
 
    SSO_SET_ID,
26
 
    get_all_cluster_nodes,
27
 
    get_master_node,
28
 
    )
29
 
 
30
 
 
31
 
con = None
32
 
options = None
33
 
 
34
 
sets = {
35
 
    LPMAIN_SET_ID: 'lpmain_set',
36
 
    SSO_SET_ID: 'sso_set',
37
 
    LPMIRROR_SET_ID: 'lpmirror_set',
38
 
    }
39
 
 
40
 
 
41
 
def outpath(filename):
42
 
    return os.path.join(options.outdir, filename)
43
 
 
44
 
 
45
 
def message(outf, msg):
46
 
    assert "'" not in msg
47
 
    print >> outf, "echo '%s';" % msg
48
 
 
49
 
 
50
 
def generate_preamble():
51
 
    outf = open(outpath('mig_preamble.sk'), 'w')
52
 
    print >> outf, replication.helpers.preamble(con)
53
 
 
54
 
    cur = con.cursor()
55
 
 
56
 
    for set_id, set_name in list(sets.items()):
57
 
        cur.execute(
58
 
            "SELECT set_origin FROM _sl.sl_set WHERE set_id=%s", [set_id])
59
 
        result = cur.fetchone()
60
 
        if result:
61
 
            origin = result[0]
62
 
            print >> outf, "define %s_origin %d;" % (set_name, origin)
63
 
        else:
64
 
            del sets[set_id]  # For testing. Production will have 3 sets.
65
 
    outf.close()
66
 
 
67
 
 
68
 
def generate_uninstall():
69
 
    outf = open(outpath('mig_uninstall.sk'), 'w')
70
 
    print >> outf, "# Uninstall Slony-I 1.2 from all nodes"
71
 
    print >> outf, "include <mig_preamble.sk>;"
72
 
 
73
 
    nodes = get_all_cluster_nodes(con)
74
 
 
75
 
    # Ensure everything is really, really synced since we will be
76
 
    # resubscribing with 'omit copy'
77
 
    for node in nodes:
78
 
        print >> outf, dedent("""\
79
 
                sync (id=%d);
80
 
                wait for event (origin=%d, confirmed=all, wait on=%d);
81
 
                """).strip() % (node.node_id, node.node_id, node.node_id)
82
 
 
83
 
    for node in nodes:
84
 
        message(outf, "Uninstall node %d" % node.node_id)
85
 
        print >> outf, "uninstall node (id=%d);" % node.node_id
86
 
    outf.close()
87
 
 
88
 
 
89
 
def generate_sync():
90
 
    outf = open(outpath('mig_sync.sk'), 'w')
91
 
    message(outf, "Waiting for sync")
92
 
    print >> outf, "sync (id=@master_node);"
93
 
    print >> outf, dedent("""\
94
 
        wait for event (
95
 
                origin=@master_node, confirmed=all, wait on=@master_node);
96
 
            """).strip()
97
 
    outf.close()
98
 
 
99
 
 
100
 
def generate_rebuild():
101
 
    outf = open(outpath('mig_rebuild.sk'), 'w')
102
 
    print >> outf, "# Rebuild the replication cluster with Slony-I 2.0"
103
 
    print >> outf, "include <mig_preamble.sk>;"
104
 
 
105
 
    nodes = get_all_cluster_nodes(con)
106
 
    first_node = nodes[0]
107
 
    remaining_nodes = nodes[1:]
108
 
 
109
 
    # Initialize the cluster
110
 
    message(outf, "Initializing cluster (node %d)" % first_node.node_id)
111
 
    print >> outf, "init cluster (id=%d);" % first_node.node_id
112
 
 
113
 
    # Create all the other nodes
114
 
    for node in remaining_nodes:
115
 
        message(outf, "Initializing node %d" % node.node_id)
116
 
        print >> outf, "store node (id=%d, event node=%d);" % (
117
 
            node.node_id, first_node.node_id)
118
 
 
119
 
    # Create paths so they can communicate.
120
 
    message(outf, "Storing %d paths" % pow(len(nodes), 2))
121
 
    for client_node in nodes:
122
 
        for server_node in nodes:
123
 
            print >> outf, (
124
 
                "store path (server=%d, client=%d, "
125
 
                "conninfo=@node%d_node_conninfo);" % (
126
 
                    server_node.node_id, client_node.node_id,
127
 
                    server_node.node_id))
128
 
 
129
 
    # sync to ensure replication is happening.
130
 
    print >> outf, "include <mig_sync.sk>;"
131
 
 
132
 
    # Create replication sets.
133
 
    for set_id, set_name in sets.items():
134
 
        generate_initialize_set(set_id, set_name, outf)
135
 
    print >> outf, "include <mig_sync.sk>;"
136
 
 
137
 
    # Subscribe slave nodes to replication sets.
138
 
    for set_id, set_name in sets.items():
139
 
        generate_subscribe_set(set_id, set_name, outf)
140
 
 
141
 
    outf.close()
142
 
 
143
 
 
144
 
def generate_initialize_set(set_id, set_name, outf):
145
 
    origin_node = get_master_node(con, set_id)
146
 
    message(outf, "Creating %s origin %d" % (set_name, origin_node.node_id))
147
 
    print >> outf, "create set (id=%d, origin=@%s_origin, comment='%s');" % (
148
 
        set_id, set_name, set_name)
149
 
    # Need to connect to a node currently replicating the desired set.
150
 
    origin_con = origin_node.connect()
151
 
    cur = origin_con.cursor()
152
 
    cur.execute("""
153
 
        SELECT tab_id, tab_nspname, tab_relname, tab_comment
154
 
        FROM _sl.sl_table WHERE tab_set=%s
155
 
        """, (set_id,))
156
 
    results = cur.fetchall()
157
 
    message(outf, "Adding %d tables to %s" % (len(results), set_name))
158
 
    for tab_id, tab_nspname, tab_relname, tab_comment in results:
159
 
        if not tab_comment:
160
 
            tab_comment = ''
161
 
        print >> outf, dedent("""\
162
 
                set add table (
163
 
                    set id=@%s, origin=@%s_origin, id=%d,
164
 
                    fully qualified name='%s.%s',
165
 
                    comment='%s');
166
 
                """).strip() % (
167
 
                    set_name, set_name, tab_id,
168
 
                    tab_nspname, tab_relname, tab_comment)
169
 
    cur.execute("""
170
 
        SELECT seq_id, seq_nspname, seq_relname, seq_comment
171
 
        FROM _sl.sl_sequence WHERE seq_set=%s
172
 
        """, (set_id,))
173
 
    results = cur.fetchall()
174
 
    message(outf, "Adding %d sequences to %s" % (len(results), set_name))
175
 
    for seq_id, seq_nspname, seq_relname, seq_comment in results:
176
 
        if not seq_comment:
177
 
            seq_comment = ''
178
 
        print >> outf, dedent("""\
179
 
                set add sequence (
180
 
                    set id=@%s, origin=@%s_origin, id=%d,
181
 
                    fully qualified name='%s.%s',
182
 
                    comment='%s');
183
 
                """).strip() % (
184
 
                    set_name, set_name, seq_id,
185
 
                    seq_nspname, seq_relname, seq_comment)
186
 
 
187
 
 
188
 
def generate_subscribe_set(set_id, set_name, outf):
189
 
    cur = con.cursor()
190
 
    cur.execute("""
191
 
        SELECT sub_receiver FROM _sl.sl_subscribe
192
 
        WHERE sub_set=%s and sub_active is true
193
 
        """, (set_id,))
194
 
    for receiver_id, in cur.fetchall():
195
 
        message(outf, "Subscribing node %d to %s" % (receiver_id, set_name))
196
 
        print >> outf, dedent("""\
197
 
                subscribe set (
198
 
                    id=%d, provider=@%s_origin, receiver=%d,
199
 
                    forward=true, omit copy=true);
200
 
                wait for event (
201
 
                    origin=@%s_origin, confirmed=all, wait on=@%s_origin);
202
 
                """).strip() % (
203
 
                    set_id, set_name, receiver_id,
204
 
                    set_name, set_name)
205
 
        print >> outf, "include <mig_sync.sk>;"
206
 
 
207
 
 
208
 
def main():
209
 
    parser = OptionParser()
210
 
    scripts.db_options(parser)
211
 
    parser.add_option(
212
 
        "-o", "--output-dir", dest='outdir', default=".",
213
 
        help="Write slonik scripts to DIR", metavar="DIR")
214
 
    global options
215
 
    options, args = parser.parse_args()
216
 
    if args:
217
 
        parser.error("Too many arguments")
218
 
    scripts.execute_zcml_for_scripts(use_web_security=False)
219
 
 
220
 
    global con
221
 
    con = connect()
222
 
 
223
 
    generate_preamble()
224
 
    generate_sync()
225
 
    generate_uninstall()
226
 
    generate_rebuild()
227
 
 
228
 
    return 0
229
 
 
230
 
 
231
 
if __name__ == '__main__':
232
 
    raise SystemExit(main())