public/kube-fetcher
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

moved configuration to fetcher object.

Browse Source
This commit is contained in:
Erik Brakkee 2025-03-02 17:03:03 +01:00
parent 58ef7f76e4
commit 1688aa6dea
2 changed files with 31 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

42
cmd/fetcher/sync.go → cmd/fetcher/fetcher.go
View File

@ -13,7 +13,15 @@ import (
"time" "time"
) )
func canonicalizeImageName(image string) string { type Fetcher struct {
KubernetesNamespace string
SocketPath string
ContainerdNamespace string
Nodename string
ReadyDuration time.Duration
}
func (fetcher *Fetcher) canonicalizeImageName(image string) string {
pattern := `^(?:(?P<registry>[a-zA-Z0-9][-a-zA-Z0-9.]*[a-zA-Z0-9](?::[0-9]+)?)/)?(?P<repository>(?:[a-z0-9]+(?:(?:[._]|__|[-]+)[a-z0-9]+)*(?:/[a-z0-9]+(?:(?:[._]|__|[-]+)[a-z0-9]+)*)*)?)?(?::(?P<tag>[\w][\w.-]{0,127}))?(?:@(?P<digest>[a-z][a-z0-9]*(?:[+.-][a-z][a-z0-9]*)*:[a-zA-Z0-9*+,-./:;=@_]{32,}))?$` pattern := `^(?:(?P<registry>[a-zA-Z0-9][-a-zA-Z0-9.]*[a-zA-Z0-9](?::[0-9]+)?)/)?(?P<repository>(?:[a-z0-9]+(?:(?:[._]|__|[-]+)[a-z0-9]+)*(?:/[a-z0-9]+(?:(?:[._]|__|[-]+)[a-z0-9]+)*)*)?)?(?::(?P<tag>[\w][\w.-]{0,127}))?(?:@(?P<digest>[a-z][a-z0-9]*(?:[+.-][a-z][a-z0-9]*)*:[a-zA-Z0-9*+,-./:;=@_]{32,}))?$`
re := regexp.MustCompile(pattern) re := regexp.MustCompile(pattern)
matches := re.FindStringSubmatch(image) matches := re.FindStringSubmatch(image)
@ -64,11 +72,11 @@ func canonicalizeImageName(image string) string {
return fullimage return fullimage
} }
func isReadyForSomeTime(pod *v1.Pod, duration time.Duration) bool { func (fetcher *Fetcher) isReadyForSomeTime(pod *v1.Pod) bool {
ready := false ready := false
for _, condition := range pod.Status.Conditions { for _, condition := range pod.Status.Conditions {
if condition.Type == corev1.PodReady && condition.Status == corev1.ConditionTrue { if condition.Type == corev1.PodReady && condition.Status == corev1.ConditionTrue {
if time.Now().Sub(condition.LastTransitionTime.Time) > duration { if time.Now().Sub(condition.LastTransitionTime.Time) > fetcher.ReadyDuration {
ready = true ready = true
} }
} }
@ -79,9 +87,8 @@ func isReadyForSomeTime(pod *v1.Pod, duration time.Duration) bool {
return ready return ready
} }
func getContainers(clientset *kubernetes.Clientset, kubernetesNamespace, nodename string, func (fetcher *Fetcher) getContainers(clientset *kubernetes.Clientset) map[string]bool {
readyDuration time.Duration) map[string]bool { pods, err := clientset.CoreV1().Pods(fetcher.KubernetesNamespace).List(context.Background(),
pods, err := clientset.CoreV1().Pods(kubernetesNamespace).List(context.Background(),
metav1.ListOptions{}) metav1.ListOptions{})
if err != nil { if err != nil {
panic(err) panic(err)
@ -94,21 +101,21 @@ func getContainers(clientset *kubernetes.Clientset, kubernetesNamespace, nodenam
klog.V(3).Infof("%s/%s\n", pod.Namespace, pod.Name) klog.V(3).Infof("%s/%s\n", pod.Namespace, pod.Name)
for _, container := range pod.Spec.InitContainers { for _, container := range pod.Spec.InitContainers {
klog.V(3).Infof(" %s\n", container.Image) klog.V(3).Infof(" %s\n", container.Image)
if pod.Spec.NodeName == nodename { if pod.Spec.NodeName == fetcher.Nodename {
containersOnCurrentNode[canonicalizeImageName(container.Image)] = true containersOnCurrentNode[fetcher.canonicalizeImageName(container.Image)] = true
} else { } else {
if isReadyForSomeTime(&pod, readyDuration) { if fetcher.isReadyForSomeTime(&pod) {
containers[canonicalizeImageName(container.Image)] = true containers[fetcher.canonicalizeImageName(container.Image)] = true
} }
} }
} }
for _, container := range pod.Spec.Containers { for _, container := range pod.Spec.Containers {
klog.V(3).Infof(" %s\n", container.Image) klog.V(3).Infof(" %s\n", container.Image)
if pod.Spec.NodeName == nodename { if pod.Spec.NodeName == fetcher.Nodename {
containersOnCurrentNode[canonicalizeImageName(container.Image)] = true containersOnCurrentNode[fetcher.canonicalizeImageName(container.Image)] = true
} else { } else {
if isReadyForSomeTime(&pod, readyDuration) { if fetcher.isReadyForSomeTime(&pod) {
containers[canonicalizeImageName(container.Image)] = true containers[fetcher.canonicalizeImageName(container.Image)] = true
} }
} }
} }
@ -120,11 +127,10 @@ func getContainers(clientset *kubernetes.Clientset, kubernetesNamespace, nodenam
return containers return containers
} }
func pullAndPin(kubernetesNamespace, socketPath, containerdNamespace, nodename string, func (fetcher *Fetcher) pullAndPin() error {
readyDuration time.Duration) error {
// Create the image manager // Create the image manager
containerd, err := NewContainerd(socketPath, containerdNamespace) containerd, err := NewContainerd(fetcher.SocketPath, fetcher.ContainerdNamespace)
if err != nil { if err != nil {
klog.Fatalf("Failed to create image manager: %v", err) klog.Fatalf("Failed to create image manager: %v", err)
} }
@ -136,7 +142,7 @@ func pullAndPin(kubernetesNamespace, socketPath, containerdNamespace, nodename s
clientset, _ := GetKubernetesConnection() clientset, _ := GetKubernetesConnection()
containers := getContainers(clientset, kubernetesNamespace, nodename, readyDuration) containers := fetcher.getContainers(clientset)
for container, _ := range containers { for container, _ := range containers {
klog.V(3).Infof("Found container %s\n", container) klog.V(3).Infof("Found container %s\n", container)
} }

19
cmd/fetcher/main.go
View File

@ -12,11 +12,7 @@ func main() {
klogFlags := goflags.NewFlagSet("", goflags.PanicOnError) klogFlags := goflags.NewFlagSet("", goflags.PanicOnError)
klog.InitFlags(klogFlags) klog.InitFlags(klogFlags)
var kubernetesNamespace string fetcher := Fetcher{}
var socketPath string
var containerdNamespace string
var nodename string
var readyDuration time.Duration
cmd := &cobra.Command{ cmd := &cobra.Command{
Use: "kube-fetcher", Use: "kube-fetcher",
@ -26,21 +22,20 @@ Queries k8s for all running pods and makes sure that all
images referenced in pods are made available on the local k8s node and pinned images referenced in pods are made available on the local k8s node and pinned
so they don't get garbage collected'`, so they don't get garbage collected'`,
RunE: func(cmd *cobra.Command, args []string) error { RunE: func(cmd *cobra.Command, args []string) error {
err := pullAndPin(kubernetesNamespace, socketPath, containerdNamespace, nodename, err := fetcher.pullAndPin()
readyDuration)
return err return err
}, },
} }
cmd.PersistentFlags().StringVar(&kubernetesNamespace, "kubernetes-namespace", cmd.PersistentFlags().StringVar(&fetcher.KubernetesNamespace, "kubernetes-namespace",
"", "Kubernetes containerdNamespace to inspect (default is all namespaces)") "", "Kubernetes containerdNamespace to inspect (default is all namespaces)")
cmd.PersistentFlags().StringVar(&socketPath, "socket", cmd.PersistentFlags().StringVar(&fetcher.SocketPath, "socket",
"/run/containerd/containerd.sock", "Containerd socket") "/run/containerd/containerd.sock", "Containerd socket")
cmd.PersistentFlags().StringVar(&containerdNamespace, "containerd-namespace", cmd.PersistentFlags().StringVar(&fetcher.ContainerdNamespace, "containerd-namespace",
"k8s.io", "Containerd namespace to use") "k8s.io", "Containerd namespace to use")
cmd.PersistentFlags().StringVar(&nodename, "nodename", "", cmd.PersistentFlags().StringVar(&fetcher.Nodename, "nodename", "",
"Kubernetes node name the fetcher is running on, it will only fetch images running on other nodes") "Kubernetes node name the fetcher is running on, it will only fetch images running on other nodes")
cmd.PersistentFlags().DurationVar(&readyDuration, "ready-duration", cmd.PersistentFlags().DurationVar(&fetcher.ReadyDuration, "ready-duration",
1*time.Hour, "Time a pod must be ready before its image will be fetched") 1*time.Hour, "Time a pod must be ready before its image will be fetched")
cmd.Flags().AddGoFlagSet(klogFlags) cmd.Flags().AddGoFlagSet(klogFlags)