ttl for regular feature views

pyalex · pyalex · commit acbfe844fe47 · 2022-05-19T17:56:58.000-07:00
Signed-off-by: Oleksii Moskalenko &lt;moskalenko.alexey@gmail.com&gt;
diff --git a/go/embedded/online_features.go b/go/embedded/online_features.go
@@ -269,12 +269,12 @@ func (s *OnlineFeatureService) StartGprcServerWithLogging(host string, port int,
 	go func() {
 		// As soon as these signals are received from OS, try to gracefully stop the gRPC server
 		<-s.grpcStopCh
-		fmt.Println("Stopping the gRPC server...")
+		log.Println("Stopping the gRPC server...")
 		grpcServer.GracefulStop()
 		if loggingService != nil {
 			loggingService.Stop()
 		}
-		fmt.Println("gRPC server terminated")
+		log.Println("gRPC server terminated")
 	}()
 
 	err = grpcServer.Serve(lis)
@@ -314,11 +314,15 @@ func (s *OnlineFeatureService) StartHttpServerWithLogging(host string, port int,
 	go func() {
 		// As soon as these signals are received from OS, try to gracefully stop the gRPC server
 		<-s.httpStopCh
-		fmt.Println("Stopping the HTTP server...")
+		log.Println("Stopping the HTTP server...")
 		err := ser.Stop()
 		if err != nil {
-			fmt.Printf("Error when stopping the HTTP server: %v\n", err)
+			log.Printf("Error when stopping the HTTP server: %v\n", err)
 		}
+		if loggingService != nil {
+			loggingService.Stop()
+		}
+		log.Println("HTTP server terminated")
 	}()
 
 	return ser.Serve(host, port)
diff --git a/sdk/python/tests/integration/e2e/test_go_feature_server.py b/sdk/python/tests/integration/e2e/test_go_feature_server.py
@@ -103,8 +103,9 @@ def server_port(environment, server_type: str):
         embedded.stop_grpc_server()
     else:
         embedded.stop_http_server()
+
     # wait for graceful stop
-    time.sleep(2)
+    time.sleep(5)
 
 
 @pytest.fixture
diff --git a/sdk/python/tests/utils/logged_features.py b/sdk/python/tests/utils/logged_features.py
@@ -9,11 +9,10 @@
 import pandas as pd
 import pyarrow
 
-from feast import FeatureService, FeatureStore
+from feast import FeatureService, FeatureStore, FeatureView
 from feast.errors import FeatureViewNotFoundException
 from feast.feature_logging import LOG_DATE_FIELD, LOG_TIMESTAMP_FIELD, REQUEST_ID_FIELD
 from feast.protos.feast.serving.ServingService_pb2 import FieldStatus
-from feast.utils import make_tzaware
 
 
 def prepare_logs(
@@ -52,13 +51,17 @@ def prepare_logs(
             logs_df[f"{destination_field}__timestamp"] = source_df[
                 "event_timestamp"
             ].dt.floor("s")
+            if logs_df[f"{destination_field}__timestamp"].dt.tz:
+                logs_df[f"{destination_field}__timestamp"] = logs_df[
+                    f"{destination_field}__timestamp"
+                ].dt.tz_convert(None)
             logs_df[f"{destination_field}__status"] = FieldStatus.PRESENT
-            if view.ttl:
+            if isinstance(view, FeatureView) and view.ttl:
                 logs_df[f"{destination_field}__status"] = logs_df[
                     f"{destination_field}__status"
                 ].mask(
-                    source_df["event_timestamp"]
-                    < (make_tzaware(datetime.datetime.utcnow()) - view.ttl),
+                    logs_df[f"{destination_field}__timestamp"]
+                    < (datetime.datetime.utcnow() - view.ttl),
                     FieldStatus.OUTSIDE_MAX_AGE,
                 )
 
