Statistics
| Branch: | Tag: | Revision:

root / logic / callbacks.py @ 583bfaa0

History | View | Annotate | Download (8.3 kB)

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

    
30
# Callback functions used by the dispatcher to process incoming notifications
31
# from AMQP queues.
32

    
33
import socket
34
import traceback
35
import json
36
import sys
37

    
38
from synnefo.db.models import VirtualMachine
39
from synnefo.logic import utils, backend, email_send, log
40

    
41
_logger = log.get_logger("synnefo.dispatcher")
42

    
43
def update_db(message):
44
    """Process the status of a VM based on a ganeti status message"""
45
    _logger.debug("Processing ganeti-op-status msg: %s", message.body)
46
    try:
47
        msg = _parse_json(message.body)
48

    
49
        if msg["type"] != "ganeti-op-status":
50
            _logger.error("Message is of unknown type %s.", msg["type"])
51
            return
52

    
53
        if msg["operation"] == "OP_INSTANCE_QUERY_DATA":
54
            return status_job_finished(message)
55

    
56
        vmid = utils.id_from_instance_name(msg["instance"])
57
        vm = VirtualMachine.objects.get(id=vmid)
58

    
59
        backend.process_op_status(vm, msg["jobId"], msg["operation"],
60
                                  msg["status"], msg["logmsg"])
61
        _logger.debug("Done processing ganeti-op-status msg for vm %s.",
62
                      msg["instance"])
63
        message.channel.basic_ack(message.delivery_tag)
64
    except KeyError:
65
        _logger.error("Malformed incoming JSON, missing attributes: %s",
66
                      message.body)
67
    except VirtualMachine.InvalidBackendIdError:
68
        _logger.debug("Ignoring msg for unknown instance %s.",
69
                      msg["instance"])
70
    except VirtualMachine.DoesNotExist:
71
        _logger.error("VM for instance %s with id %d not found in DB.",
72
                      msg["instance"], vmid)
73
    except Exception as e:
74
        _logger.error("Unexpected error:\n%s" %
75
            "".join(traceback.format_exception(*sys.exc_info())))
76

    
77

    
78
def update_net(message):
79
    """Process a network status update notification from Ganeti"""
80
    _logger.debug("Processing ganeti-net-status msg: %s", message.body)
81
    try:
82
        msg = _parse_json(message.body)
83

    
84
        if msg["type"] != "ganeti-net-status":
85
            _logger.error("Message is of unknown type %s", msg["type"])
86
            return
87

    
88
        vmid = utils.id_from_instance_name(msg["instance"])
89
        vm = VirtualMachine.objects.get(id=vmid)
90

    
91
        backend.process_net_status(vm, msg["nics"])
92
        _logger.debug("Done processing ganeti-net-status msg for vm %s.",
93
                      msg["instance"])
94
        message.channel.basic_ack(message.delivery_tag)
95
    except KeyError:
96
        _logger.error("Malformed incoming JSON, missing attributes: %s",
97
                      message.body)
98
    except VirtualMachine.InvalidBackendIdError:
99
        _logger.debug("Ignoring msg for unknown instance %s.",
100
                      msg["instance"])
101
    except VirtualMachine.DoesNotExist:
102
        _logger.error("VM for instance %s with id %d not found in DB.",
103
                      msg["instance"], vmid)
104
    except Exception as e:
105
        _logger.error("Unexpected error:\n%s" %
106
            "".join(traceback.format_exception(*sys.exc_info())))
107

    
108

    
109
def send_email(message):
110
    """Process an email submission request"""
111

    
112
    try:
113
        msg = json.loads(message.body)
114

    
115
        sent = email_send.send(sender=msg['frm'], recipient = msg['to'],
116
                        body=msg['body'], subject=msg['subject'])
117

    
118
        if not sent:
119
            _logger.warn("Failed to send email to %s", msg['to'])
120
        else:
