Adding Kafka message broker and Zipkin stream service for kafka

Author: pranav-patil

analytics-service/build.gradle

@@ -40,6 +40,9 @@ dependencies {
 	compile('org.springframework.cloud:spring-cloud-starter-netflix-eureka-client')
 	compile('org.springframework.cloud:spring-cloud-starter-netflix-hystrix')
 	compile('org.springframework.cloud:spring-cloud-starter-config')
+	compile('org.springframework.cloud:spring-cloud-starter-zipkin')
+	compile('org.springframework.cloud:spring-cloud-sleuth-zipkin')
+	compile('org.springframework.kafka:spring-kafka')
 	compile('org.springframework.security.oauth:spring-security-oauth2')
 	compile('com.google.code.gson:gson:2.8.4')
 	runtime('com.h2database:h2')

config-service/src/main/resources/shared/analytics-service.yml

@@ -1,3 +1,11 @@
+spring:
+  zipkin:
+    baseUrl: http://localhost:9411/
+    sender:
+      type: kafka
+  sleuth:
+    sampler:
+      probability: 1.0
 
 server:
   port: 8309
@@ -41,3 +49,8 @@ management:
     web:
       exposure:
         include: hystrix.stream,info,health
+
+sample:
+  zipkin:
+    # When enabled=false, traces log to the console. When enabled=true traces send to zipkin
+    enabled: true

config-service/src/main/resources/shared/finance-service.yml

@@ -1,7 +1,11 @@
 spring:
-  cloud:
-    refresh:
-      enabled: false
+  zipkin:
+    baseUrl: http://localhost:9411/
+    sender:
+      type: kafka
+  sleuth:
+    sampler:
+      probability: 1.0
 
 server:
   port: 8301
@@ -47,3 +51,8 @@ management:
     web:
       exposure:
         include: hystrix.stream,info,health
+
+sample:
+  zipkin:
+    # When enabled=false, traces log to the console. When enabled=true traces send to zipkin
+    enabled: true

finance-service/build.gradle

@@ -30,14 +30,16 @@ ext {
 
 dependencies {
 	compile('org.springframework.boot:spring-boot-starter-actuator')
-	compile('org.springframework.boot:spring-boot-starter-data-jpa')
 	compile('org.springframework.boot:spring-boot-starter-data-rest')
 	compile('org.springframework.boot:spring-boot-starter-security')
 	compile('org.springframework.cloud:spring-cloud-starter-config')
 	compile('org.springframework.cloud:spring-cloud-starter-oauth2')
 	compile('org.springframework.cloud:spring-cloud-starter-openfeign')
 	compile('org.springframework.cloud:spring-cloud-starter-netflix-eureka-client')
 	compile('org.springframework.cloud:spring-cloud-starter-netflix-hystrix')
+	compile('org.springframework.cloud:spring-cloud-starter-zipkin')
+	compile('org.springframework.cloud:spring-cloud-sleuth-zipkin')
+	compile('org.springframework.kafka:spring-kafka')
 	compile('org.springframework.security.oauth:spring-security-oauth2')
 	compile('com.google.code.gson:gson:2.8.4')
 	runtime('com.h2database:h2')

finance-service/src/main/java/com/emprovise/service/api/StockResource.java

@@ -1,5 +1,6 @@
 package com.emprovise.service.api;
 
+import brave.Tracer;
 import com.emprovise.service.config.SSLUtil;
 import com.emprovise.service.dto.StockDetailDTO;
 import com.emprovise.service.mapper.StockDetailDTOMapper;
@@ -23,6 +24,8 @@ public class StockResource {
     private StockDetailDTOMapper stockDTOMapper;
     @Value("${alphavantage.apikey}")
     private String apiKey;
+    @Autowired
+    private Tracer tracer;
 
     private static final String ALPHA_VANTAGE_URL = "https://www.alphavantage.co/query";
     private static final String TIME_SERIES_INTRADAY = "TIME_SERIES_INTRADAY";
@@ -51,7 +54,7 @@ public StockDetailDTO getStockDetailsFallback(String symbol, String interval) th
 
     @GetMapping("/greeting")
     public String greeting() {
+        tracer.currentSpan().tag("service.greeting", "hello_world");
         return "Hello World";
     }
-
 }

kafka-broker/README.md

@@ -0,0 +1,67 @@
+Kafka Message Broker
+=============
+
+Apache Kafka is a fast and scalable messaging queue, capable of handling heavy read and write operations.
+Apache Kafka requires a running Zookeeper instance, which is used for reliable distributed co-ordination.
+Apache ZooKeeper is a distributed, open-source coordination service for distributed applications. 
+ZooKeeper enables kafka to manage multiple clusters of kafka instances. 
+
+### Installation and Running of Apache Zookeeper
+
+* Download latest [Apache Zookeeper release](http://zookeeper.apache.org/releases.html#download) and extract the .tar.gz file. 
+* Copy and Rename **zoo_sample.cfg** to **zoo.cfg** file in ZOOKEEPER_HOME\conf directory, were ZOOKEEPER_HOME is the path of extracted zookeeper directory.
+* Find & edit the line in zoo.cfg file from "dataDir=/tmp/zookeeper" to "dataDir=ZOOKEEPER_HOME\data" 
+* Go to ZOOKEEPER_HOME\bin directory and execute the command "zkserver" to run zookeeper.
+* Apache Zookeeper runs on default port 2181.
+
+All the above steps can be executed directly by running below runZooKeeper gradle task for windows.
+
+    $ gradle runZooKeeper
+
+### Installation and Running of Apache Kafka message broker
+
+Before running kafka ensure that Apache Zookeeper instance is already running on default port 2181.
+
+* Download latest stable [Apache Kafka release](https://kafka.apache.org/downloads) and extract the .tgz file.
+* Go to config directory in Apache Kafka in path KAFKA_HOME\config were KAFKA_HOME is the path of extracted kafka directory.
+* Find "log.dirs" in **server.properties** in config directory, and replace line line "log.dirs=/tmp/kafka-logs" to "log.dir= KAFKA_HOME\kafka-logs".
+*  If Zookeeper is running on different machine or cluster, edit "zookeeper.connect:2181" to the custom IP and port.
+* Go to KAFKA_HOME directory and execute the below windows command to run kafka.
+* Apache Kafka runs on default port 9092.
+
+
+    .\bin\windows\kafka-server-start.bat .\config\server.properties
+ 
+All the above steps can be executed directly by running below runKafka gradle task for windows.
+
+    $ gradle runKafka
+
+### Creating topics
+
+Kafka topics can be created using the below windows command. Although **zipin** should already exist by default.
+
+    kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic zipin
+
+Alternatively **zipin** topic can be created using below createZipkinTopic gradle task.
+
+    $ gradle createZipkinTopic
+
+### Consuming topics
+
+To produce messages to Kafka topic use the below windows command.
+
+    kafka-console-producer.bat --broker-list localhost:9092 --topic zipin
+
+To consume the messages in Kafka topic use the below windows command.
+
+    kafka-console-consumer.bat --zookeeper localhost:2181 --topic zipin
+
+Alternatively messages from **zipin** topic can viewed by running consumeZipkinTopic gradle task.
+
+    $ gradle consumeZipkinTopic
+
+### Notes
+
+* Microsoft Windows XP or later have  the maximum length of command to 8191 characters. If it exceeds we get an error **The input line is too long. The syntax of the command is incorrect.**.
+Apache Kafka command has many classpath dependencies hence, running this in a nested directory could cause above error. To avoid this please run Apache Kafka in a directory within c: root seperately or clone this project in c: root itself.
+* On running the bat commands for zookeeper and kafka we might get unrelated errors like **\IBM\WebSphere was unexpected at this time.**. Not sure about the reason but this can be avoided by removing corresponding path from the environment variables.    

kafka-broker/build.gradle

@@ -0,0 +1,95 @@
+
+description = "Kafka Message Broker"
+group = 'com.emprovise.service'
+version = '0.0.1-SNAPSHOT'
+
+buildscript {
+	ext {
+		zookeeperVersion = '3.4.12'
+		scalaVersion = '2.11'
+		kafkaVersion = '1.1.0'
+	}
+	repositories {
+		jcenter()
+	}
+	dependencies {
+		classpath("de.undercouch:gradle-download-task:3.4.3")
+	}
+}
+
+apply plugin: 'de.undercouch.download'
+def pullCommonDir = new File(buildDir, 'repo')
+pullCommonDir.mkdirs()
+
+task downloadFiles(type: Download) {
+	src([
+		"http://apache.mirrors.ionfish.org/zookeeper/zookeeper-${zookeeperVersion}/zookeeper-${zookeeperVersion}.tar.gz",
+		"http://apache.mirrors.ionfish.org/kafka/${kafkaVersion}/kafka_${scalaVersion}-${kafkaVersion}.tgz"
+	])
+
+	dest "${buildDir}/repo"
+	overwrite false
+}
+
+task extractArchives(dependsOn: downloadFiles) {
+	doLast {
+		copy {
+			into new File(buildDir, "libs")
+			pullCommonDir.eachFileRecurse {
+				if (it.path.endsWith(".tar.gz") || it.path.endsWith(".tgz")) {
+					from tarTree(it)
+				}
+			}
+		}
+	}
+}
+
+task copyAndRename(dependsOn: extractArchives) {
+	doLast {
+		if (!file("${buildDir}/libs/zookeeper-${zookeeperVersion}/conf/zoo.cfg").exists()) {
+			println "copying zoo.cfg file..."
+			def buildDirPath = "${buildDir}".replace("\\", "/")
+
+			copy {
+				from "${buildDir}/libs/zookeeper-${zookeeperVersion}/conf/zoo_sample.cfg"
+				into "${buildDir}/libs/zookeeper-${zookeeperVersion}/conf"
+				filter { line -> line.replaceAll("dataDir=/tmp/zookeeper", "dataDir=${buildDirPath}/libs/zookeeper-${zookeeperVersion}/data") }
+				rename { String fileName ->
+					fileName.replace("zoo_sample.cfg", "zoo.cfg")
+				}
+			}
+		}
+
+		def kafkaConfigFile = new File("${buildDir}/libs/kafka_${scalaVersion}-${kafkaVersion}/config/server.properties")
+
+		if (kafkaConfigFile.exists()) {
+			println "modifying kafka config file..."
+			def fileText = kafkaConfigFile.text
+			kafkaConfigFile.withWriter { writer ->
+				writer.print fileText.replace("log.dirs=/tmp/kafka-logs", "log.dirs=${buildDir}\\libs\\kafka_${scalaVersion}-${kafkaVersion}\\kafka-logs")
+			}
+		}
+	}
+}
+
+def KAFKA_HOME = "${buildDir}\\libs\\kafka_${scalaVersion}-${kafkaVersion}"
+
+task runZooKeeper(type: Exec, dependsOn: copyAndRename) {
+	workingDir "${buildDir}/libs/zookeeper-${zookeeperVersion}/bin"
+	commandLine 'cmd', '/c', 'zkServer'
+}
+
+task runKafka(type: Exec, dependsOn: copyAndRename) {
+	workingDir "${KAFKA_HOME}\\bin\\windows"
+	commandLine 'cmd', '/c', "kafka-server-start.bat ..\\..\\config\\server.properties"
+}
+
+task createZipkinTopic(type: Exec) {
+	workingDir "${KAFKA_HOME}\\bin\\windows"
+	commandLine 'cmd', '/c', "kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic zipkin"
+}
+
+task consumeZipkinTopic(type: Exec) {
+	workingDir "${KAFKA_HOME}\\bin\\windows"
+	commandLine 'cmd', '/c', "kafka-console-consumer.bat --zookeeper localhost:2181 --topic zipkin"
+}

settings.gradle

@@ -6,6 +6,7 @@ include "config-service"
 include "discovery-service"
 include "finance-service"
 include "gateway-service"
-include "sqldb-service"
+include "kafka-broker"
 include "monitor-service"
-
+include "sqldb-service"
+include "zipkin-service"

zipkin-service/README.md

@@ -0,0 +1,38 @@
+Zipkin Service
+=============
+
+[OpenZipkin](https://github.com/openzipkin/zipkin) is a distributed tracing system which enables to trace requests spanning across multiple micro services. 
+Distributed tracing is a process of collecting end-to-end transaction graphs in real time. 
+Zipkin collects the request data in form of traces and spans. A trace represents entire journey of a request, while a span represents single operation call.
+The first service called has traceId and spanId, while the subsequent service call have additional parentId including traceId and spanId, which essentially is spanId of the caller service.  
+
+[Spring Cloud Sleuth](http://cloud.spring.io/spring-cloud-sleuth/single/spring-cloud-sleuth.html) is a tracer which adds unique trace and span ids to Slf4J MDC. 
+It has an instrumentation or sampling policy represented by application property spring.sleuth.sampler.probability.
+
+### Running the Zipkin default application (using http for trace requests)
+
+In order to run the default [zipkin-server](https://github.com/openzipkin/zipkin/tree/master/zipkin-server) use the gradle runZipkin task
+
+    $ gradle runZipkin
+
+The runZipkin downloads the [zipkin-server-executable jar](https://search.maven.org/remote_content?g=io.zipkin.java&a=zipkin-server&v=LATEST&c=exec) from maven repository.
+It then executes the jar using default java command. Instead of using the above gradle task, you can directly execute the below command.  
+
+    $ java -jar zipkin-server-2.10.1-exec.jar
+
+### Running the Zipkin Kafka application (using kafka message broker for trace messages)
+
+Before running zipkin with kafka message broker, go to kafka-broker service to run kafka message broker.
+By default the kafka message broker executes on port 9092. Then use the below runZipkinKafka gradle task to run zipkin stream server configured with default kafka message broker. 
+
+    $ gradle runZipkinKafka
+    
+Instead of using the above gradle task, you can directly execute the below java command with the additional kafka config argument [zipkin.collector.kafka.bootstrap-servers](https://github.com/openzipkin/zipkin/tree/master/zipkin-autoconfigure/collector-kafka).
+
+    $ java -jar zipkin-server-2.10.1-exec.jar -Dzipkin.collector.kafka.bootstrap-servers=127.0.0.1:9092
+
+### Accessing Zipin Server
+
+To view the zipin traces for micro services using zipin's brave UI, access the server with default port 9411.   
+
+http://localhost:9411/

zipkin-service/build.gradle

@@ -0,0 +1,36 @@
+
+description = "Zipkin Stream Server"
+group = 'com.emprovise.service'
+version = '0.0.1-SNAPSHOT'
+sourceCompatibility = 1.8
+
+buildscript {
+	ext {
+		zipkinVersion = '2.10.1'
+	}
+	repositories {
+		jcenter()
+	}
+	dependencies {
+		classpath("de.undercouch:gradle-download-task:3.4.3")
+	}
+}
+
+apply plugin: 'de.undercouch.download'
+
+task downloadFile(type: Download) {
+	src "https://search.maven.org/remotecontent?filepath=io/zipkin/java/zipkin-server/$zipkinVersion/zipkin-server-$zipkinVersion-exec.jar"
+	dest "$buildDir/runtime/libs/zipkin-server-$zipkinVersion-exec.jar"
+	overwrite false
+}
+
+task runZipkin(type: JavaExec, dependsOn: downloadFile) {
+	main = "-jar"
+	args = [ "$buildDir/runtime/libs/zipkin-server-$zipkinVersion-exec.jar" ]
+}
+
+task runZipkinKafka(type: JavaExec, dependsOn: downloadFile) {
+	main = "-jar"
+	jvmArgs = [ "-Dzipkin.collector.kafka.bootstrap-servers=127.0.0.1:9092" ]
+	args = [ "$buildDir/runtime/libs/zipkin-server-$zipkinVersion-exec.jar" ]
+}