Siddhi with Kerberized Kafka Source

README.md

Kerberized Kafka with Siddhi

• krb5-user package with krb.conf
• crt in JKS
• valid keytab for renewing token

carbon.sh

#!/bin/sh
# ---------------------------------------------------------------------------
#  Copyright (c) 2015, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
#
#  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.

# ----------------------------------------------------------------------------
# Main Script for the WSO2 Carbon Server
#
# Environment Variable Prequisites
#
#   CARBON_HOME   Home of WSO2 Carbon installation. If not set I will  try
#                   to figure it out.
#   RUNTIME_HOME  Home of WSO2 Carbon Runtime. .
#
#   JAVA_HOME       Must point at your Java Development Kit installation.
#
#   JAVA_OPTS       (Optional) Java runtime options used when the commands
#                   is executed.
#
# NOTE: Borrowed generously from Apache Tomcat startup scripts.
# -----------------------------------------------------------------------------

# OS specific support.  $var _must_ be set to either true or false.
#ulimit -n 100000

cygwin=false;
darwin=false;
os400=false;
mingw=false;
case "`uname`" in
CYGWIN*) cygwin=true;;
MINGW*) mingw=true;;
OS400*) os400=true;;
Darwin*) darwin=true
        if [ -z "$JAVA_VERSION" ] ; then
             JAVA_VERSION="CurrentJDK"
           else
             echo "Using Java version: $JAVA_VERSION"
           fi
           if [ -z "$JAVA_HOME" ] ; then
             JAVA_HOME=/System/Library/Frameworks/JavaVM.framework/Versions/${JAVA_VERSION}/Home
           fi
           ;;
esac

# resolve links - $0 may be a softlink
PRG="$0"

while [ -h "$PRG" ]; do
  ls=`ls -ld "$PRG"`
  link=`expr "$ls" : '.*-> \(.*\)$'`
  if expr "$link" : '.*/.*' > /dev/null; then
    PRG="$link"
  else
    PRG=`dirname "$PRG"`/"$link"
  fi
done

# Get standard environment variables
TEMPCURDIR=`dirname "$PRG"`

# Only set RUNTIME_HOME if not already set
[ -z "$RUNTIME_HOME" ] && RUNTIME_HOME=`cd "$TEMPCURDIR/.." ; pwd`

# Only set CARBON_HOME if not already set
[ -z "$CARBON_HOME" ] && CARBON_HOME=`cd "$TEMPCURDIR/../../../" ; pwd`