121
            message.channel.basic_ack(message.delivery_tag)
122
    except KeyError:
123
        _logger.error("Malformed incoming JSON, missing attributes: %s",
124
                      message.body)
125
    except socket.error as e:
126
        _logger.error("Cannot connect to SMTP server:%s\n", e)
127
    except Exception as e:
128
        _logger.error("Unexpected error:%s\n", e)
129
        raise
130

    
131

    
132
def update_credits(message):
133
    _logger.debug("Request to update credits")
134
    message.channel.basic_ack(message.delivery_tag)
135

    
136
def trigger_status_update(message):
137
    """
138
        Triggers a status update job for a specific VM id.
139
    """
140
    _logger.debug("Request to trigger status update: %s", message.body)
141

    
142
    try:
143
        msg = _parse_json(message.body)
144

    
145
        if msg["type"] != "reconcile":
146
             _logger.error("Message is of unknown type %s", msg["type"])
147
             return
148

    
149
        if msg["vmid"] == "":
150
            _logger.error("Reconciliate message does not specify a VM id")
151
            return
152

    
153
        vm = VirtualMachine.objects.get(id=msg["vmid"])
154
        backend.request_status_update(vm)
155

    
156
        message.channel.basic_ack(message.delivery_tag)
157
    except KeyError as k:
158
        _logger.error("Malformed incoming JSON, missing attributes: %s", k)
159
    except Exception as e:
160
        _logger.error("Unexpected error:%s", e)
161

    
162
def status_job_finished (message):
163
    """
164
        Updates VM status based on a previously sent status update request
165
    """
166
    try:
167
        msg = _parse_json(message.body)
168

    
169
        if msg["operation"] != 'OP_INSTANCE_QUERY_DATA':
170
            _logger.error("Message is of unknown type %s", msg["operation"])
171
            return
172

    
173
        if msg["status"] != "success":
174
            _logger.warn("Ignoring non-success status update from job %d on VM %s",
175
                          msg['jobId'], msg['instance'])
176
            message.channel.basic_ack(message.delivery_tag)
177
            return
178

    
179
        status = backend.get_job_status(msg['jobId'])
180

    
181
        _logger.debug("Node status job result: %s" % status)
182

    
183
        if status['summary'][0] != u'INSTANCE_QUERY_DATA':
184
             _logger.error("Status update is of unknown type %s", status['summary'])
185
             return
186

    
187
        conf_state = status['opresult'][0][msg['instance']]['config_state']
188
        run_state = status['opresult'][0][msg['instance']]['run_state']
189

    
190
        # XXX: The following assumes names like snf-12
191
        instid = msg['instance'].split('-')[1]
192

    
193
        vm = VirtualMachine.objects.get(id = instid)
194

    
195
        if run_state == "up":
196
            opcode = "OP_INSTANCE_REBOOT"
197
        else:
198
            opcode = "OP_INSTANCE_SHUTDOWN"
199

    
200
        backend.process_op_status(vm=vm, jobid=msg['jobId'],opcode=opcode,
201
                                  status="success",
202
                                  logmsg="Reconciliation: simulated event")
203

    
204
        message.channel.basic_ack(message.delivery_tag)
205
    except KeyError as k:
206
        _logger.error("Malformed incoming JSON, missing attributes: %s", k)
207
    except Exception as e:
208
        _logger.error("Unexpected error:%s"%e)
209

    
210
def dummy_proc(message):
211
    try:
212
        msg = _logger.debug(message.body)
213
        _logger.debug("Msg (exchange:%s) ", msg)
214
        message.channel.basic_ack(message.delivery_tag)
215
    except Exception as e:
216
        _logger.error("Could not receive message %s" % e.message)
217
        pass
218

    
219
def _parse_json(data):
220
    try:
221
        return json.loads(data)
222
    except Exception as e:
223
        _logger.error("Could not parse JSON file: %s", e)
224
        raise