Statistics
| Branch: | Tag: | Revision:

root / snf-cyclades-gtools / synnefo / ganeti / eventd.py @ b2272468

History | View | Annotate | Download (16 kB)

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

40 b024b97a Georgios Gousios
A daemon to monitor the Ganeti job queue and publish job progress
41 80bd5072 Georgios Gousios
and Ganeti VM state notifications to the ganeti exchange
42 b024b97a Georgios Gousios
"""
43 b024b97a Georgios Gousios
44 b024b97a Georgios Gousios
import sys
45 b024b97a Georgios Gousios
import os
46 b024b97a Georgios Gousios
path = os.path.normpath(os.path.join(os.getcwd(), '..'))
47 b024b97a Georgios Gousios
sys.path.append(path)
48 3bac1e87 Christos Stavrakakis
# Since Ganeti 2.7, debian package ships the majority of the python code in
49 3bac1e87 Christos Stavrakakis
# a private module under '/usr/share/ganeti'. Add this directory to path
50 3bac1e87 Christos Stavrakakis
# in order to be able to import ganeti. Also, add it to the start of path
51 3bac1e87 Christos Stavrakakis
# to allow conflicts with Ganeti RAPI client.
52 3bac1e87 Christos Stavrakakis
sys.path.insert(0, "/usr/share/ganeti")
53 b024b97a Georgios Gousios
54 b024b97a Georgios Gousios
import json
55 b024b97a Georgios Gousios
import logging
56 b024b97a Georgios Gousios
import pyinotify
57 b024b97a Georgios Gousios
import daemon
58 b024b97a Georgios Gousios
import daemon.pidlockfile
59 cf2a3529 Christos Stavrakakis
import daemon.runner
60 cf2a3529 Christos Stavrakakis
from lockfile import LockTimeout
61 b024b97a Georgios Gousios
from signal import signal, SIGINT, SIGTERM
62 1dc821c9 Christos Stavrakakis
import setproctitle
63 b024b97a Georgios Gousios
64 2b4becab Christos Stavrakakis
from ganeti import utils, jqueue, constants, serializer, pathutils, cli
65 6df16263 Christos Stavrakakis
from ganeti import errors as ganeti_errors
66 2b4becab Christos Stavrakakis
from ganeti.ssconf import SimpleStore
67 e66a7e34 Christos Stavrakakis
68 b024b97a Georgios Gousios
69 6d6b8f88 Kostas Papadimitriou
from synnefo import settings
70 c4e55622 Christos Stavrakakis
from synnefo.lib.amqp import AMQPClient
71 c4e55622 Christos Stavrakakis
72 27a8e4ae Christos Stavrakakis
73 c4e55622 Christos Stavrakakis
def get_time_from_status(op, job):
74 c4e55622 Christos Stavrakakis
    """Generate a unique message identifier for a ganeti job.
75 c4e55622 Christos Stavrakakis

76 c4e55622 Christos Stavrakakis
    The identifier is based on the timestamp of the job. Since a ganeti
77 c4e55622 Christos Stavrakakis
    job passes from multiple states, we need to pick the timestamp that
78 c4e55622 Christos Stavrakakis
    corresponds to each state.
79 c4e55622 Christos Stavrakakis

80 c4e55622 Christos Stavrakakis
    """
81 c4e55622 Christos Stavrakakis
    status = op.status
82 c4e55622 Christos Stavrakakis
    if status == constants.JOB_STATUS_QUEUED:
83 765ff3ff Christos Stavrakakis
        time = job.received_timestamp
84 27a8e4ae Christos Stavrakakis
    try:  # Compatibility with Ganeti version
85 39ccdb18 Christos Stavrakakis
        if status == constants.JOB_STATUS_WAITLOCK:
86 765ff3ff Christos Stavrakakis
            time = op.start_timestamp
87 39ccdb18 Christos Stavrakakis
    except AttributeError:
88 39ccdb18 Christos Stavrakakis
        if status == constants.JOB_STATUS_WAITING:
89 765ff3ff Christos Stavrakakis
            time = op.start_timestamp
90 c4e55622 Christos Stavrakakis
    if status == constants.JOB_STATUS_CANCELING:
91 765ff3ff Christos Stavrakakis
        time = op.start_timestamp
92 c4e55622 Christos Stavrakakis
    if status == constants.JOB_STATUS_RUNNING:
93 765ff3ff Christos Stavrakakis
        time = op.exec_timestamp
94 c4e55622 Christos Stavrakakis
    if status in constants.JOBS_FINALIZED:
95 765ff3ff Christos Stavrakakis
        time = op.end_timestamp
96 765ff3ff Christos Stavrakakis
97 765ff3ff Christos Stavrakakis
    return time and time or job.end_timestamp
98 c4e55622 Christos Stavrakakis
99 27a8e4ae Christos Stavrakakis
    raise InvalidBackendStatus(status, job)
100 c4e55622 Christos Stavrakakis
101 c4e55622 Christos Stavrakakis
102 0e1f3323 Christos Stavrakakis
def get_instance_nics(instance, logger):
103 0e1f3323 Christos Stavrakakis
    """Query Ganeti to a get the instance's NICs.
