Statistics
| Branch: | Tag: | Revision:

root / snf-cyclades-app / synnefo / logic / reconciliation.py @ bfd04b01

History | View | Annotate | Download (30.4 kB)

1
# -*- coding: utf-8 -*-
2
#
3
# Copyright 2011-2013 GRNET S.A. All rights reserved.
4
#
5
# Redistribution and use in source and binary forms, with or
6
# without modification, are permitted provided that the following
7
# conditions are met:
8
#
9
#   1. Redistributions of source code must retain the above
10
#      copyright notice, this list of conditions and the following
11
#      disclaimer.
12
#
13
#   2. Redistributions in binary form must reproduce the above
14
#      copyright notice, this list of conditions and the following
15
#      disclaimer in the documentation and/or other materials
16
#      provided with the distribution.
17
#
18
# THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS
19
# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
20
# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
21
# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR
22
# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
23
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
24
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
25
# USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
26
# AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
27
# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
28
# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
29
# POSSIBILITY OF SUCH DAMAGE.
30
#
31
# The views and conclusions contained in the software and
32
# documentation are those of the authors and should not be
33
# interpreted as representing official policies, either expressed
34
# or implied, of GRNET S.A.
35
#
36
"""Business logic for reconciliation
37

38
Reconcile the contents of the DB with the actual state of the
39
Ganeti backend.
40

41
Let D be the set of VMs in the DB, G the set of VMs in Ganeti.
42
RULES:
43
    R1. Stale servers in DB:
44
            For any v in D but not in G:
45
            Set deleted=True.
46
    R2. Orphan instances in Ganet:
47
            For any v in G with deleted=True in D:
48
            Issue OP_INSTANCE_DESTROY.
49
    R3. Unsynced operstate:
50
            For any v whose operating state differs between G and V:
51
            Set the operating state in D based on the state in G.
52
In the code, D, G are Python dicts mapping instance ids to operating state.
53
For D, the operating state is chosen from VirtualMachine.OPER_STATES.
54
For G, the operating state is True if the machine is up, False otherwise.
55

56
"""
57

    
58

    
59
from django.conf import settings
60

    
61
import logging
62
import itertools
63
import bitarray
64
from datetime import datetime, timedelta
65

    
66
from django.db import transaction
67
from synnefo.db.models import (Backend, VirtualMachine, Flavor,
68
                               pooled_rapi_client, Network,
69
                               BackendNetwork, BridgePoolTable,
70
                               MacPrefixPoolTable)
71
from synnefo.db import pools
72
from synnefo.logic import utils, backend as backend_mod
73

    
74
logger = logging.getLogger()
75
logging.basicConfig()
76

    
77
try:
78
    CHECK_INTERVAL = settings.RECONCILIATION_CHECK_INTERVAL
79
except AttributeError:
80
    CHECK_INTERVAL = 60
81

    
82
GANETI_JOB_ERROR = "error"
83
GANETI_JOBS_PENDING = ["queued", "waiting", "running", "canceling"]
84
GANETI_JOBS_FINALIZED = ["success", "error", "canceled"]
85

    
86

    
87
class BackendReconciler(object):
88
    def __init__(self, backend, logger, options=None):
89
        self.backend = backend
90
        self.log = logger
91
        self.client = backend.get_client()
92
        if options is None:
93
            self.options = {}
94
        else:
95
            self.options = options
96

    
97
    def close(self):
98
        self.backend.put_client(self.client)
99

    
100
    @transaction.commit_on_success
101
    def reconcile(self):
102
        log = self.log
103
        backend = self.backend
104
        log.debug("Reconciling backend %s", backend)
105

    
106
        self.db_servers = get_database_servers(backend)
107
        self.db_servers_keys = set(self.db_servers.keys())
108
        log.debug("Got servers info from database.")
109

    
110
        self.gnt_servers = get_ganeti_servers(backend)
111
        self.gnt_servers_keys = set(self.gnt_servers.keys())
112
        log.debug("Got servers info from Ganeti backend.")
