Skip to content

Commit 04b30e0

Browse files
pyryadvr
pyr
authored and
yadvr
committed
Add Apache Kafka event-bus support - producing only.
This commit produces event bus messages to a "cloudstack" topic in Apache Kafka. Configuration is expected to be found in /etc/cloudstack/management/kafka.producer.properties and will generally be of the form: bootstrap.servers=kafka-host1:9092,kafka-host2:9092 key.serializer=org.apache.kafka.common.serialization.StringSerializer value.serializer=org.apache.kafka.common.serialization.StringSerializer There is no way to parameterize the topic yet, and the consuming code is just place-holder. I think adding a consumer within cloudstack is very debatable and likely not needed. Signed-off-by: Rohit Yadav <[email protected]>
1 parent 05d2b0a commit 04b30e0

4 files changed

Lines changed: 160 additions & 0 deletions

File tree

‎client/pom.xml‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -241,6 +241,11 @@
241241
<artifactId>cloud-mom-inmemory</artifactId>
242242
<version>${project.version}</version>
243243
</dependency>
244+
<dependency>
245+
<groupId>org.apache.cloudstack</groupId>
246+
<artifactId>cloud-mom-kafka</artifactId>
247+
<version>${project.version}</version>
248+
</dependency>
244249
<dependency>
245250
<groupId>org.apache.cloudstack</groupId>
246251
<artifactId>cloud-framework-ipc</artifactId>

‎plugins/event-bus/kafka/pom.xml‎

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,45 @@
1+
<!--
2+
Licensed to the Apache Software Foundation (ASF) under one
3+
or more contributor license agreements. See the NOTICE file
4+
distributed with this work for additional information
5+
regarding copyright ownership. The ASF licenses this file
6+
to you under the Apache License, Version 2.0 (the
7+
"License"); you may not use this file except in compliance
8+
with the License. You may obtain a copy of the License at
9+
10+
http://www.apache.org/licenses/LICENSE-2.0
11+
12+
Unless required by applicable law or agreed to in writing,
13+
software distributed under the License is distributed on an
14+
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
KIND, either express or implied. See the License for the
16+
specific language governing permissions and limitations
17+
under the License.
18+
-->
19+
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
20+
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
21+
<modelVersion>4.0.0</modelVersion>
22+
<artifactId>cloud-mom-kafka</artifactId>
23+
<name>Apache CloudStack Plugin - Kafka Event Bus</name>
24+
<parent>
25+
<groupId>org.apache.cloudstack</groupId>
26+
<artifactId>cloudstack-plugins</artifactId>
27+
<version>4.4.2</version>
28+
<relativePath>../../pom.xml</relativePath>
29+
</parent>
30+
<dependencies>
31+
<dependency>
32+
<groupId>org.apache.cloudstack</groupId>
33+
<artifactId>cloud-framework-events</artifactId>
34+
<version>${project.version}</version>
35+
</dependency>
36+
<dependency>
37+
<groupId>org.apache.kafka</groupId>
38+
<artifactId>kafka-clients</artifactId>
39+
<version>0.8.2.0</version>
40+
</dependency>
41+
</dependencies>
42+
<build>
43+
<defaultGoal>install</defaultGoal>
44+
</build>
45+
</project>

‎plugins/event-bus/kafka/src/org/apache/cloudstack/mom/kafka/KafkaEventBus.java‎

Lines changed: 109 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,109 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
package org.apache.cloudstack.mom.kafka;
21+
22+
import java.io.FileInputStream;
23+
24+
import java.util.Map;
25+
import java.util.UUID;
26+
import java.util.Properties;
27+
28+
import javax.ejb.Local;
29+
import javax.naming.ConfigurationException;
30+
31+
import org.apache.log4j.Logger;
32+
33+
import org.apache.cloudstack.framework.events.Event;
34+
import org.apache.cloudstack.framework.events.EventBus;
35+
import org.apache.cloudstack.framework.events.EventBusException;
36+
import org.apache.cloudstack.framework.events.EventSubscriber;
37+
import org.apache.cloudstack.framework.events.EventTopic;
38+
39+
import com.cloud.utils.component.ManagerBase;
40+
41+
import org.apache.kafka.clients.producer.Producer;
42+
import org.apache.kafka.clients.producer.KafkaProducer;
43+
import org.apache.kafka.clients.producer.ProducerRecord;
44+
45+
import com.cloud.utils.PropertiesUtil;
46+
47+
@Local(value = EventBus.class)
48+
public class KafkaEventBus extends ManagerBase implements EventBus {
49+
50+
private final String _topic = "cloudstack";
51+
private Producer<String,String> _producer;
52+
private static final Logger s_logger = Logger.getLogger(KafkaEventBus.class);
53+
54+
@Override
55+
public boolean configure(String name, Map<String, Object> params) throws ConfigurationException {
56+
57+
final Properties props = new Properties();
58+
59+
try {
60+
final FileInputStream is = new FileInputStream(PropertiesUtil.findConfigFile("kafka.producer.properties"));
61+
props.load(is);
62+
is.close();
63+
} catch (Exception e) {
64+
throw new ConfigurationException("Could not read kafka properties");
65+
}
66+
67+
_producer = new KafkaProducer<String,String>(props);
68+
_name = name;
69+
70+
return true;
71+
}
72+
73+
@Override
74+
public void setName(String name) {
75+
_name = name;
76+
}
77+
78+
@Override
79+
public UUID subscribe(EventTopic topic, EventSubscriber subscriber) throws EventBusException {
80+
/* NOOP */
81+
return UUID.randomUUID();
82+
}
83+
84+
@Override
85+
public void unsubscribe(UUID subscriberId, EventSubscriber subscriber) throws EventBusException {
86+
/* NOOP */
87+
}
88+
89+
@Override
90+
public void publish(Event event) throws EventBusException {
91+
ProducerRecord<String, String> record = new ProducerRecord<String,String>(_topic, event.getResourceUUID(), event.getDescription());
92+
_producer.send(record);
93+
}
94+
95+
@Override
96+
public String getName() {
97+
return _name;
98+
}
99+
100+
@Override
101+
public boolean start() {
102+
return true;
103+
}
104+
105+
@Override
106+
public boolean stop() {
107+
return true;
108+
}
109+
}

‎plugins/pom.xml‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -54,6 +54,7 @@
5454
<module>hypervisors/kvm</module>
5555
<module>event-bus/rabbitmq</module>
5656
<module>event-bus/inmemory</module>
57+
<module>event-bus/kafka</module>
5758
<module>hypervisors/baremetal</module>
5859
<module>hypervisors/ucs</module>
5960
<module>hypervisors/hyperv</module>

0 commit comments

Comments
 (0)