1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
|
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2010 United States Government as represented by the
# Administrator of the National Aeronautics and Space Administration.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
"""
Generic Node baseclass for all workers that run on hosts
"""
import inspect
import os
import sys
import time
from eventlet import event
from eventlet import greenthread
from eventlet import greenpool
from sqlalchemy.exc import OperationalError
from nova import context
from nova import db
from nova import exception
from nova import log as logging
from nova import flags
from nova import rpc
from nova import utils
from nova import version
FLAGS = flags.FLAGS
flags.DEFINE_integer('report_interval', 10,
'seconds between nodes reporting state to datastore',
lower_bound=1)
flags.DEFINE_integer('periodic_interval', 60,
'seconds between running periodic tasks',
lower_bound=1)
flags.DEFINE_string('pidfile', None,
'pidfile to use for this service')
flags.DEFINE_flag(flags.HelpFlag())
flags.DEFINE_flag(flags.HelpshortFlag())
flags.DEFINE_flag(flags.HelpXMLFlag())
class Service(object):
"""Base class for workers that run on hosts."""
def __init__(self, host, binary, topic, manager, report_interval=None,
periodic_interval=None, *args, **kwargs):
self.host = host
self.binary = binary
self.topic = topic
self.manager_class_name = manager
self.report_interval = report_interval
self.periodic_interval = periodic_interval
super(Service, self).__init__(*args, **kwargs)
self.saved_args, self.saved_kwargs = args, kwargs
self.timers = []
def start(self):
manager_class = utils.import_class(self.manager_class_name)
self.manager = manager_class(host=self.host, *self.saved_args,
**self.saved_kwargs)
self.manager.init_host()
self.model_disconnected = False
ctxt = context.get_admin_context()
try:
service_ref = db.service_get_by_args(ctxt,
self.host,
self.binary)
self.service_id = service_ref['id']
except exception.NotFound:
self._create_service_ref(ctxt)
conn1 = rpc.Connection.instance(new=True)
conn2 = rpc.Connection.instance(new=True)
if self.report_interval:
consumer_all = rpc.AdapterConsumer(
connection=conn1,
topic=self.topic,
proxy=self)
consumer_node = rpc.AdapterConsumer(
connection=conn2,
topic='%s.%s' % (self.topic, self.host),
proxy=self)
self.timers.append(consumer_all.attach_to_eventlet())
self.timers.append(consumer_node.attach_to_eventlet())
pulse = utils.LoopingCall(self.report_state)
pulse.start(interval=self.report_interval, now=False)
self.timers.append(pulse)
if self.periodic_interval:
periodic = utils.LoopingCall(self.periodic_tasks)
periodic.start(interval=self.periodic_interval, now=False)
self.timers.append(periodic)
def _create_service_ref(self, context):
zone = FLAGS.node_availability_zone
service_ref = db.service_create(context,
{'host': self.host,
'binary': self.binary,
'topic': self.topic,
'report_count': 0,
'availability_zone': zone})
self.service_id = service_ref['id']
def __getattr__(self, key):
manager = self.__dict__.get('manager', None)
return getattr(manager, key)
@classmethod
def create(cls,
host=None,
binary=None,
topic=None,
manager=None,
report_interval=None,
periodic_interval=None):
"""Instantiates class and passes back application object.
Args:
host, defaults to FLAGS.host
binary, defaults to basename of executable
topic, defaults to bin_name - "nova-" part
manager, defaults to FLAGS.<topic>_manager
report_interval, defaults to FLAGS.report_interval
periodic_interval, defaults to FLAGS.periodic_interval
"""
if not host:
host = FLAGS.host
if not binary:
binary = os.path.basename(inspect.stack()[-1][1])
if not topic:
topic = binary.rpartition("nova-")[2]
if not manager:
manager = FLAGS.get('%s_manager' % topic, None)
if not report_interval:
report_interval = FLAGS.report_interval
if not periodic_interval:
periodic_interval = FLAGS.periodic_interval
vcs_string = version.version_string_with_vcs()
logging.audit(_("Starting %(topic)s node (version %(vcs_string)s)")
% locals())
service_obj = cls(host, binary, topic, manager,
report_interval, periodic_interval)
return service_obj
def kill(self):
"""Destroy the service object in the datastore"""
self.stop()
try:
db.service_destroy(context.get_admin_context(), self.service_id)
except exception.NotFound:
logging.warn(_("Service killed that has no database entry"))
def stop(self):
for x in self.timers:
try:
x.stop()
except Exception:
pass
self.timers = []
def periodic_tasks(self):
"""Tasks to be run at a periodic interval"""
self.manager.periodic_tasks(context.get_admin_context())
def report_state(self):
"""Update the state of this service in the datastore."""
ctxt = context.get_admin_context()
try:
try:
service_ref = db.service_get(ctxt, self.service_id)
except exception.NotFound:
logging.debug(_("The service database object disappeared, "
"Recreating it."))
self._create_service_ref(ctxt)
service_ref = db.service_get(ctxt, self.service_id)
db.service_update(ctxt,
self.service_id,
{'report_count': service_ref['report_count'] + 1})
# TODO(termie): make this pattern be more elegant.
if getattr(self, "model_disconnected", False):
self.model_disconnected = False
logging.error(_("Recovered model server connection!"))
# TODO(vish): this should probably only catch connection errors
except Exception: # pylint: disable-msg=W0702
if not getattr(self, "model_disconnected", False):
self.model_disconnected = True
logging.exception(_("model server went away"))
def serve(*services):
FLAGS(sys.argv)
logging.basicConfig()
if not services:
services = [Service.create()]
name = '_'.join(x.binary for x in services)
logging.debug(_("Serving %s"), name)
logging.debug(_("Full set of FLAGS:"))
for flag in FLAGS:
flag_get = FLAGS.get(flag, None)
logging.debug("%(flag)s : %(flag_get)s" % locals())
for x in services:
x.start()
def wait():
while True:
greenthread.sleep(5)
|