113

    
114
        self.gnt_jobs = get_ganeti_jobs(backend)
115
        log.debug("Got jobs from Ganeti backend")
116

    
117
        self.event_time = datetime.now()
118

    
119
        self.stale_servers = self.reconcile_stale_servers()
120
        self.orphan_servers = self.reconcile_orphan_servers()
121
        self.unsynced_servers = self.reconcile_unsynced_servers()
122
        self.close()
123

    
124
    def get_build_status(self, db_server):
125
        job_id = db_server.backendjobid
126
        if job_id in self.gnt_jobs:
127
            gnt_job_status = self.gnt_jobs[job_id]["status"]
128
            if gnt_job_status == GANETI_JOB_ERROR:
129
                return "ERROR"
130
            elif gnt_job_status not in GANETI_JOBS_FINALIZED:
131
                return "RUNNING"
132
            else:
133
                return "FINALIZED"
134
        else:
135
            return "ERROR"
136

    
137
    def reconcile_stale_servers(self):
138
        # Detect stale servers
139
        stale = []
140
        stale_keys = self.db_servers_keys - self.gnt_servers_keys
141
        for server_id in stale_keys:
142
            db_server = self.db_servers[server_id]
143
            if db_server.operstate == "BUILD":
144
                build_status = self.get_build_status(db_server)
145
                if build_status == "ERROR":
146
                    # Special handling of BUILD eerrors
147
                    self.reconcile_building_server(db_server)
148
                elif build_status != "RUNNING":
149
                    stale.append(server_id)
150
            elif (db_server.operstate == "ERROR" and
151
                  db_server.action != "DESTROY"):
152
                # Servers at building ERROR are stale only if the user has
153
                # asked to destroy them.
154
                pass
155
            else:
156
                stale.append(server_id)
157

    
158
        # Report them
159
        if stale:
160
            self.log.info("Found stale servers %s at backend %s",
161
                          ", ".join(map(str, stale)), self.backend)
162
        else:
163
            self.log.debug("No stale servers at backend %s", self.backend)
164

    
165
        # Fix them
166
        if stale and self.options["fix_stale"]:
167
            for server_id in stale:
168
                db_server = self.db_servers[server_id]
169
                backend_mod.process_op_status(
170
                    vm=db_server,
171
                    etime=self.event_time,
172
                    jobid=-0,
173
                    opcode='OP_INSTANCE_REMOVE', status='success',
174
                    logmsg='Reconciliation: simulated Ganeti event')
175
            self.log.debug("Simulated Ganeti removal for stale servers.")
176

    
177
    def reconcile_orphan_servers(self):
178
        orphans = self.gnt_servers_keys - self.db_servers_keys
179
        if orphans:
180
            self.log.info("Found orphan servers %s at backend %s",
181
                          ", ".join(map(str, orphans)), self.backend)
182
        else:
183
            self.log.debug("No orphan servers at backend %s", self.backend)
184

    
185
        if orphans and self.options["fix_orphans"]:
186
            for server_id in orphans:
187
                server_name = utils.id_to_instance_name(server_id)
188
                self.client.DeleteInstance(server_name)
189
            self.log.debug("Issued OP_INSTANCE_REMOVE for orphan servers.")
190

    
191
    def reconcile_unsynced_servers(self):
192
        #log = self.log
193
        for server_id in self.db_servers_keys & self.gnt_servers_keys:
194
            db_server = self.db_servers[server_id]
195
            gnt_server = self.gnt_servers[server_id]
196
            if db_server.operstate == "BUILD":
197
                build_status = self.get_build_status(db_server)
198
                if build_status == "RUNNING":
199
                    # Do not reconcile building VMs
200
                    continue
201
                elif build_status == "ERROR":
202
                    # Special handling of build errors
203
                    self.reconcile_building_server(db_server)
204
                    continue
205

    
206
            self.reconcile_unsynced_operstate(server_id, db_server,
207
                                              gnt_server)
208
            self.reconcile_unsynced_flavor(server_id, db_server,
209
                                           gnt_server)
