first commit

.gitignore

@@ -0,0 +1,30 @@
+*.class
+*.log
+
+# sbt specific
+.cache
+.history
+.lib/
+dist/*
+target/
+lib_managed/
+src_managed/
+project/boot/
+project/plugins/project/
+
+# Scala-IDE specific
+.scala_dependencies
+.worksheet
+
+# Other
+.java-version
+scm-source.json
+.ensime
+.ensime_cache/
+.idea/
+__pycache__/
+*.pyc
+
+# Stups
+user.json
+client.json

CONTRIBUTING.md

@@ -0,0 +1,23 @@
+# Contributions
+
+We are happy to accept contributions! This is what makes Open Source possible. We ask
+that you follow these guidelines when contributing.
+
+## Issues
+Found a bug or have an idea for a feature? You can help us by [creating an issue](https://github.com/zalando-incubator/remora/issues).
+Just make sure that an issue describing the bug or feature does not already exist. You can go even further and
+[open a Pull Request](https://github.com/zalando-incubator/remora/pulls) with the fix or proposed feature.
+
+## Pull Requests
+Follow these steps to contribute your work to API Discovery:
+
+1. [Open an issue](https://github.com/zalando-incubator/remora/issues) describing the problem or proposed feature. Assign yourself so we know you're working on it.
+1. Fork this repo and create a branch for your work.
+1. Push changes to your branch.
+1. Test your changes.
+1. Open a [Pull Request](https://github.com/zalando-incubator/remora/pulls) when your code is ready for review.
+    1. Mention the issue number in the comment (e.g. Fixes #37).
+    1. If you're still working on it, add the **under development** label.
+1. We will review your PR, give feedback, and merge when it is ready.
+
+Thanks for your contribution!

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2017 The Zalando Incubator
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE

MAINTAINERS

@@ -0,0 +1,2 @@
+Mark Kelly <[email protected]>
+Adam Drakeford <[email protected]>

README.md

@@ -0,0 +1,111 @@
+# Remora
+A simple API, Built using akka and scala, in front of kafka to give offset information. 
+
+## Background
+
+Due to using akka streams we could not get the current offset from the api. Instead we used linkedin [burrow](https://github.com/linkedin/Burrow)
+but contains some [performance issues](https://github.com/linkedin/Burrow/wiki/Known-Issues)
+
+All we did was wrap and hack the [kafka consumer group command](https://github.com/apache/kafka/blob/0.10.0/core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala) in akka http.
+
+## Prerequisites
+
+* Kafka 0.10.0.1
+* 2.11.8
+* Store offsets in kafka
+
+## Build
+
+* `sbt package`
+* `sbt docker:publishLocal`
+* `sbt docker:publish -Ddocker.repo=<DOCKER REPO>`
+
+## Server Arguments
+
+* SERVER_PORT - default `9000`
+* KAFKA_ENDPOINT - default `localhost:9092`
+
+## Show Active Consumers
+`http://localhost:9000/consumers`
+
+which gives 
+
+```json
+[
+    {
+      "protocolType": "consumer",
+      "groupId": "consumer-1"
+    }
+    {
+      "protocolType": "consumer",
+      "groupId": "consumer-2"
+    }
+    {
+      "protocolType": "consumer",
+      "groupId": "consumer-3"
+    }
+]
+```
+
+## Show specific consumer info
+`curl http://localhost:9000/consumers/<ConsumerGroupId>`
+
+```json
+[
+    {
+      "owner": "consumer-2_/132.34.134.12",
+      "lag": 155758,
+      "log_end_offset": 2580124,
+      "offset": 2424366,
+      "partition": 1,
+      "topic": "foobar",
+      "group": "consumer-1"
+    }
+    {
+      "owner": "consumer-2_/132.34.134.12",
+      "lag": 155758,
+      "log_end_offset": 2580124,
+      "offset": 2424366,
+      "partition": 2,
+      "topic": "foobar",
+      "group": "consumer-1"
+    }
+    {
+      "owner": "consumer-2_/132.34.134.12",
+      "lag": 155758,
+      "log_end_offset": 2580124,
+      "offset": 2424366,
+      "partition": 3,
+      "topic": "foobar",
+      "group": "consumer-1"
+    }
+]
+```
+
+## Health
+
+`curl http://localhost:9000/health` returns `OK`
+
+## Metrics
+
+`curl http://localhost:9000/metrics`
+
+### Contributing
+
+We are happy to accept contributions. First, take a look at our [contributing guidelines](CONTRIBUTING.md).
+
+You can see our current status in [this task board](https://github.com/zalando-incubator/remora/projects/1).
+
+
+### TODO
+
+Please check the [Issues Page](https://github.com/zalando-incubator/remora/issues)
+for contribution ideas.
+
+### Contact
+
+Feel free to contact one of the [maintainers](MAINTAINERS).
+
+### License
+
+MIT

build.sbt

@@ -0,0 +1,49 @@
+lazy val commonSettings = Seq(
+  name         := "remora",
+  organization := "de.zalando",
+  scalaVersion := "2.11.8"
+)
+
+lazy val dockerSettings = Seq(
+  daemonUser in Docker           := "root",
+  dockerBaseImage in Docker      := "registry.opensource.zalan.do/stups/openjdk:8u91-b14-1-22",
+  dockerExposedPorts in Docker   := Seq(9000),
+  dockerExposedVolumes in Docker := Seq("/opt/docker/logs"),
+  dockerRepository in Docker     := sys.props.get("docker.repo"),
+  maintainer in Docker           := "[email protected]"
+)
+
+lazy val gitSettings = Seq(
+  git.useGitDescribe := true
+)
+
+lazy val root = (project in file("."))
+  .settings(commonSettings)
+  .settings(dockerSettings)
+  .settings(gitSettings)
+  .enablePlugins(
+    GitVersioning,
+    JavaServerAppPackaging
+  )
+
+libraryDependencies ++= Seq(
+  "ch.qos.logback"            %  "logback-classic"              % "1.1.8",
+  "io.dropwizard.metrics"     %  "metrics-core"                 % "3.1.2",
+  "io.dropwizard.metrics"     %  "metrics-jvm"                  % "3.1.2",
+  "io.dropwizard.metrics"     %  "metrics-json"                 % "3.1.2",
+  "com.typesafe.akka"         %% "akka-actor"                   % "2.4.16",
+  "com.typesafe.akka"         %% "akka-http"                    % "10.0.5",
+  "com.typesafe.akka"         %% "akka-slf4j"                   % "2.4.16",
+  "org.apache.httpcomponents" %  "httpcore"                     % "4.4.5",
+  "org.apache.httpcomponents" %  "httpclient"                   % "4.5.2",
+  "org.scalaz"                %% "scalaz-core"                  % "7.2.8",
+  "com.typesafe.play"         %% "play-json"                    % "2.4.8",
+  "org.apache.kafka"          %% "kafka"                        % "0.10.0.1",
+  "org.scalatest"             %% "scalatest"                    % "2.2.4"   % "test",
+  "org.scalamock"             %% "scalamock-scalatest-support"  % "3.2"     % "test"
+)
+
+assemblyMergeStrategy in assembly := {
+  case PathList("org", "apache", "commons", "logging", xs @ _*) => MergeStrategy.first
+  case default => (assemblyMergeStrategy in assembly).value(default)
+}

docker-compose/consume.py

@@ -0,0 +1,18 @@
+import sys
+from kafka import KafkaConsumer
+import logging as log
+
+log.basicConfig(level=log.DEBUG)
+
+def consume(varargs):
+    ip =varargs[0]
+    print(ip+":9092")
+    consumer = KafkaConsumer(group_id='test-0-consumer', bootstrap_servers=ip+":9092")
+    consumer.subscribe(['test-0'])
+
+    for message in consumer:
+        print (message)
+
+
+if __name__ == "__main__":
+    print(consume(sys.argv[1:]))

docker-compose/docker-compose.yml

@@ -0,0 +1,36 @@
+version: '2'
+services:
+  zookeeper:
+    image: wurstmeister/zookeeper
+
+  kafka:
+    image: wurstmeister/kafka:0.10.0.1
+    hostname: kafka
+    environment:
+      KAFKA_ADVERTISED_HOST_NAME: kafka
+      KAFKA_ADVERTISED_PORT: 9092
+      KAFKA_CREATE_TOPICS: "test-0:1:1,test-1:1:1"
+      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
+    ports:
+      - "9092:9092"
+      - "2181:2181"
+
+  remora:
+#    build: ./target/docker/.
+    image: pierone.stups.zalan.do/setanta/remora:0.0.6-2-ge24db20
+    depends_on:
+      - kafka
+    hostname: remora
+    environment:
+      KAFKA_ENDPOINT: "kafka:9092"
+#      JAVA_OPTS:               >
+#        -Xmx1g
+#        -Dcom.sun.management.jmxremote.rmi.port=9090
+#        -Dcom.sun.management.jmxremote=true
+#        -Dcom.sun.management.jmxremote.port=9090
+#        -Dcom.sun.management.jmxremote.ssl=false
+#        -Dcom.sun.management.jmxremote.authenticate=false
+#        -Dcom.sun.management.jmxremote.local.only=false
+#        -Djava.rmi.server.hostname=localhost
+    ports:
+      - "9000:9000"

docker-compose/produce.py

@@ -0,0 +1,11 @@
+import sys
+from kafka import KafkProducer
+
+def produce(varargs):
+    ip =varargs[0]
+    producer = KafkProducer(bootstrap_servers=ip+":9092")
+    for _ in range(10):
+        producer.send('test-0', b'some_message_bytes')
+
+if __name__ == "__main__":
+    print(produce(sys.argv[1:]))

docker-compose/test.py

@@ -0,0 +1,26 @@
+#!/usr/bin/python
+
+import sys, requests
+
+
+def check():
+  data={}
+
+  consumers = requests.get('http://localhost:9000/consumers').json()
+
+  for consumer in consumers:
+
+      consumerInfos=requests.get('http://localhost:9000/consumers/'+consumer['groupId']).json()
+
+      for consumerInfo in consumerInfos:
+        data['{consumer_group}-{topic}-{partition}-lag'.format(consumer_group=consumer['groupId'],topic=consumerInfo['topic'],partition=consumerInfo['partition'])]=consumerInfo['lag']
+        data['{consumer_group}-{topic}-{partition}-log_end_offset'.format(consumer_group=consumer['groupId'],topic=consumerInfo['topic'],partition=consumerInfo['partition'])]=consumerInfo['log_end_offset']
+        data['{consumer_group}-{topic}-{partition}-offset'.format(consumer_group=consumer['groupId'],topic=consumerInfo['topic'],partition=consumerInfo['partition'])]=consumerInfo['offset']
+
+      print(data)
+
+  return data
+
+
+if __name__ == "__main__":
+   check()

project/build.properties

@@ -0,0 +1 @@
+sbt.version = 0.13.8

project/plugins.sbt

@@ -0,0 +1,3 @@
+addSbtPlugin("com.eed3si9n"     % "sbt-assembly"        % "0.14.3")
+addSbtPlugin("com.typesafe.sbt" % "sbt-git"             % "0.8.5")
+addSbtPlugin("com.typesafe.sbt" % "sbt-native-packager" % "1.1.5")

src/main/resources/application.conf

@@ -0,0 +1,35 @@
+akka {
+  loggers = ["akka.event.slf4j.Slf4jLogger"]
+  logging-filter = "akka.event.slf4j.Slf4jLoggingFilter"
+  loglevel = "DEBUG"
+  stdout-loglevel = "OFF"
+  log-dead-letters = off
+}
+
+api {
+  port = 9000
+  port = ${?SERVER_PORT}
+}
+
+kafka {
+  endpoint = "localhost:9092"
+  endpoint = ${?KAFKA_ENDPOINT}
+}
+
+kafka-consumer-dispatcher {
+  type = Dispatcher
+  executor = "fork-join-executor"
+  fork-join-executor {
+    parallelism-min = 2
+    parallelism-factor = 3.0
+    parallelism-max = 9
+  }
+}
+
+api-dispatcher {
+  type = Dispatcher
+  executor = "thread-pool-executor"
+  thread-pool-executor {
+    fixed-pool-size = 9
+  }
+}