Skip to content

Commit 79c230e

Browse files
Ly Caoachals
Ly Cao
authored and
achals
committed
added http json endpoint + add validation for feature names and entity keys for each requested feature view in GetOnlineFeatures
Signed-off-by: Felix Wang <wangfelix98@gmail.com> Signed-off-by: Achal Shah <achals@gmail.com>
1 parent fb186b7 commit 79c230e

7 files changed

Lines changed: 213 additions & 48 deletions

File tree

‎go/client_test/client.go‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,9 @@
1+
package main
2+
3+
import (
4+
"net/http"
5+
)
6+
7+
func main() {
8+
request :=
9+
}

‎go/feast/featurestore.go‎

Lines changed: 50 additions & 45 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,8 @@ import (
1010
timestamppb "google.golang.org/protobuf/types/known/timestamppb"
1111
"io/ioutil"
1212
"strings"
13+
"sort"
14+
"fmt"
1315
)
1416

1517
type FeatureStore struct {
@@ -62,52 +64,13 @@ func (fs *FeatureStore) GetOnlineFeatures(request *serving.GetOnlineFeaturesRequ
6264
}
6365

6466
requestEntities := request.GetEntities() // map[string]*types.RepeatedValue
65-
registryEntities := fs.registry.GetEntities() //[]*Entity
66-
entitiesInRegistry := make(map[string]bool) // used for validation of requested entities versus registry entities
67-
var requestEntitiesRowLength int
68-
69-
for _, values := range requestEntities {
70-
requestEntitiesRowLength = len(values.GetVal())
71-
break
72-
}
73-
for _, registryEntity := range registryEntities {
74-
// append(entities_in_registry, registry_entity.Spec.Name)
75-
entitiesInRegistry[registryEntity.Spec.Name] = true
76-
}
77-
joinKeyIndex := 0
78-
joinKeyToIndex := make(map[string]int)
79-
// Validate that all entities in request_entities are found in registry
80-
for entityName, values := range requestEntities {
81-
if _, ok := entitiesInRegistry[entityName]; !ok {
82-
return nil, errors.New("Requested entity not found inside the registry")
83-
}
84-
if len(values.GetVal()) != requestEntitiesRowLength {
85-
return nil, errors.New("Values of each Entity must have the same length")
86-
}
87-
joinKeyToIndex[entityName] = joinKeyIndex
88-
joinKeyIndex += 1
89-
}
67+
9068
// Construct a map of all feature_views to validate later
9169
registryFeatureViews := fs.registry.GetFeatureViews()
9270
featureViewsInRegistry := make(map[string]*core.FeatureView)
9371
for _, registryFeatureView := range registryFeatureViews {
9472
featureViewsInRegistry[registryFeatureView.Spec.Name] = registryFeatureView
9573
}
96-
numRequestJoinKeys := len(requestEntities)
97-
entityKeys := make([]types.EntityKey, requestEntitiesRowLength)
98-
for index, _ := range entityKeys {
99-
entityKey := types.EntityKey{ JoinKeys: make([]string, numRequestJoinKeys),
100-
EntityValues: make([]*types.Value, numRequestJoinKeys)}
101-
entityKeys[index] = entityKey
102-
}
103-
// Building entity keys
104-
for joinKey, values := range requestEntities {
105-
for rowEntityKeyIndex, value := range values.GetVal() {
106-
joinKeyIndex := joinKeyToIndex[joinKey]
107-
entityKeys[rowEntityKeyIndex].JoinKeys[joinKeyIndex] = joinKey
108-
entityKeys[rowEntityKeyIndex].EntityValues[joinKeyIndex] = value
109-
}
110-
}
11174

11275
response := serving.GetOnlineFeaturesResponse{Metadata: &serving.GetOnlineFeaturesResponseMetadata{FeatureNames: featureList},
11376
Results: make([]*serving.GetOnlineFeaturesResponse_FeatureVector, 0)}
@@ -123,12 +86,55 @@ func (fs *FeatureStore) GetOnlineFeatures(request *serving.GetOnlineFeaturesRequ
12386
// Obtain all join keys required by this feature view
12487
// and for each join key, create a EntityKey
12588
// and add to entity_keys
126-
entitiesRequired := featureViewSpec.GetEntities()
127-
for _, entityName := range entitiesRequired {
128-
if _, ok := requestEntities[entityName]; !ok {
129-
return nil, errors.New("All entities inside FeatureView must be provided")
89+
entitiesInFeatureView := featureViewSpec.GetEntities()
90+
sort.Strings(entitiesInFeatureView)
91+
featuresInFeatureView := featureViewSpec.GetFeatures()
92+
// Validate that all features asked for are inside this feature view
93+
featuresInFeatureViewMap := make(map[string]bool)
94+
for _, featureRef := range featuresInFeatureView {
95+
featuresInFeatureViewMap[featureRef.GetName()] = true
96+
}
97+
98+
for _, featureName := range allFeatures {
99+
if _, ok := featuresInFeatureViewMap[featureName]; !ok {
100+
return nil, errors.New(fmt.Sprintf("FeatureView: %s doesn't contain feature: %s\n", featureViewName, featureName))
130101
}
131102
}
103+
104+
var entityKeys []types.EntityKey
105+
// Construct EntityKeys
106+
if len(entitiesInFeatureView) > 0 {
107+
108+
if _, ok := requestEntities[entitiesInFeatureView[0]]; !ok {
109+
return nil, errors.New(fmt.Sprintf("EntityKey: %s is required for feature view: %s\n", entitiesInFeatureView[0], featureViewName))
110+
}
111+
requestEntitiesRowLength := len(requestEntities[entitiesInFeatureView[0]].GetVal())
112+
113+
numJoinKeysInFeatureView := len(entitiesInFeatureView)
114+
entityKeys = make([]types.EntityKey, requestEntitiesRowLength)
115+
for index, _ := range entityKeys {
116+
entityKey := types.EntityKey{ JoinKeys: make([]string, numJoinKeysInFeatureView),
117+
EntityValues: make([]*types.Value, numJoinKeysInFeatureView)}
118+
entityKeys[index] = entityKey
119+
}
120+
// Building entity keys for required for each Feature View from the Feature View's Spec
121+
for joinKeyIndex, joinKey := range entitiesInFeatureView {
122+
if values, ok := requestEntities[joinKey]; !ok {
123+
return nil, errors.New(fmt.Sprintf("EntityKey: %s is required for feature view: %s\n", joinKey, featureViewName))
124+
} else {
125+
// All requested entities must have the same number of rows
126+
if len(values.GetVal()) != requestEntitiesRowLength {
127+
return nil, errors.New("Values of each Entity must have the same length")
128+
}
129+
for rowEntityKeyIndex, value := range values.GetVal() {
130+
entityKeys[rowEntityKeyIndex].JoinKeys[joinKeyIndex] = joinKey
131+
entityKeys[rowEntityKeyIndex].EntityValues[joinKeyIndex] = value
132+
}
133+
}
134+
}
135+
136+
}
137+
132138

133139
features, err := fs.onlineStore.OnlineRead(entityKeys, featureViewName, allFeatures)
134140

@@ -149,7 +155,6 @@ func (fs *FeatureStore) GetOnlineFeatures(request *serving.GetOnlineFeaturesRequ
149155
} else if checkOutsideMaxAge(&feature.timestamp, timestamppb.Now(), featureViewSpec.GetTtl()) {
150156
status = serving.FieldStatus_OUTSIDE_MAX_AGE
151157
}
152-
153158
value := feature.value
154159
timeStamp := feature.timestamp
155160
featureVector.Values = append(featureVector.Values, &value)

‎go/server/main.go‎

Lines changed: 66 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,7 @@
11
package main
22

3+
// THIS WORKS
4+
35
import (
46
"fmt"
57
"github.com/feast-dev/feast/go/feast"
@@ -9,15 +11,25 @@ import (
911
"net"
1012
"os"
1113
"path/filepath"
14+
"net/http"
15+
"github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
16+
"google.golang.org/grpc/credentials/insecure"
17+
"context"
18+
"sync"
19+
"github.com/golang/glog"
20+
"strings"
1221
)
1322

1423
const (
1524
flagFeastRepoPath = "FEAST_REPO_PATH"
1625
flagFeastGrpcPort = "FEAST_GRPC_PORT"
1726
defaultFeastGrpcPort = "6566"
1827
feastServerVersion = "0.18.0"
28+
defaultFeastHttpPort = "8081"
1929
)
2030

31+
var wg sync.WaitGroup
32+
2133
func main() {
2234
repoPath := os.Getenv(flagFeastRepoPath)
2335
grpcPort, ok := os.LookupEnv(flagFeastGrpcPort)
@@ -44,9 +56,60 @@ func main() {
4456
}
4557
grpcServer := grpc.NewServer()
4658
serving.RegisterServingServiceServer(grpcServer, &server)
47-
err = grpcServer.Serve(lis)
48-
if err != nil {
59+
wg.Add(1)
60+
go func () {
61+
defer wg.Done()
62+
err = grpcServer.Serve(lis)
63+
if err != nil {
64+
log.Fatalln(err)
65+
}
66+
}()
67+
68+
// Implement HTTP server Endpoint
69+
log.Printf("Starting a HTTP server at port %s...", defaultFeastHttpPort)
70+
if err = runHttp(fmt.Sprintf(":%s", grpcPort)); err != nil {
4971
log.Fatalln(err)
5072
}
51-
// TODO: implement HTTP server endpoint
73+
wg.Wait()
74+
}
75+
76+
func runHttp(grpcServerEndpoint string) error {
77+
ctx := context.Background()
78+
ctx, cancel := context.WithCancel(ctx)
79+
defer cancel()
80+
81+
// Register gRPC server endpoint
82+
// Note: Make sure the gRPC server is running properly and accessible
83+
mux := runtime.NewServeMux()
84+
opts := []grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())}
85+
err := serving.RegisterServingServiceHandlerFromEndpoint(ctx, mux, grpcServerEndpoint, opts)
86+
if err != nil {
87+
return err
88+
}
89+
90+
return http.ListenAndServe(fmt.Sprintf(":%s", defaultFeastHttpPort), allowCORS(mux))
91+
}
92+
93+
func preflightHandler(w http.ResponseWriter, r *http.Request) {
94+
headers := []string{"Content-Type", "Accept"}
95+
w.Header().Set("Access-Control-Allow-Headers", strings.Join(headers, ","))
96+
methods := []string{"GET", "HEAD", "POST", "PUT", "DELETE"}
97+
w.Header().Set("Access-Control-Allow-Methods", strings.Join(methods, ","))
98+
glog.Infof("preflight request for %s", r.URL.Path)
99+
return
100+
}
101+
102+
// allowCORS allows Cross Origin Resoruce Sharing from any origin.
103+
// Don't do this without consideration in production systems.
104+
func allowCORS(h http.Handler) http.Handler {
105+
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
106+
if origin := r.Header.Get("Origin"); origin != "" {
107+
w.Header().Set("Access-Control-Allow-Origin", origin)
108+
if r.Method == "OPTIONS" && r.Header.Get("Access-Control-Request-Method") != "" {
109+
preflightHandler(w, r)
110+
return
111+
}
112+
}
113+
h.ServeHTTP(w, r)
114+
})
52115
}

‎go/server/server_http_test.go‎

Lines changed: 83 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,83 @@
1+
package main
2+
3+
import (
4+
"net/http"
5+
// "encoding/json"
6+
"github.com/feast-dev/feast/go/protos/feast/types"
7+
"github.com/feast-dev/feast/go/protos/feast/serving"
8+
"github.com/stretchr/testify/assert"
9+
"testing"
10+
"fmt"
11+
"io"
12+
"bytes"
13+
"google.golang.org/protobuf/encoding/protojson"
14+
)
15+
16+
type reqHttp struct {
17+
Features []string
18+
Entities map[string][]int64
19+
}
20+
21+
func TestServerHttp(t *testing.T) {
22+
featureViewNames := []string{"driver_hourly_stats:conv_rate",
23+
"driver_hourly_stats:acc_rate",
24+
"driver_hourly_stats:avg_daily_trips"}
25+
featureList := serving.FeatureList{Val: featureViewNames}
26+
featureListRequest := serving.GetOnlineFeaturesRequest_Features{Features: &featureList}
27+
entities := map[string]*types.RepeatedValue{"driver_id": {Val: []*types.Value{{Val: &types.Value_Int64Val{Int64Val: 1001}},
28+
{Val: &types.Value_Int64Val{Int64Val: 1002}},
29+
{Val: &types.Value_Int64Val{Int64Val: 1003}}}}}
30+
request := serving.GetOnlineFeaturesRequest{Kind: &featureListRequest, Entities: entities, FullFeatureNames: true}
31+
// request := { "Kind" : {
32+
// "Features" : {
33+
// "Val" : []string{ "driver_hourly_stats:conv_rate",
34+
// "driver_hourly_stats:acc_rate",
35+
// "driver_hourly_stats:avg_daily_trips"}
36+
// }
37+
// },
38+
// "Entities" : {
39+
// "driver_id": [1001, 1002, 1003]
40+
// },
41+
// }
42+
// request := reqHttp{ Features: []string{ "driver_hourly_stats:conv_rate",
43+
// "driver_hourly_stats:acc_rate",
44+
// "driver_hourly_stats:avg_daily_trips"},
45+
// Entities: map[string][]int64{"driver_id": []int64{1001, 1002, 1003} }}
46+
// requestBody, err := json.Marshal(request)
47+
requestBody, err := protojson.Marshal(&request)
48+
fmt.Println(string(requestBody))
49+
assert.Nil(t, err)
50+
resp, err := http.Post("http://localhost:8081/get-online-features", "application/json", bytes.NewBuffer(requestBody))
51+
if err != nil {
52+
panic(err)
53+
}
54+
assert.Nil(t, err)
55+
defer resp.Body.Close()
56+
fmt.Println("response status", resp.StatusCode)
57+
if resp.StatusCode == http.StatusOK {
58+
bodyBytes, err := io.ReadAll(resp.Body)
59+
assert.Nil(t, err)
60+
bodyString := string(bodyBytes)
61+
fmt.Println(bodyString)
62+
var response serving.GetOnlineFeaturesResponse
63+
if err = protojson.Unmarshal(bodyBytes, &response); err != nil {
64+
// panic(err)
65+
} else {
66+
// for _, featureVector := range response.Results {
67+
68+
// values := featureVector.GetValues()
69+
// statuses := featureVector.GetStatuses()
70+
// timestamps := featureVector.GetEventTimestamps()
71+
// lenValues := len(values)
72+
// for i := 0; i < lenValues; i++ {
73+
// fmt.Println(values[i].String(), statuses[i], timestamps[i].String())
74+
// }
75+
// }
76+
fmt.Println("Passed server_http_test")
77+
}
78+
} else {
79+
fmt.Println("response status", resp.StatusCode)
80+
}
81+
82+
assert.Nil(t, err)
83+
}

‎go/test_repo/.DS_Store‎

6 KB
Binary file not shown.

‎go/test_repo/data/registry.db‎

18 Bytes
Binary file not shown.

‎go/test_repo/post.sh‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
#!/bin/bash
2+
3+
curl -X POST http://localhost:8081/get-online-features -d '{"features":{"val":["driver_hourly_stats:conv_rate","driver_hourly_stats:acc_rate","driver_hourly_stats:avg_daily_trips"]},"entities":{"driver_id":{"val":[{"int64Val":"1001"},{"int64Val":"1002"},{"int64Val":"1003"}]}},"fullFeatureNames":true}'
4+
5+

0 commit comments

Comments
 (0)