210
            self.reconcile_unsynced_nics(server_id, db_server, gnt_server)
211
            self.reconcile_unsynced_disks(server_id, db_server, gnt_server)
212
            if db_server.task is not None:
213
                self.reconcile_pending_task(server_id, db_server)
214

    
215
    def reconcile_building_server(self, db_server):
216
        self.log.info("Server '%s' is BUILD in DB, but 'ERROR' in Ganeti.",
217
                      db_server.id)
218
        if self.options["fix_unsynced"]:
219
            fix_opcode = "OP_INSTANCE_CREATE"
220
            backend_mod.process_op_status(
221
                vm=db_server,
222
                etime=self.event_time,
223
                jobid=-0,
224
                opcode=fix_opcode, status='error',
225
                logmsg='Reconciliation: simulated Ganeti event')
226
            self.log.debug("Simulated Ganeti error build event for"
227
                           " server '%s'", db_server.id)
228

    
229
    def reconcile_unsynced_operstate(self, server_id, db_server, gnt_server):
230
        if db_server.operstate != gnt_server["state"]:
231
            self.log.info("Server '%s' is '%s' in DB and '%s' in Ganeti.",
232
                          server_id, db_server.operstate, gnt_server["state"])
233
            if self.options["fix_unsynced"]:
234
                # If server is in building state, you will have first to
235
                # reconcile it's creation, to avoid wrong quotas
236
                if db_server.operstate == "BUILD":
237
                    backend_mod.process_op_status(
238
                        vm=db_server, etime=self.event_time, jobid=-0,
239
                        opcode="OP_INSTANCE_CREATE", status='success',
240
                        logmsg='Reconciliation: simulated Ganeti event')
241
                fix_opcode = "OP_INSTANCE_STARTUP"\
242
                    if gnt_server["state"] == "STARTED"\
243
                    else "OP_INSTANCE_SHUTDOWN"
244
                backend_mod.process_op_status(
245
                    vm=db_server, etime=self.event_time, jobid=-0,
246
                    opcode=fix_opcode, status='success',
247
                    logmsg='Reconciliation: simulated Ganeti event')
248
                self.log.debug("Simulated Ganeti state event for server '%s'",
249
                               server_id)
250

    
251
    def reconcile_unsynced_flavor(self, server_id, db_server, gnt_server):
252
        db_flavor = db_server.flavor
253
        gnt_flavor = gnt_server["flavor"]
254
        if (db_flavor.ram != gnt_flavor["ram"] or
255
           db_flavor.cpu != gnt_flavor["vcpus"]):
256
            try:
257
                gnt_flavor = Flavor.objects.get(
258
                    ram=gnt_flavor["ram"],
259
                    cpu=gnt_flavor["vcpus"],
260
                    disk=db_flavor.disk,
261
                    disk_template=db_flavor.disk_template)
262
            except Flavor.DoesNotExist:
263
                self.log.warning("Server '%s' has unknown flavor.", server_id)
264
                return
265

    
266
            self.log.info("Server '%s' has flavor '%s' in DB and '%s' in"
267
                          " Ganeti", server_id, db_flavor, gnt_flavor)
268
            if self.options["fix_unsynced_flavors"]:
269
                old_state = db_server.operstate
270
                opcode = "OP_INSTANCE_SET_PARAMS"
271
                beparams = {"vcpus": gnt_flavor.cpu,
272
                            "minmem": gnt_flavor.ram,
273
                            "maxmem": gnt_flavor.ram}
274
                backend_mod.process_op_status(
275
                    vm=db_server, etime=self.event_time, jobid=-0,
276
                    opcode=opcode, status='success',
277
                    job_fields={"beparams": beparams},
278
                    logmsg='Reconciliation: simulated Ganeti event')
279
                # process_op_status with beparams will set the vmstate to
280
                # shutdown. Fix this be returning it to old state
281
                vm = VirtualMachine.objects.get(pk=server_id)
282
                vm.operstate = old_state
283
                vm.save()
