cs-codex-dist-tests/Framework/KubernetesWorkflow/K8sController.cs

860 lines
32 KiB
C#

using k8s;
using k8s.Models;
using KubernetesWorkflow.Recipe;
using KubernetesWorkflow.Types;
using Logging;
using Utils;
namespace KubernetesWorkflow
{
public class K8sController
{
private readonly ILog log;
private readonly K8sCluster cluster;
private readonly WorkflowNumberSource workflowNumberSource;
private readonly K8sClient client;
public const string PodLabelKey = "pod-uuid";
public K8sController(ILog log, K8sCluster cluster, WorkflowNumberSource workflowNumberSource, string k8sNamespace)
{
this.log = log;
this.cluster = cluster;
this.workflowNumberSource = workflowNumberSource;
client = new K8sClient(cluster.GetK8sClientConfig());
K8sNamespace = k8sNamespace;
}
public void Dispose()
{
client.Dispose();
}
public StartResult BringOnline(ContainerRecipe[] containerRecipes, ILocation location)
{
log.Debug();
EnsureNamespace();
var podLabel = K8sNameUtils.Format(Guid.NewGuid().ToString());
var deployment = CreateDeployment(containerRecipes, location, podLabel);
var internalService = CreateInternalService(containerRecipes);
var externalService = CreateExternalService(containerRecipes);
return new StartResult(cluster, containerRecipes, deployment, internalService, externalService);
}
public PodInfo GetPodInfo(RunningDeployment deployment)
{
var pod = GetPodForDeployment(deployment);
return CreatePodInfo(pod);
}
public void Stop(StartResult startResult)
{
log.Debug();
if (startResult.InternalService != null) DeleteService(startResult.InternalService);
if (startResult.ExternalService != null) DeleteService(startResult.ExternalService);
DeleteDeployment(startResult.Deployment);
WaitUntilPodsForDeploymentAreOffline(startResult.Deployment);
}
public void DownloadPodLog(RunningContainer container, ILogHandler logHandler, int? tailLines)
{
log.Debug();
var podName = GetPodName(container);
var recipeName = container.Recipe.Name;
using var stream = client.Run(c => c.ReadNamespacedPodLog(podName, K8sNamespace, recipeName, tailLines: tailLines));
logHandler.Log(stream);
}
public string ExecuteCommand(RunningContainer container, string command, params string[] args)
{
var containerName = container.Recipe.Name;
var cmdAndArgs = $"{containerName}: {command} ({string.Join(",", args)})";
log.Debug(cmdAndArgs);
var podName = GetPodName(container);
var runner = new CommandRunner(client, K8sNamespace, podName, containerName, command, args);
runner.Run();
var result = runner.GetStdOut();
log.Debug($"{cmdAndArgs} = '{result}'");
return result;
}
public void DeleteAllNamespacesStartingWith(string prefix)
{
log.Debug();
var all = client.Run(c => c.ListNamespace().Items);
var namespaces = all.Select(n => n.Name()).Where(n => n.StartsWith(prefix));
foreach (var ns in namespaces)
{
DeleteNamespace(ns);
}
}
public void DeleteNamespace()
{
log.Debug();
if (IsNamespaceOnline(K8sNamespace))
{
client.Run(c => c.DeleteNamespace(K8sNamespace, null, null, gracePeriodSeconds: 0));
}
}
public void DeleteNamespace(string ns)
{
log.Debug();
if (IsNamespaceOnline(ns))
{
client.Run(c => c.DeleteNamespace(ns, null, null, gracePeriodSeconds: 0));
}
}
#region Discover K8s Nodes
public K8sNodeLabel[] GetAvailableK8sNodes()
{
var nodes = client.Run(c => c.ListNode());
var optionals = nodes.Items.Select(i => CreateNodeLabel(i));
return optionals.Where(n => n != null).Select(n => n!).ToArray();
}
private K8sNodeLabel? CreateNodeLabel(V1Node i)
{
var keys = i.Metadata.Labels.Keys;
var hostnameKey = keys.SingleOrDefault(k => k.ToLowerInvariant().Contains("hostname"));
if (hostnameKey != null)
{
var hostnameValue = i.Metadata.Labels[hostnameKey];
return new K8sNodeLabel(hostnameKey, hostnameValue);
}
return null;
}
#endregion
#region Namespace management
private string K8sNamespace { get; }
private void EnsureNamespace()
{
if (IsNamespaceOnline(K8sNamespace)) return;
var namespaceSpec = new V1Namespace
{
ApiVersion = "v1",
Metadata = new V1ObjectMeta
{
Name = K8sNamespace,
Labels = new Dictionary<string, string> { { "name", K8sNamespace } }
}
};
client.Run(c => c.CreateNamespace(namespaceSpec));
WaitUntilNamespaceCreated();
CreatePolicy();
}
private bool IsNamespaceOnline(string name)
{
return client.Run(c => c.ListNamespace().Items.Any(n => n.Metadata.Name == name));
}
private void CreatePolicy()
{
client.Run(c =>
{
var body = new V1NetworkPolicy
{
Metadata = new V1ObjectMeta
{
Name = "isolate-policy",
NamespaceProperty = K8sNamespace
},
Spec = new V1NetworkPolicySpec
{
PodSelector = new V1LabelSelector {},
PolicyTypes = new[]
{
"Ingress",
"Egress"
},
Ingress = new List<V1NetworkPolicyIngressRule>
{
new V1NetworkPolicyIngressRule
{
FromProperty = new List<V1NetworkPolicyPeer>
{
new V1NetworkPolicyPeer
{
PodSelector = new V1LabelSelector {}
}
}
},
new V1NetworkPolicyIngressRule
{
FromProperty = new List<V1NetworkPolicyPeer>
{
new V1NetworkPolicyPeer
{
NamespaceSelector = new V1LabelSelector
{
MatchLabels = GetRunnerNamespaceSelector()
}
}
}
},
new V1NetworkPolicyIngressRule
{
FromProperty = new List<V1NetworkPolicyPeer>
{
new V1NetworkPolicyPeer
{
NamespaceSelector = new V1LabelSelector
{
MatchLabels = GetPrometheusNamespaceSelector()
}
}
}
}
},
Egress = new List<V1NetworkPolicyEgressRule>
{
new V1NetworkPolicyEgressRule
{
To = new List<V1NetworkPolicyPeer>
{
new V1NetworkPolicyPeer
{
PodSelector = new V1LabelSelector {}
}
}
},
new V1NetworkPolicyEgressRule
{
To = new List<V1NetworkPolicyPeer>
{
new V1NetworkPolicyPeer
{
NamespaceSelector = new V1LabelSelector
{
MatchLabels = new Dictionary<string, string> { { "kubernetes.io/metadata.name", "kube-system" } }
}
},
new V1NetworkPolicyPeer
{
PodSelector = new V1LabelSelector
{
MatchLabels = new Dictionary<string, string> { { "k8s-app", "kube-dns" } }
}
}
},
Ports = new List<V1NetworkPolicyPort>
{
new V1NetworkPolicyPort
{
Port = new IntstrIntOrString
{
Value = "53"
},
Protocol = "UDP"
}
}
},
new V1NetworkPolicyEgressRule
{
To = new List<V1NetworkPolicyPeer>
{
new V1NetworkPolicyPeer
{
IpBlock = new V1IPBlock
{
Cidr = "0.0.0.0/0"
}
}
},
Ports = new List<V1NetworkPolicyPort>
{
new V1NetworkPolicyPort
{
Port = new IntstrIntOrString
{
Value = "80"
},
Protocol = "TCP"
},
new V1NetworkPolicyPort
{
Port = new IntstrIntOrString
{
Value = "443"
},
Protocol = "TCP"
}
}
}
}
}
};
c.CreateNamespacedNetworkPolicy(body, K8sNamespace);
});
}
#endregion
#region Deployment management
private RunningDeployment CreateDeployment(ContainerRecipe[] containerRecipes, ILocation location, string podLabel)
{
var deploymentSpec = new V1Deployment
{
ApiVersion = "apps/v1",
Metadata = CreateDeploymentMetadata(containerRecipes),
Spec = new V1DeploymentSpec
{
Replicas = 1,
Selector = new V1LabelSelector
{
MatchLabels = GetSelector(containerRecipes)
},
Template = new V1PodTemplateSpec
{
Metadata = new V1ObjectMeta
{
Labels = GetSelector(containerRecipes, podLabel),
Annotations = GetAnnotations(containerRecipes)
},
Spec = new V1PodSpec
{
NodeSelector = CreateNodeSelector(location),
Containers = CreateDeploymentContainers(containerRecipes),
Volumes = CreateVolumes(containerRecipes)
}
}
}
};
client.Run(c => c.CreateNamespacedDeployment(deploymentSpec, K8sNamespace));
WaitUntilDeploymentOnline(deploymentSpec.Metadata.Name);
var name = deploymentSpec.Metadata.Name;
return new RunningDeployment(name, podLabel);
}
private void DeleteDeployment(RunningDeployment deployment)
{
client.Run(c => c.DeleteNamespacedDeployment(deployment.Name, K8sNamespace));
WaitUntilDeploymentOffline(deployment.Name);
}
private IDictionary<string, string> CreateNodeSelector(ILocation location)
{
var nodeLabel = GetNodeLabelForLocation(location);
if (nodeLabel == null) return new Dictionary<string, string>();
return new Dictionary<string, string>
{
{ nodeLabel.Key, nodeLabel.Value }
};
}
private K8sNodeLabel? GetNodeLabelForLocation(ILocation location)
{
var l = (Location)location;
return l.NodeLabel;
}
private IDictionary<string, string> GetSelector(ContainerRecipe[] containerRecipes)
{
return containerRecipes.First().PodLabels.GetLabels();
}
private IDictionary<string, string> GetSelector(ContainerRecipe[] containerRecipes, string podLabel)
{
var labels = containerRecipes.First().PodLabels.Clone();
labels.Add(PodLabelKey, podLabel);
return labels.GetLabels();
}
private IDictionary<string, string> GetRunnerNamespaceSelector()
{
return new Dictionary<string, string> { { "kubernetes.io/metadata.name", "default" } };
}
private IDictionary<string, string> GetPrometheusNamespaceSelector()
{
return new Dictionary<string, string> { { "kubernetes.io/metadata.name", "monitoring" } };
}
private IDictionary<string, string> GetAnnotations(ContainerRecipe[] containerRecipes)
{
return containerRecipes.First().PodAnnotations.GetAnnotations();
}
private V1ObjectMeta CreateDeploymentMetadata(ContainerRecipe[] containerRecipes)
{
return new V1ObjectMeta
{
Name = string.Join('-',containerRecipes.Select(r => r.Name)),
NamespaceProperty = K8sNamespace,
Labels = GetSelector(containerRecipes),
Annotations = GetAnnotations(containerRecipes)
};
}
private List<V1Container> CreateDeploymentContainers(ContainerRecipe[] containerRecipes)
{
return containerRecipes.Select(CreateDeploymentContainer).ToList();
}
private V1Container CreateDeploymentContainer(ContainerRecipe recipe)
{
return new V1Container
{
Name = recipe.Name,
Image = recipe.Image,
ImagePullPolicy = "Always",
Ports = CreateContainerPorts(recipe),
Env = CreateEnv(recipe),
VolumeMounts = CreateContainerVolumeMounts(recipe),
Resources = CreateResourceLimits(recipe)
};
}
private V1ResourceRequirements CreateResourceLimits(ContainerRecipe recipe)
{
return new V1ResourceRequirements
{
Requests = CreateResourceQuantities(recipe.Resources.Requests),
Limits = CreateResourceQuantities(recipe.Resources.Limits)
};
}
private Dictionary<string, ResourceQuantity> CreateResourceQuantities(ContainerResourceSet set)
{
var result = new Dictionary<string, ResourceQuantity>();
if (set.MilliCPUs != 0)
{
result.Add("cpu", new ResourceQuantity($"{set.MilliCPUs}m"));
}
if (set.Memory.SizeInBytes != 0)
{
result.Add("memory", new ResourceQuantity(set.Memory.ToSuffixNotation()));
}
return result;
}
private List<V1VolumeMount> CreateContainerVolumeMounts(ContainerRecipe recipe)
{
return recipe.Volumes.Select(CreateContainerVolumeMount).ToList();
}
private V1VolumeMount CreateContainerVolumeMount(VolumeMount v)
{
return new V1VolumeMount
{
Name = v.VolumeName,
MountPath = v.MountPath,
SubPath = v.SubPath,
};
}
private List<V1Volume> CreateVolumes(ContainerRecipe[] containerRecipes)
{
return containerRecipes.Where(c => c.Volumes.Any()).SelectMany(CreateVolumes).ToList();
}
private List<V1Volume> CreateVolumes(ContainerRecipe recipe)
{
return recipe.Volumes.Select(CreateVolume).ToList();
}
private V1Volume CreateVolume(VolumeMount v)
{
CreatePersistentVolumeClaimIfNeeded(v);
if (!string.IsNullOrEmpty(v.HostPath))
{
return new V1Volume
{
Name = v.VolumeName,
HostPath = new V1HostPathVolumeSource
{
Path = v.HostPath
}
};
}
if (!string.IsNullOrEmpty(v.Secret))
{
return new V1Volume
{
Name = v.VolumeName,
Secret = CreateVolumeSecret(v)
};
}
return new V1Volume
{
Name = v.VolumeName,
PersistentVolumeClaim = new V1PersistentVolumeClaimVolumeSource
{
ClaimName = v.VolumeName
}
};
}
private void CreatePersistentVolumeClaimIfNeeded(VolumeMount v)
{
var pvcs = client.Run(c => c.ListNamespacedPersistentVolumeClaim(K8sNamespace));
if (pvcs != null && pvcs.Items.Any(i => i.Name() == v.VolumeName)) return;
client.Run(c => c.CreateNamespacedPersistentVolumeClaim(new V1PersistentVolumeClaim
{
ApiVersion = "v1",
Metadata = new V1ObjectMeta
{
Name = v.VolumeName,
},
Spec = new V1PersistentVolumeClaimSpec
{
AccessModes = new List<string>
{
"ReadWriteOnce"
},
Resources = CreateVolumeResourceRequirements(v),
},
}, K8sNamespace));
}
private V1SecretVolumeSource CreateVolumeSecret(VolumeMount v)
{
if (string.IsNullOrWhiteSpace(v.Secret)) return null!;
return new V1SecretVolumeSource
{
SecretName = v.Secret
};
}
private V1ResourceRequirements CreateVolumeResourceRequirements(VolumeMount v)
{
if (v.ResourceQuantity == null) return null!;
return new V1ResourceRequirements
{
Requests = new Dictionary<string, ResourceQuantity>()
{
{"storage", new ResourceQuantity(v.ResourceQuantity) }
}
};
}
private List<V1EnvVar> CreateEnv(ContainerRecipe recipe)
{
return recipe.EnvVars.Select(CreateEnvVar).ToList();
}
private V1EnvVar CreateEnvVar(EnvVar envVar)
{
return new V1EnvVar
{
Name = envVar.Name,
Value = envVar.Value,
};
}
private List<V1ContainerPort> CreateContainerPorts(ContainerRecipe recipe)
{
var exposedPorts = recipe.ExposedPorts.SelectMany(p => CreateContainerPort(recipe, p));
var internalPorts = recipe.InternalPorts.SelectMany(p => CreateContainerPort(recipe, p));
return exposedPorts.Concat(internalPorts).ToList();
}
private List<V1ContainerPort> CreateContainerPort(ContainerRecipe recipe, Port port)
{
var result = new List<V1ContainerPort>();
if (port.IsTcp()) CreateTcpContainerPort(result, recipe, port);
if (port.IsUdp()) CreateUdpContainerPort(result, recipe, port);
return result;
}
private void CreateUdpContainerPort(List<V1ContainerPort> result, ContainerRecipe recipe, Port port)
{
result.Add(CreateContainerPort(recipe, port, "UDP"));
}
private void CreateTcpContainerPort(List<V1ContainerPort> result, ContainerRecipe recipe, Port port)
{
result.Add(CreateContainerPort(recipe, port, "TCP"));
}
private V1ContainerPort CreateContainerPort(ContainerRecipe recipe, Port port, string protocol)
{
return new V1ContainerPort
{
Name = GetNameForPort(recipe, port),
ContainerPort = port.Number,
Protocol = protocol
};
}
private string GetNameForPort(ContainerRecipe recipe, Port port)
{
var inputs = new[]
{
$"p{workflowNumberSource.WorkflowNumber}",
recipe.Number.ToString(),
port.Number.ToString(),
port.Protocol.ToString().ToLowerInvariant()
};
return K8sNameUtils.FormatPortName(string.Join(",", inputs));
}
private string GetPodName(RunningContainer container)
{
return GetPodForDeployment(container.RunningContainers.StartResult.Deployment).Metadata.Name;
}
private V1Pod GetPodForDeployment(RunningDeployment deployment)
{
return Time.Retry(() => GetPodForDeplomentInternal(deployment),
maxRetries: 2,
retryTime: TimeSpan.FromSeconds(10),
description: "Find pod by label for deployment.");
}
private V1Pod GetPodForDeplomentInternal(RunningDeployment deployment)
{
var allPods = client.Run(c => c.ListNamespacedPod(K8sNamespace));
var pods = allPods.Items.Where(p => p.GetLabel(PodLabelKey) == deployment.PodLabel).ToArray();
if (pods.Length != 1)
{
var allLabels = allPods.Items.Select(p =>
{
var labels = string.Join(",", p.Labels().Select(l => $"{l.Key}={l.Value}"));
return $"pod:'{p.Name()}' has labels: [{labels}]";
});
throw new Exception($"Expected to find 1 pod by podLabel '{deployment.PodLabel}'. Found: {pods.Length}. " +
$"Total number of pods: {allPods.Items.Count}. Their labels: {string.Join(Environment.NewLine, allLabels)}");
}
return pods[0];
}
#endregion
#region Service management
private RunningService? CreateInternalService(ContainerRecipe[] recipes)
{
return CreateService(recipes, r => r.InternalPorts.Concat(r.ExposedPorts).ToArray(), "ClusterIP", "int", false);
}
private RunningService? CreateExternalService(ContainerRecipe[] recipes)
{
return CreateService(recipes, r => r.ExposedPorts, "NodePort", "ext", true);
}
private RunningService? CreateService(ContainerRecipe[] recipes, Func<ContainerRecipe, Port[]> portSelector, string serviceType, string namePostfix, bool isNodePort)
{
var ports = CreateServicePorts(recipes, portSelector, isNodePort);
if (!ports.Any()) return null;
var serviceSpec = new V1Service
{
ApiVersion = "v1",
Metadata = CreateServiceMetadata(recipes, namePostfix),
Spec = new V1ServiceSpec
{
Type = serviceType,
Selector = GetSelector(recipes),
Ports = ports,
}
};
client.Run(c => c.CreateNamespacedService(serviceSpec, K8sNamespace));
var result = ReadBackServiceAndMapPorts(serviceSpec, recipes);
var name = serviceSpec.Metadata.Name;
return new RunningService(name, result);
}
private List<ContainerRecipePortMapEntry> ReadBackServiceAndMapPorts(V1Service serviceSpec, ContainerRecipe[] containerRecipes)
{
var result = new List<ContainerRecipePortMapEntry>();
// For each container-recipe-port, we need to figure out which service-ports it was assigned by K8s.
var readback = client.Run(c => c.ReadNamespacedService(serviceSpec.Metadata.Name, K8sNamespace));
foreach (var r in containerRecipes)
{
var recipePorts = r.ExposedPorts.Concat(r.InternalPorts).ToArray();
foreach (var port in recipePorts)
{
var portName = GetNameForPort(r, port);
var matchingServicePorts = readback.Spec.Ports.Where(p => p.Name == portName);
var ports = matchingServicePorts.Select(p => MapPortIfAble(p, port.Tag, port.Protocol)).ToArray();
if (ports.Any())
{
result.Add(new ContainerRecipePortMapEntry(r.Number, ports));
log.Debug($"Service Readback: {portName} found: {string.Join(",", ports.Select(p => p.ToString()))}");
}
}
}
return result;
}
private Port MapPortIfAble(V1ServicePort p, string tag, PortProtocol protocol)
{
if (p.NodePort != null) return new Port(p.NodePort.Value, tag, protocol);
if (p.Port > 0) return new Port(p.Port, tag, protocol);
throw new Exception("Unable to map port.");
}
private void DeleteService(RunningService service)
{
client.Run(c => c.DeleteNamespacedService(service.Name, K8sNamespace));
}
private V1ObjectMeta CreateServiceMetadata(ContainerRecipe[] containerRecipes, string namePostfix)
{
var recipeName = containerRecipes.First().Name;
var name = K8sNameUtils.Format($"{recipeName}-{workflowNumberSource.WorkflowNumber}-{namePostfix}");
log.Debug("Creating service: " + name);
return new V1ObjectMeta
{
Name = name,
NamespaceProperty = K8sNamespace,
};
}
private List<V1ServicePort> CreateServicePorts(ContainerRecipe[] recipes, Func<ContainerRecipe, Port[]> portSelector, bool isNodePort)
{
var result = new List<V1ServicePort>();
foreach (var recipe in recipes)
{
var ports = portSelector(recipe);
foreach (var port in ports)
{
result.AddRange(CreateServicePorts(recipe, port, isNodePort));
}
}
return result;
}
private List<V1ServicePort> CreateServicePorts(ContainerRecipe recipe, Port recipePort, bool isNodePort)
{
var result = new List<V1ServicePort>();
if (recipePort.IsTcp()) CreateServicePort(result, recipe, recipePort, "TCP", isNodePort);
if (recipePort.IsUdp()) CreateServicePort(result, recipe, recipePort, "UDP", isNodePort);
return result;
}
private void CreateServicePort(List<V1ServicePort> result, ContainerRecipe recipe, Port port, string protocol, bool isNodePort)
{
var p = new V1ServicePort
{
Name = GetNameForPort(recipe, port),
Protocol = protocol,
Port = port.Number,
TargetPort = GetNameForPort(recipe, port)
};
if (isNodePort) p.NodePort = port.Number;
result.Add(p);
}
#endregion
#region Waiting
private void WaitUntilNamespaceCreated()
{
WaitUntil(() => IsNamespaceOnline(K8sNamespace));
}
private void WaitUntilDeploymentOnline(string deploymentName)
{
WaitUntil(() =>
{
var deployment = client.Run(c => c.ReadNamespacedDeployment(deploymentName, K8sNamespace));
return deployment?.Status.AvailableReplicas != null && deployment.Status.AvailableReplicas > 0;
});
}
private void WaitUntilDeploymentOffline(string deploymentName)
{
WaitUntil(() =>
{
var deployments = client.Run(c => c.ListNamespacedDeployment(K8sNamespace));
var deployment = deployments.Items.SingleOrDefault(d => d.Metadata.Name == deploymentName);
return deployment == null || deployment.Status.AvailableReplicas == 0;
});
}
private void WaitUntilPodsForDeploymentAreOffline(RunningDeployment deployment)
{
WaitUntil(() =>
{
var pods = FindPodsByLabel(deployment.PodLabel);
return !pods.Any();
});
}
private void WaitUntil(Func<bool> predicate)
{
var sw = Stopwatch.Begin(log, true);
try
{
Time.WaitUntil(predicate, cluster.K8sOperationTimeout(), cluster.K8sOperationRetryDelay());
}
finally
{
sw.End("", 1);
}
}
#endregion
public CrashWatcher CreateCrashWatcher(RunningContainer container)
{
var containerName = container.Name;
var podName = GetPodName(container);
var recipeName = container.Recipe.Name;
return new CrashWatcher(log, cluster.GetK8sClientConfig(), containerName, podName, recipeName, K8sNamespace);
}
private V1Pod[] FindPodsByLabel(string podLabel)
{
var pods = client.Run(c => c.ListNamespacedPod(K8sNamespace));
return pods.Items.Where(p => p.GetLabel(PodLabelKey) == podLabel).ToArray();
}
private PodInfo CreatePodInfo(V1Pod pod)
{
var name = pod.Name();
var ip = pod.Status.PodIP;
var k8sNodeName = pod.Spec.NodeName;
if (string.IsNullOrEmpty(name)) throw new InvalidOperationException("Invalid pod name received. Test infra failure.");
if (string.IsNullOrEmpty(ip)) throw new InvalidOperationException("Invalid pod IP received. Test infra failure.");
return new PodInfo(name, ip, k8sNodeName);
}
}
}