Skip to content

Commit f05a775

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 1958df0 commit f05a775

5 files changed

Lines changed: 349 additions & 11 deletions

File tree

‎go/embedded/online_features.go‎

Lines changed: 53 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -239,15 +239,12 @@ func (s *OnlineFeatureService) StartGprcServerWithLoggingDefaultOpts(host string
239239
return s.StartGprcServerWithLogging(host, port, writeLoggedFeaturesCallback, defaultOpts)
240240
}
241241

242-
// StartGprcServerWithLogging starts gRPC server with enabled feature logging
243-
// Caller of this function must provide Python callback to flush buffered logs as well as logging configuration (loggingOpts)
244-
func (s *OnlineFeatureService) StartGprcServerWithLogging(host string, port int, writeLoggedFeaturesCallback logging.OfflineStoreWriteCallback, loggingOpts LoggingOptions) error {
242+
func (s *OnlineFeatureService) constructLoggingService(writeLoggedFeaturesCallback logging.OfflineStoreWriteCallback, loggingOpts LoggingOptions) (*logging.LoggingService, error) {
245243
var loggingService *logging.LoggingService = nil
246-
var err error
247244
if writeLoggedFeaturesCallback != nil {
248245
sink, err := logging.NewOfflineStoreSink(writeLoggedFeaturesCallback)
249246
if err != nil {
250-
return err
247+
return nil, err
251248
}
252249

253250
loggingService, err = logging.NewLoggingService(s.fs, sink, logging.LoggingOptions{
@@ -257,9 +254,19 @@ func (s *OnlineFeatureService) StartGprcServerWithLogging(host string, port int,
257254
FlushInterval: loggingOpts.FlushInterval,
258255
})
259256
if err != nil {
260-
return err
257+
return nil, err
261258
}
262259
}
260+
return loggingService, nil
261+
}
262+
263+
// StartGprcServerWithLogging starts gRPC server with enabled feature logging
264+
// Caller of this function must provide Python callback to flush buffered logs as well as logging configuration (loggingOpts)
265+
func (s *OnlineFeatureService) StartGprcServerWithLogging(host string, port int, writeLoggedFeaturesCallback logging.OfflineStoreWriteCallback, loggingOpts LoggingOptions) error {
266+
loggingService, err := s.constructLoggingService(writeLoggedFeaturesCallback, loggingOpts)
267+
if err != nil {
268+
return err
269+
}
263270
ser := server.NewGrpcServingServiceServer(s.fs, loggingService)
264271
log.Printf("Starting a gRPC server on host %s port %d\n", host, port)
265272
lis, err := net.Listen("tcp", fmt.Sprintf("%s:%d", host, port))
@@ -288,6 +295,46 @@ func (s *OnlineFeatureService) StartGprcServerWithLogging(host string, port int,
288295
return nil
289296
}
290297

298+
// StartHttpServer starts HTTP server with disabled feature logging and blocks the thread
299+
func (s *OnlineFeatureService) StartHttpServer(host string, port int) error {
300+
return s.StartHttpServerWithLogging(host, port, nil, LoggingOptions{})
301+
}
302+
303+
// StartHttpServerWithLoggingDefaultOpts starts HTTP server with enabled feature logging but default configuration for logging
304+
// Caller of this function must provide Python callback to flush buffered logs
305+
func (s *OnlineFeatureService) StartHttpServerWithLoggingDefaultOpts(host string, port int, writeLoggedFeaturesCallback logging.OfflineStoreWriteCallback) error {
306+
defaultOpts := LoggingOptions{
307+
ChannelCapacity: logging.DefaultOptions.ChannelCapacity,
308+
EmitTimeout: logging.DefaultOptions.EmitTimeout,
309+
WriteInterval: logging.DefaultOptions.WriteInterval,
310+
FlushInterval: logging.DefaultOptions.FlushInterval,
311+
}
312+
return s.StartHttpServerWithLogging(host, port, writeLoggedFeaturesCallback, defaultOpts)
313+
}
314+
315+
// StartHttpServerWithLogging starts HTTP server with enabled feature logging
316+
// Caller of this function must provide Python callback to flush buffered logs as well as logging configuration (loggingOpts)
317+
func (s *OnlineFeatureService) StartHttpServerWithLogging(host string, port int, writeLoggedFeaturesCallback logging.OfflineStoreWriteCallback, loggingOpts LoggingOptions) error {
318+
loggingService, err := s.constructLoggingService(writeLoggedFeaturesCallback, loggingOpts)
319+
if err != nil {
320+
return err
321+
}
322+
ser := server.NewHttpServer(s.fs, loggingService)
323+
log.Printf("Starting a HTTP server on host %s port %d\n", host, port)
324+
325+
go func() {
326+
// As soon as these signals are received from OS, try to gracefully stop the gRPC server
327+
<-s.grpcStopCh
328+
fmt.Println("Stopping the HTTP server...")
329+
err := ser.Stop()
330+
if err != nil {
331+
fmt.Printf("Error when stopping the HTTP server: %v\n", err)
332+
}
333+
}()
334+
335+
return ser.Serve(host, port)
336+
}
337+
291338
func (s *OnlineFeatureService) Stop() {
292339
s.grpcStopCh <- syscall.SIGINT
293340
}

‎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
@@ -610,17 +610,27 @@ def init_command(project_directory, minimal: bool, template: str):
610610
default=6566,
611611
help="Specify a port for the server [default: 6566]",
612612
)
613+
@click.option(
614+
"--type",
615+
"-t",
616+
"type_",
617+
type=click.STRING,
618+
default="http",
619+
help="Specify a server type: 'http' or 'grpc' [default: http]",
620+
)
613621
@click.option(
614622
"--no-access-log", is_flag=True, help="Disable the Uvicorn access log.",
615623
)
616624
@click.pass_context
617-
def serve_command(ctx: click.Context, host: str, port: int, no_access_log: bool):
625+
def serve_command(
626+
ctx: click.Context, host: str, port: int, type_: str, no_access_log: bool
627+
):
618628
"""Start a feature server locally on a given port."""
619629
repo = ctx.obj["CHDIR"]
620630
cli_check_repo(repo)
621631
store = FeatureStore(repo_path=str(repo))
622632

623-
store.serve(host, port, no_access_log)
633+
store.serve(host, port, type_, no_access_log)
624634

625635

626636
@cli.command("serve_transformations")

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

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -158,6 +158,28 @@ def start_grpc_server(
158158
else:
159159
self._service.StartGprcServer(host, port)
160160

161+
def start_http_server(
162+
self,
163+
host: str,
164+
port: int,
165+
enable_logging: bool = True,
166+
logging_options: Optional[LoggingOptions] = None,
167+
):
168+
if enable_logging:
169+
if logging_options:
170+
self._service.StartHttpServerWithLogging(
171+
host, port, self._logging_callback, logging_options
172+
)
173+
else:
174+
self._service.StartHttpServerWithLoggingDefaultOpts(
175+
host, port, self._logging_callback
176+
)
177+
else:
178+
self._service.StartHttpServer(host, port)
179+
180+
def stop_http_server(self):
181+
self._service.Stop()
182+
161183
def stop_grpc_server(self):
162184
self._service.Stop()
163185

0 commit comments

Comments
 (0)