pyrox.dev
1{ config, lib, pkgs, ... }:
2
3with lib;
4
5let
6 cfg = config.services.cassandra;
7 cassandraPackage = cfg.package.override {
8 jre = cfg.jre;
9 };
10 cassandraUser = {
11 name = cfg.user;
12 home = "/var/lib/cassandra";
13 description = "Cassandra role user";
14 };
15
16 cassandraRackDcProperties = ''
17 dc=${cfg.dc}
18 rack=${cfg.rack}
19 '';
20
21 cassandraConf = ''
22 cluster_name: ${cfg.clusterName}
23 num_tokens: 256
24 auto_bootstrap: ${if cfg.autoBootstrap then "true" else "false"}
25 hinted_handoff_enabled: ${if cfg.hintedHandOff then "true" else "false"}
26 hinted_handoff_throttle_in_kb: ${builtins.toString cfg.hintedHandOffThrottle}
27 max_hints_delivery_threads: 2
28 max_hint_window_in_ms: 10800000 # 3 hours
29 authenticator: ${cfg.authenticator}
30 authorizer: ${cfg.authorizer}
31 permissions_validity_in_ms: 2000
32 partitioner: org.apache.cassandra.dht.Murmur3Partitioner
33 data_file_directories:
34 ${builtins.concatStringsSep "\n" (map (v: " - "+v) cfg.dataDirs)}
35 commitlog_directory: ${cfg.commitLogDirectory}
36 disk_failure_policy: stop
37 key_cache_size_in_mb:
38 key_cache_save_period: 14400
39 row_cache_size_in_mb: 0
40 row_cache_save_period: 0
41 saved_caches_directory: ${cfg.savedCachesDirectory}
42 commitlog_sync: ${cfg.commitLogSync}
43 commitlog_sync_period_in_ms: ${builtins.toString cfg.commitLogSyncPeriod}
44 commitlog_segment_size_in_mb: 32
45 seed_provider:
46 - class_name: org.apache.cassandra.locator.SimpleSeedProvider
47 parameters:
48 - seeds: "${builtins.concatStringsSep "," cfg.seeds}"
49 concurrent_reads: ${builtins.toString cfg.concurrentReads}
50 concurrent_writes: ${builtins.toString cfg.concurrentWrites}
51 memtable_flush_queue_size: 4
52 trickle_fsync: false
53 trickle_fsync_interval_in_kb: 10240
54 storage_port: 7000
55 ssl_storage_port: 7001
56 listen_address: ${cfg.listenAddress}
57 start_native_transport: true
58 native_transport_port: 9042
59 start_rpc: true
60 rpc_address: ${cfg.rpcAddress}
61 rpc_port: 9160
62 rpc_keepalive: true
63 rpc_server_type: sync
64 thrift_framed_transport_size_in_mb: 15
65 incremental_backups: ${if cfg.incrementalBackups then "true" else "false"}
66 snapshot_before_compaction: false
67 auto_snapshot: true
68 column_index_size_in_kb: 64
69 in_memory_compaction_limit_in_mb: 64
70 multithreaded_compaction: false
71 compaction_throughput_mb_per_sec: 16
72 compaction_preheat_key_cache: true
73 read_request_timeout_in_ms: 10000
74 range_request_timeout_in_ms: 10000
75 write_request_timeout_in_ms: 10000
76 cas_contention_timeout_in_ms: 1000
77 truncate_request_timeout_in_ms: 60000
78 request_timeout_in_ms: 10000
79 cross_node_timeout: false
80 endpoint_snitch: ${cfg.snitch}
81 dynamic_snitch_update_interval_in_ms: 100
82 dynamic_snitch_reset_interval_in_ms: 600000
83 dynamic_snitch_badness_threshold: 0.1
84 request_scheduler: org.apache.cassandra.scheduler.NoScheduler
85 server_encryption_options:
86 internode_encryption: ${cfg.internodeEncryption}
87 keystore: ${cfg.keyStorePath}
88 keystore_password: ${cfg.keyStorePassword}
89 truststore: ${cfg.trustStorePath}
90 truststore_password: ${cfg.trustStorePassword}
91 client_encryption_options:
92 enabled: ${if cfg.clientEncryption then "true" else "false"}
93 keystore: ${cfg.keyStorePath}
94 keystore_password: ${cfg.keyStorePassword}
95 internode_compression: all
96 inter_dc_tcp_nodelay: false
97 preheat_kernel_page_cache: false
98 streaming_socket_timeout_in_ms: ${toString cfg.streamingSocketTimoutInMS}
99 '';
100
101 cassandraLog = ''
102 log4j.rootLogger=${cfg.logLevel},stdout
103 log4j.appender.stdout=org.apache.log4j.ConsoleAppender
104 log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
105 log4j.appender.stdout.layout.ConversionPattern=%5p [%t] %d{HH:mm:ss,SSS} %m%n
106 '';
107
108 cassandraConfFile = pkgs.writeText "cassandra.yaml" cassandraConf;
109 cassandraLogFile = pkgs.writeText "log4j-server.properties" cassandraLog;
110 cassandraRackFile = pkgs.writeText "cassandra-rackdc.properties" cassandraRackDcProperties;
111
112 cassandraEnvironment = {
113 CASSANDRA_HOME = cassandraPackage;
114 JAVA_HOME = cfg.jre;
115 CASSANDRA_CONF = "/etc/cassandra";
116 };
117
118in {
119
120 ###### interface
121
122 options.services.cassandra = {
123 enable = mkOption {
124 description = "Whether to enable cassandra.";
125 default = false;
126 type = types.bool;
127 };
128 package = mkOption {
129 description = "Cassandra package to use.";
130 default = pkgs.cassandra;
131 defaultText = "pkgs.cassandra";
132 type = types.package;
133 };
134 jre = mkOption {
135 description = "JRE package to run cassandra service.";
136 default = pkgs.jre;
137 defaultText = "pkgs.jre";
138 type = types.package;
139 };
140 user = mkOption {
141 description = "User that runs cassandra service.";
142 default = "cassandra";
143 type = types.string;
144 };
145 group = mkOption {
146 description = "Group that runs cassandra service.";
147 default = "cassandra";
148 type = types.string;
149 };
150 envFile = mkOption {
151 description = "path to cassandra-env.sh";
152 default = "${cassandraPackage}/conf/cassandra-env.sh";
153 defaultText = "\${cassandraPackage}/conf/cassandra-env.sh";
154 type = types.path;
155 };
156 clusterName = mkOption {
157 description = "set cluster name";
158 default = "cassandra";
159 example = "prod-cluster0";
160 type = types.string;
161 };
162 commitLogDirectory = mkOption {
163 description = "directory for commit logs";
164 default = "/var/lib/cassandra/commit_log";
165 type = types.string;
166 };
167 savedCachesDirectory = mkOption {
168 description = "directory for saved caches";
169 default = "/var/lib/cassandra/saved_caches";
170 type = types.string;
171 };
172 hintedHandOff = mkOption {
173 description = "enable hinted handoff";
174 default = true;
175 type = types.bool;
176 };
177 hintedHandOffThrottle = mkOption {
178 description = "hinted hand off throttle rate in kb";
179 default = 1024;
180 type = types.int;
181 };
182 commitLogSync = mkOption {
183 description = "commitlog sync method";
184 default = "periodic";
185 type = types.str;
186 example = "batch";
187 };
188 commitLogSyncPeriod = mkOption {
189 description = "commitlog sync period in ms ";
190 default = 10000;
191 type = types.int;
192 };
193 envScript = mkOption {
194 default = "${cassandraPackage}/conf/cassandra-env.sh";
195 defaultText = "\${cassandraPackage}/conf/cassandra-env.sh";
196 type = types.path;
197 description = "Supply your own cassandra-env.sh rather than using the default";
198 };
199 extraParams = mkOption {
200 description = "add additional lines to cassandra-env.sh";
201 default = [];
202 example = [''JVM_OPTS="$JVM_OPTS -Dcassandra.available_processors=1"''];
203 type = types.listOf types.str;
204 };
205 dataDirs = mkOption {
206 type = types.listOf types.path;
207 default = [ "/var/lib/cassandra/data" ];
208 description = "Data directories for cassandra";
209 };
210 logLevel = mkOption {
211 type = types.str;
212 default = "INFO";
213 description = "default logging level for log4j";
214 };
215 internodeEncryption = mkOption {
216 description = "enable internode encryption";
217 default = "none";
218 example = "all";
219 type = types.str;
220 };
221 clientEncryption = mkOption {
222 description = "enable client encryption";
223 default = false;
224 type = types.bool;
225 };
226 trustStorePath = mkOption {
227 description = "path to truststore";
228 default = ".conf/truststore";
229 type = types.str;
230 };
231 keyStorePath = mkOption {
232 description = "path to keystore";
233 default = ".conf/keystore";
234 type = types.str;
235 };
236 keyStorePassword = mkOption {
237 description = "password to keystore";
238 default = "cassandra";
239 type = types.str;
240 };
241 trustStorePassword = mkOption {
242 description = "password to truststore";
243 default = "cassandra";
244 type = types.str;
245 };
246 seeds = mkOption {
247 description = "password to truststore";
248 default = [ "127.0.0.1" ];
249 type = types.listOf types.str;
250 };
251 concurrentWrites = mkOption {
252 description = "number of concurrent writes allowed";
253 default = 32;
254 type = types.int;
255 };
256 concurrentReads = mkOption {
257 description = "number of concurrent reads allowed";
258 default = 32;
259 type = types.int;
260 };
261 listenAddress = mkOption {
262 description = "listen address";
263 default = "localhost";
264 type = types.str;
265 };
266 rpcAddress = mkOption {
267 description = "rpc listener address";
268 default = "localhost";
269 type = types.str;
270 };
271 incrementalBackups = mkOption {
272 description = "enable incremental backups";
273 default = false;
274 type = types.bool;
275 };
276 snitch = mkOption {
277 description = "snitch to use for topology discovery";
278 default = "GossipingPropertyFileSnitch";
279 example = "Ec2Snitch";
280 type = types.str;
281 };
282 dc = mkOption {
283 description = "datacenter for use in topology configuration";
284 default = "DC1";
285 example = "DC1";
286 type = types.str;
287 };
288 rack = mkOption {
289 description = "rack for use in topology configuration";
290 default = "RAC1";
291 example = "RAC1";
292 type = types.str;
293 };
294 authorizer = mkOption {
295 description = "
296 Authorization backend, implementing IAuthorizer; used to limit access/provide permissions
297 ";
298 default = "AllowAllAuthorizer";
299 example = "CassandraAuthorizer";
300 type = types.str;
301 };
302 authenticator = mkOption {
303 description = "
304 Authentication backend, implementing IAuthenticator; used to identify users
305 ";
306 default = "AllowAllAuthenticator";
307 example = "PasswordAuthenticator";
308 type = types.str;
309 };
310 autoBootstrap = mkOption {
311 description = "It makes new (non-seed) nodes automatically migrate the right data to themselves.";
312 default = true;
313 example = true;
314 type = types.bool;
315 };
316 streamingSocketTimoutInMS = mkOption {
317 description = "Enable or disable socket timeout for streaming operations";
318 default = 3600000; #CASSANDRA-8611
319 example = 120;
320 type = types.int;
321 };
322 repairStartAt = mkOption {
323 default = "Sun";
324 type = types.string;
325 description = ''
326 Defines realtime (i.e. wallclock) timers with calendar event
327 expressions. For more details re: systemd OnCalendar at
328 https://www.freedesktop.org/software/systemd/man/systemd.time.html#Displaying%20Time%20Spans
329 '';
330 example = ["weekly" "daily" "08:05:40" "mon,fri *-1/2-1,3 *:30:45"];
331 };
332 repairRandomizedDelayInSec = mkOption {
333 default = 0;
334 type = types.int;
335 description = ''Delay the timer by a randomly selected, evenly distributed
336 amount of time between 0 and the specified time value. re: systemd timer
337 RandomizedDelaySec for more details
338 '';
339 };
340 repairPostStop = mkOption {
341 default = null;
342 type = types.nullOr types.string;
343 description = ''
344 Run a script when repair is over. One can use it to send statsd events, email, etc.
345 '';
346 };
347 repairPostStart = mkOption {
348 default = null;
349 type = types.nullOr types.string;
350 description = ''
351 Run a script when repair starts. One can use it to send statsd events, email, etc.
352 It has same semantics as systemd ExecStopPost; So, if it fails, unit is consisdered
353 failed.
354 '';
355 };
356 };
357
358 ###### implementation
359
360 config = mkIf cfg.enable {
361
362 environment.etc."cassandra/cassandra-rackdc.properties" = {
363 source = cassandraRackFile;
364 };
365 environment.etc."cassandra/cassandra.yaml" = {
366 source = cassandraConfFile;
367 };
368 environment.etc."cassandra/log4j-server.properties" = {
369 source = cassandraLogFile;
370 };
371 environment.etc."cassandra/cassandra-env.sh" = {
372 text = ''
373 ${builtins.readFile cfg.envFile}
374 ${concatStringsSep "\n" cfg.extraParams}
375 '';
376 };
377 systemd.services.cassandra = {
378 description = "Cassandra Daemon";
379 wantedBy = [ "multi-user.target" ];
380 after = [ "network-interfaces.target" ];
381 environment = cassandraEnvironment;
382 restartTriggers = [ cassandraConfFile cassandraLogFile cassandraRackFile ];
383 serviceConfig = {
384
385 User = cfg.user;
386 PermissionsStartOnly = true;
387 LimitAS = "infinity";
388 LimitNOFILE = "100000";
389 LimitNPROC = "32768";
390 LimitMEMLOCK = "infinity";
391
392 };
393 script = ''
394 ${cassandraPackage}/bin/cassandra -f
395 '';
396 path = [
397 cfg.jre
398 cassandraPackage
399 pkgs.coreutils
400 ];
401 preStart = ''
402 mkdir -m 0700 -p /etc/cassandra/triggers
403 mkdir -m 0700 -p /var/lib/cassandra /var/log/cassandra
404 chown ${cfg.user} /var/lib/cassandra /var/log/cassandra /etc/cassandra/triggers
405 '';
406 postStart = ''
407 sleep 2
408 while ! nodetool status >/dev/null 2>&1; do
409 sleep 2
410 done
411 nodetool status
412 '';
413 };
414
415 environment.systemPackages = [ cassandraPackage ];
416
417 networking.firewall.allowedTCPPorts = [
418 7000
419 7001
420 9042
421 9160
422 ];
423
424 users.extraUsers.cassandra =
425 if config.ids.uids ? "cassandra"
426 then { uid = config.ids.uids.cassandra; } // cassandraUser
427 else cassandraUser ;
428
429 boot.kernel.sysctl."vm.swappiness" = pkgs.lib.mkOptionDefault 0;
430
431 systemd.timers."cassandra-repair" = {
432 timerConfig = {
433 OnCalendar = "${toString cfg.repairStartAt}";
434 RandomizedDelaySec = cfg.repairRandomizedDelayInSec;
435 };
436 };
437
438 systemd.services."cassandra-repair" = {
439 description = "Cassandra repair daemon";
440 environment = cassandraEnvironment;
441 script = "${cassandraPackage}/bin/nodetool repair -pr";
442 postStop = mkIf (cfg.repairPostStop != null) cfg.repairPostStop;
443 postStart = mkIf (cfg.repairPostStart != null) cfg.repairPostStart;
444 serviceConfig = {
445 User = cfg.user;
446 };
447 };
448 };
449}