104 0e1f3323 Christos Stavrakakis

105 6df16263 Christos Stavrakakis
    Get instance's NICs from Ganeti configuration data. If running on master,
106 6df16263 Christos Stavrakakis
    query Ganeti via Ganeti CLI client. Otherwise, get the nics from Ganeti
107 6df16263 Christos Stavrakakis
    configuration file.
108 6df16263 Christos Stavrakakis

109 0e1f3323 Christos Stavrakakis
    @type instance: string
110 0e1f3323 Christos Stavrakakis
    @param instance: the name of the instance
111 0e1f3323 Christos Stavrakakis
    @rtype: List of dicts
112 6df16263 Christos Stavrakakis
    @return: Dictionary containing the instance's NICs. Each dictionary
113 0e1f3323 Christos Stavrakakis
             contains the following keys: 'network', 'ip', 'mac', 'mode',
114 0e1f3323 Christos Stavrakakis
             'link' and 'firewall'
115 0e1f3323 Christos Stavrakakis

116 0e1f3323 Christos Stavrakakis
    """
117 6df16263 Christos Stavrakakis
    try:
118 6df16263 Christos Stavrakakis
        client = cli.GetClient()
119 dee2e82c Christos Stavrakakis
        fields = ["nic.networks.names", "nic.ips", "nic.macs", "nic.modes",
120 b2272468 Christos Stavrakakis
                  "nic.links", "nic.names", "tags"]
121 6df16263 Christos Stavrakakis
        info = client.QueryInstances([instance], fields, use_locking=False)
122 b2272468 Christos Stavrakakis
        networks, ips, macs, modes, links, names, tags = info[0]
123 b2272468 Christos Stavrakakis
        nic_keys = ["network", "ip", "mac", "mode", "link", "name"]
124 b2272468 Christos Stavrakakis
        nics = zip(networks, ips, macs, modes, links, names)
125 6df16263 Christos Stavrakakis
        nics = map(lambda x: dict(zip(nic_keys, x)), nics)
126 6df16263 Christos Stavrakakis
    except ganeti_errors.OpPrereqError:
127 6df16263 Christos Stavrakakis
        # Not running on master! Load the conf file
128 ffc4b474 Christos Stavrakakis
        raw_data = utils.ReadFile(pathutils.CLUSTER_CONF_FILE)
129 6df16263 Christos Stavrakakis
        config = serializer.LoadJson(raw_data)
130 6df16263 Christos Stavrakakis
        i = config["instances"][instance]
131 6df16263 Christos Stavrakakis
        nics = []
132 6df16263 Christos Stavrakakis
        for nic in i["nics"]:
133 6df16263 Christos Stavrakakis
            params = nic.pop("nicparams")
134 6df16263 Christos Stavrakakis
            nic["mode"] = params["mode"]
135 6df16263 Christos Stavrakakis
            nic["link"] = params["link"]
136 6df16263 Christos Stavrakakis
            nics.append(nic)
137 6df16263 Christos Stavrakakis
        tags = i.get("tags", [])
138 0e1f3323 Christos Stavrakakis
    # Get firewall from instance Tags
139 0e1f3323 Christos Stavrakakis
    # Tags are of the form synnefo:network:N:firewall_mode
140 0e1f3323 Christos Stavrakakis
    for tag in tags:
141 0e1f3323 Christos Stavrakakis
        t = tag.split(":")
142 0e1f3323 Christos Stavrakakis
        if t[0:2] == ["synnefo", "network"]:
143 0e1f3323 Christos Stavrakakis
            if len(t) != 4:
144 0e1f3323 Christos Stavrakakis
                logger.error("Malformed synefo tag %s", tag)
145 0e1f3323 Christos Stavrakakis
                continue
146 0e1f3323 Christos Stavrakakis
            try:
147 0e1f3323 Christos Stavrakakis
                index = int(t[2])
148 0e1f3323 Christos Stavrakakis
                nics[index]['firewall'] = t[3]
149 0e1f3323 Christos Stavrakakis
            except ValueError:
150 0e1f3323 Christos Stavrakakis
                logger.error("Malformed synnefo tag %s", tag)
151 0e1f3323 Christos Stavrakakis
            except IndexError:
152 0e1f3323 Christos Stavrakakis
                logger.error("Found tag %s for non-existent NIC %d",
153 0e1f3323 Christos Stavrakakis
                             tag, index)
154 0e1f3323 Christos Stavrakakis
    return nics
155 0e1f3323 Christos Stavrakakis
156 0e1f3323 Christos Stavrakakis
157 c4e55622 Christos Stavrakakis
class InvalidBackendStatus(Exception):
158 c4e55622 Christos Stavrakakis
    def __init__(self, status, job):
159 c4e55622 Christos Stavrakakis
        self.status = status
160 c4e55622 Christos Stavrakakis
        self.job = job
161 c4e55622 Christos Stavrakakis
162 c4e55622 Christos Stavrakakis
    def __str__(self):
163 c4e55622 Christos Stavrakakis
        return repr("Invalid backend status: %s in job %s"
164 c4e55622 Christos Stavrakakis
                    % (self.status, self.job))
165 c4e55622 Christos Stavrakakis
166 27a8e4ae Christos Stavrakakis
167 39ccdb18 Christos Stavrakakis
def prefix_from_name(name):
168 39ccdb18 Christos Stavrakakis
    return name.split('-')[0]
169 39ccdb18 Christos Stavrakakis
170 39ccdb18 Christos Stavrakakis
171 39ccdb18 Christos Stavrakakis
def get_field(from_, field):
172 39ccdb18 Christos Stavrakakis
    try:
173 39ccdb18 Christos Stavrakakis
        return getattr(from_, field)
174 39ccdb18 Christos Stavrakakis
    except AttributeError:
175 39ccdb18 Christos Stavrakakis
        None
176 39ccdb18 Christos Stavrakakis
177 45ebfd48 Vangelis Koukis
178 b024b97a Georgios Gousios
class JobFileHandler(pyinotify.ProcessEvent):
179 27a8e4ae Christos Stavrakakis
    def __init__(self, logger, cluster_name):
180 348f53de Georgios Gousios
        pyinotify.ProcessEvent.__init__(self)
181 348f53de Georgios Gousios
        self.logger = logger
182 27a8e4ae Christos Stavrakakis
        self.cluster_name = cluster_name
183 27a8e4ae Christos Stavrakakis
184 2ef10562 Christos Stavrakakis
        # Set max_retries to 0 for unlimited retries.
185 2ef10562 Christos Stavrakakis
        self.client = AMQPClient(hosts=settings.AMQP_HOSTS, confirm_buffer=25,
186 a8858945 Christos Stavrakakis
                                 max_retries=0, logger=logger)
187 2ef10562 Christos Stavrakakis
188 c4e55622 Christos Stavrakakis
        handler_logger.info("Attempting to connect to RabbitMQ hosts")
189 27a8e4ae Christos Stavrakakis
190 c4e55622 Christos Stavrakakis
        self.client.connect()
191 c4e55622 Christos Stavrakakis
        handler_logger.info("Connected succesfully")
192 b024b97a Georgios Gousios
193 27a8e4ae Christos Stavrakakis
        self.client.exchange_declare(settings.EXCHANGE_GANETI, type='topic')
194 27a8e4ae Christos Stavrakakis
195 39ccdb18 Christos Stavrakakis
        self.op_handlers = {"INSTANCE": self.process_instance_op,
196 39ccdb18 Christos Stavrakakis
                            "NETWORK": self.process_network_op}
197 39ccdb18 Christos Stavrakakis
                            # "GROUP": self.process_group_op}
198 39ccdb18 Christos Stavrakakis
199 b024b97a Georgios Gousios
    def process_IN_CLOSE_WRITE(self, event):
200 80dcd124 Vangelis Koukis
        self.process_IN_MOVED_TO(event)
201 80dcd124 Vangelis Koukis
202 80dcd124 Vangelis Koukis
    def process_IN_MOVED_TO(self, event):
203 b024b97a Georgios Gousios
        jobfile = os.path.join(event.path, event.name)
204 b024b97a Georgios Gousios
        if not event.name.startswith("job-"):
205 b024b97a Georgios Gousios
            self.logger.debug("Not a job file: %s" % event.path)
206 b024b97a Georgios Gousios
            return
207 b024b97a Georgios Gousios
208 b024b97a Georgios Gousios
        try:
209 b024b97a Georgios Gousios
            data = utils.ReadFile(jobfile)
210 b024b97a Georgios Gousios
        except IOError:
211 b024b97a Georgios Gousios
            return
212 b024b97a Georgios Gousios
213 b024b97a Georgios Gousios
        data = serializer.LoadJson(data)
214 2b4becab Christos Stavrakakis
        job = jqueue._QueuedJob.Restore(None, data, False, False)
215 4ed30eed Christos Stavrakakis
216 39ccdb18 Christos Stavrakakis
        job_id = int(job.id)
217 b024b97a Georgios Gousios
218 b024b97a Georgios Gousios
        for op in job.ops:
219 39ccdb18 Christos Stavrakakis
            op_id = op.input.OP_ID
220 b024b97a Georgios Gousios
221 39ccdb18 Christos Stavrakakis
            msg = None
222 b024b97a Georgios Gousios
            try:
223 39ccdb18 Christos Stavrakakis
                handler_fn = self.op_handlers[op_id.split('_')[1]]
224 39ccdb18 Christos Stavrakakis
                msg, routekey = handler_fn(op, job_id)
225 39ccdb18 Christos Stavrakakis
            except KeyError:
226 b024b97a Georgios Gousios
                pass
227 b024b97a Georgios Gousios
228 39ccdb18 Christos Stavrakakis
            if not msg:
229 39ccdb18 Christos Stavrakakis
                self.logger.debug("Ignoring job: %s: %s", job_id, op_id)
230 07e4ab22 Christos Stavrakakis
                continue
231 07e4ab22 Christos Stavrakakis
232 39ccdb18 Christos Stavrakakis
            # Generate a unique message identifier
233 39ccdb18 Christos Stavrakakis
            event_time = get_time_from_status(op, job)
234 39ccdb18 Christos Stavrakakis
235 b024b97a Georgios Gousios
            # Get the last line of the op log as message
236 b024b97a Georgios Gousios
            try:
237 b024b97a Georgios Gousios
                logmsg = op.log[-1][-1]
238 b024b97a Georgios Gousios
            except IndexError:
239 b024b97a Georgios Gousios
                logmsg = None
240 348f53de Georgios Gousios
241 39ccdb18 Christos Stavrakakis
            # Add shared attributes for all operations
242 39ccdb18 Christos Stavrakakis
            msg.update({"event_time": event_time,
243 39ccdb18 Christos Stavrakakis
                        "operation": op_id,
244 39ccdb18 Christos Stavrakakis
                        "status": op.status,
245 27a8e4ae Christos Stavrakakis
                        "cluster": self.cluster_name,
246 39ccdb18 Christos Stavrakakis
                        "logmsg": logmsg,
247 aee560b0 Christos Stavrakakis
                        "result": op.result,
248 39ccdb18 Christos Stavrakakis
                        "jobId": job_id})
249 c4e55622 Christos Stavrakakis
250 0e1f3323 Christos Stavrakakis
            if op_id in ["OP_INSTANCE_CREATE", "OP_INSTANCE_SET_PARAMS",
251 0e1f3323 Christos Stavrakakis
                         "OP_INSTANCE_STARTUP"]:
252 0e1f3323 Christos Stavrakakis
                if op.status == "success":
253 0e1f3323 Christos Stavrakakis
                    nics = get_instance_nics(msg["instance"], self.logger)
254 0e1f3323 Christos Stavrakakis
                    msg["nics"] = nics
255 0e1f3323 Christos Stavrakakis
256 727fb2f9 Christos Stavrakakis
            if op_id == "OP_INSTANCE_CREATE" and op.status == "error":
257 727fb2f9 Christos Stavrakakis
                # In case an instance creation fails send the job input
258 727fb2f9 Christos Stavrakakis
                # so that the job can be retried if needed.
259 727fb2f9 Christos Stavrakakis
                msg["job_fields"] = op.Serialize()["input"]
260 727fb2f9 Christos Stavrakakis
261 c4e55622 Christos Stavrakakis
            msg = json.dumps(msg)
262 727fb2f9 Christos Stavrakakis
263 39ccdb18 Christos Stavrakakis
            self.logger.debug("Delivering msg: %s (key=%s)", msg, routekey)
264 c4e55622 Christos Stavrakakis
265 c4e55622 Christos Stavrakakis
            # Send the message to RabbitMQ
266 db400d82 Christos Stavrakakis
            self.client.basic_publish(settings.EXCHANGE_GANETI,
267 db400d82 Christos Stavrakakis
                                      routekey,
268 db400d82 Christos Stavrakakis
                                      msg)
269 39ccdb18 Christos Stavrakakis
270 39ccdb18 Christos Stavrakakis
    def process_instance_op(self, op, job_id):
271 39ccdb18 Christos Stavrakakis
        """ Process OP_INSTANCE_* opcodes.