284
                self.log.debug("Simulated Ganeti flavor event for server '%s'",
285
                               server_id)
286

    
287
    def reconcile_unsynced_nics(self, server_id, db_server, gnt_server):
288
        building_time = (self.event_time -
289
                         timedelta(seconds=backend_mod.BUILDING_NIC_TIMEOUT))
290
        db_nics = db_server.nics.exclude(state="BUILDING",
291
                                         created__lte=building_time) \
292
                                .order_by("index")
293
        gnt_nics = gnt_server["nics"]
294
        gnt_nics_parsed = backend_mod.process_ganeti_nics(gnt_nics)
295
        nics_changed = len(db_nics) != len(gnt_nics)
296
        for db_nic, gnt_nic in zip(db_nics, sorted(gnt_nics_parsed.items())):
297
            gnt_nic_id, gnt_nic = gnt_nic
298
            if (db_nic.id == gnt_nic_id) and\
299
               backend_mod.nics_are_equal(db_nic, gnt_nic):
300
                continue
301
            else:
302
                nics_changed = True
303
                break
304
        if nics_changed:
305
            msg = "Found unsynced NICs for server '%s'.\n"\
306
                  "\tDB:\n\t\t%s\n\tGaneti:\n\t\t%s"
307
            db_nics_str = "\n\t\t".join(map(format_db_nic, db_nics))
308
            gnt_nics_str = "\n\t\t".join(map(format_gnt_nic,
309
                                         gnt_nics_parsed.items()))
310
            self.log.info(msg, server_id, db_nics_str, gnt_nics_str)
311
            if self.options["fix_unsynced_nics"]:
312
                backend_mod.process_net_status(vm=db_server,
313
                                               etime=self.event_time,
314
                                               nics=gnt_nics)
315

    
316
    def reconcile_unsynced_disks(self, server_id, db_server, gnt_server):
317
        pass
318

    
319
    def reconcile_pending_task(self, server_id, db_server):
320
        job_id = db_server.task_job_id
321
        pending_task = False
322
        if job_id not in self.gnt_jobs:
323
            pending_task = True
324
        else:
325
            gnt_job_status = self.gnt_jobs[job_id]["status"]
326
            if gnt_job_status in GANETI_JOBS_FINALIZED:
327
                pending_task = True
328

    
329
        if pending_task:
330
            self.log.info("Found server '%s' with pending task: '%s'",
331
                          server_id, db_server.task)
332
            if self.options["fix_pending_tasks"]:
333
                db_server.task = None
334
                db_server.task_job_id = None
335
                db_server.save()
336
                self.log.info("Cleared pending task for server '%s", server_id)
337

    
338

    
339
NIC_MSG = ": %s\t".join(["ID", "State", "IP", "Network", "MAC", "Index",
340
                         "Firewall"]) + ": %s"
341

    
342

    
343
def format_db_nic(nic):
344
    return NIC_MSG % (nic.id, nic.state, nic.ipv4, nic.network_id, nic.mac,
345
                      nic.index, nic.firewall_profile)
346

    
347

    
348
def format_gnt_nic(nic):
349
    nic_name, nic = nic
350
    return NIC_MSG % (nic_name, nic["state"], nic["ipv4"], nic["network"],
351
                      nic["mac"], nic["index"], nic["firewall_profile"])
352

    
353

    
354
#
355
# Networks
356
#
357

    
358

    
359
def get_networks_from_ganeti(backend):
360
    prefix = settings.BACKEND_PREFIX_ID + 'net-'
361

    
362
    networks = {}
363
    with pooled_rapi_client(backend) as c:
364
        for net in c.GetNetworks(bulk=True):
365
            if net['name'].startswith(prefix):
366
                id = utils.id_from_network_name(net['name'])
367
                networks[id] = net
368

    
369
    return networks
370

    
371

    
372
def hanging_networks(backend, GNets):
373
    """Get networks that are not connected to all Nodegroups.
374

375
    """
376
    def get_network_groups(group_list):
377
        groups = set()
