Skip to content

Commit f3ff812

Browse files
author
Tsotne Tabidze
authored
feat: CLI command 'feast serve' should start go-based server if flag is enabled (feast-dev#2617)
* CLI command 'feast serve' should start go-based server if flag is enabled Signed-off-by: Tsotne Tabidze <tsotne@tecton.ai> * Format stuff Signed-off-by: Tsotne Tabidze <tsotne@tecton.ai> * Split grpc & http methods in python and sort imports correctly Signed-off-by: Tsotne Tabidze <tsotne@tecton.ai>
1 parent 134dc5f commit f3ff812

File tree

21 files changed

+83
-101
lines changed

21 files changed

+83
-101
lines changed

‎Makefile‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -167,7 +167,7 @@ compile-protos-go: install-go-proto-dependencies install-protoc-dependencies
167167
cd sdk/python && python setup.py build_go_protos
168168

169169
compile-go-lib: install-go-proto-dependencies install-go-ci-dependencies
170-
cd sdk/python && python setup.py build_go_lib
170+
cd sdk/python && COMPILE_GO=True python setup.py build_ext --inplace
171171

172172
# Needs feast package to setup the feature store
173173
test-go: compile-protos-go
@@ -178,7 +178,7 @@ format-go:
178178
gofmt -s -w go/
179179

180180
lint-go: compile-protos-go
181-
go vet ./go/internal/feast ./go/cmd/server
181+
go vet ./go/internal/feast ./go/embedded
182182

183183
# Docker
184184

go/cmd/server/logging/feature_repo/__init__.py renamed to go/__init__.py

File renamed without changes.

‎go/cmd/server/logging/feature_repo/data/online_store.db‎

-16 KB
Binary file not shown.

‎go/cmd/server/main.go‎

Lines changed: 0 additions & 74 deletions
This file was deleted.

‎go/embedded/online_features.go‎

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,15 @@ package embedded
33
import (
44
"context"
55
"fmt"
6+
"github.com/feast-dev/feast/go/internal/feast/server"
7+
"github.com/feast-dev/feast/go/internal/feast/server/logging"
8+
"github.com/feast-dev/feast/go/protos/feast/serving"
9+
"google.golang.org/grpc"
610
"log"
11+
"net"
12+
"os"
13+
"os/signal"
14+
"syscall"
715

816
"github.com/apache/arrow/go/v8/arrow"
917
"github.com/apache/arrow/go/v8/arrow/array"
@@ -198,6 +206,36 @@ func (s *OnlineFeatureService) GetOnlineFeatures(
198206
return nil
199207
}
200208

209+
func (s *OnlineFeatureService) StartGprcServer(host string, port int) error {
210+
// TODO(oleksii): enable logging
211+
// Disable logging for now
212+
var loggingService *logging.LoggingService = nil
213+
ser := server.NewGrpcServingServiceServer(s.fs, loggingService)
214+
log.Printf("Starting a gRPC server on host %s port %d\n", host, port)
215+
lis, err := net.Listen("tcp", fmt.Sprintf("%s:%d", host, port))
216+
if err != nil {
217+
return err
218+
}
219+
grpcServer := grpc.NewServer()
220+
serving.RegisterServingServiceServer(grpcServer, ser)
221+
222+
// Notify this channel when receiving interrupt or termination signals from OS
223+
c := make(chan os.Signal, 1)
224+
signal.Notify(c, syscall.SIGINT, syscall.SIGTERM)
225+
go func() {
226+
// As soon as these signals are received from OS, try to gracefully stop the gRPC server
227+
<-c
228+
fmt.Println("Stopping the gRPC server...")
229+
grpcServer.GracefulStop()
230+
}()
231+
232+
err = grpcServer.Serve(lis)
233+
if err != nil {
234+
return err
235+
}
236+
return nil
237+
}
238+
201239
/*
202240
Read Record Batch from memory managed by Python caller.
203241
Python part uses C ABI interface to export this record into C Data Interface,

‎go/internal/__init__.py‎

Whitespace-only changes.

‎go/internal/feast/__init__.py‎

Whitespace-only changes.

‎go/internal/feast/onlinestore/onlinestore.go‎

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -43,7 +43,8 @@ type OnlineStore interface {
4343

4444
func getOnlineStoreType(onlineStoreConfig map[string]interface{}) (string, bool) {
4545
if onlineStoreType, ok := onlineStoreConfig["type"]; !ok {
46-
return "", false
46+
// If online store type isn't specified, default to sqlite
47+
return "sqlite", true
4748
} else {
4849
result, ok := onlineStoreType.(string)
4950
return result, ok
@@ -53,10 +54,11 @@ func getOnlineStoreType(onlineStoreConfig map[string]interface{}) (string, bool)
5354
func NewOnlineStore(config *registry.RepoConfig) (OnlineStore, error) {
5455
onlineStoreType, ok := getOnlineStoreType(config.OnlineStore)
5556
if !ok {
57+
return nil, fmt.Errorf("could not get online store type from online store config: %+v", config.OnlineStore)
58+
} else if onlineStoreType == "sqlite" {
5659
onlineStore, err := NewSqliteOnlineStore(config.Project, config, config.OnlineStore)
5760
return onlineStore, err
58-
}
59-
if onlineStoreType == "redis" {
61+
} else if onlineStoreType == "redis" {
6062
onlineStore, err := NewRedisOnlineStore(config.Project, config.OnlineStore)
6163
return onlineStore, err
6264
} else {

go/cmd/server/server.go renamed to go/internal/feast/server/grpc_server.go

Lines changed: 10 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -1,27 +1,29 @@
1-
package main
1+
package server
22

33
import (
44
"context"
55

6-
"github.com/feast-dev/feast/go/cmd/server/logging"
76
"github.com/feast-dev/feast/go/internal/feast"
7+
"github.com/feast-dev/feast/go/internal/feast/server/logging"
88
"github.com/feast-dev/feast/go/protos/feast/serving"
99
prototypes "github.com/feast-dev/feast/go/protos/feast/types"
1010
"github.com/feast-dev/feast/go/types"
1111
"github.com/google/uuid"
1212
)
1313

14-
type servingServiceServer struct {
14+
const feastServerVersion = "0.0.1"
15+
16+
type grpcServingServiceServer struct {
1517
fs *feast.FeatureStore
1618
loggingService *logging.LoggingService
1719
serving.UnimplementedServingServiceServer
1820
}
1921

20-
func newServingServiceServer(fs *feast.FeatureStore, loggingService *logging.LoggingService) *servingServiceServer {
21-
return &servingServiceServer{fs: fs, loggingService: loggingService}
22+
func NewGrpcServingServiceServer(fs *feast.FeatureStore, loggingService *logging.LoggingService) *grpcServingServiceServer {
23+
return &grpcServingServiceServer{fs: fs, loggingService: loggingService}
2224
}
2325

24-
func (s *servingServiceServer) GetFeastServingInfo(ctx context.Context, request *serving.GetFeastServingInfoRequest) (*serving.GetFeastServingInfoResponse, error) {
26+
func (s *grpcServingServiceServer) GetFeastServingInfo(ctx context.Context, request *serving.GetFeastServingInfoRequest) (*serving.GetFeastServingInfoResponse, error) {
2527
return &serving.GetFeastServingInfoResponse{
2628
Version: feastServerVersion,
2729
}, nil
@@ -30,7 +32,7 @@ func (s *servingServiceServer) GetFeastServingInfo(ctx context.Context, request
3032
// Returns an object containing the response to GetOnlineFeatures.
3133
// Metadata contains featurenames that corresponds to the number of rows in response.Results.
3234
// Results contains values including the value of the feature, the event timestamp, and feature status in a columnar format.
33-
func (s *servingServiceServer) GetOnlineFeatures(ctx context.Context, request *serving.GetOnlineFeaturesRequest) (*serving.GetOnlineFeaturesResponse, error) {
35+
func (s *grpcServingServiceServer) GetOnlineFeatures(ctx context.Context, request *serving.GetOnlineFeaturesRequest) (*serving.GetOnlineFeaturesResponse, error) {
3436
requestId := GenerateRequestId()
3537
featuresOrService, err := s.fs.ParseFeatures(request.GetKind())
3638
if err != nil {
@@ -74,7 +76,7 @@ func (s *servingServiceServer) GetOnlineFeatures(ctx context.Context, request *s
7476
EventTimestamps: vector.Timestamps,
7577
})
7678
}
77-
if featuresOrService.FeatureService != nil {
79+
if featuresOrService.FeatureService != nil && s.loggingService != nil {
7880
go s.loggingService.GenerateLogs(featuresOrService.FeatureService, entityValuesMap, resp.Results[len(request.Entities):], request.RequestContext, requestId)
7981
}
8082
return resp, nil

go/cmd/server/server_test.go renamed to go/internal/feast/server/grpc_server_test.go

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
package main
1+
package server
22

33
import (
44
"context"
@@ -16,8 +16,8 @@ import (
1616
"github.com/apache/arrow/go/v8/arrow/memory"
1717
"github.com/apache/arrow/go/v8/parquet/file"
1818
"github.com/apache/arrow/go/v8/parquet/pqarrow"
19-
"github.com/feast-dev/feast/go/cmd/server/logging"
2019
"github.com/feast-dev/feast/go/internal/feast"
20+
"github.com/feast-dev/feast/go/internal/feast/server/logging"
2121
"github.com/feast-dev/feast/go/internal/test"
2222
"github.com/feast-dev/feast/go/protos/feast/serving"
2323
"github.com/feast-dev/feast/go/protos/feast/types"
@@ -73,7 +73,7 @@ func getClient(ctx context.Context, offlineStoreType string, basePath string, en
7373
if err != nil {
7474
panic(err)
7575
}
76-
servingServiceServer := newServingServiceServer(fs, loggingService)
76+
servingServiceServer := NewGrpcServingServiceServer(fs, loggingService)
7777

7878
serving.RegisterServingServiceServer(server, servingServiceServer)
7979
go func() {

0 commit comments

Comments
 (0)