272 39ccdb18 Christos Stavrakakis

273 39ccdb18 Christos Stavrakakis
        """
274 39ccdb18 Christos Stavrakakis
        input = op.input
275 39ccdb18 Christos Stavrakakis
        op_id = input.OP_ID
276 39ccdb18 Christos Stavrakakis
277 39ccdb18 Christos Stavrakakis
        instances = None
278 39ccdb18 Christos Stavrakakis
        instances = get_field(input, 'instance_name')
279 39ccdb18 Christos Stavrakakis
        if not instances:
280 39ccdb18 Christos Stavrakakis
            instances = get_field(input, 'instances')
281 39ccdb18 Christos Stavrakakis
            if not instances or len(instances) > 1:
282 39ccdb18 Christos Stavrakakis
                # Do not publish messages for jobs with no or multiple
283 449787d3 Christos Stavrakakis
                # instances.  Currently snf-dispatcher can not normally handle
284 449787d3 Christos Stavrakakis
                # these messages
285 39ccdb18 Christos Stavrakakis
                return None, None
286 39ccdb18 Christos Stavrakakis
            else:
287 39ccdb18 Christos Stavrakakis
                instances = instances[0]
288 39ccdb18 Christos Stavrakakis
289 39ccdb18 Christos Stavrakakis
        self.logger.debug("Job: %d: %s(%s) %s", job_id, op_id,
290 39ccdb18 Christos Stavrakakis
                          instances, op.status)
291 39ccdb18 Christos Stavrakakis
292 39ccdb18 Christos Stavrakakis
        msg = {"type": "ganeti-op-status",
293 39ccdb18 Christos Stavrakakis
               "instance": instances,
294 39ccdb18 Christos Stavrakakis
               "operation": op_id}
295 39ccdb18 Christos Stavrakakis
296 39ccdb18 Christos Stavrakakis
        routekey = "ganeti.%s.event.op" % prefix_from_name(instances)
297 39ccdb18 Christos Stavrakakis
298 39ccdb18 Christos Stavrakakis
        return msg, routekey
299 39ccdb18 Christos Stavrakakis
300 39ccdb18 Christos Stavrakakis
    def process_network_op(self, op, job_id):
301 39ccdb18 Christos Stavrakakis
        """ Process OP_NETWORK_* opcodes.