378
        for (name, mode, link) in group_list:
379
            groups.add(name)
380
        return groups
381

    
382
    with pooled_rapi_client(backend) as c:
383
        groups = set(c.GetGroups())
384

    
385
    hanging = {}
386
    for id, info in GNets.items():
387
        group_list = get_network_groups(info['group_list'])
388
        if group_list != groups:
389
            hanging[id] = groups - group_list
390
    return hanging
391

    
392

    
393
def get_online_backends():
394
    return Backend.objects.filter(offline=False)
395

    
396

    
397
def get_database_servers(backend):
398
    servers = backend.virtual_machines.select_related("nics", "flavor")\
399
                                      .filter(deleted=False)
400
    return dict([(s.id, s) for s in servers])
401

    
402

    
403
def get_ganeti_servers(backend):
404
    gnt_instances = backend_mod.get_instances(backend)
405
    # Filter out non-synnefo instances
406
    snf_backend_prefix = settings.BACKEND_PREFIX_ID
407
    gnt_instances = filter(lambda i: i["name"].startswith(snf_backend_prefix),
408
                           gnt_instances)
409
    gnt_instances = map(parse_gnt_instance, gnt_instances)
410
    return dict([(i["id"], i) for i in gnt_instances if i["id"] is not None])
411

    
412

    
413
def parse_gnt_instance(instance):
414
    try:
415
        instance_id = utils.id_from_instance_name(instance['name'])
416
    except Exception:
417
        logger.error("Ignoring instance with malformed name %s",
418
                     instance['name'])
419
        return (None, None)
420

    
421
    beparams = instance["beparams"]
422

    
423
    vcpus = beparams["vcpus"]
424
    ram = beparams["maxmem"]
425
    state = instance["oper_state"] and "STARTED" or "STOPPED"
426

    
427
    return {
428
        "id": instance_id,
429
        "state": state,  # FIX
430
        "updated": datetime.fromtimestamp(instance["mtime"]),
431
        "disks": disks_from_instance(instance),
432
        "nics": nics_from_instance(instance),
433
        "flavor": {"vcpus": vcpus,
434
                   "ram": ram},
435
        "tags": instance["tags"]
436
    }
437

    
438

    
439
def nics_from_instance(i):
440
    ips = zip(itertools.repeat('ip'), i['nic.ips'])
441
    names = zip(itertools.repeat('name'), i['nic.names'])
442
    macs = zip(itertools.repeat('mac'), i['nic.macs'])
443
    networks = zip(itertools.repeat('network'), i['nic.networks.names'])
444
    # modes = zip(itertools.repeat('mode'), i['nic.modes'])
445
    # links = zip(itertools.repeat('link'), i['nic.links'])
446
    # nics = zip(ips,macs,modes,networks,links)
447
    nics = zip(ips, names, macs, networks)
448
    nics = map(lambda x: dict(x), nics)
449
    #nics = dict(enumerate(nics))
450
    tags = i["tags"]
451
    for tag in tags:
452
        t = tag.split(":")
453
        if t[0:2] == ["synnefo", "network"]:
454
            if len(t) != 4:
455
                logger.error("Malformed synefo tag %s", tag)
456
                continue
457
            try:
458
                index = int(t[2])
459
                nics[index]['firewall'] = t[3]
460
            except ValueError:
461
                logger.error("Malformed synnefo tag %s", tag)
462
            except IndexError:
463
                logger.error("Found tag %s for non-existent NIC %d",
464
                             tag, index)
465
    return nics
466

    
467

    
468
def get_ganeti_jobs(backend):
469
    gnt_jobs = backend_mod.get_jobs(backend)
470
    return dict([(int(j["id"]), j) for j in gnt_jobs])
471

    
472

    
473
def disks_from_instance(i):
474
    return dict([(index, {"size": size})
475
                 for index, size in enumerate(i["disk.sizes"])])
476

    
477

    
478
class NetworkReconciler(object):
479
    def __init__(self, logger, fix=False):
480
        self.log = logger
481
        self.fix = fix
