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/05/21 06:27:54 UTC
[flink-statefun] branch master updated (23b4d73 -> 1c018e1)
This is an automated email from the ASF dual-hosted git repository.
tzulitai pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink-statefun.git.
from 23b4d73 [hotfix] Fix spotless checks
new 95e8757 [FLINK-17611] [core] Support UNIX domain sockets for remote function endpoints
new 9c43dc9 [FLINK-17611] [core] Bridge OkHttp and UNIX domain sockets
new 1c018e1 [FLINK-17611] [core] Add an ITCase for Unix domain sockets
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
pom.xml | 1 +
.../statefun-e2e-tests-common/pom.xml | 10 ++
statefun-flink/statefun-flink-core/pom.xml | 17 ++-
.../flink/core/httpfn/HttpFunctionProvider.java | 33 ++++-
.../flink/core/httpfn/HttpFunctionSpec.java | 5 +
.../flink/core/httpfn/OkHttpUnixSocketBridge.java | 153 +++++++++++++++++++++
.../flink/core/httpfn/UnixDomainHttpEndpoint.java | 60 ++++++++
.../statefun/flink/core/jsonmodule/JsonModule.java | 7 +-
.../core/httpfn/UnixDomainHttpEndpointTest.java | 32 +++++
.../flink/core/httpfn/UnixDomainSocketITCase.java | 99 +++++++++++++
.../flink/core/jsonmodule/JsonModuleTest.java | 4 +-
.../src/test/resources/bar-module/module.yaml | 9 ++
.../src/main/resources/META-INF/NOTICE | 2 +
13 files changed, 419 insertions(+), 13 deletions(-)
create mode 100644 statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/OkHttpUnixSocketBridge.java
create mode 100644 statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/UnixDomainHttpEndpoint.java
create mode 100644 statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/httpfn/UnixDomainHttpEndpointTest.java
create mode 100644 statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/httpfn/UnixDomainSocketITCase.java
[flink-statefun] 01/03: [FLINK-17611] [core] Support UNIX domain
sockets for remote function endpoints
Posted by tz...@apache.org.
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 95e8757cb706d703ddde073b49dec7325a679400
Author: slinkydeveloper <fr...@gmail.com>
AuthorDate: Thu May 14 14:58:51 2020 +0200
[FLINK-17611] [core] Support UNIX domain sockets for remote function endpoints
---
pom.xml | 6 +++
.../statefun-e2e-tests-common/pom.xml | 10 ++++
statefun-flink/statefun-flink-core/pom.xml | 11 ++--
.../flink/core/httpfn/HttpFunctionProvider.java | 62 ++++++++++++++++++++--
.../flink/core/httpfn/HttpFunctionSpec.java | 5 ++
.../statefun/flink/core/jsonmodule/JsonModule.java | 7 ++-
.../core/httpfn/HttpFunctionProviderTest.java | 35 ++++++++++++
.../flink/core/jsonmodule/JsonModuleTest.java | 4 +-
.../src/test/resources/bar-module/module.yaml | 9 ++++
.../src/main/resources/META-INF/NOTICE | 2 +
10 files changed, 142 insertions(+), 9 deletions(-)
diff --git a/pom.xml b/pom.xml
index 93483b4..6f40a25 100644
--- a/pom.xml
+++ b/pom.xml
@@ -74,6 +74,7 @@ under the License.
<spotless-maven-plugin.version>1.20.0</spotless-maven-plugin.version>
<auto-service.version>1.0-rc6</auto-service.version>
<protobuf.version>3.7.1</protobuf.version>
+ <unixsocket.version>2.3.2</unixsocket.version>
<protoc-jar-maven-plugin.version>3.11.1</protoc-jar-maven-plugin.version>
<flink.version>1.10.1</flink.version>
</properties>
@@ -170,6 +171,11 @@ under the License.
</configuration>
</execution>
</executions>
+ <configuration>
+ <includeDirectories>
+ <include>/usr/include/</include>
+ </includeDirectories>
+ </configuration>
</plugin>
</plugins>
</pluginManagement>
diff --git a/statefun-e2e-tests/statefun-e2e-tests-common/pom.xml b/statefun-e2e-tests/statefun-e2e-tests-common/pom.xml
index 9ce09ec..3989414 100644
--- a/statefun-e2e-tests/statefun-e2e-tests-common/pom.xml
+++ b/statefun-e2e-tests/statefun-e2e-tests-common/pom.xml
@@ -63,9 +63,19 @@ under the License.
<groupId>org.apache.commons</groupId>
<artifactId>commons-compress</artifactId>
</exclusion>
+ <exclusion>
+ <groupId>com.kohlschutter.junixsocket</groupId>
+ <artifactId>junixsocket-common</artifactId>
+ </exclusion>
</exclusions>
</dependency>
+ <dependency>
+ <groupId>com.kohlschutter.junixsocket</groupId>
+ <artifactId>junixsocket-common</artifactId>
+ <version>${unixsocket.version}</version>
+ </dependency>
+
<!-- Flink Config -->
<dependency>
<groupId>org.apache.flink</groupId>
diff --git a/statefun-flink/statefun-flink-core/pom.xml b/statefun-flink/statefun-flink-core/pom.xml
index ff7ffb0..e99077e 100644
--- a/statefun-flink/statefun-flink-core/pom.xml
+++ b/statefun-flink/statefun-flink-core/pom.xml
@@ -19,7 +19,7 @@ under the License.
xmlns="http://maven.apache.org/POM/4.0.0"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
-
+
<parent>
<groupId>org.apache.flink</groupId>
<artifactId>statefun-flink</artifactId>
@@ -28,11 +28,11 @@ under the License.
</parent>
<artifactId>statefun-flink-core</artifactId>
-
+
<properties>
<okhttp.version>3.14.6</okhttp.version>
</properties>
-
+
<dependencies>
<!-- sdk -->
@@ -68,6 +68,11 @@ under the License.
<artifactId>okhttp</artifactId>
<version>${okhttp.version}</version>
</dependency>
+ <dependency>
+ <groupId>com.kohlschutter.junixsocket</groupId>
+ <artifactId>junixsocket-core</artifactId>
+ <version>${unixsocket.version}</version>
+ </dependency>
<!-- tests -->
<dependency>
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 24a9114..e0c9cb5 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
@@ -18,13 +18,22 @@
package org.apache.flink.statefun.flink.core.httpfn;
+import java.net.URI;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.AbstractMap;
+import java.util.Collections;
import java.util.Map;
+import java.util.stream.IntStream;
import okhttp3.HttpUrl;
import okhttp3.OkHttpClient;
+import okhttp3.Protocol;
+import org.apache.flink.annotation.VisibleForTesting;
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.StatefulFunctionProvider;
+import org.newsclub.net.unix.AFUNIXSocketFactory;
public class HttpFunctionProvider implements StatefulFunctionProvider {
private final Map<FunctionType, HttpFunctionSpec> supportedTypes;
@@ -41,12 +50,59 @@ public class HttpFunctionProvider implements StatefulFunctionProvider {
if (spec == null) {
throw new IllegalArgumentException("Unsupported type " + type);
}
+ return new RequestReplyFunction(
+ spec.states(), spec.maxNumBatchRequests(), buildHttpClient(spec));
+ }
+
+ private RequestReplyClient buildHttpClient(HttpFunctionSpec spec) {
+ // We need to build a UDS HTTP client
+ if (spec.isUnixDomainSocket()) {
+
+ // We need to split the path in order to get the sock file and the path after the sock file
+ Map.Entry<String, String> splittedFilePathAndEndpoint =
+ splitFilePathAndEndpointForUDS(spec.endpoint());
+
+ OkHttpClient specificClient =
+ sharedClient
+ .newBuilder()
+ .socketFactory(
+ new AFUNIXSocketFactory.FactoryArg(splittedFilePathAndEndpoint.getKey()))
+ // Enable HTTP/2 if available (uses H2 upgrade),
+ // otherwise fallback to HTTP/1.1
+ .protocols(Collections.singletonList(Protocol.HTTP_2))
+ .callTimeout(spec.maxRequestDuration())
+ .build();
+
+ return new HttpRequestReplyClient(
+ // Only the path matters!
+ HttpUrl.get(URI.create(splittedFilePathAndEndpoint.getValue())), specificClient);
+ }
// specific client reuses the same the connection pool and thread pool
// as the sharedClient.
OkHttpClient specificClient =
sharedClient.newBuilder().callTimeout(spec.maxRequestDuration()).build();
- RequestReplyClient httpClient =
- new HttpRequestReplyClient(HttpUrl.get(spec.endpoint()), specificClient);
- return new RequestReplyFunction(spec.states(), spec.maxNumBatchRequests(), httpClient);
+ return new HttpRequestReplyClient(HttpUrl.get(spec.endpoint()), specificClient);
+ }
+
+ @VisibleForTesting
+ static Map.Entry<String, String> splitFilePathAndEndpointForUDS(URI input) {
+ // We need to split the path in order to get the sock file and the path after the sock file
+ Path path = Paths.get(input.getPath());
+
+ int sockPath =
+ IntStream.rangeClosed(0, path.getNameCount() - 1)
+ .filter(i -> path.getName(i).toString().endsWith(".sock"))
+ .findFirst()
+ .orElseThrow(
+ () ->
+ new IllegalStateException(
+ "Unix Domain Socket path should contain a .sock file"));
+
+ String filePath = "/" + path.subpath(0, sockPath + 1).toString();
+ String endpoint = "/";
+ if (sockPath != path.getNameCount() - 1) {
+ endpoint = "/" + path.subpath(sockPath + 1, path.getNameCount()).toString();
+ }
+ return new AbstractMap.SimpleImmutableEntry<>(filePath, endpoint);
}
}
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 0f38760..61945f4 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
@@ -58,6 +58,11 @@ public final class HttpFunctionSpec implements FunctionSpec {
return endpoint;
}
+ public boolean isUnixDomainSocket() {
+ String scheme = endpoint.getScheme();
+ return "http+unix".equalsIgnoreCase(scheme) || "https+unix".equalsIgnoreCase(scheme);
+ }
+
public List<String> states() {
return states;
}
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 0d167c1..bd96211 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
@@ -302,13 +302,16 @@ final class JsonModule implements StatefulFunctionModule {
+ uri
+ "; an http or https scheme must be provided.");
}
- if (scheme.equalsIgnoreCase("http") || scheme.equalsIgnoreCase("https")) {
+ if (scheme.equalsIgnoreCase("http")
+ || scheme.equalsIgnoreCase("https")
+ || scheme.equalsIgnoreCase("http+unix")
+ || scheme.equalsIgnoreCase("https+unix")) {
return typedUri;
}
throw new IllegalArgumentException(
"Missing scheme in function endpoint "
+ uri
- + "; an http or https scheme must be provided.");
+ + "; an http or https or http+unix or https+unix scheme must be provided.");
}
private static Collector<FunctionSpec, ?, Map<FunctionType, FunctionSpec>> groupByFunctionType() {
diff --git a/statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/httpfn/HttpFunctionProviderTest.java b/statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/httpfn/HttpFunctionProviderTest.java
new file mode 100644
index 0000000..21bbccc
--- /dev/null
+++ b/statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/httpfn/HttpFunctionProviderTest.java
@@ -0,0 +1,35 @@
+package org.apache.flink.statefun.flink.core.httpfn;
+
+import static org.junit.Assert.assertEquals;
+
+import java.net.URI;
+import java.util.Map;
+import org.junit.Test;
+
+public class HttpFunctionProviderTest {
+
+ @Test
+ public void splitOnlyWithFile() {
+ Map.Entry<String, String> out =
+ HttpFunctionProvider.splitFilePathAndEndpointForUDS(
+ URI.create("http+unix:///some/path.sock"));
+
+ assertEquals("/some/path.sock", out.getKey());
+ assertEquals("/", out.getValue());
+ }
+
+ @Test
+ public void splitOnlyWithFileAndEndpoint() {
+ Map.Entry<String, String> out =
+ HttpFunctionProvider.splitFilePathAndEndpointForUDS(
+ URI.create("http+unix:///some/path.sock/hello"));
+
+ assertEquals("/some/path.sock", out.getKey());
+ assertEquals("/hello", out.getValue());
+ }
+
+ @Test(expected = IllegalStateException.class)
+ public void missingSockFile() {
+ HttpFunctionProvider.splitFilePathAndEndpointForUDS(URI.create("http+unix:///some/path/hello"));
+ }
+}
diff --git a/statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/jsonmodule/JsonModuleTest.java b/statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/jsonmodule/JsonModuleTest.java
index a261ea4..f1f3aa1 100644
--- a/statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/jsonmodule/JsonModuleTest.java
+++ b/statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/jsonmodule/JsonModuleTest.java
@@ -55,7 +55,8 @@ public class JsonModuleTest {
universe.functions(),
allOf(
hasKey(new FunctionType("com.example", "hello")),
- hasKey(new FunctionType("com.foo", "world"))));
+ hasKey(new FunctionType("com.foo", "world")),
+ hasKey(new FunctionType("com.bar", "world"))));
}
@Test
@@ -95,6 +96,7 @@ public class JsonModuleTest {
private static StatefulFunctionModule fromPath(String path) {
URL moduleUrl = JsonModuleTest.class.getClassLoader().getResource(path);
+ assertThat(moduleUrl, not(nullValue()));
ObjectMapper mapper = JsonServiceLoader.mapper();
final JsonNode json;
try {
diff --git a/statefun-flink/statefun-flink-core/src/test/resources/bar-module/module.yaml b/statefun-flink/statefun-flink-core/src/test/resources/bar-module/module.yaml
index 61aa52f..0afafc6 100644
--- a/statefun-flink/statefun-flink-core/src/test/resources/bar-module/module.yaml
+++ b/statefun-flink/statefun-flink-core/src/test/resources/bar-module/module.yaml
@@ -36,6 +36,15 @@ module:
states:
- seen_count
maxNumBatchRequests: 10000
+ - function:
+ meta:
+ kind: http
+ type: com.bar/world
+ spec:
+ endpoint: http+unix:///hello/world.sock/statefun
+ states:
+ - seen_count
+ maxNumBatchRequests: 10000
routers:
- router:
meta:
diff --git a/statefun-flink/statefun-flink-distribution/src/main/resources/META-INF/NOTICE b/statefun-flink/statefun-flink-distribution/src/main/resources/META-INF/NOTICE
index 79b470d..1cd43cc 100644
--- a/statefun-flink/statefun-flink-distribution/src/main/resources/META-INF/NOTICE
+++ b/statefun-flink/statefun-flink-distribution/src/main/resources/META-INF/NOTICE
@@ -6,6 +6,8 @@ The Apache Software Foundation (http://www.apache.org/).
This project bundles the following dependencies under the Apache Software License 2.0 (http://www.apache.org/licenses/LICENSE-2.0.txt)
+- com.kohlschutter.junixsocket:junixsocket-core:2.3.2
+- com.kohlschutter.junixsocket:junixsocket-common:2.3.2
- commons-codec:commons-codec:1.10
- commons-lang:commons-lang:2.6
- commons-logging:commons-logging:1.1.3
[flink-statefun] 02/03: [FLINK-17611] [core] Bridge OkHttp and UNIX
domain sockets
Posted by tz...@apache.org.
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 9c43dc9c7fe57f69668163502d231a0ed63ecde7
Author: Igal Shilman <ig...@gmail.com>
AuthorDate: Tue May 19 20:19:23 2020 +0200
[FLINK-17611] [core] Bridge OkHttp and UNIX domain sockets
---
pom.xml | 5 -
.../flink/core/httpfn/HttpFunctionProvider.java | 69 +++-------
.../flink/core/httpfn/OkHttpUnixSocketBridge.java | 153 +++++++++++++++++++++
.../flink/core/httpfn/UnixDomainHttpEndpoint.java | 60 ++++++++
.../core/httpfn/HttpFunctionProviderTest.java | 35 -----
.../core/httpfn/UnixDomainHttpEndpointTest.java | 32 +++++
6 files changed, 261 insertions(+), 93 deletions(-)
diff --git a/pom.xml b/pom.xml
index 6f40a25..e1f3668 100644
--- a/pom.xml
+++ b/pom.xml
@@ -171,11 +171,6 @@ under the License.
</configuration>
</execution>
</executions>
- <configuration>
- <includeDirectories>
- <include>/usr/include/</include>
- </includeDirectories>
- </configuration>
</plugin>
</plugins>
</pluginManagement>
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 e0c9cb5..03159b1 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
@@ -18,22 +18,15 @@
package org.apache.flink.statefun.flink.core.httpfn;
-import java.net.URI;
-import java.nio.file.Path;
-import java.nio.file.Paths;
-import java.util.AbstractMap;
-import java.util.Collections;
+import static org.apache.flink.statefun.flink.core.httpfn.OkHttpUnixSocketBridge.configureUnixDomainSocket;
+
import java.util.Map;
-import java.util.stream.IntStream;
import okhttp3.HttpUrl;
import okhttp3.OkHttpClient;
-import okhttp3.Protocol;
-import org.apache.flink.annotation.VisibleForTesting;
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.StatefulFunctionProvider;
-import org.newsclub.net.unix.AFUNIXSocketFactory;
public class HttpFunctionProvider implements StatefulFunctionProvider {
private final Map<FunctionType, HttpFunctionSpec> supportedTypes;
@@ -55,54 +48,24 @@ public class HttpFunctionProvider implements StatefulFunctionProvider {
}
private RequestReplyClient buildHttpClient(HttpFunctionSpec spec) {
- // We need to build a UDS HTTP client
- if (spec.isUnixDomainSocket()) {
+ OkHttpClient.Builder clientBuilder = sharedClient.newBuilder();
+ clientBuilder.callTimeout(spec.maxRequestDuration());
- // We need to split the path in order to get the sock file and the path after the sock file
- Map.Entry<String, String> splittedFilePathAndEndpoint =
- splitFilePathAndEndpointForUDS(spec.endpoint());
+ final HttpUrl url;
+ if (spec.isUnixDomainSocket()) {
+ UnixDomainHttpEndpoint endpoint = UnixDomainHttpEndpoint.parseFrom(spec.endpoint());
- OkHttpClient specificClient =
- sharedClient
- .newBuilder()
- .socketFactory(
- new AFUNIXSocketFactory.FactoryArg(splittedFilePathAndEndpoint.getKey()))
- // Enable HTTP/2 if available (uses H2 upgrade),
- // otherwise fallback to HTTP/1.1
- .protocols(Collections.singletonList(Protocol.HTTP_2))
- .callTimeout(spec.maxRequestDuration())
+ url =
+ new HttpUrl.Builder()
+ .scheme("http")
+ .host("unused")
+ .addPathSegment(endpoint.pathSegment)
.build();
- return new HttpRequestReplyClient(
- // Only the path matters!
- HttpUrl.get(URI.create(splittedFilePathAndEndpoint.getValue())), specificClient);
- }
- // specific client reuses the same the connection pool and thread pool
- // as the sharedClient.
- OkHttpClient specificClient =
- sharedClient.newBuilder().callTimeout(spec.maxRequestDuration()).build();
- return new HttpRequestReplyClient(HttpUrl.get(spec.endpoint()), specificClient);
- }
-
- @VisibleForTesting
- static Map.Entry<String, String> splitFilePathAndEndpointForUDS(URI input) {
- // We need to split the path in order to get the sock file and the path after the sock file
- Path path = Paths.get(input.getPath());
-
- int sockPath =
- IntStream.rangeClosed(0, path.getNameCount() - 1)
- .filter(i -> path.getName(i).toString().endsWith(".sock"))
- .findFirst()
- .orElseThrow(
- () ->
- new IllegalStateException(
- "Unix Domain Socket path should contain a .sock file"));
-
- String filePath = "/" + path.subpath(0, sockPath + 1).toString();
- String endpoint = "/";
- if (sockPath != path.getNameCount() - 1) {
- endpoint = "/" + path.subpath(sockPath + 1, path.getNameCount()).toString();
+ configureUnixDomainSocket(clientBuilder, endpoint.unixDomainFile);
+ } else {
+ url = HttpUrl.get(spec.endpoint());
}
- return new AbstractMap.SimpleImmutableEntry<>(filePath, endpoint);
+ return new HttpRequestReplyClient(url, clientBuilder.build());
}
}
diff --git a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/OkHttpUnixSocketBridge.java b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/OkHttpUnixSocketBridge.java
new file mode 100644
index 0000000..b658ced
--- /dev/null
+++ b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/OkHttpUnixSocketBridge.java
@@ -0,0 +1,153 @@
+/*
+ * 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.httpfn;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.InetAddress;
+import java.net.Socket;
+import java.net.SocketAddress;
+import java.net.UnknownHostException;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import javax.net.SocketFactory;
+import okhttp3.Dns;
+import okhttp3.OkHttpClient;
+import org.apache.flink.util.IOUtils;
+import org.newsclub.net.unix.AFUNIXSocket;
+import org.newsclub.net.unix.AFUNIXSocketAddress;
+
+/** The following class holds utilities needed to bridge unix domain sockets and okhttp client. */
+final class OkHttpUnixSocketBridge {
+ private OkHttpUnixSocketBridge() {}
+
+ /** Configures the {@link OkHttpClient} builder to connect over a unix domain socket. */
+ static void configureUnixDomainSocket(OkHttpClient.Builder builder, File unixSocketFile) {
+ builder.socketFactory(new UnixSocketFactory(unixSocketFile)).dns(ConstantDnsLookup.INSTANCE);
+ }
+
+ /** resolve all host names to Ipv4 0.0.0.0 and port 0. */
+ private enum ConstantDnsLookup implements Dns {
+ INSTANCE;
+
+ @SuppressWarnings("NullableProblems")
+ @Override
+ public List<InetAddress> lookup(String hostname) throws UnknownHostException {
+ InetAddress address = InetAddress.getByAddress(hostname, new byte[] {0, 0, 0, 0});
+ return Collections.singletonList(address);
+ }
+ }
+
+ /**
+ * A {@code SocketFactory} that is bound to a specific path, and would return a {@code UnixSocket}
+ * for that path.
+ */
+ private static final class UnixSocketFactory extends SocketFactory {
+ private final File unixSocketFile;
+
+ public UnixSocketFactory(File unixSocketFile) {
+ this.unixSocketFile = Objects.requireNonNull(unixSocketFile);
+ }
+
+ @Override
+ public Socket createSocket() {
+ return new UnixSocket(unixSocketFile);
+ }
+
+ @Override
+ public Socket createSocket(String s, int i) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Socket createSocket(String s, int i, InetAddress inetAddress, int i1) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Socket createSocket(InetAddress inetAddress, int i) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Socket createSocket(InetAddress inetAddress, int i, InetAddress inetAddress1, int i1) {
+ throw new UnsupportedOperationException();
+ }
+ }
+
+ /**
+ * A {@code Socket} that is bound to a specific unix socket file, and delegates the relevant
+ * operations to {@link AFUNIXSocket}.
+ */
+ private static final class UnixSocket extends Socket {
+ private final File unixSocketFile;
+ private AFUNIXSocket delegate;
+
+ UnixSocket(File unixSocketFile) {
+ this.unixSocketFile = Objects.requireNonNull(unixSocketFile);
+ }
+
+ @Override
+ public void connect(SocketAddress endpoint, int timeout) throws IOException {
+ delegate = AFUNIXSocket.newInstance();
+ delegate.connect(new AFUNIXSocketAddress(unixSocketFile), timeout);
+ delegate.setSoTimeout(timeout);
+ }
+
+ @Override
+ public void bind(SocketAddress bindpoint) throws IOException {
+ delegate.bind(bindpoint);
+ }
+
+ @Override
+ public boolean isConnected() {
+ return delegate != null && delegate.isConnected();
+ }
+
+ @Override
+ public OutputStream getOutputStream() throws IOException {
+ return delegate.getOutputStream();
+ }
+
+ @Override
+ public InputStream getInputStream() throws IOException {
+ return delegate.getInputStream();
+ }
+
+ @Override
+ public synchronized void close() {
+ IOUtils.closeSocket(delegate);
+ delegate = null;
+ }
+
+ @Override
+ public boolean isClosed() {
+ return delegate.isClosed();
+ }
+
+ @Override
+ public synchronized void setSoTimeout(int timeout) {
+ // noop.
+ // we set the timeout after connecting
+ }
+ }
+}
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
new file mode 100644
index 0000000..c628fdb
--- /dev/null
+++ b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/UnixDomainHttpEndpoint.java
@@ -0,0 +1,60 @@
+/*
+ * 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.httpfn;
+
+import java.io.File;
+import java.net.URI;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.Objects;
+
+/** Represents a Unix domain file path and an http endpoint */
+final class UnixDomainHttpEndpoint {
+
+ /** Parses a URI of the form {@code http+unix://<file system path>.sock/<http endpoint>}. */
+ static UnixDomainHttpEndpoint parseFrom(URI endpoint) {
+ final Path path = Paths.get(endpoint.getPath());
+ final int sockPathIndex = indexOfSockFile(path);
+ final String filePath = "/" + path.subpath(0, sockPathIndex + 1).toString();
+ final File unixDomainFile = new File(filePath);
+
+ if (sockPathIndex == path.getNameCount() - 1) {
+ return new UnixDomainHttpEndpoint(unixDomainFile, "/");
+ }
+ String pathSegment = "/" + path.subpath(sockPathIndex + 1, path.getNameCount()).toString();
+ return new UnixDomainHttpEndpoint(unixDomainFile, pathSegment);
+ }
+
+ private static int indexOfSockFile(Path path) {
+ for (int i = 0; i < path.getNameCount(); i++) {
+ if (path.getName(i).toString().endsWith(".sock")) {
+ return i;
+ }
+ }
+ throw new IllegalStateException("Unix Domain Socket path should contain a .sock file");
+ }
+
+ final File unixDomainFile;
+ final String pathSegment;
+
+ private UnixDomainHttpEndpoint(File unixDomainFile, String endpoint) {
+ this.unixDomainFile = Objects.requireNonNull(unixDomainFile);
+ this.pathSegment = Objects.requireNonNull(endpoint);
+ }
+}
diff --git a/statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/httpfn/HttpFunctionProviderTest.java b/statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/httpfn/HttpFunctionProviderTest.java
deleted file mode 100644
index 21bbccc..0000000
--- a/statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/httpfn/HttpFunctionProviderTest.java
+++ /dev/null
@@ -1,35 +0,0 @@
-package org.apache.flink.statefun.flink.core.httpfn;
-
-import static org.junit.Assert.assertEquals;
-
-import java.net.URI;
-import java.util.Map;
-import org.junit.Test;
-
-public class HttpFunctionProviderTest {
-
- @Test
- public void splitOnlyWithFile() {
- Map.Entry<String, String> out =
- HttpFunctionProvider.splitFilePathAndEndpointForUDS(
- URI.create("http+unix:///some/path.sock"));
-
- assertEquals("/some/path.sock", out.getKey());
- assertEquals("/", out.getValue());
- }
-
- @Test
- public void splitOnlyWithFileAndEndpoint() {
- Map.Entry<String, String> out =
- HttpFunctionProvider.splitFilePathAndEndpointForUDS(
- URI.create("http+unix:///some/path.sock/hello"));
-
- assertEquals("/some/path.sock", out.getKey());
- assertEquals("/hello", out.getValue());
- }
-
- @Test(expected = IllegalStateException.class)
- public void missingSockFile() {
- HttpFunctionProvider.splitFilePathAndEndpointForUDS(URI.create("http+unix:///some/path/hello"));
- }
-}
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
new file mode 100644
index 0000000..f96ac3e
--- /dev/null
+++ b/statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/httpfn/UnixDomainHttpEndpointTest.java
@@ -0,0 +1,32 @@
+package org.apache.flink.statefun.flink.core.httpfn;
+
+import static org.junit.Assert.assertEquals;
+
+import java.net.URI;
+import org.junit.Test;
+
+public class UnixDomainHttpEndpointTest {
+
+ @Test
+ public void splitOnlyWithFile() {
+ UnixDomainHttpEndpoint out =
+ UnixDomainHttpEndpoint.parseFrom(URI.create("http+unix:///some/path.sock"));
+
+ assertEquals("/some/path.sock", out.unixDomainFile.toString());
+ assertEquals("/", out.pathSegment);
+ }
+
+ @Test
+ public void splitOnlyWithFileAndEndpoint() {
+ UnixDomainHttpEndpoint out =
+ UnixDomainHttpEndpoint.parseFrom(URI.create("http+unix:///some/path.sock/hello"));
+
+ assertEquals("/some/path.sock", out.unixDomainFile.toString());
+ assertEquals("/hello", out.pathSegment);
+ }
+
+ @Test(expected = IllegalStateException.class)
+ public void missingSockFile() {
+ UnixDomainHttpEndpoint.parseFrom(URI.create("http+unix:///some/path/hello"));
+ }
+}
[flink-statefun] 03/03: [FLINK-17611] [core] Add an ITCase for Unix
domain sockets
Posted by tz...@apache.org.
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 1c018e19d96f44334fe1b24280337679dc2a7be8
Author: Igal Shilman <ig...@gmail.com>
AuthorDate: Wed May 20 20:52:27 2020 +0200
[FLINK-17611] [core] Add an ITCase for Unix domain sockets
Co-authored-by: slinkydeveloper <fr...@gmail.com>
This closes #114.
---
statefun-flink/statefun-flink-core/pom.xml | 6 ++
.../flink/core/httpfn/UnixDomainSocketITCase.java | 99 ++++++++++++++++++++++
2 files changed, 105 insertions(+)
diff --git a/statefun-flink/statefun-flink-core/pom.xml b/statefun-flink/statefun-flink-core/pom.xml
index e99077e..d140f9b 100644
--- a/statefun-flink/statefun-flink-core/pom.xml
+++ b/statefun-flink/statefun-flink-core/pom.xml
@@ -100,6 +100,12 @@ under the License.
<version>${flink.version}</version>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>com.squareup.okhttp3</groupId>
+ <artifactId>mockwebserver</artifactId>
+ <version>${okhttp.version}</version>
+ <scope>test</scope>
+ </dependency>
<!-- benchmarks -->
<dependency>
diff --git a/statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/httpfn/UnixDomainSocketITCase.java b/statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/httpfn/UnixDomainSocketITCase.java
new file mode 100644
index 0000000..8ba733c
--- /dev/null
+++ b/statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/httpfn/UnixDomainSocketITCase.java
@@ -0,0 +1,99 @@
+/*
+ * 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.httpfn;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.CoreMatchers.notNullValue;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.ServerSocket;
+import javax.net.ServerSocketFactory;
+import okhttp3.OkHttpClient;
+import okhttp3.OkHttpClient.Builder;
+import okhttp3.Request;
+import okhttp3.Response;
+import okhttp3.mockwebserver.MockResponse;
+import okhttp3.mockwebserver.MockWebServer;
+import org.junit.Test;
+import org.newsclub.net.unix.AFUNIXServerSocket;
+import org.newsclub.net.unix.AFUNIXSocketAddress;
+
+public class UnixDomainSocketITCase {
+
+ @Test(timeout = 10 * 1_000)
+ public void unixDomainSocket() throws IOException {
+ final File sockFile = new File("/tmp/uds-" + System.nanoTime() + ".sock");
+ sockFile.deleteOnExit();
+
+ try (MockWebServer server = new MockWebServer()) {
+ server.setServerSocketFactory(udsSocketFactory(sockFile));
+ server.enqueue(new MockResponse().setBody("hi"));
+ server.start();
+
+ OkHttpClient client = udsSocketClient(sockFile);
+
+ Response response = request(client);
+
+ assertTrue(response.isSuccessful());
+ assertThat(response.body(), is(notNullValue()));
+ assertThat(response.body().string(), is("hi"));
+ }
+ }
+
+ private static Response request(OkHttpClient client) throws IOException {
+ Request request = new Request.Builder().url("http://unused/").build();
+ return client.newCall(request).execute();
+ }
+
+ /** returns an {@link OkHttpClient} that connects trough the provided socket file. */
+ private static OkHttpClient udsSocketClient(File sockFile) {
+ Builder sharedClient = OkHttpUtils.newClient().newBuilder();
+ OkHttpUnixSocketBridge.configureUnixDomainSocket(sharedClient, sockFile);
+ return sharedClient.build();
+ }
+
+ private static ServerSocketFactory udsSocketFactory(File sockFile) {
+ return new ServerSocketFactory() {
+ @Override
+ public ServerSocket createServerSocket() throws IOException {
+ return AFUNIXServerSocket.forceBindOn(new AFUNIXSocketAddress(sockFile));
+ }
+
+ @Override
+ public ServerSocket createServerSocket(int i) throws IOException {
+ return createServerSocket();
+ }
+
+ @Override
+ public ServerSocket createServerSocket(int i, int i1) throws IOException {
+ return createServerSocket();
+ }
+
+ @Override
+ public ServerSocket createServerSocket(int i, int i1, InetAddress inetAddress)
+ throws IOException {
+ return createServerSocket();
+ }
+ };
+ }
+}