depot/nixos/modules/services/misc/apache-kafka.nix

Ignoring revisions in .git-blame-ignore-revs. Click here to bypass and see the normal blame view.

216 lines
7.7 KiB
Nix
Raw Normal View History

{ config, lib, pkgs, ... }:
let
cfg = config.services.apache-kafka;
# The `javaProperties` generator takes care of various escaping rules and
# generation of the properties file, but we'll handle stringly conversion
# ourselves in mkPropertySettings and stringlySettings, since we know more
# about the specifically allowed format eg. for lists of this type, and we
# don't want to coerce-downsample values to str too early by having the
# coercedTypes from javaProperties directly in our NixOS option types.
#
# Make sure every `freeformType` and any specific option type in `settings` is
# supported here.
mkPropertyString = let
render = {
bool = lib.boolToString;
int = toString;
list = lib.concatMapStringsSep "," mkPropertyString;
string = lib.id;
};
in
v: render.${builtins.typeOf v} v;
stringlySettings = lib.mapAttrs (_: mkPropertyString)
(lib.filterAttrs (_: v: v != null) cfg.settings);
generator = (pkgs.formats.javaProperties {}).generate;
in {
options.services.apache-kafka = {
enable = lib.mkEnableOption "Apache Kafka event streaming broker";
settings = lib.mkOption {
description = ''
[Kafka broker configuration](https://kafka.apache.org/documentation.html#brokerconfigs)
{file}`server.properties`.
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 = lib.types.submodule {
freeformType = with lib.types; let
primitive = oneOf [bool int str];
in lazyAttrsOf (nullOr (either primitive (listOf primitive)));
options = {
"broker.id" = lib.mkOption {
description = "Broker ID. -1 or null to auto-allocate in zookeeper mode.";
default = null;
type = with lib.types; nullOr int;
};
"log.dirs" = lib.mkOption {
description = "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 lib.types; listOf path;
};
"listeners" = lib.mkOption {
description = ''
Kafka Listener List.
See [listeners](https://kafka.apache.org/documentation/#brokerconfigs_listeners).
'';
type = lib.types.listOf lib.types.str;
default = [ "PLAINTEXT://localhost:9092" ];
};
};
};
};
clusterId = lib.mkOption {
description = ''
KRaft mode ClusterId used for formatting log directories. Can be generated with `kafka-storage.sh random-uuid`
'';
type = with lib.types; nullOr str;
default = null;
};
configFiles.serverProperties = lib.mkOption {
description = ''
Kafka server.properties configuration file path.
Defaults to the rendered `settings`.
'';
type = lib.types.path;
};
configFiles.log4jProperties = lib.mkOption {
description = "Kafka log4j property configuration file path";
type = lib.types.path;
default = pkgs.writeText "log4j.properties" cfg.log4jProperties;
defaultText = ''pkgs.writeText "log4j.properties" cfg.log4jProperties'';
};
formatLogDirs = lib.mkOption {
description = ''
Whether to format log dirs in KRaft mode if all log dirs are
unformatted, ie. they contain no meta.properties.
'';
type = lib.types.bool;
default = false;
};
formatLogDirsIgnoreFormatted = lib.mkOption {
description = ''
Whether to ignore already formatted log dirs when formatting log dirs,
instead of failing. Useful when replacing or adding disks.
'';
type = lib.types.bool;
default = false;
};
log4jProperties = lib.mkOption {
description = "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 = lib.types.lines;
};
jvmOptions = lib.mkOption {
description = "Extra command line options for the JVM running Kafka.";
default = [];
type = lib.types.listOf lib.types.str;
example = [
"-Djava.net.preferIPv4Stack=true"
"-Dcom.sun.management.jmxremote"
"-Dcom.sun.management.jmxremote.local.only=true"
];
};
package = lib.mkPackageOption pkgs "apacheKafka" { };
jre = lib.mkOption {
description = "The JRE with which to run Kafka";
default = cfg.package.passthru.jre;
defaultText = lib.literalExpression "pkgs.apacheKafka.passthru.jre";
type = lib.types.package;
};
};
imports = [
(lib.mkRenamedOptionModule
[ "services" "apache-kafka" "brokerId" ]
[ "services" "apache-kafka" "settings" ''broker.id'' ])
(lib.mkRenamedOptionModule
[ "services" "apache-kafka" "logDirs" ]
[ "services" "apache-kafka" "settings" ''log.dirs'' ])
(lib.mkRenamedOptionModule
[ "services" "apache-kafka" "zookeeper" ]
[ "services" "apache-kafka" "settings" ''zookeeper.connect'' ])
(lib.mkRemovedOptionModule [ "services" "apache-kafka" "port" ]
"Please see services.apache-kafka.settings.listeners and its documentation instead")
(lib.mkRemovedOptionModule [ "services" "apache-kafka" "hostname" ]
"Please see services.apache-kafka.settings.listeners and its documentation instead")
(lib.mkRemovedOptionModule [ "services" "apache-kafka" "extraProperties" ]
"Please see services.apache-kafka.settings and its documentation instead")
(lib.mkRemovedOptionModule [ "services" "apache-kafka" "serverProperties" ]
"Please see services.apache-kafka.settings and its documentation instead")
];
config = lib.mkIf cfg.enable {
services.apache-kafka.configFiles.serverProperties = generator "server.properties" stringlySettings;
users.users.apache-kafka = {
isSystemUser = true;
group = "apache-kafka";
description = "Apache Kafka daemon user";
};
users.groups.apache-kafka = {};
systemd.tmpfiles.rules = map (logDir: "d '${logDir}' 0700 apache-kafka - - -") cfg.settings."log.dirs";
systemd.services.apache-kafka = {
description = "Apache Kafka Daemon";
wantedBy = [ "multi-user.target" ];
after = [ "network.target" ];
preStart = lib.mkIf cfg.formatLogDirs
(if cfg.formatLogDirsIgnoreFormatted then ''
${cfg.package}/bin/kafka-storage.sh format -t "${cfg.clusterId}" -c ${cfg.configFiles.serverProperties} --ignore-formatted
'' else ''
if ${lib.concatMapStringsSep " && " (l: ''[ ! -f "${l}/meta.properties" ]'') cfg.settings."log.dirs"}; then
${cfg.package}/bin/kafka-storage.sh format -t "${cfg.clusterId}" -c ${cfg.configFiles.serverProperties}
fi
'');
serviceConfig = {
ExecStart = ''
${cfg.jre}/bin/java \
-cp "${cfg.package}/libs/*" \
-Dlog4j.configuration=file:${cfg.configFiles.log4jProperties} \
${toString cfg.jvmOptions} \
kafka.Kafka \
${cfg.configFiles.serverProperties}
'';
User = "apache-kafka";
SuccessExitStatus = "0 143";
};
};
};
meta.doc = ./kafka.md;
meta.maintainers = with lib.maintainers; [
srhb
];
}