Skip to content

Commit f3367f2

Browse files
author
Tsotne Tabidze
committed
feat: Add http endpoint to the Go feature server
Signed-off-by: Tsotne Tabidze <tsotne@tecton.ai>
1 parent cff0133 commit f3367f2

5 files changed

Lines changed: 299 additions & 6 deletions

File tree

‎go/embedded/online_features.go‎

Lines changed: 19 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -216,7 +216,6 @@ func (s *OnlineFeatureService) GetOnlineFeatures(
216216

217217
func (s *OnlineFeatureService) StartGprcServer(host string, port int) error {
218218
// TODO(oleksii): enable logging
219-
// Disable logging for now
220219
var loggingService *logging.LoggingService = nil
221220
ser := server.NewGrpcServingServiceServer(s.fs, loggingService)
222221
log.Printf("Starting a gRPC server on host %s port %d\n", host, port)
@@ -243,6 +242,25 @@ func (s *OnlineFeatureService) StartGprcServer(host string, port int) error {
243242
return nil
244243
}
245244

245+
func (s *OnlineFeatureService) StartHttpServer(host string, port int) error {
246+
// TODO(oleksii): enable logging
247+
var loggingService *logging.LoggingService = nil
248+
ser := server.NewHttpServer(s.fs, loggingService)
249+
log.Printf("Starting a HTTP server on host %s port %d\n", host, port)
250+
251+
go func() {
252+
// As soon as these signals are received from OS, try to gracefully stop the gRPC server
253+
<-s.grpcStopCh
254+
fmt.Println("Stopping the HTTP server...")
255+
err := ser.Stop()
256+
if err != nil {
257+
fmt.Printf("Error when stopping the HTTP server: %v\n", err)
258+
}
259+
}()
260+
261+
return ser.Serve(host, port)
262+
}
263+
246264
func (s *OnlineFeatureService) Stop() {
247265
s.grpcStopCh <- syscall.SIGINT
248266
}

‎go/internal/feast/server/http_server.go‎

Lines changed: 248 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,248 @@
1+
package server
2+
3+
import (
4+
"context"
5+
"encoding/json"
6+
"fmt"
7+
"github.com/feast-dev/feast/go/internal/feast"
8+
"github.com/feast-dev/feast/go/internal/feast/model"
9+
"github.com/feast-dev/feast/go/internal/feast/server/logging"
10+
prototypes "github.com/feast-dev/feast/go/protos/feast/types"
11+
"net/http"
12+
)
13+
14+
type httpServer struct {
15+
fs *feast.FeatureStore
16+
loggingService *logging.LoggingService
17+
server *http.Server
18+
}
19+
20+
// Some Feast types aren't supported during JSON conversion
21+
type repeatedValue struct {
22+
stringVal []string
23+
int64Val []int64
24+
doubleVal []float64
25+
boolVal []bool
26+
stringListVal [][]string
27+
int64ListVal [][]int64
28+
doubleListVal [][]float64
29+
boolListVal [][]bool
30+
}
31+
32+
func (u *repeatedValue) UnmarshalJSON(data []byte) error {
33+
isString := false
34+
isDouble := false
35+
isInt64 := false
36+
isArray := false
37+
openBraketCounter := 0
38+
for _, b := range data {
39+
if b == '"' {
40+
isString = true
41+
}
42+
if b == '.' {
43+
isDouble = true
44+
}
45+
if b >= '0' && b <= '9' {
46+
isInt64 = true
47+
}
48+
if b == '[' {
49+
openBraketCounter++
50+
if openBraketCounter > 1 {
51+
isArray = true
52+
}
53+
}
54+
}
55+
var err error
56+
if !isArray {
57+
if isString {
58+
err = json.Unmarshal(data, &u.stringVal)
59+
} else if isDouble {
60+
err = json.Unmarshal(data, &u.doubleVal)
61+
} else if isInt64 {
62+
err = json.Unmarshal(data, &u.int64Val)
63+
} else {
64+
err = json.Unmarshal(data, &u.boolVal)
65+
}
66+
} else {
67+
if isString {
68+
err = json.Unmarshal(data, &u.stringListVal)
69+
} else if isDouble {
70+
err = json.Unmarshal(data, &u.doubleListVal)
71+
} else if isInt64 {
72+
err = json.Unmarshal(data, &u.int64ListVal)
73+
} else {
74+
err = json.Unmarshal(data, &u.boolListVal)
75+
}
76+
}
77+
return err
78+
}
79+
80+
func (u *repeatedValue) ToProto() *prototypes.RepeatedValue {
81+
proto := new(prototypes.RepeatedValue)
82+
if u.stringVal != nil {
83+
for _, val := range u.stringVal {
84+
proto.Val = append(proto.Val, &prototypes.Value{Val: &prototypes.Value_StringVal{StringVal: val}})
85+
}
86+
}
87+
if u.int64Val != nil {
88+
for _, val := range u.int64Val {
89+
proto.Val = append(proto.Val, &prototypes.Value{Val: &prototypes.Value_Int64Val{Int64Val: val}})
90+
}
91+
}
92+
if u.doubleVal != nil {
93+
for _, val := range u.doubleVal {
94+
proto.Val = append(proto.Val, &prototypes.Value{Val: &prototypes.Value_DoubleVal{DoubleVal: val}})
95+
}
96+
}
97+
if u.boolVal != nil {
98+
for _, val := range u.boolVal {
99+
proto.Val = append(proto.Val, &prototypes.Value{Val: &prototypes.Value_BoolVal{BoolVal: val}})
100+
}
101+
}
102+
if u.stringListVal != nil {
103+
for _, val := range u.stringListVal {
104+
proto.Val = append(proto.Val, &prototypes.Value{Val: &prototypes.Value_StringListVal{StringListVal: &prototypes.StringList{Val: val}}})
105+
}
106+
}
107+
if u.int64ListVal != nil {
108+
for _, val := range u.int64ListVal {
109+
proto.Val = append(proto.Val, &prototypes.Value{Val: &prototypes.Value_Int64ListVal{Int64ListVal: &prototypes.Int64List{Val: val}}})
110+
}
111+
}
112+
if u.doubleListVal != nil {
113+
for _, val := range u.doubleListVal {
114+
proto.Val = append(proto.Val, &prototypes.Value{Val: &prototypes.Value_DoubleListVal{DoubleListVal: &prototypes.DoubleList{Val: val}}})
115+
}
116+
}
117+
if u.boolListVal != nil {
118+
for _, val := range u.boolListVal {
119+
proto.Val = append(proto.Val, &prototypes.Value{Val: &prototypes.Value_BoolListVal{BoolListVal: &prototypes.BoolList{Val: val}}})
120+
}
121+
}
122+
return proto
123+
}
124+
125+
type getOnlineFeaturesRequest struct {
126+
FeatureService *string `json:"feature_service"`
127+
Features []string `json:"features"`
128+
Entities map[string]repeatedValue `json:"entities"`
129+
FullFeatureNames bool `json:"full_feature_names"`
130+
RequestContext map[string]repeatedValue `json:"request_context"`
131+
}
132+
133+
func NewHttpServer(fs *feast.FeatureStore, loggingService *logging.LoggingService) *httpServer {
134+
return &httpServer{fs: fs, loggingService: loggingService}
135+
}
136+
137+
func (s *httpServer) getOnlineFeatures(w http.ResponseWriter, r *http.Request) {
138+
if r.Method != "POST" {
139+
http.NotFound(w, r)
140+
return
141+
}
142+
143+
decoder := json.NewDecoder(r.Body)
144+
var request getOnlineFeaturesRequest
145+
err := decoder.Decode(&request)
146+
if err != nil {
147+
http.Error(w, fmt.Sprintf("Error decoding JSON request data: %+v", err), http.StatusInternalServerError)
148+
return
149+
}
150+
var featureService *model.FeatureService
151+
if request.FeatureService != nil {
152+
featureService, err = s.fs.GetFeatureService(*request.FeatureService)
153+
if err != nil {
154+
http.Error(w, fmt.Sprintf("Error getting feature service from registry: %+v", err), http.StatusInternalServerError)
155+
}
156+
}
157+
entitiesProto := make(map[string]*prototypes.RepeatedValue)
158+
for key, value := range request.Entities {
159+
entitiesProto[key] = value.ToProto()
160+
}
161+
requestContextProto := make(map[string]*prototypes.RepeatedValue)
162+
for key, value := range request.RequestContext {
163+
requestContextProto[key] = value.ToProto()
164+
}
165+
166+
fmt.Printf("features: %+v\n", request.Features)
167+
for _, feature := range request.Features {
168+
fmt.Printf(" feature %+v (type %T)\n", feature, feature)
169+
}
170+
fmt.Printf("feature_service: %+v\n", featureService)
171+
for key, value := range entitiesProto {
172+
fmt.Printf(" entity %s | repeatedValue %+v\n", key, value)
173+
}
174+
for key, value := range requestContextProto {
175+
fmt.Printf(" requestKey %s | repeatedValue %+v\n", key, value)
176+
}
177+
fmt.Printf("full_feature_names: %+v\n", request.FullFeatureNames)
178+
fmt.Println()
179+
180+
featureVectors, err := s.fs.GetOnlineFeatures(
181+
r.Context(),
182+
request.Features,
183+
featureService,
184+
entitiesProto,
185+
requestContextProto,
186+
request.FullFeatureNames)
187+
188+
if err != nil {
189+
http.Error(w, fmt.Sprintf("Error getting feature vector: %+v", err), http.StatusInternalServerError)
190+
}
191+
192+
fmt.Printf("featureVectors: %+v\n", featureVectors)
193+
var featureNames []string
194+
var results []map[string]interface{}
195+
for _, vector := range featureVectors {
196+
fmt.Printf(" featureVector: %+v\n", *vector)
197+
featureNames = append(featureNames, vector.Name)
198+
result := make(map[string]interface{})
199+
var statuses []string
200+
for _, status := range vector.Statuses {
201+
statuses = append(statuses, status.String())
202+
}
203+
var timestamps []string
204+
for _, timestamp := range vector.Timestamps {
205+
timestamps = append(timestamps, timestamp.String())
206+
}
207+
208+
result["statuses"] = statuses
209+
result["event_timestamps"] = timestamps
210+
// Note, that vector.Values is an Arrow Array, but this type implements JSON Marshaller.
211+
// So, it's not necessary to pre-process it in any way.
212+
result["values"] = vector.Values
213+
214+
results = append(results, result)
215+
}
216+
217+
response := map[string]interface{}{
218+
"metadata": map[string]interface{}{
219+
"feature_names": featureNames,
220+
},
221+
"results": results,
222+
}
223+
224+
err = json.NewEncoder(w).Encode(response)
225+
226+
if err != nil {
227+
http.Error(w, fmt.Sprintf("Error encoding response: %+v", err), http.StatusInternalServerError)
228+
}
229+
230+
w.Header().Set("Content-Type", "application/json")
231+
}
232+
233+
func (s *httpServer) Serve(host string, port int) error {
234+
s.server = &http.Server{Addr: fmt.Sprintf("%s:%d", host, port), Handler: nil}
235+
http.HandleFunc("/get-online-features", s.getOnlineFeatures)
236+
err := s.server.ListenAndServe()
237+
// Don't return the error if it's caused by graceful shutdown using Stop()
238+
if err == http.ErrServerClosed {
239+
return nil
240+
}
241+
return err
242+
}
243+
func (s *httpServer) Stop() error {
244+
if s.server != nil {
245+
return s.server.Shutdown(context.Background())
246+
}
247+
return nil
248+
}

‎sdk/python/feast/cli.py‎

Lines changed: 12 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -571,17 +571,27 @@ def init_command(project_directory, minimal: bool, template: str):
571571
default=6566,
572572
help="Specify a port for the server [default: 6566]",
573573
)
574+
@click.option(
575+
"--type",
576+
"-t",
577+
"type_",
578+
type=click.STRING,
579+
default="http",
580+
help="Specify a server type: 'http' or 'grpc' [default: http]",
581+
)
574582
@click.option(
575583
"--no-access-log", is_flag=True, help="Disable the Uvicorn access log.",
576584
)
577585
@click.pass_context
578-
def serve_command(ctx: click.Context, host: str, port: int, no_access_log: bool):
586+
def serve_command(
587+
ctx: click.Context, host: str, port: int, type_: str, no_access_log: bool
588+
):
579589
"""Start a feature server locally on a given port."""
580590
repo = ctx.obj["CHDIR"]
581591
cli_check_repo(repo)
582592
store = FeatureStore(repo_path=str(repo))
583593

584-
store.serve(host, port, no_access_log)
594+
store.serve(host, port, type_, no_access_log)
585595

586596

587597
@cli.command("serve_transformations")

‎sdk/python/feast/embedded_go/online_features_service.py‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -132,9 +132,15 @@ def get_online_features(
132132
resp = record_batch_to_online_response(record_batch)
133133
return OnlineResponse(resp)
134134

135+
def start_http_server(self, host: str, port: int):
136+
self._service.StartHttpServer(host, port)
137+
135138
def start_grpc_server(self, host: str, port: int):
136139
self._service.StartGprcServer(host, port)
137140

141+
def stop_http_server(self):
142+
self._service.Stop()
143+
138144
def stop_grpc_server(self):
139145
self._service.Stop()
140146

‎sdk/python/feast/feature_store.py‎

Lines changed: 14 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1959,14 +1959,25 @@ def _get_feature_views_to_use(
19591959
return views_to_use
19601960

19611961
@log_exceptions_and_usage
1962-
def serve(self, host: str, port: int, no_access_log: bool) -> None:
1962+
def serve(self, host: str, port: int, type_: str, no_access_log: bool) -> None:
19631963
"""Start the feature consumption server locally on a given port."""
1964+
type_ = type_.lower()
19641965
if self.config.go_feature_retrieval:
19651966
# Start go server instead of python if the flag is enabled
19661967
self._lazy_init_go_server()
1967-
# TODO(tsotne) add http/grpc flag in CLI and call appropriate method here depending on that
1968-
self._go_server.start_grpc_server(host, port)
1968+
if type_ == "http":
1969+
self._go_server.start_http_server(host, port)
1970+
elif type_ == "grpc":
1971+
self._go_server.start_grpc_server(host, port)
1972+
else:
1973+
raise ValueError(
1974+
f"Unsupported server type '{type_}'. Must be one of 'http' or 'grpc'."
1975+
)
19691976
else:
1977+
if type_ != "http":
1978+
raise ValueError(
1979+
f"Python server only supports 'http'. Got '{type_}' instead."
1980+
)
19701981
# Start the python server if go server isn't enabled
19711982
feature_server.start_server(self, host, port, no_access_log)
19721983

0 commit comments

Comments
 (0)