# This test is very comprehensive. It tests whether all hadoop services work well with each other. # Run this when updating the Hadoop package or making significant changes to the hadoop module. # For a more basic test, see hdfs.nix and yarn.nix import ../make-test-python.nix ({pkgs, ...}: { nodes = let package = pkgs.hadoop; coreSite = { "fs.defaultFS" = "hdfs://ns1"; }; hdfsSite = { "dfs.namenode.rpc-bind-host" = "0.0.0.0"; "dfs.namenode.http-bind-host" = "0.0.0.0"; "dfs.namenode.servicerpc-bind-host" = "0.0.0.0"; # HA Quorum Journal Manager configuration "dfs.nameservices" = "ns1"; "dfs.ha.namenodes.ns1" = "nn1,nn2"; "dfs.namenode.shared.edits.dir.ns1.nn1" = "qjournal://jn1:8485;jn2:8485;jn3:8485/ns1"; "dfs.namenode.shared.edits.dir.ns1.nn2" = "qjournal://jn1:8485;jn2:8485;jn3:8485/ns1"; "dfs.namenode.rpc-address.ns1.nn1" = "nn1:8020"; "dfs.namenode.rpc-address.ns1.nn2" = "nn2:8020"; "dfs.namenode.servicerpc-address.ns1.nn1" = "nn1:8022"; "dfs.namenode.servicerpc-address.ns1.nn2" = "nn2:8022"; "dfs.namenode.http-address.ns1.nn1" = "nn1:9870"; "dfs.namenode.http-address.ns1.nn2" = "nn2:9870"; # Automatic failover configuration "dfs.client.failover.proxy.provider.ns1" = "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider"; "dfs.ha.automatic-failover.enabled.ns1" = "true"; "dfs.ha.fencing.methods" = "shell(true)"; "ha.zookeeper.quorum" = "zk1:2181"; }; yarnSiteHA = { "yarn.resourcemanager.zk-address" = "zk1:2181"; "yarn.resourcemanager.ha.enabled" = "true"; "yarn.resourcemanager.ha.rm-ids" = "rm1,rm2"; "yarn.resourcemanager.hostname.rm1" = "rm1"; "yarn.resourcemanager.hostname.rm2" = "rm2"; "yarn.resourcemanager.ha.automatic-failover.enabled" = "true"; "yarn.resourcemanager.cluster-id" = "cluster1"; # yarn.resourcemanager.webapp.address needs to be defined even though yarn.resourcemanager.hostname is set. This shouldn't be necessary, but there's a bug in # hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-web-proxy/src/main/java/org/apache/hadoop/yarn/server/webproxy/amfilter/AmFilterInitializer.java:70 # that causes AM containers to fail otherwise. "yarn.resourcemanager.webapp.address.rm1" = "rm1:8088"; "yarn.resourcemanager.webapp.address.rm2" = "rm2:8088"; }; in { zk1 = { ... }: { services.zookeeper.enable = true; networking.firewall.allowedTCPPorts = [ 2181 ]; }; # HDFS cluster nn1 = {pkgs, options, ...}: { services.hadoop = { inherit package coreSite hdfsSite; hdfs.namenode.enable = true; hdfs.zkfc.enable = true; }; }; nn2 = {pkgs, options, ...}: { services.hadoop = { inherit package coreSite hdfsSite; hdfs.namenode.enable = true; hdfs.zkfc.enable = true; }; }; jn1 = {pkgs, options, ...}: { services.hadoop = { inherit package coreSite hdfsSite; hdfs.journalnode.enable = true; }; }; jn2 = {pkgs, options, ...}: { services.hadoop = { inherit package coreSite hdfsSite; hdfs.journalnode.enable = true; }; }; jn3 = {pkgs, options, ...}: { services.hadoop = { inherit package coreSite hdfsSite; hdfs.journalnode.enable = true; }; }; dn1 = {pkgs, options, ...}: { services.hadoop = { inherit package coreSite hdfsSite; hdfs.datanode.enable = true; }; }; # YARN cluster rm1 = {pkgs, options, ...}: { services.hadoop = { inherit package coreSite hdfsSite; yarnSite = options.services.hadoop.yarnSite.default // yarnSiteHA; yarn.resourcemanager.enable = true; }; }; rm2 = {pkgs, options, ...}: { services.hadoop = { inherit package coreSite hdfsSite; yarnSite = options.services.hadoop.yarnSite.default // yarnSiteHA; yarn.resourcemanager.enable = true; }; }; nm1 = {pkgs, options, ...}: { virtualisation.memorySize = 2048; services.hadoop = { inherit package coreSite hdfsSite; yarnSite = options.services.hadoop.yarnSite.default // yarnSiteHA; yarn.nodemanager.enable = true; }; }; }; testScript = '' start_all() #### HDFS tests #### zk1.wait_for_unit("network.target") jn1.wait_for_unit("network.target") jn2.wait_for_unit("network.target") jn3.wait_for_unit("network.target") nn1.wait_for_unit("network.target") nn2.wait_for_unit("network.target") dn1.wait_for_unit("network.target") zk1.wait_for_unit("zookeeper") jn1.wait_for_unit("hdfs-journalnode") jn2.wait_for_unit("hdfs-journalnode") jn3.wait_for_unit("hdfs-journalnode") zk1.wait_for_open_port(2181) jn1.wait_for_open_port(8480) jn1.wait_for_open_port(8485) jn2.wait_for_open_port(8480) jn2.wait_for_open_port(8485) # Namenodes must be stopped before initializing the cluster nn1.succeed("systemctl stop hdfs-namenode") nn2.succeed("systemctl stop hdfs-namenode") nn1.succeed("systemctl stop hdfs-zkfc") nn2.succeed("systemctl stop hdfs-zkfc") # Initialize zookeeper for failover controller nn1.succeed("sudo -u hdfs hdfs zkfc -formatZK 2>&1 | systemd-cat") # Format NN1 and start it nn1.succeed("sudo -u hdfs hadoop namenode -format 2>&1 | systemd-cat") nn1.succeed("systemctl start hdfs-namenode") nn1.wait_for_open_port(9870) nn1.wait_for_open_port(8022) nn1.wait_for_open_port(8020) # Bootstrap NN2 from NN1 and start it nn2.succeed("sudo -u hdfs hdfs namenode -bootstrapStandby 2>&1 | systemd-cat") nn2.succeed("systemctl start hdfs-namenode") nn2.wait_for_open_port(9870) nn2.wait_for_open_port(8022) nn2.wait_for_open_port(8020) nn1.succeed("netstat -tulpne | systemd-cat") # Start failover controllers nn1.succeed("systemctl start hdfs-zkfc") nn2.succeed("systemctl start hdfs-zkfc") # DN should have started by now, but confirm anyway dn1.wait_for_unit("hdfs-datanode") # Print states of namenodes dn1.succeed("sudo -u hdfs hdfs haadmin -getAllServiceState | systemd-cat") # Wait for cluster to exit safemode dn1.succeed("sudo -u hdfs hdfs dfsadmin -safemode wait") dn1.succeed("sudo -u hdfs hdfs haadmin -getAllServiceState | systemd-cat") # test R/W dn1.succeed("echo testfilecontents | sudo -u hdfs hdfs dfs -put - /testfile") assert "testfilecontents" in dn1.succeed("sudo -u hdfs hdfs dfs -cat /testfile") # Test NN failover nn1.succeed("systemctl stop hdfs-namenode") assert "active" in dn1.succeed("sudo -u hdfs hdfs haadmin -getAllServiceState") dn1.succeed("sudo -u hdfs hdfs haadmin -getAllServiceState | systemd-cat") assert "testfilecontents" in dn1.succeed("sudo -u hdfs hdfs dfs -cat /testfile") nn1.succeed("systemctl start hdfs-namenode") nn1.wait_for_open_port(9870) nn1.wait_for_open_port(8022) nn1.wait_for_open_port(8020) assert "standby" in dn1.succeed("sudo -u hdfs hdfs haadmin -getAllServiceState") dn1.succeed("sudo -u hdfs hdfs haadmin -getAllServiceState | systemd-cat") #### YARN tests #### rm1.wait_for_unit("network.target") rm2.wait_for_unit("network.target") nm1.wait_for_unit("network.target") rm1.wait_for_unit("yarn-resourcemanager") rm1.wait_for_open_port(8088) rm2.wait_for_unit("yarn-resourcemanager") rm2.wait_for_open_port(8088) nm1.wait_for_unit("yarn-nodemanager") nm1.wait_for_open_port(8042) nm1.wait_for_open_port(8040) nm1.wait_until_succeeds("yarn node -list | grep Nodes:1") nm1.succeed("sudo -u yarn yarn rmadmin -getAllServiceState | systemd-cat") nm1.succeed("sudo -u yarn yarn node -list | systemd-cat") # Test RM failover rm1.succeed("systemctl stop yarn-resourcemanager") assert "standby" not in nm1.succeed("sudo -u yarn yarn rmadmin -getAllServiceState") nm1.succeed("sudo -u yarn yarn rmadmin -getAllServiceState | systemd-cat") rm1.succeed("systemctl start yarn-resourcemanager") rm1.wait_for_unit("yarn-resourcemanager") rm1.wait_for_open_port(8088) assert "standby" in nm1.succeed("sudo -u yarn yarn rmadmin -getAllServiceState") nm1.succeed("sudo -u yarn yarn rmadmin -getAllServiceState | systemd-cat") assert "Estimated value of Pi is" in nm1.succeed("HADOOP_USER_NAME=hdfs yarn jar $(readlink $(which yarn) | sed -r 's~bin/yarn~lib/hadoop-*/share/hadoop/mapreduce/hadoop-mapreduce-examples-*.jar~g') pi 2 10") assert "SUCCEEDED" in nm1.succeed("yarn application -list -appStates FINISHED") ''; })