nixos/modules/services/cluster/druid/default.nix

   1 {
   2   config,
   3   lib,
   4   pkgs,
   5   ...
   6 }:
   7 let
   8   cfg = config.services.druid;
   9   inherit (lib)
  10     concatStrings
  11     concatStringsSep
  12     mapAttrsToList
  13     concatMap
  14     attrByPath
  15     mkIf
  16     mkMerge
  17     mkEnableOption
  18     mkOption
  19     types
  20     mkPackageOption
  21     ;
  22
  23   druidServiceOption = serviceName: {
  24     enable = mkEnableOption serviceName;
  25
  26     restartIfChanged = mkOption {
  27       type = types.bool;
  28       description = ''
  29         Automatically restart the service on config change.
  30         This can be set to false to defer restarts on clusters running critical applications.
  31         Please consider the security implications of inadvertently running an older version,
  32         and the possibility of unexpected behavior caused by inconsistent versions across a cluster when disabling this option.
  33       '';
  34       default = false;
  35     };
  36
  37     config = mkOption {
  38       default = { };
  39       type = types.attrsOf types.anything;
  40       description = ''
  41         (key=value) Configuration to be written to runtime.properties of the druid ${serviceName}
  42         <https://druid.apache.org/docs/latest/configuration/index.html>
  43       '';
  44       example = {
  45         "druid.plainTextPort" = "8082";
  46         "druid.service" = "servicename";
  47       };
  48     };
  49
  50     jdk = mkPackageOption pkgs "JDK" { default = [ "jdk17_headless" ]; };
  51
  52     jvmArgs = mkOption {
  53       type = types.str;
  54       default = "";
  55       description = "Arguments to pass to the JVM";
  56     };
  57
  58     openFirewall = mkOption {
  59       type = types.bool;
  60       default = false;
  61       description = "Open firewall ports for ${serviceName}.";
  62     };
  63
  64     internalConfig = mkOption {
  65       default = { };
  66       type = types.attrsOf types.anything;
  67       internal = true;
  68       description = "Internal Option to add to runtime.properties for ${serviceName}.";
  69     };
  70   };
  71
  72   druidServiceConfig =
  73     {
  74       name,
  75       serviceOptions ? cfg."${name}",
  76       allowedTCPPorts ? [ ],
  77       tmpDirs ? [ ],
  78       extraConfig ? { },
  79     }:
  80     (mkIf serviceOptions.enable (mkMerge [
  81       {
  82         systemd = {
  83           services."druid-${name}" = {
  84             after = [ "network.target" ];
  85
  86             description = "Druid ${name}";
  87
  88             wantedBy = [ "multi-user.target" ];
  89
  90             inherit (serviceOptions) restartIfChanged;
  91
  92             path = [
  93               cfg.package
  94               serviceOptions.jdk
  95             ];
  96
  97             script =
  98               let
  99                 cfgFile =
 100                   fileName: properties:
 101                   pkgs.writeTextDir fileName (
 102                     concatStringsSep "\n" (mapAttrsToList (n: v: "${n}=${toString v}") properties)
 103                   );
 104
 105                 commonConfigFile = cfgFile "common.runtime.properties" cfg.commonConfig;
 106
 107                 configFile = cfgFile "runtime.properties" (serviceOptions.config // serviceOptions.internalConfig);
 108
 109                 extraClassPath = concatStrings (map (path: ":" + path) cfg.extraClassPaths);
 110
 111                 extraConfDir = concatStrings (map (dir: ":" + dir + "/*") cfg.extraConfDirs);
 112               in
 113               ''
 114                 run-java -Dlog4j.configurationFile=file:${cfg.log4j} \
 115                   -Ddruid.extensions.directory=${cfg.package}/extensions \
 116                   -Ddruid.extensions.hadoopDependenciesDir=${cfg.package}/hadoop-dependencies \
 117                   -classpath  ${commonConfigFile}:${configFile}:${cfg.package}/lib/\*${extraClassPath}${extraConfDir} \
 118                   ${serviceOptions.jvmArgs} \
 119                   org.apache.druid.cli.Main server ${name}
 120               '';
 121
 122             serviceConfig = {
 123               User = "druid";
 124               SyslogIdentifier = "druid-${name}";
 125               Restart = "always";
 126             };
 127           };
 128
 129           tmpfiles.rules = concatMap (x: [ "d ${x} 0755 druid druid" ]) (cfg.commonTmpDirs ++ tmpDirs);
 130         };
 131         networking.firewall.allowedTCPPorts = mkIf (attrByPath [
 132           "openFirewall"
 133         ] false serviceOptions) allowedTCPPorts;
 134
 135         users = {
 136           users.druid = {
 137             description = "Druid user";
 138             group = "druid";
 139             isNormalUser = true;
 140           };
 141           groups.druid = { };
 142         };
 143       }
 144       extraConfig
 145     ]));
 146 in
 147 {
 148   options.services.druid = {
 149     package = mkPackageOption pkgs "apache-druid" { default = [ "druid" ]; };
 150
 151     commonConfig = mkOption {
 152       default = { };
 153
 154       type = types.attrsOf types.anything;
 155
 156       description = "(key=value) Configuration to be written to common.runtime.properties";
 157
 158       example = {
 159         "druid.zk.service.host" = "localhost:2181";
 160         "druid.metadata.storage.type" = "mysql";
 161         "druid.metadata.storage.connector.connectURI" = "jdbc:mysql://localhost:3306/druid";
 162         "druid.extensions.loadList" = ''[ "mysql-metadata-storage" ]'';
 163       };
 164     };
 165
 166     commonTmpDirs = mkOption {
 167       default = [ "/var/log/druid/requests" ];
 168       type = types.listOf types.str;
 169       description = "Common List of directories used by druid processes";
 170     };
 171
 172     log4j = mkOption {
 173       type = types.path;
 174       description = "Log4j Configuration for the druid process";
 175     };
 176
 177     extraClassPaths = mkOption {
 178       default = [ ];
 179       type = types.listOf types.str;
 180       description = "Extra classpath to include in the jvm";
 181     };
 182
 183     extraConfDirs = mkOption {
 184       default = [ ];
 185       type = types.listOf types.path;
 186       description = "Extra Conf Dirs to include in the jvm";
 187     };
 188
 189     overlord = druidServiceOption "Druid Overlord";
 190
 191     coordinator = druidServiceOption "Druid Coordinator";
 192
 193     broker = druidServiceOption "Druid Broker";
 194
 195     historical = (druidServiceOption "Druid Historical") // {
 196       segmentLocations = mkOption {
 197
 198         default = null;
 199
 200         description = "Locations where the historical will store its data.";
 201
 202         type =
 203           with types;
 204           nullOr (
 205             listOf (submodule {
 206               options = {
 207                 path = mkOption {
 208                   type = path;
 209                   description = "the path to store the segments";
 210                 };
 211
 212                 maxSize = mkOption {
 213                   type = str;
 214                   description = "Max size the druid historical can occupy";
 215                 };
 216
 217                 freeSpacePercent = mkOption {
 218                   type = float;
 219                   default = 1.0;
 220                   description = "Druid Historical will fail to write if it exceeds this value";
 221                 };
 222               };
 223             })
 224           );
 225
 226       };
 227     };
 228
 229     middleManager = druidServiceOption "Druid middleManager";
 230     router = druidServiceOption "Druid Router";
 231   };
 232   config = mkMerge [
 233     (druidServiceConfig rec {
 234       name = "overlord";
 235       allowedTCPPorts = [ (attrByPath [ "druid.plaintextPort" ] 8090 cfg."${name}".config) ];
 236     })
 237
 238     (druidServiceConfig rec {
 239       name = "coordinator";
 240       allowedTCPPorts = [ (attrByPath [ "druid.plaintextPort" ] 8081 cfg."${name}".config) ];
 241     })
 242
 243     (druidServiceConfig rec {
 244       name = "broker";
 245
 246       tmpDirs = [ (attrByPath [ "druid.lookup.snapshotWorkingDir" ] "" cfg."${name}".config) ];
 247
 248       allowedTCPPorts = [ (attrByPath [ "druid.plaintextPort" ] 8082 cfg."${name}".config) ];
 249     })
 250
 251     (druidServiceConfig rec {
 252       name = "historical";
 253
 254       tmpDirs = [
 255         (attrByPath [ "druid.lookup.snapshotWorkingDir" ] "" cfg."${name}".config)
 256       ] ++ (map (x: x.path) cfg."${name}".segmentLocations);
 257
 258       allowedTCPPorts = [ (attrByPath [ "druid.plaintextPort" ] 8083 cfg."${name}".config) ];
 259
 260       extraConfig.services.druid.historical.internalConfig."druid.segmentCache.locations" = builtins.toJSON cfg.historical.segmentLocations;
 261     })
 262
 263     (druidServiceConfig rec {
 264       name = "middleManager";
 265
 266       tmpDirs = [
 267         "/var/log/druid/indexer"
 268       ] ++ [ (attrByPath [ "druid.indexer.task.baseTaskDir" ] "" cfg."${name}".config) ];
 269
 270       allowedTCPPorts = [ (attrByPath [ "druid.plaintextPort" ] 8091 cfg."${name}".config) ];
 271
 272       extraConfig = {
 273         services.druid.middleManager.internalConfig = {
 274           "druid.indexer.runner.javaCommand" = "${cfg.middleManager.jdk}/bin/java";
 275           "druid.indexer.runner.javaOpts" =
 276             (attrByPath [ "druid.indexer.runner.javaOpts" ] "" cfg.middleManager.config)
 277             + " -Dlog4j.configurationFile=file:${cfg.log4j}";
 278         };
 279
 280         networking.firewall.allowedTCPPortRanges = mkIf cfg.middleManager.openFirewall [
 281           {
 282             from = attrByPath [ "druid.indexer.runner.startPort" ] 8100 cfg.middleManager.config;
 283             to = attrByPath [ "druid.indexer.runner.endPort" ] 65535 cfg.middleManager.config;
 284           }
 285         ];
 286       };
 287     })
 288
 289     (druidServiceConfig rec {
 290       name = "router";
 291
 292       allowedTCPPorts = [ (attrByPath [ "druid.plaintextPort" ] 8888 cfg."${name}".config) ];
 293     })
 294   ];
 295
 296 }