482

    
483
    @transaction.commit_on_success
484
    def reconcile_networks(self):
485
        # Get models from DB
486
        self.backends = Backend.objects.exclude(offline=True)
487
        self.networks = Network.objects.filter(deleted=False)
488

    
489
        self.event_time = datetime.now()
490

    
491
        # Get info from all ganeti backends
492
        self.ganeti_networks = {}
493
        self.ganeti_hanging_networks = {}
494
        for b in self.backends:
495
            g_nets = get_networks_from_ganeti(b)
496
            self.ganeti_networks[b] = g_nets
497
            g_hanging_nets = hanging_networks(b, g_nets)
498
            self.ganeti_hanging_networks[b] = g_hanging_nets
499

    
500
        self._reconcile_orphan_networks()
501

    
502
        for network in self.networks:
503
            self._reconcile_network(network)
504

    
505
    @transaction.commit_on_success
506
    def _reconcile_network(self, network):
507
        """Reconcile a network with corresponging Ganeti networks.
508

509
        Reconcile a Network and the associated BackendNetworks with the
510
        corresponding Ganeti networks in all Ganeti backends.
511

512
        """
513
        network_ip_pool = network.get_pool()  # X-Lock on IP Pool
514
        for bend in self.backends:
515
            bnet = get_backend_network(network, bend)
516
            gnet = self.ganeti_networks[bend].get(network.id)
517
            if not bnet:
518
                if network.floating_ip_pool:
519
                    # Network is a floating IP pool and does not exist in
520
                    # backend. We need to create it
521
                    bnet = self.reconcile_parted_network(network, bend)
522
                elif not gnet:
523
                    # Network does not exist either in Ganeti nor in BD.
524
                    continue
525
                else:
526
                    # Network exists in Ganeti and not in DB.
527
                    if network.action != "DESTROY" and not network.public:
528
                        bnet = self.reconcile_parted_network(network, bend)
529
                    else:
530
                        continue
531

    
532
            if not gnet:
533
                # Network does not exist in Ganeti. If the network action
534
                # is DESTROY, we have to mark as deleted in DB, else we
535
                # have to create it in Ganeti.
536
                if network.action == "DESTROY":
537
                    if bnet.operstate != "DELETED":
538
                        self.reconcile_stale_network(bnet)
539
                else:
540
                    self.reconcile_missing_network(network, bend)
541
                # Skip rest reconciliation!
542
                continue
543

    
544
            try:
545
                hanging_groups = self.ganeti_hanging_networks[bend][network.id]
546
            except KeyError:
547
                # Network is connected to all nodegroups
548
                hanging_groups = []
549

    
550
            if hanging_groups:
551
                # CASE-3: Ganeti networks not connected to all nodegroups
552
                self.reconcile_hanging_groups(network, bend,
553
                                              hanging_groups)
554
                continue
555

    
556
            if bnet.operstate != 'ACTIVE':
557
                # CASE-4: Unsynced network state. At this point the network
558
                # exists and is connected to all nodes so is must be
559
                # active!
560
                self.reconcile_unsynced_network(network, bend, bnet)
561

    
562
            # Check that externally reserved IPs of the network in Ganeti are
563
            # also externally reserved to the IP pool
564
            externally_reserved = gnet['external_reservations']
565
            if externally_reserved:
566
                for ip in externally_reserved.split(","):
567
                    ip = ip.strip()
568
                    if not network_ip_pool.is_reserved(ip):
569
                        msg = ("D: IP '%s' is reserved for network '%s' in"
570
                               " backend '%s' but not in DB.")
571
                        self.log.info(msg, ip, network, bend)
572
                        if self.fix:
573
                            network_ip_pool.reserve(ip, external=True)
574
                            network_ip_pool.save()
575
                            self.log.info("F: Reserved IP '%s'", ip)
576

    
577
    def reconcile_parted_network(self, network, backend):
578
        self.log.info("D: Missing DB entry for network %s in backend %s",
579
                      network, backend)
580
        if self.fix:
