e2e tests WIP

Signed-off-by: pyalex <[email protected]>

go/embedded/online_features.go

@@ -8,23 +8,22 @@ import (
 	"os"
 	"os/signal"
 	"syscall"
-
-	"google.golang.org/grpc"
-
-	"github.com/feast-dev/feast/go/internal/feast/server"
-	"github.com/feast-dev/feast/go/internal/feast/server/logging"
-	"github.com/feast-dev/feast/go/protos/feast/serving"
+	"time"
 
 	"github.com/apache/arrow/go/v8/arrow"
 	"github.com/apache/arrow/go/v8/arrow/array"
 	"github.com/apache/arrow/go/v8/arrow/cdata"
 	"github.com/apache/arrow/go/v8/arrow/memory"
+	"google.golang.org/grpc"
 
 	"github.com/feast-dev/feast/go/internal/feast"
 	"github.com/feast-dev/feast/go/internal/feast/model"
 	"github.com/feast-dev/feast/go/internal/feast/onlineserving"
 	"github.com/feast-dev/feast/go/internal/feast/registry"
+	"github.com/feast-dev/feast/go/internal/feast/server"
+	"github.com/feast-dev/feast/go/internal/feast/server/logging"
 	"github.com/feast-dev/feast/go/internal/feast/transformation"
+	"github.com/feast-dev/feast/go/protos/feast/serving"
 	prototypes "github.com/feast-dev/feast/go/protos/feast/types"
 	"github.com/feast-dev/feast/go/types"
 )
@@ -44,6 +43,15 @@ type DataTable struct {
 	SchemaPtr uintptr
 }
 