302 39ccdb18 Christos Stavrakakis

303 39ccdb18 Christos Stavrakakis
        """
304 39ccdb18 Christos Stavrakakis
305 39ccdb18 Christos Stavrakakis
        input = op.input
306 39ccdb18 Christos Stavrakakis
        op_id = input.OP_ID
307 39ccdb18 Christos Stavrakakis
        network_name = get_field(input, 'network_name')
308 39ccdb18 Christos Stavrakakis
309 39ccdb18 Christos Stavrakakis
        if not network_name:
310 39ccdb18 Christos Stavrakakis
            return None, None
311 39ccdb18 Christos Stavrakakis
312 39ccdb18 Christos Stavrakakis
        self.logger.debug("Job: %d: %s(%s) %s", job_id, op_id,
313 39ccdb18 Christos Stavrakakis
                          network_name, op.status)
314 39ccdb18 Christos Stavrakakis
315 39ccdb18 Christos Stavrakakis
        msg = {'operation':    op_id,
316 39ccdb18 Christos Stavrakakis
               'type':         "ganeti-network-status",
317 39ccdb18 Christos Stavrakakis
               'network':      network_name,
318 39ccdb18 Christos Stavrakakis
               'subnet':       get_field(input, 'network'),
319 39ccdb18 Christos Stavrakakis
               # 'network_mode': get_field(input, 'network_mode'),
320 39ccdb18 Christos Stavrakakis
               # 'network_link': get_field(input, 'network_link'),
321 39ccdb18 Christos Stavrakakis
               'gateway':      get_field(input, 'gateway'),
322 39ccdb18 Christos Stavrakakis
               'group_name':   get_field(input, 'group_name')}
323 39ccdb18 Christos Stavrakakis
324 fd2bdbb2 Christos Stavrakakis
        if op_id == "OP_NETWORK_SET_PARAMS":
325 449787d3 Christos Stavrakakis
            msg["add_reserved_ips"] = get_field(input, "add_reserved_ips")
326 449787d3 Christos Stavrakakis
            msg["remove_reserved_ips"] = get_field(input,
327 449787d3 Christos Stavrakakis
                                                   "remove_reserved_ips")
328 39ccdb18 Christos Stavrakakis
        routekey = "ganeti.%s.event.network" % prefix_from_name(network_name)
329 39ccdb18 Christos Stavrakakis
330 39ccdb18 Christos Stavrakakis
        return msg, routekey
331 39ccdb18 Christos Stavrakakis
332 39ccdb18 Christos Stavrakakis
333 39ccdb18 Christos Stavrakakis
    # def process_group_op(self, op, job_id):
334 39ccdb18 Christos Stavrakakis
    #     """ Process OP_GROUP_* opcodes.
