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/12/04 04:09:39 UTC
[flink-statefun] 02/03: [FLINK-20334] [core] Implement V3 module
YAML with function endpoint templating
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 ee051dd25d88d9da38bc33f91788997975fa5e8e
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
AuthorDate: Mon Nov 30 15:20:42 2020 +0800
[FLINK-20334] [core] Implement V3 module YAML with function endpoint templating
This closes #182.
---
.../src/test/resources/remote-module/module.yaml | 18 +-
.../statefun/module.yaml | 15 +-
.../statefun-python-greeter-example/module.yaml | 15 +-
.../statefun-python-k8s-example/module.yaml | 13 +-
...tionSpec.java => HttpFunctionEndpointSpec.java} | 92 ++++-----
.../flink/core/httpfn/HttpFunctionProvider.java | 2 +-
.../flink/core/httpfn/HttpFunctionSpec.java | 5 -
...der.java => TemplatedHttpFunctionProvider.java} | 56 ++++--
.../flink/core/httpfn/UnixDomainHttpEndpoint.java | 8 +
.../flink/core/jsonmodule/FormatVersion.java | 5 +-
.../jsonmodule/FunctionEndpointJsonEntity.java | 206 +++++++++++++++++++++
...ormatVersion.java => FunctionEndpointSpec.java} | 44 +++--
.../statefun/flink/core/jsonmodule/JsonModule.java | 13 +-
.../core/httpfn/UnixDomainHttpEndpointTest.java | 11 ++
.../flink/core/jsonmodule/JsonModuleV3Test.java | 114 ++++++++++++
.../src/test/resources/module-v3_0/module.yaml | 85 +++++++++
16 files changed, 572 insertions(+), 130 deletions(-)
diff --git a/statefun-e2e-tests/statefun-exactly-once-remote-e2e/src/test/resources/remote-module/module.yaml b/statefun-e2e-tests/statefun-exactly-once-remote-e2e/src/test/resources/remote-module/module.yaml
index d9a80e2..1925210 100644
--- a/statefun-e2e-tests/statefun-exactly-once-remote-e2e/src/test/resources/remote-module/module.yaml
+++ b/statefun-e2e-tests/statefun-exactly-once-remote-e2e/src/test/resources/remote-module/module.yaml
@@ -13,26 +13,20 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-version: "1.0"
+version: "3.0"
module:
meta:
type: remote
spec:
- functions:
- - function:
+ endpoints:
+ - endpoint:
meta:
kind: http
- type: org.apache.flink.statefun.e2e.remote/counter
spec:
- endpoint: http://remote-function:8000/service
- maxNumBatchRequests: 10000
- - function:
- meta:
- kind: http
- type: org.apache.flink.statefun.e2e.remote/forward-function
- spec:
- endpoint: http://remote-function:8000/service
+ typename:
+ namespace: org.apache.flink.statefun.e2e.remote
+ urlPathTemplate: http://remote-function:8000/service
maxNumBatchRequests: 10000
ingresses:
- ingress:
diff --git a/statefun-examples/statefun-async-python-example/statefun/module.yaml b/statefun-examples/statefun-async-python-example/statefun/module.yaml
index 762e464..f4b197d 100644
--- a/statefun-examples/statefun-async-python-example/statefun/module.yaml
+++ b/statefun-examples/statefun-async-python-example/statefun/module.yaml
@@ -12,20 +12,23 @@
# 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.
-version: "1.0"
+version: "3.0"
module:
meta:
type: remote
spec:
- functions:
- - function:
+ endpoints:
+ - endpoint:
meta:
kind: http
- type: example/greeter
spec:
- endpoint: http://python-worker:8000/statefun
+ typename:
+ namespace: example
+ type: greeter
+ urlPathTemplate: http://python-worker:8000/statefun
maxNumBatchRequests: 500
- timeout: 2min
+ timeouts:
+ call: 2min
ingresses:
- ingress:
meta:
diff --git a/statefun-examples/statefun-python-greeter-example/module.yaml b/statefun-examples/statefun-python-greeter-example/module.yaml
index 762e464..f4b197d 100644
--- a/statefun-examples/statefun-python-greeter-example/module.yaml
+++ b/statefun-examples/statefun-python-greeter-example/module.yaml
@@ -12,20 +12,23 @@
# 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.
-version: "1.0"
+version: "3.0"
module:
meta:
type: remote
spec:
- functions:
- - function:
+ endpoints:
+ - endpoint:
meta:
kind: http
- type: example/greeter
spec:
- endpoint: http://python-worker:8000/statefun
+ typename:
+ namespace: example
+ type: greeter
+ urlPathTemplate: http://python-worker:8000/statefun
maxNumBatchRequests: 500
- timeout: 2min
+ timeouts:
+ call: 2min
ingresses:
- ingress:
meta:
diff --git a/statefun-examples/statefun-python-k8s-example/module.yaml b/statefun-examples/statefun-python-k8s-example/module.yaml
index cb59fd5..813ee1f 100644
--- a/statefun-examples/statefun-python-k8s-example/module.yaml
+++ b/statefun-examples/statefun-python-k8s-example/module.yaml
@@ -12,20 +12,23 @@
# 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.
-version: "1.0"
+version: "3.0"
module:
meta:
type: remote
spec:
- functions:
- - function:
+ endpoints:
+ - endpoint:
meta:
kind: http
- type: k8s-demo/greeter
spec:
+ typename:
+ namespace: k8s-demo
+ type: greeter
endpoint: http://statefun-python:8000/statefun
maxNumBatchRequests: 500
- timeout: 2min
+ timeouts:
+ call: 2min
ingresses:
- ingress:
meta:
diff --git a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/HttpFunctionSpec.java b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/HttpFunctionEndpointSpec.java
similarity index 68%
copy from statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/HttpFunctionSpec.java
copy to statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/HttpFunctionEndpointSpec.java
index 3a8f653..d5731d1 100644
--- a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/HttpFunctionSpec.java
+++ b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/HttpFunctionEndpointSpec.java
@@ -17,18 +17,14 @@
*/
package org.apache.flink.statefun.flink.core.httpfn;
-import java.io.Serializable;
-import java.net.URI;
import java.time.Duration;
-import java.util.ArrayList;
-import java.util.List;
import java.util.Objects;
-import org.apache.flink.statefun.flink.core.jsonmodule.FunctionSpec;
+import org.apache.flink.statefun.flink.core.jsonmodule.FunctionEndpointSpec;
import org.apache.flink.statefun.sdk.FunctionType;
+import org.apache.flink.statefun.sdk.FunctionTypeNamespaceMatcher;
+import org.apache.flink.types.Either;
-public final class HttpFunctionSpec implements FunctionSpec, Serializable {
-
- private static final long serialVersionUID = 1;
+public final class HttpFunctionEndpointSpec implements FunctionEndpointSpec {
private static final Duration DEFAULT_HTTP_TIMEOUT = Duration.ofMinutes(1);
private static final Duration DEFAULT_HTTP_CONNECT_TIMEOUT = Duration.ofSeconds(10);
@@ -36,41 +32,40 @@ public final class HttpFunctionSpec implements FunctionSpec, Serializable {
private static final Duration DEFAULT_HTTP_WRITE_TIMEOUT = Duration.ofSeconds(10);
private static final Integer DEFAULT_MAX_NUM_BATCH_REQUESTS = 1000;
- private final FunctionType functionType;
- private final URI endpoint;
- private final List<StateSpec> states;
+ private final Either<FunctionType, FunctionTypeNamespaceMatcher> target;
+ private final UrlPathTemplate urlPathTemplate;
+
private final Duration maxRequestDuration;
private final Duration connectTimeout;
private final Duration readTimeout;
private final Duration writeTimeout;
private final int maxNumBatchRequests;
- private HttpFunctionSpec(
- FunctionType functionType,
- URI endpoint,
- List<StateSpec> states,
+ public static Builder builder(
+ Either<FunctionType, FunctionTypeNamespaceMatcher> target, UrlPathTemplate urlPathTemplate) {
+ return new Builder(target, urlPathTemplate);
+ }
+
+ private HttpFunctionEndpointSpec(
+ Either<FunctionType, FunctionTypeNamespaceMatcher> target,
+ UrlPathTemplate urlPathTemplate,
Duration maxRequestDuration,
Duration connectTimeout,
Duration readTimeout,
Duration writeTimeout,
int maxNumBatchRequests) {
- this.functionType = Objects.requireNonNull(functionType);
- this.endpoint = Objects.requireNonNull(endpoint);
- this.states = Objects.requireNonNull(states);
- this.maxRequestDuration = Objects.requireNonNull(maxRequestDuration);
- this.connectTimeout = Objects.requireNonNull(connectTimeout);
- this.readTimeout = Objects.requireNonNull(readTimeout);
- this.writeTimeout = Objects.requireNonNull(writeTimeout);
+ this.target = target;
+ this.urlPathTemplate = urlPathTemplate;
+ this.maxRequestDuration = maxRequestDuration;
+ this.connectTimeout = connectTimeout;
+ this.readTimeout = readTimeout;
+ this.writeTimeout = writeTimeout;
this.maxNumBatchRequests = maxNumBatchRequests;
}
- public static Builder builder(FunctionType functionType, URI endpoint) {
- return new Builder(functionType, endpoint);
- }
-
@Override
- public FunctionType functionType() {
- return functionType;
+ public Either<FunctionType, FunctionTypeNamespaceMatcher> target() {
+ return target;
}
@Override
@@ -78,17 +73,9 @@ public final class HttpFunctionSpec implements FunctionSpec, Serializable {
return Kind.HTTP;
}
- public URI endpoint() {
- return endpoint;
- }
-
- public boolean isUnixDomainSocket() {
- String scheme = endpoint.getScheme();
- return "http+unix".equalsIgnoreCase(scheme) || "https+unix".equalsIgnoreCase(scheme);
- }
-
- public List<StateSpec> states() {
- return states;
+ @Override
+ public UrlPathTemplate urlPathTemplate() {
+ return urlPathTemplate;
}
public Duration maxRequestDuration() {
@@ -113,24 +100,20 @@ public final class HttpFunctionSpec implements FunctionSpec, Serializable {
public static final class Builder {
- private final FunctionType functionType;
- private final URI endpoint;
+ private final Either<FunctionType, FunctionTypeNamespaceMatcher> target;
+ private final UrlPathTemplate urlPathTemplate;
- private final List<StateSpec> states = new ArrayList<>();
private Duration maxRequestDuration = DEFAULT_HTTP_TIMEOUT;
private Duration connectTimeout = DEFAULT_HTTP_CONNECT_TIMEOUT;
private Duration readTimeout = DEFAULT_HTTP_READ_TIMEOUT;
private Duration writeTimeout = DEFAULT_HTTP_WRITE_TIMEOUT;
private int maxNumBatchRequests = DEFAULT_MAX_NUM_BATCH_REQUESTS;
- private Builder(FunctionType functionType, URI endpoint) {
- this.functionType = Objects.requireNonNull(functionType);
- this.endpoint = Objects.requireNonNull(endpoint);
- }
-
- public Builder withState(StateSpec stateSpec) {
- this.states.add(stateSpec);
- return this;
+ private Builder(
+ Either<FunctionType, FunctionTypeNamespaceMatcher> target,
+ UrlPathTemplate urlPathTemplate) {
+ this.target = Objects.requireNonNull(target);
+ this.urlPathTemplate = Objects.requireNonNull(urlPathTemplate);
}
public Builder withMaxRequestDuration(Duration duration) {
@@ -158,13 +141,12 @@ public final class HttpFunctionSpec implements FunctionSpec, Serializable {
return this;
}
- public HttpFunctionSpec build() {
+ public HttpFunctionEndpointSpec build() {
validateTimeouts();
- return new HttpFunctionSpec(
- functionType,
- endpoint,
- states,
+ return new HttpFunctionEndpointSpec(
+ target,
+ urlPathTemplate,
maxRequestDuration,
connectTimeout,
readTimeout,
diff --git a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/HttpFunctionProvider.java b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/HttpFunctionProvider.java
index e1ef27e..67c2c68 100644
--- a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/HttpFunctionProvider.java
+++ b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/HttpFunctionProvider.java
@@ -72,7 +72,7 @@ public class HttpFunctionProvider implements StatefulFunctionProvider, ManagingR
clientBuilder.writeTimeout(spec.writeTimeout());
final HttpUrl url;
- if (spec.isUnixDomainSocket()) {
+ if (UnixDomainHttpEndpoint.validate(spec.endpoint())) {
UnixDomainHttpEndpoint endpoint = UnixDomainHttpEndpoint.parseFrom(spec.endpoint());
url =
diff --git a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/HttpFunctionSpec.java b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/HttpFunctionSpec.java
index 3a8f653..42a7abb 100644
--- a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/HttpFunctionSpec.java
+++ b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/HttpFunctionSpec.java
@@ -82,11 +82,6 @@ public final class HttpFunctionSpec implements FunctionSpec, Serializable {
return endpoint;
}
- public boolean isUnixDomainSocket() {
- String scheme = endpoint.getScheme();
- return "http+unix".equalsIgnoreCase(scheme) || "https+unix".equalsIgnoreCase(scheme);
- }
-
public List<StateSpec> states() {
return states;
}
diff --git a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/HttpFunctionProvider.java b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/TemplatedHttpFunctionProvider.java
similarity index 59%
copy from statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/HttpFunctionProvider.java
copy to statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/TemplatedHttpFunctionProvider.java
index e1ef27e..ae4d7f4 100644
--- a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/HttpFunctionProvider.java
+++ b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/TemplatedHttpFunctionProvider.java
@@ -15,12 +15,14 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-
package org.apache.flink.statefun.flink.core.httpfn;
import static org.apache.flink.statefun.flink.core.httpfn.OkHttpUnixSocketBridge.configureUnixDomainSocket;
+import java.net.URI;
+import java.util.Collections;
import java.util.Map;
+import java.util.Objects;
import javax.annotation.Nullable;
import javax.annotation.concurrent.NotThreadSafe;
import okhttp3.HttpUrl;
@@ -30,38 +32,52 @@ import org.apache.flink.statefun.flink.core.reqreply.PersistedRemoteFunctionValu
import org.apache.flink.statefun.flink.core.reqreply.RequestReplyClient;
import org.apache.flink.statefun.flink.core.reqreply.RequestReplyFunction;
import org.apache.flink.statefun.sdk.FunctionType;
+import org.apache.flink.statefun.sdk.StatefulFunction;
import org.apache.flink.statefun.sdk.StatefulFunctionProvider;
@NotThreadSafe
-public class HttpFunctionProvider implements StatefulFunctionProvider, ManagingResources {
- private final Map<FunctionType, HttpFunctionSpec> supportedTypes;
+public final class TemplatedHttpFunctionProvider
+ implements StatefulFunctionProvider, ManagingResources {
+
+ private final Map<FunctionType, HttpFunctionEndpointSpec> specificTypeEndpointSpecs;
+ private final Map<String, HttpFunctionEndpointSpec> perNamespaceEndpointSpecs;
/** lazily initialized by {code buildHttpClient} */
@Nullable private OkHttpClient sharedClient;
private volatile boolean shutdown;
- public HttpFunctionProvider(Map<FunctionType, HttpFunctionSpec> supportedTypes) {
- this.supportedTypes = supportedTypes;
+ public TemplatedHttpFunctionProvider(
+ Map<FunctionType, HttpFunctionEndpointSpec> specificTypeEndpointSpecs,
+ Map<String, HttpFunctionEndpointSpec> perNamespaceEndpointSpecs) {
+ this.specificTypeEndpointSpecs = Objects.requireNonNull(specificTypeEndpointSpecs);
+ this.perNamespaceEndpointSpecs = Objects.requireNonNull(perNamespaceEndpointSpecs);
}
@Override
- public RequestReplyFunction functionOfType(FunctionType type) {
- HttpFunctionSpec spec = supportedTypes.get(type);
- if (spec == null) {
- throw new IllegalArgumentException("Unsupported type " + type);
- }
+ public StatefulFunction functionOfType(FunctionType functionType) {
+ final HttpFunctionEndpointSpec endpointsSpec = getEndpointsSpecOrThrow(functionType);
return new RequestReplyFunction(
- new PersistedRemoteFunctionValues(spec.states()),
- spec.maxNumBatchRequests(),
- buildHttpClient(spec));
+ new PersistedRemoteFunctionValues(Collections.emptyList()),
+ endpointsSpec.maxNumBatchRequests(),
+ buildHttpClient(endpointsSpec, functionType));
}
- public HttpFunctionSpec getFunctionSpec(FunctionType type) {
- return supportedTypes.get(type);
+ private HttpFunctionEndpointSpec getEndpointsSpecOrThrow(FunctionType functionType) {
+ HttpFunctionEndpointSpec endpointSpec = specificTypeEndpointSpecs.get(functionType);
+ if (endpointSpec != null) {
+ return endpointSpec;
+ }
+ endpointSpec = perNamespaceEndpointSpecs.get(functionType.namespace());
+ if (endpointSpec != null) {
+ return endpointSpec;
+ }
+
+ throw new IllegalStateException("Unknown type: " + functionType);
}
- private RequestReplyClient buildHttpClient(HttpFunctionSpec spec) {
+ private RequestReplyClient buildHttpClient(
+ HttpFunctionEndpointSpec spec, FunctionType functionType) {
if (sharedClient == null) {
sharedClient = OkHttpUtils.newClient();
}
@@ -71,9 +87,11 @@ public class HttpFunctionProvider implements StatefulFunctionProvider, ManagingR
clientBuilder.readTimeout(spec.readTimeout());
clientBuilder.writeTimeout(spec.writeTimeout());
+ URI endpointUrl = spec.urlPathTemplate().apply(functionType);
+
final HttpUrl url;
- if (spec.isUnixDomainSocket()) {
- UnixDomainHttpEndpoint endpoint = UnixDomainHttpEndpoint.parseFrom(spec.endpoint());
+ if (UnixDomainHttpEndpoint.validate(endpointUrl)) {
+ UnixDomainHttpEndpoint endpoint = UnixDomainHttpEndpoint.parseFrom(endpointUrl);
url =
new HttpUrl.Builder()
@@ -84,7 +102,7 @@ public class HttpFunctionProvider implements StatefulFunctionProvider, ManagingR
configureUnixDomainSocket(clientBuilder, endpoint.unixDomainFile);
} else {
- url = HttpUrl.get(spec.endpoint());
+ url = HttpUrl.get(endpointUrl);
}
return new HttpRequestReplyClient(url, clientBuilder.build(), () -> shutdown);
}
diff --git a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/UnixDomainHttpEndpoint.java b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/UnixDomainHttpEndpoint.java
index c628fdb..01c49d5 100644
--- a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/UnixDomainHttpEndpoint.java
+++ b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/UnixDomainHttpEndpoint.java
@@ -23,12 +23,20 @@ import java.net.URI;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.Objects;
+import org.apache.flink.util.Preconditions;
/** Represents a Unix domain file path and an http endpoint */
final class UnixDomainHttpEndpoint {
+ /** Checks whether or not an endpoint is using UNIX domain sockets. */
+ static boolean validate(URI endpoint) {
+ String scheme = endpoint.getScheme();
+ return "http+unix".equalsIgnoreCase(scheme) || "https+unix".equalsIgnoreCase(scheme);
+ }
+
/** Parses a URI of the form {@code http+unix://<file system path>.sock/<http endpoint>}. */
static UnixDomainHttpEndpoint parseFrom(URI endpoint) {
+ Preconditions.checkArgument(validate(endpoint));
final Path path = Paths.get(endpoint.getPath());
final int sockPathIndex = indexOfSockFile(path);
final String filePath = "/" + path.subpath(0, sockPathIndex + 1).toString();
diff --git a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/jsonmodule/FormatVersion.java b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/jsonmodule/FormatVersion.java
index debb28c..87f4ec4 100644
--- a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/jsonmodule/FormatVersion.java
+++ b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/jsonmodule/FormatVersion.java
@@ -20,7 +20,8 @@ package org.apache.flink.statefun.flink.core.jsonmodule;
enum FormatVersion {
v1_0("1.0"),
- v2_0("2.0");
+ v2_0("2.0"),
+ v3_0("3.0");
private String versionStr;
@@ -39,6 +40,8 @@ enum FormatVersion {
return v1_0;
case "2.0":
return v2_0;
+ case "3.0":
+ return v3_0;
default:
throw new IllegalArgumentException("Unrecognized format version: " + versionStr);
}
diff --git a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/jsonmodule/FunctionEndpointJsonEntity.java b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/jsonmodule/FunctionEndpointJsonEntity.java
new file mode 100644
index 0000000..fac1bbe
--- /dev/null
+++ b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/jsonmodule/FunctionEndpointJsonEntity.java
@@ -0,0 +1,206 @@
+/*
+ * 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 static java.util.stream.Collectors.groupingBy;
+import static java.util.stream.Collectors.toList;
+
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Optional;
+import java.util.OptionalInt;
+import java.util.stream.StreamSupport;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonPointer;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
+import org.apache.flink.statefun.flink.common.json.Selectors;
+import org.apache.flink.statefun.flink.core.httpfn.HttpFunctionEndpointSpec;
+import org.apache.flink.statefun.flink.core.httpfn.TemplatedHttpFunctionProvider;
+import org.apache.flink.statefun.sdk.FunctionType;
+import org.apache.flink.statefun.sdk.FunctionTypeNamespaceMatcher;
+import org.apache.flink.statefun.sdk.StatefulFunctionProvider;
+import org.apache.flink.statefun.sdk.spi.StatefulFunctionModule;
+import org.apache.flink.types.Either;
+import org.apache.flink.util.TimeUtils;
+
+public final class FunctionEndpointJsonEntity implements JsonEntity {
+
+ private static final JsonPointer FUNCTION_ENDPOINTS_POINTER = JsonPointer.compile("/endpoints");
+
+ private static final class MetaPointers {
+ private static final JsonPointer KIND = JsonPointer.compile("/endpoint/meta/kind");
+ }
+
+ private static final class SpecPointers {
+ private static final JsonPointer TYPENAME = JsonPointer.compile("/endpoint/spec/typename");
+ private static final JsonPointer URL_PATH_TEMPLATE =
+ JsonPointer.compile("/endpoint/spec/urlPathTemplate");
+ private static final JsonPointer TIMEOUTS = JsonPointer.compile("/endpoint/spec/timeouts");
+ private static final JsonPointer MAX_NUM_BATCH_REQUESTS =
+ JsonPointer.compile("/endpoint/spec/maxNumBatchRequests");
+ }
+
+ private static final class TypenamePointers {
+ private static final JsonPointer NAMESPACE = JsonPointer.compile("/namespace");
+ private static final JsonPointer FUNCTION_NAME = JsonPointer.compile("/type");
+ }
+
+ private static final class TimeoutPointers {
+ private static final JsonPointer CALL = JsonPointer.compile("/call");
+ private static final JsonPointer CONNECT = JsonPointer.compile("/connect");
+ private static final JsonPointer READ = JsonPointer.compile("/read");
+ private static final JsonPointer WRITE = JsonPointer.compile("/write");
+ }
+
+ @Override
+ public void bind(
+ StatefulFunctionModule.Binder binder, JsonNode moduleSpecNode, FormatVersion formatVersion) {
+ if (formatVersion != FormatVersion.v3_0) {
+ throw new IllegalArgumentException("endpoints is only supported with format version 3.0.");
+ }
+
+ final Iterable<? extends JsonNode> functionEndpointsSpecNodes =
+ functionEndpointSpecNodes(moduleSpecNode);
+
+ for (Map.Entry<FunctionEndpointSpec.Kind, List<FunctionEndpointSpec>> entry :
+ parseFunctionEndpointSpecs(functionEndpointsSpecNodes).entrySet()) {
+ final Map<FunctionType, FunctionEndpointSpec> specificTypeEndpointSpecs = new HashMap<>();
+ final Map<String, FunctionEndpointSpec> perNamespaceEndpointSpecs = new HashMap<>();
+
+ entry
+ .getValue()
+ .forEach(
+ spec -> {
+ Either<FunctionType, FunctionTypeNamespaceMatcher> target = spec.target();
+ if (target.isLeft()) {
+ specificTypeEndpointSpecs.put(target.left(), spec);
+ } else {
+ perNamespaceEndpointSpecs.put(target.right().targetNamespace(), spec);
+ }
+ });
+
+ StatefulFunctionProvider provider =
+ functionProvider(entry.getKey(), specificTypeEndpointSpecs, perNamespaceEndpointSpecs);
+ specificTypeEndpointSpecs
+ .keySet()
+ .forEach(specificType -> binder.bindFunctionProvider(specificType, provider));
+ perNamespaceEndpointSpecs
+ .keySet()
+ .forEach(
+ namespace ->
+ binder.bindFunctionProvider(
+ FunctionTypeNamespaceMatcher.targetNamespace(namespace), provider));
+ }
+ }
+
+ private static Iterable<? extends JsonNode> functionEndpointSpecNodes(
+ JsonNode moduleSpecRootNode) {
+ return Selectors.listAt(moduleSpecRootNode, FUNCTION_ENDPOINTS_POINTER);
+ }
+
+ private static Map<FunctionEndpointSpec.Kind, List<FunctionEndpointSpec>>
+ parseFunctionEndpointSpecs(Iterable<? extends JsonNode> functionEndpointsSpecNodes) {
+ return StreamSupport.stream(functionEndpointsSpecNodes.spliterator(), false)
+ .map(FunctionEndpointJsonEntity::parseFunctionEndpointsSpec)
+ .collect(groupingBy(FunctionEndpointSpec::kind, toList()));
+ }
+
+ private static FunctionEndpointSpec parseFunctionEndpointsSpec(
+ JsonNode functionEndpointSpecNode) {
+ FunctionEndpointSpec.Kind kind = endpointKind(functionEndpointSpecNode);
+
+ switch (kind) {
+ case HTTP:
+ final HttpFunctionEndpointSpec.Builder specBuilder =
+ HttpFunctionEndpointSpec.builder(
+ target(functionEndpointSpecNode), urlPathTemplate(functionEndpointSpecNode));
+
+ JsonNode timeoutsNode = functionEndpointSpecNode.at(SpecPointers.TIMEOUTS);
+ optionalMaxNumBatchRequests(functionEndpointSpecNode)
+ .ifPresent(specBuilder::withMaxNumBatchRequests);
+ optionalTimeoutDuration(timeoutsNode, TimeoutPointers.CALL)
+ .ifPresent(specBuilder::withMaxRequestDuration);
+ optionalTimeoutDuration(timeoutsNode, TimeoutPointers.CONNECT)
+ .ifPresent(specBuilder::withConnectTimeoutDuration);
+ optionalTimeoutDuration(timeoutsNode, TimeoutPointers.READ)
+ .ifPresent(specBuilder::withReadTimeoutDuration);
+ optionalTimeoutDuration(timeoutsNode, TimeoutPointers.WRITE)
+ .ifPresent(specBuilder::withWriteTimeoutDuration);
+
+ return specBuilder.build();
+ case GRPC:
+ throw new UnsupportedOperationException("GRPC endpoints are not supported yet.");
+ default:
+ throw new IllegalArgumentException("Unrecognized function endpoint kind " + kind);
+ }
+ }
+
+ private static FunctionEndpointSpec.Kind endpointKind(JsonNode functionEndpointSpecNode) {
+ String endpointKind = Selectors.textAt(functionEndpointSpecNode, MetaPointers.KIND);
+ return FunctionEndpointSpec.Kind.valueOf(endpointKind.toUpperCase(Locale.getDefault()));
+ }
+
+ private static Either<FunctionType, FunctionTypeNamespaceMatcher> target(
+ JsonNode functionEndpointSpecNode) {
+ JsonNode targetNode = functionEndpointSpecNode.at(SpecPointers.TYPENAME);
+ String namespace = Selectors.textAt(targetNode, TypenamePointers.NAMESPACE);
+ Optional<String> functionName =
+ Selectors.optionalTextAt(targetNode, TypenamePointers.FUNCTION_NAME);
+ return (functionName.isPresent())
+ ? Either.Left(new FunctionType(namespace, functionName.get()))
+ : Either.Right(FunctionTypeNamespaceMatcher.targetNamespace(namespace));
+ }
+
+ private static FunctionEndpointSpec.UrlPathTemplate urlPathTemplate(
+ JsonNode functionEndpointSpecNode) {
+ String template = Selectors.textAt(functionEndpointSpecNode, SpecPointers.URL_PATH_TEMPLATE);
+ return new FunctionEndpointSpec.UrlPathTemplate(template);
+ }
+
+ private static OptionalInt optionalMaxNumBatchRequests(JsonNode functionNode) {
+ return Selectors.optionalIntegerAt(functionNode, SpecPointers.MAX_NUM_BATCH_REQUESTS);
+ }
+
+ private static Optional<Duration> optionalTimeoutDuration(
+ JsonNode node, JsonPointer timeoutPointer) {
+ return Selectors.optionalTextAt(node, timeoutPointer).map(TimeUtils::parseDuration);
+ }
+
+ private static StatefulFunctionProvider functionProvider(
+ FunctionEndpointSpec.Kind kind,
+ Map<FunctionType, FunctionEndpointSpec> specificTypeEndpointSpecs,
+ Map<String, FunctionEndpointSpec> perNamespaceEndpointSpecs) {
+ switch (kind) {
+ case HTTP:
+ return new TemplatedHttpFunctionProvider(
+ castValues(specificTypeEndpointSpecs), castValues(perNamespaceEndpointSpecs));
+ case GRPC:
+ throw new UnsupportedOperationException("GRPC endpoints are not supported yet.");
+ default:
+ throw new IllegalStateException("Unexpected kind: " + kind);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private static <K, NV extends FunctionEndpointSpec> Map<K, NV> castValues(
+ Map<K, FunctionEndpointSpec> toCast) {
+ return new HashMap(toCast);
+ }
+}
diff --git a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/jsonmodule/FormatVersion.java b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/jsonmodule/FunctionEndpointSpec.java
similarity index 53%
copy from statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/jsonmodule/FormatVersion.java
copy to statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/jsonmodule/FunctionEndpointSpec.java
index debb28c..940db87 100644
--- a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/jsonmodule/FormatVersion.java
+++ b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/jsonmodule/FunctionEndpointSpec.java
@@ -15,32 +15,38 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-
package org.apache.flink.statefun.flink.core.jsonmodule;
-enum FormatVersion {
- v1_0("1.0"),
- v2_0("2.0");
+import java.net.URI;
+import java.util.Objects;
+import org.apache.flink.statefun.sdk.FunctionType;
+import org.apache.flink.statefun.sdk.FunctionTypeNamespaceMatcher;
+import org.apache.flink.types.Either;
- private String versionStr;
+public interface FunctionEndpointSpec {
- FormatVersion(String versionStr) {
- this.versionStr = versionStr;
- }
+ Either<FunctionType, FunctionTypeNamespaceMatcher> target();
+
+ Kind kind();
+
+ UrlPathTemplate urlPathTemplate();
- @Override
- public String toString() {
- return versionStr;
+ enum Kind {
+ HTTP,
+ GRPC
}
- static FormatVersion fromString(String versionStr) {
- switch (versionStr) {
- case "1.0":
- return v1_0;
- case "2.0":
- return v2_0;
- default:
- throw new IllegalArgumentException("Unrecognized format version: " + versionStr);
+ class UrlPathTemplate {
+ private static final String FUNCTION_NAME_HOLDER = "{typename.function}";
+
+ private final String template;
+
+ public UrlPathTemplate(String template) {
+ this.template = Objects.requireNonNull(template);
+ }
+
+ public URI apply(FunctionType functionType) {
+ return URI.create(template.replace(FUNCTION_NAME_HOLDER, functionType.name()));
}
}
}
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 cedeb97..658bb9b 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
@@ -37,6 +37,13 @@ final class JsonModule implements StatefulFunctionModule {
new RouterJsonEntity(),
new EgressJsonEntity());
+ private static final List<JsonEntity> V3_ENTITIES =
+ Arrays.asList(
+ new FunctionEndpointJsonEntity(),
+ new IngressJsonEntity(),
+ new RouterJsonEntity(),
+ new EgressJsonEntity());
+
private final JsonNode moduleSpecNode;
private final FormatVersion formatVersion;
private final URL moduleUrl;
@@ -49,7 +56,11 @@ final class JsonModule implements StatefulFunctionModule {
public void configure(Map<String, String> conf, Binder binder) {
try {
- ENTITIES.forEach(jsonEntity -> jsonEntity.bind(binder, moduleSpecNode, formatVersion));
+ if (formatVersion == FormatVersion.v3_0) {
+ V3_ENTITIES.forEach(jsonEntity -> jsonEntity.bind(binder, moduleSpecNode, formatVersion));
+ } else {
+ ENTITIES.forEach(jsonEntity -> jsonEntity.bind(binder, moduleSpecNode, formatVersion));
+ }
} catch (Throwable t) {
throw new ModuleConfigurationException(
format("Error while parsing module at %s", moduleUrl), t);
diff --git a/statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/httpfn/UnixDomainHttpEndpointTest.java b/statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/httpfn/UnixDomainHttpEndpointTest.java
index f96ac3e..127775a 100644
--- a/statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/httpfn/UnixDomainHttpEndpointTest.java
+++ b/statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/httpfn/UnixDomainHttpEndpointTest.java
@@ -1,6 +1,7 @@
package org.apache.flink.statefun.flink.core.httpfn;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
import java.net.URI;
import org.junit.Test;
@@ -29,4 +30,14 @@ public class UnixDomainHttpEndpointTest {
public void missingSockFile() {
UnixDomainHttpEndpoint.parseFrom(URI.create("http+unix:///some/path/hello"));
}
+
+ @Test
+ public void validateUdsEndpoint() {
+ assertFalse(UnixDomainHttpEndpoint.validate(URI.create("http:///bar.foo.com/some/path")));
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void parseNonUdsEndpoint() {
+ UnixDomainHttpEndpoint.parseFrom(URI.create("http:///bar.foo.com/some/path"));
+ }
}
diff --git a/statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/jsonmodule/JsonModuleV3Test.java b/statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/jsonmodule/JsonModuleV3Test.java
new file mode 100644
index 0000000..fcbb885
--- /dev/null
+++ b/statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/jsonmodule/JsonModuleV3Test.java
@@ -0,0 +1,114 @@
+/*
+ * 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 static org.hamcrest.Matchers.allOf;
+import static org.hamcrest.Matchers.hasKey;
+import static org.hamcrest.Matchers.not;
+import static org.hamcrest.Matchers.notNullValue;
+import static org.hamcrest.Matchers.nullValue;
+import static org.junit.Assert.assertThat;
+
+import com.google.protobuf.Any;
+import com.google.protobuf.Message;
+import java.net.URL;
+import java.util.Collections;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.flink.statefun.flink.core.StatefulFunctionsUniverse;
+import org.apache.flink.statefun.flink.core.message.MessageFactoryKey;
+import org.apache.flink.statefun.flink.core.message.MessageFactoryType;
+import org.apache.flink.statefun.sdk.FunctionType;
+import org.apache.flink.statefun.sdk.io.EgressIdentifier;
+import org.apache.flink.statefun.sdk.io.IngressIdentifier;
+import org.apache.flink.statefun.sdk.spi.StatefulFunctionModule;
+import org.junit.Test;
+
+public class JsonModuleV3Test {
+
+ private static final String modulePath = "module-v3_0/module.yaml";
+
+ @Test
+ public void exampleUsage() {
+ StatefulFunctionModule module = fromPath(modulePath);
+
+ assertThat(module, notNullValue());
+ }
+
+ @Test
+ public void testFunctions() {
+ StatefulFunctionModule module = fromPath(modulePath);
+
+ StatefulFunctionsUniverse universe = emptyUniverse();
+ module.configure(Collections.emptyMap(), universe);
+
+ assertThat(
+ universe.functions(),
+ allOf(
+ hasKey(new FunctionType("com.foo.bar", "specific_function")),
+ hasKey(new FunctionType("com.other.namespace", "hello"))));
+
+ assertThat(universe.namespaceFunctions(), hasKey("com.foo.bar"));
+ }
+
+ @Test
+ public void testRouters() {
+ StatefulFunctionModule module = fromPath(modulePath);
+
+ StatefulFunctionsUniverse universe = emptyUniverse();
+ module.configure(Collections.emptyMap(), universe);
+
+ assertThat(
+ universe.routers(),
+ hasKey(new IngressIdentifier<>(Message.class, "com.mycomp.igal", "names")));
+ }
+
+ @Test
+ public void testIngresses() {
+ StatefulFunctionModule module = fromPath(modulePath);
+
+ StatefulFunctionsUniverse universe = emptyUniverse();
+ module.configure(Collections.emptyMap(), universe);
+
+ assertThat(
+ universe.ingress(),
+ hasKey(new IngressIdentifier<>(Message.class, "com.mycomp.igal", "names")));
+ }
+
+ @Test
+ public void testEgresses() {
+ StatefulFunctionModule module = fromPath(modulePath);
+
+ StatefulFunctionsUniverse universe = emptyUniverse();
+ module.configure(Collections.emptyMap(), universe);
+
+ assertThat(
+ universe.egress(), hasKey(new EgressIdentifier<>("com.mycomp.foo", "bar", Any.class)));
+ }
+
+ private static StatefulFunctionModule fromPath(String path) {
+ URL moduleUrl = JsonModuleTest.class.getClassLoader().getResource(path);
+ assertThat(moduleUrl, not(nullValue()));
+ ObjectMapper mapper = JsonServiceLoader.mapper();
+ return JsonServiceLoader.fromUrl(mapper, moduleUrl);
+ }
+
+ private static StatefulFunctionsUniverse emptyUniverse() {
+ return new StatefulFunctionsUniverse(
+ MessageFactoryKey.forType(MessageFactoryType.WITH_PROTOBUF_PAYLOADS, null));
+ }
+}
diff --git a/statefun-flink/statefun-flink-core/src/test/resources/module-v3_0/module.yaml b/statefun-flink/statefun-flink-core/src/test/resources/module-v3_0/module.yaml
new file mode 100644
index 0000000..f7d3d7c
--- /dev/null
+++ b/statefun-flink/statefun-flink-core/src/test/resources/module-v3_0/module.yaml
@@ -0,0 +1,85 @@
+# 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.
+
+version: "3.0"
+
+module:
+ meta:
+ type: remote
+ spec:
+ endpoints:
+ - endpoint:
+ meta:
+ kind: http
+ spec:
+ typename:
+ namespace: com.foo.bar
+ urlPathTemplate: http://bar.foo.com:8080/functions/{typename.function}
+ timeouts:
+ call: 1minutes
+ connect: 10seconds
+ read: 10second
+ write: 10seconds
+ maxNumBatchRequests: 10000
+ - endpoint:
+ meta:
+ kind: http
+ spec:
+ typename:
+ namespace: com.foo.bar
+ type: specific_function
+ urlPathTemplate: http://bar.foo.com:8080/functions/abc
+ - endpoint:
+ meta:
+ kind: http
+ spec:
+ typename:
+ namespace: com.other.namespace
+ type: hello
+ urlPathTemplate: http://namespace.other.com:8080/hello
+ routers:
+ - router:
+ meta:
+ type: org.apache.flink.statefun.sdk/protobuf-router
+ spec:
+ ingress: com.mycomp.igal/names
+ target: "com.example/hello/{{$.name}}"
+ messageType: org.apache.flink.test.SimpleMessage
+ descriptorSet: classpath:test.desc
+ ingresses:
+ - ingress:
+ meta:
+ type: statefun.kafka.io/protobuf-ingress
+ id: com.mycomp.igal/names
+ spec:
+ address: kafka-broker:9092
+ topics:
+ - names
+ properties:
+ - consumer.group: greeter
+ messageType: org.apache.flink.test.SimpleMessage
+ descriptorSet: classpath:test.desc
+ egresses:
+ - egress:
+ meta:
+ type: statefun.kafka.io/generic-egress
+ id: com.mycomp.foo/bar
+ spec:
+ address: kafka-broker:9092
+ deliverySemantic:
+ type: exactly-once
+ transactionTimeoutMillis: 100000
+ properties:
+ - foo.config: bar
\ No newline at end of file