feat: Add Grpc registry store for Go feature server

go/internal/feast/registry/grpc.go

@@ -0,0 +1,147 @@
+package registry
+
+import (
+	"context"
+	"crypto/tls"
+	"crypto/x509"
+	"fmt"
+	"os"
+	"strings"
+
+	"github.com/feast-dev/feast/go/protos/feast/core"
+	registryPb "github.com/feast-dev/feast/go/protos/feast/registry"
+	"github.com/rs/zerolog/log"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/protobuf/types/known/emptypb"
+)
+
+type GrpcRegistryStore struct {
+	project string
+	client  registryPb.RegistryServerClient
+	conn    *grpc.ClientConn
+}
+
+// NewGrpcRegistryStore creates a gRPC-backed registry store.
+//
+// TLS is enabled when any of the following are true (mirrors Python RemoteRegistryConfig):
+//   - config.Path uses the "grpcs://" scheme
+//   - config.IsTls is true
+//   - config.Cert is set (path to a PEM certificate file)
+//
+// Without TLS, the connection is insecure. config.Path may be bare "host:port",
+// "grpc://host:port", or "grpcs://host:port".
+func NewGrpcRegistryStore(config *RegistryConfig, project string) (*GrpcRegistryStore, error) {
+	target, schemeIsTLS := parseGrpcTarget(config.Path)
+	useTLS := schemeIsTLS || config.IsTls || config.Cert != ""
+
+	var dialOpts []grpc.DialOption
+	if useTLS {
+		tlsCreds, err := buildTLSCredentials(config.Cert)
+		if err != nil {
+			return nil, err
+		}
+		dialOpts = append(dialOpts, grpc.WithTransportCredentials(tlsCreds))
+	} else {
+		dialOpts = append(dialOpts, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	}
+
+	conn, err := grpc.NewClient(target, dialOpts...)
+	if err != nil {
+		return nil, err
+	}
+
+	log.Info().Msgf("Using gRPC Feature Registry: %s", config.Path)
+	return &GrpcRegistryStore{
+		project: project,
+		client:  registryPb.NewRegistryServerClient(conn),
+		conn:    conn,
+	}, nil
+}
+
+// buildTLSCredentials returns TLS transport credentials. If certPath is non-empty
+// the PEM file is loaded as a custom root CA (for self-signed certs), otherwise
+// the system certificate pool is used.
+func buildTLSCredentials(certPath string) (credentials.TransportCredentials, error) {
+	if certPath == "" {
+		return credentials.NewTLS(&tls.Config{}), nil
+	}
+	pemBytes, err := os.ReadFile(certPath)
+	if err != nil {
+		return nil, fmt.Errorf("grpc registry: reading cert file %q: %w", certPath, err)
+	}
+	pool := x509.NewCertPool()
+	if !pool.AppendCertsFromPEM(pemBytes) {
+		return nil, fmt.Errorf("grpc registry: no valid PEM certificates found in %q", certPath)
+	}
+	return credentials.NewTLS(&tls.Config{RootCAs: pool}), nil
+}
+
+// parseGrpcTarget strips grpc:// or grpcs:// scheme prefixes and returns
+// the target address along with whether TLS should be used.
+func parseGrpcTarget(path string) (target string, useTLS bool) {
+	if strings.HasPrefix(path, "https://") {
+		return strings.TrimPrefix(path, "https://"), true
+	}
+	if strings.HasPrefix(path, "http://") {
+		return strings.TrimPrefix(path, "http://"), false
+	}
+	return path, false
+}
+
+func (g *GrpcRegistryStore) getEntity(name string, allowCache bool) (*core.Entity, error) {
+	return g.client.GetEntity(context.Background(), &registryPb.GetEntityRequest{
+		Name:       name,
+		Project:    g.project,
+		AllowCache: allowCache,
+	})
+}
+
+func (g *GrpcRegistryStore) getFeatureView(name string, allowCache bool) (*core.FeatureView, error) {
+	return g.client.GetFeatureView(context.Background(), &registryPb.GetFeatureViewRequest{
+		Name:       name,
+		Project:    g.project,
+		AllowCache: allowCache,
+	})
+}
+
+func (g *GrpcRegistryStore) getSortedFeatureView(name string, allowCache bool) (*core.SortedFeatureView, error) {
+	return g.client.GetSortedFeatureView(context.Background(), &registryPb.GetSortedFeatureViewRequest{
+		Name:       name,
+		Project:    g.project,
+		AllowCache: allowCache,
+	})
+}
+
+func (g *GrpcRegistryStore) getOnDemandFeatureView(name string, allowCache bool) (*core.OnDemandFeatureView, error) {
+	return g.client.GetOnDemandFeatureView(context.Background(), &registryPb.GetOnDemandFeatureViewRequest{
+		Name:       name,
+		Project:    g.project,
+		AllowCache: allowCache,
+	})
+}
+
+func (g *GrpcRegistryStore) getFeatureService(name string, allowCache bool) (*core.FeatureService, error) {
+	return g.client.GetFeatureService(context.Background(), &registryPb.GetFeatureServiceRequest{
+		Name:       name,
+		Project:    g.project,
+		AllowCache: allowCache,
+	})
+}
+
+func (g *GrpcRegistryStore) GetRegistryProto() (*core.Registry, error) {
+	return g.client.Proto(context.Background(), &emptypb.Empty{})
+}
+
+func (g *GrpcRegistryStore) UpdateRegistryProto(rp *core.Registry) error {
+	return &NotImplementedError{FunctionName: "UpdateRegistryProto"}
+}
+
+func (g *GrpcRegistryStore) Teardown() error {
+	return g.conn.Close()
+}
+
+func (g *GrpcRegistryStore) HasFallback() bool {
+	return true
+}

go/internal/feast/registry/grpc_test.go

@@ -0,0 +1,191 @@
+//go:build !integration
+
+package registry
+
+import (
+	"context"
+	"net"
+	"testing"
+
+	"github.com/feast-dev/feast/go/protos/feast/core"
+	registryPb "github.com/feast-dev/feast/go/protos/feast/registry"
+	"github.com/stretchr/testify/assert"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/connectivity"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/test/bufconn"
+	"google.golang.org/protobuf/types/known/emptypb"
+)
+
+// testRegistryServer is a minimal RegistryServer implementation for testing.
+type testRegistryServer struct {
+	registryPb.UnimplementedRegistryServerServer
+}
+
+func (s *testRegistryServer) GetEntity(_ context.Context, req *registryPb.GetEntityRequest) (*core.Entity, error) {
+	return &core.Entity{Spec: &core.EntitySpecV2{Name: req.Name}}, nil
+}
+
+func (s *testRegistryServer) GetFeatureView(_ context.Context, req *registryPb.GetFeatureViewRequest) (*core.FeatureView, error) {
+	return &core.FeatureView{Spec: &core.FeatureViewSpec{Name: req.Name}}, nil
+}
+
+func (s *testRegistryServer) GetSortedFeatureView(_ context.Context, req *registryPb.GetSortedFeatureViewRequest) (*core.SortedFeatureView, error) {
+	return &core.SortedFeatureView{Spec: &core.SortedFeatureViewSpec{Name: req.Name}}, nil
+}
+
+func (s *testRegistryServer) GetOnDemandFeatureView(_ context.Context, req *registryPb.GetOnDemandFeatureViewRequest) (*core.OnDemandFeatureView, error) {
+	return &core.OnDemandFeatureView{Spec: &core.OnDemandFeatureViewSpec{Name: req.Name}}, nil
+}
+
+func (s *testRegistryServer) GetFeatureService(_ context.Context, req *registryPb.GetFeatureServiceRequest) (*core.FeatureService, error) {
+	return &core.FeatureService{Spec: &core.FeatureServiceSpec{Name: req.Name}}, nil
+}
+
+func (s *testRegistryServer) Proto(_ context.Context, _ *emptypb.Empty) (*core.Registry, error) {
+	return &core.Registry{}, nil
+}
+
+// newTestGrpcStore spins up an in-process gRPC server using bufconn and returns
+// a GrpcRegistryStore backed by it, along with a cleanup function.
+func newTestGrpcStore(t *testing.T, project string) (*GrpcRegistryStore, func()) {
+	t.Helper()
+
+	listener := bufconn.Listen(1024 * 1024)
+	srv := grpc.NewServer()
+	registryPb.RegisterRegistryServerServer(srv, &testRegistryServer{})
+	go func() {
+		if err := srv.Serve(listener); err != nil && err != grpc.ErrServerStopped {
+			t.Errorf("bufconn server error: %v", err)
+		}
+	}()
+
+	conn, err := grpc.NewClient(
+		"passthrough:///bufconn",
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+		grpc.WithContextDialer(func(ctx context.Context, _ string) (net.Conn, error) {
+			return listener.DialContext(ctx)
+		}),
+	)
+	if err != nil {
+		t.Fatalf("failed to create grpc client: %v", err)
+	}
+
+	store := &GrpcRegistryStore{
+		project: project,
+		client:  registryPb.NewRegistryServerClient(conn),
+		conn:    conn,
+	}
+
+	cleanup := func() {
+		conn.Close()
+		srv.Stop()
+		listener.Close()
+	}
+	return store, cleanup
+}
+
+func TestGrpcGetEntity(t *testing.T) {
+	store, cleanup := newTestGrpcStore(t, "test_project")
+	defer cleanup()
+
+	result, err := store.getEntity("test_entity", true)
+
+	assert.Nil(t, err)
+	assert.Equal(t, "test_entity", result.Spec.Name)
+}
+
+func TestGrpcGetFeatureView(t *testing.T) {
+	store, cleanup := newTestGrpcStore(t, "test_project")
+	defer cleanup()
+
+	result, err := store.getFeatureView("test_feature_view", true)
+
+	assert.Nil(t, err)
+	assert.Equal(t, "test_feature_view", result.Spec.Name)
+}
+
+func TestGrpcGetSortedFeatureView(t *testing.T) {
+	store, cleanup := newTestGrpcStore(t, "test_project")
+	defer cleanup()
+
+	result, err := store.getSortedFeatureView("test_sorted_view", true)
+
+	assert.Nil(t, err)
+	assert.Equal(t, "test_sorted_view", result.Spec.Name)
+}
+
+func TestGrpcGetOnDemandFeatureView(t *testing.T) {
+	store, cleanup := newTestGrpcStore(t, "test_project")
+	defer cleanup()
+
+	result, err := store.getOnDemandFeatureView("test_odfv", true)
+
+	assert.Nil(t, err)
+	assert.Equal(t, "test_odfv", result.Spec.Name)
+}
+
+func TestGrpcGetFeatureService(t *testing.T) {
+	store, cleanup := newTestGrpcStore(t, "test_project")
+	defer cleanup()
+
+	result, err := store.getFeatureService("test_feature_service", true)
+
+	assert.Nil(t, err)
+	assert.Equal(t, "test_feature_service", result.Spec.Name)
+}
+
+func TestGrpcGetRegistryProto(t *testing.T) {
+	store, cleanup := newTestGrpcStore(t, "test_project")
+	defer cleanup()
+
+	registry, err := store.GetRegistryProto()
+
+	assert.Nil(t, err)
+	assert.IsType(t, &core.Registry{}, registry)
+}
+
+func TestGrpcUpdateRegistryProtoNotImplemented(t *testing.T) {
+	store, cleanup := newTestGrpcStore(t, "test_project")
+	defer cleanup()
+
+	err := store.UpdateRegistryProto(&core.Registry{})
+
+	assert.NotNil(t, err)
+	assert.IsType(t, &NotImplementedError{}, err)
+	assert.Equal(t, "UpdateRegistryProto", err.(*NotImplementedError).FunctionName)
+}
+
+func TestGrpcHasFallback(t *testing.T) {
+	store, cleanup := newTestGrpcStore(t, "test_project")
+	defer cleanup()
+
+	assert.True(t, store.HasFallback())
+}
+
+func TestGrpcTeardown(t *testing.T) {
+	store, cleanup := newTestGrpcStore(t, "test_project")
+	defer cleanup()
+
+	err := store.Teardown()
+
+	assert.Nil(t, err)
+	assert.Equal(t, connectivity.Shutdown, store.conn.GetState())
+}
+
+func TestParseGrpcTarget(t *testing.T) {
+	cases := []struct {
+		input  string
+		target string
+		useTLS bool
+	}{
+		{"http://host:50051", "host:50051", false},
+		{"https://host:50051", "host:50051", true},
+		{"host:50051", "host:50051", false},
+	}
+	for _, c := range cases {
+		target, useTLS := parseGrpcTarget(c.input)
+		assert.Equal(t, c.target, target, "input: %s", c.input)
+		assert.Equal(t, c.useTLS, useTLS, "input: %s", c.input)
+	}
+}