335 39ccdb18 Christos Stavrakakis
336 39ccdb18 Christos Stavrakakis
    #     """
337 39ccdb18 Christos Stavrakakis
    #     return None, None
338 39ccdb18 Christos Stavrakakis
339 39ccdb18 Christos Stavrakakis
340 27a8e4ae Christos Stavrakakis
def find_cluster_name():
341 51fc0054 Christos Stavrakakis
    global handler_logger
342 51fc0054 Christos Stavrakakis
    try:
343 2b4becab Christos Stavrakakis
        ss = SimpleStore()
344 2b4becab Christos Stavrakakis
        name = ss.GetClusterName()
345 51fc0054 Christos Stavrakakis
    except Exception as e:
346 51fc0054 Christos Stavrakakis
        handler_logger.error('Can not get the name of the Cluster: %s' % e)
347 51fc0054 Christos Stavrakakis
        raise e
348 51fc0054 Christos Stavrakakis
349 27a8e4ae Christos Stavrakakis
    return name
350 39ccdb18 Christos Stavrakakis
351 449787d3 Christos Stavrakakis
352 b024b97a Georgios Gousios
handler_logger = None
353 449787d3 Christos Stavrakakis
354 449787d3 Christos Stavrakakis
355 b024b97a Georgios Gousios
def fatal_signal_handler(signum, frame):
356 b024b97a Georgios Gousios
    global handler_logger
357 b024b97a Georgios Gousios
358 b024b97a Georgios Gousios
    handler_logger.info("Caught fatal signal %d, will raise SystemExit",
359 2cd99e7a Georgios Gousios
                        signum)
360 b024b97a Georgios Gousios
    raise SystemExit
361 b024b97a Georgios Gousios
362 449787d3 Christos Stavrakakis
363 b024b97a Georgios Gousios
def parse_arguments(args):
364 b024b97a Georgios Gousios
    from optparse import OptionParser
365 b024b97a Georgios Gousios
366 b024b97a Georgios Gousios
    parser = OptionParser()
367 b024b97a Georgios Gousios
    parser.add_option("-d", "--debug", action="store_true", dest="debug",
368 2cd99e7a Georgios Gousios
                      help="Enable debugging information")
369 b024b97a Georgios Gousios
    parser.add_option("-l", "--log", dest="log_file",
370 45ebfd48 Vangelis Koukis
                      default="/var/log/snf-ganeti-eventd.log",
371 2cd99e7a Georgios Gousios
                      metavar="FILE",
372 2cd99e7a Georgios Gousios
                      help="Write log to FILE instead of %s" %
373 449787d3 Christos Stavrakakis
                           "/var/log/snf-ganeti-eventd.log")
374 b024b97a Georgios Gousios
    parser.add_option('--pid-file', dest="pid_file",
375 45ebfd48 Vangelis Koukis
                      default="/var/run/snf-ganeti-eventd.pid",
376 2cd99e7a Georgios Gousios
                      metavar='PIDFILE',
377 2cd99e7a Georgios Gousios
                      help="Save PID to file (default: %s)" %
378 449787d3 Christos Stavrakakis
                           "/var/run/snf-ganeti-eventd.pid")
