aggregator: introduces retry mechanism

cmd/kube-apiserver/app/BUILD

@@ -79,6 +79,7 @@ go_library(
         "//staging/src/k8s.io/kube-aggregator/pkg/client/informers/externalversions/apiregistration/v1:go_default_library",
         "//staging/src/k8s.io/kube-aggregator/pkg/controllers/autoregister:go_default_library",
         "//vendor/github.com/go-openapi/spec:go_default_library",
+        "//vendor/github.com/p0lyn0mial/failure-detector:go_default_library",
         "//vendor/github.com/spf13/cobra:go_default_library",
         "//vendor/k8s.io/klog:go_default_library",
     ],

cmd/kube-apiserver/app/server.go

@@ -29,9 +29,11 @@ import (
 	"strconv"
 	"strings"
 	"time"
+	gcontext "context"
 
 	"github.com/go-openapi/spec"
 	"github.com/spf13/cobra"
+	failuredetector "github.com/p0lyn0mial/failure-detector"
 
 	extensionsapiserver "k8s.io/apiextensions-apiserver/pkg/apiserver"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -281,7 +283,7 @@ func CreateKubeAPIServerConfig(
 	[]admission.PluginInitializer,
 	error,
 ) {
-	genericConfig, versionedInformers, insecureServingInfo, serviceResolver, pluginInitializers, admissionPostStartHook, storageFactory, err := buildGenericConfig(s.ServerRunOptions, proxyTransport)
+	genericConfig, versionedInformers, insecureServingInfo, serviceResolver, serviceResolverPostStartHook, pluginInitializers, admissionPostStartHook, storageFactory, err := buildGenericConfig(s.ServerRunOptions, proxyTransport)
 	if err != nil {
 		return nil, nil, nil, nil, err
 	}
@@ -375,6 +377,12 @@ func CreateKubeAPIServerConfig(
 		return nil, nil, nil, nil, err
 	}
 
+	if serviceResolverPostStartHook != nil {
+		if err := config.GenericConfig.AddPostStartHook("start-kube-apiserver-service-resolver", serviceResolverPostStartHook); err != nil {
+			return nil, nil, nil, nil, err
+		}
+	}
+
 	if nodeTunneler != nil {
 		// Use the nodeTunneler's dialer to connect to the kubelet
 		config.ExtraConfig.KubeletClientConfig.Dial = nodeTunneler.Dial
@@ -422,6 +430,7 @@ func buildGenericConfig(
 	versionedInformers clientgoinformers.SharedInformerFactory,
 	insecureServingInfo *genericapiserver.DeprecatedInsecureServingInfo,
 	serviceResolver aggregatorapiserver.ServiceResolver,
+	serviceResolverPostStartHook genericapiserver.PostStartHookFunc,
 	pluginInitializers []admission.PluginInitializer,
 	admissionPostStartHook genericapiserver.PostStartHookFunc,
 	storageFactory *serverstorage.DefaultStorageFactory,
@@ -518,7 +527,7 @@ func buildGenericConfig(
 		LoopbackClientConfig: genericConfig.LoopbackClientConfig,
 		CloudConfigFile:      s.CloudProvider.CloudConfigFile,
 	}
-	serviceResolver = buildServiceResolver(s.EnableAggregatorRouting, genericConfig.LoopbackClientConfig.Host, versionedInformers)
+	serviceResolver, serviceResolverPostStartHook = buildServiceResolver(s.EnableAggregatorRouting, genericConfig.LoopbackClientConfig.Host, versionedInformers)
 
 	authInfoResolverWrapper := webhook.NewDefaultAuthenticationInfoResolverWrapper(proxyTransport, genericConfig.EgressSelector, genericConfig.LoopbackClientConfig)
 
@@ -725,12 +734,20 @@ func Complete(s *options.ServerRunOptions) (completedServerRunOptions, error) {
 	return options, nil
 }
 
-func buildServiceResolver(enabledAggregatorRouting bool, hostname string, informer clientgoinformers.SharedInformerFactory) webhook.ServiceResolver {
+func buildServiceResolver(enabledAggregatorRouting bool, hostname string, informer clientgoinformers.SharedInformerFactory) (webhook.ServiceResolver, genericapiserver.PostStartHookFunc) {
 	var serviceResolver webhook.ServiceResolver
+	var serviceResolverPostStartHook func(context genericapiserver.PostStartHookContext) error
 	if enabledAggregatorRouting {
-		serviceResolver = aggregatorapiserver.NewEndpointServiceResolver(
+		fd := failuredetector.NewDefaultFailureDetector()
+		serviceResolverPostStartHook = func(context genericapiserver.PostStartHookContext) error {
+			// TODO: wire context to PostStartHookContext or change the method signature to accept a chan
+			go fd.Run(gcontext.TODO())
+			return nil
+		}
+		serviceResolver = aggregatorapiserver.NewEndpointServiceResolverWithFailureDetector(
 			informer.Core().V1().Services().Lister(),
 			informer.Core().V1().Endpoints().Lister(),
+			fd,
 		)
 	} else {
 		serviceResolver = aggregatorapiserver.NewClusterIPServiceResolver(
@@ -741,7 +758,7 @@ func buildServiceResolver(enabledAggregatorRouting bool, hostname string, inform
 	if localHost, err := url.Parse(hostname); err == nil {
 		serviceResolver = aggregatorapiserver.NewLoopbackServiceResolver(serviceResolver, localHost)
 	}
-	return serviceResolver
+	return serviceResolver, serviceResolverPostStartHook
 }
 
 func getServiceIPAndRanges(serviceClusterIPRanges string) (net.IP, net.IPNet, net.IPNet, error) {

go.mod

@@ -97,6 +97,7 @@ require (
 	github.com/opencontainers/runc v1.0.0-rc10
 	github.com/opencontainers/runtime-spec v1.0.0 // indirect
 	github.com/opencontainers/selinux v1.3.1-0.20190929122143-5215b1806f52
+	github.com/p0lyn0mial/failure-detector v0.0.0-20200511131836-42e70bbc7eb0
 	github.com/pkg/errors v0.9.1
 	github.com/pmezard/go-difflib v1.0.0
 	github.com/pquerna/ffjson v0.0.0-20180717144149-af8b230fcd20 // indirect
@@ -424,6 +425,9 @@ replace (
 	github.com/opencontainers/runc => github.com/opencontainers/runc v1.0.0-rc10
 	github.com/opencontainers/runtime-spec => github.com/opencontainers/runtime-spec v1.0.0
 	github.com/opencontainers/selinux => github.com/opencontainers/selinux v1.3.1-0.20190929122143-5215b1806f52
+	github.com/p0lyn0mial/batch-working-queue => github.com/p0lyn0mial/batch-working-queue v0.0.0-20200511091501-d87326ed735a
+	github.com/p0lyn0mial/failure-detector => github.com/p0lyn0mial/failure-detector v0.0.0-20200511131836-42e70bbc7eb0
+	github.com/p0lyn0mial/ttl-cache => github.com/p0lyn0mial/ttl-cache v0.0.0-20200511091430-b21b42dbc05f
 	github.com/pelletier/go-toml => github.com/pelletier/go-toml v1.2.0
 	github.com/peterbourgon/diskv => github.com/peterbourgon/diskv v2.0.1+incompatible
 	github.com/pkg/errors => github.com/pkg/errors v0.9.1

go.sum

@@ -410,6 +410,12 @@ github.com/opencontainers/runtime-spec v1.0.0 h1:O6L965K88AilqnxeYPks/75HLpp4IG+
 github.com/opencontainers/runtime-spec v1.0.0/go.mod h1:jwyrGlmzljRJv/Fgzds9SsS/C5hL+LL3ko9hs6T5lQ0=
 github.com/opencontainers/selinux v1.3.1-0.20190929122143-5215b1806f52 h1:B8hYj3NxHmjsC3T+tnlZ1UhInqUgnyF1zlGPmzNg2Qk=
 github.com/opencontainers/selinux v1.3.1-0.20190929122143-5215b1806f52/go.mod h1:+BLncwf63G4dgOzykXAxcmnFlUaOlkDdmw/CqsW6pjs=
+github.com/p0lyn0mial/batch-working-queue v0.0.0-20200511091501-d87326ed735a h1:msrvrQTWCWOjUj2sJZhG8JrNeQ43PQsbVy13OnaGVuc=
+github.com/p0lyn0mial/batch-working-queue v0.0.0-20200511091501-d87326ed735a/go.mod h1:PTDbndC6P2gYYyqKtegbAWqprBd8/3KeeaWmZdxF8ZY=
+github.com/p0lyn0mial/failure-detector v0.0.0-20200511131836-42e70bbc7eb0 h1:x0rzTK1F1NBDFnqXuS690g67/gus8mLOVx3J39QbqKY=
+github.com/p0lyn0mial/failure-detector v0.0.0-20200511131836-42e70bbc7eb0/go.mod h1:BkbqMoib5l6c0rSJoh0ep1W+5ZrMbyFpDSQ/cr2trPk=
+github.com/p0lyn0mial/ttl-cache v0.0.0-20200511091430-b21b42dbc05f h1:plVsxZZZJqXa2pN1g54Mw/PAb3tvhmndXyN1zaU6IoA=
+github.com/p0lyn0mial/ttl-cache v0.0.0-20200511091430-b21b42dbc05f/go.mod h1:HcmP7670Zfkf8tjY2fDLgRN7onbUlR83lPx0oOBa2iM=
 github.com/pelletier/go-toml v1.2.0 h1:T5zMGML61Wp+FlcbWjRDT7yAxhJNAiPPLOFECq181zc=
 github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic=
 github.com/peterbourgon/diskv v2.0.1+incompatible h1:UBdAOUP5p4RWqPBg048CAvpKN+vxiaj6gdUUzhl4XmI=

staging/src/k8s.io/apimachinery/pkg/util/net/util.go

@@ -17,6 +17,7 @@ limitations under the License.
 package net
 
 import (
+	"errors"
 	"net"
 	"net/url"
 	"os"
@@ -40,13 +41,16 @@ func IPNetEqual(ipnet1, ipnet2 *net.IPNet) bool {
 
 // Returns if the given err is "connection reset by peer" error.
 func IsConnectionReset(err error) bool {
-	if urlErr, ok := err.(*url.Error); ok {
+	var urlErr *url.Error
+	if errors.As(err, &urlErr) {
 		err = urlErr.Err
 	}
-	if opErr, ok := err.(*net.OpError); ok {
+	var opErr *net.OpError
+	if errors.As(err, &opErr) {
 		err = opErr.Err
 	}
-	if osErr, ok := err.(*os.SyscallError); ok {
+	var osErr *os.SyscallError
+	if errors.As(err, &osErr) {
 		err = osErr.Err
 	}
 	if errno, ok := err.(syscall.Errno); ok && errno == syscall.ECONNRESET {
@@ -57,13 +61,16 @@ func IsConnectionReset(err error) bool {
 
 // Returns if the given err is "connection refused" error
 func IsConnectionRefused(err error) bool {
-	if urlErr, ok := err.(*url.Error); ok {
+	var urlErr *url.Error
+	if errors.As(err, &urlErr) {
 		err = urlErr.Err
 	}
-	if opErr, ok := err.(*net.OpError); ok {
+	var opErr *net.OpError
+	if errors.As(err, &opErr) {
 		err = opErr.Err
 	}
-	if osErr, ok := err.(*os.SyscallError); ok {
+	var osErr *os.SyscallError
+	if errors.As(err, &osErr) {
 		err = osErr.Err
 	}
 	if errno, ok := err.(syscall.Errno); ok && errno == syscall.ECONNREFUSED {

staging/src/k8s.io/apimachinery/pkg/util/proxy/transport.go

@@ -109,7 +109,7 @@ func (t *Transport) RoundTrip(req *http.Request) (*http.Response, error) {
 		}
 		resp.Header.Set("Content-Type", "text/plain; charset=utf-8")
 		resp.Header.Set("X-Content-Type-Options", "nosniff")
-		return resp, nil
+		return resp, err
 	}
 
 	if redirect := resp.Header.Get("Location"); redirect != "" {

staging/src/k8s.io/apimachinery/pkg/util/proxy/upgradeaware.go

@@ -232,6 +232,7 @@ func (h *UpgradeAwareHandler) ServeHTTP(w http.ResponseWriter, req *http.Request
 	proxy.Transport = h.Transport
 	proxy.FlushInterval = h.FlushInterval
 	proxy.ErrorLog = log.New(noSuppressPanicError{}, "", log.LstdFlags)
+	proxy.ErrorHandler = h.Responder.Error
 	proxy.ServeHTTP(w, newReq)
 }
 
@@ -412,12 +413,12 @@ func getResponse(r io.Reader) (*http.Response, []byte, error) {
 func dial(req *http.Request, transport http.RoundTripper) (net.Conn, error) {
 	conn, err := dialURL(req.Context(), req.URL, transport)
 	if err != nil {
-		return nil, fmt.Errorf("error dialing backend: %v", err)
+		return nil, fmt.Errorf("error dialing backend: %w", err)
 	}
 
 	if err = req.Write(conn); err != nil {
 		conn.Close()
-		return nil, fmt.Errorf("error sending request: %v", err)
+		return nil, fmt.Errorf("error sending request: %w", err)
 	}
 
 	return conn, err

staging/src/k8s.io/apiserver/pkg/util/proxy/BUILD

@@ -14,6 +14,7 @@ go_test(
         "//staging/src/k8s.io/api/core/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/intstr:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/util/sets:go_default_library",
         "//staging/src/k8s.io/client-go/listers/core/v1:go_default_library",
         "//staging/src/k8s.io/client-go/tools/cache:go_default_library",
     ],
@@ -27,6 +28,7 @@ go_library(
     deps = [
         "//staging/src/k8s.io/api/core/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/api/errors:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/util/sets:go_default_library",
         "//staging/src/k8s.io/client-go/listers/core/v1:go_default_library",
     ],
 )

staging/src/k8s.io/apiserver/pkg/util/proxy/proxy.go

@@ -25,6 +25,7 @@ import (
 
 	"k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/util/sets"
 	listersv1 "k8s.io/client-go/listers/core/v1"
 )
 
@@ -39,7 +40,8 @@ func findServicePort(svc *v1.Service, port int32) (*v1.ServicePort, error) {
 }
 
 // ResourceLocation returns a URL to which one can send traffic for the specified service.
-func ResolveEndpoint(services listersv1.ServiceLister, endpoints listersv1.EndpointsLister, namespace, id string, port int32) (*url.URL, error) {
+// TODO: update desc
+func ResolveEndpoint(services listersv1.ServiceLister, endpoints listersv1.EndpointsLister, namespace, id string, port int32, seenEndpoints ...*url.URL) (*url.URL, error) {
 	svc, err := services.Services(namespace).Get(id)
 	if err != nil {
 		return nil, err
@@ -68,6 +70,21 @@ func ResolveEndpoint(services listersv1.ServiceLister, endpoints listersv1.Endpo
 	// Pick a random Subset to start searching from.
 	ssSeed := rand.Intn(len(eps.Subsets))
 
+
+	seenEndpointsToAddressesFn := func(scheme, port string) (sets.String, error) {
+		ret := sets.String{}
+		for _, ep := range seenEndpoints {
+			if ep.Scheme == scheme && ep.Port() == port {
+				ip, _, err := net.SplitHostPort(ep.Host)
+				if err != nil {
+					return nil, err
+				}
+				ret.Insert(ip)
+			}
+		}
+		return ret, nil
+	}
+
 	// Find a Subset that has the port.
 	for ssi := 0; ssi < len(eps.Subsets); ssi++ {
 		ss := &eps.Subsets[(ssSeed+ssi)%len(eps.Subsets)]
@@ -77,8 +94,22 @@ func ResolveEndpoint(services listersv1.ServiceLister, endpoints listersv1.Endpo
 		for i := range ss.Ports {
 			if ss.Ports[i].Name == svcPort.Name {
 				// Pick a random address.
-				ip := ss.Addresses[rand.Intn(len(ss.Addresses))].IP
 				port := int(ss.Ports[i].Port)
+				seenAddresses, err := seenEndpointsToAddressesFn("https", strconv.Itoa(port))
+				if err != nil {
+					return nil, err
+				}
+				availablePoolOfAddresses := []string{}
+				for _, ep := range ss.Addresses {
+					if seenAddresses.Has(ep.IP) {
+						continue
+					}
+					availablePoolOfAddresses = append(availablePoolOfAddresses, ep.IP)
+				}
+				if len(availablePoolOfAddresses) == 0 {
+					continue
+				}
+				ip := availablePoolOfAddresses[rand.Intn(len(availablePoolOfAddresses))]
 				return &url.URL{
 					Scheme: "https",
 					Host:   net.JoinHostPort(ip, strconv.Itoa(port)),