581
            network.create_backend_network(backend)
582
            self.log.info("F: Created DB entry")
583
            bnet = get_backend_network(network, backend)
584
            return bnet
585

    
586
    def reconcile_stale_network(self, backend_network):
587
        self.log.info("D: Stale DB entry for network %s in backend %s",
588
                      backend_network.network, backend_network.backend)
589
        if self.fix:
590
            backend_mod.process_network_status(
591
                backend_network, self.event_time, 0,
592
                "OP_NETWORK_REMOVE",
593
                "success",
594
                "Reconciliation simulated event")
595
            self.log.info("F: Reconciled event: OP_NETWORK_REMOVE")
596

    
597
    def reconcile_missing_network(self, network, backend):
598
        self.log.info("D: Missing Ganeti network %s in backend %s",
599
                      network, backend)
600
        if self.fix:
601
            backend_mod.create_network(network, backend)
602
            self.log.info("F: Issued OP_NETWORK_CONNECT")
603

    
604
    def reconcile_hanging_groups(self, network, backend, hanging_groups):
605
        self.log.info('D: Network %s in backend %s is not connected to '
606
                      'the following groups:', network, backend)
607
        self.log.info('-  ' + '\n-  '.join(hanging_groups))
608
        if self.fix:
609
            for group in hanging_groups:
610
                self.log.info('F: Connecting network %s to nodegroup %s',
611
                              network, group)
612
                backend_mod.connect_network(network, backend, depends=[],
613
                                            group=group)
614

    
615
    def reconcile_unsynced_network(self, network, backend, backend_network):
616
        self.log.info("D: Unsynced network %s in backend %s", network, backend)
617
        if self.fix:
618
            self.log.info("F: Issuing OP_NETWORK_CONNECT")
619
            backend_mod.process_network_status(
620
                backend_network, self.event_time, 0,
621
                "OP_NETWORK_CONNECT",
622
                "success",
623
                "Reconciliation simulated eventd")
624

    
625
    def _reconcile_orphan_networks(self):
626
        db_networks = self.networks
627
        ganeti_networks = self.ganeti_networks
628
        # Detect Orphan Networks in Ganeti
629
        db_network_ids = set([net.id for net in db_networks])
630
        for back_end, ganeti_networks in ganeti_networks.items():
631
            ganeti_network_ids = set(ganeti_networks.keys())
632
            orphans = ganeti_network_ids - db_network_ids
633

    
634
            if len(orphans) > 0:
635
                self.log.info('D: Orphan Networks in backend %s:',
636
                              back_end.clustername)
637
                self.log.info('-  ' + '\n-  '.join([str(o) for o in orphans]))
638
                if self.fix:
639
                    for net_id in orphans:
640
                        self.log.info('Disconnecting and deleting network %d',
641
                                      net_id)
642
                        try:
643
                            network = Network.objects.get(id=net_id)
644
                            backend_mod.delete_network(network,
645
                                                       backend=back_end)
646
                        except Network.DoesNotExist:
647
                            self.log.info("Not entry for network %s in DB !!",
648
                                          net_id)
649

    
650

    
651
def get_backend_network(network, backend):
652
    try:
653
        return BackendNetwork.objects.get(network=network, backend=backend)
654
    except BackendNetwork.DoesNotExist:
655
        return None
656

    
657

    
658
class PoolReconciler(object):
659
    def __init__(self, logger, fix=False):
660
        self.log = logger
661
        self.fix = fix
662

    
663
    def reconcile(self):
664
        self.reconcile_bridges()
665
        self.reconcile_mac_prefixes()
666
        for network in Network.objects.filter(deleted=False):
667
            self.reconcile_ip_pool(network)
668

    
669
    @transaction.commit_on_success
670
    def reconcile_bridges(self):
671
        networks = Network.objects.filter(deleted=False,
672
                                          flavor="PHYSICAL_VLAN")
673
        check_unique_values(objects=networks, field='link', logger=self.log)
674
        try:
675
            pool = BridgePoolTable.get_pool()