379 b024b97a Georgios Gousios
380 b024b97a Georgios Gousios
    return parser.parse_args(args)
381 b024b97a Georgios Gousios
382 449787d3 Christos Stavrakakis
383 b024b97a Georgios Gousios
def main():
384 b024b97a Georgios Gousios
    global handler_logger
385 b024b97a Georgios Gousios
386 b024b97a Georgios Gousios
    (opts, args) = parse_arguments(sys.argv[1:])
387 b024b97a Georgios Gousios
388 b024b97a Georgios Gousios
    # Initialize logger
389 b024b97a Georgios Gousios
    lvl = logging.DEBUG if opts.debug else logging.INFO
390 03a4b970 Georgios Gousios
    logger = logging.getLogger("ganeti.eventd")
391 b024b97a Georgios Gousios
    logger.setLevel(lvl)
392 36cf1973 Vangelis Koukis
    formatter = logging.Formatter(
393 36cf1973 Vangelis Koukis
        "%(asctime)s %(module)s[%(process)d] %(levelname)s: %(message)s",
394 36cf1973 Vangelis Koukis
        "%Y-%m-%d %H:%M:%S")
395 b024b97a Georgios Gousios
    handler = logging.FileHandler(opts.log_file)
396 b024b97a Georgios Gousios
    handler.setFormatter(formatter)
397 b024b97a Georgios Gousios
    logger.addHandler(handler)
398 b024b97a Georgios Gousios
    handler_logger = logger
399 b024b97a Georgios Gousios
400 1dc821c9 Christos Stavrakakis
    # Rename this process so 'ps' output looks like this is a native
401 1dc821c9 Christos Stavrakakis
    # executable.  Can not seperate command-line arguments from actual name of
402 1dc821c9 Christos Stavrakakis
    # the executable by NUL bytes, so only show the name of the executable
403 1dc821c9 Christos Stavrakakis
    # instead.  setproctitle.setproctitle("\x00".join(sys.argv))
404 1dc821c9 Christos Stavrakakis
    setproctitle.setproctitle(sys.argv[0])
405 1dc821c9 Christos Stavrakakis
406 cf2a3529 Christos Stavrakakis
    # Create pidfile
407 cf2a3529 Christos Stavrakakis
    pidf = daemon.pidlockfile.TimeoutPIDLockFile(opts.pid_file, 10)
408 cf2a3529 Christos Stavrakakis
409 cf2a3529 Christos Stavrakakis
    # Remove any stale PID files, left behind by previous invocations
410 cf2a3529 Christos Stavrakakis
    if daemon.runner.is_pidfile_stale(pidf):
411 cf2a3529 Christos Stavrakakis
        logger.warning("Removing stale PID lock file %s", pidf.path)
412 cf2a3529 Christos Stavrakakis
        pidf.break_lock()
413 cf2a3529 Christos Stavrakakis
414 348f53de Georgios Gousios
    # Become a daemon:
415 348f53de Georgios Gousios
    # Redirect stdout and stderr to handler.stream to catch
416 348f53de Georgios Gousios
    # early errors in the daemonization process [e.g., pidfile creation]
417 348f53de Georgios Gousios
    # which will otherwise go to /dev/null.
