blob: 22604ef341c765368f5485d81bc86fbe5290939a [file] [log] [blame]
raghunath dudyala01482a82016-08-01 12:29:16 +05301#
2# Copyright 2015 Cisco Inc.
3#
4# Licensed under the Apache License, Version 2.0 (the "License"); you may
5# not use this file except in compliance with the License. You may obtain
6# a copy of the License at
7#
8# http://www.apache.org/licenses/LICENSE-2.0
9#
10# Unless required by applicable law or agreed to in writing, software
11# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13# License for the specific language governing permissions and limitations
14# under the License.
15
16import json
17
18import kafka
19from oslo_utils import netutils
20from six.moves.urllib import parse as urlparse
21import logging as LOG
22
23
24class KafkaBrokerPublisher():
25 def __init__(self, parsed_url):
26 self.kafka_client = None
27 self.kafka_server = None
28 self.kafka_consumer = None
29
30 self.host, self.port = netutils.parse_host_port(
31 parsed_url.netloc, default_port=9092)
32
33 self.local_queue = []
34
35 params = urlparse.parse_qs(parsed_url.query)
36 self.topic = params.get('topic', ['ceilometer'])[-1]
37 self.policy = params.get('policy', ['default'])[-1]
38 self.max_queue_length = int(params.get(
39 'max_queue_length', [1024])[-1])
40 self.max_retry = int(params.get('max_retry', [100])[-1])
41
42 if self.policy in ['default', 'drop', 'queue']:
43 LOG.info(('Publishing policy set to %s') % self.policy)
44 else:
45 LOG.warn(('Publishing policy is unknown (%s) force to default')
46 % self.policy)
47 self.policy = 'default'
48
49 try:
50 self._get_client()
51 self._get_server()
52 except Exception as e:
53 LOG.exception("Failed to connect to Kafka service: %s", e)
54
55 def _get_client(self):
56 if not self.kafka_client:
57 self.kafka_client = kafka.KafkaClient(
58 "%s:%s" % (self.host, self.port))
59 self.kafka_producer = kafka.SimpleProducer(self.kafka_client)
60
61 def _get_server(self):
62 if not self.kafka_server:
63 self.kafka_server = kafka.KafkaClient(
64 "%s:%s" % (self.host, self.port))
65 self.kafka_consumer = kafka.KafkaConsumer(self.topic,bootstrap_servers = ["%s:%s" % (self.host, self.port)])
66
67
68 def _send(self, data):
69 #for d in data:
70 try:
71 self.kafka_producer.send_messages(
72 self.topic, json.dumps(data))
73 except Exception as e:
74 LOG.exception(("Failed to send sample data: %s"), e)
75 raise