diff --git a/sdk/java/src/main/java/com/gojek/feast/FeastClient.java b/sdk/java/src/main/java/com/gojek/feast/FeastClient.java index 94836d8..eee7244 100644 --- a/sdk/java/src/main/java/com/gojek/feast/FeastClient.java +++ b/sdk/java/src/main/java/com/gojek/feast/FeastClient.java @@ -25,31 +25,19 @@ import feast.proto.serving.ServingServiceGrpc; import feast.proto.serving.ServingServiceGrpc.ServingServiceBlockingStub; import io.grpc.CallCredentials; +import io.grpc.Channel; import io.grpc.ManagedChannel; -import io.grpc.ManagedChannelBuilder; -import io.grpc.netty.shaded.io.grpc.netty.GrpcSslContexts; -import io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder; -import io.opentracing.contrib.grpc.TracingClientInterceptor; -import io.opentracing.util.GlobalTracer; -import java.io.File; import java.util.HashSet; import java.util.List; import java.util.Optional; -import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; -import javax.net.ssl.SSLException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @SuppressWarnings("WeakerAccess") -public class FeastClient implements AutoCloseable { +public class FeastClient extends GrpcManager { Logger logger = LoggerFactory.getLogger(FeastClient.class); - private static final int CHANNEL_SHUTDOWN_TIMEOUT_SEC = 5; - - private final ManagedChannel channel; - private final ServingServiceBlockingStub stub; - /** * Create a client to access Feast Serving. * @@ -73,33 +61,7 @@ public static FeastClient create(String host, int port) { * @return {@link FeastClient} */ public static FeastClient createSecure(String host, int port, SecurityConfig securityConfig) { - // Configure client TLS - ManagedChannel channel = null; - if (securityConfig.isTLSEnabled()) { - if (securityConfig.getCertificatePath().isPresent()) { - String certificatePath = securityConfig.getCertificatePath().get(); - // Use custom certificate for TLS - File certificateFile = new File(certificatePath); - try { - channel = - NettyChannelBuilder.forAddress(host, port) - .useTransportSecurity() - .sslContext(GrpcSslContexts.forClient().trustManager(certificateFile).build()) - .build(); - } catch (SSLException e) { - throw new IllegalArgumentException( - String.format("Invalid Certificate provided at path: %s", certificatePath), e); - } - } else { - // Use system certificates for TLS - channel = ManagedChannelBuilder.forAddress(host, port).useTransportSecurity().build(); - } - } else { - // Disable TLS - channel = ManagedChannelBuilder.forAddress(host, port).usePlaintext().build(); - } - - return new FeastClient(channel, securityConfig.getCredentials()); + return new FeastClient(host, port, securityConfig); } /** @@ -188,24 +150,16 @@ public List getOnlineFeatures(List featureRefs, List rows, Str .collect(Collectors.toList()); } - protected FeastClient(ManagedChannel channel, Optional credentials) { - this.channel = channel; - TracingClientInterceptor tracingInterceptor = - TracingClientInterceptor.newBuilder().withTracer(GlobalTracer.get()).build(); - - ServingServiceBlockingStub servingStub = - ServingServiceGrpc.newBlockingStub(tracingInterceptor.intercept(channel)); - - if (credentials.isPresent()) { - servingStub = servingStub.withCallCredentials(credentials.get()); - } + @Override + protected ServingServiceBlockingStub getStub(Channel channel) { + return ServingServiceGrpc.newBlockingStub(channel); + } - this.stub = servingStub; + protected FeastClient(ManagedChannel channel, Optional credentials) { + super(channel, credentials); } - public void close() throws Exception { - if (channel != null) { - channel.shutdown().awaitTermination(CHANNEL_SHUTDOWN_TIMEOUT_SEC, TimeUnit.SECONDS); - } + protected FeastClient(String host, int port, SecurityConfig securityConfig) { + super(host, port, securityConfig); } } diff --git a/sdk/java/src/main/java/com/gojek/feast/GrpcManager.java b/sdk/java/src/main/java/com/gojek/feast/GrpcManager.java new file mode 100644 index 0000000..a7fa4b1 --- /dev/null +++ b/sdk/java/src/main/java/com/gojek/feast/GrpcManager.java @@ -0,0 +1,95 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * Copyright 2018-2021 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 com.gojek.feast; + +import io.grpc.CallCredentials; +import io.grpc.Channel; +import io.grpc.ManagedChannel; +import io.grpc.ManagedChannelBuilder; +import io.grpc.netty.shaded.io.grpc.netty.GrpcSslContexts; +import io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder; +import io.opentracing.contrib.grpc.TracingClientInterceptor; +import io.opentracing.util.GlobalTracer; +import java.io.File; +import java.util.Optional; +import java.util.concurrent.TimeUnit; +import javax.net.ssl.SSLException; + +public abstract class GrpcManager> + implements AutoCloseable { + + private static final int CHANNEL_SHUTDOWN_TIMEOUT_SEC = 5; + + private final ManagedChannel channel; + protected final S stub; + + protected GrpcManager(String host, int port, SecurityConfig securityConfig) { + this(createSecureChannel(host, port, securityConfig), securityConfig.getCredentials()); + } + + protected abstract S getStub(Channel channel); + + protected GrpcManager(ManagedChannel channel, Optional credentials) { + this.channel = channel; + TracingClientInterceptor tracingInterceptor = + TracingClientInterceptor.newBuilder().withTracer(GlobalTracer.get()).build(); + + S servingStub = getStub(tracingInterceptor.intercept(channel)); + + if (credentials.isPresent()) { + servingStub = servingStub.withCallCredentials(credentials.get()); + } + + this.stub = servingStub; + } + + private static ManagedChannel createSecureChannel( + String host, int port, SecurityConfig securityConfig) { + ManagedChannel channel; + if (securityConfig.isTLSEnabled()) { + if (securityConfig.getCertificatePath().isPresent()) { + String certificatePath = securityConfig.getCertificatePath().get(); + // Use custom certificate for TLS + File certificateFile = new File(certificatePath); + try { + channel = + NettyChannelBuilder.forAddress(host, port) + .useTransportSecurity() + .sslContext(GrpcSslContexts.forClient().trustManager(certificateFile).build()) + .build(); + } catch (SSLException e) { + throw new IllegalArgumentException( + String.format("Invalid Certificate provided at path: %s", certificatePath), e); + } + } else { + // Use system certificates for TLS + channel = ManagedChannelBuilder.forAddress(host, port).useTransportSecurity().build(); + } + } else { + // Disable TLS + channel = ManagedChannelBuilder.forAddress(host, port).usePlaintext().build(); + } + return channel; + } + + @Override + public void close() throws Exception { + if (channel != null) { + channel.shutdown().awaitTermination(CHANNEL_SHUTDOWN_TIMEOUT_SEC, TimeUnit.SECONDS); + } + } +} diff --git a/sdk/java/src/main/java/com/gojek/feast/core/CoreClient.java b/sdk/java/src/main/java/com/gojek/feast/core/CoreClient.java new file mode 100644 index 0000000..895cb09 --- /dev/null +++ b/sdk/java/src/main/java/com/gojek/feast/core/CoreClient.java @@ -0,0 +1,285 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * Copyright 2018-2021 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 com.gojek.feast.core; + +import com.gojek.feast.GrpcManager; +import com.gojek.feast.SecurityConfig; +import feast.proto.core.*; +import feast.proto.core.CoreServiceGrpc.CoreServiceBlockingStub; +import io.grpc.CallCredentials; +import io.grpc.Channel; +import io.grpc.ManagedChannel; +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.stream.Collectors; + +public class CoreClient extends GrpcManager { + + /** + * Create a client to access Feast Core. + * + * @param host hostname or ip address of Feast core GRPC server + * @param port port number of Feast core GRPC server + * @return {@link CoreClient} + */ + public static CoreClient create(String host, int port) { + // configure client with no security config. + return CoreClient.createSecure(host, port, SecurityConfig.newBuilder().build()); + } + + /** + * Create an authenticated client that can access Feast core with authentication enabled. Supports + * the {@link CallCredentials} in the {@link feast.common.auth.credentials} package. + * + * @param host hostname or ip address of Feast core GRPC server + * @param port port number of Feast core GRPC server + * @param securityConfig security options to configure the Feast client. See {@link + * SecurityConfig} for options. + * @return Optional of {@link CoreClient} + */ + public static CoreClient createSecure(String host, int port, SecurityConfig securityConfig) { + return new CoreClient(host, port, securityConfig); + } + + /** + * Create a globally unique namespace to store Feature Tables in. + * + * @param name of the project. + * @return true if request returned, false otherwise. + */ + public boolean createProject(String name) { + CoreServiceProto.CreateProjectRequest request = + CoreServiceProto.CreateProjectRequest.newBuilder().setName(name).build(); + CoreServiceProto.CreateProjectResponse response = stub.createProject(request); + return response != null; + } + + /** + * Archives a project. Archived projects will continue to exist and function, but won't be visible + * through the Core API. Any existing ingestion or serving requests will continue to function, but + * will result in warning messages being logged. It is not possible to unarchive a project through + * the Core API. + * + * @param name of the project to archive. + * @return true if request returned, false otherwise. + */ + public boolean archiveProject(String name) { + CoreServiceProto.ArchiveProjectRequest request = + CoreServiceProto.ArchiveProjectRequest.newBuilder().setName(name).build(); + CoreServiceProto.ArchiveProjectResponse response = stub.archiveProject(request); + return response != null; + } + + /** + * List all active projects. + * + * @return the list of project names. + */ + public List listProjects() { + return stub.listProjects(CoreServiceProto.ListProjectsRequest.newBuilder().build()) + .getProjectsList(); + } + + /** + * Returns a specific entity. + * + * @param project name + * @param name of the entity + * @return Optional of {@link Entity} + */ + public Optional getEntity(String project, String name) { + CoreServiceProto.GetEntityRequest.Builder request = + CoreServiceProto.GetEntityRequest.newBuilder().setProject(project).setName(name); + CoreServiceProto.GetEntityResponse response = stub.getEntity(request.build()); + Entity entity = response.hasEntity() ? new Entity(project, response.getEntity()) : null; + return Optional.ofNullable(entity); + } + + /** + * Create or update and existing entity. Schema changes will update the entity if the changes are + * valid. Can't update name or type. + * + * @param entitySpec {@link Entity.Spec} + * @return Optional of {@link Entity} + */ + public Optional apply(Entity.Spec entitySpec) { + CoreServiceProto.ApplyEntityRequest.Builder request = + CoreServiceProto.ApplyEntityRequest.newBuilder() + .setProject(entitySpec.getProject()) + .setSpec(entitySpec.toProto()); + CoreServiceProto.ApplyEntityResponse response = stub.applyEntity(request.build()); + Entity entity = + response.hasEntity() ? new Entity(entitySpec.getProject(), response.getEntity()) : null; + return Optional.ofNullable(entity); + } + + /** + * List all entities under a certain project. + * + * @param project specifies the name of the project to list Entities in. + * @return all entity references under projectt. + */ + public List listEntities(String project) { + return listEntities(project, null); + } + + /** + * List all entities under a certain project. + * + * @param project specifies the name of the project to list Entities in. + * @param labels are User defined metadata for entity. Entities with all matching labels will be + * returned. + * @return all entity references and respective entities matching labels under project. + */ + public List listEntities(String project, Map labels) { + CoreServiceProto.ListEntitiesRequest.Filter.Builder filter = + CoreServiceProto.ListEntitiesRequest.Filter.newBuilder().setProject(project); + if (labels != null) filter.putAllLabels(labels); + CoreServiceProto.ListEntitiesRequest request = + CoreServiceProto.ListEntitiesRequest.newBuilder().setFilter(filter).build(); + List entities = stub.listEntities(request).getEntitiesList(); + return entities.stream().map(proto -> new Entity(project, proto)).collect(Collectors.toList()); + } + + /** + * Returns a specific feature table. + * + * @param project name + * @param name of the feature table + * @return Optional of {@link FeatureTable} + */ + public Optional getFeatureTable(String project, String name) { + CoreServiceProto.GetFeatureTableRequest.Builder request = + CoreServiceProto.GetFeatureTableRequest.newBuilder().setProject(project).setName(name); + CoreServiceProto.GetFeatureTableResponse response = stub.getFeatureTable(request.build()); + FeatureTable featureTable = + response.hasTable() ? new FeatureTable(project, response.getTable()) : null; + return Optional.ofNullable(featureTable); + } + + /** + * Create or update an existing feature table. Schema changes will update the feature table if the + * changes are valid. Can't update name, entities and feature names and types. + * + * @param featureTableSpec {@link FeatureTable.Spec} + * @return Optional of {@link FeatureTable} + */ + public Optional apply(FeatureTable.Spec featureTableSpec) { + CoreServiceProto.ApplyFeatureTableRequest.Builder request = + CoreServiceProto.ApplyFeatureTableRequest.newBuilder() + .setProject(featureTableSpec.getProject()) + .setTableSpec(featureTableSpec.toProto()); + CoreServiceProto.ApplyFeatureTableResponse response = stub.applyFeatureTable(request.build()); + FeatureTable featureTable = + response.hasTable() + ? new FeatureTable(featureTableSpec.getProject(), response.getTable()) + : null; + return Optional.ofNullable(featureTable); + } + + /** + * List all Feature Tables under a project. + * + * @param project name. + * @return a list of the matching {@link FeatureTable}. + */ + public List listFeatureTables(String project) { + return listFeatureTables(project, null); + } + + /** + * List all Feature Tables under a project. + * + * @param project name. + * @param labels on Feature Tables will be returned. + * @return a list of the matching {@link FeatureTable}. + */ + public List listFeatureTables(String project, Map labels) { + CoreServiceProto.ListFeatureTablesRequest.Filter.Builder filter = + CoreServiceProto.ListFeatureTablesRequest.Filter.newBuilder().setProject(project); + if (labels != null) filter.putAllLabels(labels); + CoreServiceProto.ListFeatureTablesRequest request = + CoreServiceProto.ListFeatureTablesRequest.newBuilder().setFilter(filter).build(); + List featureTables = + stub.listFeatureTables(request).getTablesList(); + return featureTables.stream() + .map(proto -> new FeatureTable(project, proto)) + .collect(Collectors.toList()); + } + + /** + * @param project name that the feature tables belongs to + * @param entities contained within the featureSet that the feature belongs to. Only feature + * tables with these entities will be searched for features. If none are specified all Feature + * Tables will be seeked. + * @return list of Features matching the parameters. + */ + public List listFeatures(String project, String... entities) { + return listFeatures(project, Arrays.asList(entities), null); + } + + /** + * @param project name that the feature tables belongs to + * @param entities contained within the featureSet that the feature belongs to. Only feature + * tables with these entities will be searched for features. + * @param labels are user defined metadata for feature. Features with all matching labels will be + * returned. + * @return list of Features matching the parameters. + */ + public List listFeatures( + String project, List entities, Map labels) { + CoreServiceProto.ListFeaturesRequest.Filter.Builder filter = + CoreServiceProto.ListFeaturesRequest.Filter.newBuilder().setProject(project); + if (entities != null) filter.addAllEntities(entities); + if (labels != null) filter.putAllLabels(labels); + CoreServiceProto.ListFeaturesRequest request = + CoreServiceProto.ListFeaturesRequest.newBuilder().setFilter(filter).build(); + Map features = stub.listFeatures(request).getFeaturesMap(); + return features.values().stream().map(Feature::new).collect(Collectors.toList()); + } + + /** + * Delete a specific Feature Table. + * + * @param project Name of the Project to delete the Feature Table from. + * @param name of the FeatureTable to delete. + * @return true if request returned, false otherwise. + */ + public boolean deleteFeatureTable(String project, String name) { + CoreServiceProto.DeleteFeatureTableRequest request = + CoreServiceProto.DeleteFeatureTableRequest.newBuilder() + .setProject(project) + .setName(name) + .build(); + return stub.deleteFeatureTable(request) != null; + } + + protected CoreClient(String host, int port, SecurityConfig securityConfig) { + super(host, port, securityConfig); + } + + protected CoreClient(ManagedChannel channel, Optional credentials) { + super(channel, credentials); + } + + @Override + protected CoreServiceBlockingStub getStub(Channel channel) { + return CoreServiceGrpc.newBlockingStub(channel); + } +} diff --git a/sdk/java/src/main/java/com/gojek/feast/core/Entity.java b/sdk/java/src/main/java/com/gojek/feast/core/Entity.java new file mode 100644 index 0000000..b7c3c8f --- /dev/null +++ b/sdk/java/src/main/java/com/gojek/feast/core/Entity.java @@ -0,0 +1,148 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * Copyright 2018-2021 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 com.gojek.feast.core; + +import feast.proto.core.EntityProto; +import java.util.HashMap; +import java.util.Map; + +public class Entity { + private Spec spec; + private Metadata metadata; + + public static class Spec { + private final String name; + private final String project; + private final ValueType value; + private final String description; + private final Map labels; + + protected Spec( + String project, + String name, + ValueType value, + String description, + Map labels) { + this.name = name; + this.project = project; + this.value = value; + this.description = description; + this.labels = labels; + } + + public static Builder getBuilder(String name, String project) { + return new Builder(name, project); + } + + public static class Builder { + private final String name; + private final String project; + private ValueType value; + private String description; + private final Map labels = new HashMap<>(); + + private Builder(String project, String name) { + this.project = project; + this.name = name; + } + + public Builder setValue(ValueType value) { + this.value = value; + return this; + } + + public Builder setDescription(String description) { + this.description = description; + return this; + } + + public Builder addLabel(String key, String val) { + this.labels.put(key, val); + return this; + } + + public Builder addLabels(Map labels) { + this.labels.putAll(labels); + return this; + } + + public Spec build() { + return new Spec(project, name, value, description, labels); + } + } + + public String getName() { + return name; + } + + public String getProject() { + return project; + } + + public ValueType getValue() { + return value; + } + + public String getDescription() { + return description; + } + + public Map getLabels() { + return labels; + } + + protected Spec(String project, EntityProto.EntitySpecV2 spec) { + this.project = project; + this.name = spec.getName(); + this.value = ValueType.fromProto(spec.getValueType()); + this.description = spec.getDescription(); + this.labels = spec.getLabelsMap(); + } + + protected EntityProto.EntitySpecV2 toProto() { + return EntityProto.EntitySpecV2.newBuilder() + .setName(name) + .setValueType(value.toProto()) + .setDescription(description) + .putAllLabels(labels) + .build(); + } + } + + public Spec getSpec() { + return spec; + } + + public Metadata getMetadata() { + return metadata; + } + + protected Entity(String project, EntityProto.Entity entity) { + if (entity.hasSpec()) this.spec = new Spec(project, entity.getSpec()); + if (entity.hasMeta()) this.metadata = new Metadata(entity.getMeta()); + } + + protected EntityProto.Entity toProto() { + return EntityProto.Entity.newBuilder() + .setSpec(spec.toProto()) + .setMeta( + EntityProto.EntityMeta.newBuilder() + .setCreatedTimestamp(Metadata.toProto(metadata.getCreatedTimestamp())) + .setLastUpdatedTimestamp(Metadata.toProto(metadata.getLastUpdatedTimestamp()))) + .build(); + } +} diff --git a/sdk/java/src/main/java/com/gojek/feast/core/Feature.java b/sdk/java/src/main/java/com/gojek/feast/core/Feature.java new file mode 100644 index 0000000..79cc456 --- /dev/null +++ b/sdk/java/src/main/java/com/gojek/feast/core/Feature.java @@ -0,0 +1,88 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * Copyright 2018-2021 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 com.gojek.feast.core; + +import feast.proto.core.FeatureProto; +import java.util.HashMap; +import java.util.Map; + +public class Feature { + private final String name; + private final ValueType valueType; + private final Map labels; + + public static Builder getBuilder(String name, ValueType valueType) { + return new Builder(name, valueType); + } + + protected Feature(String name, ValueType valueType, Map labels) { + this.name = name; + this.valueType = valueType; + this.labels = labels; + } + + public static class Builder { + private final String name; + private final ValueType valueType; + private final Map labels = new HashMap<>(); + + private Builder(String name, ValueType valueType) { + this.name = name; + this.valueType = valueType; + } + + public Builder addLabel(String key, String val) { + this.labels.put(key, val); + return this; + } + + public Builder addLabels(Map labels) { + this.labels.putAll(labels); + return this; + } + + public Feature build() { + return new Feature(name, valueType, labels); + } + } + + public String getName() { + return name; + } + + public ValueType getValueType() { + return valueType; + } + + public Map getLabels() { + return labels; + } + + protected Feature(FeatureProto.FeatureSpecV2 feature) { + this.name = feature.getName(); + this.valueType = ValueType.fromProto(feature.getValueType()); + this.labels = feature.getLabelsMap(); + } + + protected FeatureProto.FeatureSpecV2 toProto() { + return FeatureProto.FeatureSpecV2.newBuilder() + .setName(name) + .setValueType(valueType.toProto()) + .putAllLabels(labels) + .build(); + } +} diff --git a/sdk/java/src/main/java/com/gojek/feast/core/FeatureTable.java b/sdk/java/src/main/java/com/gojek/feast/core/FeatureTable.java new file mode 100644 index 0000000..8f68b58 --- /dev/null +++ b/sdk/java/src/main/java/com/gojek/feast/core/FeatureTable.java @@ -0,0 +1,220 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * Copyright 2018-2021 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 com.gojek.feast.core; + +import feast.proto.core.DataSourceProto; +import feast.proto.core.FeatureTableProto; +import java.time.Duration; +import java.util.*; +import java.util.stream.Collectors; + +public class FeatureTable { + private Spec spec; + private Metadata metadata; + + public static class Spec { + private final String project; + private final String name; + private final List entities; + private final List features; + private final Map labels; + private DataSourceProto.DataSource batchSource; + private DataSourceProto.DataSource streamSource; + private Duration maxAge; + + protected Spec( + String project, + String name, + List entities, + List features, + Map labels, + DataSourceProto.DataSource batchSource, + DataSourceProto.DataSource streamSource, + Duration maxAge) { + this.project = project; + this.name = name; + this.entities = entities; + this.features = features; + this.labels = labels; + this.batchSource = batchSource; + this.streamSource = streamSource; + this.maxAge = maxAge; + } + + public static Builder getBuilder(String project, String name) { + return new Builder(project, name); + } + + public static class Builder { + private final String project; + private final String name; + private final List entities = new LinkedList<>(); + private final List features = new LinkedList<>(); + private final Map labels = new HashMap<>(); + private DataSourceProto.DataSource batchSource; + private DataSourceProto.DataSource streamSource; + private Duration maxAge; + + private Builder(String project, String name) { + this.project = project; + this.name = name; + } + + public Spec build() { + return new Spec( + project, name, entities, features, labels, batchSource, streamSource, maxAge); + } + + public Builder addEntity(String entity) { + this.entities.add(entity); + return this; + } + + public Builder addEntities(Collection entities) { + this.entities.addAll(entities); + return this; + } + + public Builder addFeature(Feature feature) { + this.features.add(feature); + return this; + } + + public Builder addFeatures(Collection features) { + this.features.addAll(features); + return this; + } + + public Builder addLabel(String key, String val) { + this.labels.put(key, val); + return this; + } + + public Builder addLabels(Map labels) { + this.labels.putAll(labels); + return this; + } + + public Builder setBatchSource(DataSourceProto.DataSource batchSource) { + this.batchSource = batchSource; + return this; + } + + public Builder setStreamSource(DataSourceProto.DataSource streamSource) { + this.streamSource = streamSource; + return this; + } + + public Builder setMaxAge(Duration maxAge) { + this.maxAge = maxAge; + return this; + } + } + + public String getProject() { + return project; + } + + public String getName() { + return name; + } + + public List getEntities() { + return entities; + } + + public List getFeatures() { + return features; + } + + public Map getLabels() { + return labels; + } + + public DataSourceProto.DataSource getBatchSource() { + return batchSource; + } + + public DataSourceProto.DataSource getStreamSource() { + return streamSource; + } + + public Optional getMaxAge() { + return Optional.ofNullable(maxAge); + } + + protected Spec(String project, FeatureTableProto.FeatureTableSpec spec) { + this.project = project; + this.name = spec.getName(); + this.entities = spec.getEntitiesList(); + this.features = + spec.getFeaturesList().stream().map(Feature::new).collect(Collectors.toList()); + this.labels = spec.getLabelsMap(); + if (spec.hasBatchSource()) this.batchSource = spec.getBatchSource(); + if (spec.hasStreamSource()) this.streamSource = spec.getStreamSource(); + if (spec.hasMaxAge()) + this.maxAge = + Duration.ofSeconds(spec.getMaxAge().getSeconds(), spec.getMaxAge().getNanos()); + } + + protected FeatureTableProto.FeatureTableSpec toProto() { + FeatureTableProto.FeatureTableSpec.Builder builder = + FeatureTableProto.FeatureTableSpec.newBuilder() + .setName(name) + .addAllEntities(entities) + .addAllFeatures(features.stream().map(Feature::toProto).collect(Collectors.toList())) + .putAllLabels(labels); + if (batchSource != null) { + builder.setBatchSource(batchSource); + } + if (streamSource != null) { + builder.setStreamSource(streamSource); + } + if (maxAge != null) { + builder.setMaxAge( + com.google.protobuf.Duration.newBuilder() + .setSeconds(maxAge.getSeconds()) + .setNanos(maxAge.getNano()) + .build()); + } + return builder.build(); + } + } + + public Spec getSpec() { + return spec; + } + + public Metadata getMetadata() { + return metadata; + } + + protected FeatureTable(String project, FeatureTableProto.FeatureTable featureTable) { + if (featureTable.hasSpec()) this.spec = new Spec(project, featureTable.getSpec()); + if (featureTable.hasMeta()) this.metadata = new Metadata(featureTable.getMeta()); + } + + protected FeatureTableProto.FeatureTable toProto() { + return FeatureTableProto.FeatureTable.newBuilder() + .setSpec(spec.toProto()) + .setMeta( + FeatureTableProto.FeatureTableMeta.newBuilder() + .setLastUpdatedTimestamp(Metadata.toProto(metadata.getLastUpdatedTimestamp())) + .setCreatedTimestamp(Metadata.toProto(metadata.getCreatedTimestamp()))) + .build(); + } +} diff --git a/sdk/java/src/main/java/com/gojek/feast/core/Metadata.java b/sdk/java/src/main/java/com/gojek/feast/core/Metadata.java new file mode 100644 index 0000000..bbe9d1c --- /dev/null +++ b/sdk/java/src/main/java/com/gojek/feast/core/Metadata.java @@ -0,0 +1,53 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * Copyright 2018-2021 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 com.gojek.feast.core; + +import com.google.protobuf.Timestamp; +import feast.proto.core.EntityProto; +import feast.proto.core.FeatureTableProto; +import java.time.Instant; + +public class Metadata { + private final Instant createdTimestamp; + private final Instant lastUpdatedTimestamp; + + protected Metadata(EntityProto.EntityMeta meta) { + this.createdTimestamp = toInstant(meta.getCreatedTimestamp()); + this.lastUpdatedTimestamp = toInstant(meta.getLastUpdatedTimestamp()); + } + + protected Metadata(FeatureTableProto.FeatureTableMeta meta) { + this.createdTimestamp = toInstant(meta.getCreatedTimestamp()); + this.lastUpdatedTimestamp = toInstant(meta.getLastUpdatedTimestamp()); + } + + public Instant getCreatedTimestamp() { + return createdTimestamp; + } + + public Instant getLastUpdatedTimestamp() { + return lastUpdatedTimestamp; + } + + protected static Instant toInstant(Timestamp timestamp) { + return Instant.ofEpochSecond(timestamp.getSeconds(), timestamp.getNanos()); + } + + protected static Timestamp toProto(Instant instant) { + return Timestamp.newBuilder().setSeconds(instant.getEpochSecond()).build(); + } +} diff --git a/sdk/java/src/main/java/com/gojek/feast/core/ValueType.java b/sdk/java/src/main/java/com/gojek/feast/core/ValueType.java new file mode 100644 index 0000000..aa27fe7 --- /dev/null +++ b/sdk/java/src/main/java/com/gojek/feast/core/ValueType.java @@ -0,0 +1,62 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * Copyright 2018-2021 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 com.gojek.feast.core; + +import feast.proto.types.ValueProto; +import java.util.Arrays; + +public enum ValueType { + INVALID(0), + BYTES(1), + STRING(2), + INT32(3), + INT64(4), + DOUBLE(5), + FLOAT(6), + BOOL(7), + UNIX_TIMESTAMP(8), + BYTES_LIST(11), + STRING_LIST(12), + INT32_LIST(13), + INT64_LIST(14), + DOUBLE_LIST(15), + FLOAT_LIST(16), + BOOL_LIST(17), + UNIX_TIMESTAMP_LIST(18); + + /** @param number - is the matching {@link feast.proto.types.ValueProto.Value} enumeration */ + ValueType(int number) { + this.number = number; + } + + private final int number; + + static ValueType fromProto(ValueProto.ValueType.Enum valueType) { + return Arrays.stream(values()) + .filter(value -> value.number == valueType.getNumber()) + .findFirst() + .orElse(INVALID); + } + + ValueProto.ValueType.Enum toProto() { + return ValueProto.ValueType.Enum.forNumber(number); + } + + int getNumber() { + return this.number; + } +} diff --git a/sdk/java/src/test/java/com/gojek/feast/FeastClientTest.java b/sdk/java/src/test/java/com/gojek/feast/FeastClientTest.java index c458a06..190551b 100644 --- a/sdk/java/src/test/java/com/gojek/feast/FeastClientTest.java +++ b/sdk/java/src/test/java/com/gojek/feast/FeastClientTest.java @@ -17,6 +17,7 @@ package com.gojek.feast; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.AdditionalAnswers.delegatesTo; import static org.mockito.Mockito.mock; @@ -31,26 +32,18 @@ import feast.proto.serving.ServingServiceGrpc.ServingServiceImplBase; import feast.proto.types.ValueProto.Value; import io.grpc.*; -import io.grpc.ServerCall.Listener; -import io.grpc.inprocess.InProcessChannelBuilder; -import io.grpc.inprocess.InProcessServerBuilder; import io.grpc.stub.StreamObserver; -import io.grpc.testing.GrpcCleanupRule; import java.time.Instant; import java.util.Arrays; import java.util.HashMap; import java.util.List; import java.util.Optional; -import java.util.concurrent.atomic.AtomicBoolean; import org.junit.Before; -import org.junit.Rule; import org.junit.Test; public class FeastClientTest { - private final String AUTH_TOKEN = "test token"; - - @Rule public GrpcCleanupRule grpcRule; - private AtomicBoolean isAuthenticated; + protected final String AUTH_TOKEN = "test token"; + private GrpcMock grpcMock; private ServingServiceImplBase servingMock = mock( @@ -70,46 +63,15 @@ public void getOnlineFeaturesV2( } })); - // Mock Authentication interceptor will flag authenticated request by setting isAuthenticated to - // true. - private ServerInterceptor mockAuthInterceptor = - new ServerInterceptor() { - @Override - public Listener interceptCall( - ServerCall call, Metadata headers, ServerCallHandler next) { - final Metadata.Key authorizationKey = - Metadata.Key.of("Authorization", Metadata.ASCII_STRING_MARSHALLER); - if (headers.containsKey(authorizationKey)) { - isAuthenticated.set(true); - } - return next.startCall(call, headers); - } - }; - private FeastClient client; private FeastClient authenticatedClient; @Before public void setup() throws Exception { - this.grpcRule = new GrpcCleanupRule(); - this.isAuthenticated = new AtomicBoolean(false); - // setup fake serving service - String serverName = InProcessServerBuilder.generateName(); - this.grpcRule.register( - InProcessServerBuilder.forName(serverName) - .directExecutor() - .addService(this.servingMock) - .intercept(mockAuthInterceptor) - .build() - .start()); - - // setup test feast client target - ManagedChannel channel = - this.grpcRule.register( - InProcessChannelBuilder.forName(serverName).directExecutor().build()); - this.client = new FeastClient(channel, Optional.empty()); + this.grpcMock = new GrpcMock(this.servingMock); + this.client = new FeastClient(grpcMock.getChannel(), Optional.empty()); this.authenticatedClient = - new FeastClient(channel, Optional.of(new JwtCallCredentials(AUTH_TOKEN))); + new FeastClient(grpcMock.getChannel(), Optional.of(new JwtCallCredentials(AUTH_TOKEN))); } @Test @@ -119,9 +81,9 @@ public void shouldGetOnlineFeatures() { @Test public void shouldAuthenticateAndGetOnlineFeatures() { - isAuthenticated.set(false); + grpcMock.resetAuthentication(); shouldGetOnlineFeaturesWithClient(this.authenticatedClient); - assertEquals(isAuthenticated.get(), true); + assertTrue(grpcMock.isAuthenticated()); } private void shouldGetOnlineFeaturesWithClient(FeastClient client) { diff --git a/sdk/java/src/test/java/com/gojek/feast/GrpcMock.java b/sdk/java/src/test/java/com/gojek/feast/GrpcMock.java new file mode 100644 index 0000000..2381848 --- /dev/null +++ b/sdk/java/src/test/java/com/gojek/feast/GrpcMock.java @@ -0,0 +1,78 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * Copyright 2018-2021 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 com.gojek.feast; + +import io.grpc.*; +import io.grpc.inprocess.InProcessChannelBuilder; +import io.grpc.inprocess.InProcessServerBuilder; +import io.grpc.testing.GrpcCleanupRule; +import java.io.IOException; +import java.util.concurrent.atomic.AtomicBoolean; +import org.junit.Rule; + +public class GrpcMock { + + @Rule public GrpcCleanupRule grpcRule; + protected AtomicBoolean isAuthenticated; + private ManagedChannel channel; + + public GrpcMock(BindableService server) throws IOException { + this.grpcRule = new GrpcCleanupRule(); + this.isAuthenticated = new AtomicBoolean(false); + // setup fake serving service + String serverName = InProcessServerBuilder.generateName(); + this.grpcRule.register( + InProcessServerBuilder.forName(serverName) + .directExecutor() + .addService(server) + .intercept(mockAuthInterceptor) + .build() + .start()); + // setup test feast client target + this.channel = + this.grpcRule.register( + InProcessChannelBuilder.forName(serverName).directExecutor().build()); + } + + // Mock Authentication interceptor will flag authenticated request by setting isAuthenticated to + // true. + private final ServerInterceptor mockAuthInterceptor = + new ServerInterceptor() { + @Override + public ServerCall.Listener interceptCall( + ServerCall call, Metadata headers, ServerCallHandler next) { + final Metadata.Key authorizationKey = + Metadata.Key.of("Authorization", Metadata.ASCII_STRING_MARSHALLER); + if (headers.containsKey(authorizationKey)) { + isAuthenticated.set(true); + } + return next.startCall(call, headers); + } + }; + + public ManagedChannel getChannel() { + return channel; + } + + public boolean isAuthenticated() { + return isAuthenticated.get(); + } + + public void resetAuthentication() { + isAuthenticated.set(false); + } +} diff --git a/sdk/java/src/test/java/com/gojek/feast/core/CoreClientTest.java b/sdk/java/src/test/java/com/gojek/feast/core/CoreClientTest.java new file mode 100644 index 0000000..deecc35 --- /dev/null +++ b/sdk/java/src/test/java/com/gojek/feast/core/CoreClientTest.java @@ -0,0 +1,105 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * Copyright 2018-2021 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 com.gojek.feast.core; + +import static org.mockito.AdditionalAnswers.delegatesTo; +import static org.mockito.Mockito.mock; + +import com.gojek.feast.GrpcMock; +import feast.proto.core.CoreServiceGrpc; +import feast.proto.core.DataSourceProto; +import java.util.*; +import org.junit.Before; +import org.junit.Test; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeEach; + +public class CoreClientTest { + + private final CoreServiceGrpc.CoreServiceImplBase coreMock = + mock(CoreServiceGrpc.CoreServiceImplBase.class, delegatesTo(new CoreServiceImplMock())); + + private CoreClient client; + + @Before + public void setup() throws Exception { + GrpcMock grpcMock = new GrpcMock(this.coreMock); + this.client = new CoreClient(grpcMock.getChannel(), Optional.empty()); + } + + @BeforeEach + void setUp() { + CoreServiceImplMock.clear(); + } + + @Test + public void testProject() { + Assertions.assertEquals(CoreServiceImplMock.PROJECT, client.listProjects().get(0)); + } + + @Test + public void testEntity() { + final String name = "entity"; + Entity.Spec spec = + Entity.Spec.getBuilder(CoreServiceImplMock.PROJECT, name) + .setValue(ValueType.BOOL) + .setDescription("description") + .build(); + client.apply(spec); + Assertions.assertEquals( + spec.toProto(), + client.getEntity(CoreServiceImplMock.PROJECT, name).get().getSpec().toProto()); + spec.getLabels().put("l1", "v1"); + client.apply(spec); + Assertions.assertTrue( + client + .getEntity(CoreServiceImplMock.PROJECT, name) + .get() + .getSpec() + .getLabels() + .containsKey("l1")); + Assertions.assertEquals(1, client.listEntities(CoreServiceImplMock.PROJECT).size()); + } + + @Test + public void testFeatureTable() { + final String name = "featureTable"; + FeatureTable.Spec spec = + FeatureTable.Spec.getBuilder(CoreServiceImplMock.PROJECT, name) + .addEntity("entity") + .addLabel("key", "value") + .setBatchSource( + DataSourceProto.DataSource.newBuilder() + .setType(DataSourceProto.DataSource.SourceType.BATCH_FILE) + .build()) + .build(); + client.apply(spec); + Assertions.assertEquals( + spec.toProto(), + client.getFeatureTable(CoreServiceImplMock.PROJECT, name).get().getSpec().toProto()); + spec.getFeatures().add(Feature.getBuilder("feature", ValueType.FLOAT).build()); + client.apply(spec); + Assertions.assertTrue( + client.getFeatureTable(CoreServiceImplMock.PROJECT, name).get().getSpec().getFeatures() + .stream() + .anyMatch(f -> f.getName().equals("feature"))); + Assertions.assertEquals(1, client.listFeatureTables(CoreServiceImplMock.PROJECT).size()); + Assertions.assertEquals(1, client.listFeatures(CoreServiceImplMock.PROJECT).size()); + client.deleteFeatureTable(CoreServiceImplMock.PROJECT, name); + Assertions.assertTrue(client.listFeatureTables(CoreServiceImplMock.PROJECT).isEmpty()); + } +} diff --git a/sdk/java/src/test/java/com/gojek/feast/core/CoreServiceImplMock.java b/sdk/java/src/test/java/com/gojek/feast/core/CoreServiceImplMock.java new file mode 100644 index 0000000..a6b07f3 --- /dev/null +++ b/sdk/java/src/test/java/com/gojek/feast/core/CoreServiceImplMock.java @@ -0,0 +1,214 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * Copyright 2018-2021 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 com.gojek.feast.core; + +import com.google.protobuf.Timestamp; +import feast.proto.core.CoreServiceGrpc; +import feast.proto.core.CoreServiceProto; +import feast.proto.core.EntityProto; +import feast.proto.core.FeatureTableProto; +import io.grpc.Status; +import io.grpc.stub.StreamObserver; +import java.time.Instant; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +public class CoreServiceImplMock extends CoreServiceGrpc.CoreServiceImplBase { + protected static final String PROJECT = "project"; + + private static final Map entities = new ConcurrentHashMap<>(); + private static final Map featureTables = + new ConcurrentHashMap<>(); + + public static void clear() { + entities.clear(); + featureTables.clear(); + } + + @Override + public void getEntity( + CoreServiceProto.GetEntityRequest request, + StreamObserver responseObserver) { + if (!request.getProject().equals(PROJECT)) + responseObserver.onError(Status.FAILED_PRECONDITION.asRuntimeException()); + CoreServiceProto.GetEntityResponse.Builder response = + CoreServiceProto.GetEntityResponse.newBuilder(); + entities.computeIfPresent( + request.getName(), + (k, v) -> { + response.setEntity(v); + return v; + }); + responseObserver.onNext(response.build()); + responseObserver.onCompleted(); + } + + @Override + public void applyEntity( + CoreServiceProto.ApplyEntityRequest request, + StreamObserver responseObserver) { + if (!request.getProject().equals(PROJECT)) + responseObserver.onError(Status.FAILED_PRECONDITION.asRuntimeException()); + EntityProto.EntitySpecV2 spec = request.getSpec(); + EntityProto.Entity.Builder entity = + entities.computeIfAbsent( + request.getSpec().getName(), + k -> + EntityProto.Entity.newBuilder() + .setSpec(spec) + .setMeta(EntityProto.EntityMeta.newBuilder().setCreatedTimestamp(now()))); + EntityProto.EntitySpecV2 oldSpec = entity.getSpec(); + if (!oldSpec.getName().equals(spec.getName()) + || !oldSpec.getValueType().equals(spec.getValueType())) + responseObserver.onError(Status.INVALID_ARGUMENT.asRuntimeException()); + entity.setSpec(spec).getMetaBuilder().setLastUpdatedTimestamp(now()); + responseObserver.onNext( + CoreServiceProto.ApplyEntityResponse.newBuilder().setEntity(entity).build()); + responseObserver.onCompleted(); + } + + @Override + public void applyFeatureTable( + CoreServiceProto.ApplyFeatureTableRequest request, + StreamObserver responseObserver) { + if (!request.getProject().equals(PROJECT)) + responseObserver.onError(Status.FAILED_PRECONDITION.asRuntimeException()); + FeatureTableProto.FeatureTableSpec spec = request.getTableSpec(); + FeatureTableProto.FeatureTable.Builder featureTable = + featureTables.computeIfAbsent( + spec.getName(), + k -> + FeatureTableProto.FeatureTable.newBuilder() + .setSpec(spec) + .setMeta( + FeatureTableProto.FeatureTableMeta.newBuilder() + .setCreatedTimestamp(now()))); + FeatureTableProto.FeatureTableSpec oldSpec = featureTable.getSpec(); + if (!oldSpec.getName().equals(spec.getName()) + || !oldSpec.getEntitiesList().equals(spec.getEntitiesList())) + responseObserver.onError(Status.INVALID_ARGUMENT.asRuntimeException()); + featureTable.setSpec(spec).getMetaBuilder().setLastUpdatedTimestamp(now()); + responseObserver.onNext( + CoreServiceProto.ApplyFeatureTableResponse.newBuilder().setTable(featureTable).build()); + responseObserver.onCompleted(); + } + + @Override + public void getFeatureTable( + CoreServiceProto.GetFeatureTableRequest request, + StreamObserver responseObserver) { + if (!request.getProject().equals(PROJECT)) + responseObserver.onError(Status.FAILED_PRECONDITION.asRuntimeException()); + CoreServiceProto.GetFeatureTableResponse.Builder response = + CoreServiceProto.GetFeatureTableResponse.newBuilder(); + featureTables.computeIfPresent( + request.getName(), + (k, v) -> { + response.setTable(v); + return v; + }); + responseObserver.onNext(response.build()); + responseObserver.onCompleted(); + } + + @Override + public void listFeatures( + CoreServiceProto.ListFeaturesRequest request, + StreamObserver responseObserver) { + if (!request.getFilter().getProject().equals(PROJECT)) + responseObserver.onError(Status.FAILED_PRECONDITION.asRuntimeException()); + if (request.getFilter().getEntitiesCount() > 0 || request.getFilter().getLabelsCount() > 0) + responseObserver.onError(Status.UNIMPLEMENTED.asRuntimeException()); + CoreServiceProto.ListFeaturesResponse.Builder response = + CoreServiceProto.ListFeaturesResponse.newBuilder(); + featureTables.values().stream() + .flatMap(featureTable -> featureTable.getSpec().getFeaturesList().stream()) + .forEach(feature -> response.putFeatures(feature.getName(), feature)); + responseObserver.onNext(response.build()); + responseObserver.onCompleted(); + } + + @Override + public void listEntities( + CoreServiceProto.ListEntitiesRequest request, + StreamObserver responseObserver) { + if (!request.getFilter().getProject().equals(PROJECT)) + responseObserver.onError(Status.FAILED_PRECONDITION.asRuntimeException()); + if (request.getFilter().getLabelsCount() > 0) + responseObserver.onError(Status.UNIMPLEMENTED.asRuntimeException()); + CoreServiceProto.ListEntitiesResponse.Builder response = + CoreServiceProto.ListEntitiesResponse.newBuilder(); + entities.values().forEach(response::addEntities); + responseObserver.onNext(response.build()); + responseObserver.onCompleted(); + } + + @Override + public void createProject( + CoreServiceProto.CreateProjectRequest request, + StreamObserver responseObserver) { + responseObserver.onError(Status.UNIMPLEMENTED.asRuntimeException()); + } + + @Override + public void archiveProject( + CoreServiceProto.ArchiveProjectRequest request, + StreamObserver responseObserver) { + responseObserver.onError(Status.UNIMPLEMENTED.asRuntimeException()); + } + + @Override + public void listProjects( + CoreServiceProto.ListProjectsRequest request, + StreamObserver responseObserver) { + responseObserver.onNext( + CoreServiceProto.ListProjectsResponse.newBuilder().addProjects(PROJECT).build()); + responseObserver.onCompleted(); + } + + @Override + public void listFeatureTables( + CoreServiceProto.ListFeatureTablesRequest request, + StreamObserver responseObserver) { + if (!request.getFilter().getProject().equals(PROJECT)) + responseObserver.onError(Status.FAILED_PRECONDITION.asRuntimeException()); + if (request.getFilter().getLabelsCount() > 0) + responseObserver.onError(Status.UNIMPLEMENTED.asRuntimeException()); + CoreServiceProto.ListFeatureTablesResponse.Builder response = + CoreServiceProto.ListFeatureTablesResponse.newBuilder(); + featureTables.values().forEach(response::addTables); + responseObserver.onNext(response.build()); + responseObserver.onCompleted(); + } + + @Override + public void deleteFeatureTable( + CoreServiceProto.DeleteFeatureTableRequest request, + StreamObserver responseObserver) { + if (!request.getProject().equals(PROJECT)) + responseObserver.onError(Status.FAILED_PRECONDITION.asRuntimeException()); + featureTables.remove(request.getName()); + CoreServiceProto.DeleteFeatureTableResponse.Builder response = + CoreServiceProto.DeleteFeatureTableResponse.newBuilder(); + responseObserver.onNext(response.build()); + responseObserver.onCompleted(); + } + + private Timestamp now() { + return Timestamp.newBuilder().setSeconds(Instant.now().getEpochSecond()).build(); + } +} diff --git a/sdk/java/src/test/java/com/gojek/feast/core/ProtobufSerializationTest.java b/sdk/java/src/test/java/com/gojek/feast/core/ProtobufSerializationTest.java new file mode 100644 index 0000000..e9d9bbe --- /dev/null +++ b/sdk/java/src/test/java/com/gojek/feast/core/ProtobufSerializationTest.java @@ -0,0 +1,92 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * Copyright 2018-2021 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 com.gojek.feast.core; + +import com.google.protobuf.Descriptors; +import com.google.protobuf.Duration; +import com.google.protobuf.Timestamp; +import feast.proto.core.EntityProto; +import feast.proto.core.FeatureProto; +import feast.proto.core.FeatureTableProto; +import feast.proto.types.ValueProto; +import org.junit.Assert; +import org.junit.Test; + +public class ProtobufSerializationTest { + public static final Timestamp TIMESTAMP = Timestamp.newBuilder().setSeconds(100000).build(); + + @Test + public void testValueType() { + for (Descriptors.EnumValueDescriptor value : + ValueProto.ValueType.Enum.getDescriptor().getValues()) { + ValueType valueType = ValueType.valueOf(value.getName()); + Assert.assertNotNull("ValueType enum=" + value.getName() + " Is missing.", valueType); + Assert.assertEquals(valueType.getNumber(), value.getNumber()); + } + } + + @Test + public void testEntity() { + EntityProto.Entity protoEntity = + EntityProto.Entity.newBuilder() + .setMeta( + EntityProto.EntityMeta.newBuilder() + .setCreatedTimestamp(TIMESTAMP) + .setLastUpdatedTimestamp(TIMESTAMP)) + .setSpec( + EntityProto.EntitySpecV2.newBuilder() + .setName("name") + .setDescription("description") + .setValueType(ValueProto.ValueType.Enum.BOOL_LIST) + .putLabels("key", "val")) + .build(); + Entity entity = new Entity("project", protoEntity); + Assert.assertEquals(protoEntity, entity.toProto()); + } + + @Test + public void testFeature() { + FeatureProto.FeatureSpecV2 featureProto = + FeatureProto.FeatureSpecV2.newBuilder() + .setName("name") + .setValueType(ValueProto.ValueType.Enum.BOOL) + .putLabels("key", "val") + .build(); + Feature feature = new Feature(featureProto); + Assert.assertEquals(featureProto, feature.toProto()); + } + + @Test + public void testFeatureTable() { + FeatureTableProto.FeatureTable featureTableProto = + FeatureTableProto.FeatureTable.newBuilder() + .setMeta( + FeatureTableProto.FeatureTableMeta.newBuilder() + .setCreatedTimestamp(TIMESTAMP) + .setLastUpdatedTimestamp(TIMESTAMP)) + .setSpec( + FeatureTableProto.FeatureTableSpec.newBuilder() + .setName("name") + .setMaxAge(Duration.newBuilder().setSeconds(10).setNanos(10)) + .addEntities("entity") + .putLabels("key", "val") + .addFeatures(FeatureProto.FeatureSpecV2.getDefaultInstance())) + .build(); + FeatureTable featureTable = new FeatureTable("project", featureTableProto); + Assert.assertEquals(featureTableProto, featureTable.toProto()); + } +}