Support remote zilla configuration with change detection (#1071)

* Remove BindingConfig.readLocation * Remove GuardedConfig.readPath * Rename ConfigAdapterContext.readLocation to ConfigAdapterContext.readResource * Deprecate ConfigAdapterContext for removal * Add EngineConfiguration.configURI() to resolve absolute file path * Remove EngineConfiguration.configURL() but use to default EngineConfiguration.configURI() * Simplify EngineManager * Gather resources on NamespaceConfig from member configs * Consolidate config and resource watcher as EngineConfigWatcher * Resolve watched paths based on path filesystem provider scheme * Configure HttpFileSystem poll internval duration via Map<String,?> env * Simplify ReconfigureHttpIT scripts and include /zilla.yaml in request path * HttpFileSystem per origin (root path) not per individual path * Track HttpPath change count vs read count to implement simple caching * No watch event needed for identical response body * Handle status 204 with null body and infer delay for optional prefer wait --------- Co-authored-by: John Fallows <john.r.fallows@gmail.com>
aklivity · Jun 26, 2024 · 230eec2 · 230eec2
1 parent 89f2863
commit 230eec2
Show file tree

Hide file tree

Showing 120 changed files with 5,229 additions and 1,073 deletions.
diff --git a/cloud/docker-image/pom.xml b/cloud/docker-image/pom.xml
@@ -253,6 +253,12 @@
       <version>${project.version}</version>
       <scope>runtime</scope>
     </dependency>
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>filesystem-http</artifactId>
+      <version>${project.version}</version>
+      <scope>runtime</scope>
+    </dependency>
     <dependency>
       <groupId>${project.groupId}</groupId>
       <artifactId>metrics-stream</artifactId>

diff --git a/cloud/docker-image/src/main/docker/assembly.xml b/cloud/docker-image/src/main/docker/assembly.xml
@@ -30,6 +30,7 @@
         <include>io/aklivity/zilla/binding-*/**</include>
         <include>io/aklivity/zilla/catalog-*/**</include>
         <include>io/aklivity/zilla/exporter-*/**</include>
+        <include>io/aklivity/zilla/filesystem-*/**</include>
         <include>io/aklivity/zilla/guard-*/**</include>
         <include>io/aklivity/zilla/metrics-*/**</include>
         <include>io/aklivity/zilla/model-*/**</include>

diff --git a/cloud/docker-image/src/main/docker/zpm.json.template b/cloud/docker-image/src/main/docker/zpm.json.template
@@ -50,6 +50,7 @@
     "io.aklivity.zilla:exporter-otlp",
     "io.aklivity.zilla:exporter-prometheus",
     "io.aklivity.zilla:exporter-stdout",
+    "io.aklivity.zilla:filesystem-http",
     "io.aklivity.zilla:guard-jwt",
     "io.aklivity.zilla:metrics-stream",
     "io.aklivity.zilla:metrics-http",

diff --git a/...ava/io/aklivity/zilla/runtime/binding/asyncapi/internal/config/AsyncapiBindingConfig.java b/...ava/io/aklivity/zilla/runtime/binding/asyncapi/internal/config/AsyncapiBindingConfig.java
@@ -320,7 +320,6 @@ private void attachProxyBinding(
         namespaceGenerator.init(binding);
         final List<String> labels = configs.stream().map(c -> c.apiLabel).collect(toList());
         final NamespaceConfig composite = namespaceGenerator.generateProxy(binding, asyncapis, schemaIdsByApiId::get, labels);
-        composite.readURL = binding.readURL;
         attach.accept(composite);
         updateNamespace(configs, composite, new ArrayList<>(asyncapis.values()));
     }
@@ -349,7 +348,6 @@ private void attachServerClientBinding(
             namespaceConfig.servers.forEach(s -> s.setAsyncapiProtocol(
                 namespaceGenerator.resolveProtocol(s.protocol(), options, namespaceConfig.asyncapis, namespaceConfig.servers)));
             final NamespaceConfig composite = namespaceGenerator.generate(binding, namespaceConfig);
-            composite.readURL = binding.readURL;
             attach.accept(composite);
             updateNamespace(namespaceConfig.configs, composite, namespaceConfig.asyncapis);
         }

diff --git a/...-echo/src/test/java/io/aklivity/zilla/runtime/binding/echo/internal/bench/EchoWorker.java b/...-echo/src/test/java/io/aklivity/zilla/runtime/binding/echo/internal/bench/EchoWorker.java
@@ -16,8 +16,8 @@
 package io.aklivity.zilla.runtime.binding.echo.internal.bench;
 
 import java.net.InetAddress;
-import java.net.URL;
 import java.nio.channels.SelectableChannel;
+import java.nio.file.Path;
 import java.time.Clock;
 import java.util.function.LongSupplier;
 
@@ -319,8 +319,8 @@ public ConverterHandler supplyWriteConverter(
     }
 
     @Override
-    public URL resolvePath(
-        String path)
+    public Path resolvePath(
+        String location)
     {
         return null;
     }

diff --git a/...java/io/aklivity/zilla/runtime/binding/grpc/internal/config/GrpcOptionsConfigAdapter.java b/...java/io/aklivity/zilla/runtime/binding/grpc/internal/config/GrpcOptionsConfigAdapter.java
@@ -41,7 +41,7 @@ public final class GrpcOptionsConfigAdapter implements OptionsConfigAdapterSpi,
 
     private final GrpcProtobufParser parser = new GrpcProtobufParser();
 
-    private Function<String, String> readURL;
+    private Function<String, String> readResource;
 
     @Override
     public Kind kind()
@@ -88,7 +88,7 @@ public OptionsConfig adaptFromJson(
     public void adaptContext(
         ConfigAdapterContext context)
     {
-        this.readURL = context::readURL;
+        this.readResource = context::readResource;
     }
 
     private List<GrpcProtobufConfig> asListProtobufs(
@@ -103,7 +103,7 @@ private GrpcProtobufConfig asProtobuf(
         JsonValue value)
     {
         final String location = ((JsonString) value).getString();
-        final String protobuf = readURL.apply(location);
+        final String protobuf = readResource.apply(location);
 
         return parser.parse(location, protobuf);
     }

diff --git a/.../io/aklivity/zilla/runtime/binding/grpc/internal/config/GrpcOptionsConfigAdapterTest.java b/.../io/aklivity/zilla/runtime/binding/grpc/internal/config/GrpcOptionsConfigAdapterTest.java
@@ -66,7 +66,7 @@ public void initJson() throws IOException
         {
             content = new String(resource.readAllBytes(), UTF_8);
         }
-        Mockito.doReturn(content).when(context).readURL("protobuf/echo.proto");
+        Mockito.doReturn(content).when(context).readResource("protobuf/echo.proto");
         adapter = new OptionsConfigAdapter(OptionsConfigAdapterSpi.Kind.BINDING, context);
         adapter.adaptType("grpc");
         JsonbConfig config = new JsonbConfig()

diff --git a/...g-http/src/main/java/io/aklivity/zilla/runtime/binding/http/config/HttpOptionsConfig.java b/...g-http/src/main/java/io/aklivity/zilla/runtime/binding/http/config/HttpOptionsConfig.java
@@ -27,6 +27,7 @@
 
 import io.aklivity.zilla.runtime.binding.http.internal.types.String16FW;
 import io.aklivity.zilla.runtime.binding.http.internal.types.String8FW;
+import io.aklivity.zilla.runtime.engine.config.ModelConfig;
 import io.aklivity.zilla.runtime.engine.config.OptionsConfig;
 
 public final class HttpOptionsConfig extends OptionsConfig
@@ -55,41 +56,46 @@ public static <T> HttpOptionsConfigBuilder<T> builder(
         HttpAuthorizationConfig authorization,
         List<HttpRequestConfig> requests)
     {
-        super(requests != null && !requests.isEmpty()
-            ? requests.stream()
-                .flatMap(request -> Stream.concat(
-                    Stream.of(request.content),
-                    Stream.concat(
-                        request.headers != null
-                            ? request.headers.stream().flatMap(header -> Stream.of(header != null ? header.model : null))
-                            : Stream.empty(),
-                        Stream.concat(
-                            request.pathParams != null
-                                ? request.pathParams.stream().flatMap(param -> Stream.of(param != null ? param.model : null))
-                                : Stream.empty(),
-                            Stream.concat(
-                                    request.queryParams != null
-                                    ? request.queryParams.stream().flatMap(param -> Stream.of(param != null ? param.model : null))
-                                    : Stream.empty(),
-                                    Stream.concat(request.responses != null
-                                        ? request.responses.stream().flatMap(param -> Stream.of(param != null
-                                            ? param.content
-                                            : null))
-                                        : Stream.empty(), request.responses != null
-                                        ? request.responses.stream()
-                                            .flatMap(response -> response.headers != null
-                                                ? response.headers.stream()
-                                                    .flatMap(param -> Stream.of(param != null ? param.model : null))
-                                                : Stream.empty())
-                                        : Stream.empty())
-                            )))).filter(Objects::nonNull))
-                .collect(Collectors.toList())
-            : emptyList());
-
+        super(resolveModels(requests), List.of());
         this.versions = versions;
         this.overrides = overrides;
         this.access = access;
         this.authorization = authorization;
         this.requests = requests;
     }
+
+    private static List<ModelConfig> resolveModels(
+        List<HttpRequestConfig> requests)
+    {
+        return requests != null && !requests.isEmpty()
+            ? requests.stream()
+            .flatMap(request -> Stream.concat(
+                Stream.of(request.content),
+                Stream.concat(
+                    request.headers != null
+                        ? request.headers.stream().flatMap(header -> Stream.of(header != null ? header.model : null))
+                        : Stream.empty(),
+                    Stream.concat(
+                        request.pathParams != null
+                            ? request.pathParams.stream().flatMap(param -> Stream.of(param != null ? param.model : null))
+                            : Stream.empty(),
+                        Stream.concat(
+                            request.queryParams != null
+                                ? request.queryParams.stream().flatMap(param -> Stream.of(param != null ? param.model : null))
+                                : Stream.empty(),
+                            Stream.concat(request.responses != null
+                                ? request.responses.stream().flatMap(param -> Stream.of(param != null
+                                ? param.content
+                                : null))
+                                : Stream.empty(), request.responses != null
+                                ? request.responses.stream()
+                                .flatMap(response -> response.headers != null
+                                    ? response.headers.stream()
+                                    .flatMap(param -> Stream.of(param != null ? param.model : null))
+                                    : Stream.empty())
+                                : Stream.empty())
+                        )))).filter(Objects::nonNull))
+            .collect(Collectors.toList())
+            : emptyList();
+    }
 }
diff --git a/...afka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/config/KafkaOptionsConfig.java b/...afka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/config/KafkaOptionsConfig.java
@@ -23,6 +23,7 @@
 import java.util.function.Function;
 import java.util.stream.Stream;
 
+import io.aklivity.zilla.runtime.engine.config.ModelConfig;
 import io.aklivity.zilla.runtime.engine.config.OptionsConfig;
 
 public final class KafkaOptionsConfig extends OptionsConfig
@@ -49,15 +50,21 @@ public static <T> KafkaOptionsConfigBuilder<T> builder(
         List<KafkaServerConfig> servers,
         KafkaSaslConfig sasl)
     {
-        super(topics != null && !topics.isEmpty()
-            ? topics.stream()
-                .flatMap(t -> Stream.of(t.key, t.value))
-                .filter(Objects::nonNull)
-                .collect(toList())
-            : emptyList());
+        super(resolveModels(topics), List.of());
         this.bootstrap = bootstrap;
         this.topics = topics;
         this.servers = servers;
         this.sasl = sasl;
     }
+
+    private static List<ModelConfig> resolveModels(
+        List<KafkaTopicConfig> topics)
+    {
+        return topics != null && !topics.isEmpty()
+            ? topics.stream()
+                .flatMap(t -> Stream.of(t.key, t.value))
+                .filter(Objects::nonNull)
+                .collect(toList())
+            : emptyList();
+    }
 }
diff --git a/...g-mqtt/src/main/java/io/aklivity/zilla/runtime/binding/mqtt/config/MqttOptionsConfig.java b/...g-mqtt/src/main/java/io/aklivity/zilla/runtime/binding/mqtt/config/MqttOptionsConfig.java
@@ -26,6 +26,7 @@
 import java.util.stream.Stream;
 
 import io.aklivity.zilla.runtime.binding.mqtt.internal.config.MqttVersion;
+import io.aklivity.zilla.runtime.engine.config.ModelConfig;
 import io.aklivity.zilla.runtime.engine.config.OptionsConfig;
 
 public class MqttOptionsConfig extends OptionsConfig
@@ -50,18 +51,24 @@ public MqttOptionsConfig(
         List<MqttTopicConfig> topics,
         List<MqttVersion> versions)
     {
-        super(topics != null && !topics.isEmpty()
+        super(resolveModels(topics), List.of());
+        this.authorization = authorization;
+        this.topics = topics;
+        this.versions = versions;
+    }
+
+    private static List<ModelConfig> resolveModels(
+        List<MqttTopicConfig> topics)
+    {
+        return topics != null && !topics.isEmpty()
             ? topics.stream()
             .flatMap(topic -> Stream.concat(
-                Stream.of(topic.content),
+                    Stream.of(topic.content),
                     Optional.ofNullable(topic.userProperties).orElseGet(Collections::emptyList).stream()
-                    .flatMap(p -> Stream.of(p.value))
-                    .filter(Objects::nonNull))
+                        .flatMap(p -> Stream.of(p.value))
+                        .filter(Objects::nonNull))
                 .filter(Objects::nonNull))
             .collect(Collectors.toList())
-            : emptyList());
-        this.authorization = authorization;
-        this.topics = topics;
-        this.versions = versions;
+            : emptyList();
     }
 }
diff --git a/.../zilla/runtime/binding/openapi/asyncapi/internal/config/OpenapiAsyncapiBindingConfig.java b/.../zilla/runtime/binding/openapi/asyncapi/internal/config/OpenapiAsyncapiBindingConfig.java
@@ -144,7 +144,6 @@ public void attach(
                     Object2ObjectHashMap::new));
 
         this.composite = namespaceGenerator.generate(binding, openapis, asyncapis, openapiSchemaIdsByApiId::get);
-        this.composite.readURL = binding.readURL;
         attach.accept(this.composite);
 
         BindingConfig mappingBinding = composite.bindings.stream()

diff --git a/.../java/io/aklivity/zilla/runtime/binding/openapi/internal/config/OpenapiBindingConfig.java b/.../java/io/aklivity/zilla/runtime/binding/openapi/internal/config/OpenapiBindingConfig.java
@@ -138,7 +138,6 @@ public void attach(
         for (OpenapiNamespaceConfig namespaceConfig : namespaceConfigs.values())
         {
             final NamespaceConfig composite = namespaceGenerator.generate(binding, namespaceConfig);
-            composite.readURL = binding.readURL;
             attach.accept(composite);
             namespaceConfig.configs.forEach(c ->
             {

diff --git a/...ding-sse/src/main/java/io/aklivity/zilla/runtime/binding/sse/config/SseOptionsConfig.java b/...ding-sse/src/main/java/io/aklivity/zilla/runtime/binding/sse/config/SseOptionsConfig.java
@@ -23,6 +23,7 @@
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
+import io.aklivity.zilla.runtime.engine.config.ModelConfig;
 import io.aklivity.zilla.runtime.engine.config.OptionsConfig;
 
 public final class SseOptionsConfig extends OptionsConfig
@@ -46,14 +47,20 @@ public static <T> SseOptionsConfigBuilder<T> builder(
         int retry,
         List<SseRequestConfig> requests)
     {
-        super(requests != null && !requests.isEmpty()
+        super(resolveModels(requests), List.of());
+        this.retry = retry;
+        this.requests = requests;
+    }
+
+    private static List<ModelConfig> resolveModels(
+        List<SseRequestConfig> requests)
+    {
+        return requests != null && !requests.isEmpty()
             ? requests.stream()
             .flatMap(path ->
                 Stream.of(path.content)
                     .filter(Objects::nonNull))
             .collect(Collectors.toList())
-            : emptyList());
-        this.retry = retry;
-        this.requests = requests;
+            : emptyList();
     }
 }
diff --git a/...ing-tls/src/test/java/io/aklivity/zilla/runtime/binding/tls/internal/bench/TlsWorker.java b/...ing-tls/src/test/java/io/aklivity/zilla/runtime/binding/tls/internal/bench/TlsWorker.java
@@ -18,12 +18,10 @@
 import static io.aklivity.zilla.runtime.engine.internal.stream.StreamId.isInitial;
 import static java.lang.ThreadLocal.withInitial;
 import static java.nio.charset.StandardCharsets.UTF_8;
-import static org.agrona.LangUtil.rethrowUnchecked;
 
 import java.net.InetAddress;
-import java.net.MalformedURLException;
-import java.net.URL;
 import java.nio.channels.SelectableChannel;
+import java.nio.file.Path;
 import java.time.Clock;
 import java.util.function.IntConsumer;
 import java.util.function.LongSupplier;
@@ -85,7 +83,7 @@ public class TlsWorker implements EngineContext
     private final BindingFactory factory;
     private final VaultFactory vaultFactory;
     private final Configuration config;
-    private final URL configURL;
+    private final Path configPath;
 
     private final TlsSignaler signaler;
 
@@ -105,7 +103,7 @@ public TlsWorker(
                 .readonly(false)
                 .build()
                 .bufferPool();
-        this.configURL = config.configURL();
+        this.configPath = Path.of(config.configURI());
 
         this.signaler = new TlsSignaler();
 
@@ -387,19 +385,10 @@ public ConverterHandler supplyWriteConverter(
     }
 
     @Override
-    public URL resolvePath(
-        String path)
+    public Path resolvePath(
+        String location)
     {
-        URL resolved = null;
-        try
-        {
-            resolved = new URL(configURL, path);
-        }
-        catch (MalformedURLException ex)
-        {
-            rethrowUnchecked(ex);
-        }
-        return resolved;
+        return configPath.resolveSibling(location);
     }
 
     @Override

diff --git a/.../java/io/aklivity/zilla/runtime/catalog/filesystem/internal/FilesystemCatalogHandler.java b/.../java/io/aklivity/zilla/runtime/catalog/filesystem/internal/FilesystemCatalogHandler.java
@@ -15,7 +15,8 @@
 package io.aklivity.zilla.runtime.catalog.filesystem.internal;
 
 import java.io.InputStream;
-import java.net.URL;
+import java.nio.file.Files;
+import java.nio.file.Path;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -34,7 +35,7 @@ public class FilesystemCatalogHandler implements CatalogHandler
     private final CRC32C crc32c;
     private final FilesystemEventContext event;
     private final long catalogId;
-    private final Function<String, URL> resolvePath;
+    private final Function<String, Path> resolvePath;
 
     public FilesystemCatalogHandler(
         FilesystemOptionsConfig config,
@@ -72,8 +73,8 @@ private void registerSchema(
         {
             try
             {
-                URL storeURL = resolvePath.apply(config.path);
-                try (InputStream input = storeURL.openStream())
+                Path storePath = resolvePath.apply(config.path);
+                try (InputStream input = Files.newInputStream(storePath))
                 {
                     String schema = new String(input.readAllBytes());
                     int schemaId = generateCRC32C(schema);