418 348f53de Georgios Gousios
    daemon_context = daemon.DaemonContext(
419 449787d3 Christos Stavrakakis
        pidfile=pidf,
420 449787d3 Christos Stavrakakis
        umask=022,
421 449787d3 Christos Stavrakakis
        stdout=handler.stream,
422 449787d3 Christos Stavrakakis
        stderr=handler.stream,
423 449787d3 Christos Stavrakakis
        files_preserve=[handler.stream])
424 cf2a3529 Christos Stavrakakis
    try:
425 cf2a3529 Christos Stavrakakis
        daemon_context.open()
426 cf2a3529 Christos Stavrakakis
    except (daemon.pidlockfile.AlreadyLocked, LockTimeout):
427 cf2a3529 Christos Stavrakakis
        logger.critical("Failed to lock pidfile %s, another instance running?",
428 cf2a3529 Christos Stavrakakis
                        pidf.path)
429 cf2a3529 Christos Stavrakakis
        sys.exit(1)
430 cf2a3529 Christos Stavrakakis
431 348f53de Georgios Gousios
    logger.info("Became a daemon")
432 348f53de Georgios Gousios
433 348f53de Georgios Gousios
    # Catch signals to ensure graceful shutdown
434 348f53de Georgios Gousios
    signal(SIGINT, fatal_signal_handler)
435 348f53de Georgios Gousios
    signal(SIGTERM, fatal_signal_handler)
436 348f53de Georgios Gousios
437 b024b97a Georgios Gousios
    # Monitor the Ganeti job queue, create and push notifications
438 b024b97a Georgios Gousios
    wm = pyinotify.WatchManager()
439 449787d3 Christos Stavrakakis
    mask = (pyinotify.EventsCodes.ALL_FLAGS["IN_MOVED_TO"] |
440 449787d3 Christos Stavrakakis
            pyinotify.EventsCodes.ALL_FLAGS["IN_CLOSE_WRITE"])
441 51fc0054 Christos Stavrakakis
442 27a8e4ae Christos Stavrakakis
    cluster_name = find_cluster_name()
443 51fc0054 Christos Stavrakakis
444 27a8e4ae Christos Stavrakakis
    handler = JobFileHandler(logger, cluster_name)
445 b024b97a Georgios Gousios
    notifier = pyinotify.Notifier(wm, handler)
446 b024b97a Georgios Gousios
447 b024b97a Georgios Gousios
    try:
448 b024b97a Georgios Gousios
        # Fail if adding the inotify() watch fails for any reason
449 2b4becab Christos Stavrakakis
        res = wm.add_watch(pathutils.QUEUE_DIR, mask)
450 2b4becab Christos Stavrakakis
        if res[pathutils.QUEUE_DIR] < 0:
451 36cf1973 Vangelis Koukis
            raise Exception("pyinotify add_watch returned negative descriptor")
452 348f53de Georgios Gousios
453 2b4becab Christos Stavrakakis
        logger.info("Now watching %s of %s" % (pathutils.QUEUE_DIR,
454 2b4becab Christos Stavrakakis
                    cluster_name))
455 b024b97a Georgios Gousios
456 b024b97a Georgios Gousios
        while True:    # loop forever
457 348f53de Georgios Gousios
            # process the queue of events as explained above
458 b024b97a Georgios Gousios
            notifier.process_events()
459 b024b97a Georgios Gousios
            if notifier.check_events():
460 b024b97a Georgios Gousios
                # read notified events and enqeue them
461 b024b97a Georgios Gousios
                notifier.read_events()
462 b024b97a Georgios Gousios
    except SystemExit:
463 b024b97a Georgios Gousios
        logger.info("SystemExit")
464 b024b97a Georgios Gousios
    except:
465 b024b97a Georgios Gousios
        logger.exception("Caught exception, terminating")
466 b024b97a Georgios Gousios
    finally:
467 b024b97a Georgios Gousios
        # destroy the inotify's instance on this interrupt (stop monitoring)
468 b024b97a Georgios Gousios
        notifier.stop()
469 b024b97a Georgios Gousios
        raise
470 b024b97a Georgios Gousios
471 b024b97a Georgios Gousios
if __name__ == "__main__":
472 b024b97a Georgios Gousios
    sys.exit(main())
473 b024b97a Georgios Gousios
474 348f53de Georgios Gousios
# vim: set sta sts=4 shiftwidth=4 sw=4 et ai :