Support connecting to Heroku Kafka

[sdball]

Make changes to Kaffe that will allow declaring connections to an SSL Kafka (i.e. Heroku Kafka)

Blocked by spreedly/dev-services#36

[sdball]

Now that we've got an ssl-test topic on Heroku I'm digging into how to setup brod to talk to SSL Kafka. (And also looking around the landscape of Elixir Kafka clients.)

(time passes)

Ok, brod and the leading Elixir Kafka client (kafka_ex) have the same "problem". They expect the SSL certs to be actual files:

{:cacertfile, "cacert.pem"},
{:certfile, "cert.pem"},
{:keyfile, "cert_key.pem"},

I'm not sure how that will match up with having the certs as ENV values. I hope there's a way to hand over the cert values directly instead of handing over filenames with the cert values.

For comparison here's Ruby-Kafka: https://github.com/spreedly/ruby-kafka/blob/master/heroku-kafka-consumer.rb#L48-L61

brokers = ENV.fetch("KAFKA_URL").split(",")
ssl_ca_cert = ENV.fetch("KAFKA_TRUSTED_CERT")
ssl_client_cert = ENV.fetch("KAFKA_CLIENT_CERT")
ssl_client_cert_key = ENV.fetch("KAFKA_CLIENT_CERT_KEY")

kafka = Kafka.new(
  seed_brokers: brokers,
  client_id: client_id,
  socket_timeout: 120,
  logger: logger,
  ssl_ca_cert: ssl_ca_cert,
  ssl_client_cert: ssl_client_cert,
  ssl_client_cert_key: ssl_client_cert_key,
)

[sdball]

And here's how to connect to Heroku Kafka from kafkacat: for posterity

$ kafkacat \
  -t ssl-test \
  -o beginning \
  -e \
  -b $(echo $KAFKA_URL | sed -e 's/kafka+ssl:\/\///g') \
  -X security.protocol=ssl \
  -X ssl.key.location=<(echo $KAFKA_CLIENT_CERT_KEY) \
  -X ssl.certificate.location=<(echo $KAFKA_CLIENT_CERT) \
  -X ssl.ca.location=<(echo $KAFKA_TRUSTED_CERT)

[sdball]

Yeah it's pretty clear that brod and kafka_ex are both using the Erlang SSL module:

http://erlang.org/doc/man/ssl.html

Happily it looks like that module can accept values and not just files. I'll try setting them for brod. First with the certs as files like it wants and next with the underlying Erlang SSL module options for values.

[sdball]

Woo! After much experimentation I finally found the magical combination of extracted pieces and SSL incantations to allow us to configure the Erlang SSL module used by brod to talk to the Heroku Kafka!

The magics:

def consumer_config do
  [
    auto_start_producers: false,
    allow_topic_auto_creation: false,
    ssl: [
      cert: client_cert,
      key: client_cert_key,
    ]
  ]
end

defp client_cert do
  {_type, der, _} = "KAFKA_CLIENT_CERT"
  |> System.get_env
  |> :public_key.pem_decode
  |> List.first
  der
end

defp client_cert_key do
  {type, der, _} = "KAFKA_CLIENT_CERT_KEY"
  |> System.get_env
  |> :public_key.pem_decode
  |> List.first
  {type, der}
end

This is all hacked together but working!

Next is to nicely expose the SSL configuration pieces. I have no problem with Kaffe assuming Heroku Kafka SSL setup since it is our client after all.

Perhaps a config like…

config :kaffe,
  consumer: [
    endpoints: [kafka: 9092],
    ssl: [
      client_cert: System.get_env("KAFKA_CLIENT_CERT"),
      client_cert_key: System.get_env("KAFKA_CLIENT_CERT_KEY")
    ],
    topics: ["whitelist"],
    offset_commit_interval_seconds: 5, # default
    begin_offset: :earliest # default :latest
  ],
  producer: [
    endpoints: [kafka: 9092],
    ssl: [
      client_cert: System.get_env("KAFKA_CLIENT_CERT"),
      client_cert_key: System.get_env("KAFKA_CLIENT_CERT_KEY")
    ],
    topics: ["decorated-whitelist"],
    partition_strategy: :round_robin # default
  ]

Using default values…

config :kaffe,
  consumer: [
    endpoints: [kafka: 9092], # yeah, gotta pull these from KAFKA_URL
    ssl: [
      client_cert: System.get_env("KAFKA_CLIENT_CERT"),
      client_cert_key: System.get_env("KAFKA_CLIENT_CERT_KEY")
    ],
    topics: ["whitelist"]
  ],
  producer: [
    endpoints: [kafka: 9092], # yeah, gotta pull these from KAFKA_URL
    ssl: [
      client_cert: System.get_env("KAFKA_CLIENT_CERT"),
      client_cert_key: System.get_env("KAFKA_CLIENT_CERT_KEY")
    ],
    topics: ["decorated-whitelist"]
  ]

[sdball]

What we get from KAFKA_URL:

iex(1)> "KAFKA_URL" |> System.get_env
"kafka+ssl://34.195.140.72:9096,kafka+ssl://34.195.135.147:9096,kafka+ssl://34.195.140.195:9096"

What brod wants for its endpoints configuration:

[{:"34.195.140.72", 9096}, {:"34.195.135.147", 9096}, {:"34.195.140.195", 9096}]

A straight up data transformation problem?! Yesssss