initial integration of kafka

echo-stomp/echo-stomp.gradle → echo-kafka/echo-kafka.gradle

@@ -1,26 +1,38 @@
 /*
  * Copyright 2014 Netflix, Inc.
  *
- * Licensed under the Apache License, Version 2.0 (the 'License');
+ * 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,
+ * 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.
  */
 
 apply from: "$rootDir/gradle/groovy-module.gradle"
 
+ext {
+    kafkaVersion = '0.8.1.1'
+    scalaVersion = '2.10'
+}
+
 dependencies {
     compile project(':echo-model')
-    compile 'org.springframework.boot:spring-boot-starter-websocket:1.1.7.RELEASE'
-    compile 'org.springframework:spring-messaging:4.0.7.RELEASE'
+    compile spinnaker.dependency('bootWeb')
+    compile commonDependencies.kork
+    compile commonDependencies.jacksonDatabind
     testCompile commonDependencies.spockSpring
     testCompile commonDependencies.springTest
-    testCompile commonDependencies.objenesis
+
+    compile("org.apache.kafka:kafka_$scalaVersion:$kafkaVersion") {
+        exclude module: 'jms'
+        exclude module: 'jmxtools'
+        exclude module: 'jmxri'
+    }
+
 }

echo-kafka/src/main/groovy/com/netflix/spinnaker/echo/config/KafkaProducerConfig.groovy

@@ -0,0 +1,28 @@
+/*
+ * Copyright 2014 Netflix, Inc.
+ *
+ * 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.
+ */
+
+package com.netflix.spinnaker.echo.config
+
+import org.springframework.context.annotation.Configuration
+
+/**
+ * Enables Cassandra for this project
+ */
+@Configuration
+@SuppressWarnings('GStringExpressionWithinString')
+class KafkaProducerConfig {
+
+}

echo-stomp/src/main/groovy/com/netflix/spinnaker/echo/events/StompEventListener.groovy → echo-kafka/src/main/groovy/com/netflix/spinnaker/echo/events/KafkaEventListener.groovy

@@ -16,44 +16,42 @@
 
 package com.netflix.spinnaker.echo.events
 
+import com.fasterxml.jackson.databind.ObjectMapper
 import com.netflix.spinnaker.echo.model.Event
-import org.springframework.beans.factory.annotation.Autowired
-import org.springframework.messaging.simp.SimpMessagingTemplate
+import org.springframework.beans.factory.annotation.Value
 import org.springframework.stereotype.Component
 
 /**
- * Event listener for stomp
+ * Event listener for echo events
  */
 @Component
-class StompEventListener implements EchoEventListener {
+class KafkaEventListener implements EchoEventListener {
 
-    @Autowired
-    SimpMessagingTemplate template
+    ObjectMapper mapper = new ObjectMapper()
+
+    @Value('${kafka.prefix}')
+    String prefix
+
+    @Value('${kafka.hostname}')
+    String hostname
+
+    @Value('${kafka.app}')
+    String app
+
+    @Value('${kafka.rowid}')
+    String rowid
 
     @Override
     void processEvent(Event event) {
 
-        if (event.details.application) {
-            template.convertAndSend(
-                '/topic/application/' + event.details.application,
-                event
-            )
-        }
-
-        template.convertAndSend(
-            '/topic/source/' + event.details.source,
-            event
-        )
-
-        template.convertAndSend(
-            '/topic/type/' + event.details.type,
-            event
-        )
-
-        template.convertAndSend(
-            '/topic/events',
-            event
-        )
+        Map eventAsMap = mapper.convertValue(event, Map)
+
+        eventAsMap."${prefix}_ts" = new Date().time
+        eventAsMap."${prefix}_app" = app
+        eventAsMap."${prefix}_hostname" = hostname
+        eventAsMap."${prefix}_rowid" = rowid
+
+        println eventAsMap
 
     }
 

echo-web/echo-web.gradle

@@ -44,7 +44,7 @@ dependencies {
     compile project(':echo-model')
     compile project(':echo-cassandra')
     compile project(':echo-elasticsearch')
-    // compile project(':echo-stomp')
+    compile project(':echo-kafka')
     compile spinnaker.dependency('bootActuator')
     compile spinnaker.dependency('bootWeb')
     compile commonDependencies.springWeb

echo-web/src/main/resources/application.yml

@@ -9,3 +9,9 @@ search:
   index: 'event_history'
 # server:
   # port: 8120
+
+kafka:
+  prefix: 'suro'
+  hostname: 'echo.prod.netflix.com'
+  app: 'spinnaker'
+  rowid: 'events'

settings.gradle

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-include 'echo-core', 'echo-model', 'echo-cassandra', 'echo-web', 'echo-elasticsearch', 'echo-stomp'
+include 'echo-core', 'echo-model', 'echo-cassandra', 'echo-web', 'echo-elasticsearch', 'echo-kafka'
 
 rootProject.name = 'echo'