feat: add v2 endpoints controller

Implement the basic requirements of a new Endpoints controller that
registers Services via Consul's V2 API.

Further tests and TODOs will be addressed in follow-up changes.

control-plane/connect-inject/common/common.go

@@ -9,10 +9,13 @@ import (
 	"strings"
 
 	mapset "github.com/deckarep/golang-set"
+	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/types/known/anypb"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"github.com/hashicorp/consul-k8s/control-plane/connect-inject/constants"
+	pbcatalog "github.com/hashicorp/consul/proto-public/pbcatalog/v1alpha1"
 )
 
 // DetermineAndValidatePort behaves as follows:
@@ -116,3 +119,40 @@ func ShouldIgnore(namespace string, denySet, allowSet mapset.Set) bool {
 func ConsulNodeNameFromK8sNode(nodeName string) string {
 	return fmt.Sprintf("%s-virtual", nodeName)
 }
+
+// ********************
+// V2 Exclusive Common Code
+// ********************
+
+// ToProtoAny is a convenience function for converting proto.Message values to anypb.Any without error handling.
+// This should _only_ be used in cases where a nil or valid proto.Message value is _guaranteed_, else it will panic.
+func ToProtoAny(m proto.Message) *anypb.Any {
+	if m == nil {
+		return nil
+	}
+	a, err := anypb.New(m)
+	if err != nil {
+		panic(fmt.Errorf("unexpected error: failed to convert proto message to anypb.Any: %w", err))
+	}
+	return a
+}
+
+// GetPortProtocol matches the Kubernetes EndpointPort.AppProtocol or ServicePort.AppProtocol (*string) to a supported
+// Consul catalog port protocol. If nil or unrecognized, the default of `PROTOCOL_UNSPECIFIED` is returned.
+func GetPortProtocol(appProtocol *string) pbcatalog.Protocol {
+	if appProtocol == nil {
+		return pbcatalog.Protocol_PROTOCOL_UNSPECIFIED
+	}
+	switch *appProtocol {
+	case "tcp":
+		return pbcatalog.Protocol_PROTOCOL_TCP
+	case "http":
+		return pbcatalog.Protocol_PROTOCOL_HTTP
+	case "http2":
+		return pbcatalog.Protocol_PROTOCOL_HTTP2
+	case "grpc":
+		return pbcatalog.Protocol_PROTOCOL_GRPC
+	}
+	// If unrecognized or empty string, return default
+	return pbcatalog.Protocol_PROTOCOL_UNSPECIFIED
+}

control-plane/connect-inject/common/common_test.go

@@ -7,12 +7,17 @@ import (
 	"testing"
 
 	mapset "github.com/deckarep/golang-set"
+	"github.com/google/go-cmp/cmp"
 	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/testing/protocmp"
+	"google.golang.org/protobuf/types/known/anypb"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"github.com/hashicorp/consul-k8s/control-plane/connect-inject/constants"
 	"github.com/hashicorp/consul-k8s/control-plane/namespaces"
+	pbcatalog "github.com/hashicorp/consul/proto-public/pbcatalog/v1alpha1"
+	"github.com/hashicorp/consul/proto-public/pbresource"
 )
 
 func TestCommonDetermineAndValidatePort(t *testing.T) {
@@ -314,3 +319,75 @@ func TestShouldIgnore(t *testing.T) {
 		})
 	}
 }
+
+func TestToProtoAny(t *testing.T) {
+	t.Parallel()
+
+	t.Run("nil gets nil", func(t *testing.T) {
+		require.Nil(t, ToProtoAny(nil))
+	})
+
+	t.Run("valid proto is successfully serialized", func(t *testing.T) {
+		testMsg := &pbresource.Resource{Id: &pbresource.ID{Name: "foo"}}
+		testAny, err := anypb.New(testMsg)
+		require.NoError(t, err)
+
+		if diff := cmp.Diff(testAny, ToProtoAny(testMsg), protocmp.Transform()); diff != "" {
+			t.Errorf("unexpected difference:\n%v", diff)
+		}
+	})
+}
+
+func TestGetPortProtocol(t *testing.T) {
+	t.Parallel()
+	toStringPtr := func(s string) *string {
+		return &s
+	}
+	cases := []struct {
+		name     string
+		input    *string
+		expected pbcatalog.Protocol
+	}{
+		{
+			name:     "nil gets UNSPECIFIED",
+			input:    nil,
+			expected: pbcatalog.Protocol_PROTOCOL_UNSPECIFIED,
+		},
+		{
+			name:     "tcp gets TCP",
+			input:    toStringPtr("tcp"),
+			expected: pbcatalog.Protocol_PROTOCOL_TCP,
+		},
+		{
+			name:     "http gets HTTP",
+			input:    toStringPtr("http"),
+			expected: pbcatalog.Protocol_PROTOCOL_HTTP,
+		},
+		{
+			name:     "http2 gets HTTP2",
+			input:    toStringPtr("http2"),
+			expected: pbcatalog.Protocol_PROTOCOL_HTTP2,
+		},
+		{
+			name:     "grpc gets GRPC",
+			input:    toStringPtr("grpc"),
+			expected: pbcatalog.Protocol_PROTOCOL_GRPC,
+		},
+		{
+			name:     "case sensitive",
+			input:    toStringPtr("gRPC"),
+			expected: pbcatalog.Protocol_PROTOCOL_UNSPECIFIED,
+		},
+		{
+			name:     "unknown gets UNSPECIFIED",
+			input:    toStringPtr("foo"),
+			expected: pbcatalog.Protocol_PROTOCOL_UNSPECIFIED,
+		},
+	}
+	for _, tt := range cases {
+		t.Run(tt.name, func(t *testing.T) {
+			actual := GetPortProtocol(tt.input)
+			require.Equal(t, tt.expected, actual)
+		})
+	}
+}

control-plane/connect-inject/constants/annotations_and_labels.go

@@ -203,6 +203,7 @@ const (
 	Enabled = "enabled"
 
 	// ManagedByValue is the value for keyManagedBy.
+	//TODO(zalimeni) rename this to ManagedByLegacyEndpointsValue
 	ManagedByValue = "consul-k8s-endpoints-controller"
 )
 
@@ -220,8 +221,13 @@ const (
 	// a pod after an injection is done.
 	KeyMeshInjectStatus = "consul.hashicorp.com/mesh-inject-status"
 
+	// ManagedByEndpointsValue is used in Consul metadata to identify the manager
+	// of resources. The 'v2' suffix is used to differentiate from the legacy
+	// endpoints controller of the same name.
+	ManagedByEndpointsValue = "consul-k8s-endpoints-controller-v2"
+
 	// ManagedByPodValue is used in Consul metadata to identify the manager
-	// of this resource.
+	// of resources.
 	ManagedByPodValue = "consul-k8s-pod-controller"
 )