Added Service & PVC cache for faster inspection loops

Author: ewoutp

pkg/deployment/resources/pvcs.go

@@ -46,23 +46,11 @@ func (r *Resources) EnsurePVCs() error {
 	status, _ := r.context.GetStatus()
 	enforceAntiAffinity := r.context.GetSpec().GetEnvironment().IsProduction()
 
-	pvcs := kubecli.CoreV1().PersistentVolumeClaims(ns)
-	list, err := pvcs.List(metav1.ListOptions{})
-	if err != nil {
-		return maskAny(err)
-	}
-	pvcExists := func(name string) bool {
-		for _, pvc := range list.Items {
-			if pvc.GetName() == name {
-				return true
-			}
-		}
-		return false
-	}
+	pvcs := k8sutil.NewPersistentVolumeClaimCache(kubecli.CoreV1().PersistentVolumeClaims(ns))
 	if err := iterator.ForeachServerGroup(func(group api.ServerGroup, spec api.ServerGroupSpec, status *api.MemberStatusList) error {
 		for _, m := range *status {
 			if m.PersistentVolumeClaimName != "" {
-				if !pvcExists(m.PersistentVolumeClaimName) {
+				if _, err := pvcs.Get(m.PersistentVolumeClaimName, metav1.GetOptions{}); err != nil {
 					storageClassName := spec.GetStorageClassName()
 					role := group.AsRole()
 					resources := spec.Resources

pkg/deployment/resources/services.go

@@ -46,23 +46,9 @@ func (r *Resources) EnsureServices() error {
 	spec := r.context.GetSpec()
 
 	// Fetch existing services
-	svcs := kubecli.CoreV1().Services(ns)
-	list, err := svcs.List(metav1.ListOptions{})
-	if err != nil {
-		log.Debug().Err(err).Msg("Failed to list existing services")
-		return maskAny(err)
-	}
-	svcExists := func(name string) bool {
-		for _, svc := range list.Items {
-			if svc.GetName() == name {
-				return true
-			}
-		}
-		return false
-	}
-
+	svcs := k8sutil.NewServiceCache(kubecli.CoreV1().Services(ns))
 	// Headless service
-	if !svcExists(k8sutil.CreateHeadlessServiceName(deploymentName)) {
+	if _, err := svcs.Get(k8sutil.CreateHeadlessServiceName(deploymentName), metav1.GetOptions{}); err != nil {
 		svcName, newlyCreated, err := k8sutil.CreateHeadlessService(svcs, apiObject, owner)
 		if err != nil {
 			log.Debug().Err(err).Msg("Failed to create headless service")
@@ -75,7 +61,7 @@ func (r *Resources) EnsureServices() error {
 
 	// Internal database client service
 	single := spec.GetMode().HasSingleServers()
-	if !svcExists(k8sutil.CreateDatabaseClientServiceName(deploymentName)) {
+	if _, err := svcs.Get(k8sutil.CreateDatabaseClientServiceName(deploymentName), metav1.GetOptions{}); err != nil {
 		svcName, newlyCreated, err := k8sutil.CreateDatabaseClientService(svcs, apiObject, single, owner)
 		if err != nil {
 			log.Debug().Err(err).Msg("Failed to create database client service")
@@ -101,15 +87,15 @@ func (r *Resources) EnsureServices() error {
 	if single {
 		role = "single"
 	}
-	if err := r.ensureExternalAccessServices(eaServiceName, ns, role, "database", k8sutil.ArangoPort, false, spec.ExternalAccess, apiObject, log, kubecli); err != nil {
+	if err := r.ensureExternalAccessServices(svcs, eaServiceName, ns, role, "database", k8sutil.ArangoPort, false, spec.ExternalAccess, apiObject, log, kubecli); err != nil {
 		return maskAny(err)
 	}
 
 	if spec.Sync.IsEnabled() {
 		// External (and internal) Sync master service
 		eaServiceName := k8sutil.CreateSyncMasterClientServiceName(deploymentName)
 		role := "syncmaster"
-		if err := r.ensureExternalAccessServices(eaServiceName, ns, role, "sync", k8sutil.ArangoSyncMasterPort, true, spec.Sync.ExternalAccess.ExternalAccessSpec, apiObject, log, kubecli); err != nil {
+		if err := r.ensureExternalAccessServices(svcs, eaServiceName, ns, role, "sync", k8sutil.ArangoSyncMasterPort, true, spec.Sync.ExternalAccess.ExternalAccessSpec, apiObject, log, kubecli); err != nil {
 			return maskAny(err)
 		}
 		status, lastVersion := r.context.GetStatus()
@@ -124,12 +110,11 @@ func (r *Resources) EnsureServices() error {
 }
 
 // EnsureServices creates all services needed to service the deployment
-func (r *Resources) ensureExternalAccessServices(eaServiceName, ns, svcRole, title string, port int, noneIsClusterIP bool, spec api.ExternalAccessSpec, apiObject k8sutil.APIObject, log zerolog.Logger, kubecli kubernetes.Interface) error {
+func (r *Resources) ensureExternalAccessServices(svcs k8sutil.ServiceInterface, eaServiceName, ns, svcRole, title string, port int, noneIsClusterIP bool, spec api.ExternalAccessSpec, apiObject k8sutil.APIObject, log zerolog.Logger, kubecli kubernetes.Interface) error {
 	// Database external access service
 	createExternalAccessService := false
 	deleteExternalAccessService := false
 	eaServiceType := spec.GetType().AsServiceType() // Note: Type auto defaults to ServiceTypeLoadBalancer
-	svcs := kubecli.CoreV1().Services(ns)
 	if existing, err := svcs.Get(eaServiceName, metav1.GetOptions{}); err == nil {
 		// External access service exists
 		loadBalancerIP := spec.GetLoadBalancerIP()

pkg/util/k8sutil/pvc.go

@@ -27,11 +27,16 @@ import (
 
 	"k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	corev1 "k8s.io/client-go/kubernetes/typed/core/v1"
 
 	"github.com/arangodb/kube-arangodb/pkg/util/constants"
 )
 
+// PersistentVolumeClaimInterface has methods to work with PersistentVolumeClaim resources.
+type PersistentVolumeClaimInterface interface {
+	Create(*v1.PersistentVolumeClaim) (*v1.PersistentVolumeClaim, error)
+	Get(name string, options metav1.GetOptions) (*v1.PersistentVolumeClaim, error)
+}
+
 // IsPersistentVolumeClaimMarkedForDeletion returns true if the pod has been marked for deletion.
 func IsPersistentVolumeClaimMarkedForDeletion(pvc *v1.PersistentVolumeClaim) bool {
 	return pvc.DeletionTimestamp != nil
@@ -46,7 +51,7 @@ func CreatePersistentVolumeClaimName(deploymentName, role, id string) string {
 // CreatePersistentVolumeClaim creates a persistent volume claim with given name and configuration.
 // If the pvc already exists, nil is returned.
 // If another error occurs, that error is returned.
-func CreatePersistentVolumeClaim(pvcs corev1.PersistentVolumeClaimInterface, pvcName, deploymentName, ns, storageClassName, role string, enforceAntiAffinity bool, resources v1.ResourceRequirements, finalizers []string, owner metav1.OwnerReference) error {
+func CreatePersistentVolumeClaim(pvcs PersistentVolumeClaimInterface, pvcName, deploymentName, ns, storageClassName, role string, enforceAntiAffinity bool, resources v1.ResourceRequirements, finalizers []string, owner metav1.OwnerReference) error {
 	labels := LabelsForDeployment(deploymentName, role)
 	volumeMode := v1.PersistentVolumeFilesystem
 	pvc := &v1.PersistentVolumeClaim{

pkg/util/k8sutil/pvc_cache.go

@@ -0,0 +1,75 @@
+//
+// DISCLAIMER
+//
+// Copyright 2018 ArangoDB GmbH, Cologne, Germany
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+// Copyright holder is ArangoDB GmbH, Cologne, Germany
+//
+// Author Ewout Prangsma
+//
+
+package k8sutil
+
+import (
+	"k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	corev1 "k8s.io/client-go/kubernetes/typed/core/v1"
+)
+
+// pvcsCache implements a cached version of a PersistentVolumeClaimInterface.
+// It is NOT go-routine safe.
+type pvcsCache struct {
+	cli   corev1.PersistentVolumeClaimInterface
+	cache []v1.PersistentVolumeClaim
+}
+
+// NewPersistentVolumeClaimCache creates a cached version of the given PersistentVolumeClaimInterface.
+func NewPersistentVolumeClaimCache(cli corev1.PersistentVolumeClaimInterface) PersistentVolumeClaimInterface {
+	return &pvcsCache{cli: cli}
+}
+
+var (
+	pvcGroupResource = schema.GroupResource{
+		Group:    v1.GroupName,
+		Resource: "PersistentVolumeClaim",
+	}
+)
+
+func (sc *pvcsCache) Create(s *v1.PersistentVolumeClaim) (*v1.PersistentVolumeClaim, error) {
+	sc.cache = nil
+	result, err := sc.cli.Create(s)
+	if err != nil {
+		return nil, maskAny(err)
+	}
+	return result, nil
+}
+
+func (sc *pvcsCache) Get(name string, options metav1.GetOptions) (*v1.PersistentVolumeClaim, error) {
+	if sc.cache == nil {
+		list, err := sc.cli.List(metav1.ListOptions{})
+		if err != nil {
+			return nil, maskAny(err)
+		}
+		sc.cache = list.Items
+	}
+	for _, s := range sc.cache {
+		if s.GetName() == name {
+			return &s, nil
+		}
+	}
+	return nil, maskAny(apierrors.NewNotFound(pvcGroupResource, name))
+}

pkg/util/k8sutil/services.go

@@ -31,9 +31,15 @@ import (
 
 	"k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	corev1 "k8s.io/client-go/kubernetes/typed/core/v1"
 )
 
+// ServiceInterface has methods to work with Service resources.
+type ServiceInterface interface {
+	Create(*v1.Service) (*v1.Service, error)
+	Delete(name string, options *metav1.DeleteOptions) error
+	Get(name string, options metav1.GetOptions) (*v1.Service, error)
+}
+
 // CreateHeadlessServiceName returns the name of the headless service for the given
 // deployment name.
 func CreateHeadlessServiceName(deploymentName string) string {
@@ -63,7 +69,7 @@ func CreateSyncMasterClientServiceName(deploymentName string) string {
 // If the service already exists, nil is returned.
 // If another error occurs, that error is returned.
 // The returned bool is true if the service is created, or false when the service already existed.
-func CreateHeadlessService(svcs corev1.ServiceInterface, deployment metav1.Object, owner metav1.OwnerReference) (string, bool, error) {
+func CreateHeadlessService(svcs ServiceInterface, deployment metav1.Object, owner metav1.OwnerReference) (string, bool, error) {
 	deploymentName := deployment.GetName()
 	svcName := CreateHeadlessServiceName(deploymentName)
 	ports := []v1.ServicePort{
@@ -86,7 +92,7 @@ func CreateHeadlessService(svcs corev1.ServiceInterface, deployment metav1.Objec
 // If the service already exists, nil is returned.
 // If another error occurs, that error is returned.
 // The returned bool is true if the service is created, or false when the service already existed.
-func CreateDatabaseClientService(svcs corev1.ServiceInterface, deployment metav1.Object, single bool, owner metav1.OwnerReference) (string, bool, error) {
+func CreateDatabaseClientService(svcs ServiceInterface, deployment metav1.Object, single bool, owner metav1.OwnerReference) (string, bool, error) {
 	deploymentName := deployment.GetName()
 	svcName := CreateDatabaseClientServiceName(deploymentName)
 	ports := []v1.ServicePort{
@@ -115,7 +121,7 @@ func CreateDatabaseClientService(svcs corev1.ServiceInterface, deployment metav1
 // If the service already exists, nil is returned.
 // If another error occurs, that error is returned.
 // The returned bool is true if the service is created, or false when the service already existed.
-func CreateExternalAccessService(svcs corev1.ServiceInterface, svcName, role string, deployment metav1.Object, serviceType v1.ServiceType, port, nodePort int, loadBalancerIP string, owner metav1.OwnerReference) (string, bool, error) {
+func CreateExternalAccessService(svcs ServiceInterface, svcName, role string, deployment metav1.Object, serviceType v1.ServiceType, port, nodePort int, loadBalancerIP string, owner metav1.OwnerReference) (string, bool, error) {
 	deploymentName := deployment.GetName()
 	ports := []v1.ServicePort{
 		v1.ServicePort{
@@ -137,7 +143,7 @@ func CreateExternalAccessService(svcs corev1.ServiceInterface, svcName, role str
 // If the service already exists, nil is returned.
 // If another error occurs, that error is returned.
 // The returned bool is true if the service is created, or false when the service already existed.
-func createService(svcs corev1.ServiceInterface, svcName, deploymentName, ns, clusterIP, role string, serviceType v1.ServiceType,
+func createService(svcs ServiceInterface, svcName, deploymentName, ns, clusterIP, role string, serviceType v1.ServiceType,
 	ports []v1.ServicePort, loadBalancerIP string, publishNotReadyAddresses bool, owner metav1.OwnerReference) (bool, error) {
 	labels := LabelsForDeployment(deploymentName, role)
 	svc := &v1.Service{

pkg/util/k8sutil/services_cache.go

@@ -0,0 +1,83 @@
+//
+// DISCLAIMER
+//
+// Copyright 2018 ArangoDB GmbH, Cologne, Germany
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+// Copyright holder is ArangoDB GmbH, Cologne, Germany
+//
+// Author Ewout Prangsma
+//
+
+package k8sutil
+
+import (
+	"k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	corev1 "k8s.io/client-go/kubernetes/typed/core/v1"
+)
+
+// servicesCache implements a cached version of a ServiceInterface.
+// It is NOT go-routine safe.
+type servicesCache struct {
+	cli   corev1.ServiceInterface
+	cache []v1.Service
+}
+
+// NewServiceCache creates a cached version of the given ServiceInterface.
+func NewServiceCache(cli corev1.ServiceInterface) ServiceInterface {
+	return &servicesCache{cli: cli}
+}
+
+var (
+	serviceGroupResource = schema.GroupResource{
+		Group:    v1.GroupName,
+		Resource: "Service",
+	}
+)
+
+func (sc *servicesCache) Create(s *v1.Service) (*v1.Service, error) {
+	sc.cache = nil
+	result, err := sc.cli.Create(s)
+	if err != nil {
+		return nil, maskAny(err)
+	}
+	return result, nil
+}
+
+func (sc *servicesCache) Delete(name string, options *metav1.DeleteOptions) error {
+	sc.cache = nil
+	if err := sc.cli.Delete(name, options); err != nil {
+		return maskAny(err)
+	}
+	return nil
+}
+
+func (sc *servicesCache) Get(name string, options metav1.GetOptions) (*v1.Service, error) {
+	if sc.cache == nil {
+		list, err := sc.cli.List(metav1.ListOptions{})
+		if err != nil {
+			return nil, maskAny(err)
+		}
+		sc.cache = list.Items
+	}
+	for _, s := range sc.cache {
+		if s.GetName() == name {
+			return &s, nil
+		}
+	}
+	return nil, maskAny(apierrors.NewNotFound(serviceGroupResource, name))
+}