Fix

Signed-off-by: Kevin Zhang <kzhang@tecton.ai>
feast-dev · feast-ci-bot · Apr 13, 2022 · Mar 9, 2022 · Mar 11, 2022 · Mar 17, 2022
commit a4587c326134ec9a532245799cbde1fc652f9561
@@ -31,12 +31,6 @@ func GetFileConfig(config *registry.RepoConfig) (*OfflineLogStoreConfig, error)
 		fileConfig.path = path
 	} else {
 		return nil, errors.New("need path for file log storage")
-		// absPath, err := filepath.Abs(filepath.Join(".", "log.parquet"))
-		// if err != nil {
-		// 	return nil, err
-		// }
-		// fileConfig.path = absPath
-		// return &fileConfig, nil
 	}
 	return &fileConfig, nil
 }
@@ -60,7 +54,7 @@ func NewFileOfflineStore(project string, offlineStoreConfig *OfflineLogStoreConf
 	return &store, nil
 }
 
-func CreateOrOpenLogFile(absPath string) (*os.File, error) {
+func OpenLogFile(absPath string) (*os.File, error) {
 	var _, err = os.Stat(absPath)
 
 	// create file if not exists
@@ -71,16 +65,12 @@ func CreateOrOpenLogFile(absPath string) (*os.File, error) {
 		}
 		return file, nil
 	} else {
-		var file, err = os.OpenFile(absPath, os.O_RDWR, 0644)
-		if err != nil {
-			return nil, err
-		}
-		return file, nil
+		return nil, fmt.Errorf("path %s already exists", absPath)
 	}
 }
 
 func (f *FileLogStorage) FlushToStorage(tbl array.Table) error {
-	w, err := CreateOrOpenLogFile(f.path)
+	w, err := OpenLogFile(f.path)
 	var writer io.Writer = w
 	if err != nil {
 		return err

@@ -1,9 +1,9 @@
 package logging
 
 import (
+	"context"
 	"path/filepath"
 
-	"context"
 	"testing"
 
 	"github.com/apache/arrow/go/v8/arrow/array"
@@ -15,6 +15,7 @@ import (
 )
 
 func TestFlushToStorage(t *testing.T) {
+	ctx := context.Background()
 	table, expectedSchema, expectedColumns, err := GetTestArrowTableAndExpectedResults()
 	defer table.Release()
 	assert.Nil(t, err)
@@ -28,11 +29,7 @@ func TestFlushToStorage(t *testing.T) {
 	assert.Nil(t, err)
 	logPath, err := filepath.Abs(offlineStoreConfig.path)
 	assert.Nil(t, err)
-	w, err := CreateOrOpenLogFile(logPath)
-	assert.Nil(t, err)
-	ctx := context.Background()
-
-	pf, err := file.NewParquetReader(w)
+	pf, err := file.OpenParquetFile(logPath, false)
 	assert.Nil(t, err)
 
 	reader, err := pqarrow.NewFileReader(pf, pqarrow.ArrowReadProperties{}, memory.DefaultAllocator)

@@ -17,6 +17,9 @@ import (
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
+const DEFAULT_LOG_FLUSH_INTERVAL = 100 * time.Millisecond
+const DEFAULT_LOG_INSERT_TIMEOUT = 20 * time.Millisecond
+
 type Log struct {
 	// Example: val{int64_val: 5017}, val{int64_val: 1003}
 	EntityValue []*types.Value
@@ -37,16 +40,19 @@ type LoggingService struct {
 	logChannel        chan *Log
 	fs                *feast.FeatureStore
 	offlineLogStorage OfflineLogStorage
+	logInsertTTl      time.Duration
+	logFlushInterval  time.Duration
 }
 
-func NewLoggingService(fs *feast.FeatureStore, logChannelCapacity int, featureServiceName string, enableLogging bool) (*LoggingService, error) {
+func NewLoggingService(fs *feast.FeatureStore, logChannelCapacity int, featureServiceName string, enableLogProcessing bool) (*LoggingService, error) {
 	var featureService *model.FeatureService = nil
 	var err error
-	if enableLogging {
+	if fs != nil {
 		featureService, err = fs.GetFeatureService(featureServiceName)
 		if err != nil {
 			return nil, err
 		}
+
 	}
 
 	loggingService := &LoggingService{
@@ -55,20 +61,23 @@ func NewLoggingService(fs *feast.FeatureStore, logChannelCapacity int, featureSe
 			logs:           make([]*Log, 0),
 			featureService: featureService,
 		},
-		fs: fs,
+		fs:               fs,
+		logInsertTTl:     DEFAULT_LOG_INSERT_TIMEOUT,
+		logFlushInterval: DEFAULT_LOG_FLUSH_INTERVAL,
 	}
 
-	if !enableLogging || fs == nil {
-		loggingService.offlineLogStorage = nil
-	} else {
+	if fs != nil {
 		offlineLogStorage, err := NewOfflineStore(fs.GetRepoConfig())
 		loggingService.offlineLogStorage = offlineLogStorage
-
 		if err != nil {
 			return nil, err
 		}
-		// Start goroutine to process logs
+	}
+
+	// Start goroutine to process logs
+	if enableLogProcessing {
 		go loggingService.processLogs()
+
 	}
 	return loggingService, nil
 }
@@ -77,15 +86,15 @@ func (s *LoggingService) EmitLog(l *Log) error {
 	select {
 	case s.logChannel <- l:
 		return nil
-	case <-time.After(20 * time.Millisecond):
+	case <-time.After(s.logInsertTTl):
 		return fmt.Errorf("could not add to log channel with capacity %d. Operation timed out. Current log channel length is %d", cap(s.logChannel), len(s.logChannel))
 	}
 }
 
 func (s *LoggingService) processLogs() {
 	// start a periodic flush
 	// TODO(kevjumba): set param so users can configure flushing duration
-	ticker := time.NewTicker(100 * time.Millisecond)
+	ticker := time.NewTicker(s.logFlushInterval)
 	defer ticker.Stop()
 
 	for {
@@ -263,9 +272,8 @@ func GetSchemaFromFeatureService(featureService *model.FeatureService, entities
 		// Create copies of FeatureView that may contains the same *FeatureView but
 		// each differentiated by a *FeatureViewProjection
 		featureViewName := featureProjection.Name
-		if fv, ok := fvs[featureViewName]; ok {
-			for _, f := range fv.Base.Projection.Features {
-				// add feature to map
+		if _, ok := fvs[featureViewName]; ok {
+			for _, f := range featureProjection.Features {
 				features = append(features, f.Name)
 				allFeatureTypes[f.Name] = f.Dtype
 			}
@@ -278,6 +286,7 @@ func GetSchemaFromFeatureService(featureService *model.FeatureService, entities
 			return nil, fmt.Errorf("no such feature view found in feature service %s", featureViewName)
 		}
 	}
+
 	schema := &Schema{
 		Entities:      joinKeys,
 		Features:      features,

@@ -28,7 +28,7 @@ func TestLoggingChannelTimeout(t *testing.T) {
 	}
 	loggingService.EmitLog(&newLog)
 	// Wait for memory buffer flush
-	assert.Eventually(t, func() bool { return true }, 100*time.Millisecond, 20*time.Millisecond)
+	assert.Eventually(t, func() bool { return true }, 100*time.Millisecond, DEFAULT_LOG_INSERT_TIMEOUT)
 
 	newTs := timestamppb.New(time.Now())
 

@@ -19,18 +19,21 @@ type OfflineLogStorage interface {
 }
 
 func getOfflineStoreType(offlineStoreConfig map[string]interface{}) (string, bool) {
-	if onlineStoreType, ok := offlineStoreConfig["type"]; !ok {
+	if onlineStoreType, ok := offlineStoreConfig["storeType"]; !ok {
 		// Assume file for case of no specified.
-		return "file", true
+		return "", true
 	} else {
 		result, ok := onlineStoreType.(string)
 		return result, ok
 	}
 }
 
 func NewOfflineStore(config *registry.RepoConfig) (OfflineLogStorage, error) {
-	onlineStoreType, _ := getOfflineStoreType(config.OfflineStore)
-	if onlineStoreType == "file" {
+	offlineStoreType, _ := getOfflineStoreType(config.OfflineStore)
+	if offlineStoreType == "" {
+		// No offline store specified.
+		return nil, nil
+	} else if offlineStoreType == "file" {
 		fileConfig, err := GetFileConfig(config)
 		if err != nil {
 			return nil, err

@@ -42,7 +42,7 @@ func getRepoPath(basePath string) string {
 }
 
 // Starts a new grpc server, registers the serving service and returns a client.
-func getClient(ctx context.Context, basePath string, enableLogging bool) (serving.ServingServiceClient, func()) {
+func getClient(ctx context.Context, offlineStoreType string, basePath string, enableLogging bool) (serving.ServingServiceClient, func()) {
 	buffer := 1024 * 1024
 	listener := bufconn.Listen(buffer)
 
@@ -60,6 +60,7 @@ func getClient(ctx context.Context, basePath string, enableLogging bool) (servin
 			panic(err)
 		}
 		config.OfflineStore["path"] = absPath
+		config.OfflineStore["storeType"] = offlineStoreType
 	}
 
 	if err != nil {
@@ -104,7 +105,7 @@ func TestGetFeastServingInfo(t *testing.T) {
 	err := test.SetupFeatureRepo(dir)
 	assert.Nil(t, err)
 	defer test.CleanUpRepo(dir)
-	client, closer := getClient(ctx, dir, false)
+	client, closer := getClient(ctx, "", dir, false)
 	defer closer()
 	response, err := client.GetFeastServingInfo(ctx, &serving.GetFeastServingInfoRequest{})
 	assert.Nil(t, err)
@@ -118,7 +119,7 @@ func TestGetOnlineFeaturesSqlite(t *testing.T) {
 	err := test.SetupFeatureRepo(dir)
 	assert.Nil(t, err)
 	defer test.CleanUpRepo(dir)
-	client, closer := getClient(ctx, dir, false)
+	client, closer := getClient(ctx, "", dir, false)
 	defer closer()
 	entities := make(map[string]*types.RepeatedValue)
 	entities["driver_id"] = &types.RepeatedValue{
@@ -176,7 +177,7 @@ func TestGetOnlineFeaturesSqliteWithLogging(t *testing.T) {
 	err := test.SetupFeatureRepo(dir)
 	assert.Nil(t, err)
 	defer test.CleanUpRepo(dir)
-	client, closer := getClient(ctx, dir, true)
+	client, closer := getClient(ctx, "file", dir, true)
 	defer closer()
 	entities := make(map[string]*types.RepeatedValue)
 	entities["driver_id"] = &types.RepeatedValue{
@@ -195,7 +196,7 @@ func TestGetOnlineFeaturesSqliteWithLogging(t *testing.T) {
 	}
 	response, err := client.GetOnlineFeatures(ctx, request)
 	// Wait for logger to flush.
-	assert.Eventually(t, func() bool { return true }, 1*time.Second, 200*time.Millisecond)
+	assert.Eventually(t, func() bool { return true }, 1*time.Second, logging.DEFAULT_LOG_FLUSH_INTERVAL)
 	assert.Nil(t, err)
 	assert.NotNil(t, response)
 
@@ -233,8 +234,8 @@ func TestGetOnlineFeaturesSqliteWithLogging(t *testing.T) {
 			}
 		}
 	}
-	err = test.CleanUpLogs(logPath)
-	assert.Nil(t, err)
+	// err = test.CleanUpLogs(logPath)
+	// assert.Nil(t, err)
 }
 
 // Generate the expected log rows based on the resulting feature vector returned from GetOnlineFeatures.