Project Namespacing (#393)

* Implement project namespacing (without auth) * Update Protos, Java SDK, Golang SDK to support namespacing * Fixed Python SDK to support project namespacing protos * Add integration with projects, update code to be compliant with new protos * Move name, version and project back to spec * Update Feast Core and Feast Ingestion to support project namespacing * Update Core and Ingestion based on refactored FeatureSet proto * Remove entity dataset validation * Register feature sets first to speed up tests * Apply PR #392 * Apply spotless * Order test output Co-authored-by: Chen Zhiling <[email protected]>
feast-dev · Dec 27, 2019 · 5b69331 · 5b69331
1 parent 51cbdc7
commit 5b69331
Show file tree

Hide file tree

Showing 139 changed files with 4,666 additions and 2,855 deletions.
diff --git a/.prow/scripts/test-end-to-end-batch.sh b/.prow/scripts/test-end-to-end-batch.sh
@@ -123,7 +123,9 @@ feast:
 
 spring:
   jpa:
-    properties.hibernate.format_sql: true
+    properties.hibernate:
+      format_sql: true
+      event.merge.entity_copy_observer: allow
     hibernate.naming.physical-strategy=org.hibernate.boot.model.naming: PhysicalNamingStrategyStandardImpl
     hibernate.ddl-auto: update
   datasource:
@@ -167,7 +169,8 @@ bigquery_config:
   datasetId: $DATASET_NAME
 subscriptions:
   - name: "*"
-    version: ">0"
+    version: "*"
+    project: "*"
 EOF
 
 cat <<EOF > /tmp/serving.warehouse.application.yml

diff --git a/.prow/scripts/test-end-to-end.sh b/.prow/scripts/test-end-to-end.sh
@@ -115,9 +115,12 @@ feast:
 
 spring:
   jpa:
-    properties.hibernate.format_sql: true
+    properties.hibernate:
+      format_sql: true
+      event.merge.entity_copy_observer: allow
     hibernate.naming.physical-strategy=org.hibernate.boot.model.naming: PhysicalNamingStrategyStandardImpl
     hibernate.ddl-auto: update
+
   datasource:
     url: jdbc:postgresql://localhost:5432/postgres
     username: postgres
@@ -153,7 +156,8 @@ redis_config:
   port: 6379
 subscriptions:
   - name: "*"
-    version: ">0"
+    version: "*"
+    project: "*"
 EOF
 
 cat <<EOF > /tmp/serving.online.application.yml
@@ -182,6 +186,7 @@ grpc:
 spring:
   main:
     web-environment: false
+
 EOF
 
 nohup java -jar serving/target/feast-serving-$REVISION.jar \

diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
@@ -64,8 +64,9 @@ store {
   name: "SERVING"
   type: REDIS
   subscriptions {
+    project: "*"
     name: "*"
-    version: ">0"
+    version: "*"
   }
   redis_config {
     host: "localhost"
@@ -76,8 +77,9 @@ store {
   name: "WAREHOUSE"
   type: BIGQUERY
   subscriptions {
+    project: "*"
     name: "*"
-    version: ">0"
+    version: "*"
   }
   bigquery_config {
     project_id: "my-google-project-id"

diff --git a/core/src/main/java/feast/core/config/FeatureStreamConfig.java b/core/src/main/java/feast/core/config/FeatureStreamConfig.java
@@ -52,8 +52,8 @@ public Source getDefaultSource(FeastProperties feastProperties) {
         String topicName = streamProperties.getOptions().get("topic");
         Map<String, Object> map = new HashMap<>();
         map.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
-        map.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG,
-            DEFAULT_KAFKA_REQUEST_TIMEOUT_MS_CONFIG);
+        map.put(
+            AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, DEFAULT_KAFKA_REQUEST_TIMEOUT_MS_CONFIG);
         AdminClient client = AdminClient.create(map);
 
         NewTopic newTopic =

diff --git a/core/src/main/java/feast/core/dao/FeatureSetRepository.java b/core/src/main/java/feast/core/dao/FeatureSetRepository.java
@@ -19,28 +19,31 @@
 import feast.core.model.FeatureSet;
 import java.util.List;
 import org.springframework.data.jpa.repository.JpaRepository;
-import org.springframework.data.jpa.repository.Query;
 
 /** JPA repository supplying FeatureSet objects keyed by id. */
 public interface FeatureSetRepository extends JpaRepository<FeatureSet, String> {
 
   long count();
 
-  // Find feature set by name and version
-  FeatureSet findFeatureSetByNameAndVersion(String name, Integer version);
+  // Find single feature set by project, name, and version
+  FeatureSet findFeatureSetByNameAndProject_NameAndVersion(
+      String name, String project, Integer version);
 
-  // Find latest version of a feature set by name
-  FeatureSet findFirstFeatureSetByNameOrderByVersionDesc(String name);
-
-  // find all versions of featureSets matching the given name.
-  List<FeatureSet> findByName(String name);
-
-  // find all versions of featureSets with names matching the regex
-  @Query(
-      nativeQuery = true,
-      value = "SELECT * FROM feature_sets " + "WHERE name LIKE ?1 ORDER BY name ASC, version ASC")
-  List<FeatureSet> findByNameWithWildcardOrderByNameAscVersionAsc(String name);
+  // Find single latest version of a feature set by project and name (LIKE)
+  FeatureSet findFirstFeatureSetByNameLikeAndProject_NameOrderByVersionDesc(
+      String name, String project);
 
   // find all feature sets and order by name and version
   List<FeatureSet> findAllByOrderByNameAscVersionAsc();
+
+  // find all feature sets within a project and order by name and version
+  List<FeatureSet> findAllByProject_NameOrderByNameAscVersionAsc(String project_name);
+
+  // find all versions of feature sets matching the given name pattern with a specific project.
+  List<FeatureSet> findAllByNameLikeAndProject_NameOrderByNameAscVersionAsc(
+      String name, String project_name);
+
+  // find all versions of feature sets matching the given name pattern and project pattern
+  List<FeatureSet> findAllByNameLikeAndProject_NameLikeOrderByNameAscVersionAsc(
+      String name, String project_name);
 }
diff --git a/core/src/main/java/feast/core/dao/ProjectRepository.java b/core/src/main/java/feast/core/dao/ProjectRepository.java
@@ -0,0 +1,27 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ * Copyright 2018-2019 The Feast Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package feast.core.dao;
+
+import feast.core.model.Project;
+import java.util.List;
+import org.springframework.data.jpa.repository.JpaRepository;
+
+/** JPA repository supplying Project objects keyed by id. */
+public interface ProjectRepository extends JpaRepository<Project, String> {
+
+  List<Project> findAllByArchivedIsFalse();
+}
diff --git a/core/src/main/java/feast/core/grpc/CoreServiceImpl.java b/core/src/main/java/feast/core/grpc/CoreServiceImpl.java
@@ -16,25 +16,35 @@
  */
 package feast.core.grpc;
 
-import com.google.protobuf.InvalidProtocolBufferException;
 import feast.core.CoreServiceGrpc.CoreServiceImplBase;
 import feast.core.CoreServiceProto.ApplyFeatureSetRequest;
 import feast.core.CoreServiceProto.ApplyFeatureSetResponse;
+import feast.core.CoreServiceProto.ArchiveProjectRequest;
+import feast.core.CoreServiceProto.ArchiveProjectResponse;
+import feast.core.CoreServiceProto.CreateProjectRequest;
+import feast.core.CoreServiceProto.CreateProjectResponse;
 import feast.core.CoreServiceProto.GetFeastCoreVersionRequest;
 import feast.core.CoreServiceProto.GetFeastCoreVersionResponse;
 import feast.core.CoreServiceProto.GetFeatureSetRequest;
 import feast.core.CoreServiceProto.GetFeatureSetResponse;
 import feast.core.CoreServiceProto.ListFeatureSetsRequest;
 import feast.core.CoreServiceProto.ListFeatureSetsResponse;
+import feast.core.CoreServiceProto.ListProjectsRequest;
+import feast.core.CoreServiceProto.ListProjectsResponse;
 import feast.core.CoreServiceProto.ListStoresRequest;
 import feast.core.CoreServiceProto.ListStoresResponse;
 import feast.core.CoreServiceProto.UpdateStoreRequest;
 import feast.core.CoreServiceProto.UpdateStoreResponse;
 import feast.core.exception.RetrievalException;
 import feast.core.grpc.interceptors.MonitoringInterceptor;
+import feast.core.model.Project;
+import feast.core.service.AccessManagementService;
 import feast.core.service.SpecService;
+import io.grpc.Status;
 import io.grpc.StatusRuntimeException;
 import io.grpc.stub.StreamObserver;
+import java.util.List;
+import java.util.stream.Collectors;
 import lombok.extern.slf4j.Slf4j;
 import org.lognet.springboot.grpc.GRpcService;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -45,10 +55,12 @@
 public class CoreServiceImpl extends CoreServiceImplBase {
 
   private SpecService specService;
+  private AccessManagementService accessManagementService;
 
   @Autowired
-  public CoreServiceImpl(SpecService specService) {
+  public CoreServiceImpl(SpecService specService, AccessManagementService accessManagementService) {
     this.specService = specService;
+    this.accessManagementService = accessManagementService;
   }
 
   @Override
@@ -65,9 +77,10 @@ public void getFeatureSet(
       GetFeatureSetResponse response = specService.getFeatureSet(request);
       responseObserver.onNext(response);
       responseObserver.onCompleted();
-    } catch (RetrievalException | InvalidProtocolBufferException | StatusRuntimeException e) {
+    } catch (RetrievalException | StatusRuntimeException e) {
       log.error("Exception has occurred in GetFeatureSet method: ", e);
-      responseObserver.onError(e);
+      responseObserver.onError(
+          Status.INTERNAL.withDescription(e.getMessage()).withCause(e).asRuntimeException());
     }
   }
 
@@ -78,9 +91,10 @@ public void listFeatureSets(
       ListFeatureSetsResponse response = specService.listFeatureSets(request.getFilter());
       responseObserver.onNext(response);
       responseObserver.onCompleted();
-    } catch (RetrievalException | InvalidProtocolBufferException e) {
+    } catch (RetrievalException | IllegalArgumentException e) {
       log.error("Exception has occurred in ListFeatureSet method: ", e);
-      responseObserver.onError(e);
+      responseObserver.onError(
+          Status.INTERNAL.withDescription(e.getMessage()).withCause(e).asRuntimeException());
     }
   }
 
@@ -93,7 +107,8 @@ public void listStores(
       responseObserver.onCompleted();
     } catch (RetrievalException e) {
       log.error("Exception has occurred in ListStores method: ", e);
-      responseObserver.onError(e);
+      responseObserver.onError(
+          Status.INTERNAL.withDescription(e.getMessage()).withCause(e).asRuntimeException());
     }
   }
 
@@ -104,9 +119,17 @@ public void applyFeatureSet(
       ApplyFeatureSetResponse response = specService.applyFeatureSet(request.getFeatureSet());
       responseObserver.onNext(response);
       responseObserver.onCompleted();
+    } catch (org.hibernate.exception.ConstraintViolationException e) {
+      log.error(
+          "Unable to persist this feature set due to a constraint violation. Please ensure that"
+              + " field names are unique within the project namespace: ",
+          e);
+      responseObserver.onError(
+          Status.ALREADY_EXISTS.withDescription(e.getMessage()).withCause(e).asRuntimeException());
     } catch (Exception e) {
       log.error("Exception has occurred in ApplyFeatureSet method: ", e);
-      responseObserver.onError(e);
+      responseObserver.onError(
+          Status.INTERNAL.withDescription(e.getMessage()).withCause(e).asRuntimeException());
     }
   }
 
@@ -119,7 +142,53 @@ public void updateStore(
       responseObserver.onCompleted();
     } catch (Exception e) {
       log.error("Exception has occurred in UpdateStore method: ", e);
-      responseObserver.onError(e);
+      responseObserver.onError(
+          Status.INTERNAL.withDescription(e.getMessage()).withCause(e).asRuntimeException());
+    }
+  }
+
+  @Override
+  public void createProject(
+      CreateProjectRequest request, StreamObserver<CreateProjectResponse> responseObserver) {
+    try {
+      accessManagementService.createProject(request.getName());
+      responseObserver.onNext(CreateProjectResponse.getDefaultInstance());
+      responseObserver.onCompleted();
+    } catch (Exception e) {
+      log.error("Exception has occurred in the createProject method: ", e);
+      responseObserver.onError(
+          Status.INTERNAL.withDescription(e.getMessage()).withCause(e).asRuntimeException());
+    }
+  }
+
+  @Override
+  public void archiveProject(
+      ArchiveProjectRequest request, StreamObserver<ArchiveProjectResponse> responseObserver) {
+    try {
+      accessManagementService.archiveProject(request.getName());
+      responseObserver.onNext(ArchiveProjectResponse.getDefaultInstance());
+      responseObserver.onCompleted();
+    } catch (Exception e) {
+      log.error("Exception has occurred in the createProject method: ", e);
+      responseObserver.onError(
+          Status.INTERNAL.withDescription(e.getMessage()).withCause(e).asRuntimeException());
+    }
+  }
+
+  @Override
+  public void listProjects(
+      ListProjectsRequest request, StreamObserver<ListProjectsResponse> responseObserver) {
+    try {
+      List<Project> projects = accessManagementService.listProjects();
+      responseObserver.onNext(
+          ListProjectsResponse.newBuilder()
+              .addAllProjects(projects.stream().map(Project::getName).collect(Collectors.toList()))
+              .build());
+      responseObserver.onCompleted();
+    } catch (Exception e) {
+      log.error("Exception has occurred in the listProjects method: ", e);
+      responseObserver.onError(
+          Status.INTERNAL.withDescription(e.getMessage()).withCause(e).asRuntimeException());
     }
   }
 }