You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tz...@apache.org on 2020/02/10 03:43:38 UTC
[flink-statefun] 09/09: [FLINK-15955] Configure a remote function
acorrding to its spec
This is an automated email from the ASF dual-hosted git repository.
tzulitai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-statefun.git
commit 1158cd18b3a357de7b3265115e05bd278619acb3
Author: Igal Shilman <ig...@gmail.com>
AuthorDate: Fri Feb 7 18:35:28 2020 +0100
[FLINK-15955] Configure a remote function acorrding to its spec
This commit would instantiate the correct remote function
provider with its corresponding function spec, as specified
in the module.yaml.
This closes #18.
---
.../core/jsonmodule/HttpFunctionProvider.java | 33 +++++++++
.../statefun/flink/core/jsonmodule/JsonModule.java | 78 +++++++++++++++++-----
2 files changed, 96 insertions(+), 15 deletions(-)
diff --git a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/jsonmodule/HttpFunctionProvider.java b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/jsonmodule/HttpFunctionProvider.java
new file mode 100644
index 0000000..58c1c7c
--- /dev/null
+++ b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/jsonmodule/HttpFunctionProvider.java
@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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
+ *
+ * http://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 org.apache.flink.statefun.flink.core.jsonmodule;
+
+import java.util.Map;
+import org.apache.flink.statefun.sdk.FunctionType;
+import org.apache.flink.statefun.sdk.StatefulFunction;
+import org.apache.flink.statefun.sdk.StatefulFunctionProvider;
+
+public class HttpFunctionProvider implements StatefulFunctionProvider {
+ public HttpFunctionProvider(Map<FunctionType, HttpFunctionSpec> specs) {}
+
+ @Override
+ public StatefulFunction functionOfType(FunctionType type) {
+ throw new UnsupportedOperationException("Not implemented yet");
+ }
+}
diff --git a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/jsonmodule/JsonModule.java b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/jsonmodule/JsonModule.java
index 4465864..7f2219f 100644
--- a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/jsonmodule/JsonModule.java
+++ b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/jsonmodule/JsonModule.java
@@ -18,27 +18,37 @@
package org.apache.flink.statefun.flink.core.jsonmodule;
import static java.lang.String.format;
+import static java.util.stream.Collectors.groupingBy;
import static java.util.stream.Collectors.toMap;
+import static org.apache.flink.statefun.flink.core.common.Maps.transformValues;
import com.google.protobuf.Descriptors;
import com.google.protobuf.Message;
import java.io.IOException;
import java.net.InetSocketAddress;
+import java.net.URI;
import java.net.URL;
+import java.util.List;
+import java.util.Locale;
import java.util.Map;
+import java.util.Map.Entry;
import java.util.Objects;
import java.util.Optional;
+import java.util.Set;
import java.util.function.Function;
+import java.util.stream.Collector;
import java.util.stream.StreamSupport;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
import org.apache.flink.statefun.flink.common.ResourceLocator;
import org.apache.flink.statefun.flink.common.json.NamespaceNamePair;
import org.apache.flink.statefun.flink.common.json.Selectors;
import org.apache.flink.statefun.flink.common.protobuf.ProtobufDescriptorMap;
+import org.apache.flink.statefun.flink.core.jsonmodule.FunctionSpec.Kind;
import org.apache.flink.statefun.flink.core.protorouter.ProtobufRouter;
import org.apache.flink.statefun.flink.io.spi.JsonIngressSpec;
import org.apache.flink.statefun.sdk.FunctionType;
import org.apache.flink.statefun.sdk.IngressType;
+import org.apache.flink.statefun.sdk.StatefulFunctionProvider;
import org.apache.flink.statefun.sdk.io.IngressIdentifier;
import org.apache.flink.statefun.sdk.io.Router;
import org.apache.flink.statefun.sdk.spi.StatefulFunctionModule;
@@ -64,17 +74,31 @@ final class JsonModule implements StatefulFunctionModule {
}
private void configureFunctions(Binder binder, Iterable<? extends JsonNode> functions) {
- final Map<FunctionType, GrpcFunctionSpec> definedFunctions =
+ final Map<Kind, Map<FunctionType, FunctionSpec>> definedFunctions =
StreamSupport.stream(functions.spliterator(), false)
- .map(JsonModule::parseRemoteFunctionSpec)
- .collect(toMap(GrpcFunctionSpec::functionType, Function.identity()));
-
- // currently we had a single function type that can be specified at a module.yaml
- // and this is the RemoteFunction. Therefore we translate immediately the function spec
- // to a (GRPC) RemoteFunctionProvider.
- GrpcFunctionProvider provider = new GrpcFunctionProvider(definedFunctions);
- for (FunctionType type : definedFunctions.keySet()) {
- binder.bindFunctionProvider(type, provider);
+ .map(JsonModule::parseFunctionSpec)
+ .collect(groupingBy(FunctionSpec::kind, groupByFunctionType()));
+
+ for (Entry<Kind, Map<FunctionType, FunctionSpec>> entry : definedFunctions.entrySet()) {
+ StatefulFunctionProvider provider = functionProvider(entry.getKey(), entry.getValue());
+ Set<FunctionType> functionTypes = entry.getValue().keySet();
+ for (FunctionType type : functionTypes) {
+ binder.bindFunctionProvider(type, provider);
+ }
+ }
+ }
+
+ private static StatefulFunctionProvider functionProvider(
+ Kind kind, Map<FunctionType, FunctionSpec> definedFunctions) {
+ switch (kind) {
+ case HTTP:
+ return new HttpFunctionProvider(
+ transformValues(definedFunctions, HttpFunctionSpec.class::cast));
+ case GRPC:
+ return new GrpcFunctionProvider(
+ transformValues(definedFunctions, GrpcFunctionSpec.class::cast));
+ default:
+ throw new IllegalStateException("Unexpected value: " + kind);
}
}
@@ -103,7 +127,6 @@ final class JsonModule implements StatefulFunctionModule {
// ----------------------------------------------------------------------------------------------------------
// Ingresses
// ----------------------------------------------------------------------------------------------------------
-
private static IngressType ingressType(JsonNode spec) {
String typeString = Selectors.textAt(spec, Pointers.Ingress.META_TYPE);
NamespaceNamePair nn = NamespaceNamePair.from(typeString);
@@ -119,7 +142,6 @@ final class JsonModule implements StatefulFunctionModule {
// ----------------------------------------------------------------------------------------------------------
// Routers
// ----------------------------------------------------------------------------------------------------------
-
private static Router<Message> dynamicRouter(JsonNode router) {
String addressTemplate = Selectors.textAt(router, Pointers.Routers.SPEC_TARGET);
String descriptorSetPath = Selectors.textAt(router, Pointers.Routers.SPEC_DESCRIPTOR);
@@ -142,6 +164,10 @@ final class JsonModule implements StatefulFunctionModule {
private static ProtobufDescriptorMap protobufDescriptorMap(String descriptorSetPath) {
try {
URL url = ResourceLocator.findNamedResource(descriptorSetPath);
+ if (url == null) {
+ throw new IllegalArgumentException(
+ "Unable to locate a Protobuf descriptor set at " + descriptorSetPath);
+ }
return ProtobufDescriptorMap.from(url);
} catch (IOException e) {
throw new IllegalStateException(
@@ -168,10 +194,23 @@ final class JsonModule implements StatefulFunctionModule {
// Functions
// ----------------------------------------------------------------------------------------------------------
- private static GrpcFunctionSpec parseRemoteFunctionSpec(JsonNode functionNode) {
+ private static FunctionSpec parseFunctionSpec(JsonNode functionNode) {
+ String functionKind = Selectors.textAt(functionNode, Pointers.Functions.META_KIND);
+ FunctionSpec.Kind kind = Kind.valueOf(functionKind.toUpperCase(Locale.getDefault()));
FunctionType functionType = functionType(functionNode);
- InetSocketAddress functionAddress = functionAddress(functionNode);
- return new GrpcFunctionSpec(functionType, functionAddress);
+ switch (kind) {
+ case HTTP:
+ return new HttpFunctionSpec(
+ functionType, functionUri(functionNode), functionStates(functionNode));
+ case GRPC:
+ return new GrpcFunctionSpec(functionType, functionAddress(functionNode));
+ default:
+ throw new IllegalArgumentException("Unrecognized function kind " + functionKind);
+ }
+ }
+
+ private static List<String> functionStates(JsonNode functionNode) {
+ return Selectors.textListAt(functionNode, Pointers.Functions.FUNCTION_STATES);
}
private static FunctionType functionType(JsonNode functionNode) {
@@ -185,4 +224,13 @@ final class JsonModule implements StatefulFunctionModule {
int port = Selectors.integerAt(functionNode, Pointers.Functions.FUNCTION_PORT);
return new InetSocketAddress(host, port);
}
+
+ private static URI functionUri(JsonNode functionNode) {
+ String uri = Selectors.textAt(functionNode, Pointers.Functions.FUNCTION_ENDPOINT);
+ return URI.create(uri);
+ }
+
+ private static Collector<FunctionSpec, ?, Map<FunctionType, FunctionSpec>> groupByFunctionType() {
+ return toMap(FunctionSpec::functionType, Function.identity());
+ }
}