services-flake/nix/apache-kafka.nix

199 lines
6.4 KiB
Nix
Raw Normal View History

2023-08-29 14:34:04 +03:00
# Based on: https://github.com/NixOS/nixpkgs/blob/master/nixos/modules/services/misc/apache-kafka.nix
{ config, lib, pkgs, name, ... }:
let
mkPropertyString =
let
render = {
bool = lib.boolToString;
int = toString;
list = lib.concatMapStringsSep "," mkPropertyString;
string = lib.id;
};
in
v: render.${lib.strings.typeOf v} v;
stringlySettings = lib.mapAttrs (_: mkPropertyString)
(lib.filterAttrs (_: v: v != null) config.settings);
2023-08-29 14:34:04 +03:00
generator = (pkgs.formats.javaProperties { }).generate;
in
2023-08-29 14:34:04 +03:00
with lib;
{
options = {
enable = mkEnableOption (lib.mdDoc "Apache Kafka event streaming broker");
2023-08-29 14:34:04 +03:00
port = mkOption {
description = lib.mdDoc "Port number the broker should listen on.";
default = 9092;
type = types.port;
};
dataDir = lib.mkOption {
type = types.str;
default = "./data/${name}";
description = lib.mdDoc "The apache-kafka data directory";
2023-08-29 14:34:04 +03:00
};
settings = mkOption {
description = lib.mdDoc ''
[Kafka broker configuration](https://kafka.apache.org/documentation.html#brokerconfigs)
{file}`server.properties`.
2023-08-29 14:34:04 +03:00
Note that .properties files contain mappings from string to string.
Keys with dots are NOT represented by nested attrs in these settings,
but instead as quoted strings (ie. `settings."broker.id"`, NOT
`settings.broker.id`).
'';
type = types.submodule {
freeformType = with types; let
primitive = oneOf [ bool int str ];
in
lazyAttrsOf (nullOr (either primitive (listOf primitive)));
options = {
"broker.id" = mkOption {
description = lib.mdDoc "Broker ID. -1 or null to auto-allocate in zookeeper mode.";
default = null;
type = with types; nullOr int;
};
"log.dirs" = mkOption {
description = lib.mdDoc "Log file directories.";
# Deliberaly leave out old default and use the rewrite opportunity
# to have users choose a safer value -- /tmp might be volatile and is a
# slightly scary default choice.
# default = [ "/tmp/apache-kafka" ];
type = with types; listOf str;
default = [ (config.dataDir + "/logs") ];
};
"listeners" = mkOption {
description = lib.mdDoc ''
Kafka Listener List.
See [listeners](https://kafka.apache.org/documentation/#brokerconfigs_listeners).
'';
type = types.listOf types.str;
default = [ "PLAINTEXT://localhost:${builtins.toString config.port}" ];
};
};
};
2023-08-29 14:34:04 +03:00
};
clusterId = mkOption {
description = lib.mdDoc ''
KRaft mode ClusterId used for formatting log directories. Can be generated with `kafka-storage.sh random-uuid`
'';
type = with types; nullOr str;
2023-08-29 14:34:04 +03:00
default = null;
};
configFiles.serverProperties = mkOption {
2023-08-29 14:34:04 +03:00
description = lib.mdDoc ''
Kafka server.properties configuration file path.
Defaults to the rendered `settings`.
2023-08-29 14:34:04 +03:00
'';
type = types.path;
default = generator "server.properties" stringlySettings;
};
configFiles.log4jProperties = mkOption {
description = lib.mdDoc "Kafka log4j property configuration file path";
type = types.path;
default = pkgs.writeText "log4j.properties" config.log4jProperties;
defaultText = ''pkgs.writeText "log4j.properties" config.log4jProperties'';
};
formatLogDirs = mkOption {
description = lib.mdDoc ''
Whether to format log dirs in KRaft mode if all log dirs are
unformatted, ie. they contain no meta.properties.
'';
type = types.bool;
default = false;
};
formatLogDirsIgnoreFormatted = mkOption {
description = lib.mdDoc ''
Whether to ignore already formatted log dirs when formatting log dirs,
instead of failing. Useful when replacing or adding disks.
'';
type = types.bool;
default = false;
2023-08-29 14:34:04 +03:00
};
log4jProperties = mkOption {
description = lib.mdDoc "Kafka log4j property configuration.";
default = ''
log4j.rootLogger=INFO, stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=[%d] %p %m (%c)%n
'';
type = types.lines;
};
jvmOptions = mkOption {
description = lib.mdDoc "Extra command line options for the JVM running Kafka.";
default = [ ];
type = types.listOf types.str;
example = [
"-Djava.net.preferIPv4Stack=true"
"-Dcom.sun.management.jmxremote"
"-Dcom.sun.management.jmxremote.local.only=true"
];
};
package = mkPackageOption pkgs "apacheKafka" { };
2023-08-29 14:34:04 +03:00
jre = mkOption {
description = lib.mdDoc "The JRE with which to run Kafka";
default = config.package.passthru.jre;
defaultText = literalExpression "pkgs.apacheKafka.passthru.jre";
type = types.package;
};
outputs.settings = lib.mkOption {
type = types.deferredModule;
internal = true;
readOnly = true;
default = {
processes = {
"${name}" =
let
startScript = pkgs.writeShellApplication {
name = "start-kafka";
runtimeInputs = [ config.jre ];
text = ''
java \
-cp "${config.package}/libs/*" \
-Dlog4j.configuration=file:${config.configFiles.log4jProperties} \
${toString config.jvmOptions} \
kafka.Kafka \
${config.configFiles.serverProperties}
'';
};
2023-08-29 14:34:04 +03:00
in
{
command = startScript;
2023-08-29 14:34:04 +03:00
readiness_probe = {
# TODO: need to find a better way to check if kafka is ready. Maybe use one of the scripts in bin?
exec.command = "${pkgs.netcat.nc}/bin/nc -z localhost ${builtins.toString config.port}";
2023-08-29 14:34:04 +03:00
initial_delay_seconds = 2;
period_seconds = 10;
timeout_seconds = 4;
success_threshold = 1;
failure_threshold = 5;
};
2023-10-04 20:20:59 +03:00
namespace = name;
2023-08-29 14:34:04 +03:00
availability.restart = "on_failure";
};
};
};
};
};
}