# Only set RUNTIME if not already set
[ -z "$RUNTIME" ] && RUNTIME=${RUNTIME_HOME##*/}

# For Cygwin, ensure paths are in UNIX format before anything is touched
if $cygwin; then
  [ -n "$JAVA_HOME" ] && JAVA_HOME=`cygpath --unix "$JAVA_HOME"`
  [ -n "$CARBON_HOME" ] && CARBON_HOME=`cygpath --unix "$CARBON_HOME"`
  [ -n "$RUNTIME_HOME" ] && RUNTIME_HOME=`cygpath --unix "$RUNTIME_HOME"`
fi

# For OS400
if $os400; then
  # Set job priority to standard for interactive (interactive - 6) by using
  # the interactive priority - 6, the helper threads that respond to requests
  # will be running at the same priority as interactive jobs.
  COMMAND='chgjob job('$JOBNAME') runpty(6)'
  system $COMMAND

  # Enable multi threading
  QIBM_MULTI_THREADED=Y
  export QIBM_MULTI_THREADED
fi

# For Migwn, ensure paths are in UNIX format before anything is touched
if $mingw ; then
  [ -n "$CARBON_HOME" ] &&
    CARBON_HOME="`(cd "$CARBON_HOME"; pwd)`"
  [ -n "$JAVA_HOME" ] &&
    JAVA_HOME="`(cd "$JAVA_HOME"; pwd)`"
  # TODO classpath?
fi

if [ -z "$JAVACMD" ] ; then
  if [ -n "$JAVA_HOME"  ] ; then
    if [ -x "$JAVA_HOME/jre/sh/java" ] ; then
      # IBM's JDK on AIX uses strange locations for the executables
      JAVACMD="$JAVA_HOME/jre/sh/java"
    else
      JAVACMD="$JAVA_HOME/bin/java"
    fi
  else
    JAVACMD=java
  fi
fi

if [ ! -x "$JAVACMD" ] ; then
  echo "Error: JAVA_HOME is not defined correctly."
  echo " CARBON cannot execute $JAVACMD"
  exit 1
fi

# if JAVA_HOME is not set we're not happy
if [ -z "$JAVA_HOME" ]; then
  echo "You must set the JAVA_HOME variable before running CARBON."
  exit 1
fi

if [ -e "$RUNTIME_HOME/runtime.pid" ]; then
  PID=`cat "$RUNTIME_HOME"/runtime.pid`
fi

# ----- Process the input command ----------------------------------------------
args=""
for c in $*
do
    if [ "$c" = "--debug" ] || [ "$c" = "-debug" ] || [ "$c" = "debug" ]; then
          CMD="--debug"
          continue
    elif [ "$CMD" = "--debug" ]; then
          if [ -z "$PORT" ]; then
                PORT=$c
          fi
    elif [ "$c" = "--stop" ] || [ "$c" = "-stop" ] || [ "$c" = "stop" ]; then
          CMD="stop"
    elif [ "$c" = "--start" ] || [ "$c" = "-start" ] || [ "$c" = "start" ]; then
          CMD="start"
    elif [ "$c" = "--version" ] || [ "$c" = "-version" ] || [ "$c" = "version" ]; then
          CMD="version"
    elif [ "$c" = "--restart" ] || [ "$c" = "-restart" ] || [ "$c" = "restart" ]; then
          CMD="restart"
    elif [ "$c" = "--test" ] || [ "$c" = "-test" ] || [ "$c" = "test" ]; then
          CMD="test"
    else
        args="$args $c"
    fi
done

if [ "$CMD" = "--debug" ]; then
  if [ "$PORT" = "" ]; then
    echo " Please specify the debug port after the --debug option"
    exit 1
  fi
  if [ -n "$JAVA_OPTS" ]; then
    echo "Warning !!!. User specified JAVA_OPTS will be ignored, once you give the --debug option."
  fi
  CMD="RUN"
  JAVA_OPTS="-Xdebug -Xnoagent -Djava.compiler=NONE -Xrunjdwp:transport=dt_socket,server=y,suspend=y,address=$PORT"
  echo "Please start the remote debugging client to continue..."
elif [ "$CMD" = "start" ]; then
  if [ -e "$RUNTIME_HOME/runtime.pid" ]; then
    if  ps -p $PID > /dev/null ; then
      echo "Process is already running"
      exit 0
    fi
  fi
  export CARBON_HOME=$CARBON_HOME
# using nohup bash to avoid erros in solaris OS.TODO
  nohup bash $RUNTIME_HOME/bin/carbon.sh $args > /dev/null 2>&1 &
  exit 0
elif [ "$CMD" = "stop" ]; then
  export CARBON_HOME=$CARBON_HOME
  kill -term `cat $RUNTIME_HOME/runtime.pid`
  exit 0
elif [ "$CMD" = "restart" ]; then
  export CARBON_HOME=$CARBON_HOME
  kill -term `cat $RUNTIME_HOME/runtime.pid`
  process_status=0
  pid=`cat $RUNTIME_HOME/runtime.pid`
  while [ "$process_status" -eq "0" ]
  do
        sleep 1;
        ps -p$pid 2>&1 > /dev/null
        process_status=$?
  done

# using nohup bash to avoid erros in solaris OS.TODO
  nohup bash $RUNTIME_HOME/bin/carbon.sh $args > /dev/null 2>&1 &
  exit 0
elif [ "$CMD" = "test" ]; then
    JAVACMD="exec "$JAVACMD""
elif [ "$CMD" = "version" ]; then
  cat $CARBON_HOME/bin/kernel-version.txt
  exit 0
fi

# ---------- Handle the SSL Issue with proper JDK version --------------------
jdk_18=`$JAVA_HOME/bin/java -version 2>&1 | grep "1.[8]"`
if [ "$jdk_18" = "" ]; then
   echo " Starting WSO2 Carbon (in unsupported JDK)"
   echo " [ERROR] CARBON is supported only on JDK 1.8"
fi

CARBON_XBOOTCLASSPATH=""
for f in "$CARBON_HOME"/bin/bootstrap/xboot/*.jar
do
    if [ "$f" != "$CARBON_HOME/bin/bootstrap/xboot/*.jar" ];then
        CARBON_XBOOTCLASSPATH="$CARBON_XBOOTCLASSPATH":$f
    fi
done

JAVA_ENDORSED_DIRS="$CARBON_HOME/bin/bootstrap/endorsed":"$JAVA_HOME/jre/lib/endorsed":"$JAVA_HOME/lib/endorsed"

CARBON_CLASSPATH=""
if [ -e "$JAVA_HOME/bin/bootstrap/tools.jar" ]; then
    CARBON_CLASSPATH="$JAVA_HOME/lib/tools.jar"
fi
for f in "$CARBON_HOME"/bin/bootstrap/*.jar
do
    if [ "$f" != "$CARBON_HOME/bin/bootstrap/*.jar" ];then
        CARBON_CLASSPATH="$CARBON_CLASSPATH":$f
    fi
done
for t in "$CARBON_HOME"/bin/bootstrap/commons-lang*.jar
do
    CARBON_CLASSPATH="$CARBON_CLASSPATH":$t
done
# For Cygwin, switch paths to Windows format before running java
if $cygwin; then
  JAVA_HOME=`cygpath --absolute --windows "$JAVA_HOME"`
  CARBON_HOME=`cygpath --absolute --windows "$CARBON_HOME"`
  RUNTIME_HOME=`cygpath --absolute --windows "$RUNTIME_HOME"`
  CLASSPATH=`cygpath --path --windows "$CLASSPATH"`
  JAVA_ENDORSED_DIRS=`cygpath --path --windows "$JAVA_ENDORSED_DIRS"`
  CARBON_CLASSPATH=`cygpath --path --windows "$CARBON_CLASSPATH"`
  CARBON_XBOOTCLASSPATH=`cygpath --path --windows "$CARBON_XBOOTCLASSPATH"`
fi

# ----- Execute The Requested Command -----------------------------------------

echo JAVA_HOME environment variable is set to $JAVA_HOME
echo CARBON_HOME environment variable is set to $CARBON_HOME
echo RUNTIME_HOME environment variable is set to $RUNTIME_HOME

cd "$RUNTIME_HOME"

START_EXIT_STATUS=121
status=$START_EXIT_STATUS

#To monitor a Carbon server in remote JMX mode on linux host machines, set the below system property.
#   -Djava.rmi.server.hostname="your.IP.goes.here"

while [ "$status" = "$START_EXIT_STATUS" ]
do
    $JAVACMD \
    -Xbootclasspath/a:"$CARBON_XBOOTCLASSPATH" \
    -Xms256m -Xmx1024m \
    -XX:+HeapDumpOnOutOfMemoryError \
    -XX:HeapDumpPath="$RUNTIME_HOME/logs/heap-dump.hprof" \
    $JAVA_OPTS \
    -classpath "$CARBON_CLASSPATH" \
    -Djava.endorsed.dirs="$JAVA_ENDORSED_DIRS" \
    -Djava.io.tmpdir="$CARBON_HOME/tmp" \
    -Dcarbon.registry.root=/ \
    -Djava.command="$JAVACMD" \
    -Dcarbon.home="$CARBON_HOME" \
    -Dwso2.runtime.path="$RUNTIME_HOME" \
    -Dwso2.runtime="$RUNTIME" \
    -Djava.util.logging.config.file="$RUNTIME_HOME/bin/bootstrap/logging.properties" \
    -Djava.security.egd=file:/dev/./urandom \
    -Dfile.encoding=UTF8 \
    -Djavax.net.ssl.keyStore="$CARBON_HOME/resources/security/wso2carbon.jks" \
    -Djavax.net.ssl.keyStorePassword="wso2carbon" \
    -Djavax.net.ssl.trustStore="$CARBON_HOME/resources/security/client-truststore.jks" \
    -Djavax.net.ssl.trustStorePassword="wso2carbon" \
    -Djava.security.auth.login.config="/etc/kafka/secrets/consumer_jaas.conf" \
    -Djava.security.krb5.conf="/etc/kafka/secrets/krb.conf" \
    org.wso2.carbon.launcher.Main $*
    status=$?
done

consumer.config

ssl.truststore.location=/etc/kafka/secrets/kafka.producer.truststore.jks
ssl.truststore.password=confluent

ssl.keystore.location=/etc/kafka/secrets/kafka.producer.keystore.jks
ssl.keystore.password=confluent
ssl.key.password=confluent

security.protocol=SASL_SSL
sasl.mechanism=GSSAPI
sasl.kerberos.service.name=kafka

consumer_jaas.conf

KafkaClient {
    com.sun.security.auth.module.Krb5LoginModule required
    useKeyTab=true
    storeKey=true
    keyTab="/etc/kafka/secrets/saslconsumer.keytab"
    principal="saslconsumer/quickstart.confluent.io@TEST.CONFLUENT.IO";
};

krb.conf

[logging]
 default = FILE:/var/log/kerberos/krb5libs.log
 kdc = FILE:/var/log/kerberos/krb5kdc.log
 admin_server = FILE:/var/log/kerberos/kadmind.log

[libdefaults]
 default_realm = TEST.CONFLUENT.IO
 dns_lookup_realm = false
 dns_lookup_kdc = false
 ticket_lifetime = 24h
 renew_lifetime = 7d
 forwardable = true

[realms]
 TEST.CONFLUENT.IO = {
  kdc = quickstart.confluent.io
  admin_server = quickstart.confluent.io
 }

[domain_realm]
 .TEST.CONFLUENT.IO = TEST.CONFLUENT.IO
 TEST.CONFLUENT.IO = TEST.CONFLUENT.IO

logging.properties

############################################################
# Default Logging Configuration File For Wso2 Carbon
#
############################################################

############################################################
# Global properties
############################################################

############################################################
# "handlers" specifies a comma separated list of log Handler
# classes.  These handlers will be installed during VM startup.
# Note that these classes must be on the system classpath.
# By default we only configure a ConsoleHandler and a
# FileLogHandler which will only show messages at
# the INFO and above levels.
handlers= java.util.logging.FileHandler, java.util.logging.ConsoleHandler

# Default global logging level.
# This specifies which kinds of events are logged across
# all loggers.  For any given facility this global level
# can be overriden by a facility specific level
# Note that the ConsoleHandler also has a separate level
# setting to limit messages printed to the console.
.level= DEBUG

############################################################
# Handler specific properties.
# Describes specific configuration info for Handlers.
############################################################

# default file output is relative to in user's home directory.
java.util.logging.FileHandler.level = DEBUG
java.util.logging.FileHandler.pattern = logs/carbon.log
java.util.logging.FileHandler.limit = 50000
java.util.logging.FileHandler.count = 1
java.util.logging.FileHandler.formatter = java.util.logging.SimpleFormatter
java.util.logging.SimpleFormatter.format = [%1$tY-%1$tm-%1$td %1$tk:%1$tM:%1$tS,%1$tL]  %4$s {%2$s} - %5$s %6$s %n

# Limit the message that are printed on the console to INFO and above.
java.util.logging.ConsoleHandler.level = DEBUG
java.util.logging.ConsoleHandler.formatter = java.util.logging.SimpleFormatter

# Example to customize the SimpleFormatter output format
# to print one-line log message like this:
#     <level>: <log message> [<date/time>]
#
# java.util.logging.SimpleFormatter.format=%4$s: %5$s [%1$tc]%n

############################################################
# Facility specific properties.
# Provides extra control for each logger.
############################################################

# For example, set the com.xyz.foo logger to only log SEVERE
# messages:
com.xyz.foo.level = SEVERE
org.apache.kafka = DEUBG

SiddhiApp.siddhi

@App:name("SiddhiApp")
@App:description("Description of the plan")

-- Please refer to https://docs.wso2.com/display/SP400/Quick+Start+Guide on getting started with SP editor. 


@source(type='kafka',
        topic.list='bar',
        partition.no.list='0',
        threading.option='single.thread',
        group.id="ssl-asl-host",
        bootstrap.servers='quickstart.confluent.io:29094,quickstart.confluent.io:39094,quickstart.confluent.io:49094',
        optional.configuration='ssl.truststore.location:/etc/kafka/secrets/kafka.consumer.truststore.jks,ssl.truststore.password:confluent,ssl.keystore.location:/etc/kafka/secrets/kafka.consumer.keystore.jks,ssl.keystore.password:confluent,ssl.key.password:confluent,security.protocol:SASL_SSL,sasl.mechanism:GSSAPI,sasl.kerberos.service.name:kafka',
      @map(type='json'))
      
define stream SweetProductionStream ( amount string);