OpenStack Telemetry (Ceilometer)
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

base.py 11KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342
  1. #
  2. # Copyright 2013 Intel Corp.
  3. # Copyright 2014 Red Hat, Inc
  4. #
  5. # Licensed under the Apache License, Version 2.0 (the "License"); you may
  6. # not use this file except in compliance with the License. You may obtain
  7. # a copy of the License at
  8. #
  9. # http://www.apache.org/licenses/LICENSE-2.0
  10. #
  11. # Unless required by applicable law or agreed to in writing, software
  12. # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  13. # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  14. # License for the specific language governing permissions and limitations
  15. # under the License.
  16. import abc
  17. from oslo_config import cfg
  18. from oslo_log import log
  19. import oslo_messaging
  20. import six
  21. from ceilometer import agent
  22. from ceilometer import publisher
  23. OPTS = [
  24. cfg.StrOpt('pipeline_cfg_file',
  25. default="pipeline.yaml",
  26. help="Configuration file for pipeline definition."
  27. ),
  28. cfg.StrOpt('event_pipeline_cfg_file',
  29. default="event_pipeline.yaml",
  30. deprecated_for_removal=True,
  31. help="Configuration file for event pipeline definition."
  32. ),
  33. ]
  34. LOG = log.getLogger(__name__)
  35. class PipelineException(agent.ConfigException):
  36. def __init__(self, message, cfg):
  37. super(PipelineException, self).__init__('Pipeline', message, cfg)
  38. class PublishContext(object):
  39. def __init__(self, pipelines):
  40. self.pipelines = pipelines or []
  41. def __enter__(self):
  42. def p(data):
  43. for p in self.pipelines:
  44. p.publish_data(data)
  45. return p
  46. def __exit__(self, exc_type, exc_value, traceback):
  47. for p in self.pipelines:
  48. p.flush()
  49. class PipelineSource(agent.Source):
  50. """Represents a source of samples or events."""
  51. def __init__(self, cfg):
  52. try:
  53. super(PipelineSource, self).__init__(cfg)
  54. except agent.SourceException as err:
  55. raise PipelineException(err.msg, cfg)
  56. try:
  57. self.sinks = cfg['sinks']
  58. except KeyError as err:
  59. raise PipelineException(
  60. "Required field %s not specified" % err.args[0], cfg)
  61. def check_sinks(self, sinks):
  62. if not self.sinks:
  63. raise PipelineException(
  64. "No sink defined in source %s" % self,
  65. self.cfg)
  66. for sink in self.sinks:
  67. if sink not in sinks:
  68. raise PipelineException(
  69. "Dangling sink %s from source %s" % (sink, self),
  70. self.cfg)
  71. class Sink(object):
  72. """Represents a sink for the transformation and publication of data.
  73. Each sink config is concerned *only* with the transformation rules
  74. and publication conduits for data.
  75. In effect, a sink describes a chain of handlers. The chain ends with one or
  76. more publishers.
  77. At the end of the chain, publishers publish the data. The exact
  78. publishing method depends on publisher type, for example, pushing
  79. into data storage via the message bus providing guaranteed delivery,
  80. or for loss-tolerant data UDP may be used.
  81. """
  82. def __init__(self, conf, cfg, publisher_manager):
  83. self.conf = conf
  84. self.cfg = cfg
  85. try:
  86. self.name = cfg['name']
  87. except KeyError as err:
  88. raise PipelineException(
  89. "Required field %s not specified" % err.args[0], cfg)
  90. if not cfg.get('publishers'):
  91. raise PipelineException("No publisher specified", cfg)
  92. self.publishers = []
  93. for p in cfg['publishers']:
  94. if '://' not in p:
  95. # Support old format without URL
  96. p = p + "://"
  97. try:
  98. self.publishers.append(publisher_manager.get(p))
  99. except Exception:
  100. LOG.error("Unable to load publisher %s", p,
  101. exc_info=True)
  102. self.multi_publish = True if len(self.publishers) > 1 else False
  103. def __str__(self):
  104. return self.name
  105. @staticmethod
  106. def flush():
  107. """Flush data after all events have been injected to pipeline."""
  108. @six.add_metaclass(abc.ABCMeta)
  109. class Pipeline(object):
  110. """Represents a coupling between a sink and a corresponding source."""
  111. def __init__(self, conf, source, sink):
  112. self.conf = conf
  113. self.source = source
  114. self.sink = sink
  115. self.name = str(self)
  116. def __str__(self):
  117. return (self.source.name if self.source.name == self.sink.name
  118. else '%s:%s' % (self.source.name, self.sink.name))
  119. def flush(self):
  120. self.sink.flush()
  121. @property
  122. def publishers(self):
  123. return self.sink.publishers
  124. @abc.abstractmethod
  125. def publish_data(self, data):
  126. """Publish data from pipeline."""
  127. @abc.abstractmethod
  128. def supported(self, data):
  129. """Attribute to filter on. Pass if no partitioning."""
  130. class PublisherManager(object):
  131. def __init__(self, conf, purpose):
  132. self._loaded_publishers = {}
  133. self._conf = conf
  134. self._purpose = purpose
  135. def get(self, url):
  136. if url not in self._loaded_publishers:
  137. p = publisher.get_publisher(
  138. self._conf, url,
  139. 'ceilometer.%s.publisher' % self._purpose)
  140. self._loaded_publishers[url] = p
  141. return self._loaded_publishers[url]
  142. class PipelineManager(agent.ConfigManagerBase):
  143. """Pipeline Manager
  144. Pipeline manager sets up pipelines according to config file
  145. """
  146. def __init__(self, conf, cfg_file):
  147. """Setup the pipelines according to config.
  148. The configuration is supported as follows:
  149. Decoupled: the source and sink configuration are separately
  150. specified before being linked together. This allows source-
  151. specific configuration, such as meter handling, to be
  152. kept focused only on the fine-grained source while avoiding
  153. the necessity for wide duplication of sink-related config.
  154. The configuration is provided in the form of separate lists
  155. of dictionaries defining sources and sinks, for example:
  156. {"sources": [{"name": source_1,
  157. "meters" : ["meter_1", "meter_2"],
  158. "sinks" : ["sink_1", "sink_2"]
  159. },
  160. {"name": source_2,
  161. "meters" : ["meter_3"],
  162. "sinks" : ["sink_2"]
  163. },
  164. ],
  165. "sinks": [{"name": sink_1,
  166. "publishers": ["publisher_1", "publisher_2"]
  167. },
  168. {"name": sink_2,
  169. "publishers": ["publisher_3"]
  170. },
  171. ]
  172. }
  173. Valid meter format is '*', '!meter_name', or 'meter_name'.
  174. '*' is wildcard symbol means any meters; '!meter_name' means
  175. "meter_name" will be excluded; 'meter_name' means 'meter_name'
  176. will be included.
  177. Valid meters definition is all "included meter names", all
  178. "excluded meter names", wildcard and "excluded meter names", or
  179. only wildcard.
  180. Publisher's name is plugin name in setup.cfg
  181. """
  182. super(PipelineManager, self).__init__(conf)
  183. cfg = self.load_config(cfg_file)
  184. self.pipelines = []
  185. if not ('sources' in cfg and 'sinks' in cfg):
  186. raise PipelineException("Both sources & sinks are required",
  187. cfg)
  188. publisher_manager = PublisherManager(self.conf, self.pm_type)
  189. unique_names = set()
  190. sources = []
  191. for s in cfg.get('sources'):
  192. name = s.get('name')
  193. if name in unique_names:
  194. raise PipelineException("Duplicated source names: %s" %
  195. name, self)
  196. else:
  197. unique_names.add(name)
  198. sources.append(self.pm_source(s))
  199. unique_names.clear()
  200. sinks = {}
  201. for s in cfg.get('sinks'):
  202. name = s.get('name')
  203. if name in unique_names:
  204. raise PipelineException("Duplicated sink names: %s" %
  205. name, self)
  206. else:
  207. unique_names.add(name)
  208. sinks[s['name']] = self.pm_sink(self.conf, s,
  209. publisher_manager)
  210. unique_names.clear()
  211. for source in sources:
  212. source.check_sinks(sinks)
  213. for target in source.sinks:
  214. pipe = self.pm_pipeline(self.conf, source, sinks[target])
  215. if pipe.name in unique_names:
  216. raise PipelineException(
  217. "Duplicate pipeline name: %s. Ensure pipeline"
  218. " names are unique. (name is the source and sink"
  219. " names combined)" % pipe.name, cfg)
  220. else:
  221. unique_names.add(pipe.name)
  222. self.pipelines.append(pipe)
  223. unique_names.clear()
  224. @abc.abstractproperty
  225. def pm_type(self):
  226. """Pipeline manager type."""
  227. @abc.abstractproperty
  228. def pm_pipeline(self):
  229. """Pipeline class"""
  230. @abc.abstractproperty
  231. def pm_source(self):
  232. """Pipeline source class"""
  233. @abc.abstractproperty
  234. def pm_sink(self):
  235. """Pipeline sink class"""
  236. def publisher(self):
  237. """Build publisher for pipeline publishing."""
  238. return PublishContext(self.pipelines)
  239. def get_main_endpoints(self):
  240. """Return endpoints for main queue."""
  241. pass
  242. class NotificationEndpoint(object):
  243. """Base Endpoint for plugins that support the notification API."""
  244. event_types = []
  245. """List of strings to filter messages on."""
  246. def __init__(self, conf, publisher):
  247. super(NotificationEndpoint, self).__init__()
  248. # NOTE(gordc): this is filter rule used by oslo.messaging to dispatch
  249. # messages to an endpoint.
  250. if self.event_types:
  251. self.filter_rule = oslo_messaging.NotificationFilter(
  252. event_type='|'.join(self.event_types))
  253. self.conf = conf
  254. self.publisher = publisher
  255. @abc.abstractmethod
  256. def process_notifications(self, priority, notifications):
  257. """Return a sequence of Counter instances for the given message.
  258. :param message: Message to process.
  259. """
  260. @classmethod
  261. def _consume_and_drop(cls, notifications):
  262. """RPC endpoint for useless notification level"""
  263. # NOTE(sileht): nothing special todo here, but because we listen
  264. # for the generic notification exchange we have to consume all its
  265. # queues
  266. audit = _consume_and_drop
  267. critical = _consume_and_drop
  268. debug = _consume_and_drop
  269. error = _consume_and_drop
  270. info = _consume_and_drop
  271. sample = _consume_and_drop
  272. warn = _consume_and_drop