7544717b96
Previously, when reaping child processes, if there were no remaining reapable children, the parent strand would only nap 1, which puts unnecessary load on respirate unless at least one child strand exits in the next second. Change this approach by having the exiting child strands, after they release the lease, schedule their parent immediately if the parent has no non-exited child strands. When doing this, you need to be careful to make sure there are not race conditions that would delay the scheduling of the parent. There are two potential situations you need to handle: 1. Multiple children exiting at the same time 2. Parent currently running while child is exiting By waiting until after the child strand leases are released, you still have a race condition with 1, but the race condition is that multiple child strands exiting concurrently could both reschedule the parent strand. However, that isn't a problem. You want to avoid the case where neither child strand schedules the parent, which rescheduling after releasing the lease should do. To handle 2, inside reap use Model#lock! to lock the parent strand. This will make exiting child strands block if they UPDATE the parent strand with a new schedule, until the parent strand's transaction commits. However, it's possible that a child strand already UPDATED the parent. To handle this situation, before calling lock!, store the cached schedule value in a local variable. lock! implicitly does a reload, so compare the schedule value after reload. If the schedule has changed, likely a child scheduled the parent for immediate execution, so nap 0 in that case. Just in case there are unforeseen race conditions that are not handled, only nap for 120 seconds if there are active children. Worst case scenario, this results in a 2 minute delay for running the parent. However, this can potentially result in 120x less load from parent strands polling children.
332 lines
14 KiB
Ruby
332 lines
14 KiB
Ruby
# frozen_string_literal: true
|
|
|
|
require_relative "../../model/spec_helper"
|
|
|
|
RSpec.describe Prog::Kubernetes::ProvisionKubernetesNode do
|
|
subject(:prog) { described_class.new(st) }
|
|
|
|
let(:st) { Strand.new }
|
|
|
|
let(:project) {
|
|
Project.create(name: "default")
|
|
}
|
|
let(:subnet) {
|
|
Prog::Vnet::SubnetNexus.assemble(Config.kubernetes_service_project_id, name: "test", ipv4_range: "172.19.0.0/16", ipv6_range: "fd40:1a0a:8d48:182a::/64").subject
|
|
}
|
|
|
|
let(:vm) {
|
|
nic = Prog::Vnet::NicNexus.assemble(subnet.id, ipv4_addr: "172.19.145.64/26", ipv6_addr: "fd40:1a0a:8d48:182a::/79").subject
|
|
vm = Prog::Vm::Nexus.assemble("pub key", Config.kubernetes_service_project_id, name: "test-vm", private_subnet_id: subnet.id, nic_id: nic.id).subject
|
|
vm.update(ephemeral_net6: "2001:db8:85a3:73f2:1c4a::/79")
|
|
}
|
|
|
|
let(:kubernetes_cluster) {
|
|
kc = KubernetesCluster.create(
|
|
name: "k8scluster",
|
|
version: "v1.32",
|
|
cp_node_count: 3,
|
|
private_subnet_id: subnet.id,
|
|
location_id: Location::HETZNER_FSN1_ID,
|
|
project_id: project.id,
|
|
target_node_size: "standard-4",
|
|
target_node_storage_size_gib: 37
|
|
)
|
|
|
|
lb = LoadBalancer.create(private_subnet_id: subnet.id, name: "somelb", health_check_endpoint: "/foo", project_id: Config.kubernetes_service_project_id)
|
|
LoadBalancerPort.create(load_balancer_id: lb.id, src_port: 123, dst_port: 456)
|
|
kc.add_cp_vm(vm)
|
|
kc.add_cp_vm(create_vm)
|
|
|
|
kc.update(api_server_lb_id: lb.id)
|
|
kc
|
|
}
|
|
|
|
let(:kubernetes_nodepool) { KubernetesNodepool.create(name: "k8stest-np", node_count: 2, kubernetes_cluster_id: kubernetes_cluster.id, target_node_size: "standard-8", target_node_storage_size_gib: 78) }
|
|
|
|
before do
|
|
allow(Config).to receive(:kubernetes_service_project_id).and_return(Project.create(name: "UbicloudKubernetesService").id)
|
|
allow(prog).to receive_messages(kubernetes_cluster: kubernetes_cluster, frame: {"vm_id" => vm.id})
|
|
end
|
|
|
|
describe "random_ula_cidr" do
|
|
it "returns a /108 subnet" do
|
|
cidr = prog.random_ula_cidr
|
|
expect(cidr.netmask.prefix_len).to eq(108)
|
|
end
|
|
|
|
it "returns an address in the fd00::/8 range" do
|
|
cidr = prog.random_ula_cidr
|
|
ula_range = NetAddr::IPv6Net.parse("fd00::/8")
|
|
expect(ula_range.cmp(cidr)).to be(-1)
|
|
end
|
|
end
|
|
|
|
describe "#before_run" do
|
|
it "destroys itself if the kubernetes cluster is getting deleted" do
|
|
Strand.create(id: kubernetes_cluster.id, label: "something", prog: "KubernetesClusterNexus")
|
|
kubernetes_cluster.reload
|
|
expect(kubernetes_cluster.strand.label).to eq("something")
|
|
prog.before_run # Nothing happens
|
|
|
|
kubernetes_cluster.strand.label = "destroy"
|
|
expect { prog.before_run }.to exit({"msg" => "provisioning canceled"})
|
|
|
|
prog.strand.label = "destroy"
|
|
prog.before_run # Nothing happens
|
|
end
|
|
end
|
|
|
|
describe "#start" do
|
|
it "creates a CP VM and hops if a nodepool is not given" do
|
|
expect(prog.kubernetes_nodepool).to be_nil
|
|
expect(kubernetes_cluster.cp_vms.count).to eq(2)
|
|
expect(kubernetes_cluster.api_server_lb).to receive(:add_vm)
|
|
|
|
expect { prog.start }.to hop("bootstrap_rhizome")
|
|
|
|
expect(kubernetes_cluster.cp_vms.count).to eq(3)
|
|
|
|
new_vm = kubernetes_cluster.cp_vms.last
|
|
expect(new_vm.name).to start_with("#{kubernetes_cluster.ubid}-")
|
|
expect(new_vm.sshable).not_to be_nil
|
|
expect(new_vm.vcpus).to eq(4)
|
|
expect(new_vm.strand.stack.first["storage_volumes"].first["size_gib"]).to eq(37)
|
|
expect(new_vm.boot_image).to eq("kubernetes-v1_32")
|
|
end
|
|
|
|
it "creates a worker VM and hops if a nodepool is given" do
|
|
expect(prog).to receive(:frame).and_return({"nodepool_id" => kubernetes_nodepool.id})
|
|
expect(kubernetes_nodepool.vms.count).to eq(0)
|
|
|
|
expect { prog.start }.to hop("bootstrap_rhizome")
|
|
|
|
expect(kubernetes_nodepool.reload.vms.count).to eq(1)
|
|
|
|
new_vm = kubernetes_nodepool.vms.last
|
|
expect(new_vm.name).to start_with("#{kubernetes_nodepool.ubid}-")
|
|
expect(new_vm.sshable).not_to be_nil
|
|
expect(new_vm.vcpus).to eq(8)
|
|
expect(new_vm.strand.stack.first["storage_volumes"].first["size_gib"]).to eq(78)
|
|
expect(new_vm.boot_image).to eq("kubernetes-v1_32")
|
|
end
|
|
|
|
it "assigns the default storage size if not specified" do
|
|
kubernetes_cluster.update(target_node_storage_size_gib: nil)
|
|
expect(kubernetes_cluster.api_server_lb).to receive(:add_vm)
|
|
|
|
expect(kubernetes_cluster.cp_vms.count).to eq(2)
|
|
|
|
expect { prog.start }.to hop("bootstrap_rhizome")
|
|
|
|
expect(kubernetes_cluster.cp_vms.count).to eq(3)
|
|
|
|
new_vm = kubernetes_cluster.cp_vms.last
|
|
expect(new_vm.strand.stack.first["storage_volumes"].first["size_gib"]).to eq 80
|
|
end
|
|
end
|
|
|
|
describe "#bootstrap_rhizome" do
|
|
it "waits until the VM is ready" do
|
|
st = instance_double(Strand, label: "non-wait")
|
|
expect(prog.vm).to receive(:strand).and_return(st)
|
|
expect { prog.bootstrap_rhizome }.to nap(5)
|
|
end
|
|
|
|
it "enables kubelet and buds a bootstrap rhizome process" do
|
|
sshable = instance_double(Sshable)
|
|
st = instance_double(Strand, label: "wait")
|
|
expect(prog.vm).to receive_messages(sshable: sshable, strand: st)
|
|
expect(sshable).to receive(:cmd).with("sudo iptables-nft -t nat -A POSTROUTING -s 172.19.145.64/26 -o ens3 -j MASQUERADE")
|
|
expect(sshable).to receive(:cmd).with(
|
|
"sudo nft --file -",
|
|
stdin: <<~TEMPLATE
|
|
table ip6 pod_access;
|
|
delete table ip6 pod_access;
|
|
table ip6 pod_access {
|
|
chain ingress_egress_control {
|
|
type filter hook forward priority filter; policy drop;
|
|
# allow access to the vm itself in order to not break the normal functionality of Clover and SSH
|
|
ip6 daddr 2001:db8:85a3:73f2:1c4a::2 ct state established,related,new counter accept
|
|
ip6 saddr 2001:db8:85a3:73f2:1c4a::2 ct state established,related,new counter accept
|
|
|
|
# not allow new connections from internet but allow new connections from inside
|
|
ip6 daddr 2001:db8:85a3:73f2:1c4a::/79 ct state established,related counter accept
|
|
ip6 saddr 2001:db8:85a3:73f2:1c4a::/79 ct state established,related,new counter accept
|
|
|
|
# used for internal private ipv6 communication between pods
|
|
ip6 saddr fd40:1a0a:8d48:182a::/64 ct state established,related,new counter accept
|
|
ip6 daddr fd40:1a0a:8d48:182a::/64 ct state established,related,new counter accept
|
|
}
|
|
}
|
|
TEMPLATE
|
|
)
|
|
expect(sshable).to receive(:cmd).with("sudo systemctl enable --now kubelet")
|
|
|
|
expect(prog).to receive(:bud).with(Prog::BootstrapRhizome, {"target_folder" => "kubernetes", "subject_id" => prog.vm.id, "user" => "ubi"})
|
|
expect { prog.bootstrap_rhizome }.to hop("wait_bootstrap_rhizome")
|
|
end
|
|
end
|
|
|
|
describe "#wait_bootstrap_rhizome" do
|
|
it "hops to assign_role if there are no sub-programs running" do
|
|
st.update(prog: "Kubernetes::ProvisionKubernetesNode", label: "wait_bootstrap_rhizome", stack: [{}])
|
|
expect { prog.wait_bootstrap_rhizome }.to hop("assign_role")
|
|
end
|
|
|
|
it "donates if there are sub-programs running" do
|
|
st.update(prog: "Kubernetes::ProvisionKubernetesNode", label: "wait_bootstrap_rhizome", stack: [{}])
|
|
Strand.create(parent_id: st.id, prog: "BootstrapRhizome", label: "start", stack: [{}], lease: Time.now + 10)
|
|
expect { prog.wait_bootstrap_rhizome }.to nap(120)
|
|
end
|
|
end
|
|
|
|
describe "#assign_role" do
|
|
it "hops to init_cluster if this is the first vm of the cluster" do
|
|
expect(prog.kubernetes_cluster.cp_vms).to receive(:count).and_return(1)
|
|
expect { prog.assign_role }.to hop("init_cluster")
|
|
end
|
|
|
|
it "hops to join_control_plane if this is the not the first vm of the cluster" do
|
|
expect(prog.kubernetes_cluster.cp_vms.count).to eq(2)
|
|
expect { prog.assign_role }.to hop("join_control_plane")
|
|
end
|
|
|
|
it "hops to join_worker if a nodepool is specified to the prog" do
|
|
expect(prog).to receive(:kubernetes_nodepool).and_return(kubernetes_nodepool)
|
|
expect { prog.assign_role }.to hop("join_worker")
|
|
end
|
|
end
|
|
|
|
describe "#init_cluster" do
|
|
before { allow(prog.vm).to receive(:sshable).and_return(instance_double(Sshable)) }
|
|
|
|
it "runs the init_cluster script if it's not started" do
|
|
expect(prog.vm.sshable).to receive(:d_check).with("init_kubernetes_cluster").and_return("NotStarted")
|
|
expect(prog.vm.sshable).to receive(:d_run).with(
|
|
"init_kubernetes_cluster", "/home/ubi/kubernetes/bin/init-cluster",
|
|
stdin: /{"node_name":"test-vm","cluster_name":"k8scluster","lb_hostname":"somelb\..*","port":"443","private_subnet_cidr4":"172.19.0.0\/16","private_subnet_cidr6":"fd40:1a0a:8d48:182a::\/64","node_ipv4":"172.19.145.65","node_ipv6":"2001:db8:85a3:73f2:1c4a::2"/, log: false
|
|
)
|
|
|
|
expect { prog.init_cluster }.to nap(30)
|
|
end
|
|
|
|
it "naps if the init_cluster script is in progress" do
|
|
expect(prog.vm.sshable).to receive(:d_check).with("init_kubernetes_cluster").and_return("InProgress")
|
|
expect { prog.init_cluster }.to nap(10)
|
|
end
|
|
|
|
it "naps and does nothing (for now) if the init_cluster script is failed" do
|
|
expect(prog.vm.sshable).to receive(:d_check).with("init_kubernetes_cluster").and_return("Failed")
|
|
expect { prog.init_cluster }.to nap(65536)
|
|
end
|
|
|
|
it "pops if the init_cluster script is successful" do
|
|
expect(prog.vm.sshable).to receive(:d_check).with("init_kubernetes_cluster").and_return("Succeeded")
|
|
expect { prog.init_cluster }.to hop("install_cni")
|
|
end
|
|
|
|
it "naps forever if the daemonizer check returns something unknown" do
|
|
expect(prog.vm.sshable).to receive(:d_check).with("init_kubernetes_cluster").and_return("Unknown")
|
|
expect { prog.init_cluster }.to nap(65536)
|
|
end
|
|
end
|
|
|
|
describe "#join_control_plane" do
|
|
before { allow(prog.vm).to receive(:sshable).and_return(instance_double(Sshable)) }
|
|
|
|
it "runs the join_control_plane script if it's not started" do
|
|
expect(prog.vm.sshable).to receive(:d_check).with("join_control_plane").and_return("NotStarted")
|
|
|
|
sshable = instance_double(Sshable)
|
|
allow(kubernetes_cluster.cp_vms.first).to receive(:sshable).and_return(sshable)
|
|
expect(sshable).to receive(:cmd).with("sudo kubeadm token create --ttl 24h --usages signing,authentication", log: false).and_return("jt\n")
|
|
expect(sshable).to receive(:cmd).with("sudo kubeadm init phase upload-certs --upload-certs", log: false).and_return("something\ncertificate key:\nck")
|
|
expect(sshable).to receive(:cmd).with("sudo kubeadm token create --print-join-command", log: false).and_return("discovery-token-ca-cert-hash dtcch")
|
|
expect(prog.vm.sshable).to receive(:d_run).with(
|
|
"join_control_plane", "kubernetes/bin/join-node",
|
|
stdin: /{"is_control_plane":true,"node_name":"test-vm","endpoint":"somelb\..*:443","join_token":"jt","certificate_key":"ck","discovery_token_ca_cert_hash":"dtcch","node_ipv4":"172.19.145.65","node_ipv6":"2001:db8:85a3:73f2:1c4a::2"}/,
|
|
log: false
|
|
)
|
|
|
|
expect { prog.join_control_plane }.to nap(15)
|
|
end
|
|
|
|
it "naps if the join_control_plane script is in progress" do
|
|
expect(prog.vm.sshable).to receive(:d_check).with("join_control_plane").and_return("InProgress")
|
|
expect { prog.join_control_plane }.to nap(10)
|
|
end
|
|
|
|
it "naps and does nothing (for now) if the join_control_plane script is failed" do
|
|
expect(prog.vm.sshable).to receive(:d_check).with("join_control_plane").and_return("Failed")
|
|
expect { prog.join_control_plane }.to nap(65536)
|
|
end
|
|
|
|
it "pops if the join_control_plane script is successful" do
|
|
expect(prog.vm.sshable).to receive(:d_check).with("join_control_plane").and_return("Succeeded")
|
|
expect { prog.join_control_plane }.to hop("install_cni")
|
|
end
|
|
|
|
it "naps forever if the daemonizer check returns something unknown" do
|
|
expect(prog.vm.sshable).to receive(:d_check).with("join_control_plane").and_return("Unknown")
|
|
expect { prog.join_control_plane }.to nap(65536)
|
|
end
|
|
end
|
|
|
|
describe "#join_worker" do
|
|
before {
|
|
allow(prog.vm).to receive(:sshable).and_return(instance_double(Sshable))
|
|
allow(prog).to receive(:kubernetes_nodepool).and_return(kubernetes_nodepool)
|
|
}
|
|
|
|
it "runs the join-worker-node script if it's not started" do
|
|
expect(prog.vm.sshable).to receive(:d_check).with("join_worker").and_return("NotStarted")
|
|
|
|
sshable = instance_double(Sshable)
|
|
allow(kubernetes_cluster.cp_vms.first).to receive(:sshable).and_return(sshable)
|
|
expect(sshable).to receive(:cmd).with("sudo kubeadm token create --ttl 24h --usages signing,authentication", log: false).and_return("\njt\n")
|
|
expect(sshable).to receive(:cmd).with("sudo kubeadm token create --print-join-command", log: false).and_return("discovery-token-ca-cert-hash dtcch")
|
|
expect(prog.vm.sshable).to receive(:d_run).with(
|
|
"join_worker", "kubernetes/bin/join-node",
|
|
stdin: /{"is_control_plane":false,"node_name":"test-vm","endpoint":"somelb\..*:443","join_token":"jt","discovery_token_ca_cert_hash":"dtcch","node_ipv4":"172.19.145.65","node_ipv6":"2001:db8:85a3:73f2:1c4a::2"}/,
|
|
log: false
|
|
)
|
|
|
|
expect { prog.join_worker }.to nap(15)
|
|
end
|
|
|
|
it "naps if the join-worker-node script is in progress" do
|
|
expect(prog.vm.sshable).to receive(:d_check).with("join_worker").and_return("InProgress")
|
|
expect { prog.join_worker }.to nap(10)
|
|
end
|
|
|
|
it "naps and does nothing (for now) if the join-worker-node script is failed" do
|
|
expect(prog.vm.sshable).to receive(:d_check).with("join_worker").and_return("Failed")
|
|
expect { prog.join_worker }.to nap(65536)
|
|
end
|
|
|
|
it "pops if the join-worker-node script is successful" do
|
|
expect(prog.vm.sshable).to receive(:d_check).with("join_worker").and_return("Succeeded")
|
|
expect { prog.join_worker }.to hop("install_cni")
|
|
end
|
|
|
|
it "naps for a long time if the daemonizer check returns something unknown" do
|
|
expect(prog.vm.sshable).to receive(:d_check).with("join_worker").and_return("Unknown")
|
|
expect { prog.join_worker }.to nap(65536)
|
|
end
|
|
end
|
|
|
|
describe "#install_cni" do
|
|
it "configures ubicni" do
|
|
sshable = instance_double(Sshable)
|
|
allow(prog.vm).to receive_messages(
|
|
sshable: sshable,
|
|
nics: [instance_double(Nic, private_ipv4: "10.0.0.37", private_ipv6: "0::1")],
|
|
ephemeral_net6: NetAddr::IPv6Net.new(NetAddr::IPv6.parse("2001:db8::"), NetAddr::Mask128.new(64))
|
|
)
|
|
|
|
expect(sshable).to receive(:cmd).with("sudo tee /etc/cni/net.d/ubicni-config.json", stdin: /"type": "ubicni"/)
|
|
expect { prog.install_cni }.to exit({vm_id: prog.vm.id})
|
|
end
|
|
end
|
|
end
|