Queue algorithm simplification

Also, prioritize checking previously failed images

Author: Andrey Klimentyev

go.mod

@@ -3,10 +3,11 @@ module github.com/flant/k8s-image-availability-exporter
 go 1.16
 
 require (
-	github.com/emirpasic/gods v1.12.0
+	github.com/gammazero/deque v0.1.0
 	github.com/google/go-containerregistry v0.6.0
 	github.com/prometheus/client_golang v1.11.0
 	github.com/sirupsen/logrus v1.8.1
+	github.com/stretchr/testify v1.7.0
 	github.com/vdemeester/k8s-pkg-credentialprovider v1.21.0
 	k8s.io/api v0.22.2
 	k8s.io/apimachinery v0.22.2

go.sum

@@ -304,8 +304,6 @@ github.com/dustin/go-humanize v1.0.0/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25Kn
 github.com/elazarl/goproxy v0.0.0-20180725130230-947c36da3153/go.mod h1:/Zj4wYkgs4iZTTu3o/KG3Itv/qCCa8VVMlb3i9OVuzc=
 github.com/emicklei/go-restful v0.0.0-20170410110728-ff4f55a20633/go.mod h1:otzb+WCGbkyDHkqmQmT5YD2WR4BBwUdeQoFo8l/7tVs=
 github.com/emicklei/go-restful v2.9.5+incompatible/go.mod h1:otzb+WCGbkyDHkqmQmT5YD2WR4BBwUdeQoFo8l/7tVs=
-github.com/emirpasic/gods v1.12.0 h1:QAUIPSaCu4G+POclxeqb3F+WPpdKqFGlw36+yOzGlrg=
-github.com/emirpasic/gods v1.12.0/go.mod h1:YfzfFFoVP/catgzJb4IKIqXjX78Ha8FMSDh3ymbK86o=
 github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
 github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
 github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1mIlRU8Am5FuJP05cCM98=
@@ -330,6 +328,8 @@ github.com/fsnotify/fsnotify v1.4.9 h1:hsms1Qyu0jgnwNXIxa+/V/PDsU6CfLf6CNO8H7IWo
 github.com/fsnotify/fsnotify v1.4.9/go.mod h1:znqG4EE+3YCdAaPaxE2ZRY/06pZUdp0tY4IgpuI1SZQ=
 github.com/fullsailor/pkcs7 v0.0.0-20190404230743-d7302db945fa/go.mod h1:KnogPXtdwXqoenmZCw6S+25EAm2MkxbG0deNDu4cbSA=
 github.com/fvbommel/sortorder v1.0.1/go.mod h1:uk88iVf1ovNn1iLfgUVU2F9o5eO30ui720w+kxuqRs0=
+github.com/gammazero/deque v0.1.0 h1:f9LnNmq66VDeuAlSAapemq/U7hJ2jpIWa4c09q8Dlik=
+github.com/gammazero/deque v0.1.0/go.mod h1:KQw7vFau1hHuM8xmI9RbgKFbAsQFWmBpqQ2KenFLk6M=
 github.com/garyburd/redigo v0.0.0-20150301180006-535138d7bcd7/go.mod h1:NR3MbYisc3/PwhQ00EMzDiPmrwpPxAn5GI05/YaO1SY=
 github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
 github.com/globalsign/mgo v0.0.0-20180905125535-1ca0a4f7cbcb/go.mod h1:xkRDCp4j0OGD1HRkm4kmhM+pmpv3AKq5SU7GMg4oO/Q=

main.go

@@ -65,6 +65,7 @@ func main() {
 	prometheus.MustRegister(liveTicksCounter)
 
 	registryChecker := registry_checker.NewRegistryChecker(
+		stopCh,
 		kubeClient,
 		*insecureSkipVerify,
 		strings.Split(*ignoredImagesStr, ","),
@@ -73,8 +74,6 @@ func main() {
 	)
 	prometheus.MustRegister(registryChecker)
 
-	registryChecker.Run(stopCh)
-
 	http.Handle("/metrics", promhttp.Handler())
 	http.HandleFunc("/healthz", handlers.Healthz)
 	go func() {
@@ -84,7 +83,7 @@ func main() {
 	handlers.UpdateHealth(true)
 
 	wait.Until(func() {
-		registryChecker.Check()
+		registryChecker.Tick()
 		liveTicksCounter.Inc()
 	}, *imageCheckInterval, stopCh)
 }

pkg/context/context.go

@@ -1,14 +1,18 @@
 package registry_checker
 
 import (
+	"context"
 	"crypto/tls"
 	"errors"
-	appsv1 "k8s.io/api/apps/v1"
-	"k8s.io/apimachinery/pkg/util/wait"
 	"net/http"
-	"sync"
+	"path"
 	"time"
 
+	"github.com/prometheus/client_golang/prometheus"
+	appsv1 "k8s.io/api/apps/v1"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/tools/cache"
+
 	"github.com/google/go-containerregistry/pkg/name"
 	"github.com/google/go-containerregistry/pkg/v1/remote"
 	"github.com/sirupsen/logrus"
@@ -21,11 +25,12 @@ import (
 
 	"github.com/flant/k8s-image-availability-exporter/pkg/store"
 	"k8s.io/client-go/kubernetes"
-	"k8s.io/client-go/tools/cache"
 )
 
 const (
-	resyncPeriod = time.Hour
+	resyncPeriod         = time.Hour
+	failedCheckBatchSize = 20
+	checkBatchSize       = 50
 )
 
 type registryCheckerConfig struct {
@@ -53,6 +58,7 @@ type RegistryChecker struct {
 }
 
 func NewRegistryChecker(
+	stopCh <-chan struct{},
 	kubeClient *kubernetes.Clientset,
 	skipVerify bool,
 	ignoredImages []string,
@@ -71,8 +77,6 @@ func NewRegistryChecker(
 	}
 
 	rc := &RegistryChecker{
-		imageStore: store.NewImageStore(),
-
 		deploymentsInformer:  informerFactory.Apps().V1().Deployments(),
 		statefulSetsInformer: informerFactory.Apps().V1().StatefulSets(),
 		daemonSetsInformer:   informerFactory.Apps().V1().DaemonSets(),
@@ -94,10 +98,8 @@ func NewRegistryChecker(
 		rc.ignoredImages[image] = struct{}{}
 	}
 
-	return rc
-}
+	rc.imageStore = store.NewImageStore(rc.Check, checkBatchSize, failedCheckBatchSize)
 
-func (rc *RegistryChecker) Run(stopCh <-chan struct{}) {
 	rc.deploymentsInformer.Informer().AddEventHandlerWithResyncPeriod(cache.ResourceEventHandlerFuncs{
 		AddFunc: func(obj interface{}) {
 			rc.reconcile(obj)
@@ -177,6 +179,26 @@ func (rc *RegistryChecker) Run(stopCh <-chan struct{}) {
 	cache.WaitForCacheSync(stopCh, rc.deploymentsInformer.Informer().HasSynced, rc.statefulSetsInformer.Informer().HasSynced,
 		rc.daemonSetsInformer.Informer().HasSynced, rc.cronJobsInformer.Informer().HasSynced, rc.secretsInformer.Informer().HasSynced)
 	logrus.Info("Caches populated successfully")
+
+	rc.imageStore.RunGC(rc.controllerIndexers.GetContainerInfosForImage)
+
+	return rc
+}
+
+// Collect implements prometheus.Collector.
+func (rc *RegistryChecker) Collect(ch chan<- prometheus.Metric) {
+	metrics := rc.imageStore.ExtractMetrics()
+
+	for _, m := range metrics {
+		ch <- m
+	}
+}
+
+// Describe implements prometheus.Collector.
+func (rc *RegistryChecker) Describe(_ chan<- *prometheus.Desc) {}
+
+func (rc RegistryChecker) Tick() {
+	rc.imageStore.Check()
 }
 
 func (rc *RegistryChecker) reconcile(obj interface{}) {
@@ -208,12 +230,13 @@ func (rc *RegistryChecker) reconcile(obj interface{}) {
 			}
 		}
 
-		if skipObject || !rc.controllerIndexers.CheckImageExistence(image) {
-			rc.imageStore.RemoveImage(image)
+		if skipObject {
+			rc.imageStore.ReconcileImage(image, []store.ContainerInfo{})
 			continue
 		}
 
-		rc.imageStore.AddOrUpdateImage(image, time.Time{})
+		containerInfos := rc.controllerIndexers.GetContainerInfosForImage(image)
+		rc.imageStore.ReconcileImage(image, containerInfos)
 	}
 }
 
@@ -225,30 +248,11 @@ func (rc *RegistryChecker) reconcileUpdate(a, b interface{}) {
 	rc.reconcile(b)
 }
 
-func (rc *RegistryChecker) Check() {
-	// TODO: tweak const
-	oldImages := rc.imageStore.PopOldestImages(rc.imageStore.Length() / 40)
-
-	var processingGroup sync.WaitGroup
-	for _, image := range oldImages {
-		keyChain := rc.controllerIndexers.GetKeychainForImage(rc.kubeClient, image)
+func (rc *RegistryChecker) Check(imageName string) store.AvailabilityMode {
+	keyChain := rc.controllerIndexers.GetKeychainForImage(rc.kubeClient, imageName)
 
-		// TODO: backoff
-		processingGroup.Add(1)
-		go func(imageName string, kc *keychain) {
-			defer processingGroup.Done()
-
-			log := logrus.WithField("image_name", imageName)
-			availMode := rc.checkImageAvailability(log, imageName, kc)
-
-			rc.imageStore.AddOrUpdateImage(imageName, time.Now(), availMode)
-		}(image, keyChain)
-
-		containerInfos := rc.controllerIndexers.GetContainerInfosForImage(image)
-		rc.imageStore.UpdateContainerAssociations(image, containerInfos)
-	}
-
-	processingGroup.Wait()
+	log := logrus.WithField("image_name", imageName)
+	return rc.checkImageAvailability(log, imageName, keyChain)
 }
 
 func checkImageNameParseErr(log *logrus.Entry, err error) store.AvailabilityMode {
@@ -286,12 +290,16 @@ func (rc *RegistryChecker) checkImageAvailability(log *logrus.Entry, imageName s
 		Factor:   2,
 		Steps:    2,
 	}, func() (bool, error) {
+		ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
+		defer cancel()
+
 		if kc != nil {
 			for i := 0; i < kc.size; i++ {
 				indexedKeychain := *kc
 				indexedKeychain.index = i
 
-				_, imgErr = remote.Head(ref, remote.WithAuthFromKeychain(&indexedKeychain), remote.WithTransport(rc.registryTransport))
+				_, imgErr = remote.Head(ref, remote.WithAuthFromKeychain(&indexedKeychain), remote.WithTransport(rc.registryTransport),
+					remote.WithContext(ctx))
 
 				if imgErr != nil {
 					continue
@@ -302,7 +310,7 @@ func (rc *RegistryChecker) checkImageAvailability(log *logrus.Entry, imageName s
 				}
 			}
 		} else {
-			_, imgErr = remote.Head(ref, remote.WithTransport(rc.registryTransport))
+			_, imgErr = remote.Head(ref, remote.WithTransport(rc.registryTransport), remote.WithContext(ctx))
 		}
 
 		availMode = store.Available

pkg/registry_checker/checker.go

@@ -2,6 +2,7 @@ package registry_checker
 
 import (
 	"errors"
+
 	"github.com/google/go-containerregistry/pkg/v1/remote"
 	"github.com/google/go-containerregistry/pkg/v1/remote/transport"
 )

pkg/registry_checker/image_pull.go

@@ -208,12 +208,6 @@ func (ci ControllerIndexers) GetKeysByIndex(image string) (ret []string) {
 	return
 }
 
-func (ci ControllerIndexers) CheckImageExistence(image string) bool {
-	keys := ci.GetKeysByIndex(image)
-
-	return len(keys) > 0
-}
-
 func (ci ControllerIndexers) GetContainerInfosForImage(image string) (ret []store.ContainerInfo) {
 	objs := ci.GetObjectsByIndex(image)