+// LoggingOptions is a public (embedded) copy of logging.LoggingOptions struct.
+// See logging.LoggingOptions for properties description
+type LoggingOptions struct {
+	ChannelCapacity int
+	EmitTimeout     time.Duration
+	WriteInterval   time.Duration
+	FlushInterval   time.Duration
+}
+
 func NewOnlineFeatureService(conf *OnlineFeatureServiceConfig, transformationCallback transformation.TransformationCallback) *OnlineFeatureService {
 	repoConfig, err := registry.NewRepoConfigFromJSON(conf.RepoPath, conf.RepoConfig)
 	if err != nil {
@@ -215,10 +223,20 @@ func (s *OnlineFeatureService) GetOnlineFeatures(
 }
 
 func (s *OnlineFeatureService) StartGprcServer(host string, port int) error {
-	return s.StartGprcServerWithLogging(host, port, nil)
+	return s.StartGprcServerWithLogging(host, port, nil, LoggingOptions{})
+}
+
+func (s *OnlineFeatureService) StartGprcServerWithLoggingDefaultOpts(host string, port int, writeLoggedFeaturesCallback logging.OfflineStoreWriteCallback) error {
+	defaultOpts := LoggingOptions{
+		ChannelCapacity: logging.DefaultOptions.ChannelCapacity,
+		EmitTimeout:     logging.DefaultOptions.EmitTimeout,
+		WriteInterval:   logging.DefaultOptions.WriteInterval,
+		FlushInterval:   logging.DefaultOptions.FlushInterval,
+	}
+	return s.StartGprcServerWithLogging(host, port, writeLoggedFeaturesCallback, defaultOpts)
 }
 
-func (s *OnlineFeatureService) StartGprcServerWithLogging(host string, port int, writeLoggedFeaturesCallback logging.OfflineStoreWriteCallback) error {
+func (s *OnlineFeatureService) StartGprcServerWithLogging(host string, port int, writeLoggedFeaturesCallback logging.OfflineStoreWriteCallback, loggingOpts LoggingOptions) error {
 	var loggingService *logging.LoggingService = nil
 	var err error
 	if writeLoggedFeaturesCallback != nil {
@@ -227,7 +245,12 @@ func (s *OnlineFeatureService) StartGprcServerWithLogging(host string, port int,
 			return err
 		}
 
-		loggingService, err = logging.NewLoggingService(s.fs, sink)
+		loggingService, err = logging.NewLoggingService(s.fs, sink, logging.LoggingOptions{
+			ChannelCapacity: loggingOpts.ChannelCapacity,
+			EmitTimeout:     loggingOpts.EmitTimeout,
+			WriteInterval:   loggingOpts.WriteInterval,
+			FlushInterval:   loggingOpts.FlushInterval,
+		})
 		if err != nil {
 			return err
 		}
@@ -238,7 +261,6 @@ func (s *OnlineFeatureService) StartGprcServerWithLogging(host string, port int,
 	if err != nil {
 		return err
 	}
-	log.Printf("Listening a gRPC server on host %s port %d\n", host, port)
 
 	grpcServer := grpc.NewServer()
 	serving.RegisterServingServiceServer(grpcServer, ser)

go/internal/feast/server/logging/logger.go

@@ -37,7 +37,7 @@ type LogSink interface {
 	// Flush actually send data to a sink.
 	// We want to control amount to interaction with sink, since it could be a costly operation.
 	// Also, some sinks like BigQuery might have quotes and physically limit amount of write requests per day.
-	Flush(featureSeviceName string) error
+	Flush(featureServiceName string) error
 }
 
 type Logger interface {
@@ -135,6 +135,10 @@ func (l *LoggerImpl) loggerLoop() (lErr error) {
 			lErr = errors.WithStack(rErr)
 		}
 	}()
+
+	writeTicker := time.NewTicker(l.config.WriteInterval)
+	flushTicker := time.NewTicker(l.config.FlushInterval)
+
 	for {
 		shouldStop := false
 
@@ -149,12 +153,12 @@ func (l *LoggerImpl) loggerLoop() (lErr error) {
 				log.Printf("Log flush failed: %+v", err)
 			}
 			shouldStop = true
-		case <-time.After(l.config.WriteInterval):
+		case <-writeTicker.C:
 			err := l.buffer.writeBatch(l.sink)
 			if err != nil {
 				log.Printf("Log write failed: %+v", err)
 			}
-		case <-time.After(l.config.FlushInterval):
+		case <-flushTicker.C:
 			err := l.sink.Flush(l.featureServiceName)
 			if err != nil {
 				log.Printf("Log flush failed: %+v", err)
@@ -171,6 +175,9 @@ func (l *LoggerImpl) loggerLoop() (lErr error) {
 		}
 	}
 
+	writeTicker.Stop()
+	flushTicker.Stop()
+
 	// Notify all waiters for graceful stop
 	l.cond.L.Lock()
 	l.isStopped = true
@@ -259,7 +266,7 @@ func (l *LoggerImpl) Log(joinKeyToEntityValues map[string][]*types.Value, featur
 			EventTimestamps: eventTimestamps,
 
 			RequestId:    requestId,
-			LogTimestamp: time.Now(),
+			LogTimestamp: time.Now().UTC(),
 		}
 		err := l.EmitLog(&newLog)
 		if err != nil {

go/internal/feast/server/logging/memorybuffer.go

@@ -2,7 +2,6 @@ package logging
 
 import (
 	"fmt"
-	"time"
 
 	"github.com/apache/arrow/go/v8/arrow"
 	"github.com/apache/arrow/go/v8/arrow/array"
@@ -143,7 +142,7 @@ func (b *MemoryBuffer) convertToArrowRecord() (arrow.Record, error) {
 		}
 
 		logTimestamp := arrow.Timestamp(logRow.LogTimestamp.UnixMicro())
-		logDate := arrow.Date32(logRow.LogTimestamp.Truncate(24 * time.Hour).Unix())
+		logDate := arrow.Date32FromTime(logRow.LogTimestamp)
 
 		builder.Field(fieldNameToIdx[LOG_TIMESTAMP_FIELD]).(*array.TimestampBuilder).UnsafeAppend(logTimestamp)
 		builder.Field(fieldNameToIdx[LOG_DATE_FIELD]).(*array.Date32Builder).UnsafeAppend(logDate)

go/internal/feast/server/logging/memorybuffer_test.go

@@ -158,7 +158,7 @@ func TestSerializeToArrowTable(t *testing.T) {
 	// log date
 	today := time.Now().Truncate(24 * time.Hour)
 	builder.Field(8).(*array.Date32Builder).AppendValues(
-		[]arrow.Date32{arrow.Date32(today.Unix()), arrow.Date32(today.Unix())}, []bool{true, true})
+		[]arrow.Date32{arrow.Date32FromTime(today), arrow.Date32FromTime(today)}, []bool{true, true})
 
 	// request id
 	builder.Field(9).(*array.StringBuilder).AppendValues(

go/internal/feast/server/logging/offlinestoresink.go

@@ -3,15 +3,16 @@ package logging
 import (
 	"errors"
 	"fmt"
+	"io"
+	"io/ioutil"
+	"os"
+	"path/filepath"
+
 	"github.com/apache/arrow/go/v8/arrow"
 	"github.com/apache/arrow/go/v8/arrow/array"
 	"github.com/apache/arrow/go/v8/parquet"
 	"github.com/apache/arrow/go/v8/parquet/pqarrow"
 	"github.com/google/uuid"
-	"io"
-	"io/ioutil"
-	"os"
-	"path/filepath"
 )
 
 type OfflineStoreWriteCallback func(featureServiceName, datasetDir string) string
@@ -41,7 +42,7 @@ func (s *OfflineStoreSink) getOrCreateDatasetDir() (string, error) {
 }
 
 func (s *OfflineStoreSink) cleanCurrentDatasetDir() error {
-	if s.datasetDir != "" {
+	if s.datasetDir == "" {
 		return nil
 	}
 	datasetDir := s.datasetDir

go/internal/test/go_integration_test_utils.go

@@ -138,7 +138,7 @@ func SetupInitializedRepo(basePath string) error {
 	// var stderr bytes.Buffer
 	// var stdout bytes.Buffer
 	applyCommand.Dir = featureRepoPath
-	out, err := applyCommand.Output()
+	out, err := applyCommand.CombinedOutput()
 	if err != nil {
 		log.Println(string(out))
 		return err
@@ -152,7 +152,7 @@ func SetupInitializedRepo(basePath string) error {
 	materializeCommand := exec.Command("feast", "materialize-incremental", formattedTime)
 	materializeCommand.Env = os.Environ()
 	materializeCommand.Dir = featureRepoPath
-	out, err = materializeCommand.Output()
+	out, err = materializeCommand.CombinedOutput()
 	if err != nil {
 		log.Println(string(out))
 		return err

sdk/python/feast/embedded_go/online_features_service.py

@@ -18,7 +18,12 @@
 from feast.types import from_value_type
 from feast.value_type import ValueType
 
-from .lib.embedded import DataTable, NewOnlineFeatureService, OnlineFeatureServiceConfig
+from .lib.embedded import (
+    DataTable,
+    LoggingOptions,
+    NewOnlineFeatureService,
+    OnlineFeatureServiceConfig,
+)
 from .lib.go import Slice_string
 from .type_map import FEAST_TYPE_TO_ARROW_TYPE, arrow_array_to_array_of_proto
 
@@ -134,9 +139,22 @@ def get_online_features(
         resp = record_batch_to_online_response(record_batch)
         return OnlineResponse(resp)
 
-    def start_grpc_server(self, host: str, port: int, enable_logging=True):
+    def start_grpc_server(
+        self,
+        host: str,
+        port: int,
+        enable_logging: bool = True,
+        logging_options: Optional[LoggingOptions] = None,
+    ):
         if enable_logging:
-            self._service.StartGprcServerWithLogging(host, port, self._logging_callback)
+            if logging_options:
+                self._service.StartGprcServerWithLogging(
+                    host, port, self._logging_callback, logging_options
+                )
+            else:
+                self._service.StartGprcServerWithLoggingDefaultOpts(
+                    host, port, self._logging_callback
+                )
         else:
             self._service.StartGprcServer(host, port)
 
@@ -189,14 +207,14 @@ def transformation_callback(
 
 def logging_callback(
     fs: "FeatureStore", feature_service_name: str, dataset_dir: str,
-) -> str:
+) -> bytes:
     feature_service = fs.get_feature_service(feature_service_name, allow_cache=True)
     try:
         fs.write_logged_features(logs=Path(dataset_dir), source=feature_service)
     except Exception as exc:
-        return repr(exc)
+        return repr(exc).encode()
 
-    return ""  # no error
+    return "".encode()  # no error
 
 
 def allocate_schema_and_array():

sdk/python/feast/feature_logging.py

@@ -108,12 +108,12 @@ class _DestinationRegistry(type):
 
     def __new__(cls, name, bases, dct):
         kls = type.__new__(cls, name, bases, dct)
-        if dct.get("_proto_attr_name"):
-            cls.classes_by_proto_attr_name[dct["_proto_attr_name"]] = kls
+        if dct.get("_proto_kind"):
+            cls.classes_by_proto_attr_name[dct["_proto_kind"]] = kls
         return kls
 
 
-class LoggingDestination:
+class LoggingDestination(metaclass=_DestinationRegistry):
     """
     Logging destination contains details about where exactly logs should be written inside an offline store.
     It is implementation specific - each offline store must implement LoggingDestination subclass.

sdk/python/feast/infra/provider.py

@@ -208,8 +208,8 @@ def write_feature_service_logs(
     def retrieve_feature_service_logs(
         self,
         feature_service: FeatureService,
-        from_: datetime,
-        to: datetime,
+        start_date: datetime,
+        end_date: datetime,
         config: RepoConfig,
         registry: Registry,
     ) -> RetrievalJob:

sdk/python/tests/foo_provider.py

@@ -94,8 +94,8 @@ def write_feature_service_logs(
     def retrieve_feature_service_logs(
         self,
         feature_service: FeatureService,
-        from_: datetime,
-        to: datetime,
+        start_date: datetime,
+        end_date: datetime,
         config: RepoConfig,
         registry: Registry,
     ) -> RetrievalJob: