fix(destination): return retry-able error on ip conflict (#15627)

When encountering a cluster ip conflict (multiple services mapping onto
a single IP) return 'Unavailable' versus 'FailedPrecondition.'

The FailedPrecondiction error code is used to indicate the error state
is terminal, and the client should not retry without an external change.

The Unavailable error code is used when the client can retry the call.

The conflict is the result of rescheduling a node, combined with the
underlying IPAM implementation re-provisioning the same address.

This change will result in the caller (proxy) issuing a retry w/
backoff.

Author: raykroeker

controller/api/destination/server.go

@@ -28,6 +28,12 @@ import (
 // updates buffered per stream before the stream is closed.
 const DefaultStreamQueueCapacity = 100
 
+var (
+	// errIPConflict is a sentinel error returned when multiple services are
+	// indexed by the same underlying IP address (clusterIP).
+	errIPConflict = status.Error(codes.Unavailable, "found multiple services with conflicting cluster IP")
+)
+
 type (
 	Config struct {
 		ControllerNS,
@@ -642,7 +648,7 @@ func getSvcID(k8sAPI *k8s.API, clusterIP string, log *logging.Entry) (*watcher.S
 			conflictingServices = append(conflictingServices, fmt.Sprintf("%s:%s", service.Namespace, service.Name))
 		}
 		log.Warnf("found conflicting %s cluster IP: %s", clusterIP, strings.Join(conflictingServices, ","))
-		return nil, status.Errorf(codes.FailedPrecondition, "found %d services with conflicting cluster IP %s", len(services), clusterIP)
+		return nil, errIPConflict
 	}
 	if len(services) == 0 {
 		return nil, nil

controller/api/destination/server_test.go

@@ -1277,6 +1277,55 @@ spec:
 			t.Fatalf("Expected not to find service mapped to [%s]", badClusterIP)
 		}
 	})
+
+	t.Run("Return Unavailable when services conflict on the same cluster IP", func(t *testing.T) {
+		conflictingClusterIP := "10.245.0.99"
+		k8sConfigs := []string{`
+apiVersion: v1
+kind: Service
+metadata:
+  name: service-a
+  namespace: test
+spec:
+  type: ClusterIP
+  clusterIP: 10.245.0.99
+  clusterIPs:
+  - 10.245.0.99
+  ports:
+  - port: 1234`, `
+apiVersion: v1
+kind: Service
+metadata:
+  name: service-b
+  namespace: test
+spec:
+  type: ClusterIP
+  clusterIP: 10.245.0.99
+  clusterIPs:
+  - 10.245.0.99
+  ports:
+  - port: 1234`}
+		k8sAPI, err := k8s.NewFakeAPI(k8sConfigs...)
+		if err != nil {
+			t.Fatalf("NewFakeAPI returned an error: %s", err)
+		}
+		err = watcher.InitializeIndexers(k8sAPI)
+		if err != nil {
+			t.Fatalf("InitializeIndexers returned an error: %s", err)
+		}
+		k8sAPI.Sync(nil)
+
+		svc, err := getSvcID(k8sAPI, conflictingClusterIP, logging.WithFields(nil))
+		if svc != nil {
+			t.Fatalf("Expected no service to be returned for conflicting cluster IP [%s]", conflictingClusterIP)
+		}
+		if err == nil {
+			t.Fatalf("Expected an error for conflicting cluster IP [%s]", conflictingClusterIP)
+		}
+		if code := status.Code(err); code != codes.Unavailable {
+			t.Fatalf("Expected error code %s, but got %s: %s", codes.Unavailable, code, err)
+		}
+	})
 }
 
 func testReturnEndpoints(t *testing.T, fqdn, ip string, port uint32) {