676
        except pools.EmptyPool:
677
            self.log.info("There is no available pool for bridges.")
678
            return
679

    
680
        used_bridges = set(networks.values_list('link', flat=True))
681
        check_pool_consistent(pool=pool, pool_class=pools.BridgePool,
682
                              used_values=used_bridges, fix=self.fix,
683
                              logger=self.log)
684

    
685
    @transaction.commit_on_success
686
    def reconcile_mac_prefixes(self):
687
        networks = Network.objects.filter(deleted=False, flavor="MAC_FILTERED")
688
        check_unique_values(objects=networks, field='mac_prefix',
689
                            logger=self.log)
690
        try:
691
            pool = MacPrefixPoolTable.get_pool()
692
        except pools.EmptyPool:
693
            self.log.info("There is no available pool for MAC prefixes.")
694
            return
695

    
696
        used_mac_prefixes = set(networks.values_list('mac_prefix', flat=True))
697
        check_pool_consistent(pool=pool, pool_class=pools.MacPrefixPool,
698
                              used_values=used_mac_prefixes, fix=self.fix,
699
                              logger=self.log)
700

    
701
    @transaction.commit_on_success
702
    def reconcile_ip_pool(self, network):
703
        # Check that all NICs have unique IPv4 address
704
        nics = network.nics.filter(ipv4__isnull=False)
705
        check_unique_values(objects=nics, field='ipv4', logger=self.log)
706

    
707
        # Check that all Floating IPs have unique IPv4 address
708
        floating_ips = network.floating_ips.filter(deleted=False)
709
        check_unique_values(objects=floating_ips, field='ipv4',
710
                            logger=self.log)
711

    
712
        # First get(lock) the IP pool of the network to prevent new NICs
713
        # from being created.
714
        network_ip_pool = network.get_pool()
715
        used_ips = set(list(nics.values_list("ipv4", flat=True)) +
716
                       list(floating_ips.values_list("ipv4", flat=True)))
717

    
718
        check_pool_consistent(pool=network_ip_pool,
719
                              pool_class=pools.IPPool,
720
                              used_values=used_ips,
721
                              fix=self.fix, logger=self.log)
722

    
723

    
724
def check_unique_values(objects, field, logger):
725
    used_values = list(objects.values_list(field, flat=True))
726
    if len(used_values) != len(set(used_values)):
727
        duplicate_values = [v for v in used_values if used_values.count(v) > 1]
728
        for value in duplicate_values:
729
            filter_args = {field: value}
730
            using_objects = objects.filter(**filter_args)
731
            msg = "Value '%s' is used as %s for more than one objects: %s"
732
            logger.error(msg, value, field, ",".join(map(str, using_objects)))
733
        return False
734
    logger.debug("Values for field '%s' are unique.", field)
735
    return True
736

    
737

    
738
def check_pool_consistent(pool, pool_class, used_values, fix, logger):
739
    dummy_pool = create_empty_pool(pool, pool_class)
740
    [dummy_pool.reserve(value) for value in used_values]
741
    if dummy_pool.available != pool.available:
742
        msg = "'%s' is not consistent!\nPool: %s\nUsed: %s"
743
        pool_diff = dummy_pool.available ^ pool.available
744
        for index in pool_diff.itersearch(bitarray.bitarray("1")):
745
            value = pool.index_to_value(int(index))
746
            msg = "%s is incosistent! Value '%s' is %s but should be %s."
747
            value1 = pool.is_available(value) and "available" or "unavailable"
748
            value2 = dummy_pool.is_available(value) and "available"\
749
                or "unavailable"
750
            logger.error(msg, pool, value, value1, value2)
751
        if fix:
752
            pool.available = dummy_pool.available
753
            pool.save()
754
            logger.info("Fixed available map of pool '%s'", pool)
755

    
756

    
757
def create_empty_pool(pool, pool_class):
758
    pool_row = pool.pool_table
759
    pool_row.available_map = ""
760
    pool_row.reserved_map = ""
761
    return pool_class(pool_row)